unkin-agent
9da206dc7c
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.
248 lines
11 KiB
Go
248 lines
11 KiB
Go
/*
|
|
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 handler
|
|
|
|
import (
|
|
"context"
|
|
"reflect"
|
|
"time"
|
|
|
|
"k8s.io/client-go/util/workqueue"
|
|
"k8s.io/utils/ptr"
|
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
|
"sigs.k8s.io/controller-runtime/pkg/controller/priorityqueue"
|
|
"sigs.k8s.io/controller-runtime/pkg/event"
|
|
"sigs.k8s.io/controller-runtime/pkg/reconcile"
|
|
)
|
|
|
|
// EventHandler enqueues reconcile.Requests in response to events (e.g. Pod Create). EventHandlers map an Event
|
|
// for one object to trigger Reconciles for either the same object or different objects - e.g. if there is an
|
|
// Event for object with type Foo (using source.Kind) then reconcile one or more object(s) with type Bar.
|
|
//
|
|
// Identical reconcile.Requests will be batched together through the queuing mechanism before reconcile is called.
|
|
//
|
|
// * Use EnqueueRequestForObject to reconcile the object the event is for
|
|
// - do this for events for the type the Controller Reconciles. (e.g. Deployment for a Deployment Controller)
|
|
//
|
|
// * Use EnqueueRequestForOwner to reconcile the owner of the object the event is for
|
|
// - do this for events for the types the Controller creates. (e.g. ReplicaSets created by a Deployment Controller)
|
|
//
|
|
// * Use EnqueueRequestsFromMapFunc to transform an event for an object to a reconcile of an object
|
|
// of a different type - do this for events for types the Controller may be interested in, but doesn't create.
|
|
// (e.g. If Foo responds to cluster size events, map Node events to Foo objects.)
|
|
//
|
|
// Unless you are implementing your own EventHandler, you can ignore the functions on the EventHandler interface.
|
|
// Most users shouldn't need to implement their own EventHandler.
|
|
type EventHandler = TypedEventHandler[client.Object, reconcile.Request]
|
|
|
|
// TypedEventHandler enqueues reconcile.Requests in response to events (e.g. Pod Create). TypedEventHandlers map an Event
|
|
// for one object to trigger Reconciles for either the same object or different objects - e.g. if there is an
|
|
// Event for object with type Foo (using source.Kind) then reconcile one or more object(s) with type Bar.
|
|
//
|
|
// Identical reconcile.Requests will be batched together through the queuing mechanism before reconcile is called.
|
|
//
|
|
// * Use TypedEnqueueRequestForObject to reconcile the object the event is for
|
|
// - do this for events for the type the Controller Reconciles. (e.g. Deployment for a Deployment Controller)
|
|
//
|
|
// * Use TypedEnqueueRequestForOwner to reconcile the owner of the object the event is for
|
|
// - do this for events for the types the Controller creates. (e.g. ReplicaSets created by a Deployment Controller)
|
|
//
|
|
// * Use TypedEnqueueRequestsFromMapFunc to transform an event for an object to a reconcile of an object
|
|
// of a different type - do this for events for types the Controller may be interested in, but doesn't create.
|
|
// (e.g. If Foo responds to cluster size events, map Node events to Foo objects.)
|
|
//
|
|
// Unless you are implementing your own TypedEventHandler, you can ignore the functions on the TypedEventHandler interface.
|
|
// Most users shouldn't need to implement their own TypedEventHandler.
|
|
//
|
|
// TypedEventHandler is experimental and subject to future change.
|
|
type TypedEventHandler[object any, request comparable] interface {
|
|
// Create is called in response to a create event - e.g. Pod Creation.
|
|
Create(context.Context, event.TypedCreateEvent[object], workqueue.TypedRateLimitingInterface[request])
|
|
|
|
// Update is called in response to an update event - e.g. Pod Updated.
|
|
Update(context.Context, event.TypedUpdateEvent[object], workqueue.TypedRateLimitingInterface[request])
|
|
|
|
// Delete is called in response to a delete event - e.g. Pod Deleted.
|
|
Delete(context.Context, event.TypedDeleteEvent[object], workqueue.TypedRateLimitingInterface[request])
|
|
|
|
// Generic is called in response to an event of an unknown type or a synthetic event triggered as a cron or
|
|
// external trigger request - e.g. reconcile Autoscaling, or a Webhook.
|
|
Generic(context.Context, event.TypedGenericEvent[object], workqueue.TypedRateLimitingInterface[request])
|
|
}
|
|
|
|
var _ EventHandler = Funcs{}
|
|
|
|
// Funcs implements eventhandler.
|
|
type Funcs = TypedFuncs[client.Object, reconcile.Request]
|
|
|
|
// TypedFuncs implements eventhandler.
|
|
//
|
|
// TypedFuncs is experimental and subject to future change.
|
|
type TypedFuncs[object any, request comparable] struct {
|
|
// Create is called in response to an add event. Defaults to no-op.
|
|
// RateLimitingInterface is used to enqueue reconcile.Requests.
|
|
CreateFunc func(context.Context, event.TypedCreateEvent[object], workqueue.TypedRateLimitingInterface[request])
|
|
|
|
// Update is called in response to an update event. Defaults to no-op.
|
|
// RateLimitingInterface is used to enqueue reconcile.Requests.
|
|
UpdateFunc func(context.Context, event.TypedUpdateEvent[object], workqueue.TypedRateLimitingInterface[request])
|
|
|
|
// Delete is called in response to a delete event. Defaults to no-op.
|
|
// RateLimitingInterface is used to enqueue reconcile.Requests.
|
|
DeleteFunc func(context.Context, event.TypedDeleteEvent[object], workqueue.TypedRateLimitingInterface[request])
|
|
|
|
// GenericFunc is called in response to a generic event. Defaults to no-op.
|
|
// RateLimitingInterface is used to enqueue reconcile.Requests.
|
|
GenericFunc func(context.Context, event.TypedGenericEvent[object], workqueue.TypedRateLimitingInterface[request])
|
|
}
|
|
|
|
var typeForClientObject = reflect.TypeFor[client.Object]()
|
|
|
|
func implementsClientObject[object any]() bool {
|
|
return reflect.TypeFor[object]().Implements(typeForClientObject)
|
|
}
|
|
|
|
func isPriorityQueue[request comparable](q workqueue.TypedRateLimitingInterface[request]) bool {
|
|
_, ok := q.(priorityqueue.PriorityQueue[request])
|
|
return ok
|
|
}
|
|
|
|
// Create implements EventHandler.
|
|
func (h TypedFuncs[object, request]) Create(ctx context.Context, e event.TypedCreateEvent[object], q workqueue.TypedRateLimitingInterface[request]) {
|
|
if h.CreateFunc != nil {
|
|
if !implementsClientObject[object]() || !isPriorityQueue(q) || isNil(e.Object) {
|
|
h.CreateFunc(ctx, e, q)
|
|
return
|
|
}
|
|
|
|
wq := workqueueWithDefaultPriority[request]{
|
|
// We already know that we have a priority queue, that event.Object implements
|
|
// client.Object and that its not nil
|
|
PriorityQueue: q.(priorityqueue.PriorityQueue[request]),
|
|
}
|
|
if e.IsInInitialList {
|
|
wq.priority = ptr.To(LowPriority)
|
|
}
|
|
h.CreateFunc(ctx, e, wq)
|
|
}
|
|
}
|
|
|
|
// Delete implements EventHandler.
|
|
func (h TypedFuncs[object, request]) Delete(ctx context.Context, e event.TypedDeleteEvent[object], q workqueue.TypedRateLimitingInterface[request]) {
|
|
if h.DeleteFunc != nil {
|
|
h.DeleteFunc(ctx, e, q)
|
|
}
|
|
}
|
|
|
|
// Update implements EventHandler.
|
|
func (h TypedFuncs[object, request]) Update(ctx context.Context, e event.TypedUpdateEvent[object], q workqueue.TypedRateLimitingInterface[request]) {
|
|
if h.UpdateFunc != nil {
|
|
if !implementsClientObject[object]() || !isPriorityQueue(q) || isNil(e.ObjectOld) || isNil(e.ObjectNew) {
|
|
h.UpdateFunc(ctx, e, q)
|
|
return
|
|
}
|
|
|
|
wq := workqueueWithDefaultPriority[request]{
|
|
// We already know that we have a priority queue, that event.ObjectOld and ObjectNew implement
|
|
// client.Object and that they are not nil
|
|
PriorityQueue: q.(priorityqueue.PriorityQueue[request]),
|
|
}
|
|
if any(e.ObjectOld).(client.Object).GetResourceVersion() == any(e.ObjectNew).(client.Object).GetResourceVersion() {
|
|
wq.priority = ptr.To(LowPriority)
|
|
}
|
|
h.UpdateFunc(ctx, e, wq)
|
|
}
|
|
}
|
|
|
|
// Generic implements EventHandler.
|
|
func (h TypedFuncs[object, request]) Generic(ctx context.Context, e event.TypedGenericEvent[object], q workqueue.TypedRateLimitingInterface[request]) {
|
|
if h.GenericFunc != nil {
|
|
h.GenericFunc(ctx, e, q)
|
|
}
|
|
}
|
|
|
|
// LowPriority is the priority set by WithLowPriorityWhenUnchanged
|
|
const LowPriority = -100
|
|
|
|
// WithLowPriorityWhenUnchanged reduces the priority of events stemming from the initial listwatch or from a resync if
|
|
// and only if a priorityqueue.PriorityQueue is used. If not, it does nothing.
|
|
func WithLowPriorityWhenUnchanged[object client.Object, request comparable](u TypedEventHandler[object, request]) TypedEventHandler[object, request] {
|
|
// TypedFuncs already implements this so just wrap
|
|
return TypedFuncs[object, request]{
|
|
CreateFunc: u.Create,
|
|
UpdateFunc: u.Update,
|
|
DeleteFunc: u.Delete,
|
|
GenericFunc: u.Generic,
|
|
}
|
|
}
|
|
|
|
type workqueueWithDefaultPriority[request comparable] struct {
|
|
priorityqueue.PriorityQueue[request]
|
|
priority *int
|
|
}
|
|
|
|
func (w workqueueWithDefaultPriority[request]) Add(item request) {
|
|
w.PriorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: w.priority}, item)
|
|
}
|
|
|
|
func (w workqueueWithDefaultPriority[request]) AddAfter(item request, after time.Duration) {
|
|
w.PriorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: w.priority, After: after}, item)
|
|
}
|
|
|
|
func (w workqueueWithDefaultPriority[request]) AddRateLimited(item request) {
|
|
w.PriorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: w.priority, RateLimited: true}, item)
|
|
}
|
|
|
|
func (w workqueueWithDefaultPriority[request]) AddWithOpts(o priorityqueue.AddOpts, items ...request) {
|
|
if o.Priority == nil {
|
|
o.Priority = w.priority
|
|
}
|
|
w.PriorityQueue.AddWithOpts(o, items...)
|
|
}
|
|
|
|
// addToQueueCreate adds the reconcile.Request to the priorityqueue in the handler
|
|
// for Create requests if and only if the workqueue being used is of type priorityqueue.PriorityQueue[reconcile.Request]
|
|
func addToQueueCreate[T client.Object, request comparable](q workqueue.TypedRateLimitingInterface[request], evt event.TypedCreateEvent[T], item request) {
|
|
priorityQueue, isPriorityQueue := q.(priorityqueue.PriorityQueue[request])
|
|
if !isPriorityQueue {
|
|
q.Add(item)
|
|
return
|
|
}
|
|
|
|
var priority *int
|
|
if evt.IsInInitialList {
|
|
priority = ptr.To(LowPriority)
|
|
}
|
|
priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: priority}, item)
|
|
}
|
|
|
|
// addToQueueUpdate adds the reconcile.Request to the priorityqueue in the handler
|
|
// for Update requests if and only if the workqueue being used is of type priorityqueue.PriorityQueue[reconcile.Request]
|
|
func addToQueueUpdate[T client.Object, request comparable](q workqueue.TypedRateLimitingInterface[request], evt event.TypedUpdateEvent[T], item request) {
|
|
priorityQueue, isPriorityQueue := q.(priorityqueue.PriorityQueue[request])
|
|
if !isPriorityQueue {
|
|
q.Add(item)
|
|
return
|
|
}
|
|
|
|
var priority *int
|
|
if evt.ObjectOld.GetResourceVersion() == evt.ObjectNew.GetResourceVersion() {
|
|
priority = ptr.To(LowPriority)
|
|
}
|
|
priorityQueue.AddWithOpts(priorityqueue.AddOpts{Priority: priority}, item)
|
|
}
|