From 5fa7aaa51b49784144498f563f9ce2818ae2777a Mon Sep 17 00:00:00 2001 From: pikastech Date: Sun, 19 Jul 2026 06:21:08 +0200 Subject: [PATCH] fix(cicd): shard PaC snapshots by namespace --- .agents/skills/unidesk-cicd/SKILL.md | 3 +- .../native/cicd/pac-node-status-snapshot.sh | 55 ++++++++++++++++++- .../platform-infra-pac-node-status.test.ts | 6 +- scripts/src/platform-infra-pac-node-status.ts | 15 ++++- 4 files changed, 73 insertions(+), 6 deletions(-) diff --git a/.agents/skills/unidesk-cicd/SKILL.md b/.agents/skills/unidesk-cicd/SKILL.md index 25c7110a..c2414640 100644 --- a/.agents/skills/unidesk-cicd/SKILL.md +++ b/.agents/skills/unidesk-cicd/SKILL.md @@ -107,7 +107,8 @@ bun scripts/cli.ts hwlab nodes control-plane legacy-cicd --help - 唯一推荐入口是 `platform-infra pipelines-as-code delivery-observe --target --consumer --source-commit `； - 一次 target-side capture 直接返回匹配的 PipelineRun、Argo sync/health 与首个退化资源或 condition、同 namespace exact-source workloads、ready/digest、首断点和 `complete|partial|blocked`； - 该入口只复用节点级轻量快照，一次批量读取 PipelineRun、Argo Application 与 Deployment； - - PipelineRun list 必须使用 PaC ownership label 做服务端过滤，单 consumer 取证同时使用 owning Repository label，禁止无选择器扫描节点上全部 PipelineRun； + - PipelineRun 必须按 owning YAML consumer namespace 在同一次 target-side capture 内并行读取、合并，并使用 PaC ownership 与 owning Repository label 做服务端过滤； + - 禁止通过 `-A` 无界扫描节点全部 PipelineRun，也禁止把 namespace 分片变成多个调用侧 capture； - 禁止在入口内部调用完整 `status`、同步读取 TaskRun/log、执行 Git fetch 或拼接第二个 history 状态源； - 任一批量读取失败时必须保留 typed `readErrors` 和最后成功阶段，不得提高 YAML timeout 或把读取失败投影为对象缺失； - 固定 `remoteCaptureCount=1`、`mutation=false`，不得再手工串联 `status`、`history`、`delivery-timing` 或裸 Kubernetes 对象查询； diff --git a/scripts/native/cicd/pac-node-status-snapshot.sh b/scripts/native/cicd/pac-node-status-snapshot.sh index 0a0141ff..75f902e7 100644 --- a/scripts/native/cicd/pac-node-status-snapshot.sh +++ b/scripts/native/cicd/pac-node-status-snapshot.sh @@ -47,7 +47,60 @@ capture_list() { printf '{"items":[]}' >"$output" } -capture_list pipeline-runs pipelinerun.tekton.dev "$payload_dir/pipelineruns.json" "$UNIDESK_PAC_PIPELINERUN_SELECTOR" +capture_pipeline_runs() { + output="$1" + selector="$2" + namespace_dir="$payload_dir/pipeline-run-namespaces" + mkdir -p "$namespace_dir" + printf '{"event":"pac-node-snapshot-progress","stage":"pipeline-runs","status":"started","valuesPrinted":false}\n' >&2 + jobs="" + for namespace in $UNIDESK_PAC_PIPELINERUN_NAMESPACES; do + ( + timeout "$UNIDESK_PAC_READ_STEP_TIMEOUT_SECONDS" \ + kubectl -n "$namespace" get pipelinerun.tekton.dev -l "$selector" -o json \ + >"$namespace_dir/$namespace.json" 2>"$namespace_dir/$namespace.stderr" + ) & + jobs="$jobs $!:$namespace" + done + failed=false + for job in $jobs; do + pid="${job%%:*}" + namespace="${job#*:}" + if wait "$pid"; then + : + else + exit_code=$? + failed=true + printf '{"stage":"pipeline-runs","namespace":"%s","code":"pac-node-pipeline-runs-read-failed","exitCode":%s,"valuesPrinted":false}\n' \ + "$namespace" "$exit_code" >>"$read_errors" + printf '{"items":[]}' >"$namespace_dir/$namespace.json" + fi + done + node - "$namespace_dir" >"$output" <<'NODE' +const fs = require("node:fs"); +const path = require("node:path"); +const directory = process.argv[2]; +const items = fs.readdirSync(directory) + .filter((name) => name.endsWith(".json")) + .sort() + .flatMap((name) => { + try { + const parsed = JSON.parse(fs.readFileSync(path.join(directory, name), "utf8")); + return Array.isArray(parsed.items) ? parsed.items : []; + } catch { + return []; + } + }); +process.stdout.write(JSON.stringify({ items })); +NODE + if [ "$failed" = true ]; then + printf '{"event":"pac-node-snapshot-progress","stage":"pipeline-runs","status":"failed","valuesPrinted":false}\n' >&2 + else + printf '{"event":"pac-node-snapshot-progress","stage":"pipeline-runs","status":"completed","valuesPrinted":false}\n' >&2 + fi +} + +capture_pipeline_runs "$payload_dir/pipelineruns.json" "$UNIDESK_PAC_PIPELINERUN_SELECTOR" capture_list argo-applications application.argoproj.io "$payload_dir/applications.json" capture_list deployments deployment.apps "$payload_dir/deployments.json" diff --git a/scripts/src/platform-infra-pac-node-status.test.ts b/scripts/src/platform-infra-pac-node-status.test.ts index f7c25eac..6f252e34 100644 --- a/scripts/src/platform-infra-pac-node-status.test.ts +++ b/scripts/src/platform-infra-pac-node-status.test.ts @@ -110,13 +110,15 @@ test("node snapshot preserves typed read failure instead of deriving pipeline-mi test("native node snapshot uses one remote process with three bounded list operations", () => { const shell = readFileSync(resolve(import.meta.dir, "../native/cicd/pac-node-status-snapshot.sh"), "utf8"); - expect(shell.match(/^capture_list /gmu)).toHaveLength(3); + expect(shell.match(/^capture_pipeline_runs /gmu)).toHaveLength(1); + expect(shell.match(/^capture_list /gmu)).toHaveLength(2); expect(shell).toContain("pipelinerun.tekton.dev"); expect(shell).toContain("UNIDESK_PAC_PIPELINERUN_SELECTOR"); + expect(shell).toContain("UNIDESK_PAC_PIPELINERUN_NAMESPACES"); expect(shell).toContain("application.argoproj.io"); expect(shell).toContain("deployment.apps"); expect(shell).not.toContain("kubectl apply"); - expect(shell.match(/pac-node-snapshot-progress/gmu)).toHaveLength(3); + expect(shell.match(/pac-node-snapshot-progress/gmu)).toHaveLength(6); }); test("node snapshot selects an exact source and expands only its runtime namespace", () => { diff --git a/scripts/src/platform-infra-pac-node-status.ts b/scripts/src/platform-infra-pac-node-status.ts index db1c0d5d..9453dba1 100644 --- a/scripts/src/platform-infra-pac-node-status.ts +++ b/scripts/src/platform-infra-pac-node-status.ts @@ -40,9 +40,14 @@ const evaluatorFile = rootPath("scripts", "native", "cicd", "pac-status-evaluato export async function capturePacNodeStatusSnapshot(options: PacNodeStatusCaptureOptions): Promise> { const startedAt = Date.now(); const pipelineRunSelector = pacPipelineRunSelector(options.consumers); + const pipelineRunNamespaces = [...new Set(options.consumers + .map((consumer) => consumer.namespace) + .filter((namespace) => /^[a-z0-9]([-a-z0-9]*[a-z0-9])?$/u.test(namespace)))] + .sort(); const script = [ `export UNIDESK_PAC_READ_STEP_TIMEOUT_SECONDS=${shQuote(String(Math.max(1, Math.ceil(options.timeoutMs / 1_000))))}`, `export UNIDESK_PAC_PIPELINERUN_SELECTOR=${shQuote(pipelineRunSelector)}`, + `export UNIDESK_PAC_PIPELINERUN_NAMESPACES=${shQuote(pipelineRunNamespaces.join(" "))}`, `export UNIDESK_PAC_NODE_CONSUMERS_B64=${shQuote(Buffer.from(JSON.stringify(options.consumers), "utf8").toString("base64"))}`, `export UNIDESK_PAC_EVALUATOR_B64=${shQuote(Buffer.from(readFileSync(evaluatorFile, "utf8"), "utf8").toString("base64"))}`, readFileSync(snapshotScriptFile, "utf8"), @@ -105,8 +110,14 @@ export async function capturePacNodeStatusSnapshot(options: PacNodeStatusCapture function pacPipelineRunSelector(consumers: readonly PacNodeStatusConsumer[]): string { const labels = ["app.kubernetes.io/managed-by=pipelinesascode.tekton.dev"]; - if (consumers.length === 1 && /^[a-z0-9]([-a-z0-9.]*[a-z0-9])?$/u.test(consumers[0].repository)) { - labels.push(`pipelinesascode.tekton.dev/repository=${consumers[0].repository}`); + const repositories = [...new Set(consumers + .map((consumer) => consumer.repository) + .filter((repository) => /^[a-z0-9]([-a-z0-9.]*[a-z0-9])?$/u.test(repository)))] + .sort(); + if (repositories.length === 1) { + labels.push(`pipelinesascode.tekton.dev/repository=${repositories[0]}`); + } else if (repositories.length > 1) { + labels.push(`pipelinesascode.tekton.dev/repository in (${repositories.join(",")})`); } return labels.join(","); }