Allow users to set a logging level when running a program

qiskit_ibm_runtime/__init__.py

@@ -202,6 +202,7 @@ def interim_result_callback(job_id, interim_result):
    RuntimeEncoder
    RuntimeDecoder
    ParameterNamespace
+   RuntimeOptions
 """
 # """
 # ===================================================
@@ -284,6 +285,7 @@ def interim_result_callback(job_id, interim_result):
 from .program.program_backend import ProgramBackend
 from .program.result_decoder import ResultDecoder
 from .utils.json import RuntimeEncoder, RuntimeDecoder
+from .runtime_options import RuntimeOptions
 
 # Setup the logger for the IBM Quantum Provider package.
 logger = logging.getLogger(__name__)

qiskit_ibm_runtime/api/clients/runtime.py

@@ -118,8 +118,9 @@ def program_run(
         program_id: str,
         backend_name: Optional[str],
         params: Dict,
-        image: str,
+        image: Optional[str],
         hgp: Optional[str],
+        log_level: Optional[str],
     ) -> Dict:
         """Run the specified program.
 
@@ -129,6 +130,7 @@ def program_run(
             params: Parameters to use.
             image: The runtime image to use.
             hgp: Hub/group/project to use.
+            log_level: Log level to use.
 
         Returns:
             JSON response.
@@ -142,6 +144,7 @@ def program_run(
             backend_name=backend_name,
             params=params,
             image=image,
+            log_level=log_level,
             **hgp_dict
         )
 

qiskit_ibm_runtime/api/rest/runtime.py

@@ -115,10 +115,11 @@ def program_run(
         program_id: str,
         backend_name: Optional[str],
         params: Dict,
-        image: str,
+        image: Optional[str] = None,
         hub: Optional[str] = None,
         group: Optional[str] = None,
         project: Optional[str] = None,
+        log_level: Optional[str] = None,
     ) -> Dict:
         """Execute the program.
 
@@ -130,6 +131,7 @@ def program_run(
             hub: Hub to be used.
             group: Group to be used.
             project: Project to be used.
+            log_level: Log level to use.
 
         Returns:
             JSON response.
@@ -138,8 +140,11 @@ def program_run(
         payload = {
             "program_id": program_id,
             "params": params,
-            "runtime": image,
         }
+        if image:
+            payload["runtime"] = image
+        if log_level:
+            payload["log_level"] = log_level
         if backend_name:
             payload["backend"] = backend_name
         if all([hub, group, project]):

qiskit_ibm_runtime/ibm_runtime_service.py

@@ -14,7 +14,6 @@
 
 import json
 import logging
-import re
 import traceback
 import warnings
 from collections import OrderedDict
@@ -47,6 +46,7 @@
 from .utils.backend_decoder import configuration_from_server_data
 from .utils.hgp import to_instance_format, from_instance_format
 from .api.client_parameters import ClientParameters
+from .runtime_options import RuntimeOptions
 
 logger = logging.getLogger(__name__)
 
@@ -768,10 +768,9 @@ def run(
         self,
         program_id: str,
         inputs: Union[Dict, ParameterNamespace],
-        options: Optional[Dict] = None,
+        options: Optional[Union[RuntimeOptions, Dict]] = None,
         callback: Optional[Callable] = None,
         result_decoder: Optional[Type[ResultDecoder]] = None,
-        image: str = "",
         instance: Optional[str] = None,
     ) -> RuntimeJob:
         """Execute the runtime program.
@@ -780,9 +779,8 @@ def run(
             program_id: Program ID.
             inputs: Program input parameters. These input values are passed
                 to the runtime program.
-            options: Runtime options that control the execution environment.
-                Currently the only available option is ``backend_name``, which is required if
-                you are using legacy runtime.
+            options: Runtime options that control the execution environment. See
+                :class:`RuntimeOptions` for all available options.
             callback: Callback function to be invoked for any interim results.
                 The callback function will receive 2 positional parameters:
 
@@ -791,8 +789,6 @@ def run(
 
             result_decoder: A :class:`ResultDecoder` subclass used to decode job results.
                 ``ResultDecoder`` is used if not specified.
-            image: The runtime image used to execute the program, specified in the form
-                of image_name:tag. Not all accounts are authorized to select a different image.
             instance: This is only supported for legacy runtime and is in the
                 hub/group/project format.
 
@@ -814,35 +810,27 @@ def run(
             inputs.validate()
             inputs = vars(inputs)
 
-        if image and not re.match(
-            "[a-zA-Z0-9]+([/.\\-_][a-zA-Z0-9]+)*:[a-zA-Z0-9]+([.\\-_][a-zA-Z0-9]+)*$",
-            image,
-        ):
-            raise IBMInputValueError('"image" needs to be in form of image_name:tag')
+        if isinstance(options, dict):
+            options = RuntimeOptions(**options)
+        options.validate(self.auth)
 
-        options = options or {}
-        backend_name = options.get("backend_name", None)
         backend = None
-
         hgp_name = None
         if self._auth == "legacy":
-            if not backend_name:
-                raise IBMInputValueError(
-                    '"backend_name" is required field in "options" for legacy runtime.'
-                )
             # Find the right hgp
-            hgp = self._get_hgp(instance=instance, backend_name=backend_name)
-            backend = hgp.backend(backend_name)
+            hgp = self._get_hgp(instance=instance, backend_name=options.backend_name)
+            backend = hgp.backend(options.backend_name)
             hgp_name = hgp.name
 
         result_decoder = result_decoder or ResultDecoder
         try:
             response = self._api_client.program_run(
                 program_id=program_id,
-                backend_name=backend_name,
+                backend_name=options.backend_name,
                 params=inputs,
-                image=image,
+                image=options.image,
                 hgp=hgp_name,
+                log_level=options.log_level,
             )
         except RequestsApiError as ex:
             if ex.status_code == 404:
@@ -863,7 +851,7 @@ def run(
             params=inputs,
             user_callback=callback,
             result_decoder=result_decoder,
-            image=image,
+            image=options.image,
         )
         return job
 

qiskit_ibm_runtime/runtime_options.py

@@ -0,0 +1,67 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2022.
+#
+# This code is licensed under the Apache License, Version 2.0. You may
+# obtain a copy of this license in the LICENSE.txt file in the root directory
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+
+"""Runtime options that control the execution environment."""
+
+import re
+import logging
+from dataclasses import dataclass
+from typing import Optional
+
+from .exceptions import IBMInputValueError
+
+
+@dataclass
+class RuntimeOptions:
+    """Class for representing runtime execution options.
+
+    Args:
+        backend_name: target backend to run on. This is required for legacy runtime.
+        image: the runtime image used to execute the program, specified in
+            the form of ``image_name:tag``. Not all accounts are
+            authorized to select a different image.
+        log_level: logging level to set in the execution environment. The valid
+            log levels are: ``DEBUG``, ``INFO``, ``WARNING``, ``ERROR``, and ``CRITICAL``.
+            The default level is ``WARNING``.
+    """
+
+    backend_name: Optional[str] = None
+    image: Optional[str] = None
+    log_level: Optional[str] = None
+
+    def validate(self, auth: str) -> None:
+        """Validate options.
+
+        Args:
+            auth: authentication type.
+
+        Raises:
+            IBMInputValueError: If one or more option is invalid.
+        """
+        if self.image and not re.match(
+            "[a-zA-Z0-9]+([/.\\-_][a-zA-Z0-9]+)*:[a-zA-Z0-9]+([.\\-_][a-zA-Z0-9]+)*$",
+            self.image,
+        ):
+            raise IBMInputValueError('"image" needs to be in form of image_name:tag')
+
+        if auth == "legacy" and not self.backend_name:
+            raise IBMInputValueError(
+                '"backend_name" is required field in "options" for legacy runtime.'
+            )
+
+        if self.log_level and not isinstance(
+            logging.getLevelName(self.log_level.upper()), int
+        ):
+            raise IBMInputValueError(
+                f"{self.log_level} is not a valid log level. The valid log levels are: `DEBUG`, "
+                f"`INFO`, `WARNING`, `ERROR`, and `CRITICAL`."
+            )

test/ibm_test_case.py

@@ -174,6 +174,7 @@ def _run_program(
         final_result=None,
         callback=None,
         backend=None,
+        log_level=None,
     ):
         """Run a program."""
         self.log.debug("Running program on %s", service.auth)
@@ -190,7 +191,7 @@ def _run_program(
         backend_name = (
             backend if backend is not None else self.sim_backends[service.auth]
         )
-        options = {"backend_name": backend_name}
+        options = {"backend_name": backend_name, "log_level": log_level}
         job = service.run(
             program_id=pid, inputs=inputs, options=options, callback=callback
         )

test/mock/fake_runtime_client.py

@@ -113,6 +113,7 @@ def __init__(
         final_status,
         params,
         image,
+        log_level=None,
     ):
         """Initialize a fake job."""
         self._job_id = job_id
@@ -125,6 +126,7 @@ def __init__(
         self._params = params
         self._image = image
         self._interim_results = json.dumps("foo")
+        self.log_level = log_level
         if final_status is None:
             self._future = self._executor.submit(self._auto_progress)
             self._result = None
@@ -345,6 +347,7 @@ def program_run(
         params: Dict,
         image: str,
         hgp: Optional[str],
+        log_level: Optional[str],
     ):
         """Run the specified program."""
         _ = self._get_program(program_id)
@@ -372,6 +375,7 @@ def program_run(
             params=params,
             final_status=self._final_status,
             image=image,
+            log_level=log_level,
             **self._job_kwargs,
         )
         self._jobs[job_id] = job

test/test_integration_job.py

@@ -62,6 +62,21 @@ def test_run_program_cloud_no_backend(self):
         job = self._run_program(service, backend="")
         self.assertTrue(job.backend, f"Job {job.job_id} has no backend.")
 
+    @run_cloud_legacy_real
+    def test_run_program_log_level(self, service):
+        """Test running with a custom log level."""
+        levels = ["INFO", "ERROR"]
+        for level in levels:
+            with self.subTest(level=level):
+                job = self._run_program(service, log_level=level)
+                job.wait_for_final_state()
+                expect_info_msg = level == "INFO"
+                self.assertEqual(
+                    "info log" in job.logs(),
+                    expect_info_msg,
+                    f"Job log is {job.logs()}",
+                )
+
     @run_cloud_legacy_real
     def test_run_program_failed(self, service):
         """Test a failed program execution."""

test/test_jobs.py

@@ -150,6 +150,13 @@ def test_run_program_with_custom_runtime_image(self, service):
         self.assertTrue(job.result())
         self.assertEqual(job.image, image)
 
+    @run_legacy_and_cloud_fake
+    def test_run_program_with_custom_log_level(self, service):
+        """Test running program with a custom image."""
+        job = run_program(service=service, log_level="DEBUG")
+        job_raw = service._api_client._get_job(job.job_id)
+        self.assertEqual(job_raw.log_level, "DEBUG")
+
     @run_legacy_and_cloud_fake
     def test_run_program_failed(self, service):
         """Test a failed program execution."""

test/utils/program.py

@@ -72,10 +72,11 @@ def run_program(
     image="",
     instance=None,
     backend_name=None,
+    log_level=None,
 ):
     """Run a program."""
     backend_name = backend_name if backend_name is not None else "common_backend"
-    options = {"backend_name": backend_name}
+    options = {"backend_name": backend_name, "image": image, "log_level": log_level}
     if final_status is not None:
         service._api_client.set_final_status(final_status)
     elif job_classes:
@@ -87,7 +88,6 @@ def run_program(
         options=options,
         inputs=inputs,
         result_decoder=decoder,
-        image=image,
         instance=instance,
     )
     return job

test/utils/templates.py

@@ -16,10 +16,13 @@
 import random
 import time
 import warnings
+import logging
 
 from qiskit import transpile
 from qiskit.circuit.random import random_circuit
 
+logger = logging.getLogger("qiskit-test")
+
 def prepare_circuits(backend):
     circuit = random_circuit(num_qubits=5, depth=4, measure=True,
                              seed=random.randint(0, 1000))
@@ -39,6 +42,7 @@ def main(backend, user_messenger, **kwargs):
     user_messenger.publish(final_result, final=True)
     print("this is a stdout message")
     warnings.warn("this is a stderr message")
+    logger.info("this is an info log")
     """
 
 RUNTIME_PROGRAM_METADATA = {