Kubernetes Operators

package controllers

import (
    "context"
    "fmt"
    
    "k8s.io/apimachinery/pkg/runtime"
    ctrl "sigs.k8s.io/controller-runtime"
    "sigs.k8s.io/controller-runtime/pkg/client"
    "sigs.k8s.io/controller-runtime/pkg/log"
    
    examplev1 "example.com/api/v1"
    appsv1 "k8s.io/api/apps/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

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

// Reconcile is the main logic of the controller
func (r *DatabaseReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    log := log.FromContext(ctx)
    
    // Fetch the Database instance
    database := &examplev1.Database{}
    err := r.Get(ctx, req.NamespacedName, database)
    if err != nil {
        if errors.IsNotFound(err) {
            // Object not found, could have been deleted
            return ctrl.Result{}, nil
        }
        return ctrl.Result{}, err
    }
    
    // Define desired StatefulSet
    statefulSet := r.statefulSetForDatabase(database)
    
    // Check if StatefulSet exists
    found := &appsv1.StatefulSet{}
    err = r.Get(ctx, types.NamespacedName{
        Name:      statefulSet.Name,
        Namespace: statefulSet.Namespace,
    }, found)
    
    if err != nil && errors.IsNotFound(err) {
        // Create StatefulSet
        log.Info("Creating StatefulSet", "name", statefulSet.Name)
        err = r.Create(ctx, statefulSet)
        if err != nil {
            return ctrl.Result{}, err
        }
        // Update status
        database.Status.Phase = "Creating"
        r.Status().Update(ctx, database)
        return ctrl.Result{Requeue: true}, nil
    } else if err != nil {
        return ctrl.Result{}, err
    }
    
    // Update StatefulSet if needed
    if !reflect.DeepEqual(statefulSet.Spec, found.Spec) {
        found.Spec = statefulSet.Spec
        err = r.Update(ctx, found)
        if err != nil {
            return ctrl.Result{}, err
        }
    }
    
    // Update Database status
    database.Status.Phase = "Running"
    database.Status.Replicas = found.Status.Replicas
    err = r.Status().Update(ctx, database)
    
    return ctrl.Result{RequeueAfter: time.Minute}, nil
}

// statefulSetForDatabase returns a StatefulSet for the Database
func (r *DatabaseReconciler) statefulSetForDatabase(db *examplev1.Database) *appsv1.StatefulSet {
    labels := map[string]string{
        "app":      "database",
        "database": db.Name,
    }
    
    replicas := int32(db.Spec.Replicas)
    
    return &appsv1.StatefulSet{
        ObjectMeta: metav1.ObjectMeta{
            Name:      db.Name + "-statefulset",
            Namespace: db.Namespace,
            OwnerReferences: []metav1.OwnerReference{
                *metav1.NewControllerRef(db, examplev1.GroupVersion.WithKind("Database")),
            },
        },
        Spec: appsv1.StatefulSetSpec{
            Replicas: &replicas,
            Selector: &metav1.LabelSelector{
                MatchLabels: labels,
            },
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: labels,
                },
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{{
                        Name:  db.Spec.Engine,
                        Image: fmt.Sprintf("%s:%s", db.Spec.Engine, db.Spec.Version),
                        Ports: []corev1.ContainerPort{{
                            ContainerPort: 5432,
                            Name:          "database",
                        }},
                        VolumeMounts: []corev1.VolumeMount{{
                            Name:      "data",
                            MountPath: "/var/lib/data",
                        }},
                    }},
                },
            },
            VolumeClaimTemplates: []corev1.PersistentVolumeClaim{{
                ObjectMeta: metav1.ObjectMeta{
                    Name: "data",
                },
                Spec: corev1.PersistentVolumeClaimSpec{
                    AccessModes: []corev1.PersistentVolumeAccessMode{
                        corev1.ReadWriteOnce,
                    },
                    Resources: corev1.ResourceRequirements{
                        Requests: corev1.ResourceList{
                            corev1.ResourceStorage: resource.MustParse(db.Spec.Storage.Size),
                        },
                    },
                    StorageClassName: &db.Spec.Storage.Class,
                },
            }},
        },
    }
}

// SetupWithManager sets up the controller with the Manager
func (r *DatabaseReconciler) SetupWithManager(mgr ctrl.Manager) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&examplev1.Database{}).
        Owns(&appsv1.StatefulSet{}).
        Complete(r)
}

const databaseFinalizer = "database.example.com/finalizer"

func (r *DatabaseReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    database := &examplev1.Database{}
    err := r.Get(ctx, req.NamespacedName, database)
    if err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }
    
    // Check if the resource is marked for deletion
    if database.ObjectMeta.DeletionTimestamp != nil {
        if controllerutil.ContainsFinalizer(database, databaseFinalizer) {
            // Perform cleanup
            if err := r.deleteExternalResources(database); err != nil {
                return ctrl.Result{}, err
            }
            
            // Remove finalizer
            controllerutil.RemoveFinalizer(database, databaseFinalizer)
            err := r.Update(ctx, database)
            if err != nil {
                return ctrl.Result{}, err
            }
        }
        return ctrl.Result{}, nil
    }
    
    // Add finalizer if not present
    if !controllerutil.ContainsFinalizer(database, databaseFinalizer) {
        controllerutil.AddFinalizer(database, databaseFinalizer)
        err = r.Update(ctx, database)
        if err != nil {
            return ctrl.Result{}, err
        }
    }
    
    // Normal reconciliation logic...
    return ctrl.Result{}, nil
}

apiVersion: operators.coreos.com/v1alpha1
kind: ClusterServiceVersion
metadata:
  name: database-operator.v1.0.0
  namespace: operators
spec:
  displayName: Database Operator
  description: |
    The Database Operator manages PostgreSQL, MySQL, and MongoDB instances
    with automated backups, scaling, and failover capabilities.
  
  version: 1.0.0
  replaces: database-operator.v0.9.0
  
  icon:
  - base64data: "iVBORw0KGgoAAAANS..."
    mediatype: "image/png"
  
  customresourcedefinitions:
    owned:
    - name: databases.example.com
      version: v1
      kind: Database
      displayName: Database
      description: Represents a database instance
  
  install:
    strategy: deployment
    spec:
      deployments:
      - name: database-operator
        spec:
          replicas: 1
          selector:
            matchLabels:
              name: database-operator
          template:
            metadata:
              labels:
                name: database-operator
            spec:
              serviceAccountName: database-operator
              containers:
              - name: database-operator
                image: example/database-operator:v1.0.0
                command:
                - database-operator
                env:
                - name: WATCH_NAMESPACE
                  value: ""
                - name: OPERATOR_NAME
                  value: "database-operator"
  
  permissions:
  - serviceAccountName: database-operator
    rules:
    - apiGroups: [""]
      resources: ["pods", "services", "endpoints", "persistentvolumeclaims", "events", "configmaps", "secrets"]
      verbs: ["*"]
    - apiGroups: ["apps"]
      resources: ["deployments", "daemonsets", "replicasets", "statefulsets"]
      verbs: ["*"]
    - apiGroups: ["example.com"]
      resources: ["databases", "databases/status", "databases/finalizers"]
      verbs: ["*"]

# Create a CatalogSource
apiVersion: operators.coreos.com/v1alpha1
kind: CatalogSource
metadata:
  name: database-operators
  namespace: olm
spec:
  sourceType: grpc
  image: example/database-operator-catalog:latest
---
# Create a Subscription
apiVersion: operators.coreos.com/v1alpha1
kind: Subscription
metadata:
  name: database-operator
  namespace: operators
spec:
  channel: stable
  name: database-operator
  source: database-operators
  sourceNamespace: olm
  installPlanApproval: Automatic

package controllers_test

import (
    "context"
    "testing"
    "time"
    
    . "github.com/onsi/ginkgo"
    . "github.com/onsi/gomega"
    
    appsv1 "k8s.io/api/apps/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "sigs.k8s.io/controller-runtime/pkg/client"
    
    examplev1 "example.com/api/v1"
)

var _ = Describe("Database Controller", func() {
    Context("When creating a Database", func() {
        It("Should create a StatefulSet", func() {
            ctx := context.Background()
            database := &examplev1.Database{
                ObjectMeta: metav1.ObjectMeta{
                    Name:      "test-database",
                    Namespace: "default",
                },
                Spec: examplev1.DatabaseSpec{
                    Engine:   "postgres",
                    Version:  "14",
                    Replicas: 3,
                    Storage: examplev1.StorageSpec{
                        Size:  "10Gi",
                        Class: "standard",
                    },
                },
            }
            
            Expect(k8sClient.Create(ctx, database)).Should(Succeed())
            
            statefulSet := &appsv1.StatefulSet{}
            Eventually(func() bool {
                err := k8sClient.Get(ctx, client.ObjectKey{
                    Name:      "test-database-statefulset",
                    Namespace: "default",
                }, statefulSet)
                return err == nil
            }, time.Second*10, time.Second).Should(BeTrue())
            
            Expect(*statefulSet.Spec.Replicas).Should(Equal(int32(3)))
            Expect(statefulSet.Spec.Template.Spec.Containers[0].Image).Should(Equal("postgres:14"))
        })
        
        It("Should update Database status", func() {
            ctx := context.Background()
            database := &examplev1.Database{}
            
            Eventually(func() string {
                err := k8sClient.Get(ctx, client.ObjectKey{
                    Name:      "test-database",
                    Namespace: "default",
                }, database)
                if err != nil {
                    return ""
                }
                return database.Status.Phase
            }, time.Second*10, time.Second).Should(Equal("Running"))
        })
    })
})

// Add Prometheus metrics to your controller
import (
    "github.com/prometheus/client_golang/prometheus"
    "sigs.k8s.io/controller-runtime/pkg/metrics"
)

var (
    reconciliationDuration = prometheus.NewHistogramVec(
        prometheus.HistogramOpts{
            Name: "database_operator_reconciliation_duration_seconds",
            Help: "Duration of reconciliation in seconds",
        },
        []string{"database", "namespace"},
    )
    
    databasesTotal = prometheus.NewGaugeVec(
        prometheus.GaugeOpts{
            Name: "database_operator_databases_total",
            Help: "Total number of databases managed",
        },
        []string{"engine", "status"},
    )
)

func init() {
    metrics.Registry.MustRegister(reconciliationDuration, databasesTotal)
}

func (r *DatabaseReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    start := time.Now()
    defer func() {
        reconciliationDuration.WithLabelValues(
            req.Name, 
            req.Namespace,
        ).Observe(time.Since(start).Seconds())
    }()
    
    // Reconciliation logic...
}

func main() {
    mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
        Scheme:                 scheme,
        MetricsBindAddress:     ":8080",
        Port:                   9443,
        HealthProbeBindAddress: ":8081",
        LeaderElection:         true,
        LeaderElectionID:       "database-operator.example.com",
        LeaderElectionNamespace: "operators",
        LeaseDuration:          15 * time.Second,
        RenewDeadline:          10 * time.Second,
        RetryPeriod:            2 * time.Second,
    })
    if err != nil {
        setupLog.Error(err, "unable to start manager")
        os.Exit(1)
    }
    
    // Setup controllers...
}

# Redis CRD
apiVersion: redis.example.com/v1
kind: RedisCluster
metadata:
  name: redis-production
spec:
  replicas: 6          # 3 masters, 3 replicas
  version: "7.0"
  persistence:
    enabled: true
    size: 10Gi
  auth:
    enabled: true
    secretName: redis-auth
  backup:
    enabled: true
    schedule: "0 */6 * * *"
    destination: s3://backups/redis
  monitoring:
    enabled: true
    serviceMonitor: true
---
# Operator will create:
# - StatefulSet for Redis nodes
# - Services for client access
# - ConfigMaps for configuration
# - Secrets for authentication
# - PodDisruptionBudget for availability
# - ServiceMonitor for Prometheus
# - CronJob for backups

// Auto-scaling based on memory usage
func (r *RedisReconciler) handleAutoScaling(ctx context.Context, redis *redisv1.RedisCluster) error {
    // Get current metrics
    metrics, err := r.getRedisMetrics(redis)
    if err != nil {
        return err
    }
    
    currentReplicas := redis.Spec.Replicas
    desiredReplicas := currentReplicas
    
    // Scale up if memory usage > 80%
    if metrics.MemoryUsagePercent > 80 {
        desiredReplicas = min(currentReplicas+2, redis.Spec.MaxReplicas)
        log.Info("Scaling up Redis cluster", "from", currentReplicas, "to", desiredReplicas)
    }
    
    // Scale down if memory usage < 30%
    if metrics.MemoryUsagePercent < 30 && currentReplicas > redis.Spec.MinReplicas {
        desiredReplicas = max(currentReplicas-2, redis.Spec.MinReplicas)
        log.Info("Scaling down Redis cluster", "from", currentReplicas, "to", desiredReplicas)
    }
    
    if desiredReplicas != currentReplicas {
        redis.Spec.Replicas = desiredReplicas
        return r.Update(ctx, redis)
    }
    
    return nil
}

// Automated failover handling
func (r *RedisReconciler) handleFailover(ctx context.Context, redis *redisv1.RedisCluster) error {
    masters, replicas, err := r.getRedisTopology(redis)
    if err != nil {
        return err
    }
    
    for _, master := range masters {
        if !master.IsHealthy() {
            log.Info("Master node unhealthy, initiating failover", "node", master.Name)
            
            // Find best replica to promote
            bestReplica := r.selectBestReplica(replicas, master)
            if bestReplica == nil {
                return fmt.Errorf("no suitable replica found for failover")
            }
            
            // Promote replica to master
            if err := r.promoteReplica(bestReplica); err != nil {
                return err
            }
            
            // Update cluster configuration
            redis.Status.Topology = r.updateTopology(masters, replicas, master, bestReplica)
            redis.Status.LastFailover = metav1.Now()
            
            return r.Status().Update(ctx, redis)
        }
    }
    
    return nil
}