FluxCD：GitOps驱动的持续部署

# Git仓库目录结构（唯一真实来源）# repositories/ai-platform/# ├── base/                  # 所有环境共享的通用配置# │   ├── deployment.yaml# │   ├── service.yaml# │   └── kustomization.yaml# └── overlays/#     ├── dev/               # 开发环境差异化配置#     │   └── kustomization.yaml#     └── prod/              # 生产环境差异化配置#         └── kustomization.yaml---# GitRepository CRD：定义Git仓库地址和分支策略apiVersion: source.toolkit.fluxcd.io/v1kind: GitRepositorymetadata:  name: ai-platform  namespace: flux-systemspec:  interval: 1m               # 监控Git仓库的间隔  url: https://github.com/team/ai-platform  ref:    branch: main             # 跟踪main分支作为期望状态来源---# Kustomization CRD：指定同步目录路径与调和参数apiVersion: kustomize.toolkit.fluxcd.io/v1kind: Kustomizationmetadata:  name: ai-platform-prod  namespace: flux-systemspec:  interval: 5m               # 调和循环同步间隔  path: ./overlays/prod      # 同步生产环境overlay目录  sourceRef:    kind: GitRepository    name: ai-platform  prune: true                # 开启prune清理已删除资源  timeout: 2m                # 单次同步超时

# base/kustomization.yaml —— 通用配置（所有环境共享）apiVersion: kustomize.config.k8s.io/v1beta1kind: Kustomizationresources:  - deployment.yaml  - service.yamlcommonLabels:  app: ai-inference---# overlays/prod/kustomization.yaml —— 生产环境差异化配置apiVersion: kustomize.config.k8s.io/v1beta1kind: Kustomizationnamespace: ai-prod                # 生产环境独立namespace隔离resources:  - ../../base                    # 引用通用base配置patches:  - path: patch-replicas.yaml     # 覆盖副本数等差异化参数images:  - name: ai-server    newName: registry.example.com/ai-server    newTag: v2.1.0                # 生产环境固定镜像版本---# overlays/prod/patch-replicas.yaml —— 环境差异化patchapiVersion: apps/v1kind: Deploymentmetadata:  name: ai-serverspec:  replicas: 5                     # 生产环境5副本，高于开发环境

# RBAC：FluxCD ServiceAccount 最小权限原则apiVersion: v1kind: ServiceAccountmetadata:  name: flux-kustomize  namespace: flux-system---apiVersion: rbac.authorization.k8s.io/v1kind: Role                       # 仅限目标namespace，遵循最小权限metadata:  name: flux-reconciler  namespace: ai-prodrules:  - apiGroups: ["apps"]    resources: ["deployments"]    verbs: ["get", "list", "watch", "patch", "update"]   # 仅必要操作  - apiGroups: [""]    resources: ["services", "configmaps"]    verbs: ["get", "list", "watch", "create", "update", "patch"]---# SOPS加密Secret：敏感信息加密后安全存入GitapiVersion: v1kind: Secretmetadata:  name: ai-model-token  namespace: ai-prodtype: OpaquestringData:  # SOPS使用KMS/PGP加密，此处为密文，FluxCD应用时自动解密  token: ENC[AES256_GCM,data:加密内容,iv:初始向量,tag:认证标签]---# Kustomization启用SOPS解密apiVersion: kustomize.toolkit.fluxcd.io/v1kind: Kustomizationmetadata:  name: ai-prod-secrets  namespace: flux-systemspec:  decryption:    provider: sops              # 声明使用SOPS解密私钥  path: ./overlays/prod  sourceRef:    kind: GitRepository    name: ai-platform

# AI推理服务Deployment（模型版本与镜像原子化绑定）apiVersion: apps/v1kind: Deploymentmetadata:  name: ai-inference  namespace: ai-prodspec:  replicas: 3  template:    spec:      containers:        - name: inference-server          image: registry.example.com/ai-server:v2.1.0   # 模型版本即镜像tag          env:            - name: MODEL_VERSION              value: "v2.1.0"                  # 配置与镜像版本保持一致          volumeMounts:            - name: model-weights              mountPath: /models               # 大模型权重来自外部对象存储      volumes:        - name: model-weights          persistentVolumeClaim:            claimName: model-weights-pvc---# FluxCD Kustomization：声明AI服务健康检查apiVersion: kustomize.toolkit.fluxcd.io/v1kind: Kustomizationmetadata:  name: ai-inference  namespace: flux-systemspec:  path: ./overlays/prod  sourceRef:    kind: GitRepository    name: ai-platform  healthChecks:                 # 验证模型加载与推理服务就绪状态    - apiVersion: apps/v1      kind: Deployment      name: ai-inference      namespace: ai-prod  postBuild:    substitute:      model_tag: v2.1.0         # 与Git提交原子化关联的模型版本

from __future__ import annotationsimport timeimport hashlibfrom dataclasses import dataclass, fieldfrom typing import Dict, List, Optional @dataclassclass GitCommit:    commit_id: str    message: str    manifests: Dict    timestamp: float = field(default_factory=time.time) @dataclassclass ClusterResource:    name: str    kind: str    spec: Dict    healthy: bool = False class GitRepository:    def __init__(self):        self.commits: List[GitCommit] = []        self._seed()     def _seed(self):        self.commit("init", {"app-deploy": {"kind": "Deployment", "replicas": 2, "image": "ai-server:v1"}})     def commit(self, message: str, manifests: Dict) -> GitCommit:        commit_id = hashlib.sha256(f"{message}{time.time()}".encode()).hexdigest()[:8]        c = GitCommit(commit_id, message, manifests, time.time())        self.commits.append(c)        print(f"[Git] new commit {commit_id}: {message}")        return c     def latest(self) -> GitCommit:        return self.commits[-1] if self.commits else None class FluxReconciler:    def __init__(self, repo: GitRepository, prune: bool = True, interval: float = 2.0):        self.repo = repo        self.prune = prune        self.interval = interval        self.cluster: Dict[str, ClusterResource] = {}        self.last_synced_commit: Optional[str] = None     def _diff(self, desired: Dict) -> List[str]:        changes = []        for name in desired:            if name not in self.cluster:                changes.append(f"+ {name}")            elif self.cluster[name].spec != desired[name]:                changes.append(f"~ {name}")        if self.prune:            for name in list(self.cluster):                if name not in desired:                    changes.append(f"- {name}")        return changes     def _health_check(self, res: ClusterResource) -> bool:        return bool(res.spec.get("image") and res.spec.get("replicas", 0) > 0)     def reconcile(self) -> bool:        head = self.repo.latest()        if head is None or head.commit_id == self.last_synced_commit:            return False        desired = head.manifests        changes = self._diff(desired)        if not changes:            self.last_synced_commit = head.commit_id            return False        print(f"[FluxCD] drift detected: {changes}")        for name in list(self.cluster):            if name not in desired and self.prune:                del self.cluster[name]        for name, spec in desired.items():            res = ClusterResource(name, spec.get("kind", "Unknown"), spec)            res.healthy = self._health_check(res)            self.cluster[name] = res        self.last_synced_commit = head.commit_id        healthy = all(r.healthy for r in self.cluster.values())        print(f"[FluxCD] sync done commit={head.commit_id} healthy={healthy}")        return True     def run_loop(self, rounds: int = 5):        for i in range(rounds):            self.reconcile()            time.sleep(self.interval * 0.01) def main():    repo = GitRepository()    flux = FluxReconciler(repo, prune=True, interval=2.0)    print("=== start reconciliation loop ===")    flux.run_loop(2)    print("\n=== data team commits new model version ===")    repo.commit("upgrade model to v2", {"app-deploy": {"kind": "Deployment", "replicas": 3, "image": "ai-server:v2"}})    flux.run_loop(2)    print("\n=== add canary deployment ===")    repo.commit("add canary", {"app-deploy": {"kind": "Deployment", "replicas": 3, "image": "ai-server:v2"}, "canary": {"kind": "Deployment", "replicas": 1, "image": "ai-server:v2-canary"}})    flux.run_loop(2)    print("\n=== current cluster state ===")    for r in flux.cluster.values():        print(f"  {r.kind}/{r.name} spec={r.spec} healthy={r.healthy}") if __name__ == "__main__":    main()