ArgoCD核心架构：GitOps交付实践

# GitOps 声明式应用示例apiVersion: argoproj.io/v1alpha1kind: Applicationmetadata:  name: my-app  namespace: argocdspec:  source:    repoURL: https://github.com/org/repo.git    targetRevision: HEAD    path: manifests/overlays/prod  destination:    server: https://kubernetes.default.svc    namespace: production  syncPolicy:    automated:      prune: true      selfHeal: true

# API Server 与 Repo Server 协作示意（伪代码）def get_manifests(app):    # API Server 委托 Repo Server 生成清单    repo_request = {        "repo": app.source.repoURL,        "revision": app.source.targetRevision,        "path": app.source.path,    }    manifests = repo_server.generate_manifests(repo_request)  # gRPC 调用    return manifests  # 返回渲染后的 K8s 清单列表

# 三路 diff 判定逻辑示意def three_way_diff(git_manifest, live_state, last_applied):    # 1. 比较 Git 与 Live    desired_change = diff(git_manifest, live_state)    # 2. 排除集群自动管理的字段    managed_change = diff(git_manifest, last_applied)    # 3. 真正漂移 = 期望变更 与 非自动字段差异的交集    real_drift = desired_change - auto_managed_fields(live_state, last_applied)    return SyncStatus.OUT_OF_SYNC if real_drift else SyncStatus.SYNCED

# sync-wave 注解示例：先建命名空间与 CRD，再建资源apiVersion: v1kind: Namespacemetadata:  name: ml-platform  annotations:    argocd.argoproj.io/sync-wave: "-1"---apiVersion: apps/v1kind: Deploymentmetadata:  name: inference-server  annotations:    argocd.argoproj.io/sync-wave: "0"

# AppProject 多租户隔离示例apiVersion: argoproj.io/v1alpha1kind: AppProjectmetadata:  name: team-ml  namespace: argocdspec:  sourceRepos:    - https://github.com/org/ml-repo.git  destinations:    - server: https://kubernetes.default.svc      namespace: ml-*  clusterResourceWhitelist:    - group: ''      kind: Namespace  roles:    - name: ml-developer      policies:        - p, proj:team-ml:ml-developer, applications, sync, team-ml/*, allow

import jsonfrom enum import Enumfrom dataclasses import dataclass, fieldfrom typing import Dict, List, Optional  class SyncStatus(Enum):    SYNCED = "Synced"    OUT_OF_SYNC = "OutOfSync"    UNKNOWN = "Unknown"  class HealthStatus(Enum):    HEALTHY = "Healthy"    DEGRADED = "Degraded"    PROGRESSING = "Progressing"    MISSING = "Missing"  @dataclassclass Resource:    name: str    kind: str    git_manifest: Dict    live_state: Optional[Dict] = None    health: HealthStatus = HealthStatus.MISSING  def detect_sync_status(resource: Resource) -> SyncStatus:    if resource.live_state is None:        return SyncStatus.OUT_OF_SYNC    if resource.git_manifest == resource.live_state:        return SyncStatus.SYNCED    return SyncStatus.OUT_OF_SYNC  def assess_health(resource: Resource) -> HealthStatus:    if resource.live_state is None:        return HealthStatus.MISSING    if resource.kind == "Deployment":        spec = resource.live_state.get("spec", {})        status = resource.live_state.get("status", {})        desired = spec.get("replicas", 1)        ready = status.get("readyReplicas", 0)        if ready >= desired:            return HealthStatus.HEALTHY        return HealthStatus.PROGRESSING    return HealthStatus.HEALTHY  HOOK_ORDER = ["PreSync", "Sync", "PostSync"]  def execute_hooks(hooks: Dict[str, List[str]]) -> List[str]:    executed = []    for phase in HOOK_ORDER:        for hook in hooks.get(phase, []):            executed.append(f"[{phase}] {hook}")    return executed  def reconciliation_loop(resources: List[Resource], auto_sync: bool = True) -> Dict:    results = {"synced": 0, "out_of_sync": 0, "health": {}}    for res in resources:        sync = detect_sync_status(res)        res.health = assess_health(res)        if sync == SyncStatus.SYNCED:            results["synced"] += 1        else:            results["out_of_sync"] += 1            if auto_sync:                res.live_state = json.loads(json.dumps(res.git_manifest))                results["synced"] += 1                results["out_of_sync"] -= 1        results["health"][res.name] = res.health.value    return results  def main():    resources = [        Resource("api-deploy", "Deployment",                 {"spec": {"replicas": 3}, "status": {"readyReplicas": 3}},                 {"spec": {"replicas": 3}, "status": {"readyReplicas": 3}}),        Resource("worker-deploy", "Deployment",                 {"spec": {"replicas": 2}, "status": {}}, None),    ]    hooks = {        "PreSync": ["backup-database"],        "Sync": ["apply-manifests"],        "PostSync": ["smoke-test", "notify-slack"],    }    print("Hook 执行顺序:", execute_hooks(hooks))    result = reconciliation_loop(resources, auto_sync=True)    print("协调循环结果:", json.dumps(result, ensure_ascii=False, indent=2))  if __name__ == "__main__":    main()