QFw: write a job's story in five log lines, stitched to its trace

backends/qfw_telemetry/README.md

@@ -63,7 +63,7 @@ change and never a code change.
 | `QFW_TELEMETRY_SAMPLE` | `off`, `always`, ratio | `off` | Trace sampling |
 | `QFW_TELEMETRY_DIR` | path | node-local tmp | Export directory, file profile |
 | `QFW_TELEMETRY_TRANSPORT` | `0`, `1` | `0` | DEFw RPC spans |
-| `QFW_TELEMETRY_LOGS` | `off`, `error`, `warning`, `info`, `debug` | `off` | The logs tier: the root logger's records at that level or above, exported with the current span's ids. See below |
+| `QFW_TELEMETRY_LOGS` | `off`, `error`, `warning`, `info`, `debug`, `all` | `off` | The logs tier: the root logger's records at that level or above, exported with the current span's ids; `all` adds DEFw's transport internals. See below |
 | `QFW_TELEMETRY_ENDPOINT` | URL | SDK default | The collector's OTLP/HTTP base URL for the otlp profile, such as `http://otel-collector:4318`; the signal paths are appended. Unset, the exporters read the standard `OTEL_EXPORTER_OTLP_*` variables |
 | `OTEL_METRIC_EXPORT_INTERVAL` | milliseconds | `10000` | How often metrics export. The SDK's own default is a minute; ten seconds suits a dashboard and bounds what a killed service loses |
 
@@ -184,14 +184,26 @@ call site changes: DEFw's own logging goes through the root logger, so the
 lines a QPM writes while it executes a circuit carry that job's trace, and
 the collector profile can show them under the job's waterfall.
 
+With the tier on, QFw's instrumentation layer also writes a job's story
+itself, five debug lines stitched to the job's trace whatever the backend:
+`submitted job ...` and `job ... completed after ... ms` on the client,
+`received circuit ...`, `executing circuit ... on <device> via <api path>` and
+`circuit ... completed on <device> after ... ms` on the QPM. A backend that
+logs nothing of its own, the fake IQM for instance, still shows those under
+its waterfall. With the tier off they cost a boolean test and write nothing.
+
 Three things to know before turning it on:
 
+- QFw's own code writes a job's story at `debug`: a QPM's device query,
+  the circuit's qubit cap, the driver's progress. `debug` is therefore the
+  tier that shows a job's lines under its trace, and `error` carries real
+  errors only.
 - DEFw's levels 30 to 35 are categories, not severities: CORE, WORKER,
   SERVICE, APP, RPC and STACKTRACE. Its service and application lines
-  (SERVICE, APP) go out at `warning` and `info`, like a warning would.
-  Its transport internals (CORE, WORKER, RPC, STACKTRACE), hundreds of
-  lines per job about work requests and RPC handling with routine stack
-  dumps, leave only at `debug`. `error` carries real errors only. The root
+  (SERVICE, APP) go out at `warning` and above, like a warning would. Its
+  transport internals (CORE, WORKER, RPC, STACKTRACE), hundreds of lines
+  per job about work requests and RPC handling with routine stack dumps,
+  leave only with `all`, whatever tier they would otherwise pass. The root
   logger's own level, which DEFw sets from `DEFW_LOG_LEVEL`, applies first;
   the handler only narrows, and a process whose DEFw level selects nothing
   exports nothing.
@@ -203,8 +215,11 @@ Three things to know before turning it on:
 
 Under the file profile the records go to `<service>-<rank>-<pid>.logs.jsonl`
 beside the spans and metrics files, one OTLP/JSON export per line. DEFw's
-`set_logging_level_helper` removes every handler on the root logger, so a
-process that changes its DEFw log level after `configure()` loses the tier.
+`set_logging_level_helper` removes every handler on the root logger, and a
+service does set its level after `configure()`, so `logs_enabled()` puts the
+tier's handler back whenever it finds it gone; QFw's own call sites ask it
+before they write a line, and the first line of the next job restores the
+tier for everything that follows.
 
 ## Trace context across DEFw RPC
 

backends/qfw_telemetry/__init__.py

@@ -38,16 +38,16 @@
 QFW_TELEMETRY_DIR        export directory for the file profile
 QFW_TELEMETRY_SAMPLE     off | always | <ratio>   (default: off)
 QFW_TELEMETRY_TRANSPORT  0 | 1                    (default: 0)
-QFW_TELEMETRY_LOGS       off | error | warning | info | debug (default: off)
+QFW_TELEMETRY_LOGS       off | error | warning | info | debug | all (default: off)
 QFW_TELEMETRY_ENDPOINT   the collector's OTLP/HTTP base URL, otlp profile
 
 The logs tier is the optional third signal. With a level set, the SDK's
 logging handler joins the root logger, so every record Python's logging
 carries at that level or above is exported with the trace and span ids of
 whatever span is current when it is written. That is how a log line is
-stitched to its job without a change to any call site. DEFw's service and
-application lines count as warnings; its transport internals leave only
-at debug.
+stitched to its job without a change to any call site. QFw's own lines
+are mostly debug, so debug is the tier that shows a job's story; DEFw's
+transport internals leave only with all.
 """
 
 import logging
@@ -66,7 +66,9 @@
 	"warning": logging.WARNING,
 	"info": logging.INFO,
 	"debug": logging.DEBUG,
+	"all": logging.DEBUG,
 }
+LOGS_ALL = "all"
 
 # Loggers whose records never leave through the logs tier: the SDK and the
 # HTTP stack it exports with. A failed export logs a warning; exporting that
@@ -77,9 +79,11 @@
 # DEFw's levels 30 to 35 are categories, not severities: CORE, WORKER,
 # SERVICE, APP, RPC and STACKTRACE, registered by name. Four of them are the
 # transport's own internals, hundreds of lines per job about work requests
-# and RPC handling, with routine stack dumps. Those leave only at debug.
-# SERVICE and APP are what QFw's services and applications write, and go
-# out at warning and above like a warning would.
+# and RPC handling, with routine stack dumps. Those leave only with "all".
+# SERVICE and APP are what QFw's services and applications write through
+# DEFw, and go out at warning and above like a warning would. QFw's own
+# code mostly logs at debug, so "debug" is the tier that tells a job's
+# story without the transport's.
 _DEFW_INTERNAL_CATEGORIES = (
 	"DEFW_CORE", "DEFW_WORKER", "DEFW_RPC", "DEFW_STACKTRACE")
 
@@ -241,6 +245,11 @@ def _logs_level():
 		f"{', '.join(repr(name) for name in LOG_LEVELS)}: got {value!r}")
 
 
+def _logs_keep_internals():
+	"""True only for the "all" tier, which carries DEFw's transport chatter."""
+	return _env(TELEMETRY_LOGS_ENV, "off").lower() == LOGS_ALL
+
+
 def _export_dir():
 	configured = _env(TELEMETRY_DIR_ENV)
 	if configured:
@@ -376,12 +385,12 @@ def _file_log_processor(service_name):
 class _LogSourceFilter(logging.Filter):
 	"""
 	Keeps the exporter's own loggers out of the export, and DEFw's transport
-	internals out of every tier but debug.
+	internals out of every tier but "all".
 	"""
 
-	def __init__(self, level):
+	def __init__(self, keep_internals):
 		super().__init__()
-		self._keep_internals = level <= logging.DEBUG
+		self._keep_internals = keep_internals
 
 	def filter(self, record):
 		name = record.name or ""
@@ -393,7 +402,7 @@ def filter(self, record):
 		return record.levelname not in _DEFW_INTERNAL_CATEGORIES
 
 
-def _install_log_handler(logger_provider, level):
+def _install_log_handler(logger_provider, level, keep_internals=False):
 	"""
 	Put the SDK's handler on the root logger at the given level.
 
@@ -411,7 +420,7 @@ def _install_log_handler(logger_provider, level):
 		from opentelemetry.sdk._logs import LoggingHandler
 
 		handler = LoggingHandler(level=level, logger_provider=logger_provider)
-	handler.addFilter(_LogSourceFilter(level))
+	handler.addFilter(_LogSourceFilter(keep_internals))
 	logging.getLogger().addHandler(handler)
 	return handler
 
@@ -570,7 +579,7 @@ def configure(service_name, service_version=None, role=None, attributes=None):
 			_STATE.logger_provider = logger_provider
 			_STATE.logs_level = logs_level
 			_STATE.log_handler = _install_log_handler(
-				logger_provider, logs_level)
+				logger_provider, logs_level, _logs_keep_internals())
 
 		_STATE.profile = profile
 		_STATE.tracer_provider = tracer_provider
@@ -616,8 +625,22 @@ def transport_spans_enabled():
 
 
 def logs_enabled():
-	"""True when the logs tier is exporting the root logger's records."""
-	return _STATE.log_handler is not None
+	"""
+	True when the logs tier is exporting the root logger's records.
+
+	DEFw's set_logging_level_helper removes every handler from the root
+	logger whenever a process sets or changes its DEFw log level, which a
+	service does after QFw has configured telemetry. So this does not only
+	answer; it puts the tier's handler back if it has gone, and QFw's own
+	call sites ask before they write a line.
+	"""
+	handler = _STATE.log_handler
+	if handler is None:
+		return False
+	root = logging.getLogger()
+	if handler not in root.handlers:
+		root.addHandler(handler)
+	return True
 
 
 def tracer():
@@ -676,7 +699,7 @@ def counter(name):
 
 
 def use_providers(tracer_provider, meter_provider=None, profile=PROFILE_FILE,
-		  logger_provider=None, logs_level=None):
+		  logger_provider=None, logs_level=None, logs_internals=False):
 	"""
 	Adopt providers built elsewhere instead of building them from the
 	environment, and register the DEFw propagation hooks for them.
@@ -708,7 +731,7 @@ def use_providers(tracer_provider, meter_provider=None, profile=PROFILE_FILE,
 			else logs_level if logs_level is not None else logging.INFO)
 		if logger_provider is not None:
 			_STATE.log_handler = _install_log_handler(
-				logger_provider, _STATE.logs_level)
+				logger_provider, _STATE.logs_level, logs_internals)
 		_STATE.defw_hooks = _register_defw_trace_hooks()
 		_STATE.configured = True
 		return _STATE.profile

docs/design/benchmarking.md

@@ -222,9 +222,13 @@ and SDKs for all benchmarking/profiling telemetry, using all three signals:
   timeline. Implemented (October 2026) as an optional logs tier,
   `QFW_TELEMETRY_LOGS`, off by default: the SDK's logging handler on the
   root logger exports each record with the current span's trace and span
-  ids, so the stitching costs no call-site change. DEFw's levels are
-  categories: its service and application lines go out at `warning`, its
-  transport internals only at `debug`, and `error` carries real errors only.
+  ids, so the stitching costs no call-site change. QFw's own lines are
+  mostly `debug`, so that tier shows a job's story; DEFw's levels are
+  categories, its service and application lines go out at `warning`, its
+  transport internals only with `all`, and `error` carries real errors only.
+  The instrumentation layer writes the story's five milestone lines itself
+  (submitted, received, executing, done, job done), while the job's span is
+  current, so every backend has lines under its waterfall.
 
 Adopting the standard has consequences the bespoke format could not offer:
 

services/util/instrumentation.py

@@ -115,6 +115,27 @@
 _EXECUTION_LABELS = contextvars.ContextVar(
 	"qfw_execution_labels", default=None)
 
+# A job's story in a few log lines, for the logs tier. Each line is written
+# while the job's span is current, so the tier stamps it with the trace and
+# the logs panel under a waterfall has something to show for every backend,
+# the fake IQM included. Debug, like the rest of what QFw logs about a job,
+# and only when the tier is on: with it off these cost a boolean test.
+_QPM_LOG = logging.getLogger("qfw.qpm")
+_CLIENT_LOG = logging.getLogger("qfw.client")
+
+
+def _story(span, logger, message, *args):
+	logs_enabled = getattr(_telemetry, "logs_enabled", None)
+	if logs_enabled is None or not logs_enabled():
+		return
+	try:
+		with _trace_api().use_span(
+				span, end_on_exit=False, record_exception=False,
+				set_status_on_exception=False):
+			logger.debug(message, *args)
+	except Exception as exc:
+		_LOG.debug("story line not written: %s", exc)
+
 
 def configure_process(role, device=None, attributes=None):
 	"""
@@ -248,6 +269,12 @@ def bind_circuit(circuit, qtask_id=None):
 		from opentelemetry import context
 		circuit.otel_context = context.get_current()
 		info = getattr(circuit, "info", None) or {}
+		_story(
+			_trace_api().get_current_span(), _QPM_LOG,
+			"received circuit %s (qtask %s): %s qubits, %s shots",
+			circuit.get_cid(), _text(info.get("qtask_id")),
+			_int(info.get("num_qubits")),
+			_int(info.get("num_shots", info.get("shots"))))
 		_set(_trace_api().get_current_span(), {
 			ATTR_CID: circuit.get_cid(),
 			ATTR_QTASK_ID: _text(qtask_id),
@@ -384,6 +411,9 @@ def begin_backend_execution(circuit, api_path, device=None, backend_kind=None):
 			ATTR_NUM_QUBITS: _int(info.get("num_qubits")),
 		})
 		_set(span, values)
+		_story(
+			span, _QPM_LOG, "executing circuit %s on %s via %s",
+			circuit.get_cid(), device or "?", api_path)
 		return _Execution(span, labels, time.monotonic())
 	except Exception as exc:
 		_LOG.debug("qfw.backend.execute not opened: %s", exc)
@@ -416,6 +446,11 @@ def finish_backend_execution(execution, circuit, error=None, cancel_event=None):
 		if error is not None:
 			_mark_error(execution.span, error)
 		_set(execution.span, {ATTR_OUTCOME: outcome})
+		_story(
+			execution.span, _QPM_LOG, "circuit %s %s on %s after %.1f ms",
+			circuit.get_cid(), outcome,
+			execution.labels.get(ATTR_DEVICE, "?"),
+			(time.monotonic() - execution.started) * 1000.0)
 		execution.span.end()
 	except Exception as exc:
 		_LOG.debug("qfw.backend.execute not closed: %s", exc)
@@ -546,6 +581,9 @@ def start_job(job_id, circuits, shots, properties=None):
 			ATTR_SHOTS: _int(shots),
 		})
 		_set(span, values)
+		_story(
+			span, _CLIENT_LOG, "submitted job %s: %s circuit(s), %s shots",
+			job_id, _int(circuits), _int(shots))
 		return _JobTrace(span, labels, time.monotonic())
 	except Exception as exc:
 		_LOG.debug("qfw.app.job not opened: %s", exc)
@@ -576,6 +614,10 @@ def end_job(job, outcome, error=None):
 		if error is not None:
 			_mark_error(job.span, error)
 		_set(job.span, {ATTR_OUTCOME: outcome})
+		_story(
+			job.span, _CLIENT_LOG, "job %s %s after %.1f ms",
+			job.span.attributes.get(ATTR_JOB_ID) if getattr(job.span, "attributes", None) else "?",
+			outcome, (time.monotonic() - job.started) * 1000.0)
 		job.span.end()
 	except Exception as exc:
 		_LOG.debug("qfw.app.job not closed: %s", exc)

tests/mock/test_instrumentation.py

@@ -99,10 +99,14 @@ def recording():
 		SimpleLogRecordProcessor(log_exporter))
 	qfw_telemetry.use_providers(
 		tracer_provider, meter_provider,
-		logger_provider=logger_provider, logs_level=logging.WARNING)
+		logger_provider=logger_provider, logs_level=logging.DEBUG)
 	assert instrumentation.enabled()
 	assert qfw_telemetry.logs_enabled()
+	# The story lines are debug; the root logger must let them through.
+	root_level = logging.getLogger().level
+	logging.getLogger().setLevel(logging.DEBUG)
 	yield _Recording(exporter, reader, log_exporter)
+	logging.getLogger().setLevel(root_level)
 	qfw_telemetry.shutdown()
 	assert not instrumentation.enabled()
 
@@ -164,6 +168,16 @@ def test_qpm_run_is_one_trace_with_every_hop(monkeypatch, tmp_path, recording):
 	assert completion["outcome"] == "COMPLETED"
 	assert ("job job-trace submitted", job.get_span_context().trace_id) in \
 		recording.logs()
+	# The QPM's own story of the circuit, each line stitched to the trace:
+	# received while the receive handler ran, then executing and done.
+	trace_id = job.get_span_context().trace_id
+	story = [body for body, tid in recording.logs() if tid == trace_id
+		and body.startswith(("received circuit", "executing circuit", "circuit "))]
+	cid = response["cid"]
+	assert story[0] == f"received circuit {cid} (qtask 1): 4 qubits, 64 shots"
+	assert story[1] == f"executing circuit {cid} on {FAKE_IQM_TARGET_ID} via simulator"
+	assert story[2].startswith(f"circuit {cid} completed on {FAKE_IQM_TARGET_ID} after ")
+	assert len(story) == 3
 
 	spans = recording.spans_by_name()
 	assert set(spans) == {
@@ -313,6 +327,12 @@ def test_qiskit_job_is_the_trace_root_and_counts_itself(monkeypatch, recording):
 	assert fake_qpm.contexts == [root.context]
 
 	assert root.attributes["qfw.job.id"] == job._job_id
+	# The client's two lines of the story, under the job's trace.
+	assert [body for body, tid in recording.logs()
+		if tid == root.context.trace_id and body.startswith(("submitted job", "job "))] == [
+		f"submitted job {job._job_id}: 1 circuit(s), 3 shots",
+		next(body for body, _ in recording.logs()
+			if body.startswith(f"job {job._job_id} completed after"))]
 	assert root.attributes["qfw.app.circuits"] == 1
 	assert root.attributes["qfw.circuit.shots"] == 3
 	assert root.attributes["qfw.device.name"] == FAKE_IQM_TARGET_ID