Merge pull request #331 from pikasTech/fix/330-runner-associated-secret-delete

修复 runner retention 关联 Secret 删除失败

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<number> {
+async function deleteAssociatedResources(options: RunnerRetentionOptions, runner: RunnerResourceEntry): Promise<number> {
   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<K8sObject[]> {
+  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 } : {}) });
 }

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;

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<void> {
+  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<void> {
+  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<void> {
+  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<void> {
+  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<void> {
   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}`;

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`);