From 03c22b38f4bc852c4c7d042ca0a9006b931f37d7 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 21 Jun 2026 14:20:57 +0000 Subject: [PATCH] fix: parallelize node git mirror post flush --- scripts/src/hwlab-node-control-plane.ts | 32 ++++++++- scripts/src/hwlab-node-impl.ts | 87 +++++++++++++++++++++++-- 2 files changed, 112 insertions(+), 7 deletions(-) diff --git a/scripts/src/hwlab-node-control-plane.ts b/scripts/src/hwlab-node-control-plane.ts index f6e4237a..4b16559b 100644 --- a/scripts/src/hwlab-node-control-plane.ts +++ b/scripts/src/hwlab-node-control-plane.ts @@ -271,6 +271,7 @@ function infraStatus(_config: ControlPlaneConfig, node: ControlPlaneNodeSpec, ta const argo = record(components.argo); const argoInstall = record(argo.install); const gitMirror = record(components.gitMirror); + const gitMirrorGithubTransport = record(gitMirror.githubTransport); const tekton = record(components.tekton); const ciNamespace = record(components.ciNamespace); const registry = record(components.registry); @@ -286,6 +287,7 @@ function infraStatus(_config: ControlPlaneConfig, node: ControlPlaneNodeSpec, ta && boolField(gitMirror, "namespaceExists") && boolField(gitMirror, "readServiceExists") && boolField(gitMirror, "writeServiceExists") + && (gitMirrorGithubTransport.required !== true || boolField(gitMirrorGithubTransport, "ready")) && (boolField(gitMirror, "cachePvcExists") || boolField(gitMirror, "cacheHostPathReady")) && boolField(registry, "ready") && boolField(registry, "toolsImageReady") @@ -313,6 +315,7 @@ function infraStatus(_config: ControlPlaneConfig, node: ControlPlaneNodeSpec, ta gitMirrorNamespaceExists: boolField(gitMirror, "namespaceExists"), gitMirrorReadServiceExists: boolField(gitMirror, "readServiceExists"), gitMirrorWriteServiceExists: boolField(gitMirror, "writeServiceExists"), + gitMirrorGithubTransportReady: gitMirrorGithubTransport.required !== true || boolField(gitMirrorGithubTransport, "ready"), gitMirrorCachePvcExists: boolField(gitMirror, "cachePvcExists"), gitMirrorCacheHostPathReady: boolField(gitMirror, "cacheHostPathReady"), gitMirrorReadReady: boolField(gitMirror, "readDeploymentReady"), @@ -1691,6 +1694,11 @@ read_svc=${shQuote(target.gitMirror.serviceReadName)} write_svc=${shQuote(target.gitMirror.serviceWriteName)} cache_pvc=${shQuote(target.gitMirror.cachePvcName)} cache_host_path=${shQuote(target.gitMirror.cacheHostPath ?? "")} +github_transport_mode=${shQuote(target.gitMirror.githubTransport.mode)} +github_token_secret=${shQuote(target.gitMirror.githubTransport.mode === "https" ? target.gitMirror.githubTransport.tokenSecretName : "")} +github_token_key=${shQuote(target.gitMirror.githubTransport.mode === "https" ? target.gitMirror.githubTransport.tokenSecretKey : "")} +github_token_source_ref=${shQuote(target.gitMirror.githubTransport.mode === "https" ? target.gitMirror.githubTransport.tokenSourceRef : "")} +github_token_source_key=${shQuote(target.gitMirror.githubTransport.mode === "https" ? target.gitMirror.githubTransport.tokenSourceKey : "")} pipeline=${shQuote(target.tekton.pipelineName)} service_account=${shQuote(target.tekton.serviceAccountName)} argo_ns=${shQuote(target.argo.namespace)} @@ -1712,6 +1720,26 @@ exists_res() { kubectl -n "$1" get "$2" "$3" >/dev/null 2>&1 && printf true || p deploy_ready() { desired=$(kubectl -n "$1" get deploy "$2" -o 'jsonpath={.spec.replicas}' 2>/dev/null || true); ready=$(kubectl -n "$1" get deploy "$2" -o 'jsonpath={.status.readyReplicas}' 2>/dev/null || true); [ -n "$desired" ] && [ "$desired" -gt 0 ] 2>/dev/null && [ "\${ready:-0}" = "$desired" ] && printf true || printf false; } sts_ready() { desired=$(kubectl -n "$1" get statefulset "$2" -o 'jsonpath={.spec.replicas}' 2>/dev/null || true); ready=$(kubectl -n "$1" get statefulset "$2" -o 'jsonpath={.status.readyReplicas}' 2>/dev/null || true); [ -n "$desired" ] && [ "$desired" -gt 0 ] 2>/dev/null && [ "\${ready:-0}" = "$desired" ] && printf true || printf false; } endpoint_ready() { endpoints=$(kubectl -n "$1" get endpoints "$2" -o 'jsonpath={.subsets[*].addresses[*].ip}' 2>/dev/null || true); [ -n "$endpoints" ] && printf true || printf false; } +github_transport_json=$(python3 - "$github_transport_mode" "$gitmirror_ns" "$github_token_secret" "$github_token_key" "$github_token_source_ref" "$github_token_source_key" <<'PY' +import hashlib, json, subprocess, sys +mode, namespace, secret_name, secret_key, source_ref, source_key = sys.argv[1:7] +if mode != "https": + print(json.dumps({"mode": mode, "required": False, "ready": True, "valuesPrinted": False})) + raise SystemExit(0) +proc = subprocess.run(["kubectl", "-n", namespace, "get", "secret", secret_name, "-o", "json"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) +exists = proc.returncode == 0 +data = {} +if exists: + try: + data = json.loads(proc.stdout).get("data") or {} + except Exception: + data = {} +encoded = data.get(secret_key) if isinstance(data, dict) else None +present = isinstance(encoded, str) and len(encoded) > 0 +fingerprint = "sha256:" + hashlib.sha256(encoded.encode()).hexdigest()[:16] if present else None +print(json.dumps({"mode": mode, "required": True, "ready": exists and present, "tokenSecretName": secret_name, "tokenSecretKey": secret_key, "tokenSourceRef": source_ref, "tokenSourceKey": source_key, "tokenSecretExists": exists, "tokenKeyPresent": present, "tokenKeyBytes": len(encoded) if present else 0, "tokenFingerprint": fingerprint, "valuesPrinted": False})) +PY +) registry_ready=false if command -v curl >/dev/null 2>&1; then curl -fsS --max-time 3 "http://$registry/v2/" >/tmp/hwlab-registry.out 2>/tmp/hwlab-registry.err && registry_ready=true; fi tools_repo_tag=\${tools_image#\${registry}/} @@ -1804,7 +1832,7 @@ print(json.dumps({"crds": crds, "deployments": deploy, "statefulSets": sts, "crd PY argo_fragment=$(cat /tmp/hwlab-node-status-fragments.json 2>/dev/null || printf '{}') cat </dev/null 2>&1 && printf true || printf false),"controllerReady":$(deploy_ready tekton-pipelines tekton-pipelines-controller),"webhookReady":$(deploy_ready tekton-pipelines tekton-pipelines-webhook)},"ciNamespace":{"name":"$ci_ns","exists":$(exists_ns "$ci_ns"),"serviceAccountExists":$(exists_res "$ci_ns" serviceaccount "$service_account"),"pipelineExists":$(exists_res "$ci_ns" pipeline "$pipeline")},"gitMirror":{"namespace":"$gitmirror_ns","namespaceExists":$(exists_ns "$gitmirror_ns"),"readDeploymentReady":$(deploy_ready "$gitmirror_ns" "$read_deploy"),"writeDeploymentReady":$(deploy_ready "$gitmirror_ns" "$write_deploy"),"readServiceExists":$(exists_res "$gitmirror_ns" service "$read_svc"),"writeServiceExists":$(exists_res "$gitmirror_ns" service "$write_svc"),"readEndpointsReady":$(endpoint_ready "$gitmirror_ns" "$read_svc"),"writeEndpointsReady":$(endpoint_ready "$gitmirror_ns" "$write_svc"),"cachePvcExists":$(exists_res "$gitmirror_ns" pvc "$cache_pvc"),"cacheHostPath":"$cache_host_path","cacheHostPathReady":$cache_host_path_ready,"summary":{"localSource":null,"githubSource":null,"localGitops":null,"githubGitops":null,"pendingFlush":null,"flushNeeded":null,"githubInSync":null}},"argo":{"namespace":"$argo_ns","namespaceExists":$(exists_ns "$argo_ns"),"installed":$(kubectl get crd applications.argoproj.io appprojects.argoproj.io >/dev/null 2>&1 && printf true || printf false),"projectExists":$(kubectl -n "$argo_ns" get appproject "$argo_project" >/dev/null 2>&1 && printf true || printf false),"applicationExists":$(kubectl -n "$argo_ns" get application "$argo_app" >/dev/null 2>&1 && printf true || printf false),"install":$argo_fragment},"registry":{"endpoint":"$registry","ready":$registry_ready,"toolsImage":"$tools_image","toolsImageReady":$tools_image_ready},"runtimeNamespace":{"name":"$runtime_ns","exists":$(exists_ns "$runtime_ns")}}} +{"observedAt":"$(date -u +%Y-%m-%dT%H:%M:%SZ)","node":"$node","lane":"$lane","components":{"k3sNodeConfig":$k3s_fragment,"tekton":{"installed":$(kubectl get crd pipelines.tekton.dev pipelineruns.tekton.dev >/dev/null 2>&1 && printf true || printf false),"controllerReady":$(deploy_ready tekton-pipelines tekton-pipelines-controller),"webhookReady":$(deploy_ready tekton-pipelines tekton-pipelines-webhook)},"ciNamespace":{"name":"$ci_ns","exists":$(exists_ns "$ci_ns"),"serviceAccountExists":$(exists_res "$ci_ns" serviceaccount "$service_account"),"pipelineExists":$(exists_res "$ci_ns" pipeline "$pipeline")},"gitMirror":{"namespace":"$gitmirror_ns","namespaceExists":$(exists_ns "$gitmirror_ns"),"readDeploymentReady":$(deploy_ready "$gitmirror_ns" "$read_deploy"),"writeDeploymentReady":$(deploy_ready "$gitmirror_ns" "$write_deploy"),"readServiceExists":$(exists_res "$gitmirror_ns" service "$read_svc"),"writeServiceExists":$(exists_res "$gitmirror_ns" service "$write_svc"),"readEndpointsReady":$(endpoint_ready "$gitmirror_ns" "$read_svc"),"writeEndpointsReady":$(endpoint_ready "$gitmirror_ns" "$write_svc"),"cachePvcExists":$(exists_res "$gitmirror_ns" pvc "$cache_pvc"),"cacheHostPath":"$cache_host_path","cacheHostPathReady":$cache_host_path_ready,"githubTransport":$github_transport_json,"summary":{"localSource":null,"githubSource":null,"localGitops":null,"githubGitops":null,"pendingFlush":null,"flushNeeded":null,"githubInSync":null}},"argo":{"namespace":"$argo_ns","namespaceExists":$(exists_ns "$argo_ns"),"installed":$(kubectl get crd applications.argoproj.io appprojects.argoproj.io >/dev/null 2>&1 && printf true || printf false),"projectExists":$(kubectl -n "$argo_ns" get appproject "$argo_project" >/dev/null 2>&1 && printf true || printf false),"applicationExists":$(kubectl -n "$argo_ns" get application "$argo_app" >/dev/null 2>&1 && printf true || printf false),"install":$argo_fragment},"registry":{"endpoint":"$registry","ready":$registry_ready,"toolsImage":"$tools_image","toolsImageReady":$tools_image_ready},"runtimeNamespace":{"name":"$runtime_ns","exists":$(exists_ns "$runtime_ns")}}} JSON `; } @@ -2135,6 +2163,8 @@ function statusNext( if (!boolField(registry, "ready")) blockers.push("node-local-registry-not-ready"); if (!boolField(registry, "toolsImageReady")) blockers.push("tools-image-missing"); if (bootstrapMissing) blockers.push("control-plane-bootstrap-missing"); + const gitMirrorGithubTransport = record(gitMirror.githubTransport); + if (gitMirrorGithubTransport.required === true && !boolField(gitMirrorGithubTransport, "ready")) blockers.push("git-mirror-github-token-secret-not-ready"); const argoInstall = record(argo.install); if (!boolField(argo, "installed")) blockers.push("argocd-not-installed"); else if (!boolField(argoInstall, "crdsReady")) blockers.push("argocd-crds-not-ready"); diff --git a/scripts/src/hwlab-node-impl.ts b/scripts/src/hwlab-node-impl.ts index d9092a13..53b871c3 100644 --- a/scripts/src/hwlab-node-impl.ts +++ b/scripts/src/hwlab-node-impl.ts @@ -2677,7 +2677,9 @@ function nodeRuntimeTriggerCurrent(scoped: ReturnType nodeRuntimeOpportunisticGitMirrorFlush(scoped, sourceCommit, pipelineRun), + }) : { ok: true, status: "already-succeeded", pipelineRun: before, polls: 0, elapsedMs: 0 }; const waitedPipelineRun = record(pipelineWait.pipelineRun); const pipelineFailureSummary = nodeRuntimePipelineFailureSummary(pipelineWait); @@ -2765,7 +2767,11 @@ function nodeRuntimeTriggerCurrent(scoped: ReturnType nodeRuntimeOpportunisticGitMirrorFlush(scoped, sourceCommit, pipelineRun), + }) + : null; const waitedPipelineRun = record(pipelineWait?.pipelineRun); const pipelineFailureSummary = nodeRuntimePipelineFailureSummary(pipelineWait); const postFlush = waitedPipelineRun.status === "True" @@ -2817,9 +2823,34 @@ function nodeRuntimeGitMirrorStatus(scoped: ReturnType/dev/null || true); ready=$(kubectl -n \"$1\" get deploy \"$2\" -o 'jsonpath={.status.readyReplicas}' 2>/dev/null || true); [ -n \"$desired\" ] && [ \"$desired\" -gt 0 ] 2>/dev/null && [ \"${ready:-0}\" = \"$desired\" ] && printf true || printf false; }", "exists_res() { kubectl -n \"$1\" get \"$2\" \"$3\" >/dev/null 2>&1 && printf true || printf false; }", "endpoint_ready() { endpoints=$(kubectl -n \"$1\" get endpoints \"$2\" -o 'jsonpath={.subsets[*].addresses[*].ip}' 2>/dev/null || true); [ -n \"$endpoints\" ] && printf true || printf false; }", + "github_transport_json=$(python3 - \"$github_transport_mode\" \"$namespace\" \"$github_token_secret\" \"$github_token_key\" \"$github_token_source_ref\" \"$github_token_source_key\" <<'PY'", + "import hashlib, json, subprocess, sys", + "mode, namespace, secret_name, secret_key, source_ref, source_key = sys.argv[1:7]", + "if mode != 'https':", + " print(json.dumps({'mode': mode, 'required': False, 'ready': True, 'valuesPrinted': False}))", + " raise SystemExit(0)", + "proc = subprocess.run(['kubectl', '-n', namespace, 'get', 'secret', secret_name, '-o', 'json'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)", + "exists = proc.returncode == 0", + "data = {}", + "if exists:", + " try:", + " data = json.loads(proc.stdout).get('data') or {}", + " except Exception:", + " data = {}", + "encoded = data.get(secret_key) if isinstance(data, dict) else None", + "present = isinstance(encoded, str) and len(encoded) > 0", + "fingerprint = 'sha256:' + hashlib.sha256(encoded.encode()).hexdigest()[:16] if present else None", + "print(json.dumps({'mode': mode, 'required': True, 'ready': exists and present, 'tokenSecretName': secret_name, 'tokenSecretKey': secret_key, 'tokenSourceRef': source_ref, 'tokenSourceKey': source_key, 'tokenSecretExists': exists, 'tokenKeyPresent': present, 'tokenKeyBytes': len(encoded) if present else 0, 'tokenFingerprint': fingerprint, 'valuesPrinted': False}))", + "PY", + ")", "summary_json=$(kubectl -n \"$namespace\" exec deploy/\"$read_deploy\" -- sh -lc '/etc/git-mirror/status.sh' 2>/tmp/hwlab-node-gitmirror-status.err || true)", "if [ -z \"$summary_json\" ]; then summary_json='{}'; fi", "read_deployment_ready=$(deploy_ready \"$namespace\" \"$read_deploy\")", @@ -2831,10 +2862,11 @@ function nodeRuntimeGitMirrorStatus(scoped: ReturnType/dev/null 2>&1; then cache_host_path_exists=true; fi", - "SUMMARY_JSON=\"$summary_json\" read_deployment_ready=\"$read_deployment_ready\" write_deployment_ready=\"$write_deployment_ready\" read_service_exists=\"$read_service_exists\" write_service_exists=\"$write_service_exists\" read_endpoints_ready=\"$read_endpoints_ready\" write_endpoints_ready=\"$write_endpoints_ready\" cache_pvc_exists=\"$cache_pvc_exists\" cache_host_path=\"$cache_host_path\" cache_host_path_exists=\"$cache_host_path_exists\" node <<'NODE'", + "SUMMARY_JSON=\"$summary_json\" GITHUB_TRANSPORT_JSON=\"$github_transport_json\" read_deployment_ready=\"$read_deployment_ready\" write_deployment_ready=\"$write_deployment_ready\" read_service_exists=\"$read_service_exists\" write_service_exists=\"$write_service_exists\" read_endpoints_ready=\"$read_endpoints_ready\" write_endpoints_ready=\"$write_endpoints_ready\" cache_pvc_exists=\"$cache_pvc_exists\" cache_host_path=\"$cache_host_path\" cache_host_path_exists=\"$cache_host_path_exists\" node <<'NODE'", "const summary = (() => { try { return JSON.parse(process.env.SUMMARY_JSON || '{}'); } catch { return {}; } })();", + "const githubTransport = (() => { try { return JSON.parse(process.env.GITHUB_TRANSPORT_JSON || '{}'); } catch { return {}; } })();", "const env = process.env;", - "const ok = env.read_deployment_ready === 'true' && env.write_deployment_ready === 'true' && env.read_service_exists === 'true' && env.write_service_exists === 'true' && env.read_endpoints_ready === 'true' && env.write_endpoints_ready === 'true' && (env.cache_pvc_exists === 'true' || env.cache_host_path_exists === 'true') && summary.localSource;", + "const ok = env.read_deployment_ready === 'true' && env.write_deployment_ready === 'true' && env.read_service_exists === 'true' && env.write_service_exists === 'true' && env.read_endpoints_ready === 'true' && env.write_endpoints_ready === 'true' && (env.cache_pvc_exists === 'true' || env.cache_host_path_exists === 'true') && githubTransport.ready !== false && summary.localSource;", "console.log(JSON.stringify({", " ok: Boolean(ok),", " resources: {", @@ -2848,6 +2880,7 @@ function nodeRuntimeGitMirrorStatus(scoped: ReturnType, - phase: "pre" | "post", + phase: "pre" | "post" | "parallel", sourceCommit: string, pipelineRun: string | null, statusInput: Record | null = null, @@ -3214,6 +3265,16 @@ function nodeRuntimeEnsureGitMirrorFlushed( }; } +function nodeRuntimeOpportunisticGitMirrorFlush( + scoped: ReturnType, + sourceCommit: string, + pipelineRun: string, +): Record | null { + const status = nodeRuntimeGitMirrorStatus({ ...scoped, action: "status", dryRun: true, confirm: false }); + if (status.ok !== true || !nodeRuntimeGitMirrorNeedsFlush(status)) return null; + return nodeRuntimeEnsureGitMirrorFlushed(scoped, "parallel", sourceCommit, pipelineRun, status); +} + function nodeRuntimeGitMirrorNeedsFlush(status: Record): boolean { const summary = record(status.summary); const localGitops = typeof summary.localGitops === "string" ? summary.localGitops : null; @@ -5634,11 +5695,18 @@ function createNodeRuntimePipelineRun(spec: HwlabRuntimeLaneSpec, sourceCommit: return runNodeK3sScript(spec, script, timeoutSeconds); } -function waitForNodeRuntimePipelineRunTerminal(spec: HwlabRuntimeLaneSpec, pipelineRun: string, timeoutSeconds: number): Record { +function waitForNodeRuntimePipelineRunTerminal( + spec: HwlabRuntimeLaneSpec, + pipelineRun: string, + timeoutSeconds: number, + options: { opportunisticPostFlush?: () => Record | null } = {}, +): Record { const startedAt = Date.now(); const deadline = startedAt + timeoutSeconds * 1000; let polls = 0; let last: Record = { exists: false, name: pipelineRun }; + let lastOpportunisticPostFlushAt = 0; + const opportunisticPostFlushes: Record[] = []; printNodeRuntimeTriggerProgress(spec, { stage: "pipelinerun-wait", status: "started", pipelineRun, timeoutSeconds }); while (Date.now() <= deadline) { polls += 1; @@ -5659,9 +5727,15 @@ function waitForNodeRuntimePipelineRunTerminal(spec: HwlabRuntimeLaneSpec, pipel failureSummary, polls, elapsedMs: Date.now() - startedAt, + opportunisticPostFlushes: opportunisticPostFlushes.length > 0 ? opportunisticPostFlushes : undefined, degradedReason: ok ? undefined : "node-runtime-pipelinerun-failed", }; } + if (options.opportunisticPostFlush !== undefined && Date.now() - lastOpportunisticPostFlushAt >= 15_000) { + lastOpportunisticPostFlushAt = Date.now(); + const flush = options.opportunisticPostFlush(); + if (flush !== null) opportunisticPostFlushes.push(flush); + } sleepSync(Math.min(10_000, Math.max(1000, deadline - Date.now()))); } printNodeRuntimeTriggerProgress(spec, { stage: "pipelinerun-wait", status: "timeout", pipelineRun, polls, elapsedMs: Date.now() - startedAt }); @@ -5675,6 +5749,7 @@ function waitForNodeRuntimePipelineRunTerminal(spec: HwlabRuntimeLaneSpec, pipel failureSummary, polls, elapsedMs: Date.now() - startedAt, + opportunisticPostFlushes: opportunisticPostFlushes.length > 0 ? opportunisticPostFlushes : undefined, degradedReason: "node-runtime-pipelinerun-wait-timeout", }; }