diff --git a/.env.sample b/.env.sample index 20ec9860..75b02a9e 100644 --- a/.env.sample +++ b/.env.sample @@ -122,3 +122,5 @@ ENV= OTEL_SDK_DISABLED=true # gRPC endpoint of the OTel collector, e.g. http://localhost:4317 OTEL_EXPORTER_OTLP_ENDPOINT= +# Port for the PIXL RabbitMQ metrics endpoint, which is scraped by Prometheus to collect queue backlog depth. +RABBITMQ_METRICS_PORT=15692 diff --git a/docker-compose.yml b/docker-compose.yml index ded0743e..a78df44f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -66,7 +66,7 @@ x-otel-common: &otel-common OTEL_EXPORTER_OTLP_PROTOCOL: grpc OTEL_LOGS_EXPORTER: none # we define our own loguru sink for exporting logs OTEL_TRACES_EXPORTER: otlp - OTEL_METRICS_EXPORTER: none + OTEL_METRICS_EXPORTER: otlp x-logs-volume: &logs-volume type: volume @@ -265,6 +265,7 @@ services: ports: - "127.0.0.1:${RABBITMQ_PORT}:5672" - "${RABBITMQ_ADMIN_PORT}:15672" + - "${RABBITMQ_METRICS_PORT}:15692" networks: - pixl-net volumes: diff --git a/docs/setup/developer.md b/docs/setup/developer.md index 7917c71c..6981915b 100644 --- a/docs/setup/developer.md +++ b/docs/setup/developer.md @@ -169,3 +169,35 @@ When adding context: - Use the same field names across services so logs can be joined up - If a field has been pseudonymised, bind a new field prefixed with `pseudo_`, e.g. `study_uid` becomes `pseudo_study_uid` after pseudonymisation + +### Adding metrics + +Custom metrics are defined centrally in [`core.metrics`](../../pixl_core/src/core/metrics.py). +To add a new metric: + +1. Add a field for it on the `PixlMetrics` dataclass, and create the instrument + (e.g. a counter) in `initialise_metrics()`. Metric names use dots as + separators, e.g. `pixl.studies.exported`. +2. Add a `record_*` helper that records a value on the instrument. Guard against + the instrument being `None` (it is unset when telemetry is disabled) and + return early if so. +3. Call the `record_*` helper from the relevant service(s). + +You can pass `attributes` to the metric that can later be used for filtering and +aggregation, e.g. `project_name`. It's highly recommended to keep attribute values +**low-cardinality** - each distinct combination of attribute values creates a separate +time series, so avoid unbounded values like raw IDs or full tracebacks. + +### RabbitMQ queue metrics + +The RabbitMQ Docker image includes a Prometheus endpoint that exposes queue backlog depth metrics. +We use this to scrape queue depth metrics for each queue in the PIXL RabbitMQ broker, rather +than manually defining a metric within PIXL. This does, however, require defining a scrape job in +the [Prometheus configuration](../../test/prometheus.yaml) of the OTel Collector, +although the configuration is fairly minimal. + +Because metrics are scraped from the Prometheus endpoint, they are independent of +`OTEL_SDK_DISABLED`. This means queue metrics will always be collected whenever +something is scraping the endpoint. + +Note, queue depth is per-queue only, and cannot be broken down by project. diff --git a/orthanc/orthanc-anon/plugin/pixl.py b/orthanc/orthanc-anon/plugin/pixl.py index 8d1c5b04..9050fa56 100644 --- a/orthanc/orthanc-anon/plugin/pixl.py +++ b/orthanc/orthanc-anon/plugin/pixl.py @@ -35,8 +35,12 @@ import pydicom import requests from core.exceptions import PixlDiscardError, PixlSkipInstanceError +from core.metrics import ( + record_instance_deidentification_failure, + record_study_deidentification_failure, +) from core.project_config.pixl_config_model import load_project_config -from core.telemetry import configure_logging, configure_tracing +from core.telemetry import configure_logging, configure_metrics, configure_tracing from decouple import config from loguru import logger from opentelemetry import trace @@ -52,6 +56,7 @@ write_dataset_to_bytes, ) from pydicom import dcmread +from sqlalchemy.exc import DBAPIError import orthanc @@ -84,6 +89,8 @@ RequestsInstrumentor().instrument() tracer = trace.get_tracer("pixl.orthanc_anon") +configure_metrics() + logger.warning("Running logging at level {}", logging_level) # Set up a thread pool executor for non-blocking calls to Orthanc @@ -347,9 +354,33 @@ def _anonymise_study_and_upload( logger.warning( "Failed to anonymize project: '{}', {}: {}", project_name, study_info, discard ) + record_study_deidentification_failure( + project_name=project_name, + failure_type="PixlDiscardError", + message="All instances have been skipped", + ) + return None + except DBAPIError as e: + logger.exception( + "Failed to anonymize project: '{}', {}: {}", project_name, study_info, e + ) + # Keep only the first line of the error message as otherwise the message contains + # the entire SQL query that failed. This would make the message have too high + # cardinality for the metric to be useful, and would make it hard to query for + # specific failure messages. + record_study_deidentification_failure( + project_name=project_name, + failure_type=type(e.orig).__name__, + message=str(e.orig).splitlines()[0], + ) return None - except Exception: # noqa: BLE001 + except Exception as e: # noqa: BLE001 logger.exception("Failed to anonymize project: '{}', {}", project_name, study_info) + record_study_deidentification_failure( + project_name=project_name, + failure_type=type(e).__name__, + message=str(e).splitlines()[0], + ) return None with logger.contextualize(pseudo_study_uid=anonymised_study_uid): @@ -412,6 +443,12 @@ def _anonymise_study_instances( ) key = "DICOM instance discarded as series not requested" skipped_instance_counts[key] += 1 + record_instance_deidentification_failure( + project_name=project_name, + study_uid=study_info.study_uid, + failure_type="PixlSkipSeriesError", + message=key, + ) continue if dataset.SeriesInstanceUID in series_to_skip: @@ -422,6 +459,12 @@ def _anonymise_study_instances( ) key = "DICOM instance discarded as series has too few instances" skipped_instance_counts[key] += 1 + record_instance_deidentification_failure( + project_name=project_name, + study_uid=study_info.study_uid, + failure_type="PixlSkipSeriesError", + message=key, + ) continue try: @@ -436,6 +479,12 @@ def _anonymise_study_instances( e, ) skipped_instance_counts[str(e)] += 1 + record_instance_deidentification_failure( + project_name=project_name, + study_uid=study_info.study_uid, + failure_type="PixlSkipInstanceError", + message=str(e), + ) else: anonymised_instances_bytes.append(anonymised_instance) anonymised_study_uid = dataset[0x0020, 0x000D].value diff --git a/pixl_core/src/core/metrics.py b/pixl_core/src/core/metrics.py new file mode 100644 index 00000000..e610e14b --- /dev/null +++ b/pixl_core/src/core/metrics.py @@ -0,0 +1,142 @@ +# Copyright (c) University College London Hospitals NHS Foundation Trust +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Define custom metrics for PIXL.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from opentelemetry import metrics + +__all__ = [ + "initialise_metrics", + "record_instance_deidentification_failure", + "record_study_deidentification_failure", + "record_study_exported", +] + + +@dataclass +class PixlMetrics: + """Custom metrics for PIXL.""" + + studies_exported: metrics.Counter | None = None + deidentification_failures: metrics.Counter | None = None + instance_deidentification_failures: metrics.Counter | None = None + + +pixl_metrics = PixlMetrics() + + +def initialise_metrics() -> None: + """ + Initialise custom metrics for PIXL. + + This must be done after the metrics provider has been set up, + otherwise the metrics will be no-ops. + """ + meter = metrics.get_meter(__name__) + + pixl_metrics.studies_exported = meter.create_counter( + name="pixl.studies.exported", + description="Number of studies exported, by project.", + unit="1", + ) + + description = ( + "Number of studies that failed to be de-identified, by project and failure reason." + ) + pixl_metrics.deidentification_failures = meter.create_counter( + name="pixl.studies.deidentification.failures", + unit="1", + description=description, + ) + + pixl_metrics.instance_deidentification_failures = meter.create_counter( + name="pixl.instances.deidentification.failures", + unit="1", + description=( + "Number of instances that failed to be de-identified, by project and failure reason." + ), + ) + + +def record_study_exported(project_name: str) -> None: + """ + Record a study exported metric. + + Args: + project_name (str): The name of the project for which the study was exported. + + """ + if pixl_metrics.studies_exported is None: + return + + pixl_metrics.studies_exported.add( + amount=1, + attributes={"project_name": project_name}, + ) + + +def record_study_deidentification_failure( + project_name: str, + failure_type: str, + message: str, +) -> None: + """ + Record a study de-identification failure metric. + + Args: + project_name: The name of the project for which the de-identification failure occurred. + failure_type: The type of the failure. + message (str): The message for the de-identification failure. + + """ + if pixl_metrics.deidentification_failures is None: + return + + pixl_metrics.deidentification_failures.add( + amount=1, + attributes={"project_name": project_name, "type": failure_type, "message": message}, + ) + + +def record_instance_deidentification_failure( + project_name: str, + study_uid: str, + failure_type: str, + message: str, +) -> None: + """ + Record an instance de-identification failure metric. + + Args: + project_name: The name of the project for which the de-identification failure occurred. + study_uid: The UID of the study for which the de-identification failure occurred. + failure_type: The type of the failure. + message (str): The message for the de-identification failure. + + """ + if pixl_metrics.instance_deidentification_failures is None: + return + + pixl_metrics.instance_deidentification_failures.add( + amount=1, + attributes={ + "project_name": project_name, + "study_uid": study_uid, + "type": failure_type, + "message": message, + }, + ) diff --git a/pixl_core/src/core/telemetry.py b/pixl_core/src/core/telemetry.py index 17157e91..ae3ef86c 100644 --- a/pixl_core/src/core/telemetry.py +++ b/pixl_core/src/core/telemetry.py @@ -21,16 +21,21 @@ from decouple import config from loguru import logger -from opentelemetry import trace +from opentelemetry import metrics, trace +from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from core.logging import OTelSink +from core.metrics import initialise_metrics __all__ = [ "configure_logging", + "configure_metrics", "configure_tracing", "telemetry_is_enabled", ] @@ -98,3 +103,28 @@ def configure_tracing() -> None: provider.add_span_processor(processor) trace.set_tracer_provider(provider) atexit.register(provider.shutdown) + + +def configure_metrics() -> None: + """ + Set up an OTLP metric exporter when OTEL_SDK_DISABLED is false + and OTEL_EXPORTER_OTLP_ENDPOINT is set in the environment. + """ + if not telemetry_is_enabled(): + return + + # If we have auto-instrumented the service, there's no way to tell the OTel SDK not to + # create the provider. So we have to reuse it here to avoid warnings in the logs. + # The provider created by the OTel SDK is equivalent to the one we create below. + existing_provider = metrics.get_meter_provider() + if isinstance(existing_provider, MeterProvider): + logger.debug("Existing MeterProvider detected (auto-instrumentation). Re-using it.") + initialise_metrics() + return + + exporter = OTLPMetricExporter() + reader = PeriodicExportingMetricReader(exporter) + provider = MeterProvider(resource=Resource.create(), metric_readers=[reader]) + metrics.set_meter_provider(provider) + atexit.register(provider.shutdown) + initialise_metrics() diff --git a/pixl_core/tests/test_metrics.py b/pixl_core/tests/test_metrics.py new file mode 100644 index 00000000..bb280a02 --- /dev/null +++ b/pixl_core/tests/test_metrics.py @@ -0,0 +1,93 @@ +# Copyright (c) University College London Hospitals NHS Foundation Trust +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests for PIXL custom OpenTelemetry metrics.""" + +from __future__ import annotations + +from unittest.mock import Mock + +import pytest + +from core.metrics import ( + pixl_metrics, + record_instance_deidentification_failure, + record_study_deidentification_failure, + record_study_exported, +) + + +@pytest.fixture +def mock_instruments(monkeypatch: pytest.MonkeyPatch) -> dict[str, Mock]: + """ + Replace each instrument on pixl_metrics with a Mock counter. + + Lets the record_* helpers be tested by asserting on the counter's `.add` + calls, with no OTel provider or exporter involved. + """ + mocks = { + "studies_exported": Mock(), + "deidentification_failures": Mock(), + "instance_deidentification_failures": Mock(), + } + for name, mock in mocks.items(): + monkeypatch.setattr(pixl_metrics, name, mock) + return mocks + + +def test_record_study_exported(mock_instruments: dict[str, Mock]) -> None: + """Test the exported-studies counter is incremented with the project attribute.""" + record_study_exported(project_name="test-project") + + mock_instruments["studies_exported"].add.assert_called_once_with( + amount=1, + attributes={"project_name": "test-project"}, + ) + + +def test_record_study_deidentification_failure(mock_instruments: dict[str, Mock]) -> None: + """Test the study de-id failure counter records project, type and message attributes.""" + record_study_deidentification_failure( + project_name="test-project", + failure_type="StringDataRightTruncation", + message="value too long for type character varying(255)", + ) + + mock_instruments["deidentification_failures"].add.assert_called_once_with( + amount=1, + attributes={ + "project_name": "test-project", + "type": "StringDataRightTruncation", + "message": "value too long for type character varying(255)", + }, + ) + + +def test_record_instance_deidentification_failure(mock_instruments: dict[str, Mock]) -> None: + """Test the instance de-id failure counter records the study_uid alongside the rest.""" + record_instance_deidentification_failure( + project_name="test-project", + study_uid="1.2.3", + failure_type="ValueError", + message="bad tag", + ) + + mock_instruments["instance_deidentification_failures"].add.assert_called_once_with( + amount=1, + attributes={ + "project_name": "test-project", + "study_uid": "1.2.3", + "type": "ValueError", + "message": "bad tag", + }, + ) diff --git a/pixl_export/src/pixl_export/main.py b/pixl_export/src/pixl_export/main.py index fc7b8554..26091321 100644 --- a/pixl_export/src/pixl_export/main.py +++ b/pixl_export/src/pixl_export/main.py @@ -23,8 +23,9 @@ from typing import Annotated from core.exports import ParquetExport +from core.metrics import record_study_exported from core.rest_api.router import router -from core.telemetry import configure_logging +from core.telemetry import configure_logging, configure_metrics from core.uploader import get_uploader from decouple import config # type: ignore [import-untyped] from fastapi import Body, FastAPI, HTTPException @@ -37,6 +38,8 @@ configure_logging(level=logging_level) logger.warning("Running logging at level {}", logging_level) +configure_metrics() + app = FastAPI( title="export-api", description="Export service", @@ -108,3 +111,4 @@ def export_dicom_from_orthanc( uploader = get_uploader(project_name) logger.debug("Sending {} via '{}'", study_id, type(uploader).__name__) uploader.upload_dicom_and_update_database(study_id) + record_study_exported(project_name) diff --git a/test/.env b/test/.env index 0210b332..aa38e552 100644 --- a/test/.env +++ b/test/.env @@ -8,6 +8,7 @@ PIXL_MAX_MESSAGES_IN_FLIGHT=5 TZ=Europe/London OTEL_SDK_DISABLED=false OTEL_EXPORTER_OTLP_ENDPOINT=http://lgtm:4317 +RABBITMQ_METRICS_PORT=15692 # PIXL PostgreSQL instance PIXL_DB_HOST=postgres diff --git a/test/docker-compose.yml b/test/docker-compose.yml index 38dbff42..26dd454d 100644 --- a/test/docker-compose.yml +++ b/test/docker-compose.yml @@ -27,6 +27,8 @@ services: ports: - "127.0.0.1:3000:3000" # Grafana UI - "127.0.0.1:4317:4317" # OTLP gRPC + volumes: + - ./prometheus.yaml:/otel-lgtm/prometheus.yaml:ro networks: pixl-net: diff --git a/test/prometheus.yaml b/test/prometheus.yaml new file mode 100644 index 00000000..db99219d --- /dev/null +++ b/test/prometheus.yaml @@ -0,0 +1,66 @@ +# Copyright (c) University College London Hospitals NHS Foundation Trust +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +--- +# Based on https://github.com/grafana/docker-otel-lgtm/blob/main/docker/prometheus.yaml +# Adds the PIXL RabbitMQ metrics endpoint to the scrape configs. +global: + scrape_native_histograms: true +otlp: + keep_identifying_resource_attributes: true + # Recommended attributes to be promoted to labels. + promote_resource_attributes: + - service.instance.id + - service.name + - service.namespace + - service.version + - cloud.availability_zone + - cloud.region + - container.name + - deployment.environment # backward compatibility + - deployment.environment.name + - k8s.cluster.name + - k8s.container.name + - k8s.cronjob.name + - k8s.daemonset.name + - k8s.deployment.name + - k8s.job.name + - k8s.namespace.name + - k8s.node.name + - k8s.pod.name + - k8s.replicaset.name + - k8s.statefulset.name + - host.name + - postgresql.database.name + - postgresql.schema.name + - postgresql.table.name + - postgresql.index.name + - database # used by otelcol/receiver/mongodb + - kafka.cluster.alias +storage: + tsdb: + # A 10min time window is enough because it can easily absorb retries and network delays. + out_of_order_time_window: 10m +scrape_configs: + - job_name: rabbitmq + metrics_path: /metrics/detailed + params: + family: + - queue_coarse_metrics + - queue_consumer_count + static_configs: + - targets: + - queue:15692 + labels: + service_name: queue + service_namespace: pixl