diff --git a/Fournos_Design_Document.md b/Fournos_Design_Document.md index 7cbdc5d..abc3aa5 100644 --- a/Fournos_Design_Document.md +++ b/Fournos_Design_Document.md @@ -55,7 +55,7 @@ Jobs are submitted as `FournosJob` custom resources ([manifests/crd.yaml](manife | `spec.displayName` | no | Human-readable job name (defaults to `metadata.name`) | | `spec.pipeline` | no | Tekton Pipeline name (default: `fournos-full`) | | `spec.priority` | no | Kueue WorkloadPriorityClass name | -| `spec.secretRefs` | no | Vault-synced K8s Secret names (`vault-`) to mount into the pipeline. Populated by Forge during the Resolving phase. Each must be a K8s Secret with `fournos.dev/vault-entry=true`. | +| `spec.secretRefs` | no | Vault-synced K8s Secret names (`vault-`) to mount into the pipeline. Populated by Forge during the Resolving phase. Each must be a K8s Secret with `fournos.dev/vault-entry=true` in `FOURNOS_SECRETS_NAMESPACE`. During the Admitted phase the operator copies them into the operator namespace and mounts them as a projected volume at `/var/run/secrets/fournos//`. | | `spec.exclusive` | no | If `true`, locks the target cluster so no other FournosJob can run there. Requires `spec.cluster`. | | `spec.shutdown` | no | Shutdown action: `Stop` (graceful, runs finally tasks) or `Terminate` (immediate, skips finally tasks). Both wait for the PipelineRun to finish before releasing Kueue quota. | @@ -192,7 +192,7 @@ sequenceDiagram 1. **on_create**: Operator validates the spec (cluster exists if specified, `exclusive` requires `cluster`). If `spec.shutdown` is set (`Stop` or `Terminate`), immediately sets `phase=Stopped`. Otherwise sets `phase=Resolving`. 2. **timer (Resolving)**: Launches a Forge resolve K8s Job that patches the FournosJob spec with `hardware` (if not user-provided) and `secretRefs`. Polls the Job for completion. On success, reads the FournosJob spec, validates hardware (GPU type checked against Kueue), validates `secretRefs` against Vault secrets, creates the Kueue Workload (exclusive jobs request all 100 `fournos/cluster-slot` units; normal jobs request 1), and sets `phase=Pending`. Failed resolve Jobs are preserved for debugging. 3. **timer (Pending)**: Polls the Workload for Kueue admission. On admission, extracts the assigned cluster and sets `phase=Admitted`. -4. **timer (Admitted)**: Reads `secretRefs` from the FournosJob spec, resolves the kubeconfig Secret, creates the Tekton PipelineRun with `ownerReferences` pointing at the FournosJob, sets `phase=Running`. +4. **timer (Admitted)**: Reads `secretRefs` from the FournosJob spec, copies each referenced secret from `secrets_namespace` into the operator namespace (per-job name `-`, with `ownerReferences` for automatic cleanup), resolves the kubeconfig Secret, creates the Tekton PipelineRun with a projected `vault-secrets` volume mounting all copied secrets at `/var/run/secrets/fournos//` and `ownerReferences` pointing at the FournosJob, sets `phase=Running`. 5. **timer (Running)**: Polls the PipelineRun for completion. On success/failure, deletes the Workload and sets `phase=Succeeded` or `phase=Failed`. 6. **timer (any non-terminal phase, shutdown)**: If `spec.shutdown` is set (`Stop` or `Terminate`) and the job has a PipelineRun (Admitted/Running), the timer cancels the PipelineRun — `Stop` uses Tekton's `CancelledRunFinally` (runs `finally` tasks), `Terminate` uses `Cancelled` (skips `finally` tasks) — and sets `phase=Stopping`. The Workload is **not** deleted yet — it stays alive to hold the cluster slot while the PipelineRun winds down. If no PipelineRun exists (Pending), the Workload is deleted immediately and the job goes straight to `phase=Stopped`. 7. **timer (Stopping)**: Polls the PipelineRun until it reaches a terminal state (`succeeded` or `failed`). Once complete, deletes the Workload to release Kueue quota and sets `phase=Stopped`. @@ -359,7 +359,7 @@ fournos/ execution.py # reconcile_admitted, reconcile_running core/ constants.py # Shared label keys, Phase enum, cluster-slot constants - clusters.py # ClusterRegistry (kubeconfig lookup, secretRef resolution) + clusters.py # ClusterRegistry (kubeconfig lookup, secretRef resolution, cross-namespace secret copying) resolve.py # ResolveClient (Forge resolve Job management) tekton.py # TektonClient (PipelineRun CRUD) kueue.py # KueueClient (Workload CRUD, admission checks, cluster-slot requests) @@ -406,11 +406,11 @@ README.md - **Stateless operator** — all job state lives in Kubernetes resources (FournosJob CRs, PipelineRuns, Workloads), not in memory. Crash-safe via `on_resume`. - **Timer-based reconciliation** — the operator polls Workload admission and PipelineRun completion via a kopf timer (5s interval), eliminating the need for callback tasks or watch streams on third-party resources - **Operator cleans up on completion** — Kueue Workloads are deleted when the PipelineRun reaches a terminal state, releasing quota without relying on external callbacks -- **ownerReferences for cascade deletion** — Workloads and PipelineRuns carry `ownerReferences` pointing at their FournosJob, so Kubernetes automatically cascade-deletes them when the job is removed +- **ownerReferences for cascade deletion** — Workloads, PipelineRuns, resolve Jobs, and copied secrets all carry `ownerReferences` pointing at their FournosJob, so Kubernetes automatically cascade-deletes them when the job is removed - **Exclusive locking via Kueue semaphore** — each cluster flavor has 100 `fournos/cluster-slot` units. Normal jobs request 1 slot; exclusive jobs request all 100. Kueue enforces mutual exclusion atomically — no operator-level blocking, labels, or in-memory state needed. Hardware-only jobs are automatically steered to clusters with available slots. - **Shutdown via spec field** — the `spec.shutdown` enum supports two modes: `Stop` (Tekton `CancelledRunFinally` — runs `finally` cleanup tasks) and `Terminate` (Tekton `Cancelled` — skips `finally` tasks). Both transition to an intermediate `Stopping` phase while the PipelineRun winds down. The Workload (and its quota) is kept alive until the PipelineRun completes, ensuring the cluster slot is not released prematurely. Only then does the operator delete the Workload and set `phase=Stopped`. The enum is extensible for future shutdown strategies. The FournosJob stays around in `Stopped` phase for inspection, unlike deletion which cascades and removes the record. - **Mandatory Resolving phase** — every job passes through a `Resolving` phase before entering `Pending`. During this phase, a Forge K8s Job runs to determine hardware requirements (`gpuType`, `gpuCount`) and secret references (`secretRefs`). Forge patches these values directly into the FournosJob spec (hardware only when not already user-provided). The operator validates them after the Job completes. Failed resolve Jobs are preserved for debugging. -- **Vault-based secret management** — pipeline secrets originate in a HashiCorp Vault and are synced to K8s Secrets on demand via `hacks/sync_vault_secrets.py` into a dedicated secrets namespace (`FOURNOS_SECRETS_NAMESPACE`, default `psap-secrets`). The K8s Secret name uses a `vault-` prefix followed by the Vault entry name (e.g. `vault-my-creds`); entries that are not valid DNS-1123 names are rejected during sync. Each synced Secret carries a `fournos.dev/vault-entry=true` label. Secret references are populated by Forge during the Resolving phase on `spec.secretRefs`. The operator validates them in the secrets namespace before creating the Workload (during Resolving) and reads them from spec when creating the PipelineRun (during Admitted). Missing or non-vault refs fail the job immediately rather than creating a broken PipelineRun. +- **Vault-based secret management with cross-namespace injection** — pipeline secrets originate in a HashiCorp Vault and are synced to K8s Secrets on demand via `hacks/sync_vault_secrets.py` into a dedicated secrets namespace (`FOURNOS_SECRETS_NAMESPACE`, default `psap-secrets`). The K8s Secret name uses a `vault-` prefix followed by the Vault entry name (e.g. `vault-my-creds`); entries that are not valid DNS-1123 names are rejected during sync. Each synced Secret carries a `fournos.dev/vault-entry=true` label. Secret references are populated by Forge during the Resolving phase on `spec.secretRefs`. The operator validates them in the secrets namespace before creating the Workload (during Resolving). During the Admitted phase, the operator copies each referenced secret from the secrets namespace into the operator namespace with a per-job name (`-`) and `ownerReferences` to the FournosJob for automatic cleanup. The copies are combined into a single projected volume (`vault-secrets`) mounted at `/var/run/secrets/fournos//`, with each secret's keys placed under a subdirectory matching its original name. This avoids key collisions across secrets and works with a static `volumeMount` in the Task YAML regardless of how many secrets a job uses. An empty projected volume (no secrets) is always emitted so the static mount never fails. Missing or non-vault refs fail the job during Resolving rather than creating a broken PipelineRun. - **Multiple pipelines** — `fournos-full` (prepare → run → cleanup) and `fournos-run-only` (run only), selectable per job - **Target clusters need nothing installed** — FORGE runs on the hub cluster inside Tekton Task pods and communicates with targets via remote `oc`/`kubectl` commands through kubeconfig Secrets (stored in the dedicated secrets namespace) diff --git a/README.md b/README.md index 6c343f2..1797c7c 100644 --- a/README.md +++ b/README.md @@ -85,7 +85,7 @@ oc delete FournosJob -n $FOURNOS_NAMESPACE # cleanup | `spec.displayName` | no | Human-readable job name (defaults to `metadata.name`) | | `spec.pipeline` | no | Tekton Pipeline name (default: `fournos-full`) | | `spec.priority` | no | Kueue WorkloadPriorityClass name | -| `spec.secretRefs` | no | Vault-synced K8s Secret names (prefixed with `vault-`) to mount into the pipeline. Populated by Forge during the Resolving phase. The operator verifies each name as a K8s Secret with the `fournos.dev/vault-entry=true` label. | +| `spec.secretRefs` | no | Vault-synced K8s Secret names (prefixed with `vault-`) to mount into the pipeline. Populated by Forge during the Resolving phase. The operator validates each name in `FOURNOS_SECRETS_NAMESPACE`, copies the secrets into the operator namespace, and mounts them as a projected volume at `/var/run/secrets/fournos//`. | | `spec.exclusive` | no | If `true`, locks the target cluster so no other FournosJob can run there. Requires `spec.cluster`. | | `spec.shutdown` | no | Shutdown action: `Stop` cancels gracefully (Tekton `CancelledRunFinally` — runs `finally` tasks); `Terminate` cancels immediately (Tekton `Cancelled` — skips `finally` tasks). Both wait for the PipelineRun to finish before releasing Kueue quota. | @@ -269,9 +269,25 @@ make sync-vault-secrets-dry-run # preview only The synced secrets are labelled `fournos.dev/vault-entry=true` and `app.kubernetes.io/managed-by=fournos-vault-sync` for easy identification. Secret references are populated by Forge during the Resolving phase directly -on the FournosJob `spec.secretRefs` field. The operator verifies each +on the FournosJob `spec.secretRefs` field. The operator validates each referenced Secret exists in the secrets namespace and carries the vault -label before proceeding. +label during the Resolving phase, then copies them into the operator +namespace during the Admitted phase and mounts them as a projected volume +into the PipelineRun pods. Each secret's keys are placed under a +subdirectory matching the original name: + +``` +/var/run/secrets/fournos/ + vault-my-creds/ + username + password + vault-other-creds/ + token +``` + +Copied secrets are named `-` and carry +`ownerReferences` back to the FournosJob, so Kubernetes garbage-collects +them automatically when the job is deleted. ## Configuration @@ -304,7 +320,7 @@ The operator runs as a single-replica Deployment using 1. **Resolves** job requirements by launching a Forge K8s Job that populates the FournosJob spec with GPU type/count and secret references 2. **Creates** a Kueue Workload with the resolved GPU resources (owned by the FournosJob via `ownerReferences`) 3. **Polls** (5 s timer) for Kueue admission and assigned cluster -4. **Launches** a Tekton PipelineRun with FORGE parameters (owned by the FournosJob via `ownerReferences`) +4. **Copies** referenced Vault secrets from the secrets namespace into the operator namespace (per-job copies with `ownerReferences` for automatic cleanup) and **launches** a Tekton PipelineRun with FORGE parameters and the secrets mounted as a projected volume at `/var/run/secrets/fournos/` (owned by the FournosJob via `ownerReferences`) 5. **Watches** the PipelineRun until completion 6. **Deletes** the Workload to release Kueue quota diff --git a/config/forge/workflows/tasks.yaml b/config/forge/workflows/tasks.yaml index 1005112..abc9304 100644 --- a/config/forge/workflows/tasks.yaml +++ b/config/forge/workflows/tasks.yaml @@ -26,7 +26,7 @@ spec: imagePullPolicy: Always env: - name: KUBECONFIG - value: /workspace/kubeconfig/kubeconfig + value: /var/run/secrets/fournos-kubeconfig/kubeconfig - name: FORGE_CONFIG value: "$(params.forge-config)" - name: FOURNOS_ENV @@ -35,7 +35,10 @@ spec: value: "$(params.job-step)" volumeMounts: - name: kubeconfig - mountPath: /workspace/kubeconfig + mountPath: /var/run/secrets/fournos-kubeconfig + readOnly: true + - name: vault-secrets + mountPath: /var/run/secrets/fournos readOnly: true script: | #!/usr/bin/env bash diff --git a/config/fournos-validation/workflows/tasks.yaml b/config/fournos-validation/workflows/tasks.yaml index 25eb21e..1a5886b 100644 --- a/config/fournos-validation/workflows/tasks.yaml +++ b/config/fournos-validation/workflows/tasks.yaml @@ -16,10 +16,10 @@ spec: image: bitnami/kubectl:latest env: - name: KUBECONFIG - value: /workspace/kubeconfig/kubeconfig + value: /var/run/secrets/fournos-kubeconfig/kubeconfig volumeMounts: - name: kubeconfig - mountPath: /workspace/kubeconfig + mountPath: /var/run/secrets/fournos-kubeconfig readOnly: true script: | #!/usr/bin/env bash diff --git a/dev/job-secret-demo.yaml b/dev/job-secret-demo.yaml new file mode 100644 index 0000000..8b935a0 --- /dev/null +++ b/dev/job-secret-demo.yaml @@ -0,0 +1,13 @@ +apiVersion: fournos.dev/v1 +kind: FournosJob +metadata: + generateName: mock-secret-demo- +spec: + owner: dev + displayName: mock-secret-demo + cluster: cluster-1 + pipeline: fournos-run-only + forge: + project: secret-demo + args: + - demo diff --git a/dev/mock-pipelines/tasks.yaml b/dev/mock-pipelines/tasks.yaml index b47e151..34cbde8 100644 --- a/dev/mock-pipelines/tasks.yaml +++ b/dev/mock-pipelines/tasks.yaml @@ -51,6 +51,10 @@ spec: name: fournos-mock-config key: sleep optional: true + volumeMounts: + - name: vault-secrets + mountPath: /var/run/secrets/fournos + readOnly: true script: | #!/bin/sh MOCK_SLEEP="${MOCK_SLEEP:-3}" @@ -61,6 +65,8 @@ spec: echo "$(params.env)" | sed 's/^/ /' echo "[mock-run] gpu-count=$(params.gpu-count)" echo "[mock-run] job=$(params.job-name)" + echo "[mock-run] vault-secrets:" + find /var/run/secrets/fournos -type f 2>/dev/null | sort | while read f; do echo " $f ($(wc -c < "$f") bytes)"; done echo "[mock-run] simulating workload (${MOCK_SLEEP}s)..." sleep "$MOCK_SLEEP" echo "[mock-run] done" diff --git a/dev/mock-resolve/resolve.sh b/dev/mock-resolve/resolve.sh index f6f79f9..45e48b3 100644 --- a/dev/mock-resolve/resolve.sh +++ b/dev/mock-resolve/resolve.sh @@ -1,8 +1,10 @@ #!/usr/bin/env bash # Mock Forge resolve script — patches the FournosJob spec with resolved values. # -# Forge writes hardware (only when not user-provided) and secretRefs -# directly into the FournosJob spec. +# In production, Forge determines hardware requirements and secret +# references by inspecting the project. This mock sets hard-coded +# defaults for hardware only when the user hasn't provided them, and +# always sets secretRefs. # # Expected env vars (set by the operator): # FOURNOS_JOB_NAME — FournosJob name to patch @@ -22,25 +24,15 @@ if [[ -z "${EXISTING_HW}" ]]; then kubectl patch fournosjob "${FOURNOS_JOB_NAME}" \ -n "${FOURNOS_NAMESPACE}" \ --type=merge \ - -p '{ - "spec": { - "hardware": { - "gpuType": "a100", - "gpuCount": 2 - }, - "secretRefs": [] - } - }' + -p '{"spec":{"hardware":{"gpuType":"a100","gpuCount":2}}}' else - echo "[mock-resolve] user-provided hardware found (${EXISTING_HW}), patching secretRefs only" - kubectl patch fournosjob "${FOURNOS_JOB_NAME}" \ - -n "${FOURNOS_NAMESPACE}" \ - --type=merge \ - -p '{ - "spec": { - "secretRefs": [] - } - }' + echo "[mock-resolve] user-provided hardware found (${EXISTING_HW}), keeping" fi +echo "[mock-resolve] setting secretRefs" +kubectl patch fournosjob "${FOURNOS_JOB_NAME}" \ + -n "${FOURNOS_NAMESPACE}" \ + --type=merge \ + -p '{"spec":{"secretRefs":["vault-placeholder"]}}' + echo "[mock-resolve] done" diff --git a/dev/mock-secrets.yaml b/dev/mock-secrets.yaml index eedd9f6..fc11456 100644 --- a/dev/mock-secrets.yaml +++ b/dev/mock-secrets.yaml @@ -37,3 +37,16 @@ metadata: type: Opaque stringData: kubeconfig: "mock" + +--- +# Mock Vault-synced secret for the secret-demo sample job. +apiVersion: v1 +kind: Secret +metadata: + name: vault-placeholder + labels: + fournos.dev/vault-entry: "true" + app.kubernetes.io/managed-by: fournos-vault-sync +type: Opaque +stringData: + placeholder: "mock-secret-value" diff --git a/fournos/core/clusters.py b/fournos/core/clusters.py index eeb6162..588b5c5 100644 --- a/fournos/core/clusters.py +++ b/fournos/core/clusters.py @@ -1,13 +1,25 @@ +from __future__ import annotations + import logging +from dataclasses import dataclass from kubernetes import client -from fournos.core.constants import LABEL_VAULT_ENTRY +from fournos.core.constants import LABEL_MANAGED_BY, LABEL_VAULT_ENTRY from fournos.settings import settings logger = logging.getLogger(__name__) +@dataclass(frozen=True) +class ResolvedSecret: + """A secret that has been copied into the pod namespace.""" + + name: str + original_name: str + keys: list[str] + + class ClusterRegistry: def __init__(self, k8s_client: client.CoreV1Api) -> None: self._k8s = k8s_client @@ -30,8 +42,7 @@ def cluster_exists(self, cluster_name: str) -> bool: def _resolve_secret_ref(self, ref: str) -> str: """Verify that *ref* is a Vault-synced K8s Secret and return its name. - Vault-synced secrets use a ``vault-`` prefix: the K8s Secret - name is ``vault-``. The + The Secret is read from ``secrets_namespace``. The ``fournos.dev/vault-entry=true`` label is checked to confirm the Secret was actually imported from Vault. @@ -59,3 +70,67 @@ def _resolve_secret_ref(self, ref: str) -> str: def resolve_secret_refs(self, refs: list[str]) -> list[str]: """Resolve a list of secretRefs to their K8s Secret names.""" return [self._resolve_secret_ref(r) for r in refs] + + def copy_secret(self, ref: str, fjob_name: str, owner_ref: dict) -> ResolvedSecret: + """Copy a Vault-synced Secret from the secrets namespace into the pod namespace. + + The copy is named ``-`` and carries an ownerReference + back to the FournosJob so K8s GC cleans it up automatically. + Idempotent: a 409 (AlreadyExists) is silently ignored. + """ + source = self._k8s.read_namespaced_secret(ref, settings.secrets_namespace) + + labels = source.metadata.labels or {} + if labels.get(LABEL_VAULT_ENTRY) != "true": + raise KeyError( + f"Secret {ref!r} in {settings.secrets_namespace} is not a " + f"Vault-synced secret (missing {LABEL_VAULT_ENTRY}=true label)" + ) + + keys = sorted((source.data or {}).keys()) + copied_name = f"{fjob_name}-{ref}" + + copy_body = client.V1Secret( + metadata=client.V1ObjectMeta( + name=copied_name, + namespace=settings.namespace, + labels={ + LABEL_MANAGED_BY: "fournos", + LABEL_VAULT_ENTRY: "true", + }, + owner_references=[ + client.V1OwnerReference( + api_version=owner_ref["apiVersion"], + kind=owner_ref["kind"], + name=owner_ref["name"], + uid=owner_ref["uid"], + controller=False, + block_owner_deletion=True, + ) + ], + ), + type=source.type, + data=source.data, + ) + + try: + self._k8s.create_namespaced_secret(settings.namespace, copy_body) + logger.info( + "Copied secret %s from %s as %s", + ref, + settings.secrets_namespace, + copied_name, + ) + except client.exceptions.ApiException as exc: + if exc.status == 409: + logger.debug("Secret copy %s already exists (409)", copied_name) + else: + raise + + return ResolvedSecret(name=copied_name, original_name=ref, keys=keys) + + def copy_secrets( + self, refs: list[str], fjob_name: str, owner_ref: dict + ) -> list[ResolvedSecret]: + """Copy all *refs* and return the resolved list.""" + return [self.copy_secret(r, fjob_name, owner_ref) for r in refs] diff --git a/fournos/core/tekton.py b/fournos/core/tekton.py index 70e6b2c..56fcde8 100644 --- a/fournos/core/tekton.py +++ b/fournos/core/tekton.py @@ -8,6 +8,7 @@ from kubernetes import client +from fournos.core.clusters import ResolvedSecret from fournos.core.constants import LABEL_JOB_NAME, LABEL_MANAGED_BY from fournos.settings import settings @@ -35,6 +36,31 @@ def serialize_env(env: dict) -> str: return "".join(lines) +def _build_secrets_volume(resolved: list[ResolvedSecret]) -> dict: + """Build a single projected volume combining all per-job secret copies. + + Always returns a valid volume spec -- when *resolved* is empty the + ``sources`` list is empty, which produces an empty directory so + static volumeMounts in Task YAMLs remain valid. + """ + return { + "name": "vault-secrets", + "projected": { + "sources": [ + { + "secret": { + "name": r.name, + "items": [ + {"key": k, "path": f"{r.original_name}/{k}"} for k in r.keys + ], + }, + } + for r in resolved + ], + }, + } + + class TektonClient: def __init__(self, k8s_client: client.CustomObjectsApi) -> None: self._k8s = k8s_client @@ -50,7 +76,7 @@ def create_pipeline_run( env: dict, kubeconfig_secret: str, gpu_count: int, - secret_refs: list[str], + resolved_secrets: list[ResolvedSecret], cluster: str, owner_ref: dict | None = None, ) -> dict: @@ -69,6 +95,8 @@ def create_pipeline_run( if owner_ref: metadata["ownerReferences"] = [owner_ref] + secret_ref_names = [r.original_name for r in resolved_secrets] + body = { "apiVersion": f"{TEKTON_GROUP}/{TEKTON_VERSION}", "kind": "PipelineRun", @@ -79,6 +107,9 @@ def create_pipeline_run( "metadata": { "labels": labels, }, + "podTemplate": { + "volumes": [_build_secrets_volume(resolved_secrets)], + }, }, "params": [ {"name": "job-name", "value": display_name}, @@ -93,7 +124,7 @@ def create_pipeline_run( }, {"name": "kubeconfig-secret", "value": kubeconfig_secret}, {"name": "gpu-count", "value": str(gpu_count)}, - {"name": "secret-refs", "value": secret_refs}, + {"name": "secret-refs", "value": secret_ref_names}, ], }, } diff --git a/fournos/handlers/execution.py b/fournos/handlers/execution.py index 666a399..ac0be19 100644 --- a/fournos/handlers/execution.py +++ b/fournos/handlers/execution.py @@ -154,17 +154,20 @@ def reconcile_admitted(spec, name, namespace, status, patch, body): secret_refs_raw = spec.get("secretRefs") or [] try: - resolved_refs = ctx.registry.resolve_secret_refs(secret_refs_raw) - except KeyError as exc: + resolved_secrets = ctx.registry.copy_secrets( + secret_refs_raw, name, owner_ref(body) + ) + except (KeyError, client.exceptions.ApiException) as exc: + msg = str(exc).strip("'\"") if isinstance(exc, KeyError) else exc.reason patch.status["phase"] = Phase.FAILED - patch.status["message"] = str(exc).strip("'\"") + patch.status["message"] = msg set_condition( patch, conditions, COND_PIPELINE_RUN_READY, "False", "SecretRefNotFound", - str(exc).strip("'\""), + msg, ) ctx.kueue.delete_workload(name) logger.error("Job %s: %s", name, exc) @@ -182,7 +185,7 @@ def reconcile_admitted(spec, name, namespace, status, patch, body): env=spec.get("env", {}), kubeconfig_secret=secret, gpu_count=gpu_count, - secret_refs=resolved_refs, + resolved_secrets=resolved_secrets, cluster=cluster, owner_ref=owner_ref(body), ) diff --git a/fournos/handlers/resolving.py b/fournos/handlers/resolving.py index 6e77302..79b769b 100644 --- a/fournos/handlers/resolving.py +++ b/fournos/handlers/resolving.py @@ -223,6 +223,11 @@ def _create_workload_and_transition( owner_ref=owner_ref(body), ) except client.exceptions.ApiException as exc: + # The 409 in that case means that the Workload already exists in Kueue. + # This may happen when the reconciliation loop re-enters this function for a FJob that has been already processed in a previous iteration: + # the Workload was created, but the status patch (setting phase=Pending) may have not been persisted yet. + # Passing on 409 allows the execution to proceed to phase setting, which applies the correct status in an idempotent way. + # This is essentially a create-if-not-exists pattern: the first creation wins, subsequent attempts skip the API call and just ensure the status is set correctly. if exc.status == 409: pass else: diff --git a/manifests/rbac/role_fournos.yaml b/manifests/rbac/role_fournos.yaml index b75ca0f..fda7952 100644 --- a/manifests/rbac/role_fournos.yaml +++ b/manifests/rbac/role_fournos.yaml @@ -18,4 +18,4 @@ rules: verbs: ["create", "get"] - apiGroups: [""] resources: ["secrets"] - verbs: ["get"] + verbs: ["get", "create"] diff --git a/tests/conftest.py b/tests/conftest.py index a275060..e125195 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -278,6 +278,17 @@ def get_pipelinerun_param(name: str, param_name: str) -> Any: return None +def get_pipelinerun_volumes(name: str) -> list[dict]: + """Return the podTemplate volumes from the PipelineRun's taskRunTemplate.""" + pr = get_k8s_resource("pipelinerun", name) + return ( + pr.get("spec", {}) + .get("taskRunTemplate", {}) + .get("podTemplate", {}) + .get("volumes", []) + ) + + def resolve_job_exists(name: str) -> bool: """Check whether the resolve Job for *name* exists.""" result = subprocess.run( diff --git a/tests/test_secret_refs.py b/tests/test_secret_refs.py index e323fd2..70cd1d0 100644 --- a/tests/test_secret_refs.py +++ b/tests/test_secret_refs.py @@ -26,6 +26,7 @@ SECRETS_NAMESPACE, create_job, get_pipelinerun_param, + get_pipelinerun_volumes, job_status_summary, poll_phase, poll_resolve_job_complete, @@ -114,6 +115,8 @@ def test_vault_sync_then_fjob(k8s, core_v1): ) assert rc == 0, "sync_vault_secrets.sync() returned non-zero" + expected_copy = f"test-e2e-secret-{VAULT_SECRET}" + try: secret = core_v1.read_namespaced_secret(VAULT_SECRET, SECRETS_NAMESPACE) assert secret.metadata.labels[LABEL_VAULT_ENTRY] == "true" @@ -149,6 +152,27 @@ def test_vault_sync_then_fjob(k8s, core_v1): f"got {refs_param!r}" ) + volumes = get_pipelinerun_volumes("test-e2e-secret") + vault_vol = next((v for v in volumes if v.get("name") == "vault-secrets"), None) + assert vault_vol is not None, ( + f"Expected a 'vault-secrets' projected volume, got {volumes!r}" + ) + sources = vault_vol.get("projected", {}).get("sources", []) + source_names = [s.get("secret", {}).get("name", "") for s in sources] + assert expected_copy in source_names, ( + f"Projected volume should reference copied secret {expected_copy!r}, " + f"got sources: {source_names!r}" + ) + + copied = core_v1.read_namespaced_secret(expected_copy, NAMESPACE) + owner_refs = copied.metadata.owner_references or [] + assert any( + o.kind == "FournosJob" and o.name == "test-e2e-secret" for o in owner_refs + ), f"Copied secret should have FournosJob ownerRef, got {owner_refs!r}" + assert sorted(copied.data.keys()) == ["password", "username"], ( + f"Copied secret data keys mismatch: {sorted(copied.data.keys())}" + ) + phase = poll_phase( k8s, "test-e2e-secret", @@ -159,6 +183,7 @@ def test_vault_sync_then_fjob(k8s, core_v1): finally: _delete_secret_if_exists(core_v1, VAULT_SECRET) + _delete_secret_if_exists(core_v1, expected_copy, namespace=NAMESPACE) def test_missing_secret_ref_fails(k8s):