sidecar: add better message to user

README.md

@@ -87,8 +87,16 @@ Admission control is a check that asks "Can we support a new hybrid queue?" wher
 | `FLUENCE_RESOURCE_CAPACITY` | device plugin | per node ceiling per type, default 1000 |
 | `FLUENCE_ADMISSION` | webhook | off turns admission control off, default on, read at startup |
 | `FLUENCE_RESERVE_CORES` | webhook | cores held back from the admission budget, default 0 |
-| `FLUENCE_WORKER_TIMEOUT` | sidecar | seconds to wait for the workers after ungating, 0 waits forever, default 0 |
-| `FLUENCE_SESSION_HOLD_TIMEOUT` | scout container | seconds the scout's release waits for the workers, 0 waits forever, default 0 |
+| `FLUENCE_WORKER_TIMEOUT` | sidecar | seconds to wait for the workers after ungating, 0 waits forever, default 3600 |
+| `FLUENCE_SESSION_HOLD_TIMEOUT` | scout container | seconds the scout's release waits for the workers, 0 waits forever, default 3600 |
+| `FLUENCE_POLL_INTERVAL` | sidecar | seconds between checks on the scout's task, default 2. Each check is one status call to the vendor |
+| `FLUENCE_POLL_MAX_ERRORS` | sidecar | vendor errors in a row before the sidecar gives up and releases the workers, default 10. A programming error stops it at once |
+| `FLUENCE_QRMI_SESSION_MODE` | workload container | batch or dedicated for the IBM session the scout opens, default batch |
+
+### What an IBM session costs
+
+IBM bills a dedicated session for wall clock time from the moment its first job starts until the session is closed or has sat idle for a minute. This includes idle time. In a shared gang there is always idle time between the scout's circuit finishing and the workers' circuits arriving, while the worker pods schedule,
+start and submit, and again after the last worker finishes before the scout releases. With dedicated mode that gap is billed at full rate and can come to about a minute per run. A batch session is billed only for time on the hardware, so the gap costs nothing. Thus, we choose to open batch sessions unless `FLUENCE_QRMI_SESSION_MODE=dedicated` is set. Use dedicated only when exclusive access is what you are measuring, and set a cost limit on the IBM instance first.
 
 The scheduler logs the graph when it starts, and for every match it logs the jobspec it sent, the allocation it got back, and how it read it, all prefixed with [fluence]. The webhook logs its environment contract at startup. You can look at logs to debug.
 

examples/qrmi-workload/sampler.py

@@ -5,6 +5,13 @@
 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.
+
+The session is opened in batch mode unless FLUENCE_QRMI_SESSION_MODE says
+dedicated. IBM bills a dedicated session for wall clock time from its first
+job until it closes or sits idle for a minute, so the gap between the scout's
+circuit and the workers' circuits is billed at full rate. A batch session is
+billed only for time on the hardware. Dedicated is for measuring exclusive
+access on purpose, and costs about a minute of idle time per run.
 """
 
 import argparse
@@ -56,6 +63,24 @@ def backend_from_fluence():
     return backend, rtype
 
 
+SESSION_MODES = ("batch", "dedicated")
+
+
+def session_mode(backend, rtype):
+    """Set the QRMI session mode for this backend, batch unless asked for
+    dedicated, and return it. An explicit <backend>_QRMI_IBM_QRS_SESSION_MODE
+    in the environment wins."""
+    name = backend + RESOURCE_TYPES[rtype][1] + "_SESSION_MODE"
+    mode = (
+        os.environ.get(name) or os.environ.get("FLUENCE_QRMI_SESSION_MODE") or "batch"
+    )
+    mode = mode.lower()
+    if mode not in SESSION_MODES:
+        sys.exit("session mode must be batch or dedicated, not {}".format(mode))
+    os.environ[name] = mode
+    return mode
+
+
 def missing_credentials(backend, rtype):
     """Derive the QRMI variables from IBM_CLOUD_TOKEN and IBM_CLOUD_CRN, and
     return the names still unset."""
@@ -162,6 +187,16 @@ def main():
 
     print_environment()
     backend, rtype = backend_from_fluence()
+    mode = session_mode(backend, rtype)
+    print(
+        "mode:    ",
+        mode,
+        (
+            "(billed only for time on the hardware)"
+            if mode == "batch"
+            else "(billed for wall clock time while the session is active, idle gaps included)"
+        ),
+    )
     missing = missing_credentials(backend, rtype)
     if missing:
         print("missing: ", ", ".join(missing))

python/fluence/providers/qrmi.py

@@ -252,7 +252,10 @@ def _status(self, task: QRMITask) -> "str | None":
             log(f"[qrmi] task_status error (will retry): {e}")
             return None
 
-    def is_ready_to_ungate(self, task: QRMITask) -> bool:
+    def is_ready_to_ungate(self, task: QRMITask, position=None) -> bool:
+        # QRMI reports no queue position, so position is accepted to match the
+        # interface and the sidecar's call, and ignored. Ready means the task
+        # has left the queue.
         name = self._status(task)
         return name in _READY_STATES if name else False
 

python/fluence/sidecar.py

@@ -15,7 +15,11 @@
   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)
-  FLUENCE_POLL_INTERVAL           seconds between polls (default 30)
+  FLUENCE_POLL_INTERVAL           seconds between polls (default 2). A dedicated
+                                  session bills the time between the scout's
+                                  circuit and the workers' circuits, and half
+                                  an interval of that is this poll. Each poll
+                                  is one status call to the vendor
   FLUENCE_UNGATE_POSITION         ungate at this queue position or closer (default 1)
   FLUENCE_WORKER_TIMEOUT          seconds to wait for the workers after ungating,
                                   0 waits without limit (default 0)
@@ -45,6 +49,13 @@ def _poll(provider, task, poll_interval, ungate, position=1):
     mode = "gang" if ungate else "observe-only"
     log(f"{mode} mode: polling queue position, ungating at {position} or closer")
     last = object()
+    # A vendor call can fail for a while and recover, so those are retried.
+    # A bug in our own code fails the same way every time and never
+    # recovers, and while it retries a session that already started keeps
+    # billing with nothing using it. So programming errors stop the poll at
+    # once, and so does any error that keeps happening.
+    max_errors = int(os.environ.get("FLUENCE_POLL_MAX_ERRORS", 10))
+    errors = 0
     while True:
         try:
             if provider.task_failed(task):
@@ -57,8 +68,18 @@ def _poll(provider, task, poll_interval, ungate, position=1):
             if pos != last:
                 log(f"queue position: {pos}")
                 last = pos
+            errors = 0
+        except (TypeError, AttributeError, NameError) as e:
+            log(f"ERROR: poll failed with a programming error, not retrying: {e!r}")
+            return False
         except Exception as e:
-            log(f"poll error (will retry): {e}")
+            errors += 1
+            log(f"poll error {errors} of {max_errors} (will retry): {e}")
+            if errors >= max_errors:
+                log(
+                    "ERROR: poll kept failing, giving up so the session is not left idle"
+                )
+                return False
         time.sleep(poll_interval)
 
 
@@ -84,7 +105,7 @@ def main():
     backend = os.environ.get("FLUXION_BACKEND", "")
     observe = os.environ.get("FLUENCE_OBSERVE", "").lower() == "true"
     discovery_timeout = int(os.environ.get("FLUENCE_TASK_DISCOVERY_TIMEOUT", 300))
-    poll_interval = int(os.environ.get("FLUENCE_POLL_INTERVAL", 30))
+    poll_interval = int(os.environ.get("FLUENCE_POLL_INTERVAL", 2))
     ungate_timeout = int(os.environ.get("FLUENCE_UNGATE_TIMEOUT", 120))
     ungate_at = ungate_position()
 
@@ -147,7 +168,9 @@ def main():
         sys.exit(0 if ready else 1)
 
     if not ready:
-        # fail open so the workers are not stranded. They find no result and exit
+        # fail open so the workers are not stranded. The session goes with
+        # them: the scout is still holding it, and workers that ran outside
+        # it would leave it open and idle, which a dedicated session bills
         log(
             "ERROR: ungating anyway so the gang is not stranded, but the task "
             "produced no result"
@@ -159,6 +182,7 @@ def main():
             ),
             job_id,
             namespace,
+            session_id=provider.session_id(task),
             backend=placed_backend,
             attributes=placed_attrs,
         )

python/fluence/ungate.py

@@ -207,7 +207,7 @@ def pods_done(namespace, group, exclude=""):
     )
 
 
-def wait_for_pods_done(namespace, group, exclude="", timeout=0, interval=10):
+def wait_for_pods_done(namespace, group, exclude="", timeout=0, interval=2):
     """Wait until the other pods in the group have all finished. A timeout of
     zero waits without limit. Returns True when they are done."""
     if not group:

python/tests/test_qrmi.py

@@ -395,3 +395,114 @@ def test_find_my_task_times_out_to_none():
             print(f"FAIL {fn.__name__}: {e}")
     print(f"\n{len(fns) - failed}/{len(fns)} passed")
     sys.exit(1 if failed else 0)
+
+
+def test_ungate_takes_the_position_the_sidecar_passes():
+    """The sidecar calls is_ready_to_ungate(task, position) for every provider.
+    QRMI has no position, but it must accept the argument, or every poll
+    raises and the gang is never ungated."""
+    import inspect
+
+    from fluence.providers.base import Provider
+    from fluence.providers import qrmi as q
+
+    sig = inspect.signature(q.QRMIProvider.is_ready_to_ungate)
+    assert "position" in sig.parameters
+    base = inspect.signature(Provider.is_ready_to_ungate)
+    assert list(sig.parameters) == list(base.parameters)
+
+
+def test_poll_stops_on_a_programming_error(monkeypatch):
+    """A bug in a provider raises the same error every poll. Retrying it
+    forever left a started session idle and billing, so it must stop."""
+    from fluence import sidecar
+
+    monkeypatch.setattr(sidecar.time, "sleep", lambda s: None)
+
+    class Broken:
+        def task_failed(self, task):
+            return False
+
+        def is_ready_to_ungate(self, task):  # the old, wrong signature
+            return True
+
+        def queue_position(self, task):
+            return None
+
+    assert sidecar._poll(Broken(), object(), 0, ungate=True) is False
+
+
+def test_poll_gives_up_after_repeated_vendor_errors(monkeypatch):
+    from fluence import sidecar
+
+    monkeypatch.setattr(sidecar.time, "sleep", lambda s: None)
+    monkeypatch.setenv("FLUENCE_POLL_MAX_ERRORS", "3")
+    calls = {"n": 0}
+
+    class Flaky:
+        def task_failed(self, task):
+            calls["n"] += 1
+            raise RuntimeError("503 from the vendor")
+
+    assert sidecar._poll(Flaky(), object(), 0, ungate=True) is False
+    assert calls["n"] == 3
+
+
+def test_poll_retries_a_vendor_error_then_succeeds(monkeypatch):
+    from fluence import sidecar
+
+    monkeypatch.setattr(sidecar.time, "sleep", lambda s: None)
+    answers = iter([RuntimeError("timeout"), False, True])
+
+    class Recovering:
+        def task_failed(self, task):
+            return False
+
+        def is_ready_to_ungate(self, task, position=None):
+            a = next(answers)
+            if isinstance(a, Exception):
+                raise a
+            return a
+
+        def queue_position(self, task):
+            return 1
+
+    assert sidecar._poll(Recovering(), object(), 0, ungate=True) is True
+
+
+def test_fail_open_hands_the_workers_the_session(monkeypatch):
+    """When the poll fails the workers are released anyway. They must get
+    the session id, or they run outside a session the scout still holds."""
+    from fluence import sidecar
+
+    calls = {}
+    monkeypatch.setattr(sidecar, "_poll", lambda *a, **k: False)
+    monkeypatch.setattr(sidecar, "_hold_for_workers", lambda *a, **k: None)
+    monkeypatch.setattr(sidecar, "gated_pods_from_env", lambda: ["worker-0"])
+    monkeypatch.setattr(
+        sidecar,
+        "ungate_pods",
+        lambda pods, job_id, ns, **kw: calls.update(kw) or len(pods),
+    )
+    monkeypatch.setattr(sidecar, "placement_from_env", lambda: ("ibm_x", {}))
+    monkeypatch.setattr(sidecar, "namespace_from_env", lambda: "default")
+
+    class P:
+        name = "fake"
+
+        def find_my_task(self, uid, backend, timeout):
+            return "task"
+
+        def job_id(self, task):
+            return "job-1"
+
+        def session_id(self, task):
+            return "sess-1"
+
+    monkeypatch.setattr(sidecar, "resolve_from_env", lambda: P())
+    monkeypatch.setenv("FLUENCE_OBSERVE", "false")
+    try:
+        sidecar.main()
+    except SystemExit as e:
+        assert e.code == 1
+    assert calls.get("session_id") == "sess-1"