Open telemetry

.github/workflows/pytest.yml

@@ -53,6 +53,7 @@ jobs:
           cp config.EXAMPLE/exporter.env.EXAMPLE ../config/exporter.env
           cp config.EXAMPLE/hasher.env.EXAMPLE ../config/hasher.env
           cp config.EXAMPLE/controller.env.EXAMPLE ../config/controller.env
+          cp config.EXAMPLE/monitoring.env.EXAMPLE ../config/monitoring.env
           {
             echo ""
             echo "AZURE_CLIENT_ID=${AZURE_CLIENT_ID}"

.pre-commit-config.yaml

@@ -36,7 +36,7 @@ repos:
             "types-pika",
             "types-requests",
           ]
-        files: src/
+        files: ^(src|monitoring)/
   # a collection of sanity checks: check for merge conflicts, check the end of
   # lines, check for DOS vs unix stuff, and a newline at the end of files.
   - repo: https://github.com/pre-commit/pre-commit-hooks

config.EXAMPLE/controller.env.EXAMPLE

@@ -13,3 +13,6 @@ RABBITMQ_PASSWORD="my_pw"
 RABBITMQ_HOST="localhost"
 RABBITMQ_PORT=5672
 RABBITMQ_QUEUE="waveform"
+# OpenTelemetry OTLP/HTTP endpoint of the LGTM collector.
+OTEL_EXPORTER_OTLP_ENDPOINT="http://lgtm:4318"
+OTEL_SERVICE_NAME=waveform-controller

config.EXAMPLE/exporter.env.EXAMPLE

@@ -32,3 +32,7 @@ ONLY_USE_CSV_FROM_YESTERDAY=TRUE
 # specify a date to process format YYYY-MM-DD also accepts a regular
 # expression to match multiple date
 PROCESS_CSV_FROM_DATE=
+
+# OpenTelemetry OTLP/HTTP endpoint of the LGTM collector.
+OTEL_EXPORTER_OTLP_ENDPOINT="http://lgtm:4318"
+OTEL_SERVICE_NAME=waveform-exporter

config.EXAMPLE/monitoring.env.EXAMPLE

@@ -0,0 +1,7 @@
+# This is an EXAMPLE file, do not put real secrets in here.
+# Copy it to ../config/monitoring.env and then DELETE THIS COMMENT.
+# When does the monitoring job run
+MONITORING_CRON_SCHEDULE="*/5 * * * *"
+# OpenTelemetry OTLP/HTTP endpoint of the LGTM collector.
+OTEL_EXPORTER_OTLP_ENDPOINT="http://lgtm:4318"
+OTEL_SERVICE_NAME=waveform-monitoring

docker-compose.lgtm.yml

@@ -0,0 +1,37 @@
+# Local Grafana LGTM stack (Loki, Grafana, Tempo, Mimir/Prometheus, plus
+# an OpenTelemetry Collector).
+# In production, this would be an external service.
+#
+# Docs: https://grafana.com/docs/opentelemetry/docker-lgtm/
+#
+# Start with the waveform stack (same Compose project/network, hostname `lgtm`):
+#   docker compose -f docker-compose.yml -f docker-compose.lgtm.yml up -d
+#
+# Or start only LGTM:
+#   docker compose -p lgtm -f docker-compose.lgtm.yml up -d
+#
+# Grafana UI: http://127.0.0.1:3000
+
+services:
+  lgtm:
+    image: grafana/otel-lgtm:0.30.2
+    restart: unless-stopped
+    profiles:
+      - lgtm
+    ports:
+      # Grafana UI
+      - "127.0.0.1:3000:3000"
+      # OTLP gRPC (OpenTelemetry default)
+      - "127.0.0.1:4317:4317"
+      # OTLP HTTP (OpenTelemetry default)
+      - "127.0.0.1:4318:4318"
+      # Optional: query backends directly (Grafana already has them as datasources)
+      - "127.0.0.1:3200:3200" # Tempo
+      - "127.0.0.1:4040:4040" # Pyroscope
+      - "127.0.0.1:9090:9090" # Prometheus / Mimir
+    volumes:
+      # https://github.com/grafana/docker-otel-lgtm#persist-data-across-container-instantiation
+      - lgtm-data:/data
+
+volumes:
+  lgtm-data:

docker-compose.yml

@@ -55,3 +55,24 @@ services:
     env_file:
       - ../config/hasher.env
     restart: unless-stopped
+  waveform-monitoring:
+    # This service exists as a partly temporary measure, covering a few different cases:
+    #   * Until LGTM on the GAE can tell us disk free amount for /gae
+    #   * Stats on HL7 bz2 files on disk. Arguably done better in Emap but
+    #     I'm putting all the telemetry in one project for now.
+    #   * Disk usage stats for the waveform processing (CSVs, parquets, etc). This will likely stay here.
+    build:
+      context: .
+      dockerfile: monitoring/Dockerfile
+      args:
+        HTTP_PROXY: ${HTTP_PROXY}
+        http_proxy: ${http_proxy}
+        HTTPS_PROXY: ${HTTPS_PROXY}
+        https_proxy: ${https_proxy}
+    env_file:
+      - ../config/monitoring.env
+    volumes:
+      # we are assuming this FS layout to emap saved messages
+      - ../../waveform-saved-messages:/waveform-saved-messages:ro
+      - ../waveform-export:/waveform-export:ro
+    restart: unless-stopped

monitoring/Dockerfile

@@ -0,0 +1,40 @@
+FROM python:3.13-slim-bookworm@sha256:8092ae2ef67061f9db412458dbdce44dbf16748fb3cae5cdbd020f467a9712d0
+LABEL authors="Stephen Thompson, Jeremy Stein"
+
+RUN export DEBIAN_FRONTEND=noninteractive && \
+    apt-get update && \
+    apt-get install --yes --no-install-recommends ca-certificates curl && \
+    apt-get autoremove --yes && apt-get clean --yes && rm -rf /var/lib/apt/lists/*
+
+# Not in apt and no official docker image, so install supercronic from: https://github.com/aptible/supercronic/releases
+ARG SUPERCRONIC_VERSION=v0.2.49
+ARG TARGETARCH
+# don't use SHA1 as per instructions as it's insecure
+RUN case "${TARGETARCH}" in \
+    amd64) SUPERCRONIC_SHA256SUM=a53ae236602c7338aba3fbaff40bda6300eae3b9fedb8261eb06cfe3724430c1 ;; \
+    arm64) SUPERCRONIC_SHA256SUM=02aa0cb229ba09050cba6638059dadb9eedc2276632ea43d6a57a2f8c1629dd5 ;; \
+    *) echo "Unsupported TARGETARCH: ${TARGETARCH}" >&2; exit 1 ;; \
+    esac \
+    && SUPERCRONIC=supercronic-linux-${TARGETARCH} \
+    && SUPERCRONIC_URL=https://github.com/aptible/supercronic/releases/download/${SUPERCRONIC_VERSION}/${SUPERCRONIC} \
+    && curl -fsSLO "$SUPERCRONIC_URL" \
+    && echo "${SUPERCRONIC_SHA256SUM}  ${SUPERCRONIC}" | sha256sum --check - \
+    && chmod +x "$SUPERCRONIC" \
+    && mv "$SUPERCRONIC" "/usr/local/bin/${SUPERCRONIC}" \
+    && ln -s "/usr/local/bin/${SUPERCRONIC}" /usr/local/bin/supercronic
+
+# uv image label "0.12.5"
+COPY --from=ghcr.io/astral-sh/uv@sha256:e85be844203885286c60ffad8a858d48afb6c5a5c237ca0e67f12e74b8f174b1 /uv /uvx /bin/
+
+ARG UVCACHE=/root/.cache/uv
+WORKDIR /app
+
+COPY monitoring/monitor.py monitoring/monitor.py.lock /app/
+RUN uv lock --check --script monitor.py
+
+RUN --mount=type=cache,target=${UVCACHE} uv export --script monitor.py --locked | uv pip install --system -r -
+
+COPY monitoring/entrypoint.sh /app/entrypoint.sh
+RUN chmod +x /app/entrypoint.sh
+
+ENTRYPOINT ["/app/entrypoint.sh"]

monitoring/entrypoint.sh

@@ -0,0 +1,14 @@
+#!/bin/bash
+
+set -eo pipefail
+
+if [ -z "$MONITORING_CRON_SCHEDULE" ]; then
+  echo "You must set MONITORING_CRON_SCHEDULE when running this container"
+  exit 1
+fi
+
+cat > /etc/crontab <<EOF
+$MONITORING_CRON_SCHEDULE python /app/monitor.py
+EOF
+
+exec supercronic /etc/crontab

monitoring/monitor.py

@@ -0,0 +1,197 @@
+#!/usr/bin/env python3
+"""Scan saved HL7 messages and emit OpenTelemetry metrics.
+
+Run in this command in dev to update the lockfile: `uv lock --script monitoring/monitor.py`
+"""
+
+import logging
+import os
+import sys
+import time
+from pathlib import Path
+from time import perf_counter
+
+from opentelemetry.metrics import Meter
+from opentelemetry import metrics
+from opentelemetry.exporter.otlp.proto.http.metric_exporter import (
+    OTLPMetricExporter,
+)
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+
+# /// script
+# requires-python = ">=3.13"
+# dependencies = [
+#   "opentelemetry-exporter-otlp-proto-http==1.42.0",
+# ]
+# ///
+
+INSTRUMENTATION_SCOPE = "waveform-monitoring.meter"
+SAVED_MESSAGES_DIR = Path("/waveform-saved-messages")
+WAVEFORM_EXPORT_DIR = Path("/waveform-export")
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(message)s",
+    stream=sys.stdout,
+)
+logger = logging.getLogger(__name__)
+
+
+def _env(name: str, default: str | None = None) -> str:
+    value = os.environ.get(name)
+    if value is None or value == "":
+        if default is not None:
+            return default
+        else:
+            raise RuntimeError(f"Environment variable {name} not set")
+    return value
+
+
+def _scan_directory_ages(path: Path) -> tuple[int, float | None, float | None]:
+    """Return file count, newest age (seconds), oldest age (seconds)."""
+    now = time.time()
+    mtimes: list[float] = []
+
+    for entry in path.rglob("*.hl7archive.bz2"):
+        if entry.is_file():
+            mtimes.append(entry.stat().st_mtime)
+
+    if not mtimes:
+        return 0, None, None
+
+    newest_mtime = max(mtimes)
+    oldest_mtime = min(mtimes)
+    return len(mtimes), now - newest_mtime, now - oldest_mtime
+
+
+def _setup_metrics(service_name: str, otlp_endpoint: str | None) -> None:
+    if not otlp_endpoint:
+        logger.error(
+            "OTEL_EXPORTER_OTLP_ENDPOINT not set; metrics will not be exported"
+        )
+        return
+
+    metrics.set_meter_provider(
+        MeterProvider(
+            resource=Resource.create({SERVICE_NAME: service_name}),
+            metric_readers=[
+                PeriodicExportingMetricReader(
+                    OTLPMetricExporter(), export_interval_millis=15000
+                )
+            ],
+        )
+    )
+
+
+def scan_hl7_bz2(meter: Meter):
+    file_count = meter.create_up_down_counter(
+        "waveform.monitoring.hl7bz2_files.count",
+        unit="{file}",
+        description="Total HL7 files in saved-messages directory",
+    )
+    newest_age = meter.create_gauge(
+        "waveform.monitoring.hl7bz2_files.newest_age_seconds",
+        unit="s",
+        description="Age of the most recently modified HL7 file",
+    )
+    oldest_age = meter.create_gauge(
+        "waveform.monitoring.hl7bz2_files.oldest_age_seconds",
+        unit="s",
+        description="Age of the oldest HL7 file",
+    )
+    total_bytes = meter.create_gauge(
+        "waveform.monitoring.hl7bz2_files.total_bytes",
+        unit="By",
+        description="Total byte count of hl7 bz2 files",
+    )
+
+    scan_time_hist = meter.create_histogram(
+        "waveform.monitoring.hl7bz2_files.meta.disk_scan_time",
+        unit="s",
+        description="Duration of disk scan for metrics generation",
+    )
+
+    start_time = perf_counter()
+    byte_count = _bytes_in_regular_files(SAVED_MESSAGES_DIR)
+    total_bytes.set(byte_count)
+    count, newest_age_seconds, oldest_age_seconds = _scan_directory_ages(
+        SAVED_MESSAGES_DIR
+    )
+    time_taken = perf_counter() - start_time
+    scan_time_hist.record(time_taken)
+    file_count.add(count)
+    if newest_age_seconds is not None:
+        newest_age.set(newest_age_seconds)
+    if oldest_age_seconds is not None:
+        oldest_age.set(oldest_age_seconds)
+    logger.info(
+        "Scanned %s in %ss: count=%d newest_age=%s oldest_age=%s, bytes=%s",
+        SAVED_MESSAGES_DIR,
+        time_taken,
+        count,
+        f"{newest_age_seconds:.0f}s" if newest_age_seconds is not None else "n/a",
+        f"{oldest_age_seconds:.0f}s" if oldest_age_seconds is not None else "n/a",
+        byte_count,
+    )
+
+
+def scan_waveform_exporter_files(meter):
+    scan_time_hist = meter.create_histogram(
+        "waveform.monitoring.exporter.meta.disk_scan_time",
+        unit="s",
+        description="Duration of disk scan for metrics generation",
+    )
+    start_time = perf_counter()
+    # dirs that contain large files where we need to track disk usage
+    big_top_level_dirs = ["original-csv", "original-parquet", "pseudonymised"]
+    # dirs that won't get too large but do have other info we'll want to track
+    for tld_name in big_top_level_dirs:
+        tld = WAVEFORM_EXPORT_DIR / tld_name
+        tld_meter_name = tld_name.replace("-", "_")
+        gauge = meter.create_gauge(
+            f"waveform.monitoring.{tld_meter_name}_bytes",
+            unit="By",
+            description=f"Bytes in the {tld_name} directory",
+        )
+        byte_count = _bytes_in_regular_files(tld)
+        gauge.set(byte_count)
+    time_taken = perf_counter() - start_time
+    scan_time_hist.record(time_taken)
+    logger.info("Scanned %s in %ss", WAVEFORM_EXPORT_DIR, time_taken)
+
+
+def _bytes_in_regular_files(tld: Path):
+    """Get sum of bytes in all regular files under the given directory."""
+    byte_count = 0
+    for dn, _, files in tld.walk():
+        for f in files:
+            f_path = dn / f
+            if f_path.is_file():
+                byte_count += f_path.stat().st_size
+    return byte_count
+
+
+def main() -> int:
+    service_name = _env("OTEL_SERVICE_NAME")
+    otlp_endpoint = _env("OTEL_EXPORTER_OTLP_ENDPOINT")
+
+    # setup
+    _setup_metrics(service_name, otlp_endpoint)
+    meter = metrics.get_meter(INSTRUMENTATION_SCOPE)
+
+    # things to measure
+    scan_hl7_bz2(meter)
+    scan_waveform_exporter_files(meter)
+
+    # shutdown, flush data
+    provider = metrics.get_meter_provider()
+    if isinstance(provider, MeterProvider):
+        provider.force_flush(timeout_millis=15000)
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())