Building a Stateful Kubernetes Operator with Kubebuilder and Finalizers

// api/v1/clusteredcache_types.go

package v1

import (
	appsv1 "k8s.io/api/apps/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

// ClusteredCacheSpec defines the desired state of ClusteredCache
type ClusteredCacheSpec struct {
	// Number of cache node replicas.
	// +kubebuilder:validation:Minimum=1
	Size int32 `json:"size"`

	// Version of the cache node image to deploy.
	// +kubebuilder:validation:MinLength=1
	Version string `json:"version"`
}

// ClusteredCacheStatus defines the observed state of ClusteredCache
type ClusteredCacheStatus struct {
	// Conditions represent the latest available observations of the ClusteredCache's state.
	// +optional
	// +patchMergeKey=type
	// +patchStrategy=merge
	Conditions []metav1.Condition `json:"conditions,omitempty" patchStrategy:"merge" patchMergeKey:"type"`

	// ReadyReplicas is the number of cache nodes that are currently running and ready.
	ReadyReplicas int32 `json:"readyReplicas,omitempty"`

	// CurrentVersion is the version observed on the running StatefulSet.
	CurrentVersion string `json:"currentVersion,omitempty"`
}

//+kubebuilder:object:root=true
//+kubebuilder:subresource:status
//+kubebuilder:printcolumn:name="Size",type="integer",JSONPath=".spec.size"
//+kubebuilder:printcolumn:name="Ready",type="string",JSONPath=".status.readyReplicas"
//+kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.conditions[?(@.type==\"Available\")].status"
//+kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"

// ClusteredCache is the Schema for the clusteredcaches API
type ClusteredCache struct {
	metav1.TypeMeta   `json:",inline"`
	metav1.ObjectMeta `json:"metadata,omitempty"`

	Spec   ClusteredCacheSpec   `json:"spec,omitempty"`
	Status ClusteredCacheStatus `json:"status,omitempty"`
}

//+kubebuilder:object:root=true

// ClusteredCacheList contains a list of ClusteredCache
type ClusteredCacheList struct {
	metav1.TypeMeta `json:",inline"`
	metav1.ListMeta `json:"metadata,omitempty"`
	Items           []ClusteredCache `json:"items"`
}

func init() {
	SchemeBuilder.Register(&ClusteredCache{}, &ClusteredCacheList{})
}

// internal/controller/clusteredcache_controller.go

// ... imports

const ( 
    cacheFinalizer = "cache.example.com/finalizer"
)

type ClusteredCacheReconciler struct {
	client.Client
	Scheme *runtime.Scheme
}

func (r *ClusteredCacheReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := log.FromContext(ctx)

	// 1. Fetch the ClusteredCache instance
	cache := &cachev1.ClusteredCache{}
	err := r.Get(ctx, req.NamespacedName, cache)
	if err != nil {
		if errors.IsNotFound(err) {
			log.Info("ClusteredCache resource not found. Ignoring since object must be deleted")
			return ctrl.Result{}, nil
		}
		log.Error(err, "Failed to get ClusteredCache")
		return ctrl.Result{}, err
	}

    // ... Finalizer and deletion logic will go here ...

	// 2. Reconcile the child StatefulSet
	foundSts := &appsv1.StatefulSet{}
	err = r.Get(ctx, types.NamespacedName{Name: cache.Name, Namespace: cache.Namespace}, foundSts)
	if err != nil && errors.IsNotFound(err) {
		// Define and create a new StatefulSet
		sts := r.statefulSetForClusteredCache(cache)
		log.Info("Creating a new StatefulSet", "StatefulSet.Namespace", sts.Namespace, "StatefulSet.Name", sts.Name)
		if err = r.Create(ctx, sts); err != nil {
			log.Error(err, "Failed to create new StatefulSet", "StatefulSet.Namespace", sts.Namespace, "StatefulSet.Name", sts.Name)
			return ctrl.Result{}, err
		}
		// StatefulSet created successfully - return and requeue
		return ctrl.Result{Requeue: true}, nil
	} else if err != nil {
		log.Error(err, "Failed to get StatefulSet")
		return ctrl.Result{}, err
	}

	// 3. Ensure the StatefulSet size and version are correct
	size := cache.Spec.Size
	version := cache.Spec.Version

	// Check if the spec has changed and requires an update
	needsUpdate := false
	if *foundSts.Spec.Replicas != size {
		log.Info("StatefulSet replica count mismatch", "Expected", size, "Found", *foundSts.Spec.Replicas)
		foundSts.Spec.Replicas = &size
		needsUpdate = true
	}

	// Assuming the first container is our cache container
	if len(foundSts.Spec.Template.Spec.Containers) > 0 && foundSts.Spec.Template.Spec.Containers[0].Image != "my-cache-image:"+version {
		log.Info("StatefulSet image version mismatch", "Expected", version, "Found", foundSts.Spec.Template.Spec.Containers[0].Image)
		foundSts.Spec.Template.Spec.Containers[0].Image = "my-cache-image:" + version
		needsUpdate = true
	}

	if needsUpdate {
		if err = r.Update(ctx, foundSts); err != nil {
			log.Error(err, "Failed to update StatefulSet", "StatefulSet.Namespace", foundSts.Namespace, "StatefulSet.Name", foundSts.Name)
			return ctrl.Result{}, err
		}
		// Spec updated, requeue to check status after update
		return ctrl.Result{Requeue: true}, nil
	}

    // ... Status update logic will go here ...

	return ctrl.Result{}, nil
}

// statefulSetForClusteredCache returns a ClusteredCache StatefulSet object
func (r *ClusteredCacheReconciler) statefulSetForClusteredCache(c *cachev1.ClusteredCache) *appsv1.StatefulSet {
	labels := map[string]string{"app": "clustered-cache", "cache_cr": c.Name}
	replicas := c.Spec.Size
	image := "my-cache-image:" + c.Spec.Version

	sts := &appsv1.StatefulSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:      c.Name,
			Namespace: c.Namespace,
		},
		Spec: appsv1.StatefulSetSpec{
			Replicas: &replicas,
			Selector: &metav1.LabelSelector{
				MatchLabels: labels,
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: labels,
				},
				Spec: corev1.PodSpec{
					Containers: []corev1.Container{{
						Image: image,
						Name:  "cache-node",
						Ports: []corev1.ContainerPort{{
							ContainerPort: 8080,
							Name:          "cache",
						}},
					}},
				},
			},
		},
	}

	// Set ClusteredCache instance as the owner and controller
	ctrl.SetControllerReference(c, sts, r.Scheme)
	return sts
}

// ... SetupWithManager function ...

// internal/controller/clusteredcache_controller.go (inside Reconcile)

func (r *ClusteredCacheReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := log.FromContext(ctx)

	cache := &cachev1.ClusteredCache{}
	err := r.Get(ctx, req.NamespacedName, cache)
	// ... (error handling as before)

	// Check if the object is being deleted
	isCacheMarkedToBeDeleted := cache.GetDeletionTimestamp() != nil
	if isCacheMarkedToBeDeleted {
		if controllerutil.ContainsFinalizer(cache, cacheFinalizer) {
			// Run our finalizer logic. If it fails, we requeue the request.
			if err := r.finalizeClusteredCache(ctx, log, cache); err != nil {
				// Don't remove the finalizer if cleanup fails.
				return ctrl.Result{}, err
			}

			// Cleanup was successful. Remove finalizer and update the object.
			// This is what allows the deletion to proceed.
			log.Info("Successfully finalized ClusteredCache. Removing finalizer.")
			controllerutil.RemoveFinalizer(cache, cacheFinalizer)
			err := r.Update(ctx, cache)
			if err != nil {
				return ctrl.Result{}, err
			}
		}
		// Stop reconciliation as the item is being deleted
		return ctrl.Result{}, nil
	}

	// Add finalizer for this CR if it doesn't have one
	if !controllerutil.ContainsFinalizer(cache, cacheFinalizer) {
		log.Info("Adding finalizer for the ClusteredCache")
		controllerutil.AddFinalizer(cache, cacheFinalizer)
		err = r.Update(ctx, cache)
		if err != nil {
			return ctrl.Result{}, err
		}
	}

    // ... (rest of the reconciliation logic for StatefulSet, Service, etc.)
}

// finalizeClusteredCache performs the cleanup logic before the CR is deleted.
func (r *ClusteredCacheReconciler) finalizeClusteredCache(ctx context.Context, log logr.Logger, c *cachev1.ClusteredCache) error {
	// This is our placeholder for complex cleanup logic.
	// In a real-world scenario, this could involve:
	// 1. Calling an external API to deregister the cluster.
	// 2. Creating a Kubernetes Job to backup data from a PersistentVolume to S3.
	// 3. Waiting for all connections to drain gracefully.

	log.Info("Starting finalizer logic for ClusteredCache", "name", c.Name)

	// Example: Simulate calling an external API that takes time.
	log.Info("Deregistering from external monitoring service...")
	time.Sleep(2 * time.Second) // Simulate network latency

	// Example: Simulate backing up data.
	// We could check the status of a backup Job here.
	log.Info("Flushing cache data to persistent storage...")
	time.Sleep(3 * time.Second) // Simulate I/O operations

	log.Info("Finalizer logic completed successfully.")
	return nil // Return nil on success
}

// internal/controller/clusteredcache_controller.go (end of Reconcile)

// ... after ensuring StatefulSet spec is correct ...

// 4. Update the ClusteredCache status

// First, create a deep copy of the status to avoid modifying the cache in-place
// before the update call.
newStatus := *cache.Status.DeepCopy()

newStatus.ReadyReplicas = foundSts.Status.ReadyReplicas
newStatus.CurrentVersion = cache.Spec.Version // Assume update succeeded

// Set the "Available" condition based on the StatefulSet's status.
if foundSts.Status.ReadyReplicas == cache.Spec.Size {
	availableCond := metav1.Condition{
		Type:               "Available",
		Status:             metav1.ConditionTrue,
		Reason:             "Ready",
		Message:            "All cache nodes are ready.",
		LastTransitionTime: metav1.Now(),
	}
	meta.SetStatusCondition(&newStatus.Conditions, availableCond)
} else {
	progressingCond := metav1.Condition{
		Type:               "Available",
		Status:             metav1.ConditionFalse,
		Reason:             "Progressing",
		Message:            fmt.Sprintf("Waiting for all cache nodes to be ready (%d/%d)", foundSts.Status.ReadyReplicas, cache.Spec.Size),
		LastTransitionTime: metav1.Now(),
	}
	meta.SetStatusCondition(&newStatus.Conditions, progressingCond)
}

// Only update the status if it has changed
if !reflect.DeepEqual(cache.Status, newStatus) {
	cache.Status = newStatus
	log.Info("Updating ClusteredCache status")
	if err := r.Status().Update(ctx, cache); err != nil {
		log.Error(err, "Failed to update ClusteredCache status")
		return ctrl.Result{}, err
	}
}

return ctrl.Result{}, nil

// Example of controlled requeue
configMapName := "some-required-config"
foundConfigMap := &corev1.ConfigMap{}
err := r.Get(ctx, types.NamespacedName{Name: configMapName, Namespace: cache.Namespace}, foundConfigMap)
if err != nil {
    if errors.IsNotFound(err) {
        log.Info("Dependent ConfigMap not found, requeueing after 10 seconds")
        // Don't return an error, just ask to be reconciled again after a delay.
        return ctrl.Result{RequeueAfter: 10 * time.Second}, nil
    }
    // For other errors (e.g., permissions), return the error for exponential backoff.
    return ctrl.Result{}, err
}

// internal/controller/clusteredcache_controller.go

func (r *ClusteredCacheReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&cachev1.ClusteredCache{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
		Owns(&appsv1.StatefulSet{}).
		// You can also add predicates to owned resources.
		// For example, only reconcile if the StatefulSet's replica count status changes.
		// Owns(&appsv1.StatefulSet{}, builder.WithPredicates(predicate.ResourceVersionChangedPredicate{})).
		Complete(r)
}

// cmd/main.go

// ...

ctrl.NewControllerManagedBy(mgr). 
    For(&cachev1.ClusteredCache{}). 
    // ... 
    WithOptions(controller.Options{MaxConcurrentReconciles: 10}). 
    Complete(&controller.ClusteredCacheReconciler{ ... })

// ...