feat: exporter prometheus_client /metrics (JEP-0013 Phase 2) - #934
feat: exporter prometheus_client /metrics (JEP-0013 Phase 2)#934RoddieKieley wants to merge 1 commit 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.
|
Warning Review limit reached
Next review available in: 59 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (9)
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?
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).