gracefully shutdown reconcilers and catalogd FBC

Signed-off-by: Joe Lanford <[email protected]>

Author: joelanford

.golangci.yaml

@@ -66,6 +66,8 @@ linters-settings:
       alias: ctrl
     - pkg: github.com/blang/semver/v4
       alias: bsemver
+    - pkg: "^github.com/operator-framework/operator-controller/internal/util/([^/]+)$"
+      alias: "${1}util"
 
 output:
   formats:

catalogd/config/base/manager/manager.yaml

@@ -85,7 +85,7 @@ spec:
         imagePullPolicy: IfNotPresent
         terminationMessagePolicy: FallbackToLogsOnError
       serviceAccountName: controller-manager
-      terminationGracePeriodSeconds: 10
+      terminationGracePeriodSeconds: 60
       volumes:
         - name: cache
           emptyDir: {}

catalogd/internal/controllers/core/clustercatalog_controller.go

@@ -40,6 +40,7 @@ import (
 	catalogdv1 "github.com/operator-framework/operator-controller/catalogd/api/v1"
 	"github.com/operator-framework/operator-controller/catalogd/internal/source"
 	"github.com/operator-framework/operator-controller/catalogd/internal/storage"
+	contextutil "github.com/operator-framework/operator-controller/internal/util/context"
 )
 
 const (
@@ -91,8 +92,11 @@ func (r *ClusterCatalogReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
+	delayedCtx, delayedCancel := contextutil.WithDelay(ctx, time.Minute)
+	defer delayedCancel()
+
 	reconciledCatsrc := existingCatsrc.DeepCopy()
-	res, reconcileErr := r.reconcile(ctx, reconciledCatsrc)
+	res, reconcileErr := r.reconcile(delayedCtx, reconciledCatsrc)
 
 	// If we encounter an error, we should delete the stored catalog metadata
 	// which represents the state of a successfully unpacked catalog. Deleting
@@ -118,15 +122,15 @@ func (r *ClusterCatalogReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 	finalizers := reconciledCatsrc.Finalizers
 
 	if updateStatus {
-		if err := r.Client.Status().Update(ctx, reconciledCatsrc); err != nil {
+		if err := r.Client.Status().Update(delayedCtx, reconciledCatsrc); err != nil {
 			reconcileErr = errors.Join(reconcileErr, fmt.Errorf("error updating status: %v", err))
 		}
 	}
 
 	reconciledCatsrc.Finalizers = finalizers
 
 	if updateFinalizers {
-		if err := r.Client.Update(ctx, reconciledCatsrc); err != nil {
+		if err := r.Client.Update(delayedCtx, reconciledCatsrc); err != nil {
 			reconcileErr = errors.Join(reconcileErr, fmt.Errorf("error updating finalizers: %v", err))
 		}
 	}

catalogd/internal/serverutil/serverutil.go

@@ -41,7 +41,7 @@ func AddCatalogServerToManager(mgr ctrl.Manager, cfg CatalogServerConfig, tlsFil
 		listener = tls.NewListener(listener, config)
 	}
 
-	shutdownTimeout := 30 * time.Second
+	shutdownTimeout := 60 * time.Second
 
 	l := mgr.GetLogger().WithName("catalogd-http-server")
 	handler := catalogdmetrics.AddMetricsToHandler(cfg.LocalStorage.StorageServerHandler())

config/base/manager/manager.yaml

@@ -85,7 +85,7 @@ spec:
             memory: 64Mi
         terminationMessagePolicy: FallbackToLogsOnError
       serviceAccountName: operator-controller-controller-manager
-      terminationGracePeriodSeconds: 10
+      terminationGracePeriodSeconds: 60
       volumes:
         - name: cache
           emptyDir: {}

internal/controllers/clusterextension_controller.go

@@ -56,6 +56,7 @@ import (
 	"github.com/operator-framework/operator-controller/internal/labels"
 	"github.com/operator-framework/operator-controller/internal/resolve"
 	rukpaksource "github.com/operator-framework/operator-controller/internal/rukpak/source"
+	contextutil "github.com/operator-framework/operator-controller/internal/util/context"
 )
 
 const (
@@ -110,8 +111,11 @@ func (r *ClusterExtensionReconciler) Reconcile(ctx context.Context, req ctrl.Req
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
+	delayedCtx, delayedCancel := contextutil.WithDelay(ctx, time.Minute)
+	defer delayedCancel()
+
 	reconciledExt := existingExt.DeepCopy()
-	res, reconcileErr := r.reconcile(ctx, reconciledExt)
+	res, reconcileErr := r.reconcile(delayedCtx, reconciledExt)
 
 	// Do checks before any Update()s, as Update() may modify the resource structure!
 	updateStatus := !equality.Semantic.DeepEqual(existingExt.Status, reconciledExt.Status)
@@ -129,14 +133,14 @@ func (r *ClusterExtensionReconciler) Reconcile(ctx context.Context, req ctrl.Req
 	// reconciledExt before updating the object.
 	finalizers := reconciledExt.Finalizers
 	if updateStatus {
-		if err := r.Client.Status().Update(ctx, reconciledExt); err != nil {
+		if err := r.Client.Status().Update(delayedCtx, reconciledExt); err != nil {
 			reconcileErr = errors.Join(reconcileErr, fmt.Errorf("error updating status: %v", err))
 		}
 	}
 	reconciledExt.Finalizers = finalizers
 
 	if updateFinalizers {
-		if err := r.Client.Update(ctx, reconciledExt); err != nil {
+		if err := r.Client.Update(delayedCtx, reconciledExt); err != nil {
 			reconcileErr = errors.Join(reconcileErr, fmt.Errorf("error updating finalizers: %v", err))
 		}
 	}

internal/util/context/context.go

@@ -0,0 +1,73 @@
+package context
+
+import (
+	"context"
+	"time"
+)
+
+type delayContext struct {
+	valuesContext       context.Context
+	cancellationContext context.Context
+}
+
+func (d *delayContext) Deadline() (time.Time, bool) {
+	return d.cancellationContext.Deadline()
+}
+
+func (d *delayContext) Done() <-chan struct{} {
+	return d.cancellationContext.Done()
+}
+
+func (d *delayContext) Err() error {
+	return d.cancellationContext.Err()
+}
+
+func (d *delayContext) Value(key interface{}) interface{} {
+	return d.valuesContext.Value(key)
+}
+
+func WithDelay(ctx context.Context, delay time.Duration) (context.Context, context.CancelFunc) {
+	delayedCtx, delayedCancel := context.WithCancel(context.Background())
+	done := make(chan struct{})
+	go func() {
+		select {
+		case <-ctx.Done():
+			// If there was no delay, cancel the delayed context immediately
+			// after the parent context is canceled, then return
+			if delay <= 0 {
+				delayedCancel()
+				return
+			}
+
+			// Otherwise, start a timer for our delayed cancellation
+			timer := time.NewTimer(delay)
+			select {
+			case <-timer.C:
+				// if the timer expires first, cancel the delayed context.
+				// timer expiration here means that the function using this
+				// context was still unable to complete even after the delay
+				delayedCancel()
+			case <-done:
+				// if done is closed first, cleanup the timer to avoid a
+				// resource leak. This case means that the function using the
+				// context was able to complete within the allotted delay
+				// duration.
+				timer.Stop()
+			}
+		case <-done:
+			// In this case, the original parent context was never cancelled
+			// so we never even started our delayed context cancellation timer.
+			// Just return to avoid leaking the goroutine.
+		}
+	}()
+
+	cancelFunc := func() {
+		close(done)     // signal the goroutine to exit
+		delayedCancel() // cancel the graceful context immediately without waiting for the timer to expire.
+	}
+
+	return &delayContext{
+		valuesContext:       ctx,
+		cancellationContext: delayedCtx,
+	}, cancelFunc
+}