rehearsal: make explicit

docs/quantum-scheduling.md

@@ -63,7 +63,7 @@ The quantum handler also enforces admission control, which is always on unless t
 5. When the task is next in line, or already running, annotate each worker with the job id and the session, then remove the gate.
 6. Wait for the workers to reach Succeeded or Failed, write the done marker so the scout's release proceeds, and exit.
 
-The sidecar ungates the workers only when the scout's task is running or has completed. If the task fails, is cancelled, or is never found, the workers stay gated, the sidecar tells the scout to release its session at once, and it exits non zero so it shows. Delete the job to clear the gated workers. The one exception is a rehearsal, where the scout exits without submitting anything, for example with no credentials. Then the workers are released with no session.
+The sidecar ungates the workers only when the scout's task is running or has completed. If the task fails, is cancelled, or is never found, the workers stay gated, the sidecar tells the scout to release its session at once, and it exits non zero so it shows. Delete the job to clear the gated workers. The one exception is a rehearsal, which you ask for with the pod label `fluence.flux-framework.org/rehearsal: "true"`. The webhook passes it to the workload and the sidecar as `FLUENCE_REHEARSAL=true`. The workload submits nothing and the sidecar releases the workers with no session. Without the label, a scout that exits without submitting, for example because its credentials are missing, is an error. The workers stay gated and the sidecar exits non zero.
 
 ## Priority and preemption
 

examples/qrmi-workload/sampler.py

@@ -3,8 +3,10 @@
 
 fluence chooses the backend and the webhook forwards via FLUXION_BACKEND.
 This acquires a session on it, runs a Bell circuit through the QRMI sampler,
-prints the counts, and releases the session. Without credentials it prints
-what it would need and exits, so the same pod works as a rehearsal.
+prints the counts, and releases the session. With FLUENCE_REHEARSAL=true,
+set by the webhook from the fluence.flux-framework.org/rehearsal pod label,
+it prints what it would need and exits without touching the vendor. Without
+the label, missing credentials is an error.
 The session is opened in batch mode unless FLUENCE_QRMI_SESSION_MODE is set
 to dedicated.
 """
@@ -193,12 +195,20 @@ def main():
         ),
     )
     missing = missing_credentials(backend, rtype)
-    if missing:
-        print("missing: ", ", ".join(missing))
+    if os.environ.get("FLUENCE_REHEARSAL", "").lower() == "true":
+        if missing:
+            print("missing: ", ", ".join(missing))
         print(
             "\nrehearsal, nothing acquired or submitted. The placement and the handoff worked."
         )
         return
+    if missing:
+        sys.exit(
+            "missing credentials: {}. Nothing was submitted. Label the pods "
+            "fluence.flux-framework.org/rehearsal=true to run without them".format(
+                ", ".join(missing)
+            )
+        )
 
     qr = resource(backend, rtype)
     session = os.environ.get("FLUENCE_QUANTUM_SESSION_ID")

pkg/webhook/handlers/quantum.go

@@ -37,6 +37,13 @@ const (
 	// does not ungate.
 	ObserveLabel = "fluence.flux-framework.org/observe"
 
+	// RehearsalLabel says a run is meant to submit nothing. The webhook sets
+	// RehearsalEnv on the workload and the sidecar. Only then may the sidecar
+	// release workers when the scout exits without a task. Without it a scout
+	// that submitted nothing is an error and the workers stay gated.
+	RehearsalLabel = "fluence.flux-framework.org/rehearsal"
+	RehearsalEnv   = "FLUENCE_REHEARSAL"
+
 	// DependencyKindQuantumSubmit is the readiness Kind for the quantum resource
 	// type: worker pods wait for a quantum submission to reach the device queue.
 	// First concrete instance of the general Dependency primitive (dependency.go).
@@ -160,6 +167,7 @@ func (h *quantumHandler) Mutate(ctx context.Context, m webhook.MutatorAPI, pod *
 	n := resolveGangSize(ctx, m, pod, g)
 	mode := coordinationMode(pod)
 	observe := spec.Label(pod, ObserveLabel) == "true"
+	rehearsal := rehearsalEnv(pod)
 	cores := placement.PodCores(pod)
 
 	// No coordination. A standalone quantum pod, or an explicitly independent
@@ -170,10 +178,11 @@ func (h *quantumHandler) Mutate(ctx context.Context, m webhook.MutatorAPI, pod *
 		ops := admissionOps(pod, pod.Namespace+"/"+pod.Name, cores)
 		ops = append(ops, interceptorOps(pod)...)
 		ops = append(ops, handoffOps(pod)...)
+		ops = append(ops, rehearsalOps(pod, rehearsal)...)
 		if observe {
 			sc := sidecarFor(m)
 			sc.EnsureRBAC(ctx, pod.Namespace)
-			ops = append(ops, sc.ContainerOps(pod, true, nil)...)
+			ops = append(ops, sc.ContainerOps(pod, true, rehearsal)...)
 		}
 		log.Printf("[fluence-webhook] quantum %s/%s mode=%s (standalone/independent, observe=%v)",
 			pod.Namespace, pod.Name, mode, observe)
@@ -186,18 +195,37 @@ func (h *quantumHandler) Mutate(ctx context.Context, m webhook.MutatorAPI, pod *
 	m.EnsurePodGroup(ctx, pod.Namespace, g, pod.Name, n-1)
 	m.RecordAdmission(ctx, pod.Namespace, g, cores*int(n))
 	ops := admissionOps(pod, pod.Namespace+"/"+g, 0)
+	ops = append(ops, rehearsalOps(pod, rehearsal)...)
 	if isScout(ctx, m, pod, g) {
-		return append(ops, h.mutateScout(ctx, m, pod, g)...)
+		return append(ops, h.mutateScout(ctx, m, pod, g, rehearsal)...)
 	}
 	return append(ops, h.mutateWorker(ctx, m, pod, g, n)...)
 }
 
+// rehearsalEnv is the env that marks a rehearsal, or nil when the pod does
+// not carry the label.
+func rehearsalEnv(pod *corev1.Pod) []corev1.EnvVar {
+	if spec.Label(pod, RehearsalLabel) != "true" {
+		return nil
+	}
+	return []corev1.EnvVar{{Name: RehearsalEnv, Value: "true"}}
+}
+
+// rehearsalOps puts the rehearsal env on the workload containers.
+func rehearsalOps(pod *corev1.Pod, env []corev1.EnvVar) []spec.Op {
+	var ops []spec.Op
+	for _, e := range env {
+		ops = append(ops, setContainerEnvOps(pod, e)...)
+	}
+	return ops
+}
+
 // mutateScout wires the single scout member (indexed-Job completion index
 // 0). Its own group-of-one <group>-scout (minCount 1) so it schedules alone
 // and runs the REAL submit, the interceptor in tag mode, RBAC, and the sidecar
 // told which worker group to ungate (FLUENCE_GANG_GROUP). The scout is one
 // of the N members, so the application is NOT run an extra time. Never gated.
-func (h *quantumHandler) mutateScout(ctx context.Context, m webhook.MutatorAPI, pod *corev1.Pod, group string) []spec.Op {
+func (h *quantumHandler) mutateScout(ctx context.Context, m webhook.MutatorAPI, pod *corev1.Pod, group string, rehearsal []corev1.EnvVar) []spec.Op {
 	pg := group + ScoutGroupSuffix
 	m.EnsurePodGroup(ctx, pod.Namespace, pg, pod.Name, 1)
 	ops := linkGroupOps(pod, pg)
@@ -206,7 +234,7 @@ func (h *quantumHandler) mutateScout(ctx context.Context, m webhook.MutatorAPI,
 	ops = append(ops, roleEnvOps(pod, RoleScout)...) // FLUENCE_COORDINATION_ROLE=scout
 	sc := sidecarFor(m)
 	sc.EnsureRBAC(ctx, pod.Namespace)
-	extra := []corev1.EnvVar{{Name: GangGroupEnv, Value: group}}
+	extra := append([]corev1.EnvVar{{Name: GangGroupEnv, Value: group}}, rehearsal...)
 	ops = append(ops, sc.ContainerOps(pod, false, extra)...)
 	log.Printf("[fluence-webhook] quantum scout %s/%s, group %s (ungates workers %q)",
 		pod.Namespace, pod.Name, pg, group)

pkg/webhook/handlers/quantum_test.go

@@ -189,6 +189,40 @@ func TestSharedScoutWiredAsRealSidecar(t *testing.T) {
 	}
 }
 
+// The rehearsal label is the only thing that lets the sidecar release the
+// workers when the scout submits nothing. It must reach the workload and the
+// scout's sidecar, and be absent from both without the label.
+func TestRehearsalLabelReachesWorkloadAndSidecar(t *testing.T) {
+	ns, group, job := "default", "qr", "qr-job"
+	par := int32(2)
+	for _, labeled := range []bool{true, false} {
+		cs := fake.NewSimpleClientset(&batchv1.Job{
+			ObjectMeta: metav1.ObjectMeta{Name: job, Namespace: ns},
+			Spec:       batchv1.JobSpec{Parallelism: &par, Completions: &par}})
+		m := &webhook.Mutator{Clientset: cs}
+		pod := sharedQPUPod(ns, group, "qr-0", job, "0")
+		if labeled {
+			pod.Labels[RehearsalLabel] = "true"
+		}
+		ops := m.Mutate(context.Background(), pod)
+
+		_, onWorkload := envOp(ops, RehearsalEnv)
+		onSidecar := false
+		for _, op := range ops {
+			if c, ok := op.Value.(corev1.Container); ok && c.Name == SidecarContainerName {
+				for _, e := range c.Env {
+					if e.Name == RehearsalEnv && e.Value == "true" {
+						onSidecar = true
+					}
+				}
+			}
+		}
+		if onWorkload != labeled || onSidecar != labeled {
+			t.Errorf("labeled=%v: %s on workload=%v, on sidecar=%v", labeled, RehearsalEnv, onWorkload, onSidecar)
+		}
+	}
+}
+
 // Shared mode never creates an extra pod. A full gang (scout index 0 +
 // workers) is N members, so the application runs exactly N times (not N+1 as
 // the old submitter-pod model did).

python/fluence/handoff.py

@@ -88,8 +88,9 @@ def wait(timeout, interval=5, path=None, log=None) -> "dict | None":
 
 
 # the interceptor writes this when the scout's interpreter exits with no task
-# published, so a rehearsal with no credentials does not leave the workers
-# gated until the discovery timeout
+# published, so the sidecar stops waiting instead of running to the discovery
+# timeout. It says the scout is gone, not that releasing the workers is safe.
+# Only FLUENCE_REHEARSAL says that
 EXITED_FILENAME = "scout-exited"
 
 

python/fluence/interceptor.py

@@ -40,8 +40,8 @@ def install() -> None:
 
 
 def _mark_exit() -> None:
-    # tell the sidecar the application ended, so a scout that never submitted
-    # releases its workers instead of leaving them gated
+    # tell the sidecar the application ended, so it stops waiting for a task.
+    # The sidecar releases the workers only if the run is a rehearsal
     try:
         from fluence import handoff
 

python/fluence/sidecar.py

@@ -12,6 +12,11 @@
   FLUENCE_GANG_GROUP              group label of the workers to ungate
   FLUENCE_GATED_PODS              optional explicit list of worker pod names
   FLUENCE_OBSERVE                 true for observe mode
+  FLUENCE_REHEARSAL               true for a run meant to submit nothing. Only
+                                  then are the workers released, with no
+                                  session, when the scout exits without a task.
+                                  Set from the fluence.flux-framework.org/rehearsal
+                                  pod label
   FLUXION_BACKEND                 the backend the scheduler chose
   FLUENCE_TASK_DISCOVERY_TIMEOUT  seconds to wait for the task (default 300)
   FLUENCE_WORKER_TIMEOUT          seconds to hold for the workers (default 3600)
@@ -100,6 +105,7 @@ def main():
     gang_group = os.environ.get("FLUENCE_GANG_GROUP", "")
     backend = os.environ.get("FLUXION_BACKEND", "")
     observe = os.environ.get("FLUENCE_OBSERVE", "").lower() == "true"
+    rehearsal = os.environ.get("FLUENCE_REHEARSAL", "").lower() == "true"
     discovery_timeout = int(os.environ.get("FLUENCE_TASK_DISCOVERY_TIMEOUT", 300))
     poll_interval = int(os.environ.get("FLUENCE_POLL_INTERVAL", 2))
     ungate_timeout = int(os.environ.get("FLUENCE_UNGATE_TIMEOUT", 120))
@@ -110,7 +116,8 @@ def main():
     log("starting the fluence sidecar")
     log(
         f"  pod_uid={pod_uid} namespace={namespace} group={group} "
-        f"gang_group={gang_group} backend={backend} observe={observe}"
+        f"gang_group={gang_group} backend={backend} observe={observe} "
+        f"rehearsal={rehearsal}"
     )
 
     placed_backend, placed_attrs = placement_from_env()
@@ -121,12 +128,10 @@ def main():
     log(f"resolved provider: {provider.name}")
 
     task = provider.find_my_task(pod_uid, backend, discovery_timeout)
-    if task is None and handoff.scout_exited() and not observe:
-        # a rehearsal, or an application that never submitted. Release the
-        # workers with no session and hold for them as usual
-        log(
-            "the scout finished without submitting a task, releasing the workers with no session"
-        )
+    if task is None and rehearsal and not observe:
+        # Asked for a run that submits nothing. Release the workers with no
+        # session and hold for them as usual.
+        log("rehearsal, no task expected, releasing the workers with no session")
         ungate_pods(
             wait_for_gated_pods(
                 namespace, gang_group, exclude=pod_name, timeout=ungate_timeout
@@ -139,14 +144,29 @@ def main():
         _hold_for_workers(namespace, gang_group, pod_name)
         sys.exit(0)
     if task is None:
-        # The scout neither published a task nor exited, so something is
-        # wrong with it. Releasing the workers would start them on a run that
-        # has no quantum task, so they stay gated. Delete the job to clear them.
+        # No task and this is not a rehearsal. The scout failed before it
+        # submitted, for example with missing credentials, or never finished
+        # submitting. Releasing the workers would start them on a run with no
+        # quantum task, so they stay gated. Delete the job to clear them.
+        if handoff.scout_exited():
+            log(
+                "ERROR: the scout exited without submitting a task. The workers "
+                "stay gated, delete the job to clear them. If this run was meant "
+                "to submit nothing, label the pods "
+                "fluence.flux-framework.org/rehearsal=true"
+            )
+        else:
+            log(
+                "ERROR: could not discover the quantum task. The workers stay "
+                "gated, delete the job to clear them"
+            )
+        sys.exit(1)
+
+    if rehearsal:
         log(
-            "ERROR: could not discover the quantum task. The workers stay gated, "
-            "delete the job to clear them"
+            "WARNING: this is labeled a rehearsal but the scout submitted a task. "
+            "Using it, since it is already paid for"
         )
-        sys.exit(1)
 
     job_id = provider.job_id(task)
     log(f"discovered task, job_id={job_id}")

python/tests/test_qrmi.py

@@ -470,7 +470,7 @@ def queue_position(self, task):
     assert sidecar._poll(Recovering(), object(), 0, ungate=True) is True
 
 
-def _sidecar_with(monkeypatch, task, ready, scout_exited=False):
+def _sidecar_with(monkeypatch, task, ready, scout_exited=False, rehearsal=False):
     """Run the sidecar's main with a fake provider. Returns what it did."""
     from fluence import sidecar
 
@@ -508,6 +508,7 @@ def session_id(self, t):
 
     monkeypatch.setattr(sidecar, "resolve_from_env", lambda: P())
     monkeypatch.setenv("FLUENCE_OBSERVE", "false")
+    monkeypatch.setenv("FLUENCE_REHEARSAL", "true" if rehearsal else "false")
     code = 0
     try:
         sidecar.main()
@@ -538,10 +539,21 @@ def test_a_task_that_cannot_be_found_leaves_the_workers_gated(monkeypatch):
     assert did["code"] == 1
 
 
-def test_a_rehearsal_still_releases_the_workers(monkeypatch):
-    """The scout exited with nothing submitted, no credentials. That is the
-    mock run, and the workers go with no session."""
+def test_a_scout_that_exits_without_a_task_is_an_error(monkeypatch):
+    """No task and no rehearsal label means the scout failed, for example
+    with missing credentials. The workers must not start on a run with no
+    quantum task."""
     did = _sidecar_with(monkeypatch, task=None, ready=False, scout_exited=True)
+    assert did["ungated"] is None
+    assert did["code"] == 1
+
+
+def test_a_rehearsal_still_releases_the_workers(monkeypatch):
+    """Labeled a rehearsal, so no task is expected and the workers go with
+    no session."""
+    did = _sidecar_with(
+        monkeypatch, task=None, ready=False, scout_exited=True, rehearsal=True
+    )
     assert did["ungated"] is not None
     assert did["ungated"]["job_id"] == ""
     assert not did["ungated"].get("session_id")