From be736f21b9301e8ae736f037f41e1c372c38763a Mon Sep 17 00:00:00 2001 From: root Date: Sun, 12 Jul 2026 18:21:48 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=8C=89=E5=BD=92=E5=B1=9E=E6=B8=85?= =?UTF-8?q?=E7=90=86=20runner=20=E5=85=B3=E8=81=94=20Secret?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mgr/runner-retention.ts | 68 ++++++++++++-- src/selftest/cases/20-runner-k8s-job.ts | 6 +- .../22-runner-retention-stale-pending.ts | 92 ++++++++++++++++++- .../fake-kubectl-retention-stale-pending.mjs | 11 +++ 4 files changed, 167 insertions(+), 10 deletions(-) diff --git a/src/mgr/runner-retention.ts b/src/mgr/runner-retention.ts index dd990bb..ae3ed85 100644 --- a/src/mgr/runner-retention.ts +++ b/src/mgr/runner-retention.ts @@ -196,7 +196,7 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu deletedRunnerJobCount += retired.deletedRunnerJobCount; deletedRunnerPodCount += retired.deletedRunnerPodCount; repairedTerminalClaimCount += retired.repairedTerminalClaimCount; - if (item.resourceKind === "job") deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName); + if (item.resourceKind === "job") deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item); } const after = selected.length > 0 ? await runnerSnapshot(input.store, input.options) : before; if (after.liveRunnerCount > targetBeforeCreate) { @@ -277,7 +277,7 @@ export async function reconcileStalePendingRunners(input: { store: AgentRunStore terminalizedCommandCount += item.candidateKind === "stale-pending" ? 1 : 0; deletedRunnerJobCount += retired.deletedRunnerJobCount; deletedRunnerPodCount += retired.deletedRunnerPodCount; - const deletedAssociatedForRunner = item.resourceKind === "job" ? await deleteAssociatedResources(input.options, item.jobName) : 0; + const deletedAssociatedForRunner = item.resourceKind === "job" ? await deleteAssociatedResources(input.options, item) : 0; deletedAssociatedResourceCount += deletedAssociatedForRunner; outcomes.push({ ...resourceIdentitySummary(item), state: "retired", terminalized: item.candidateKind === "stale-pending", ...retired, deletedAssociatedResourceCount: deletedAssociatedForRunner, valuesPrinted: false }); } catch (error) { @@ -695,17 +695,69 @@ async function kubernetesGetPodEvents(options: RunnerRetentionOptions): Promise< return Array.isArray(items) ? items.filter((entry) => typeof entry === "object" && entry !== null && !Array.isArray(entry)).map((entry) => entry as unknown as K8sObject) : []; } -async function deleteAssociatedResources(options: RunnerRetentionOptions, jobName: string): Promise { +async function deleteAssociatedResources(options: RunnerRetentionOptions, runner: RunnerResourceEntry): Promise { let deleted = 0; - const selector = `agentrun.pikastech.local/runner-job=${jobName}`; - for (const resource of ["secret", "pvc"] as const) { - const result = await retentionKubernetes(options).deleteCollection(resource, { labelSelector: selector, ignoreNotFound: true, propagationPolicy: "Background" }); - if (result.code !== 0) throw new AgentRunError("infra-failed", `Kubernetes delete associated ${resource} failed`, { httpStatus: 502, details: retentionFailureDetails(result, { reason: "runner-associated-resource-delete-failed", namespace: options.namespace, resource, jobName }) }); - deleted += result.deletedCount ?? countDeletedLines(result.stdout); + const selector = `agentrun.pikastech.local/runner-job=${runner.jobName}`; + const secrets = await kubernetesGetAssociatedResources(options, "secret", selector, runner.jobName); + for (const secret of secrets) { + const name = objectName(secret); + if (!name || !isExclusiveRunnerSecret(secret, runner)) { + throw new AgentRunError("infra-failed", "runner retention associated Secret ownership is not exclusive", { + httpStatus: 409, + details: { + reason: "runner-associated-resource-ownership-unknown", + namespace: options.namespace, + resource: "secret", + name, + jobName: runner.jobName, + valuesPrinted: false, + }, + }); + } + const result = await retentionKubernetes(options).delete("secret", name, { ignoreNotFound: true, propagationPolicy: "Background" }); + if (result.code !== 0) { + throw new AgentRunError("infra-failed", "Kubernetes delete associated secret failed", { + httpStatus: 502, + details: retentionFailureDetails(result, { reason: "runner-associated-resource-delete-failed", namespace: options.namespace, resource: "secret", name, jobName: runner.jobName }), + }); + } + if (result.transport === "in-cluster-https" ? result.kubernetesReason !== "NotFound" : countDeletedLines(result.stdout) > 0) deleted++; } return deleted; } +async function kubernetesGetAssociatedResources(options: RunnerRetentionOptions, resource: "secret", selector: string, jobName: string): Promise { + const result = await retentionKubernetes(options).list(resource, { labelSelector: selector }); + if (result.code !== 0) { + throw new AgentRunError("infra-failed", `Kubernetes get associated ${resource} failed`, { + httpStatus: 502, + details: retentionFailureDetails(result, { reason: "runner-associated-resource-observation-failed", namespace: options.namespace, resource, jobName }), + }); + } + const parsed = parseJsonObject(result.stdout, `Kubernetes get associated ${resource}`, "runner-associated-resource-observation-failed"); + const items = parsed.items; + return Array.isArray(items) + ? items.filter((entry) => typeof entry === "object" && entry !== null && !Array.isArray(entry)).map((entry) => entry as unknown as K8sObject) + : []; +} + +function isExclusiveRunnerSecret(secret: K8sObject, runner: RunnerResourceEntry): boolean { + const labels = secret.metadata?.labels ?? {}; + const annotations = secret.metadata?.annotations ?? {}; + const identity = { + "agentrun.pikastech.local/run-id": runner.runId ?? "", + "agentrun.pikastech.local/command-id": runner.commandId ?? "", + "agentrun.pikastech.local/runner-id": runner.runnerId ?? "", + "agentrun.pikastech.local/runner-job": runner.jobName, + }; + if (labels["app.kubernetes.io/component"] !== "runner-transient-env") return false; + if (labels["agentrun.pikastech.local/runner-job"] !== runner.jobName) return false; + if (annotations["agentrun.pikastech.local/runner-job"] !== runner.jobName) return false; + if (!resourceIdentityMatches(identity, annotations)) return false; + const owners = secret.metadata?.ownerReferences ?? []; + return owners.length === 0 || owners.every((owner) => owner.kind === "Job" && owner.name === runner.jobName); +} + function retentionKubernetes(options: RunnerRetentionOptions): KubernetesClient { return createKubernetesClient({ namespace: options.namespace, ...(options.kubectlCommand ? { kubectlCommand: options.kubectlCommand } : {}), ...(options.signal ? { signal: options.signal } : {}) }); } diff --git a/src/selftest/cases/20-runner-k8s-job.ts b/src/selftest/cases/20-runner-k8s-job.ts index 0134d03..929a988 100644 --- a/src/selftest/cases/20-runner-k8s-job.ts +++ b/src/selftest/cases/20-runner-k8s-job.ts @@ -361,6 +361,10 @@ if (args[0] === "get" && args[1] === "pods") { console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items: [] })); process.exit(0); } +if (args[0] === "get" && args[1] === "secrets") { + console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items: [] })); + process.exit(0); +} if (args[0] === "get" && args[1] === "events") { console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items: [] })); process.exit(0); } if (args[0] === "delete") { const deleted = await readDeleted(); @@ -444,7 +448,7 @@ process.exit(1); assert.match(String(metadata.name), /^agentrun-v02-runner-[a-f0-9]{12}$/u); const deletedItems = JSON.parse(await readFile(retentionDeleted, "utf8")) as JsonRecord[]; assert.equal(deletedItems.filter((item) => item.resource === "job").length, 2); - assert.ok(deletedItems.some((item) => String((item.args as string[]).join(" ")).includes("agentrun.pikastech.local/runner-job=agentrun-v02-runner-old-a")), "old runner associated resources should be selected by runner-job label"); + assert.equal(retention.deletedAssociatedResourceCount, 0); assertNoSecretLeak(retentionCreated); } finally { if (previousRetentionDbFacts === undefined) delete process.env.AGENTRUN_SELFTEST_RETENTION_DB_FACTS; diff --git a/src/selftest/cases/22-runner-retention-stale-pending.ts b/src/selftest/cases/22-runner-retention-stale-pending.ts index a2d7e0e..baef697 100644 --- a/src/selftest/cases/22-runner-retention-stale-pending.ts +++ b/src/selftest/cases/22-runner-retention-stale-pending.ts @@ -39,6 +39,8 @@ interface RetentionFixture { bumpedJobResourceVersion?: string; deleteErrorResources?: string[]; retainDeletedJobs?: boolean; + secrets?: JsonRecord[]; + secretDisappearsBeforeDelete?: boolean; } class RetentionFactStore extends MemoryAgentRunStore { @@ -128,6 +130,10 @@ const selfTest: SelfTestCase = async (context) => { await assertPodCasRaceFailsClosed({ fixturePath, statePath, kubectlCommand }); await assertDeleteFailureFailsClosed({ fixturePath, statePath, kubectlCommand }); await assertDeletionPostconditionFailsClosed({ fixturePath, statePath, kubectlCommand }); + await assertAssociatedSecretCleanup({ fixturePath, statePath, kubectlCommand }); + await assertAssociatedSecretNotFoundIsIdempotent({ fixturePath, statePath, kubectlCommand }); + await assertAssociatedSecretDeleteFailureIsTyped({ fixturePath, statePath, kubectlCommand }); + await assertAssociatedSecretUnknownOwnershipFailsClosed({ fixturePath, statePath, kubectlCommand }); await assertMemoryDbFenceRejectsConcurrentClaim(); await assertRunnerCreateFenceSerializesNamespace(); await assertPeriodicReconcilerCleansAtCapacity({ fixturePath, statePath, kubectlCommand }); @@ -171,6 +177,10 @@ const selfTest: SelfTestCase = async (context) => { "pod-resource-version-cas-fails-closed", "runner-delete-cas-exit-code-checked", "runner-delete-postcondition-checked", + "runner-associated-secret-delete-by-exclusive-identity", + "runner-associated-secret-not-found-is-idempotent", + "runner-associated-secret-delete-failure-is-typed", + "runner-associated-secret-unknown-ownership-fails-closed", "memory-db-fence-rejects-concurrent-claim", "namespace-runner-create-fence-prevents-capacity-overshoot", "periodic-reconciler-cleans-stale-pending-at-capacity", @@ -214,7 +224,7 @@ async function assertTwentyStalePendingSelectsOnlyOldest(input: { fixturePath: s assert.equal(cleanup.selectedRunnerCount, 1); assert.equal(cleanup.deletedRunnerJobCount, 1); assert.equal(cleanup.deletedRunnerPodCount, 0); - assert.equal(cleanup.deletedAssociatedResourceCount, 2); + assert.equal(cleanup.deletedAssociatedResourceCount, 0); const selected = firstSummary(cleanup, "selected"); assert.equal(selected.name, "agentrun-v02-runner-stale-00"); assert.equal(selected.candidateKind, "stale-pending"); @@ -515,6 +525,64 @@ async function assertDeletionPostconditionFailsClosed(input: { fixturePath: stri assert.deepEqual(deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "job").map((item) => item.name), ["agentrun-v02-runner-stale-93"]); } +async function assertAssociatedSecretCleanup(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(107, new Date(Date.now() - 60 * 60_000).toISOString()); + const job = runnerJob(fact, 107); + await writeFixture(input, { + jobs: [job], + pods: [runnerPod(fact, 107, "failed-config")], + events: [], + secrets: [runnerTransientEnvSecret(fact, 107)], + }); + const summary = await enforceRunnerRetentionBeforeCreate({ store: new RetentionFactStore([fact]), options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1 }); + assert.equal(summary.deletedAssociatedResourceCount, 1); + assert.deepEqual(deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "secret").map((item) => item.name), ["agentrun-v02-runner-env-107"]); +} + +async function assertAssociatedSecretNotFoundIsIdempotent(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(108, new Date(Date.now() - 60 * 60_000).toISOString()); + await writeFixture(input, { + jobs: [runnerJob(fact, 108)], + pods: [runnerPod(fact, 108, "failed-config")], + events: [], + secrets: [runnerTransientEnvSecret(fact, 108)], + secretDisappearsBeforeDelete: true, + }); + const summary = await enforceRunnerRetentionBeforeCreate({ store: new RetentionFactStore([fact]), options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1 }); + assert.equal(summary.deletedAssociatedResourceCount, 0); +} + +async function assertAssociatedSecretDeleteFailureIsTyped(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(109, new Date(Date.now() - 60 * 60_000).toISOString()); + await writeFixture(input, { + jobs: [runnerJob(fact, 109)], + pods: [runnerPod(fact, 109, "failed-config")], + events: [], + secrets: [runnerTransientEnvSecret(fact, 109)], + deleteErrorResources: ["secret"], + }); + const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({ store: new RetentionFactStore([fact]), options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1 })); + assert.equal(error.details?.reason, "runner-associated-resource-delete-failed"); + assert.equal(error.details?.resource, "secret"); + assert.equal(error.details?.exitCode, 42); +} + +async function assertAssociatedSecretUnknownOwnershipFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise { + const fact = pendingFact(110, new Date(Date.now() - 60 * 60_000).toISOString()); + const shared = runnerTransientEnvSecret(fact, 110); + ((shared.metadata as JsonRecord).labels as JsonRecord)["app.kubernetes.io/component"] = "provider-profile"; + await writeFixture(input, { + jobs: [runnerJob(fact, 110)], + pods: [runnerPod(fact, 110, "failed-config")], + events: [], + secrets: [shared], + }); + const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({ store: new RetentionFactStore([fact]), options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1 })); + assert.equal(error.details?.reason, "runner-associated-resource-ownership-unknown"); + assert.equal(error.details?.resource, "secret"); + assert.equal(deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "secret").length, 0); +} + async function assertMemoryDbFenceRejectsConcurrentClaim(): Promise { const store = new MemoryAgentRunStore(); const run = store.createRun({ tenantId: "unidesk", projectId: "pikasTech/agentrun", workspaceRef: { kind: "host-path", path: "/tmp/agentrun-retention-memory" }, providerId: "NC01", backendProfile: "codex", executionPolicy: { sandbox: "workspace-write", approval: "never", timeoutMs: 60_000, network: "default", secretScope: { allowCredentialEcho: false, providerCredentials: [] } }, traceSink: null }); @@ -950,6 +1018,28 @@ function runnerPod(fact: RetentionFact, index: number, state: "pending" | "runni }; } +function runnerTransientEnvSecret(fact: RetentionFact, index: number): JsonRecord { + const suffix = String(index).padStart(2, "0"); + const jobName = `agentrun-v02-runner-stale-${suffix}`; + return { + apiVersion: "v1", + kind: "Secret", + metadata: { + name: `agentrun-v02-runner-env-${suffix}`, + namespace: "agentrun-nc01-v02", + labels: { + "app.kubernetes.io/component": "runner-transient-env", + "agentrun.pikastech.local/runner-job": jobName, + }, + annotations: { + ...identityAnnotations(fact), + "agentrun.pikastech.local/runner-job": jobName, + }, + ownerReferences: [{ kind: "Job", name: jobName }], + }, + }; +} + function failedMountEvent(fact: RetentionFact, index: number): JsonRecord { const suffix = String(index).padStart(2, "0"); const jobName = `agentrun-v02-runner-stale-${suffix}`; diff --git a/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs b/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs index c589587..1190f01 100755 --- a/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs +++ b/src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs @@ -30,6 +30,7 @@ const deletedJobs = fixture.retainDeletedJobs === true ? new Set() : new Set(state.filter((item) => item.resource === "job").map((item) => item.name)); const deletedPods = new Set(state.filter((item) => item.resource === "pod").map((item) => item.name)); +const deletedSecrets = new Set(state.filter((item) => item.resource === "secret").map((item) => item.name)); function currentJob(name) { const base = fixture.jobs.find((item) => item?.metadata?.name === name); @@ -64,6 +65,12 @@ if (args[0] === "get" && args[1] === "pods") { process.exit(0); } +if (args[0] === "get" && args[1] === "secrets") { + const items = (fixture.secrets ?? []).filter((item) => !deletedSecrets.has(item?.metadata?.name)); + console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items })); + process.exit(0); +} + if (args[0] === "get" && args[1] === "events") { if (fixture.eventsError === true) { console.error("events access denied by self-test fixture"); @@ -124,6 +131,10 @@ if (args[0] === "delete") { console.error(`self-test forced ${resource} deletion failure`); process.exit(42); } + if (resource === "secret" && fixture.secretDisappearsBeforeDelete === true) { + console.error(`Error from server (NotFound): secret/${name} was not found`); + process.exit(args.includes("--ignore-not-found=true") ? 0 : 1); + } state.push({ resource, name, selector, valuesPrinted: false }); await Bun.write(statePath, `${JSON.stringify(state, null, 2)}\n`); console.log(`${resource}/${name.replace(/[^a-z0-9.-]+/giu, "-")} deleted`);