Export metrics to OTel collector and define custom metrics

.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

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:

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.

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

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,
+        },
+    )

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()