“

系列：S6 平台工程 · 第 19 篇 | 难度：高级 | 阅读时间：18 分钟

Kubernetes Operator：声明式平台生命周期管理

#TL;DR

coomia-dip Kubernetes Operator 实现了平台的声明式生命周期管理，通过自定义资源定义（CRD）描述平台拓扑，Operator 控制器自动完成服务部署、配置管理、滚动升级、弹性伸缩和故障恢复。本文从 Operator 架构、CRD 设计、控制器实现、升级策略到运维能力，完整解析这一生产级部署方案。

#1. 为什么需要 Operator

#1.1 Docker Compose 的局限

Docker Compose 适合开发和测试环境，但生产部署面临：

无自愈能力：服务崩溃不会自动重启和重调度
无弹性伸缩：无法根据负载自动调整实例数
滚动升级困难：需要手动编排升级顺序
缺少声明式管理：配置变更需要手动应用

#1.2 Operator 模式

Operator 模式将运维知识编码为软件，实现自动化的Day-2运维：

Code

用户 → 声明期望状态 (CRD) → Operator → 协调实际状态 → Kubernetes
         │                      │              │
    OntoPlatform CR         Reconcile      Deployment,
    (YAML)                  Controller     Service, ConfigMap

#1.3 对标 Palantir Foundry

能力	Palantir Foundry	coomia-dip
部署方式	专有安装器	K8s Operator
声明式管理	不支持	CRD 完整支持
弹性伸缩	有限	HPA + VPA
滚动升级	内部	Operator 自动编排
多租户	支持	Namespace 隔离

#2. CRD 设计

#2.1 OntoPlatform CRD

YAML

apiVersion: onto.paas/v1alpha1
kind: OntoPlatform
metadata:
  name: onto-production
  namespace: onto-system
spec:
  # 版本
  version: "1.5.0"

  # 全局配置
  global:
    imagePullPolicy: IfNotPresent
    imageRegistry: registry.example.com/coomia-dip
    storageClass: fast-ssd
    tlsEnabled: true
    tlsSecretName: onto-tls

  # 基础设施
  infrastructure:
    postgres:
      replicas: 3
      storage: 100Gi
      resources:
        requests: { cpu: "2", memory: "4Gi" }
        limits: { cpu: "4", memory: "8Gi" }

    redis:
      replicas: 3
      mode: sentinel
      storage: 10Gi

    minio:
      replicas: 4
      storage: 500Gi
      buckets:
        - name: onto-data
        - name: onto-iceberg

    kafka:
      replicas: 3
      storage: 100Gi
      topics:
        - name: ontology-events
          partitions: 12
          replicationFactor: 3

  # Control Layer (B)
  controlPlane:
    ontologyService:
      replicas: 3
      resources:
        requests: { cpu: "1", memory: "2Gi" }
        limits: { cpu: "2", memory: "4Gi" }
      autoscaling:
        enabled: true
        minReplicas: 2
        maxReplicas: 10
        targetCPUUtilization: 70

    schemaRegistry:
      replicas: 2
      resources:
        requests: { cpu: "500m", memory: "1Gi" }

    authService:
      replicas: 2

  # Data Layer (C)
  dataPlane:
    dataService:
      replicas: 3
      autoscaling:
        enabled: true
        minReplicas: 2
        maxReplicas: 20
        targetCPUUtilization: 60

    nessie:
      replicas: 2
      storage: 50Gi

  # Intelligence Layer (D/E)
  intelligencePlane:
    reasoningService:
      replicas: 2
      gpu:
        enabled: true
        count: 1
        type: nvidia.com/gpu

    agentRuntime:
      replicas: 2
      temporal:
        enabled: true

  # API Gateway
  gateway:
    replicas: 3
    ingress:
      enabled: true
      host: platform.example.com
      tls: true

  # 可观测性
  observability:
    metrics:
      enabled: true
      serviceMonitor: true
    tracing:
      enabled: true
      jaegerEndpoint: http://jaeger:14268/api/traces
    logging:
      level: INFO
      format: json

status:
  phase: Running
  version: "1.5.0"
  conditions:
    - type: Available
      status: "True"
    - type: Progressing
      status: "False"
  components:
    controlPlane: Ready
    dataPlane: Ready
    intelligencePlane: Ready
    gateway: Ready

#3. Operator 控制器

#3.1 Reconciliation 循环

Python

class OntoPlatformReconciler:
    """OntoPlatform 控制器 - 协调循环"""

    async def reconcile(self, request: ReconcileRequest) -> ReconcileResult:
        """主协调循环"""
        platform = await self._get_platform(request)
        if platform is None:
            return ReconcileResult(requeue=False)

        try:
            # 1. 协调基础设施
            await self._reconcile_infrastructure(platform)

            # 2. 协调 Control Layer
            await self._reconcile_control_plane(platform)

            # 3. 协调 Data Layer
            await self._reconcile_data_plane(platform)

            # 4. 协调 Intelligence Layer
            await self._reconcile_intelligence_plane(platform)

            # 5. 协调 Gateway
            await self._reconcile_gateway(platform)

            # 6. 协调可观测性
            await self._reconcile_observability(platform)

            # 7. 更新状态
            await self._update_status(platform, phase="Running")

            return ReconcileResult(requeue_after=timedelta(minutes=5))

        except Exception as e:
            await self._update_status(platform, phase="Error", message=str(e))
            return ReconcileResult(requeue_after=timedelta(seconds=30))

    async def _reconcile_control_plane(self, platform: OntoPlatform) -> None:
        """协调 Control Layer"""
        spec = platform.spec.control_plane

        # Ontology Service
        await self._ensure_deployment(
            name="ontology-service",
            namespace=platform.metadata.namespace,
            image=f"{platform.spec.global_.image_registry}/ontology-service:{platform.spec.version}",
            replicas=spec.ontology_service.replicas,
            resources=spec.ontology_service.resources,
            env=self._build_ontology_env(platform),
            health_check=GrpcHealthCheck(port=9090),
        )

        await self._ensure_service(
            name="ontology-service",
            namespace=platform.metadata.namespace,
            port=9090,
            protocol="gRPC",
        )

        if spec.ontology_service.autoscaling.enabled:
            await self._ensure_hpa(
                name="ontology-service",
                namespace=platform.metadata.namespace,
                min_replicas=spec.ontology_service.autoscaling.min_replicas,
                max_replicas=spec.ontology_service.autoscaling.max_replicas,
                target_cpu=spec.ontology_service.autoscaling.target_cpu_utilization,
            )

#3.2 滚动升级

Python

class RollingUpgradeController:
    """滚动升级控制器"""

    UPGRADE_ORDER = [
        "infrastructure",   # 先升级基础设施
        "control-Layer",    # 然后 Control Layer
        "data-Layer",       # 然后 Data Layer
        "intelligence",     # 然后 Intelligence Layer
        "gateway",          # 最后 Gateway
    ]

    async def upgrade(
        self,
        platform: OntoPlatform,
        target_version: str,
    ) -> UpgradeResult:
        """执行滚动升级"""
        current_version = platform.status.version

        for component_group in self.UPGRADE_ORDER:
            components = self._get_components(platform, component_group)

            for component in components:
                # 1. 金丝雀部署（单实例验证）
                await self._canary_deploy(component, target_version)
                healthy = await self._wait_for_healthy(component, timeout=300)
                if not healthy:
                    await self._rollback(component, current_version)
                    return UpgradeResult(
                        success=False,
                        failed_component=component.name,
                        rolled_back=True,
                    )

                # 2. 逐步替换剩余实例
                await self._rolling_replace(component, target_version)

        return UpgradeResult(success=True, version=target_version)

#4. 弹性伸缩

#4.1 HPA 配置

YAML

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: ontology-service
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: ontology-service
  minReplicas: 2
  maxReplicas: 10
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70
    - type: Pods
      pods:
        metric:
          name: grpc_requests_per_second
        target:
          type: AverageValue
          averageValue: "1000"
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 60
      policies:
        - type: Pods
          value: 2
          periodSeconds: 60
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
        - type: Pods
          value: 1
          periodSeconds: 120

#5. 故障恢复

#5.1 自愈策略

Python

class SelfHealingController:
    """自愈控制器"""

    async def on_pod_failure(self, event: PodEvent) -> None:
        if event.reason == "OOMKilled":
            # 内存不足：增加内存限制
            await self._increase_memory_limit(event.pod, factor=1.5)
        elif event.reason == "CrashLoopBackOff":
            # 反复崩溃：检查配置和依赖
            await self._diagnose_crash_loop(event.pod)
        elif event.reason == "Evicted":
            # 被驱逐：检查节点资源
            await self._check_node_resources(event.node)

    async def on_node_failure(self, event: NodeEvent) -> None:
        # 节点故障：触发 Pod 重调度
        affected_pods = await self._get_pods_on_node(event.node)
        for pod in affected_pods:
            await self._reschedule_pod(pod)

#6. 多租户支持

YAML

apiVersion: onto.paas/v1alpha1
kind: OntoPlatform
metadata:
  name: tenant-alpha
  namespace: tenant-alpha
spec:
  version: "1.5.0"
  tenancy:
    mode: namespace  # namespace 隔离
    resourceQuota:
      requests.cpu: "10"
      requests.memory: "20Gi"
      limits.cpu: "20"
      limits.memory: "40Gi"
    networkPolicy:
      isolate: true
      allowedNamespaces:
        - onto-system

#7. 测试策略

Python

class TestK8sOperator:
    def test_crd_validation(self):
        """测试 CRD 验证"""
        with pytest.raises(ValidationError):
            OntoPlatform(spec={"version": ""})  # 版本不能为空

    async def test_reconciliation(self):
        """测试协调循环"""
        platform = make_test_platform()
        reconciler = OntoPlatformReconciler(k8s_client)
        result = await reconciler.reconcile(make_request(platform))
        assert result.requeue_after is not None

    async def test_rolling_upgrade(self):
        """测试滚动升级"""
        controller = RollingUpgradeController(k8s_client)
        result = await controller.upgrade(platform, "1.6.0")
        assert result.success

    async def test_self_healing(self):
        """测试自愈"""
        controller = SelfHealingController(k8s_client)
        event = PodEvent(reason="OOMKilled", pod=make_pod())
        await controller.on_pod_failure(event)

#8. 生产最佳实践

#8.1 部署建议

使用专用节点池隔离平台服务和业务负载
为关键服务设置 Pod 反亲和性，确保跨节点分布
使用 PDB（Pod Disruption Budget）保护高可用性
定期测试故障恢复流程

#8.2 监控建议

Operator 自身的健康状态需要监控
设置 CRD 状态变更的告警规则
记录所有协调操作的审计日志
监控升级过程的成功率和耗时

#9. 总结

coomia-dip Kubernetes Operator 通过声明式 CRD 和自动化控制器，实现了生产级的平台生命周期管理。关键设计亮点：

声明式管理：通过 CRD 描述平台拓扑，Operator 自动协调
滚动升级：按依赖顺序逐组件升级，支持金丝雀验证和自动回滚
弹性伸缩：HPA + 自定义指标实现智能伸缩
自愈能力：自动检测和恢复常见故障
多租户：Namespace 隔离 + 资源配额

下一篇将探讨 coomia-dip 的 OpenTelemetry 可观测性方案。