diff --git a/python/fluence/providers/qrmi.py b/python/fluence/providers/qrmi.py index b145a8e..fe6df3b 100644 --- a/python/fluence/providers/qrmi.py +++ b/python/fluence/providers/qrmi.py @@ -18,10 +18,26 @@ # qrmi_type values whose ResourceType name cannot be derived from them _RESOURCE_TYPE_ALIASES = { + "qiskit-runtime-service": "IBMQiskitRuntimeService", "direct-access": "IBMQuantumSystem", # legacy name for ibm-quantum-system "ibm-direct-access": "IBMQuantumSystem", } +# the infix QRMI puts in its variable names for each IBM type, as in +# _QRMI_IBM_QRS_IAM_APIKEY +_ENV_INFIX = { + "IBMQiskitRuntimeService": "_QRMI_IBM_QRS", + "IBMQuantumSystem": "_QRMI_IBM_QS", +} + +# what the infix variables default to, or are derived from, when only the +# IBM_CLOUD_TOKEN and IBM_CLOUD_CRN a workload Secret provides are present. +# The sampler derives the same set for the scout. +_IBM_DEFAULTS = { + "_ENDPOINT": "https://quantum.cloud.ibm.com/api/v1", + "_IAM_ENDPOINT": "https://iam.cloud.ibm.com", +} + # kept upper case when deriving a ResourceType name _ACRONYMS = {"ibm", "iqm", "qrmi", "id", "api", "hpc"} @@ -56,8 +72,9 @@ def _pascal(qrmi_type: str, acronyms=True) -> str: def _resource_type_enum(qrmi_type: str): """The QRMI ResourceType for a qrmi_type string. An alias first, then the - name derived from the string. Falls back to the first member with a warning - rather than failing on a QRMI version we have not seen.""" + name derived from the string. Raises when nothing matches. A guessed type + reads the wrong credentials and asks the wrong service about the task, so + it is never better than stopping.""" from qrmi import ResourceType q = (qrmi_type or "").lower() @@ -70,12 +87,29 @@ def _resource_type_enum(qrmi_type: str): rt = getattr(ResourceType, name, None) if rt is not None: return rt - log( - f"[qrmi] WARNING: no ResourceType matched {qrmi_type!r} " - f"(tried {candidates}); falling back to a default member" + members = [k for k in vars(ResourceType) if not k.startswith("_")] + raise ValueError( + f"no QRMI ResourceType matches {qrmi_type!r}, tried {candidates}, " + f"QRMI has {members}" ) - members = [v for k, v in vars(ResourceType).items() if not k.startswith("_")] - return members[0] if members else None + + +def _derive_ibm_env(backend: str, rtype) -> None: + """Fill in _* from IBM_CLOUD_TOKEN and IBM_CLOUD_CRN, + never overwriting a variable that is already set. The webhook copies the + workload's Secret env onto the sidecar, and that Secret holds the plain + token and CRN, not QRMI's names.""" + name = rtype if isinstance(rtype, str) else getattr(rtype, "name", str(rtype)) + infix = _ENV_INFIX.get(name.rsplit(".", 1)[-1]) + if not infix: + return + values = dict(_IBM_DEFAULTS) + if os.environ.get("IBM_CLOUD_TOKEN"): + values["_IAM_APIKEY"] = os.environ["IBM_CLOUD_TOKEN"] + if os.environ.get("IBM_CLOUD_CRN"): + values["_SERVICE_CRN"] = os.environ["IBM_CLOUD_CRN"] + for suffix, value in values.items(): + os.environ.setdefault(backend + infix + suffix, value) def _status_name(status) -> str: @@ -235,35 +269,37 @@ def _qrmi_resource(self, task: QRMITask): import qrmi rt = _resource_type_enum(task.resource_type) + _derive_ibm_env( + task.backend, + _RESOURCE_TYPE_ALIASES.get((task.resource_type or "").lower(), rt), + ) task._resource = qrmi.QuantumResource(task.backend, rt) return task._resource - def _status(self, task: QRMITask) -> "str | None": - try: - status = self._qrmi_resource(task).task_status(task.job_id) - return _status_name(status) - except Exception as e: - log(f"[qrmi] task_status error (will retry): {e}") - return None + def _status(self, task: QRMITask) -> str: + # Errors are raised, not swallowed. The sidecar's poll loop counts + # them and gives up after FLUENCE_POLL_MAX_ERRORS, which tells the + # scout to release its session. Swallowed, a status call that can + # never work polled forever while the scout held the session. + return _status_name(self._qrmi_resource(task).task_status(task.job_id)) 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 # is running on the device or has completed. Failed and cancelled are # not ready, task_failed reports them. - name = self._status(task) - return name in _READY_STATES if name else False + return self._status(task) in _READY_STATES def task_failed(self, task: QRMITask) -> bool: - name = self._status(task) - return name in _FAILED_STATES if name else False + return self._status(task) in _FAILED_STATES def queue_position(self, task: QRMITask): # QRMI has no queue position, so report 1 once the task runs and None before - name = self._status(task) - if name and name in _READY_STATES: - return 1 - return None + try: + name = self._status(task) + except Exception: + return None + return 1 if name in _READY_STATES else None def job_id(self, task: QRMITask) -> str: return task.job_id diff --git a/python/fluence/sidecar.py b/python/fluence/sidecar.py index 7e078d1..4f1cc16 100644 --- a/python/fluence/sidecar.py +++ b/python/fluence/sidecar.py @@ -70,8 +70,8 @@ def _poll(provider, task, poll_interval, ungate, position=1): 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}") + except (TypeError, AttributeError, NameError, ValueError) as e: + log(f"ERROR: poll failed with an error retrying cannot fix: {e!r}") return False except Exception as e: errors += 1 diff --git a/python/tests/test_qrmi.py b/python/tests/test_qrmi.py index ef8b5ea..47e8252 100644 --- a/python/tests/test_qrmi.py +++ b/python/tests/test_qrmi.py @@ -85,8 +85,9 @@ def test_resource_type_derivation_covers_multiple_vendors(): # just a new qrmi_type in config. Verify with a fake ResourceType enum. from fluence.providers import qrmi as qmod + # the member names QRMI actually has fake = types.SimpleNamespace( - QiskitRuntimeService="QiskitRuntimeService", + IBMQiskitRuntimeService="IBMQiskitRuntimeService", IBMQuantumSystem="IBMQuantumSystem", PasqalCloud="PasqalCloud", ) @@ -95,15 +96,59 @@ def test_resource_type_derivation_covers_multiple_vendors(): sys.modules["qrmi"] = fake_mod try: assert ( - qmod._resource_type_enum("qiskit-runtime-service") == "QiskitRuntimeService" + qmod._resource_type_enum("qiskit-runtime-service") + == "IBMQiskitRuntimeService" ) assert qmod._resource_type_enum("ibm-quantum-system") == "IBMQuantumSystem" assert qmod._resource_type_enum("direct-access") == "IBMQuantumSystem" # alias assert qmod._resource_type_enum("pasqal-cloud") == "PasqalCloud" + # an unknown type stops, it is never guessed + try: + qmod._resource_type_enum("no-such-type") + except ValueError: + pass + else: + raise AssertionError("an unknown type was guessed") finally: sys.modules.pop("qrmi", None) +def test_sidecar_derives_qrmi_env_from_the_secret(): + """The Secret gives the sidecar IBM_CLOUD_TOKEN and IBM_CLOUD_CRN. QRMI + reads _QRMI_IBM_QRS_*, so those are derived, and an explicit + value is never overwritten. No fixtures, the file's own runner calls + tests with no arguments.""" + from fluence.providers import qrmi as qmod + + keys = ("IBM_CLOUD_TOKEN", "IBM_CLOUD_CRN") + tuple( + k for k in os.environ if k.startswith("ibm_fez_QRMI") + ) + saved = {k: os.environ.get(k) for k in keys} + try: + for k in keys: + os.environ.pop(k, None) + os.environ["IBM_CLOUD_TOKEN"] = "tok" + os.environ["IBM_CLOUD_CRN"] = "crn" + os.environ["ibm_fez_QRMI_IBM_QRS_ENDPOINT"] = "https://mine" + qmod._derive_ibm_env("ibm_fez", "IBMQiskitRuntimeService") + assert os.environ["ibm_fez_QRMI_IBM_QRS_IAM_APIKEY"] == "tok" + assert os.environ["ibm_fez_QRMI_IBM_QRS_SERVICE_CRN"] == "crn" + assert ( + os.environ["ibm_fez_QRMI_IBM_QRS_IAM_ENDPOINT"] + == "https://iam.cloud.ibm.com" + ) + assert os.environ["ibm_fez_QRMI_IBM_QRS_ENDPOINT"] == "https://mine" + finally: + for k in list(os.environ): + if k.startswith("ibm_fez_QRMI"): + del os.environ[k] + for k, v in saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + def test_pascal_transform_acronym_aware(): from fluence.providers.qrmi import _pascal @@ -303,11 +348,15 @@ def __str__(self): assert prov.is_ready_to_ungate(task) is True -def test_status_error_is_not_ready_not_raising(): +def test_status_error_reaches_the_poll_loop_and_it_gives_up(): + """A status call that keeps failing must end the poll, so the scout is + told to release its session. Swallowed, it polled forever.""" + class Boom: def task_status(self, job_id): - raise RuntimeError("transient") + raise RuntimeError("ibm_fez_QRMI_IBM_QRS_ENDPOINT is not set") + from fluence import sidecar from fluence.providers.qrmi import QRMIProvider, QRMITask prov = QRMIProvider() @@ -315,7 +364,21 @@ def task_status(self, job_id): job_id="j", backend="ibm_fez", resource_type="qiskit-runtime-service" ) task._resource = Boom() - assert prov.is_ready_to_ungate(task) is False + try: + prov.is_ready_to_ungate(task) + except RuntimeError: + pass + else: + raise AssertionError("the status error was swallowed") + saved = os.environ.get("FLUENCE_POLL_MAX_ERRORS") + os.environ["FLUENCE_POLL_MAX_ERRORS"] = "3" + try: + assert sidecar._poll(prov, task, 0, ungate=True) is False + finally: + if saved is None: + del os.environ["FLUENCE_POLL_MAX_ERRORS"] + else: + os.environ["FLUENCE_POLL_MAX_ERRORS"] = saved # ── job id / session id accessors ─────────────────────────────────────────────