Kubernetes Operator框架实战

package com.example.operator.controller;

import io.fabric8.kubernetes.api.model.*;
import io.fabric8.kubernetes.client.*;
import io.fabric8.kubernetes.client.informer.*;
import io.fabric8.kubernetes.client.informer.cache.*;
import io.quarkus.runtime.ShutdownEvent;
import io.quarkus.runtime.StartupEvent;
import org.jboss.logging.Logger;
import io.quarkus.scheduler.Scheduled;

import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.event.Observes;
import javax.inject.Inject;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

@ApplicationScoped
public class ConfigTemplateInstanceController {

    private static final Logger LOG = Logger.getLogger(ConfigTemplateInstanceController.class);
    private static final String FINALIZER = "operator-sdk.io/configmap-cleanup";

    private final KubernetesClient client;
    private final ReconcileService reconcileService;
    private final WorkQueue<String> workQueue;
    private final Map<String, AtomicInteger> retryCounters = new ConcurrentHashMap<>();

    // SharedIndexInformers for ConfigTemplate and ConfigTemplateInstance
    private SharedIndexInformer<ConfigTemplate> templateInformer;
    private SharedIndexInformer<ConfigTemplateInstance> instanceInformer;
    private final ExecutorService workers = Executors.newFixedThreadPool(4);

    @Inject
    public ConfigTemplateInstanceController(KubernetesClient client,
                                              ReconcileService reconcileService) {
        this.client = client;
        this.reconcileService = reconcileService;
        this.workQueue = new WorkQueue<>(new ExponentialBackoff(1000, 300000, 2.0));
    }

    void onStart(@Observes StartupEvent ev) {
        LOG.info("Starting ConfigTemplateInstance Controller...");
        final String namespace = client.getNamespace();
        final var templateRes = client.resources(ConfigTemplate.class);
        final var instanceRes = client.resources(ConfigTemplateInstance.class);

        // 创建 SharedIndexInformers，共享同一个 informerFactory
        var informerFactory = new SharedInformerFactory(client);
        templateInformer = informerFactory.sharedIndexInformerFor(
            templateRes, ConfigTemplate.class, 30_000L);
        instanceInformer = informerFactory.sharedIndexInformerFor(
            instanceRes, ConfigTemplateInstance.class, 30_000L);

        // Lister 注册：提供只读缓存查询接口
        Lister<ConfigTemplate> templateLister =
            new Lister<>(templateInformer.getIndexer(), namespace);
        Lister<ConfigTemplateInstance> instanceLister =
            new Lister<>(instanceInformer.getIndexer(), namespace);

        // 注册 EventHandler：将变化事件 enqueue 到 WorkQueue
        instanceInformer.addEventHandler(new ResourceEventHandler<ConfigTemplateInstance>() {
            @Override
            public void onAdd(ConfigTemplateInstance instance) {
                enqueue("ADD", instance);
            }
            @Override
            public void onUpdate(ConfigTemplateInstance old, ConfigTemplateInstance instance) {
                enqueue("UPDATE", instance);
            }
            @Override
            public void onDelete(ConfigTemplateInstance instance, boolean deletedFinalStateUnknown) {
                enqueue("DELETE", instance);
            }
        });

        // 启动 informer 和工厂
        informerFactory.startAll();
        LOG.info("Informers started, beginning worker threads...");

        // 启动 4 个 Worker 线程处理 WorkQueue
        for (int i = 0; i < 4; i++) {
            workers.submit(() -> workerLoop(instanceLister, templateLister));
        }
    }

    private void enqueue(String action, ConfigTemplateInstance instance) {
        String key = instance.getMetadata().getNamespace() + "/" + instance.getMetadata().getName();
        if (action.equals("DELETE")) {
            // 删除时从缓存中恢复 name 以获取完整 key
            workQueue.add(key);
        } else {
            workQueue.add(key);
        }
        LOG.debugf("Enqueued %s event for %s", action, key);
    }

    private void workerLoop(Lister<ConfigTemplateInstance> instanceLister,
                             Lister<ConfigTemplate> templateLister) {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                String key = workQueue.take(); // 阻塞获取
                processKey(key, instanceLister, templateLister);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    private void processKey(String key, Lister<ConfigTemplateInstance> instanceLister,
                              Lister<ConfigTemplate> templateLister) {
        String[] parts = key.split("/");
        String namespace = parts[0];
        String name = parts[1];

        ConfigTemplateInstance instance = instanceLister.namespace(namespace).get(name);
        if (instance == null) {
            // 资源已被删除，执行清理逻辑
            handleDelete(key);
            workQueue.forget(key);
            return;
        }

        // 执行 Reconcile（带指数退避重试）
        doReconcile(instance, templateLister);
        workQueue.forget(key);
    }

    private void doReconcile(ConfigTemplateInstance instance,
                              Lister<ConfigTemplate> templateLister) {
        String instanceName = instance.getMetadata().getName();
        String instanceNs = instance.getMetadata().getNamespace();

        try {
            // === 阶段一：Finalizer 管理 ===
            if (!instance.getMetadata().getFinalizers().contains(FINALIZER)) {
                instance.getMetadata().getFinalizers().add(FINALIZER);
                client.resource(instance).update();
                LOG.infof("Added finalizer to %s", instanceName);
                return; // 需要重新入队以继续后续逻辑
            }

            // === 阶段二：获取引用的模板 ===
            String templateName = instance.getSpec().getTemplateRef();
            ConfigTemplate template = templateLister.namespace(instanceNs).get(templateName);
            if (template == null) {
                reconcileService.recordCondition(instance, "TemplateNotFound",
                    "Referenced template " + templateName + " does not exist");
                return;
            }

            // === 阶段三：遍历目标集群，执行分发 ===
            List<ClusterTarget> targets = instance.getSpec().getClusters();
            List<ClusterStatusResult> clusterResults = new ArrayList<>();

            for (ClusterTarget target : targets) {
                ClusterStatusResult result = reconcileToCluster(instance, template, target);
                clusterResults.add(result);
            }

            // === 阶段四：更新状态 ===
            updateInstanceStatus(instance, clusterResults);

            // === 阶段五：检查是否需要删除（Instance 被请求删除）===
            if (instance.getMetadata().getDeletionTimestamp() != null) {
                performCleanup(instance);
                removeFinalizer(instance);
            }

            LOG.infof("Reconcile complete for %s: %d clusters processed",
                instanceName, clusterResults.size());

        } catch (KubernetesClientException e) {
            // API Server 临时错误：触发指数退避重试
            handleRetryableError(instance, e);
        } catch (Exception e) {
            // 非重试性错误：记录事件但不重试
            reconcileService.recordNonRetryableError(instance, e);
        }
    }

    private ClusterStatusResult reconcileToCluster(ConfigTemplateInstance instance,
                                                     ConfigTemplate template,
                                                     ClusterTarget target) {
        String clusterName = target.getName();
        String targetNamespace = target.getNamespace();

        try {
            // 从 template 构建 ConfigMap，并应用 overrides
            ConfigMap configMap = buildConfigMap(template, instance, target);

            // 目标集群的 KubernetesClient
            try (KubernetesClient targetClient = buildClusterClient(target.getKubeconfig())) {
                // 尝试 Patch（避免资源版本冲突）
                ConfigMap existing = targetClient.configMaps()
                    .inNamespace(targetNamespace)
                    .withName(configMap.getMetadata().getName())
                    .get();

                if (existing == null) {
                    targetClient.configMaps().inNamespace(targetNamespace).create(configMap);
                    recordEvent(instance, "Normal", "Created",
                        "ConfigMap " + configMap.getMetadata().getName() + " in " + clusterName);
                } else {
                    targetClient.configMaps().inNamespace(targetNamespace)
                        .withName(configMap.getMetadata().getName())
                        .patch(configMap);
                    recordEvent(instance, "Normal", "Updated",
                        "ConfigMap " + configMap.getMetadata().getName() + " in " + clusterName);
                }
            }

            return ClusterStatusResult.synced(clusterName);

        } catch (Exception e) {
            LOG.errorf("Failed to reconcile to cluster %s: %s", clusterName, e.getMessage());
            return ClusterStatusResult.failed(clusterName, e.getMessage());
        }
    }

    private void handleRetryableError(ConfigTemplateInstance instance, Exception e) {
        String key = instance.getMetadata().getNamespace() + "/" +
                      instance.getMetadata().getName();
        AtomicInteger retries = retryCounters.computeIfAbsent(key,
            k -> new AtomicInteger(0));
        int count = retries.incrementAndGet();

        if (count <= 10) {
            // 指数退避重试：1s → 2s → 4s → ... → 最大 5min
            long delayMs = Math.min(1000L * (1L << (count - 1)), 300_000L);
            LOG.warnf("Retryable error for %s, attempt %d, delay %d ms: %s",
                key, count, delayMs, e.getMessage());
            workQueue.addWithDelay(key, delayMs);
        } else {
            LOG.errorf("Max retries exceeded for %s, recording failure event", key);
            reconcileService.recordCondition(instance, "ReconcileFailed",
                "Max retries exceeded: " + e.getMessage());
            retryCounters.remove(key);
        }
    }
}