Merge pull request #1578 from pikasTech/fix/1572-pac-gitops-flush

fix: auto flush PaC GitOps mirror closeout

config/platform-infra/pipelines-as-code.yaml

@@ -26,6 +26,12 @@ release:
   controllerServicePort: 8080
   waitTimeoutSeconds: 55
 
+closeout:
+  gitOpsMirrorFlush:
+    enabled: true
+    waitTimeoutSeconds: 55
+    maxAttempts: 5
+
 targets:
   - id: JD01
     route: JD01:k3s
@@ -154,3 +160,4 @@ consumers:
     pipelineRunPrefix: hwlab-jd01-v03-ci-poll
     argoNamespace: argocd
     argoApplication: hwlab-node-v03
+    closeoutGitOpsMirrorFlush: true

scripts/src/platform-infra-pipelines-as-code.ts

@@ -3,6 +3,8 @@ import { chmodSync, existsSync, mkdirSync, readFileSync, writeFileSync } from "n
 import { dirname, join } from "node:path";
 import type { UniDeskConfig } from "./config";
 import { rootPath } from "./config";
+import { parseNodeScopedDelegatedOptions } from "./hwlab-node/plan";
+import { nodeRuntimeEnsureGitMirrorFlushed } from "./hwlab-node/status";
 import type { RenderedCliResult } from "./output";
 import {
   capture,
@@ -61,6 +63,13 @@ interface PacConfig {
     controllerServicePort: number;
     waitTimeoutSeconds: number;
   };
+  closeout: {
+    gitOpsMirrorFlush: {
+      enabled: boolean;
+      waitTimeoutSeconds: number;
+      maxAttempts: number;
+    };
+  };
   targets: PacTarget[];
   gitea: {
     configRef: string;
@@ -110,6 +119,7 @@ interface PacConsumer {
     argoNamespace: string;
     argoApplication: string;
     repositoryRef: string;
+    closeoutGitOpsMirrorFlush: boolean;
 }
 
 interface CommonOptions {
@@ -205,6 +215,8 @@ function help(): Record<string, unknown> {
 function readPacConfig(): PacConfig {
   const root = readYamlRecord<Record<string, unknown>>(configFile, "platform-infra-pipelines-as-code");
   const release = y.objectField(root, "release", "");
+  const closeout = y.objectField(root, "closeout", "");
+  const gitOpsMirrorFlush = y.objectField(closeout, "gitOpsMirrorFlush", "closeout");
   const gitea = y.objectField(root, "gitea", "");
   const display = y.objectField(root, "display", "");
   const admin = y.objectField(gitea, "admin", "gitea");
@@ -240,6 +252,13 @@ function readPacConfig(): PacConfig {
       controllerServicePort: y.portField(release, "controllerServicePort", "release"),
       waitTimeoutSeconds: positiveInteger(release, "waitTimeoutSeconds", "release"),
     },
+    closeout: {
+      gitOpsMirrorFlush: {
+        enabled: y.booleanField(gitOpsMirrorFlush, "enabled", "closeout.gitOpsMirrorFlush"),
+        waitTimeoutSeconds: positiveInteger(gitOpsMirrorFlush, "waitTimeoutSeconds", "closeout.gitOpsMirrorFlush"),
+        maxAttempts: positiveInteger(gitOpsMirrorFlush, "maxAttempts", "closeout.gitOpsMirrorFlush"),
+      },
+    },
     targets: y.arrayOfRecords(root.targets, "targets").map(parseTarget),
     gitea: {
       configRef: y.stringField(gitea, "configRef", "gitea"),
@@ -297,6 +316,7 @@ function parseConsumer(consumer: Record<string, unknown>, path: string, defaultR
     argoNamespace: y.kubernetesNameField(consumer, "argoNamespace", path),
     argoApplication: y.kubernetesNameField(consumer, "argoApplication", path),
     repositoryRef: typeof consumer.repositoryRef === "string" && consumer.repositoryRef.length > 0 ? consumer.repositoryRef : defaultRepositoryRef,
+    closeoutGitOpsMirrorFlush: consumer.closeoutGitOpsMirrorFlush === undefined ? false : y.booleanField(consumer, "closeoutGitOpsMirrorFlush", path),
   };
 }
 
@@ -315,6 +335,7 @@ function validateConfig(config: PacConfig): void {
   if (config.version !== 1) throw new Error(`${configLabel}.version must be 1`);
   resolveTarget(config, config.defaults.targetId);
   if (config.release.waitTimeoutSeconds > 55) throw new Error(`${configLabel}.release.waitTimeoutSeconds must fit the 60s trans budget`);
+  if (config.closeout.gitOpsMirrorFlush.waitTimeoutSeconds > 55) throw new Error(`${configLabel}.closeout.gitOpsMirrorFlush.waitTimeoutSeconds must fit the 60s trans budget`);
   resolveConsumer(config, config.defaults.consumerId);
   const ids = new Set<string>();
   for (const repository of config.repositories) {
@@ -443,7 +464,10 @@ async function closeout(config: UniDeskConfig, options: CloseoutOptions): Promis
   const diagnostics = record(summary.diagnostics);
   const observedSourceCommit = stringValue(latest.sourceCommit) !== "-" ? stringValue(latest.sourceCommit) : stringValue(diagnostics.sourceCommit);
   const sourceMatched = options.sourceCommit === null || observedSourceCommit === options.sourceCommit;
-  const ready = current.ok === true && sourceMatched;
+  const baseReady = current.ok === true && sourceMatched;
+  const gitOpsMirrorFlush = runCloseoutGitOpsMirrorFlush(pac, consumer, latest, baseReady, observedSourceCommit);
+  const gitOpsMirrorFlushReady = record(gitOpsMirrorFlush).ok !== false;
+  const ready = baseReady && gitOpsMirrorFlushReady;
   return {
     ok: ready,
     action: "platform-infra-pipelines-as-code-closeout",
@@ -464,7 +488,8 @@ async function closeout(config: UniDeskConfig, options: CloseoutOptions): Promis
     },
     summary,
     status: current,
-    blocker: ready ? null : closeoutBlocker(current, options.sourceCommit, observedSourceCommit, sourceMatched),
+    gitOpsMirrorFlush,
+    blocker: ready ? null : closeoutBlocker(current, options.sourceCommit, observedSourceCommit, sourceMatched, gitOpsMirrorFlush),
     next: nextCommands(target.id, consumer.id, pac.defaults.consumerId),
     valuesPrinted: false,
   };
@@ -897,6 +922,7 @@ function renderCloseout(result: Record<string, unknown>): RenderedCliResult {
   const runtime = record(summary.runtime);
   const diagnostics = record(summary.diagnostics);
   const blocker = record(result.blocker);
+  const gitOpsMirrorFlush = record(result.gitOpsMirrorFlush);
   const lines = [
     "PLATFORM-INFRA PIPELINES-AS-CODE CLOSEOUT",
     ...table(["TARGET", "CONSUMER", "READY", "SOURCE_MATCHED", "MUTATION"], [[stringValue(target.id), stringValue(consumer.id), boolText(result.ready), boolText(result.sourceMatched), boolText(result.mutation)]]),
@@ -913,6 +939,17 @@ function renderCloseout(result: Record<string, unknown>): RenderedCliResult {
     "CICD DIAGNOSIS",
     ...table(["OK", "CODE", "PHASE", "HINT"], [[boolText(diagnostics.ok !== false), stringValue(diagnostics.code), stringValue(diagnostics.phase), compactLine(stringValue(diagnostics.hint))]]),
     "",
+    "GITOPS MIRROR FLUSH",
+    ...table(["ENABLED", "REQUIRED", "OK", "MODE", "EXECUTED", "PENDING", "IN_SYNC"], [[
+      boolText(gitOpsMirrorFlush.enabled),
+      boolText(gitOpsMirrorFlush.required),
+      boolText(gitOpsMirrorFlush.ok !== false),
+      stringValue(gitOpsMirrorFlush.mode),
+      boolText(gitOpsMirrorFlush.executed),
+      stringValue(record(gitOpsMirrorFlush.afterSummary ?? gitOpsMirrorFlush.beforeSummary).pendingFlush),
+      stringValue(record(gitOpsMirrorFlush.afterSummary ?? gitOpsMirrorFlush.beforeSummary).githubInSync),
+    ]]),
+    "",
     ...(Object.keys(blocker).length === 0
       ? ["BLOCKER", "-"]
       : ["BLOCKER", ...table(["CODE", "REASON"], [[stringValue(blocker.code), compactLine(stringValue(blocker.reason))]])]),
@@ -1103,7 +1140,7 @@ function closeoutReady(value: Record<string, unknown>, sourceCommit: string | nu
   return observedSourceCommit === sourceCommit;
 }
 
-function closeoutBlocker(value: Record<string, unknown>, expected: string | null, observed: string, sourceMatched: boolean): Record<string, unknown> {
+function closeoutBlocker(value: Record<string, unknown>, expected: string | null, observed: string, sourceMatched: boolean, gitOpsMirrorFlush: unknown): Record<string, unknown> {
   if (!sourceMatched) {
     return {
       code: "pac-closeout-source-mismatch",
@@ -1125,6 +1162,15 @@ function closeoutBlocker(value: Record<string, unknown>, expected: string | null
       valuesPrinted: false,
     };
   }
+  const flush = record(gitOpsMirrorFlush);
+  if (flush.ok === false) {
+    const next = record(flush.next);
+    return {
+      code: stringValue(flush.degradedReason, "pac-closeout-gitops-mirror-flush-failed"),
+      reason: `GitOps mirror flush did not complete: mode=${stringValue(flush.mode)} required=${stringValue(flush.required)} next=${stringValue(next.flush ?? next.status)}`,
+      valuesPrinted: false,
+    };
+  }
   const diagnostics = record(summary.diagnostics);
   return {
     code: stringValue(diagnostics.code, "pac-closeout-not-ready"),
@@ -1137,6 +1183,100 @@ function sleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
 
+function runCloseoutGitOpsMirrorFlush(pac: PacConfig, consumer: PacConsumer, latest: Record<string, unknown>, baseReady: boolean, observedSourceCommit: string): Record<string, unknown> {
+  const cfg = pac.closeout.gitOpsMirrorFlush;
+  const configSource = `${configLabel}.closeout.gitOpsMirrorFlush`;
+  const required = cfg.enabled && consumer.closeoutGitOpsMirrorFlush;
+  if (!required) {
+    return {
+      ok: true,
+      enabled: cfg.enabled,
+      required: false,
+      mode: cfg.enabled ? "consumer-not-enabled" : "disabled",
+      executed: false,
+      configSource,
+      valuesPrinted: false,
+    };
+  }
+  if (!baseReady) {
+    return {
+      ok: true,
+      enabled: true,
+      required: true,
+      mode: "waiting-for-pac-ready",
+      executed: false,
+      configSource,
+      valuesPrinted: false,
+    };
+  }
+  const pipelineRun = stringValue(latest.name);
+  if (observedSourceCommit === "-" || pipelineRun === "-") {
+    return {
+      ok: false,
+      enabled: true,
+      required: true,
+      mode: "missing-closeout-context",
+      executed: false,
+      configSource,
+      degradedReason: "pac-closeout-gitops-mirror-context-missing",
+      valuesPrinted: false,
+    };
+  }
+  const deadlineMs = Date.now() + (cfg.waitTimeoutSeconds * 1000);
+  const scoped = parseNodeScopedDelegatedOptions("git-mirror", [
+    "flush",
+    "--node",
+    consumer.node,
+    "--lane",
+    consumer.lane,
+    "--confirm",
+    "--wait",
+    "--timeout-seconds",
+    String(cfg.waitTimeoutSeconds),
+  ]);
+  const progressLines: string[] = [];
+  const flush = captureStderrLines(progressLines, () => nodeRuntimeEnsureGitMirrorFlushed(scoped, "post", observedSourceCommit, pipelineRun, null, {
+    deadlineMs,
+    maxAttempts: cfg.maxAttempts,
+  }));
+  return {
+    ok: flush.ok === true,
+    enabled: true,
+    required: true,
+    mode: stringValue(flush.mode, flush.ok === true ? "flushed" : "failed"),
+    executed: flush.executed === true,
+    sourceCommit: observedSourceCommit,
+    pipelineRun,
+    configSource,
+    waitTimeoutSeconds: cfg.waitTimeoutSeconds,
+    maxAttempts: cfg.maxAttempts,
+    beforeSummary: flush.beforeSummary ?? null,
+    afterSummary: flush.afterSummary ?? null,
+    jobName: flush.jobName ?? null,
+    degradedReason: flush.degradedReason ?? undefined,
+    next: flush.next ?? undefined,
+    flush,
+    progressLines,
+    valuesPrinted: false,
+  };
+}
+
+function captureStderrLines<T>(lines: string[], callback: () => T): T {
+  const stderr = process.stderr as typeof process.stderr & { write: (...args: unknown[]) => boolean };
+  const originalWrite = stderr.write.bind(process.stderr);
+  stderr.write = ((chunk: unknown, ...args: unknown[]): boolean => {
+    lines.push(Buffer.isBuffer(chunk) ? chunk.toString("utf8") : String(chunk));
+    const callbackArg = args.find((arg) => typeof arg === "function") as (() => void) | undefined;
+    callbackArg?.();
+    return true;
+  }) as typeof stderr.write;
+  try {
+    return callback();
+  } finally {
+    stderr.write = originalWrite as typeof stderr.write;
+  }
+}
+
 function pipelineRunGate(latest: Record<string, unknown>): Record<string, unknown> {
   const name = stringValue(latest.name);
   if (name === "-") {