Stateful Kubernetes Operators: Finalizers & Leader Election Patterns

# 1. Initialize the project
mkdir managedcache-operator && cd managedcache-operator
kubebuilder init --domain my.domain --repo my.domain/managedcache-operator

# 2. Create the API for our ManagedCache resource
kubebuilder create api --group cache --version v1alpha1 --kind ManagedCache

// api/v1alpha1/managedcache_types.go

package v1alpha1

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

// ManagedCacheSpec defines the desired state of ManagedCache
type ManagedCacheSpec struct {
	// Size is the number of Redis replicas
	// +kubebuilder:validation:Minimum=1
	Size int32 `json:"size"`

	// Image is the Redis image to use
	// +kubebuilder:default="redis:6.2.5"
	Image string `json:"image,omitempty"`
}

// ManagedCacheStatus defines the observed state of ManagedCache
type ManagedCacheStatus struct {
	// ReadyReplicas is the number of ready Redis pods.
	ReadyReplicas int32 `json:"readyReplicas,omitempty"`

	// Conditions represent the latest available observations of an object's state
	Conditions []metav1.Condition `json:"conditions,omitempty"`
}

//+kubebuilder:object:root=true
//+kubebuilder:subresource:status

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

	Spec   ManagedCacheSpec   `json:"spec,omitempty"`
	Status ManagedCacheStatus `json:"status,omitempty"`
}

//+kubebuilder:object:root=true

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

func init() {
	SchemeBuilder.Register(&ManagedCache{}, &ManagedCacheList{})
}

// controllers/managedcache_controller.go

const managedCacheFinalizer = "cache.my.domain/finalizer"

// controllers/managedcache_controller.go

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

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

	// 2. Check if the object is being deleted
	isMarkedForDeletion := cache.GetDeletionTimestamp() != nil
	if isMarkedForDeletion {
		if controllerutil.ContainsFinalizer(cache, managedCacheFinalizer) {
			// Run our finalization logic. If it fails, we return the error
			// so we can retry.
			if err := r.finalizeManagedCache(ctx, log, cache); err != nil {
				return ctrl.Result{}, err
			}

			// Remove the finalizer from the list and update it.
			controllerutil.RemoveFinalizer(cache, managedCacheFinalizer)
			if err := r.Update(ctx, cache); err != nil {
				return ctrl.Result{}, err
			}
		}
		return ctrl.Result{}, nil
	}

	// 3. Add finalizer for this CR if it doesn't exist
	if !controllerutil.ContainsFinalizer(cache, managedCacheFinalizer) {
		controllerutil.AddFinalizer(cache, managedCacheFinalizer)
		if err := r.Update(ctx, cache); err != nil {
			return ctrl.Result{}, err
		}
	}

	// ... Normal reconciliation logic (create/update Deployment, etc.) goes here ...
	// We'll implement this part later

    // For now, let's just create the deployment
    deployment, err := r.deploymentForManagedCache(cache)
    if err != nil {
        log.Error(err, "Failed to define deployment for ManagedCache")
        return ctrl.Result{}, err
    }

    // Set ManagedCache instance as the owner and controller
	if err := controllerutil.SetControllerReference(cache, deployment, r.Scheme); err != nil {
		return ctrl.Result{}, err
	}

    // Check if this Deployment already exists
    found := &appsv1.Deployment{}
    err = r.Get(ctx, types.NamespacedName{Name: deployment.Name, Namespace: deployment.Namespace}, found)
    if err != nil && errors.IsNotFound(err) {
        log.Info("Creating a new Deployment", "Deployment.Namespace", deployment.Namespace, "Deployment.Name", deployment.Name)
        err = r.Create(ctx, deployment)
        if err != nil {
            log.Error(err, "Failed to create new Deployment")
            return ctrl.Result{}, err
        }
        // Deployment created successfully - return and requeue
        return ctrl.Result{Requeue: true}, nil
    } else if err != nil {
        log.Error(err, "Failed to get Deployment")
        return ctrl.Result{}, err
    }

	return ctrl.Result{}, nil
}

// finalizeManagedCache performs cleanup actions before the CR is deleted.
func (r *ManagedCacheReconciler) finalizeManagedCache(ctx context.Context, log logr.Logger, cache *cachev1alpha1.ManagedCache) error {
	// This is where you would put your real-world cleanup logic.
	// For this example, we'll just log a message.
	log.Info("Performing finalization tasks for ManagedCache", "name", cache.Name)

	// Example: Call an external API to deregister the cache instance.
	// Simulating this with a sleep and log.
	log.Info("Simulating call to external backup service...")
	time.Sleep(2 * time.Second)
	log.Info("Backup service call complete.")

	// If your cleanup logic fails, you should return an error here.
	// The reconciliation will be retried, and the finalizer will not be removed.
	// This prevents the object from being deleted until cleanup is successful.

	log.Info("Finalization successful for ManagedCache", "name", cache.Name)
	return nil
}

// A helper function to create the deployment
func (r *ManagedCacheReconciler) deploymentForManagedCache(m *cachev1alpha1.ManagedCache) (*appsv1.Deployment, error) {
    // ... implementation to build a Redis deployment spec ...
    // This is standard boilerplate code, omitted for brevity but available in the full example repo.
    ls := map[string]string{"app": "managedcache", "managedcache_cr": m.Name}
	
	depl := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:      m.Name + "-redis",
			Namespace: m.Namespace,
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: &m.Spec.Size,
			Selector: &metav1.LabelSelector{
				MatchLabels: ls,
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: ls,
				},
				Spec: corev1.PodSpec{
					Containers: []corev1.Container{{
						Image: m.Spec.Image,
						Name:  "redis",
						Ports: []corev1.ContainerPort{{
							ContainerPort: 6379,
							Name:          "redis",
						}},
					}},
				},
			},
		},
	}
	return depl, nil
}

// main.go

// ... imports

func main() {
	// ... setup code

	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		Scheme:                 scheme,
		MetricsBindAddress:     metricsAddr,
		Port:                   9443,
		HealthProbeBindAddress: probeAddr,
		LeaderElection:         true, // This is the magic flag!
		LeaderElectionID:       "a6d7123.my.domain",
	})

	// ... controller setup
}

// main.go - Aggressive tuning
	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		// ... other options
		LeaderElection:                true,
		LeaderElectionID:              "a6d7123.my.domain",
		LeaderElectionLeaseDuration:   &metav1.Duration{Duration: 10 * time.Second},
		LeaderElectionRenewDeadline:   &metav1.Duration{Duration: 7 * time.Second},
		LeaderElectionRetryPeriod:     &metav1.Duration{Duration: 2 * time.Second},
	})

// main.go - Conservative tuning
	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		// ... other options
		LeaderElection:                true,
		LeaderElectionID:              "a6d7123.my.domain",
		LeaderElectionLeaseDuration:   &metav1.Duration{Duration: 60 * time.Second},
		LeaderElectionRenewDeadline:   &metav1.Duration{Duration: 45 * time.Second},
		LeaderElectionRetryPeriod:     &metav1.Duration{Duration: 10 * time.Second},
	})

// controllers/managedcache_controller.go

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

	// ... (1) Fetch instance, (2) Handle finalizer logic as before ...

    // (3) Add finalizer logic as before ...

	// 4. Reconcile the Deployment
	deployment := &appsv1.Deployment{}
	err = r.Get(ctx, types.NamespacedName{Name: cache.Name + "-redis", Namespace: cache.Namespace}, deployment)

	// If the deployment does not exist, create it
	if err != nil && errors.IsNotFound(err) {
		dep, err := r.deploymentForManagedCache(cache)
		if err != nil {
			log.Error(err, "Failed to build new Deployment spec")
			return ctrl.Result{}, err
		}
		if err := controllerutil.SetControllerReference(cache, dep, r.Scheme); err != nil {
			return ctrl.Result{}, err
		}
		log.Info("Creating new Deployment")
		if err := r.Create(ctx, dep); err != nil {
			log.Error(err, "Failed to create new Deployment")
			return ctrl.Result{}, err
		}
		// Requeue to update status after deployment is created
		return ctrl.Result{Requeue: true}, nil
	} else if err != nil {
		log.Error(err, "Failed to get Deployment")
		return ctrl.Result{}, err
	}

	// 5. Ensure the deployment size matches the spec
	size := cache.Spec.Size
	if *deployment.Spec.Replicas != size {
		deployment.Spec.Replicas = &size
		if err = r.Update(ctx, deployment); err != nil {
			log.Error(err, "Failed to update Deployment size")
			return ctrl.Result{}, err
		}
        // Requeue to check status after update
		return ctrl.Result{Requeue: true}, nil
	}

	// 6. Update the status
	// IMPORTANT: Use a deep copy to avoid modifying the cache object from the informer cache.
	cacheCopy := cache.DeepCopy()
	cacheCopy.Status.ReadyReplicas = deployment.Status.ReadyReplicas

	// Use the status writer to update the status subresource.
	if err := r.Status().Update(ctx, cacheCopy); err != nil {
		log.Error(err, "Failed to update ManagedCache status")
		return ctrl.Result{}, err
	}

	return ctrl.Result{}, nil
}

// In SetupWithManager, ensure you own Deployments to trigger reconciliation on their changes.
func (r *ManagedCacheReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&cachev1alpha1.ManagedCache{}).
		Owns(&appsv1.Deployment{}). // This is key!
		Complete(r)
}