Build a Self-Healing Kubernetes Operator

# crd/self-healing-policy.yaml
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: healingpolicies.platform.example.com
spec:
  group: platform.example.com
  versions:
    - name: v1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                targetDeployment:
                  type: string
                policies:
                  type: array
                  items:
                    type: object
                    properties:
                      name:
                        type: string
                      condition:
                        type: object
                        properties:
                          type:
                            type: string
                            enum: [crash_loop, high_memory, high_cpu, queue_depth, error_rate, deploy_failed]
                          threshold:
                            type: number
                          durationSeconds:
                            type: integer
                      action:
                        type: object
                        properties:
                          type:
                            type: string
                            enum: [restart, scale_up, scale_down, rollback, drain_node, alert_only]
                          params:
                            type: object
                            x-kubernetes-preserve-unknown-fields: true
                      cooldownSeconds:
                        type: integer
  scope: Namespaced
  names:
    plural: healingpolicies
    singular: healingpolicy
    kind: HealingPolicy

// src/operator/controller.ts
import * as k8s from '@kubernetes/client-node';
import { Redis } from 'ioredis';

const kc = new k8s.KubeConfig();
kc.loadFromCluster();
const k8sApi = kc.makeApiClient(k8s.AppsV1Api);
const coreApi = kc.makeApiClient(k8s.CoreV1Api);
const redis = new Redis(process.env.REDIS_URL!);

interface HealingPolicy {
  targetDeployment: string;
  policies: Array<{
    name: string;
    condition: { type: string; threshold: number; durationSeconds: number };
    action: { type: string; params?: Record<string, any> };
    cooldownSeconds: number;
  }>;
}

export async function reconcile(policy: HealingPolicy, namespace: string): Promise<void> {
  for (const rule of policy.policies) {
    // Check cooldown
    const cooldownKey = `healing:cooldown:${namespace}:${policy.targetDeployment}:${rule.name}`;
    const inCooldown = await redis.get(cooldownKey);
    if (inCooldown) continue;

    const triggered = await checkCondition(rule.condition, policy.targetDeployment, namespace);
    if (!triggered) continue;

    console.log(`[Healing] ${rule.name} triggered for ${namespace}/${policy.targetDeployment}`);

    // Execute remediation
    await executeAction(rule.action, policy.targetDeployment, namespace);

    // Set cooldown
    await redis.setex(cooldownKey, rule.cooldownSeconds, '1');

    // Record action for audit
    await recordAction(namespace, policy.targetDeployment, rule.name, rule.action.type);
  }
}

async function checkCondition(
  condition: { type: string; threshold: number; durationSeconds: number },
  deployment: string,
  namespace: string
): Promise<boolean> {
  switch (condition.type) {
    case 'crash_loop': {
      const pods = await coreApi.listNamespacedPod(
        namespace, undefined, undefined, undefined, undefined,
        `app=${deployment}`
      );
      return pods.body.items.some(pod =>
        pod.status?.containerStatuses?.some(cs =>
          cs.state?.waiting?.reason === 'CrashLoopBackOff' &&
          (cs.restartCount ?? 0) > condition.threshold
        )
      );
    }

    case 'high_memory': {
      // Query Prometheus via API
      const query = `container_memory_usage_bytes{pod=~"${deployment}.*",namespace="${namespace}"} / container_spec_memory_limit_bytes{pod=~"${deployment}.*"} * 100`;
      const result = await queryPrometheus(query);
      return result > condition.threshold;
    }

    case 'error_rate': {
      const query = `rate(http_requests_total{status=~"5..",deployment="${deployment}"}[5m]) / rate(http_requests_total{deployment="${deployment}"}[5m]) * 100`;
      const result = await queryPrometheus(query);
      return result > condition.threshold;
    }

    case 'deploy_failed': {
      const deploy = await k8sApi.readNamespacedDeployment(deployment, namespace);
      const conditions = deploy.body.status?.conditions ?? [];
      return conditions.some(c =>
        c.type === 'Progressing' && c.status === 'False' && c.reason === 'ProgressDeadlineExceeded'
      );
    }

    default:
      return false;
  }
}

async function executeAction(
  action: { type: string; params?: Record<string, any> },
  deployment: string,
  namespace: string
): Promise<void> {
  switch (action.type) {
    case 'restart': {
      // Rolling restart by patching annotation
      await k8sApi.patchNamespacedDeployment(deployment, namespace, {
        spec: { template: { metadata: { annotations: {
          'kubectl.kubernetes.io/restartedAt': new Date().toISOString(),
        }}}}
      }, undefined, undefined, undefined, undefined, undefined, {
        headers: { 'Content-Type': 'application/strategic-merge-patch+json' },
      });
      console.log(`[Healing] Restarted ${deployment}`);
      break;
    }

    case 'scale_up': {
      const maxReplicas = action.params?.maxReplicas ?? 10;
      const deploy = await k8sApi.readNamespacedDeployment(deployment, namespace);
      const current = deploy.body.spec?.replicas ?? 1;
      const target = Math.min(current + (action.params?.increment ?? 1), maxReplicas);

      await k8sApi.patchNamespacedDeployment(deployment, namespace, {
        spec: { replicas: target },
      }, undefined, undefined, undefined, undefined, undefined, {
        headers: { 'Content-Type': 'application/strategic-merge-patch+json' },
      });
      console.log(`[Healing] Scaled ${deployment} from ${current} to ${target}`);
      break;
    }

    case 'rollback': {
      // Get previous ReplicaSet
      const rsList = await k8sApi.listNamespacedReplicaSet(
        namespace, undefined, undefined, undefined, undefined, `app=${deployment}`
      );
      const sorted = rsList.body.items
        .filter(rs => (rs.status?.replicas ?? 0) >= 0)
        .sort((a, b) => {
          const aRev = parseInt(a.metadata?.annotations?.['deployment.kubernetes.io/revision'] ?? '0');
          const bRev = parseInt(b.metadata?.annotations?.['deployment.kubernetes.io/revision'] ?? '0');
          return bRev - aRev;
        });

      if (sorted.length >= 2) {
        const previousRS = sorted[1];
        const previousImage = previousRS.spec?.template?.spec?.containers?.[0]?.image;
        if (previousImage) {
          await k8sApi.patchNamespacedDeployment(deployment, namespace, {
            spec: { template: { spec: { containers: [{ name: deployment, image: previousImage }] }}},
          }, undefined, undefined, undefined, undefined, undefined, {
            headers: { 'Content-Type': 'application/strategic-merge-patch+json' },
          });
          console.log(`[Healing] Rolled back ${deployment} to ${previousImage}`);
        }
      }
      break;
    }
  }
}

async function queryPrometheus(query: string): Promise<number> {
  const res = await fetch(`${process.env.PROMETHEUS_URL}/api/v1/query?query=${encodeURIComponent(query)}`);
  const data = await res.json() as any;
  return parseFloat(data.data?.result?.[0]?.value?.[1] ?? '0');
}

async function recordAction(ns: string, deploy: string, rule: string, action: string): Promise<void> {
  console.log(`[Audit] ${new Date().toISOString()} | ${ns}/${deploy} | ${rule} | ${action}`);
}

# policies/api-service.yaml
apiVersion: platform.example.com/v1
kind: HealingPolicy
metadata:
  name: api-service-healing
  namespace: production
spec:
  targetDeployment: api-service
  policies:
    - name: crash-loop-restart
      condition:
        type: crash_loop
        threshold: 5          # more than 5 restarts
        durationSeconds: 300
      action:
        type: restart
      cooldownSeconds: 600    # don't restart again for 10 min

    - name: memory-pressure-scale
      condition:
        type: high_memory
        threshold: 85          # 85% memory usage
        durationSeconds: 120
      action:
        type: scale_up
        params:
          increment: 2
          maxReplicas: 12
      cooldownSeconds: 300

    - name: failed-deploy-rollback
      condition:
        type: deploy_failed
        threshold: 1
        durationSeconds: 300
      action:
        type: rollback
      cooldownSeconds: 1800   # 30 min cooldown

    - name: high-error-rate
      condition:
        type: error_rate
        threshold: 10          # 10% 5xx rate
        durationSeconds: 60
      action:
        type: alert_only
      cooldownSeconds: 300