feat: exporter prometheus_client /metrics (JEP-0013 Phase 2) - #934
feat: exporter prometheus_client /metrics (JEP-0013 Phase 2)#934RoddieKieley wants to merge 2 commits into
Conversation
- Add exporter-local `prometheus_client` registry with JEP-named series: `jumpstarter_operations_total`, `jumpstarter_operation_duration_seconds`, `jumpstarter_operation_errors_total`, `jumpstarter_stream_bytes_total`, `jumpstarter_active_sessions`, plus exemplars (`client`, `lease_id`). - Expose HTTP `GET /metrics` on the exporter process for lab/dev scrape (same registry Phase 3 will later reverse-scrape via MetricsStream). - Minimal core-path wiring so series increment under test; full per-driver telemetry architecture remains Phase 4.
📝 WalkthroughWalkthroughThe exporter adds a local Prometheus registry, metrics HTTP server, driver and stream instrumentation, active-session tracking, and configurable CLI bind-address propagation. ChangesExporter metrics
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant CLI as run command
participant Child as exporter child
participant Server as metrics HTTP server
participant Registry as MetricsRegistry
participant Driver as driver handler
CLI->>Child: pass metrics_bind_address
Child->>Server: start metrics server
Server->>Registry: expose /metrics
Driver->>Registry: record operations and stream bytes
Registry-->>Server: return OpenMetrics data
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
| @@ -0,0 +1,17 @@ | |||
| """Exporter-local Prometheus metrics (JEP-0013 Phase 2).""" | |||
There was a problem hiding this comment.
I see we do use top level comments in some places, but i'd drop the JEP part at least, from the others as well
| except Exception as e: | ||
| self._record_operation_metrics( | ||
| operation=op, | ||
| result="failure", | ||
| duration_seconds=time.perf_counter() - started, | ||
| error_type="internal_error", | ||
| ) | ||
| self.logger.warning( | ||
| "Operation failed", | ||
| extra={"operation": op, "driver_type": self.driver_type, |
There was a problem hiding this comment.
Consider adding except grpc.aio.AbortError: raise immediately before except Exception in both methods to let abort errors propagate without distorting metrics or status codes.
| self, | ||
| *, | ||
| exporter: str, | ||
| operation: str, |
There was a problem hiding this comment.
Using plain str means a typo at any call site silently creates a new Prometheus time series. Consider using Literal types for all bounded label parameters, e.g. result: Literal["success", "failure"].
| result=encode_value(result), | ||
| ) | ||
| except NotImplementedError as e: | ||
| self._record_operation_metrics( | ||
| operation=op, | ||
| result="failure", | ||
| duration_seconds=time.perf_counter() - started, | ||
| error_type="not_implemented", | ||
| ) | ||
| self.logger.warning( | ||
| "Operation failed", | ||
| extra={"operation": op, "driver_type": self.driver_type, | ||
| "result": "failure", "error_type": "not_implemented"}, | ||
| ) | ||
| await context.abort(StatusCode.UNIMPLEMENTED, str(e)) | ||
| except ValueError as e: | ||
| self._record_operation_metrics( | ||
| operation=op, | ||
| result="failure", | ||
| duration_seconds=time.perf_counter() - started, | ||
| error_type="validation_error", | ||
| ) | ||
| self.logger.warning( | ||
| "Operation failed", | ||
| extra={"operation": op, "driver_type": self.driver_type, | ||
| "result": "failure", "error_type": "validation_error"}, | ||
| ) | ||
| await context.abort(StatusCode.INVALID_ARGUMENT, str(e)) | ||
| except TimeoutError as e: | ||
| self._record_operation_metrics( | ||
| operation=op, | ||
| result="failure", | ||
| duration_seconds=time.perf_counter() - started, | ||
| error_type="timeout", | ||
| ) | ||
| self.logger.warning( | ||
| "Operation failed", | ||
| extra={"operation": op, "driver_type": self.driver_type, | ||
| "result": "failure", "error_type": "timeout"}, | ||
| ) | ||
| await context.abort(StatusCode.DEADLINE_EXCEEDED, str(e)) | ||
| except ConnectionError as e: | ||
| self._record_operation_metrics( | ||
| operation=op, | ||
| result="failure", | ||
| duration_seconds=time.perf_counter() - started, | ||
| error_type="connection_error", | ||
| ) | ||
| self.logger.warning( | ||
| "Operation failed", | ||
| extra={"operation": op, "driver_type": self.driver_type, | ||
| "result": "failure", "error_type": "connection_error"}, | ||
| ) | ||
| await context.abort(StatusCode.UNAVAILABLE, str(e)) | ||
| except OSError as e: | ||
| self._record_operation_metrics( | ||
| operation=op, | ||
| result="failure", | ||
| duration_seconds=time.perf_counter() - started, | ||
| error_type="device_error", | ||
| ) | ||
| self.logger.warning( | ||
| "Operation failed", | ||
| extra={"operation": op, "driver_type": self.driver_type, | ||
| "result": "failure", "error_type": "device_error"}, | ||
| ) | ||
| await context.abort(StatusCode.INTERNAL, str(e)) | ||
| except Exception as e: | ||
| self._record_operation_metrics( | ||
| operation=op, | ||
| result="failure", | ||
| duration_seconds=time.perf_counter() - started, | ||
| error_type="internal_error", | ||
| ) | ||
| self.logger.warning( | ||
| "Operation failed", | ||
| extra={"operation": op, "driver_type": self.driver_type, |
There was a problem hiding this comment.
Consider adding test drivers whose @export methods raise each exception type (e.g., TimeoutError), then assert the correct error_type label appears in the metrics output.
| try: | ||
| get_registry().inc_active_sessions(exporter=self.name, delta=-1.0) | ||
| except Exception: | ||
| pass |
There was a problem hiding this comment.
Log exception at WARNING or DEBUG level before suppressing it.
| def extra_labels(self) -> dict[str, str]: | ||
| return {} | ||
|
|
||
| def _record_operation_metrics( | ||
| self, | ||
| *, | ||
| operation: str, | ||
| result: str, | ||
| duration_seconds: float, | ||
| error_type: str | None = None, | ||
| ) -> None: | ||
| from jumpstarter.metrics.registry import ( | ||
| exemplars_from_log_context, | ||
| exporter_from_log_context, | ||
| get_registry, | ||
| ) | ||
|
|
||
| get_registry().record_operation( | ||
| exporter=exporter_from_log_context(default=self.name if hasattr(self, "name") else "unknown"), | ||
| operation=operation, | ||
| result=result, | ||
| driver_type=self.driver_type, | ||
| duration_seconds=duration_seconds, | ||
| exemplars=exemplars_from_log_context(), | ||
| error_type=error_type, | ||
| ) | ||
|
|
||
| async def DriverCall(self, request, context): | ||
| """ | ||
| :meta private: | ||
| """ | ||
| op = request.method | ||
| started = time.perf_counter() | ||
| self.logger.info( | ||
| "Operation started", | ||
| extra={"operation": op, "driver_type": self.driver_type}, |
There was a problem hiding this comment.
The operation label passed to _record_operation_metrics comes from request.method, a client-controlled gRPC field. When __lookup_drivercall rejects an unknown method, the resulting AbortError falls through to the catch-all handler, which records metrics with the untrusted value as a Prometheus label. Each unique method name creates a new time series, allowing an authenticated client to exhaust exporter memory by sending random method names. Fixing the AbortError handling (catching it before the generic handler) would largely mitigate this.
AI-generated, human reviewed
| def filter_exemplars(exemplars: dict[str, str] | None) -> dict[str, str] | None: | ||
| """Keep only the JEP default exemplar keys with non-empty values.""" | ||
| if not exemplars: | ||
| return None | ||
| filtered = { | ||
| key: str(exemplars[key]) | ||
| for key in DEFAULT_EXEMPLAR_KEYS | ||
| if key in exemplars and exemplars[key] is not None and str(exemplars[key]) != "" | ||
| } | ||
| return filtered or None | ||
|
|
||
|
|
||
| def exemplars_from_log_context() -> dict[str, str] | None: | ||
| """Build default exemplars from the current structlog contextvars.""" | ||
| ctx = structlog.contextvars.get_contextvars() | ||
| return filter_exemplars({key: str(ctx[key]) for key in DEFAULT_EXEMPLAR_KEYS if key in ctx}) | ||
|
|
||
|
|
||
| def exporter_from_log_context(default: str = "unknown") -> str: | ||
| ctx = structlog.contextvars.get_contextvars() | ||
| value = ctx.get("exporter") | ||
| return str(value) if value else default | ||
|
|
There was a problem hiding this comment.
filter_exemplars, exemplars_from_log_context, and exporter_from_log_context contain non-trivial logic (filtering empty values, reading from structlog contextvars, defaulting to "unknown") but have no direct test coverage.
| with serve(MockPower()) as client: | ||
| client.on() | ||
| body = get_registry().generate_latest().decode() | ||
| assert "jumpstarter_active_sessions" in body | ||
| assert "jumpstarter_operations_total" in body | ||
| assert 'driver_type="power"' in body | ||
| assert 'result="success"' in body | ||
| assert 'operation="on"' in body or 'operation="On"' in body |
There was a problem hiding this comment.
Consider parsing the exposition output or using get_sample_value() to verify jumpstarter_operations_total equals 1.0 for the specific label combination.
| def _record_operation_metrics( | ||
| self, | ||
| *, | ||
| operation: str, | ||
| result: str, | ||
| duration_seconds: float, | ||
| error_type: str | None = None, | ||
| ) -> None: | ||
| from jumpstarter.metrics.registry import ( | ||
| exemplars_from_log_context, | ||
| exporter_from_log_context, | ||
| get_registry, | ||
| ) | ||
|
|
||
| get_registry().record_operation( | ||
| exporter=exporter_from_log_context(default=self.name if hasattr(self, "name") else "unknown"), | ||
| operation=operation, | ||
| result=result, | ||
| driver_type=self.driver_type, | ||
| duration_seconds=duration_seconds, | ||
| exemplars=exemplars_from_log_context(), | ||
| error_type=error_type, | ||
| ) |
There was a problem hiding this comment.
Consider moving the imports to module level or caching the resolved functions. Capture the exporter name and exemplars once before the stream loop.
| ) | ||
| self._errors = Counter( | ||
| "jumpstarter_operation_errors_total", | ||
| "Errors by class (timeout, device, …).", |
There was a problem hiding this comment.
Consider replacing ellipsis with ...
| ) -> None: | ||
| if nbytes <= 0: | ||
| return | ||
| self._stream_bytes.labels( | ||
| exporter=exporter, | ||
| driver_type=driver_type, | ||
| direction=direction, | ||
| ).inc(nbytes, exemplar=filter_exemplars(exemplars)) |
There was a problem hiding this comment.
What happens to the metric increment for zero or negative byte counts?
There was a problem hiding this comment.
In theory this should not happen, according to the guided review of the functionality anyway, as ..."the program deliberately does not record zero/negative amounts into the counter."
Re-raise grpc AbortError before generic handlers to avoid client-controlled operation label cardinality, tighten bounded metric label types, default metrics bind host to loopback, log session metric decrements on failure, and expand exporter metrics unit coverage.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
python/packages/jumpstarter/jumpstarter/driver/base.py (1)
161-173: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winRecord unary success after response serialization.
encode_value(result)can raise after_record_operation_metrics(..., result="success")records success, then the generic handler records the same RPC asresult="failure". Build theDriverCallResponsebefore recording success.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/packages/jumpstarter/jumpstarter/driver/base.py` around lines 161 - 173, The unary operation handler currently records success before response serialization can fail. In the surrounding operation flow, construct the DriverCallResponse and run encode_value(result) before calling _record_operation_metrics with result="success"; keep the success log and return after serialization succeeds so encoding failures are handled only as failures.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@python/packages/jumpstarter/jumpstarter/driver/base.py`:
- Around line 161-173: The unary operation handler currently records success
before response serialization can fail. In the surrounding operation flow,
construct the DriverCallResponse and run encode_value(result) before calling
_record_operation_metrics with result="success"; keep the success log and return
after serialization succeeds so encoding failures are handled only as failures.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 75605a6c-9f92-4070-b4a6-707a6f3fe110
⛔ Files ignored due to path filters (1)
python/uv.lockis excluded by!**/*.lock
📒 Files selected for processing (9)
python/packages/jumpstarter-cli/jumpstarter_cli/run.pypython/packages/jumpstarter/jumpstarter/driver/base.pypython/packages/jumpstarter/jumpstarter/exporter/session.pypython/packages/jumpstarter/jumpstarter/metrics/__init__.pypython/packages/jumpstarter/jumpstarter/metrics/metrics_test.pypython/packages/jumpstarter/jumpstarter/metrics/registry.pypython/packages/jumpstarter/jumpstarter/metrics/server.pypython/packages/jumpstarter/jumpstarter/streams/common.pypython/packages/jumpstarter/pyproject.toml
3 participants
prometheus_clientregistry with JEP-named series:jumpstarter_operations_total,jumpstarter_operation_duration_seconds,jumpstarter_operation_errors_total,jumpstarter_stream_bytes_total,jumpstarter_active_sessions, plus exemplars (client,lease_id).GET /metricson the exporter process for lab/dev scrape (same registry Phase 3 will later reverse-scrape via MetricsStream).