Implement autobackup-operator controllers, tests, CI and packaging
PVCs and CloudNativePG Clusters need S3 buckets and backup schedules
provisioned consistently. This operator watches the
backups.unkin.net/{schedule,destination} annotations on those objects and
provisions everything needed to back them up, with no new CRDs.
- Add a PVC controller that provisions cephrgw ObjectStoreUser/Bucket/BucketAccess,
auto-generates a restic repo-password Secret and creates a k8up Schedule scoped
to the PVC via spec.backup.volumes[].persistentVolumeClaim.claimName.
- Add a CNPG Cluster controller that provisions the same bucket stack, idempotently
patches spec.backup.barmanObjectStore (leaving a user-set destinationPath alone
with a Warning event) and creates a ScheduledBackup.
- Resolve destinations through a ConfigMap lookup table; requeue until the
BucketAccess is Ready before creating schedule resources; own-reference created
resources and retain bucket data by default.
- Add schedule-mapping helpers (k8up 5-field/shortcut pass-through, CNPG 6-field
seconds-first) and deterministic, length-bounded name derivation.
- Add unit tests (schedule mapping, name derivation, destination resolution) and
envtest controller tests for both paths, wiring the external CRDs into envtest.
- Add kubebuilder-generated RBAC, a Dockerfile (distroless/nonroot), Woodpecker
lint/test/build pipelines and a tag-triggered image push to the artifactapi
docker-internal registry, plus a version-bump Makefile and deploy manifests.
This commit is contained in:
+291
@@ -0,0 +1,291 @@
|
||||
/*
|
||||
Copyright 2018 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package controller
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/go-logr/logr"
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
"k8s.io/klog/v2"
|
||||
"k8s.io/utils/ptr"
|
||||
|
||||
"sigs.k8s.io/controller-runtime/pkg/config"
|
||||
"sigs.k8s.io/controller-runtime/pkg/controller/priorityqueue"
|
||||
"sigs.k8s.io/controller-runtime/pkg/internal/controller"
|
||||
"sigs.k8s.io/controller-runtime/pkg/manager"
|
||||
"sigs.k8s.io/controller-runtime/pkg/reconcile"
|
||||
"sigs.k8s.io/controller-runtime/pkg/source"
|
||||
)
|
||||
|
||||
// Options are the arguments for creating a new Controller.
|
||||
type Options = TypedOptions[reconcile.Request]
|
||||
|
||||
// TypedOptions are the arguments for creating a new Controller.
|
||||
type TypedOptions[request comparable] struct {
|
||||
// SkipNameValidation allows skipping the name validation that ensures that every controller name is unique.
|
||||
// Unique controller names are important to get unique metrics and logs for a controller.
|
||||
// Defaults to the Controller.SkipNameValidation setting from the Manager if unset.
|
||||
// Defaults to false if Controller.SkipNameValidation setting from the Manager is also unset.
|
||||
SkipNameValidation *bool
|
||||
|
||||
// MaxConcurrentReconciles is the maximum number of concurrent Reconciles which can be run. Defaults to 1.
|
||||
MaxConcurrentReconciles int
|
||||
|
||||
// CacheSyncTimeout refers to the time limit set to wait for syncing caches.
|
||||
// Defaults to 2 minutes if not set.
|
||||
CacheSyncTimeout time.Duration
|
||||
|
||||
// RecoverPanic indicates whether the panic caused by reconcile should be recovered.
|
||||
// Defaults to the Controller.RecoverPanic setting from the Manager if unset.
|
||||
// Defaults to true if Controller.RecoverPanic setting from the Manager is also unset.
|
||||
RecoverPanic *bool
|
||||
|
||||
// NeedLeaderElection indicates whether the controller needs to use leader election.
|
||||
// Defaults to true, which means the controller will use leader election.
|
||||
NeedLeaderElection *bool
|
||||
|
||||
// Reconciler reconciles an object
|
||||
Reconciler reconcile.TypedReconciler[request]
|
||||
|
||||
// RateLimiter is used to limit how frequently requests may be queued.
|
||||
// Defaults to MaxOfRateLimiter which has both overall and per-item rate limiting.
|
||||
// The overall is a token bucket and the per-item is exponential.
|
||||
RateLimiter workqueue.TypedRateLimiter[request]
|
||||
|
||||
// NewQueue constructs the queue for this controller once the controller is ready to start.
|
||||
// With NewQueue a custom queue implementation can be used, e.g. a priority queue to prioritize with which
|
||||
// priority/order objects are reconciled (e.g. to reconcile objects with changes first).
|
||||
// This is a func because the standard Kubernetes work queues start themselves immediately, which
|
||||
// leads to goroutine leaks if something calls controller.New repeatedly.
|
||||
// The NewQueue func gets the controller name and the RateLimiter option (defaulted if necessary) passed in.
|
||||
// NewQueue defaults to NewRateLimitingQueueWithConfig.
|
||||
//
|
||||
// NOTE: LOW LEVEL PRIMITIVE!
|
||||
// Only use a custom NewQueue if you know what you are doing.
|
||||
NewQueue func(controllerName string, rateLimiter workqueue.TypedRateLimiter[request]) workqueue.TypedRateLimitingInterface[request]
|
||||
|
||||
// Logger will be used to build a default LogConstructor if unset.
|
||||
Logger logr.Logger
|
||||
|
||||
// LogConstructor is used to construct a logger used for this controller and passed
|
||||
// to each reconciliation via the context field.
|
||||
LogConstructor func(request *request) logr.Logger
|
||||
|
||||
// UsePriorityQueue configures the controllers queue to use the controller-runtime provided
|
||||
// priority queue.
|
||||
//
|
||||
// Note: This flag is enabled by default.
|
||||
// For more details, see: https://github.com/kubernetes-sigs/controller-runtime/issues/2374.
|
||||
UsePriorityQueue *bool
|
||||
|
||||
// EnableWarmup specifies whether the controller should start its sources when the manager is not
|
||||
// the leader. This is useful for cases where sources take a long time to start, as it allows
|
||||
// for the controller to warm up its caches even before it is elected as the leader. This
|
||||
// improves leadership failover time, as the caches will be prepopulated before the controller
|
||||
// transitions to be leader.
|
||||
//
|
||||
// Setting EnableWarmup to true and NeedLeaderElection to true means the controller will start its
|
||||
// sources without waiting to become leader.
|
||||
// Setting EnableWarmup to true and NeedLeaderElection to false is a no-op as controllers without
|
||||
// leader election do not wait on leader election to start their sources.
|
||||
// Defaults to false.
|
||||
//
|
||||
// Note: This feature is currently in beta and subject to change.
|
||||
// For more details, see: https://github.com/kubernetes-sigs/controller-runtime/issues/3220.
|
||||
EnableWarmup *bool
|
||||
|
||||
// ReconciliationTimeout is used as the timeout passed to the context of each Reconcile call.
|
||||
// By default, there is no timeout.
|
||||
ReconciliationTimeout time.Duration
|
||||
}
|
||||
|
||||
// DefaultFromConfig defaults the config from a config.Controller
|
||||
func (options *TypedOptions[request]) DefaultFromConfig(config config.Controller) {
|
||||
if options.Logger.GetSink() == nil {
|
||||
options.Logger = config.Logger
|
||||
}
|
||||
|
||||
if options.SkipNameValidation == nil {
|
||||
options.SkipNameValidation = config.SkipNameValidation
|
||||
}
|
||||
|
||||
if options.MaxConcurrentReconciles <= 0 && config.MaxConcurrentReconciles > 0 {
|
||||
options.MaxConcurrentReconciles = config.MaxConcurrentReconciles
|
||||
}
|
||||
|
||||
if options.CacheSyncTimeout == 0 && config.CacheSyncTimeout > 0 {
|
||||
options.CacheSyncTimeout = config.CacheSyncTimeout
|
||||
}
|
||||
|
||||
if options.UsePriorityQueue == nil {
|
||||
options.UsePriorityQueue = config.UsePriorityQueue
|
||||
}
|
||||
|
||||
if options.RecoverPanic == nil {
|
||||
options.RecoverPanic = config.RecoverPanic
|
||||
}
|
||||
|
||||
if options.NeedLeaderElection == nil {
|
||||
options.NeedLeaderElection = config.NeedLeaderElection
|
||||
}
|
||||
|
||||
if options.EnableWarmup == nil {
|
||||
options.EnableWarmup = config.EnableWarmup
|
||||
}
|
||||
|
||||
if options.ReconciliationTimeout == 0 {
|
||||
options.ReconciliationTimeout = config.ReconciliationTimeout
|
||||
}
|
||||
}
|
||||
|
||||
// Controller implements an API. A Controller manages a work queue fed reconcile.Requests
|
||||
// from source.Sources. Work is performed through the reconcile.Reconciler for each enqueued item.
|
||||
// Work typically is reads and writes Kubernetes objects to make the system state match the state specified
|
||||
// in the object Spec.
|
||||
type Controller = TypedController[reconcile.Request]
|
||||
|
||||
// TypedController implements an API.
|
||||
type TypedController[request comparable] interface {
|
||||
// Reconciler is called to reconcile an object by Namespace/Name
|
||||
reconcile.TypedReconciler[request]
|
||||
|
||||
// Watch watches the provided Source.
|
||||
Watch(src source.TypedSource[request]) error
|
||||
|
||||
// Start starts the controller. Start blocks until the context is closed or a
|
||||
// controller has an error starting.
|
||||
Start(ctx context.Context) error
|
||||
|
||||
// GetLogger returns this controller logger prefilled with basic information.
|
||||
GetLogger() logr.Logger
|
||||
}
|
||||
|
||||
// New returns a new Controller registered with the Manager. The Manager will ensure that shared Caches have
|
||||
// been synced before the Controller is Started.
|
||||
//
|
||||
// The name must be unique as it is used to identify the controller in metrics and logs.
|
||||
func New(name string, mgr manager.Manager, options Options) (Controller, error) {
|
||||
return NewTyped(name, mgr, options)
|
||||
}
|
||||
|
||||
// NewTyped returns a new typed controller registered with the Manager,
|
||||
//
|
||||
// The name must be unique as it is used to identify the controller in metrics and logs.
|
||||
func NewTyped[request comparable](name string, mgr manager.Manager, options TypedOptions[request]) (TypedController[request], error) {
|
||||
options.DefaultFromConfig(mgr.GetControllerOptions())
|
||||
c, err := NewTypedUnmanaged(name, options)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Add the controller as a Manager components
|
||||
return c, mgr.Add(c)
|
||||
}
|
||||
|
||||
// NewUnmanaged returns a new controller without adding it to the manager. The
|
||||
// caller is responsible for starting the returned controller.
|
||||
//
|
||||
// The name must be unique as it is used to identify the controller in metrics and logs.
|
||||
func NewUnmanaged(name string, options Options) (Controller, error) {
|
||||
return NewTypedUnmanaged(name, options)
|
||||
}
|
||||
|
||||
// NewTypedUnmanaged returns a new typed controller without adding it to the manager.
|
||||
//
|
||||
// The name must be unique as it is used to identify the controller in metrics and logs.
|
||||
func NewTypedUnmanaged[request comparable](name string, options TypedOptions[request]) (TypedController[request], error) {
|
||||
if options.Reconciler == nil {
|
||||
return nil, fmt.Errorf("must specify Reconciler")
|
||||
}
|
||||
|
||||
if len(name) == 0 {
|
||||
return nil, fmt.Errorf("must specify Name for Controller")
|
||||
}
|
||||
|
||||
if options.SkipNameValidation == nil || !*options.SkipNameValidation {
|
||||
if err := checkName(name); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
if options.LogConstructor == nil {
|
||||
log := options.Logger.WithValues(
|
||||
"controller", name,
|
||||
)
|
||||
options.LogConstructor = func(in *request) logr.Logger {
|
||||
log := log
|
||||
if req, ok := any(in).(*reconcile.Request); ok && req != nil {
|
||||
log = log.WithValues(
|
||||
"object", klog.KRef(req.Namespace, req.Name),
|
||||
"namespace", req.Namespace, "name", req.Name,
|
||||
)
|
||||
}
|
||||
return log
|
||||
}
|
||||
}
|
||||
|
||||
if options.MaxConcurrentReconciles <= 0 {
|
||||
options.MaxConcurrentReconciles = 1
|
||||
}
|
||||
|
||||
if options.CacheSyncTimeout == 0 {
|
||||
options.CacheSyncTimeout = 2 * time.Minute
|
||||
}
|
||||
|
||||
if options.RateLimiter == nil {
|
||||
if ptr.Deref(options.UsePriorityQueue, true) {
|
||||
options.RateLimiter = workqueue.NewTypedItemExponentialFailureRateLimiter[request](5*time.Millisecond, 1000*time.Second)
|
||||
} else {
|
||||
options.RateLimiter = workqueue.DefaultTypedControllerRateLimiter[request]()
|
||||
}
|
||||
}
|
||||
|
||||
if options.NewQueue == nil {
|
||||
options.NewQueue = func(controllerName string, rateLimiter workqueue.TypedRateLimiter[request]) workqueue.TypedRateLimitingInterface[request] {
|
||||
if ptr.Deref(options.UsePriorityQueue, true) {
|
||||
return priorityqueue.New(controllerName, func(o *priorityqueue.Opts[request]) {
|
||||
o.Log = options.Logger.WithValues("controller", controllerName)
|
||||
o.RateLimiter = rateLimiter
|
||||
})
|
||||
}
|
||||
return workqueue.NewTypedRateLimitingQueueWithConfig(rateLimiter, workqueue.TypedRateLimitingQueueConfig[request]{
|
||||
Name: controllerName,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Create controller with dependencies set
|
||||
return controller.New[request](controller.Options[request]{
|
||||
Do: options.Reconciler,
|
||||
RateLimiter: options.RateLimiter,
|
||||
NewQueue: options.NewQueue,
|
||||
MaxConcurrentReconciles: options.MaxConcurrentReconciles,
|
||||
CacheSyncTimeout: options.CacheSyncTimeout,
|
||||
Name: name,
|
||||
LogConstructor: options.LogConstructor,
|
||||
RecoverPanic: options.RecoverPanic,
|
||||
LeaderElected: options.NeedLeaderElection,
|
||||
EnableWarmup: options.EnableWarmup,
|
||||
ReconciliationTimeout: options.ReconciliationTimeout,
|
||||
}), nil
|
||||
}
|
||||
|
||||
// ReconcileIDFromContext gets the reconcileID from the current context.
|
||||
var ReconcileIDFromContext = controller.ReconcileIDFromContext
|
||||
Generated
Vendored
+534
@@ -0,0 +1,534 @@
|
||||
/*
|
||||
Copyright 2018 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package controllerutil
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"slices"
|
||||
|
||||
"k8s.io/apimachinery/pkg/api/equality"
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/utils/ptr"
|
||||
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
|
||||
)
|
||||
|
||||
// AlreadyOwnedError is an error returned if the object you are trying to assign
|
||||
// a controller reference is already owned by another controller Object is the
|
||||
// subject and Owner is the reference for the current owner.
|
||||
type AlreadyOwnedError struct {
|
||||
Object metav1.Object
|
||||
Owner metav1.OwnerReference
|
||||
}
|
||||
|
||||
func (e *AlreadyOwnedError) Error() string {
|
||||
return fmt.Sprintf("Object %s/%s is already owned by another %s controller %s", e.Object.GetNamespace(), e.Object.GetName(), e.Owner.Kind, e.Owner.Name)
|
||||
}
|
||||
|
||||
func newAlreadyOwnedError(obj metav1.Object, owner metav1.OwnerReference) *AlreadyOwnedError {
|
||||
return &AlreadyOwnedError{
|
||||
Object: obj,
|
||||
Owner: owner,
|
||||
}
|
||||
}
|
||||
|
||||
// OwnerReferenceOption is a function that can modify a `metav1.OwnerReference`.
|
||||
type OwnerReferenceOption func(*metav1.OwnerReference)
|
||||
|
||||
// WithBlockOwnerDeletion allows configuring the BlockOwnerDeletion field on the `metav1.OwnerReference`.
|
||||
func WithBlockOwnerDeletion(blockOwnerDeletion bool) OwnerReferenceOption {
|
||||
return func(ref *metav1.OwnerReference) {
|
||||
ref.BlockOwnerDeletion = &blockOwnerDeletion
|
||||
}
|
||||
}
|
||||
|
||||
// SetControllerReference sets owner as a Controller OwnerReference on controlled.
|
||||
// This is used for garbage collection of the controlled object and for
|
||||
// reconciling the owner object on changes to controlled (with a Watch + EnqueueRequestForOwner).
|
||||
// Since only one OwnerReference can be a controller, it returns an error if
|
||||
// there is another OwnerReference with Controller flag set.
|
||||
func SetControllerReference(owner, controlled metav1.Object, scheme *runtime.Scheme, opts ...OwnerReferenceOption) error {
|
||||
// Validate the owner.
|
||||
ro, ok := owner.(runtime.Object)
|
||||
if !ok {
|
||||
return fmt.Errorf("%T is not a runtime.Object, cannot call SetControllerReference", owner)
|
||||
}
|
||||
if err := validateOwner(owner, controlled); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Create a new controller ref.
|
||||
gvk, err := apiutil.GVKForObject(ro, scheme)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ref := metav1.OwnerReference{
|
||||
APIVersion: gvk.GroupVersion().String(),
|
||||
Kind: gvk.Kind,
|
||||
Name: owner.GetName(),
|
||||
UID: owner.GetUID(),
|
||||
BlockOwnerDeletion: ptr.To(true),
|
||||
Controller: ptr.To(true),
|
||||
}
|
||||
for _, opt := range opts {
|
||||
opt(&ref)
|
||||
}
|
||||
|
||||
// Return early with an error if the object is already controlled.
|
||||
if existing := metav1.GetControllerOf(controlled); existing != nil && !referSameObject(*existing, ref) {
|
||||
return newAlreadyOwnedError(controlled, *existing)
|
||||
}
|
||||
|
||||
// Update owner references and return.
|
||||
upsertOwnerRef(ref, controlled)
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetOwnerReference is a helper method to make sure the given object contains an object reference to the object provided.
|
||||
// This allows you to declare that owner has a dependency on the object without specifying it as a controller.
|
||||
// If a reference to the same object already exists, it'll be overwritten with the newly provided version.
|
||||
func SetOwnerReference(owner, object metav1.Object, scheme *runtime.Scheme, opts ...OwnerReferenceOption) error {
|
||||
// Validate the owner.
|
||||
ro, ok := owner.(runtime.Object)
|
||||
if !ok {
|
||||
return fmt.Errorf("%T is not a runtime.Object, cannot call SetOwnerReference", owner)
|
||||
}
|
||||
if err := validateOwner(owner, object); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Create a new owner ref.
|
||||
gvk, err := apiutil.GVKForObject(ro, scheme)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ref := metav1.OwnerReference{
|
||||
APIVersion: gvk.GroupVersion().String(),
|
||||
Kind: gvk.Kind,
|
||||
UID: owner.GetUID(),
|
||||
Name: owner.GetName(),
|
||||
}
|
||||
for _, opt := range opts {
|
||||
opt(&ref)
|
||||
}
|
||||
|
||||
// Update owner references and return.
|
||||
upsertOwnerRef(ref, object)
|
||||
return nil
|
||||
}
|
||||
|
||||
// RemoveOwnerReference is a helper method to make sure the given object removes an owner reference to the object provided.
|
||||
// This allows you to remove the owner to establish a new owner of the object in a subsequent call.
|
||||
func RemoveOwnerReference(owner, object metav1.Object, scheme *runtime.Scheme) error {
|
||||
owners := object.GetOwnerReferences()
|
||||
length := len(owners)
|
||||
if length < 1 {
|
||||
return fmt.Errorf("%T does not have any owner references", object)
|
||||
}
|
||||
ro, ok := owner.(runtime.Object)
|
||||
if !ok {
|
||||
return fmt.Errorf("%T is not a runtime.Object, cannot call RemoveOwnerReference", owner)
|
||||
}
|
||||
gvk, err := apiutil.GVKForObject(ro, scheme)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
index := indexOwnerRef(owners, metav1.OwnerReference{
|
||||
APIVersion: gvk.GroupVersion().String(),
|
||||
Name: owner.GetName(),
|
||||
Kind: gvk.Kind,
|
||||
})
|
||||
if index == -1 {
|
||||
return fmt.Errorf("%T does not have an owner reference for %T", object, owner)
|
||||
}
|
||||
|
||||
owners = append(owners[:index], owners[index+1:]...)
|
||||
object.SetOwnerReferences(owners)
|
||||
return nil
|
||||
}
|
||||
|
||||
// HasControllerReference returns true if the object
|
||||
// has an owner ref with controller equal to true
|
||||
func HasControllerReference(object metav1.Object) bool {
|
||||
owners := object.GetOwnerReferences()
|
||||
for _, owner := range owners {
|
||||
isTrue := owner.Controller
|
||||
if owner.Controller != nil && *isTrue {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// HasOwnerReference returns true if the owners list contains an owner reference
|
||||
// that matches the object's group, kind, and name.
|
||||
func HasOwnerReference(ownerRefs []metav1.OwnerReference, obj client.Object, scheme *runtime.Scheme) (bool, error) {
|
||||
gvk, err := apiutil.GVKForObject(obj, scheme)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
idx := indexOwnerRef(ownerRefs, metav1.OwnerReference{
|
||||
APIVersion: gvk.GroupVersion().String(),
|
||||
Name: obj.GetName(),
|
||||
Kind: gvk.Kind,
|
||||
})
|
||||
return idx != -1, nil
|
||||
}
|
||||
|
||||
// RemoveControllerReference removes an owner reference where the controller
|
||||
// equals true
|
||||
func RemoveControllerReference(owner, object metav1.Object, scheme *runtime.Scheme) error {
|
||||
if ok := HasControllerReference(object); !ok {
|
||||
return fmt.Errorf("%T does not have a owner reference with controller equals true", object)
|
||||
}
|
||||
ro, ok := owner.(runtime.Object)
|
||||
if !ok {
|
||||
return fmt.Errorf("%T is not a runtime.Object, cannot call RemoveControllerReference", owner)
|
||||
}
|
||||
gvk, err := apiutil.GVKForObject(ro, scheme)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ownerRefs := object.GetOwnerReferences()
|
||||
index := indexOwnerRef(ownerRefs, metav1.OwnerReference{
|
||||
APIVersion: gvk.GroupVersion().String(),
|
||||
Name: owner.GetName(),
|
||||
Kind: gvk.Kind,
|
||||
})
|
||||
|
||||
if index == -1 {
|
||||
return fmt.Errorf("%T does not have an controller reference for %T", object, owner)
|
||||
}
|
||||
|
||||
if ownerRefs[index].Controller == nil || !*ownerRefs[index].Controller {
|
||||
return fmt.Errorf("%T owner is not the controller reference for %T", owner, object)
|
||||
}
|
||||
|
||||
ownerRefs = append(ownerRefs[:index], ownerRefs[index+1:]...)
|
||||
object.SetOwnerReferences(ownerRefs)
|
||||
return nil
|
||||
}
|
||||
|
||||
func upsertOwnerRef(ref metav1.OwnerReference, object metav1.Object) {
|
||||
owners := object.GetOwnerReferences()
|
||||
if idx := indexOwnerRef(owners, ref); idx == -1 {
|
||||
owners = append(owners, ref)
|
||||
} else {
|
||||
owners[idx] = ref
|
||||
}
|
||||
object.SetOwnerReferences(owners)
|
||||
}
|
||||
|
||||
// indexOwnerRef returns the index of the owner reference in the slice if found, or -1.
|
||||
func indexOwnerRef(ownerReferences []metav1.OwnerReference, ref metav1.OwnerReference) int {
|
||||
for index, r := range ownerReferences {
|
||||
if referSameObject(r, ref) {
|
||||
return index
|
||||
}
|
||||
}
|
||||
return -1
|
||||
}
|
||||
|
||||
func validateOwner(owner, object metav1.Object) error {
|
||||
ownerNs := owner.GetNamespace()
|
||||
if ownerNs != "" {
|
||||
objNs := object.GetNamespace()
|
||||
if objNs == "" {
|
||||
return fmt.Errorf("cluster-scoped resource must not have a namespace-scoped owner, owner's namespace %s", ownerNs)
|
||||
}
|
||||
if ownerNs != objNs {
|
||||
return fmt.Errorf("cross-namespace owner references are disallowed, owner's namespace %s, obj's namespace %s", owner.GetNamespace(), object.GetNamespace())
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Returns true if a and b point to the same object.
|
||||
func referSameObject(a, b metav1.OwnerReference) bool {
|
||||
aGV, err := schema.ParseGroupVersion(a.APIVersion)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
|
||||
bGV, err := schema.ParseGroupVersion(b.APIVersion)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return aGV.Group == bGV.Group && a.Kind == b.Kind && a.Name == b.Name
|
||||
}
|
||||
|
||||
// OperationResult is the action result of a CreateOrUpdate or CreateOrPatch call.
|
||||
type OperationResult string
|
||||
|
||||
const ( // They should complete the sentence "Deployment default/foo has been ..."
|
||||
// OperationResultNone means that the resource has not been changed.
|
||||
OperationResultNone OperationResult = "unchanged"
|
||||
// OperationResultCreated means that a new resource is created.
|
||||
OperationResultCreated OperationResult = "created"
|
||||
// OperationResultUpdated means that an existing resource is updated.
|
||||
OperationResultUpdated OperationResult = "updated"
|
||||
// OperationResultUpdatedStatus means that an existing resource and its status is updated.
|
||||
OperationResultUpdatedStatus OperationResult = "updatedStatus"
|
||||
// OperationResultUpdatedStatusOnly means that only an existing status is updated.
|
||||
OperationResultUpdatedStatusOnly OperationResult = "updatedStatusOnly"
|
||||
)
|
||||
|
||||
// CreateOrUpdate attempts to fetch the given object from the Kubernetes cluster.
|
||||
// If the object didn't exist, MutateFn will be called, and it will be created.
|
||||
// If the object did exist, MutateFn will be called, and if it changed the
|
||||
// object, it will be updated.
|
||||
// Otherwise, it will be left unchanged.
|
||||
// The executed operation (and an error) will be returned.
|
||||
//
|
||||
// WARNING: If the MutateFn resets a value on obj that has a default value,
|
||||
// CreateOrUpdate will *always* perform an update. This is because when the
|
||||
// object is fetched from the API server, the value will have taken on the
|
||||
// default value, and the check for equality will fail. For example, Deployments
|
||||
// must have a Replicas value set. If the MutateFn sets a Deployment's Replicas
|
||||
// to nil, then it will never match with the object returned from the API
|
||||
// server, which defaults the value to 1.
|
||||
//
|
||||
// WARNING: CreateOrUpdate assumes that no values have been set on obj aside
|
||||
// from the Name/Namespace. Values other than Name and Namespace that existed on
|
||||
// obj may be overwritten by the corresponding values in the object returned
|
||||
// from the Kubernetes API server. When this happens, the Update will not work
|
||||
// as expected.
|
||||
//
|
||||
// Note: changes made by MutateFn to any sub-resource (status...), will be
|
||||
// discarded.
|
||||
func CreateOrUpdate(ctx context.Context, c client.Client, obj client.Object, f MutateFn) (OperationResult, error) {
|
||||
key := client.ObjectKeyFromObject(obj)
|
||||
if err := c.Get(ctx, key, obj); err != nil {
|
||||
if !apierrors.IsNotFound(err) {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
if f != nil {
|
||||
if err := mutate(f, key, obj); err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
}
|
||||
|
||||
if err := c.Create(ctx, obj); err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
return OperationResultCreated, nil
|
||||
}
|
||||
|
||||
existing := obj.DeepCopyObject()
|
||||
if f != nil {
|
||||
if err := mutate(f, key, obj); err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
}
|
||||
|
||||
if equality.Semantic.DeepEqual(existing, obj) {
|
||||
return OperationResultNone, nil
|
||||
}
|
||||
|
||||
if err := c.Update(ctx, obj); err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
return OperationResultUpdated, nil
|
||||
}
|
||||
|
||||
// CreateOrPatch attempts to fetch the given object from the Kubernetes cluster.
|
||||
// If the object didn't exist, MutateFn will be called, and it will be created.
|
||||
// If the object did exist, MutateFn will be called, and if it changed the
|
||||
// object, it will be patched.
|
||||
// Otherwise, it will be left unchanged.
|
||||
// The executed operation (and an error) will be returned.
|
||||
//
|
||||
// WARNING: If the MutateFn resets a value on obj that has a default value,
|
||||
// CreateOrPatch will *always* perform a patch. This is because when the
|
||||
// object is fetched from the API server, the value will have taken on the
|
||||
// default value, and the check for equality will fail.
|
||||
// For example, Deployments must have a Replicas value set. If the MutateFn sets
|
||||
// a Deployment's Replicas to nil, then it will never match with the object
|
||||
// returned from the API server, which defaults the value to 1.
|
||||
//
|
||||
// WARNING: CreateOrPatch assumes that no values have been set on obj aside
|
||||
// from the Name/Namespace. Values other than Name and Namespace that existed on
|
||||
// obj may be overwritten by the corresponding values in the object returned
|
||||
// from the Kubernetes API server. When this happens, the Patch will not work
|
||||
// as expected.
|
||||
//
|
||||
// Note: changes to any sub-resource other than status will be ignored.
|
||||
// Changes to the status sub-resource will only be applied if the object
|
||||
// already exist. To change the status on object creation, the easiest
|
||||
// way is to requeue the object in the controller if OperationResult is
|
||||
// OperationResultCreated
|
||||
func CreateOrPatch(ctx context.Context, c client.Client, obj client.Object, f MutateFn) (OperationResult, error) {
|
||||
key := client.ObjectKeyFromObject(obj)
|
||||
if err := c.Get(ctx, key, obj); err != nil {
|
||||
if !apierrors.IsNotFound(err) {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
if f != nil {
|
||||
if err := mutate(f, key, obj); err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
}
|
||||
if err := c.Create(ctx, obj); err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
return OperationResultCreated, nil
|
||||
}
|
||||
|
||||
// Create patches for the object and its possible status.
|
||||
objPatch := client.MergeFrom(obj.DeepCopyObject().(client.Object))
|
||||
statusPatch := client.MergeFrom(obj.DeepCopyObject().(client.Object))
|
||||
|
||||
// Create a copy of the original object as well as converting that copy to
|
||||
// unstructured data.
|
||||
before, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj.DeepCopyObject())
|
||||
if err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
|
||||
// Attempt to extract the status from the resource for easier comparison later
|
||||
beforeStatus, hasBeforeStatus, err := unstructured.NestedFieldCopy(before, "status")
|
||||
if err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
|
||||
// If the resource contains a status then remove it from the unstructured
|
||||
// copy to avoid unnecessary patching later.
|
||||
if hasBeforeStatus {
|
||||
unstructured.RemoveNestedField(before, "status")
|
||||
}
|
||||
|
||||
// Mutate the original object.
|
||||
if f != nil {
|
||||
if err := mutate(f, key, obj); err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
}
|
||||
|
||||
// Convert the resource to unstructured to compare against our before copy.
|
||||
after, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
|
||||
if err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
|
||||
// Attempt to extract the status from the resource for easier comparison later
|
||||
afterStatus, hasAfterStatus, err := unstructured.NestedFieldCopy(after, "status")
|
||||
if err != nil {
|
||||
return OperationResultNone, err
|
||||
}
|
||||
|
||||
// If the resource contains a status then remove it from the unstructured
|
||||
// copy to avoid unnecessary patching later.
|
||||
if hasAfterStatus {
|
||||
unstructured.RemoveNestedField(after, "status")
|
||||
}
|
||||
|
||||
result := OperationResultNone
|
||||
|
||||
if !reflect.DeepEqual(before, after) {
|
||||
// Only issue a Patch if the before and after resources (minus status) differ
|
||||
if err := c.Patch(ctx, obj, objPatch); err != nil {
|
||||
return result, err
|
||||
}
|
||||
result = OperationResultUpdated
|
||||
}
|
||||
|
||||
if (hasBeforeStatus || hasAfterStatus) && !reflect.DeepEqual(beforeStatus, afterStatus) {
|
||||
// Only issue a Status Patch if the resource has a status and the beforeStatus
|
||||
// and afterStatus copies differ
|
||||
if result == OperationResultUpdated {
|
||||
// If Status was replaced by Patch before, set it to afterStatus
|
||||
objectAfterPatch, err := runtime.DefaultUnstructuredConverter.ToUnstructured(obj)
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
if err = unstructured.SetNestedField(objectAfterPatch, afterStatus, "status"); err != nil {
|
||||
return result, err
|
||||
}
|
||||
// If Status was replaced by Patch before, restore patched structure to the obj
|
||||
if err = runtime.DefaultUnstructuredConverter.FromUnstructured(objectAfterPatch, obj); err != nil {
|
||||
return result, err
|
||||
}
|
||||
}
|
||||
if err := c.Status().Patch(ctx, obj, statusPatch); err != nil {
|
||||
return result, err
|
||||
}
|
||||
if result == OperationResultUpdated {
|
||||
result = OperationResultUpdatedStatus
|
||||
} else {
|
||||
result = OperationResultUpdatedStatusOnly
|
||||
}
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// mutate wraps a MutateFn and applies validation to its result.
|
||||
func mutate(f MutateFn, key client.ObjectKey, obj client.Object) error {
|
||||
if err := f(); err != nil {
|
||||
return err
|
||||
}
|
||||
if newKey := client.ObjectKeyFromObject(obj); key != newKey {
|
||||
return fmt.Errorf("MutateFn cannot mutate object name and/or object namespace")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// MutateFn is a function which mutates the existing object into its desired state.
|
||||
type MutateFn func() error
|
||||
|
||||
// AddFinalizer accepts an Object and adds the provided finalizer if not present.
|
||||
// It returns an indication of whether it updated the object's list of finalizers.
|
||||
func AddFinalizer(o client.Object, finalizer string) (finalizersUpdated bool) {
|
||||
f := o.GetFinalizers()
|
||||
if slices.Contains(f, finalizer) {
|
||||
return false
|
||||
}
|
||||
o.SetFinalizers(append(f, finalizer))
|
||||
return true
|
||||
}
|
||||
|
||||
// RemoveFinalizer accepts an Object and removes the provided finalizer if present.
|
||||
// It returns an indication of whether it updated the object's list of finalizers.
|
||||
func RemoveFinalizer(o client.Object, finalizer string) (finalizersUpdated bool) {
|
||||
f := o.GetFinalizers()
|
||||
length := len(f)
|
||||
|
||||
index := 0
|
||||
for i := range length {
|
||||
if f[i] == finalizer {
|
||||
continue
|
||||
}
|
||||
f[index] = f[i]
|
||||
index++
|
||||
}
|
||||
o.SetFinalizers(f[:index])
|
||||
return length != index
|
||||
}
|
||||
|
||||
// ContainsFinalizer checks an Object that the provided finalizer is present.
|
||||
func ContainsFinalizer(o client.Object, finalizer string) bool {
|
||||
f := o.GetFinalizers()
|
||||
return slices.Contains(f, finalizer)
|
||||
}
|
||||
+20
@@ -0,0 +1,20 @@
|
||||
/*
|
||||
Copyright 2018 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
/*
|
||||
Package controllerutil contains utility functions for working with and implementing Controllers.
|
||||
*/
|
||||
package controllerutil
|
||||
+25
@@ -0,0 +1,25 @@
|
||||
/*
|
||||
Copyright 2018 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
/*
|
||||
Package controller provides types and functions for building Controllers. Controllers implement Kubernetes APIs.
|
||||
|
||||
# Creation
|
||||
|
||||
To create a new Controller, first create a manager.Manager and pass it to the controller.New function.
|
||||
The Controller MUST be started by calling Manager.Start.
|
||||
*/
|
||||
package controller
|
||||
+43
@@ -0,0 +1,43 @@
|
||||
/*
|
||||
Copyright 2020 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package controller
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
)
|
||||
|
||||
var nameLock sync.Mutex
|
||||
var usedNames sets.Set[string]
|
||||
|
||||
func checkName(name string) error {
|
||||
nameLock.Lock()
|
||||
defer nameLock.Unlock()
|
||||
if usedNames == nil {
|
||||
usedNames = sets.Set[string]{}
|
||||
}
|
||||
|
||||
if usedNames.Has(name) {
|
||||
return fmt.Errorf("controller with name %s already exists. Controller names must be unique to avoid multiple controllers reporting the same metric. This validation can be disabled via the SkipNameValidation option", name)
|
||||
}
|
||||
|
||||
usedNames.Insert(name)
|
||||
|
||||
return nil
|
||||
}
|
||||
+172
@@ -0,0 +1,172 @@
|
||||
package priorityqueue
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
"k8s.io/utils/clock"
|
||||
"sigs.k8s.io/controller-runtime/pkg/internal/metrics"
|
||||
)
|
||||
|
||||
// This file is mostly a copy of unexported code from
|
||||
// https://github.com/kubernetes/kubernetes/blob/1d8828ce707ed9dd7a6a9756385419cce1d202ac/staging/src/k8s.io/client-go/util/workqueue/metrics.go
|
||||
//
|
||||
// The only two differences are the addition of mapLock in defaultQueueMetrics and converging retryMetrics into queueMetrics.
|
||||
|
||||
type queueMetrics[T comparable] interface {
|
||||
add(item T, priority int)
|
||||
get(item T, priority int)
|
||||
updateDepthWithPriorityMetric(oldPriority, newPriority int)
|
||||
done(item T)
|
||||
updateUnfinishedWork()
|
||||
retry()
|
||||
}
|
||||
|
||||
func newQueueMetrics[T comparable](mp workqueue.MetricsProvider, name string, clock clock.Clock) queueMetrics[T] {
|
||||
if len(name) == 0 {
|
||||
return noMetrics[T]{}
|
||||
}
|
||||
|
||||
dqm := &defaultQueueMetrics[T]{
|
||||
clock: clock,
|
||||
adds: mp.NewAddsMetric(name),
|
||||
latency: mp.NewLatencyMetric(name),
|
||||
workDuration: mp.NewWorkDurationMetric(name),
|
||||
unfinishedWorkSeconds: mp.NewUnfinishedWorkSecondsMetric(name),
|
||||
longestRunningProcessor: mp.NewLongestRunningProcessorSecondsMetric(name),
|
||||
addTimes: map[T]time.Time{},
|
||||
processingStartTimes: map[T]time.Time{},
|
||||
retries: mp.NewRetriesMetric(name),
|
||||
}
|
||||
|
||||
if mpp, ok := mp.(metrics.MetricsProviderWithPriority); ok {
|
||||
dqm.depthWithPriority = mpp.NewDepthMetricWithPriority(name)
|
||||
} else {
|
||||
dqm.depth = mp.NewDepthMetric(name)
|
||||
}
|
||||
return dqm
|
||||
}
|
||||
|
||||
// defaultQueueMetrics expects the caller to lock before setting any metrics.
|
||||
type defaultQueueMetrics[T comparable] struct {
|
||||
clock clock.Clock
|
||||
|
||||
// current depth of a workqueue
|
||||
depth workqueue.GaugeMetric
|
||||
depthWithPriority metrics.DepthMetricWithPriority
|
||||
// total number of adds handled by a workqueue
|
||||
adds workqueue.CounterMetric
|
||||
// how long an item stays in a workqueue
|
||||
latency workqueue.HistogramMetric
|
||||
// how long processing an item from a workqueue takes
|
||||
workDuration workqueue.HistogramMetric
|
||||
|
||||
mapLock sync.RWMutex
|
||||
addTimes map[T]time.Time
|
||||
processingStartTimes map[T]time.Time
|
||||
|
||||
// how long have current threads been working?
|
||||
unfinishedWorkSeconds workqueue.SettableGaugeMetric
|
||||
longestRunningProcessor workqueue.SettableGaugeMetric
|
||||
|
||||
retries workqueue.CounterMetric
|
||||
}
|
||||
|
||||
// add is called for ready items only
|
||||
func (m *defaultQueueMetrics[T]) add(item T, priority int) {
|
||||
if m == nil {
|
||||
return
|
||||
}
|
||||
|
||||
m.adds.Inc()
|
||||
if m.depthWithPriority != nil {
|
||||
m.depthWithPriority.Inc(priority)
|
||||
} else {
|
||||
m.depth.Inc()
|
||||
}
|
||||
|
||||
m.mapLock.Lock()
|
||||
defer m.mapLock.Unlock()
|
||||
|
||||
if _, exists := m.addTimes[item]; !exists {
|
||||
m.addTimes[item] = m.clock.Now()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *defaultQueueMetrics[T]) get(item T, priority int) {
|
||||
if m == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if m.depthWithPriority != nil {
|
||||
m.depthWithPriority.Dec(priority)
|
||||
} else {
|
||||
m.depth.Dec()
|
||||
}
|
||||
|
||||
m.mapLock.Lock()
|
||||
defer m.mapLock.Unlock()
|
||||
|
||||
m.processingStartTimes[item] = m.clock.Now()
|
||||
if startTime, exists := m.addTimes[item]; exists {
|
||||
m.latency.Observe(m.sinceInSeconds(startTime))
|
||||
delete(m.addTimes, item)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *defaultQueueMetrics[T]) updateDepthWithPriorityMetric(oldPriority, newPriority int) {
|
||||
if m.depthWithPriority != nil {
|
||||
m.depthWithPriority.Dec(oldPriority)
|
||||
m.depthWithPriority.Inc(newPriority)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *defaultQueueMetrics[T]) done(item T) {
|
||||
if m == nil {
|
||||
return
|
||||
}
|
||||
|
||||
m.mapLock.Lock()
|
||||
defer m.mapLock.Unlock()
|
||||
if startTime, exists := m.processingStartTimes[item]; exists {
|
||||
m.workDuration.Observe(m.sinceInSeconds(startTime))
|
||||
delete(m.processingStartTimes, item)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *defaultQueueMetrics[T]) updateUnfinishedWork() {
|
||||
m.mapLock.RLock()
|
||||
defer m.mapLock.RUnlock()
|
||||
// Note that a summary metric would be better for this, but prometheus
|
||||
// doesn't seem to have non-hacky ways to reset the summary metrics.
|
||||
var total float64
|
||||
var oldest float64
|
||||
for _, t := range m.processingStartTimes {
|
||||
age := m.sinceInSeconds(t)
|
||||
total += age
|
||||
if age > oldest {
|
||||
oldest = age
|
||||
}
|
||||
}
|
||||
m.unfinishedWorkSeconds.Set(total)
|
||||
m.longestRunningProcessor.Set(oldest)
|
||||
}
|
||||
|
||||
// Gets the time since the specified start in seconds.
|
||||
func (m *defaultQueueMetrics[T]) sinceInSeconds(start time.Time) float64 {
|
||||
return m.clock.Since(start).Seconds()
|
||||
}
|
||||
|
||||
func (m *defaultQueueMetrics[T]) retry() {
|
||||
m.retries.Inc()
|
||||
}
|
||||
|
||||
type noMetrics[T any] struct{}
|
||||
|
||||
func (noMetrics[T]) add(item T, priority int) {}
|
||||
func (noMetrics[T]) get(item T, priority int) {}
|
||||
func (noMetrics[T]) updateDepthWithPriorityMetric(oldPriority, newPriority int) {}
|
||||
func (noMetrics[T]) done(item T) {}
|
||||
func (noMetrics[T]) updateUnfinishedWork() {}
|
||||
func (noMetrics[T]) retry() {}
|
||||
Generated
Vendored
+569
@@ -0,0 +1,569 @@
|
||||
package priorityqueue
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/go-logr/logr"
|
||||
"github.com/google/btree"
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
"k8s.io/client-go/util/workqueue"
|
||||
"k8s.io/utils/clock"
|
||||
"k8s.io/utils/ptr"
|
||||
|
||||
"sigs.k8s.io/controller-runtime/pkg/internal/metrics"
|
||||
)
|
||||
|
||||
// AddOpts describes the options for adding items to the queue.
|
||||
type AddOpts struct {
|
||||
After time.Duration
|
||||
RateLimited bool
|
||||
// Priority is the priority of the item. Higher values
|
||||
// indicate higher priority.
|
||||
// Defaults to zero if unset.
|
||||
Priority *int
|
||||
}
|
||||
|
||||
// PriorityQueue is a priority queue for a controller. It
|
||||
// internally de-duplicates all items that are added to
|
||||
// it. It will use the max of the passed priorities and the
|
||||
// min of possible durations.
|
||||
//
|
||||
// When an item that is already enqueued at a lower priority
|
||||
// is re-enqueued with a higher priority, it will be placed at
|
||||
// the end among items of the new priority, in order to
|
||||
// preserve FIFO semantics within each priority level.
|
||||
// The effective duration (i.e. the ready time) is still
|
||||
// computed as the minimum across all enqueues.
|
||||
type PriorityQueue[T comparable] interface {
|
||||
workqueue.TypedRateLimitingInterface[T]
|
||||
AddWithOpts(o AddOpts, Items ...T)
|
||||
GetWithPriority() (item T, priority int, shutdown bool)
|
||||
}
|
||||
|
||||
// Opts contains the options for a PriorityQueue.
|
||||
type Opts[T comparable] struct {
|
||||
// Ratelimiter is being used when AddRateLimited is called. Defaults to a per-item exponential backoff
|
||||
// limiter with an initial delay of five milliseconds and a max delay of 1000 seconds.
|
||||
RateLimiter workqueue.TypedRateLimiter[T]
|
||||
MetricProvider workqueue.MetricsProvider
|
||||
Log logr.Logger
|
||||
}
|
||||
|
||||
// Opt allows to configure a PriorityQueue.
|
||||
type Opt[T comparable] func(*Opts[T])
|
||||
|
||||
type bufferItem[T comparable] struct {
|
||||
opts AddOpts
|
||||
items []T
|
||||
}
|
||||
|
||||
// New constructs a new PriorityQueue.
|
||||
func New[T comparable](name string, o ...Opt[T]) PriorityQueue[T] {
|
||||
opts := &Opts[T]{}
|
||||
for _, f := range o {
|
||||
f(opts)
|
||||
}
|
||||
|
||||
if opts.RateLimiter == nil {
|
||||
opts.RateLimiter = workqueue.NewTypedItemExponentialFailureRateLimiter[T](5*time.Millisecond, 1000*time.Second)
|
||||
}
|
||||
|
||||
if opts.MetricProvider == nil {
|
||||
opts.MetricProvider = metrics.WorkqueueMetricsProvider{}
|
||||
}
|
||||
|
||||
pq := &priorityqueue[T]{
|
||||
log: opts.Log,
|
||||
itemAddedToAddBuffer: make(chan struct{}, 1),
|
||||
items: map[T]*item[T]{},
|
||||
ready: btree.NewG(32, lessReady[T]),
|
||||
waiting: btree.NewG(32, lessWaiting[T]),
|
||||
metrics: newQueueMetrics[T](opts.MetricProvider, name, clock.RealClock{}),
|
||||
// readyItemOrWaiterAdded indicates that a ready item or
|
||||
// waiter was added. It must be buffered, because
|
||||
// if we currently process items we can't tell
|
||||
// if that included the new item/waiter.
|
||||
readyItemOrWaiterAdded: make(chan struct{}, 1),
|
||||
waitingItemAddedOrUpdated: make(chan struct{}, 1),
|
||||
rateLimiter: opts.RateLimiter,
|
||||
locked: sets.Set[T]{},
|
||||
done: make(chan struct{}),
|
||||
get: make(chan item[T]),
|
||||
now: time.Now,
|
||||
tick: time.Tick,
|
||||
}
|
||||
|
||||
go pq.handleAddBuffer()
|
||||
go pq.handleReadyItems()
|
||||
go pq.handleWaitingItems()
|
||||
go pq.logState()
|
||||
if _, ok := pq.metrics.(noMetrics[T]); !ok {
|
||||
go pq.updateUnfinishedWorkLoop()
|
||||
}
|
||||
|
||||
return pq
|
||||
}
|
||||
|
||||
type priorityqueue[T comparable] struct {
|
||||
log logr.Logger
|
||||
|
||||
addBufferLock sync.Mutex
|
||||
addBuffer []bufferItem[T]
|
||||
itemAddedToAddBuffer chan struct{}
|
||||
|
||||
// lock has to be acquired for any access to any of items, ready, waiting,
|
||||
// addedCounter or waiters.
|
||||
lock sync.Mutex
|
||||
items map[T]*item[T]
|
||||
ready bTree[*item[T]]
|
||||
waiting bTree[*item[T]]
|
||||
|
||||
// addedCounter is a counter of elements added, we need it
|
||||
// to provide FIFO semantics.
|
||||
addedCounter uint64
|
||||
|
||||
metrics queueMetrics[T]
|
||||
|
||||
readyItemOrWaiterAdded chan struct{}
|
||||
waitingItemAddedOrUpdated chan struct{}
|
||||
|
||||
rateLimiter workqueue.TypedRateLimiter[T]
|
||||
|
||||
// locked contains the keys we handed out through Get() and that haven't
|
||||
// yet been returned through Done().
|
||||
locked sets.Set[T]
|
||||
lockedLock sync.Mutex
|
||||
|
||||
shutdown atomic.Bool
|
||||
done chan struct{}
|
||||
|
||||
get chan item[T]
|
||||
|
||||
// waiters is the number of routines blocked in Get, we use it to determine
|
||||
// if we can push items. Every manipulation has to be protected with the lock.
|
||||
waiters int64
|
||||
|
||||
// Configurable for testing
|
||||
now func() time.Time
|
||||
tick func(time.Duration) <-chan time.Time
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) AddWithOpts(o AddOpts, items ...T) {
|
||||
if w.shutdown.Load() {
|
||||
return
|
||||
}
|
||||
|
||||
if len(items) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
w.addBufferLock.Lock()
|
||||
w.addBuffer = append(w.addBuffer, bufferItem[T]{
|
||||
opts: o,
|
||||
items: items,
|
||||
})
|
||||
w.addBufferLock.Unlock()
|
||||
|
||||
w.notifyItemAddedToAddBuffer()
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) handleAddBuffer() {
|
||||
for {
|
||||
select {
|
||||
case <-w.done:
|
||||
return
|
||||
case <-w.itemAddedToAddBuffer:
|
||||
}
|
||||
|
||||
w.lock.Lock()
|
||||
w.lockedFlushAddBuffer()
|
||||
w.lock.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) lockedFlushAddBuffer() {
|
||||
w.addBufferLock.Lock()
|
||||
buffer := w.addBuffer
|
||||
w.addBuffer = make([]bufferItem[T], 0, len(buffer))
|
||||
w.addBufferLock.Unlock()
|
||||
|
||||
for _, v := range buffer {
|
||||
w.lockedAddWithOpts(v.opts, v.items...)
|
||||
}
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) lockedAddWithOpts(o AddOpts, items ...T) {
|
||||
if w.shutdown.Load() {
|
||||
return
|
||||
}
|
||||
|
||||
var readyItemAdded bool
|
||||
var waitingItemAddedOrUpdated bool
|
||||
|
||||
for _, key := range items {
|
||||
after := o.After
|
||||
if o.RateLimited {
|
||||
rlAfter := w.rateLimiter.When(key)
|
||||
if after == 0 || rlAfter < after {
|
||||
after = rlAfter
|
||||
}
|
||||
}
|
||||
|
||||
var readyAt *time.Time
|
||||
if after > 0 {
|
||||
readyAt = ptr.To(w.now().Add(after))
|
||||
w.metrics.retry()
|
||||
}
|
||||
if _, ok := w.items[key]; !ok {
|
||||
item := &item[T]{
|
||||
Key: key,
|
||||
AddedCounter: w.addedCounter,
|
||||
Priority: ptr.Deref(o.Priority, 0),
|
||||
ReadyAt: readyAt,
|
||||
}
|
||||
w.addedCounter++
|
||||
w.items[key] = item
|
||||
if readyAt != nil {
|
||||
w.waiting.ReplaceOrInsert(item)
|
||||
waitingItemAddedOrUpdated = true
|
||||
} else {
|
||||
w.ready.ReplaceOrInsert(item)
|
||||
w.metrics.add(key, item.Priority)
|
||||
readyItemAdded = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
if w.items[key].ReadyAt == nil {
|
||||
readyAt = nil
|
||||
} else if readyAt != nil && w.items[key].ReadyAt.Before(*readyAt) {
|
||||
readyAt = w.items[key].ReadyAt
|
||||
}
|
||||
|
||||
priority := w.items[key].Priority
|
||||
addedCounter := w.items[key].AddedCounter
|
||||
if newPriority := ptr.Deref(o.Priority, 0); newPriority > w.items[key].Priority {
|
||||
// Update depth metric only if the item was already ready
|
||||
if w.items[key].ReadyAt == nil {
|
||||
w.metrics.updateDepthWithPriorityMetric(w.items[key].Priority, newPriority)
|
||||
}
|
||||
priority = newPriority
|
||||
addedCounter = w.addedCounter
|
||||
w.addedCounter++
|
||||
}
|
||||
|
||||
var tree, previousTree bTree[*item[T]]
|
||||
switch {
|
||||
case readyAt == nil && w.items[key].ReadyAt == nil:
|
||||
tree, previousTree = w.ready, w.ready
|
||||
case readyAt == nil && w.items[key].ReadyAt != nil:
|
||||
tree, previousTree = w.ready, w.waiting
|
||||
readyItemAdded = true
|
||||
w.metrics.add(key, priority)
|
||||
case readyAt != nil:
|
||||
// We are in the update path and we set readyAt to nil if the
|
||||
// existing item has a nil readyAt, so we can be sure here that
|
||||
// it has a non-nil readyAt/is in w.waiting.
|
||||
tree, previousTree = w.waiting, w.waiting
|
||||
waitingItemAddedOrUpdated = true
|
||||
}
|
||||
|
||||
item, _ := previousTree.Delete(w.items[key])
|
||||
item.ReadyAt = readyAt
|
||||
item.Priority = priority
|
||||
item.AddedCounter = addedCounter
|
||||
tree.ReplaceOrInsert(item)
|
||||
}
|
||||
|
||||
if readyItemAdded {
|
||||
w.notifyReadyItemOrWaiterAdded()
|
||||
}
|
||||
if waitingItemAddedOrUpdated {
|
||||
w.notifyWaitingItemAddedOrUpdated()
|
||||
}
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) notifyItemAddedToAddBuffer() {
|
||||
select {
|
||||
case w.itemAddedToAddBuffer <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) notifyReadyItemOrWaiterAdded() {
|
||||
select {
|
||||
case w.readyItemOrWaiterAdded <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) notifyWaitingItemAddedOrUpdated() {
|
||||
select {
|
||||
case w.waitingItemAddedOrUpdated <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) handleWaitingItems() {
|
||||
blockForever := make(chan time.Time)
|
||||
var nextReady <-chan time.Time
|
||||
nextReady = blockForever
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-w.done:
|
||||
return
|
||||
case <-w.waitingItemAddedOrUpdated:
|
||||
case <-nextReady:
|
||||
nextReady = blockForever
|
||||
}
|
||||
|
||||
func() {
|
||||
w.lock.Lock()
|
||||
defer w.lock.Unlock()
|
||||
|
||||
var toMove []*item[T]
|
||||
w.waiting.Ascend(func(item *item[T]) bool {
|
||||
readyIn := item.ReadyAt.Sub(w.now()) // Store this to prevent TOCTOU issues
|
||||
if readyIn <= 0 {
|
||||
toMove = append(toMove, item)
|
||||
return true
|
||||
}
|
||||
|
||||
nextReady = w.tick(readyIn)
|
||||
return false
|
||||
})
|
||||
|
||||
// Don't manipulate the tree from within Ascend
|
||||
for _, toMove := range toMove {
|
||||
w.waiting.Delete(toMove)
|
||||
toMove.ReadyAt = nil
|
||||
|
||||
// Bump added counter so items get sorted by when
|
||||
// they became ready, not when they were added.
|
||||
toMove.AddedCounter = w.addedCounter
|
||||
w.addedCounter++
|
||||
|
||||
w.metrics.add(toMove.Key, toMove.Priority)
|
||||
w.ready.ReplaceOrInsert(toMove)
|
||||
}
|
||||
|
||||
if len(toMove) > 0 {
|
||||
w.notifyReadyItemOrWaiterAdded()
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) handleReadyItems() {
|
||||
for {
|
||||
select {
|
||||
case <-w.done:
|
||||
return
|
||||
case <-w.readyItemOrWaiterAdded:
|
||||
}
|
||||
|
||||
func() {
|
||||
w.lock.Lock()
|
||||
defer w.lock.Unlock()
|
||||
|
||||
// Flush is performed before reading items to avoid errors caused by asynchronous behavior,
|
||||
// primarily for unit testing purposes.
|
||||
// Successfully adding a ready item may result in an additional call to handleReadyItems(),
|
||||
// but the cost is negligible.
|
||||
w.lockedFlushAddBuffer()
|
||||
|
||||
if w.waiters == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
w.lockedLock.Lock()
|
||||
defer w.lockedLock.Unlock()
|
||||
|
||||
// manipulating the tree from within Ascend might lead to panics, so
|
||||
// track what we want to delete and do it after we are done ascending.
|
||||
var toDelete []*item[T]
|
||||
|
||||
w.ready.Ascend(func(item *item[T]) bool {
|
||||
// Item is locked, we can not hand it out
|
||||
if w.locked.Has(item.Key) {
|
||||
return true
|
||||
}
|
||||
|
||||
w.metrics.get(item.Key, item.Priority)
|
||||
w.locked.Insert(item.Key)
|
||||
w.waiters--
|
||||
delete(w.items, item.Key)
|
||||
toDelete = append(toDelete, item)
|
||||
w.get <- *item
|
||||
|
||||
return w.waiters > 0
|
||||
})
|
||||
|
||||
for _, item := range toDelete {
|
||||
w.ready.Delete(item)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) Add(item T) {
|
||||
w.AddWithOpts(AddOpts{}, item)
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) AddAfter(item T, after time.Duration) {
|
||||
w.AddWithOpts(AddOpts{After: after}, item)
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) AddRateLimited(item T) {
|
||||
w.AddWithOpts(AddOpts{RateLimited: true}, item)
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) GetWithPriority() (_ T, priority int, shutdown bool) {
|
||||
if w.shutdown.Load() {
|
||||
var zero T
|
||||
return zero, 0, true
|
||||
}
|
||||
|
||||
w.lock.Lock()
|
||||
w.waiters++
|
||||
w.lock.Unlock()
|
||||
|
||||
w.notifyReadyItemOrWaiterAdded()
|
||||
|
||||
select {
|
||||
case <-w.done:
|
||||
// Return if the queue was shutdown while we were already waiting for an item here.
|
||||
// For example controller workers are continuously calling GetWithPriority and
|
||||
// GetWithPriority is blocking the workers if there are no items in the queue.
|
||||
// If the controller and accordingly the queue is then shut down, without this code
|
||||
// branch the controller workers remain blocked here and are unable to shut down.
|
||||
var zero T
|
||||
return zero, 0, true
|
||||
case item := <-w.get:
|
||||
return item.Key, item.Priority, w.shutdown.Load()
|
||||
}
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) Get() (item T, shutdown bool) {
|
||||
key, _, shutdown := w.GetWithPriority()
|
||||
return key, shutdown
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) Forget(item T) {
|
||||
w.rateLimiter.Forget(item)
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) NumRequeues(item T) int {
|
||||
return w.rateLimiter.NumRequeues(item)
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) ShuttingDown() bool {
|
||||
return w.shutdown.Load()
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) Done(item T) {
|
||||
w.lockedLock.Lock()
|
||||
defer w.lockedLock.Unlock()
|
||||
w.locked.Delete(item)
|
||||
w.metrics.done(item)
|
||||
w.notifyReadyItemOrWaiterAdded()
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) ShutDown() {
|
||||
w.shutdown.Store(true)
|
||||
close(w.done)
|
||||
}
|
||||
|
||||
// ShutDownWithDrain just calls ShutDown, as the draining
|
||||
// functionality is not used by controller-runtime.
|
||||
func (w *priorityqueue[T]) ShutDownWithDrain() {
|
||||
w.ShutDown()
|
||||
}
|
||||
|
||||
// Len returns the number of items that are ready to be
|
||||
// picked up. It does not include items that are not yet
|
||||
// ready.
|
||||
func (w *priorityqueue[T]) Len() int {
|
||||
w.lock.Lock()
|
||||
defer w.lock.Unlock()
|
||||
|
||||
// Flush is performed before reading items to avoid errors caused by asynchronous behavior,
|
||||
// primarily for unit testing purposes.
|
||||
w.lockedFlushAddBuffer()
|
||||
|
||||
return w.ready.Len()
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) logState() {
|
||||
t := time.Tick(10 * time.Second)
|
||||
for {
|
||||
select {
|
||||
case <-w.done:
|
||||
return
|
||||
case <-t:
|
||||
}
|
||||
|
||||
// Log level may change at runtime, so keep the
|
||||
// loop going even if a given level is currently
|
||||
// not enabled.
|
||||
if !w.log.V(5).Enabled() {
|
||||
continue
|
||||
}
|
||||
w.lock.Lock()
|
||||
items := make([]*item[T], 0, len(w.items))
|
||||
w.waiting.Ascend(func(item *item[T]) bool {
|
||||
items = append(items, item)
|
||||
return true
|
||||
})
|
||||
w.ready.Ascend(func(item *item[T]) bool {
|
||||
items = append(items, item)
|
||||
return true
|
||||
})
|
||||
w.lock.Unlock()
|
||||
|
||||
w.log.V(5).Info("workqueue_items", "items", items)
|
||||
}
|
||||
}
|
||||
|
||||
func lessWaiting[T comparable](a, b *item[T]) bool {
|
||||
if !a.ReadyAt.Equal(*b.ReadyAt) {
|
||||
return a.ReadyAt.Before(*b.ReadyAt)
|
||||
}
|
||||
return lessReady(a, b)
|
||||
}
|
||||
|
||||
func lessReady[T comparable](a, b *item[T]) bool {
|
||||
if a.Priority != b.Priority {
|
||||
return a.Priority > b.Priority
|
||||
}
|
||||
return a.AddedCounter < b.AddedCounter
|
||||
}
|
||||
|
||||
type item[T comparable] struct {
|
||||
Key T `json:"key"`
|
||||
AddedCounter uint64 `json:"addedCounter"`
|
||||
Priority int `json:"priority"`
|
||||
ReadyAt *time.Time `json:"readyAt,omitempty"`
|
||||
}
|
||||
|
||||
func (w *priorityqueue[T]) updateUnfinishedWorkLoop() {
|
||||
t := time.Tick(500 * time.Millisecond) // borrowed from workqueue: https://github.com/kubernetes/kubernetes/blob/67a807bf142c7a2a5ecfdb2a5d24b4cdea4cc79c/staging/src/k8s.io/client-go/util/workqueue/queue.go#L182
|
||||
for {
|
||||
select {
|
||||
case <-w.done:
|
||||
return
|
||||
case <-t:
|
||||
}
|
||||
w.metrics.updateUnfinishedWork()
|
||||
}
|
||||
}
|
||||
|
||||
type bTree[T any] interface {
|
||||
ReplaceOrInsert(item T) (T, bool)
|
||||
Delete(item T) (T, bool)
|
||||
Ascend(iterator btree.ItemIteratorG[T])
|
||||
Len() int
|
||||
}
|
||||
Reference in New Issue
Block a user