03 - Kubernetes Operators: CRD, Controller Pattern e Operator SDK

Kubernetes Operators: CRD, Controller Pattern e Operator SDK

Come gestisci un cluster PostgreSQL in produzione? Devi monitorare il primario, rilevare i fallimenti, promuovere una replica, aggiornare le configurazioni, eseguire backup schedulati e gestire la rotazione dei certificati. Sono operazioni che un DBA esperto sa fare a memoria, ma che richiedono ore di lavoro manuale ogni volta che qualcosa va storto alle 3 di notte.

Il pattern Kubernetes Operator permette di codificare questa knowledge operativa in software Kubernetes-native: un controller che osserva lo stato del cluster, lo confronta con lo stato desiderato, e prende le azioni necessarie per riconciliarli. Automaticamente. Continuamente. Senza intervento umano. In questo articolo costruiremo un Operator completo usando Operator SDK e Kubebuilder, capendo a fondo il controller pattern e il reconcile loop.

Cos'e un Kubernetes Operator

Il termine "Operator" fu introdotto da CoreOS nel 2016 per descrivere un pattern: un software che incapsula il know-how operativo di una specifica applicazione come un insieme di controller Kubernetes. La definizione formale di Google:

Un Operator e un metodo di packaging, deployment e gestione di un'applicazione Kubernetes. Un Operator implementa e automatizza attivita comuni di un operatore umano quando gestisce quel tipo di applicazione: deploy, aggiornamenti, backup, failover, scaling.

Un Operator estende il modello dichiarativo di Kubernetes a domini specifici. Invece di "dammi un Pod", puoi dire "dammi un cluster PostgreSQL con 3 repliche, backup giornaliero su S3, failover automatico e certificati TLS". L'Operator sa come trasformare questa specifica ad alto livello in risorse Kubernetes concrete.

Operator Maturity Model

Il Capability Model degli Operator definisce 5 livelli di maturita crescente:

Custom Resource Definition (CRD)

Un CRD estende l'API di Kubernetes con tipi di risorse custom. Invece di usare solo risorse native (Pod, Deployment, Service), puoi definire risorse del dominio specifico come PostgresCluster, KafkaTopic, MLModel.

Definire una CRD

# postgres-cluster-crd.yaml
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: postgresclusters.database.example.com
spec:
  group: database.example.com
  scope: Namespaced
  names:
    plural: postgresclusters
    singular: postgrescluster
    kind: PostgresCluster
    shortNames:
      - pgc
  versions:
    - name: v1alpha1
      served: true
      storage: true
      # Schema di validazione OpenAPI v3
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              required:
                - replicas
                - version
              properties:
                replicas:
                  type: integer
                  minimum: 1
                  maximum: 5
                  description: "Numero di repliche PostgreSQL"
                version:
                  type: string
                  enum: ["14", "15", "16"]
                  description: "Versione PostgreSQL"
                storage:
                  type: object
                  properties:
                    size:
                      type: string
                      pattern: "^[0-9]+Gi$"
                      default: "10Gi"
                    storageClass:
                      type: string
                      default: "fast-ssd"
                backup:
                  type: object
                  properties:
                    enabled:
                      type: boolean
                      default: false
                    schedule:
                      type: string
                      description: "Cron expression per backup schedulato"
                    s3Bucket:
                      type: string
                resources:
                  type: object
                  properties:
                    requests:
                      type: object
                      properties:
                        memory:
                          type: string
                        cpu:
                          type: string
                    limits:
                      type: object
                      properties:
                        memory:
                          type: string
                        cpu:
                          type: string
            status:
              type: object
              properties:
                phase:
                  type: string
                  enum: ["Pending", "Creating", "Running", "Degraded", "Failed"]
                readyReplicas:
                  type: integer
                primaryEndpoint:
                  type: string
                conditions:
                  type: array
                  items:
                    type: object
                    properties:
                      type:
                        type: string
                      status:
                        type: string
                      reason:
                        type: string
                      message:
                        type: string
                      lastTransitionTime:
                        type: string
                        format: date-time
      # Stampa colonne aggiuntive in kubectl get
      additionalPrinterColumns:
        - name: Replicas
          type: integer
          jsonPath: .spec.replicas
        - name: Version
          type: string
          jsonPath: .spec.version
        - name: Status
          type: string
          jsonPath: .status.phase
        - name: Age
          type: date
          jsonPath: .metadata.creationTimestamp
      # Subresource status (necessario per UpdateStatus)
      subresources:
        status: {}

Una Custom Resource in Azione

# my-postgres-cluster.yaml
apiVersion: database.example.com/v1alpha1
kind: PostgresCluster
metadata:
  name: myapp-db
  namespace: production
spec:
  replicas: 3
  version: "16"
  storage:
    size: "100Gi"
    storageClass: fast-ssd
  backup:
    enabled: true
    schedule: "0 2 * * *"  # ogni notte alle 2:00
    s3Bucket: "my-postgres-backups"
  resources:
    requests:
      memory: "2Gi"
      cpu: "1000m"
    limits:
      memory: "4Gi"
      cpu: "2000m"

Il Controller Pattern e il Reconcile Loop

Il cuore di un Operator e il controller: un processo che continuamente osserva lo stato attuale delle risorse nel cluster e lo confronta con lo stato desiderato dichiarato nella Custom Resource. Quando c'e una differenza (drift), il controller esegue le azioni necessarie per riconciliare i due stati. Questo ciclo si chiama reconcile loop.

// Pseudocodice del reconcile loop
for {
    desiredState = getDesiredState(customResource)
    currentState = getCurrentState(cluster)

    if currentState != desiredState {
        actions = computeActions(desiredState, currentState)
        execute(actions)
    }

    // Attendi il prossimo trigger (evento API server o requeueing)
    waitForTrigger()
}

Il controller non usa un approccio "event-driven puro" (dove ogni evento scatena un'azione specifica) ma un approccio level-based: osserva lo stato totale e lo riconcilia. Questo rende i controller piu robusti: se mancano eventi (crash, restart), il controller ripartira e converge comunque allo stato corretto.

Kubebuilder: Costruire un Operator

Kubebuilder e il framework ufficiale CNCF per costruire Operator in Go. Genera il scaffolding del progetto, gestisce la comunicazione con l'API server e fornisce helper per il reconcile loop. Operator SDK si basa su Kubebuilder e aggiunge supporto per Helm e Ansible operator.

Setup del Progetto

# Installa Kubebuilder
curl -L -o kubebuilder "https://go.kubebuilder.io/dl/latest/$(go env GOOS)/$(go env GOARCH)"
chmod +x kubebuilder
sudo mv kubebuilder /usr/local/bin/

# Crea un nuovo progetto Operator
mkdir postgres-operator && cd postgres-operator
kubebuilder init \
  --domain database.example.com \
  --repo github.com/myorg/postgres-operator

# Genera l'API e il controller per PostgresCluster
kubebuilder create api \
  --group database \
  --version v1alpha1 \
  --kind PostgresCluster \
  --resource \
  --controller

# Struttura generata:
# api/v1alpha1/
#   postgrescluster_types.go   <- Definizione della struct CRD
#   groupversion_info.go
# internal/controller/
#   postgrescluster_controller.go  <- Logica del reconcile loop
# config/crd/                      <- Manifest YAML della CRD

Definire il Tipo API

// api/v1alpha1/postgrescluster_types.go
package v1alpha1

import (
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    corev1 "k8s.io/api/core/v1"
)

// PostgresClusterSpec definisce lo stato desiderato
type PostgresClusterSpec struct {
    // +kubebuilder:validation:Minimum=1
    // +kubebuilder:validation:Maximum=5
    Replicas int32 `json:"replicas"`

    // +kubebuilder:validation:Enum={"14","15","16"}
    Version string `json:"version"`

    Storage PostgresStorageSpec `json:"storage,omitempty"`

    Backup PostgresBackupSpec `json:"backup,omitempty"`

    Resources corev1.ResourceRequirements `json:"resources,omitempty"`
}

type PostgresStorageSpec struct {
    // +kubebuilder:default="10Gi"
    Size string `json:"size,omitempty"`

    StorageClass string `json:"storageClass,omitempty"`
}

type PostgresBackupSpec struct {
    Enabled  bool   `json:"enabled,omitempty"`
    Schedule string `json:"schedule,omitempty"`
    S3Bucket string `json:"s3Bucket,omitempty"`
}

// PostgresClusterStatus descrive lo stato osservato
type PostgresClusterStatus struct {
    Phase           string             `json:"phase,omitempty"`
    ReadyReplicas   int32              `json:"readyReplicas,omitempty"`
    PrimaryEndpoint string             `json:"primaryEndpoint,omitempty"`
    Conditions      []metav1.Condition `json:"conditions,omitempty"`
}

// +kubebuilder:object:root=true
// +kubebuilder:subresource:status
// +kubebuilder:printcolumn:name="Replicas",type=integer,JSONPath=".spec.replicas"
// +kubebuilder:printcolumn:name="Status",type=string,JSONPath=".status.phase"
// +kubebuilder:printcolumn:name="Age",type=date,JSONPath=".metadata.creationTimestamp"
type PostgresCluster struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   PostgresClusterSpec   `json:"spec,omitempty"`
    Status PostgresClusterStatus `json:"status,omitempty"`
}

Il Reconcile Loop: Implementazione

// internal/controller/postgrescluster_controller.go
package controller

import (
    "context"
    "fmt"

    appsv1 "k8s.io/api/apps/v1"
    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/api/errors"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "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"

    databasev1alpha1 "github.com/myorg/postgres-operator/api/v1alpha1"
)

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

// +kubebuilder:rbac:groups=database.example.com,resources=postgresclusters,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=database.example.com,resources=postgresclusters/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete

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

    // 1. Ottieni la Custom Resource
    pgCluster := &databasev1alpha1.PostgresCluster{}
    if err := r.Get(ctx, req.NamespacedName, pgCluster); err != nil {
        if errors.IsNotFound(err) {
            // CR eliminata, pulizia gia gestita dai finalizer
            return ctrl.Result{}, nil
        }
        return ctrl.Result{}, err
    }

    logger.Info("Reconciling PostgresCluster",
        "name", pgCluster.Name,
        "namespace", pgCluster.Namespace,
        "replicas", pgCluster.Spec.Replicas)

    // 2. Reconcilia il Service headless
    if err := r.reconcileHeadlessService(ctx, pgCluster); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to reconcile headless service: %w", err)
    }

    // 3. Reconcilia lo StatefulSet
    sts, err := r.reconcileStatefulSet(ctx, pgCluster)
    if err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to reconcile statefulset: %w", err)
    }

    // 4. Aggiorna lo status della CR
    pgCluster.Status.ReadyReplicas = sts.Status.ReadyReplicas
    pgCluster.Status.PrimaryEndpoint = fmt.Sprintf(
        "%s-0.%s.%s.svc.cluster.local:5432",
        pgCluster.Name,
        pgCluster.Name,
        pgCluster.Namespace,
    )

    if sts.Status.ReadyReplicas == pgCluster.Spec.Replicas {
        pgCluster.Status.Phase = "Running"
    } else if sts.Status.ReadyReplicas > 0 {
        pgCluster.Status.Phase = "Degraded"
    } else {
        pgCluster.Status.Phase = "Creating"
    }

    if err := r.Status().Update(ctx, pgCluster); err != nil {
        return ctrl.Result{}, fmt.Errorf("failed to update status: %w", err)
    }

    logger.Info("Reconciliation complete",
        "phase", pgCluster.Status.Phase,
        "readyReplicas", pgCluster.Status.ReadyReplicas)

    return ctrl.Result{}, nil
}

func (r *PostgresClusterReconciler) reconcileStatefulSet(
    ctx context.Context,
    pgCluster *databasev1alpha1.PostgresCluster,
) (*appsv1.StatefulSet, error) {
    desired := r.buildStatefulSet(pgCluster)

    // Imposta il owner reference per la garbage collection automatica
    if err := ctrl.SetControllerReference(pgCluster, desired, r.Scheme); err != nil {
        return nil, err
    }

    existing := &appsv1.StatefulSet{}
    err := r.Get(ctx, client.ObjectKeyFromObject(desired), existing)

    if errors.IsNotFound(err) {
        // StatefulSet non esiste: crealo
        if err := r.Create(ctx, desired); err != nil {
            return nil, fmt.Errorf("failed to create StatefulSet: %w", err)
        }
        return desired, nil
    }

    if err != nil {
        return nil, err
    }

    // StatefulSet esiste: aggiornalo se necessario
    existing.Spec.Replicas = desired.Spec.Replicas
    existing.Spec.Template = desired.Spec.Template
    if err := r.Update(ctx, existing); err != nil {
        return nil, fmt.Errorf("failed to update StatefulSet: %w", err)
    }

    return existing, nil
}

func (r *PostgresClusterReconciler) buildStatefulSet(
    pgCluster *databasev1alpha1.PostgresCluster,
) *appsv1.StatefulSet {
    image := fmt.Sprintf("postgres:%s", pgCluster.Spec.Version)

    return &appsv1.StatefulSet{
        ObjectMeta: metav1.ObjectMeta{
            Name:      pgCluster.Name,
            Namespace: pgCluster.Namespace,
        },
        Spec: appsv1.StatefulSetSpec{
            Replicas:    &pgCluster.Spec.Replicas,
            ServiceName: pgCluster.Name,
            Selector: &metav1.LabelSelector{
                MatchLabels: map[string]string{"app": pgCluster.Name},
            },
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: map[string]string{"app": pgCluster.Name},
                },
                Spec: corev1.PodSpec{
                    Containers: []corev1.Container{
                        {
                            Name:      "postgres",
                            Image:     image,
                            Resources: pgCluster.Spec.Resources,
                        },
                    },
                },
            },
            VolumeClaimTemplates: []corev1.PersistentVolumeClaim{
                {
                    ObjectMeta: metav1.ObjectMeta{
                        Name: "data",
                    },
                    Spec: corev1.PersistentVolumeClaimSpec{
                        AccessModes: []corev1.PersistentVolumeAccessMode{
                            corev1.ReadWriteOnce,
                        },
                        StorageClassName: &pgCluster.Spec.Storage.StorageClass,
                        Resources: corev1.VolumeResourceRequirements{
                            Requests: corev1.ResourceList{
                                corev1.ResourceStorage: pgCluster.Spec.Storage.ParsedSize(),
                            },
                        },
                    },
                },
            },
        },
    }
}

// SetupWithManager registra il controller con il manager
func (r *PostgresClusterReconciler) SetupWithManager(
    mgr ctrl.Manager,
) error {
    return ctrl.NewControllerManagedBy(mgr).
        For(&databasev1alpha1.PostgresCluster{}).
        Owns(&appsv1.StatefulSet{}).  // Reconcilia quando cambia lo StatefulSet owned
        Owns(&corev1.Service{}).
        Complete(r)
}

Finalizer: Pulizia delle Risorse

I finalizer permettono di eseguire operazioni di cleanup prima che una risorsa venga eliminata. Senza finalizer, cancellare una CR PostgresCluster eliminerebbe la CR ma non necessariamente i dati su S3 o i backup. Con i finalizer puoi gestire questa pulizia:

// Aggiungi finalizer handling al Reconcile
const pgClusterFinalizer = "database.example.com/finalizer"

func (r *PostgresClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    pgCluster := &databasev1alpha1.PostgresCluster{}
    if err := r.Get(ctx, req.NamespacedName, pgCluster); err != nil {
        return ctrl.Result{}, client.IgnoreNotFound(err)
    }

    // Gestione eliminazione
    if !pgCluster.DeletionTimestamp.IsZero() {
        if controllerutil.ContainsFinalizer(pgCluster, pgClusterFinalizer) {
            // Esegui cleanup
            if err := r.cleanupExternalResources(ctx, pgCluster); err != nil {
                return ctrl.Result{}, err
            }
            // Rimuovi il finalizer
            controllerutil.RemoveFinalizer(pgCluster, pgClusterFinalizer)
            if err := r.Update(ctx, pgCluster); err != nil {
                return ctrl.Result{}, err
            }
        }
        return ctrl.Result{}, nil
    }

    // Aggiungi finalizer se non presente
    if !controllerutil.ContainsFinalizer(pgCluster, pgClusterFinalizer) {
        controllerutil.AddFinalizer(pgCluster, pgClusterFinalizer)
        if err := r.Update(ctx, pgCluster); err != nil {
            return ctrl.Result{}, err
        }
    }

    // ... resto della logica di reconcile
    return ctrl.Result{}, nil
}

Testing degli Operator con envtest

Kubebuilder fornisce envtest, un framework di test che avvia un API server Kubernetes reale (senza kubelet e nodi) per testare i controller in modo integrato:

// internal/controller/postgrescluster_controller_test.go
package controller

import (
    "context"
    "time"

    . "github.com/onsi/ginkgo/v2"
    . "github.com/onsi/gomega"
    appsv1 "k8s.io/api/apps/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/types"

    databasev1alpha1 "github.com/myorg/postgres-operator/api/v1alpha1"
)

var _ = Describe("PostgresCluster Controller", func() {
    const (
        timeout  = time.Second * 10
        interval = time.Millisecond * 250
    )

    Context("Quando crea un PostgresCluster", func() {
        It("Deve creare lo StatefulSet corrispondente", func() {
            ctx := context.Background()

            pgCluster := &databasev1alpha1.PostgresCluster{
                ObjectMeta: metav1.ObjectMeta{
                    Name:      "test-postgres",
                    Namespace: "default",
                },
                Spec: databasev1alpha1.PostgresClusterSpec{
                    Replicas: 1,
                    Version:  "16",
                    Storage: databasev1alpha1.PostgresStorageSpec{
                        Size:         "10Gi",
                        StorageClass: "standard",
                    },
                },
            }

            Expect(k8sClient.Create(ctx, pgCluster)).Should(Succeed())

            // Verifica che lo StatefulSet venga creato
            stsLookupKey := types.NamespacedName{
                Name:      "test-postgres",
                Namespace: "default",
            }
            createdSts := &appsv1.StatefulSet{}

            Eventually(func() bool {
                err := k8sClient.Get(ctx, stsLookupKey, createdSts)
                return err == nil
            }, timeout, interval).Should(BeTrue())

            // Verifica le specifiche dello StatefulSet
            Expect(*createdSts.Spec.Replicas).Should(Equal(int32(1)))
            Expect(createdSts.Spec.Template.Spec.Containers[0].Image).
                Should(Equal("postgres:16"))

            // Cleanup
            Expect(k8sClient.Delete(ctx, pgCluster)).Should(Succeed())
        })
    })
})

Build e Deploy dell'Operator

# Build dell'immagine
make docker-build docker-push IMG="myregistry/postgres-operator:v0.1.0"

# Deploy dell'Operator nel cluster
make deploy IMG="myregistry/postgres-operator:v0.1.0"

# Verifica il deployment
kubectl get pods -n postgres-operator-system
kubectl logs -n postgres-operator-system deployment/postgres-operator-controller-manager

# Applica una CR
kubectl apply -f my-postgres-cluster.yaml
kubectl get postgresclusters -n production
kubectl describe postgrescluster myapp-db -n production

Operator di Produzione: Esempi Reali

Non e necessario costruire un Operator per ogni applicazione comune. L'ecosistema Kubernetes offre Operator maturi per le principali applicazioni stateful:

Zalando Postgres Operator

# Installa il Postgres Operator di Zalando (level 5 maturity)
helm repo add postgres-operator-charts \
  https://opensource.zalando.com/postgres-operator/charts/postgres-operator
helm install postgres-operator \
  postgres-operator-charts/postgres-operator \
  -n postgres-operator --create-namespace

# Crea un cluster PostgreSQL con HA e backup su S3
apiVersion: "acid.zalan.do/v1"
kind: postgresql
metadata:
  name: myapp-postgres
  namespace: production
spec:
  teamId: "myteam"
  volume:
    size: 100Gi
    storageClass: fast-ssd
  numberOfInstances: 3
  users:
    myapp:
      - superuser
      - createdb
  databases:
    myapp: myapp
  postgresql:
    version: "16"
    parameters:
      shared_buffers: "1GB"
      max_connections: "200"
  resources:
    requests:
      cpu: 1000m
      memory: 2Gi
    limits:
      cpu: 2000m
      memory: 4Gi
  patroni:
    failsafe_mode: false
  # Backup automatico su S3 con WAL-G
  enableLogicalBackup: true
  logicalBackupSchedule: "00 02 * * *"

Strimzi: Kafka su Kubernetes

# Kafka cluster con Strimzi (level 5 maturity)
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: production-cluster
  namespace: kafka
spec:
  kafka:
    version: 3.7.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      default.replication.factor: 3
      min.insync.replicas: 2
      inter.broker.protocol.version: "3.7"
    storage:
      type: jbod
      volumes:
        - id: 0
          type: persistent-claim
          size: 200Gi
          class: fast-ssd
          deleteClaim: false
    resources:
      requests:
        memory: 4Gi
        cpu: 2000m
      limits:
        memory: 8Gi
        cpu: 4000m
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 10Gi
      class: fast-ssd
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}

Operator Lifecycle Manager (OLM)

L'OLM gestisce l'installazione, l'aggiornamento e la gestione del ciclo di vita degli Operator nel cluster. E il meccanismo usato da OperatorHub.io per distribuire gli Operator.

# Installa OLM nel cluster
curl -sL https://github.com/operator-framework/operator-lifecycle-manager/releases/download/v0.28.0/install.sh | bash -s v0.28.0

# Installa un Operator da OperatorHub tramite OLM
kubectl create -f https://operatorhub.io/install/postgres-operator.yaml

# Verifica gli Operator installati
kubectl get csv -n operators  # ClusterServiceVersion
kubectl get subscription -n operators

Best Practices per gli Operator

Conclusioni e Prossimi Passi

Il pattern Kubernetes Operator e la naturale estensione della filosofia dichiarativa di Kubernetes a domini applicativi complessi. Invece di gestire manualmente database, sistemi di messaggistica e servizi stateful, codifichi la knowledge operativa in un controller che lavora 24/7 per mantenere il sistema nello stato desiderato.

Kubebuilder e Operator SDK forniscono le fondamenta: scaffolding del progetto, gestione dell'API server, framework di reconcile. Ma la logica di business - come gestire un failover PostgreSQL, come scalare un cluster Kafka, come ruotare i certificati TLS - deve essere implementata con profonda conoscenza dell'applicazione specifica.