Otel Common Code

CHANGELOG.md

@@ -1,5 +1,9 @@
 # CHANGELOG
 
+## 100.1.0
+
+* Add otel common code. Add common config base class.
+
 ## 100.0.0
 
 * `notification_type` is now a required argument of `clients.redis.daily_limit_cache_key` (apps have already been updated)

notifications_utils/clients/otel/otel_client.py

@@ -0,0 +1,204 @@
+import os
+
+from flask import Flask
+from opentelemetry import metrics
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+from opentelemetry.processor.baggage import BaggageSpanProcessor
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import (
+    ConsoleMetricExporter,
+    PeriodicExportingMetricReader,
+)
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
+from opentelemetry.trace import get_tracer_provider, set_tracer_provider
+
+
+def init_otel_app(app: Flask) -> None:
+    """
+    Initialize the OpenTelemetry instrumentation for the Flask app.
+    """
+
+    if getattr(app, "_otel_instrumented", False):
+        app.logger.debug("OpenTelemetry instrumentation already applied, skipping.")
+        return
+
+    export_mode = app.config.get("OTEL_EXPORT_TYPE", "none").lower().strip()
+    metric_readers = []
+
+    if export_mode == "console":
+        app.logger.info("OpenTelemetry metrics and spans will be exported to console")
+        metric_readers.append(PeriodicExportingMetricReader(ConsoleMetricExporter()))
+        span_processor = BatchSpanProcessor(ConsoleSpanExporter())
+    elif export_mode == "otlp":
+        endpoint = app.config.get("OTEL_COLLECTOR_ENDPOINT", "localhost:4317")
+        app.logger.info("OpenTelemetry metrics and spans will be exported to OTLP collector at %s", endpoint)
+        otlp_exporter = OTLPMetricExporter(endpoint=endpoint, insecure=True)
+        metric_readers.append(PeriodicExportingMetricReader(otlp_exporter))
+
+        os.environ["OTEL_METRIC_EXPORT_INTERVAL"] = app.config.get("OTEL_METRIC_EXPORT_INTERVAL")
+        os.environ["OTEL_METRIC_EXPORT_TIMEOUT"] = app.config.get("OTEL_METRIC_EXPORT_TIMEOUT")
+
+        span_processor = BatchSpanProcessor(
+            OTLPSpanExporter(
+                endpoint=endpoint,
+                insecure=app.config.get("OTEL_COLLECTOR_INSECURE", True),
+            )
+        )
+    elif export_mode == "none":
+        app.logger.info("OpenTelemetry metrics and spans will not be exported")
+        return
+    else:
+        raise ValueError(f"Invalid OTEL_EXPORT_TYPE: {export_mode}. Expected 'console', 'otlp', or 'none'.")
+
+    # TODO: Look into replacing the resource name inside the otel collector. This config would need to
+    # look something like:
+    #
+    # traces:
+    #     receivers: [otlp]
+    #     processors: [resourcedetection, transform/spans, batch/traces]
+    #     exporters: [otlp/traces]
+    #
+    # resourcedetection:
+    #     detectors:
+    #     - env
+    #     - system
+    #     - ecs
+    # transform/spans:
+    #     trace_statements:
+    #     - context: span
+    #         statements:
+    #         - >
+    #           set(resource.attributes["service.name"], resource.attributes["aws.ecs.service.name"])
+    #           where resource.attributes["aws.ecs.service.name"] != nil
+
+    resource = Resource.create(
+        {"service.name": os.getenv("NOTIFY_APP_NAME") or app.config.get("NOTIFY_APP_NAME") or "notifications"}
+    )
+
+    provider = MeterProvider(metric_readers=metric_readers, resource=resource)
+    metrics.set_meter_provider(provider)
+
+    set_tracer_provider(TracerProvider(resource=resource))
+
+    def public_baggage_predicate(baggage_key: str) -> bool:
+        """
+        Filter to only include baggage keys starting with 'public-'.
+        This ensures only public baggage items are automatically added as span attributes.
+        """
+        return baggage_key.startswith("public-")
+
+    get_tracer_provider().add_span_processor(BaggageSpanProcessor(public_baggage_predicate))
+    get_tracer_provider().add_span_processor(span_processor)
+
+    _instrument_app(app)
+
+    app._otel_instrumented = True
+
+
+def _instrument_app(app: Flask) -> None:
+    """
+    Apply OpenTelemetry instrumentation based on available optional dependencies.
+    """
+
+    # Affects both requests and Flask instrumentation
+    os.environ["OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST"] = app.config.get(
+        "OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST"
+    )
+
+    os.environ["OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE"] = app.config.get(
+        "OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE"
+    )
+
+    os.environ["OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS"] = app.config.get(
+        "OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS"
+    )
+
+    instrument_map = {
+        "celery": (_instrument_celery, "opentelemetry.instrumentation.celery"),
+        "flask": (_instrument_flask, "opentelemetry.instrumentation.flask"),
+        "redis": (_instrument_redis, "opentelemetry.instrumentation.redis"),
+        "requests": (_instrument_requests, "opentelemetry.instrumentation.requests"),
+        "sqlalchemy": (_instrument_sqlalchemy, "opentelemetry.instrumentation.sqlalchemy"),
+        "botocore": (_instrument_botocore, "opentelemetry.instrumentation.botocore"),
+    }
+
+    for name, (func, module_name) in instrument_map.items():
+        try:
+            __import__(module_name)
+            func(app)
+            app.logger.info("Enabled OpenTelemetry instrumentation: %s", name)
+        except ImportError:
+            app.logger.debug("Optional instrumentation '%s' not installed, skipping.", name)
+
+
+def _instrument_celery(app: Flask) -> None:
+    from opentelemetry.instrumentation.celery import CeleryInstrumentor
+
+    CeleryInstrumentor().instrument()
+
+
+def _instrument_flask(app: Flask) -> None:
+    from opentelemetry.instrumentation.flask import FlaskInstrumentor
+
+    FlaskInstrumentor().instrument_app(
+        app,
+        excluded_urls=app.config.get("OTEL_PYTHON_FLASK_EXCLUDED_URLS"),
+    )
+
+
+def _instrument_redis(app: Flask) -> None:
+    from opentelemetry.instrumentation.redis import RedisInstrumentor
+
+    def redis_request_hook(span, conn, args, kwargs):
+        if span and args and len(args) > 1:
+            # For multi-key commands, keys are all args[1:]
+            # For single-key commands, key is args[1]
+            keys = []
+            for arg in args[1:]:
+                if isinstance(arg, str | bytes):
+                    if isinstance(arg, bytes):
+                        try:
+                            arg = arg.decode("utf-8")
+                        except Exception:
+                            arg = repr(arg)
+                    keys.append(arg)
+                else:
+                    # Stop at first non-key argument (e.g., value for SET)
+                    break
+            if keys:
+                span.set_attribute("db.redis.keys", ",".join(keys))
+
+    def redis_response_hook(span, *args, **kwargs):
+        if span:
+            span.update_name(f"redis/{span.name}")
+
+    RedisInstrumentor().instrument(
+        request_hook=redis_request_hook,
+        response_hook=redis_response_hook,
+    )
+
+
+def _instrument_requests(app: Flask) -> None:
+    from opentelemetry.instrumentation.requests import RequestsInstrumentor
+
+    # Work around for span names not being unique in Requests instrumentation
+    def requests_response_hook(span, *args, **kwargs):
+        if span:
+            span.update_name(f"requests/{span.name}")
+
+    RequestsInstrumentor().instrument(response_hook=requests_response_hook)
+
+
+def _instrument_sqlalchemy(app: Flask) -> None:
+    from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
+
+    SQLAlchemyInstrumentor().instrument(enable_commenter=True, commenter_options={})
+
+
+def _instrument_botocore(app: Flask) -> None:
+    from opentelemetry.instrumentation.botocore import BotocoreInstrumentor
+
+    BotocoreInstrumentor().instrument()

notifications_utils/clients/otel/utils.py

@@ -0,0 +1,116 @@
+import time
+from collections.abc import Callable
+from contextlib import AbstractContextManager, contextmanager
+from functools import wraps
+from typing import Any
+
+from opentelemetry.metrics import get_meter
+from opentelemetry.trace import Span, Status, StatusCode, Tracer
+
+default_histogram_bucket = [
+    0.005,
+    0.01,
+    0.025,
+    0.05,
+    0.075,
+    0.1,
+    0.25,
+    0.5,
+    0.75,
+    1.0,
+    2.5,
+    5.0,
+    7.5,
+    10.0,
+    float("inf"),
+]
+
+# Examples of how to use the otel_duration_histogram decorator:
+
+# Example 1: Static attributes (current behavior)
+# @otel_duration_histogram("my_function_duration", attributes={"operation": "send_email"})
+# def send_email(to, subject):
+#     ...
+
+# Example 2: Dynamic attributes based on function arguments
+# @otel_duration_histogram("process_user_duration", attributes=lambda args, kwargs: {"user_id": kwargs.get("user_id")})
+# def process_user(user_id):
+#     ...
+
+# Example 3: Dynamic attributes using both args and kwargs
+# @otel_duration_histogram("do_something_duration", attributes=lambda args, kwargs: {
+#     "first_arg": args[0] if args else None,
+#     "keyword": kwargs.get("keyword")
+# })
+# def do_something(a, keyword=None):
+#     ...
+
+# If you are considering using this decorator, think about if a span would be more appropriate for your use case.
+# If we are using spanmetrics inside the otel collector you will automatically get a histogram for the span duration.
+
+
+def otel_duration_histogram(
+    name: str,
+    *,
+    unit: str = "seconds",
+    description: str = "",
+    attributes: dict[str, Any] | Callable[[tuple, dict], dict[str, Any]] | None = None,
+):
+    """
+    Decorator to record function execution time in an OpenTelemetry histogram.
+
+    Args:
+        name (str): Name of the histogram metric.
+        unit (str): Unit of measurement (default: "seconds").
+        description (str): Description of the metric.
+        attributes (dict or callable, optional): Static or dynamic attributes.
+
+    Returns:
+        function: Wrapped function with histogram instrumentation.
+    """
+    meter = get_meter(__name__)
+    histogram = meter.create_histogram(name, unit=unit, description=description)
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            start = time.perf_counter()
+            status = "success"
+            try:
+                return func(*args, **kwargs)
+            except Exception:
+                status = "error"
+                raise
+            finally:
+                elapsed = time.perf_counter() - start
+                base_attrs = attributes(args, kwargs) if callable(attributes) else (attributes or {})
+                record_attrs = {**base_attrs, "status": status}
+                histogram.record(elapsed, attributes=record_attrs)
+
+        return wrapper
+
+    return decorator
+
+
+@contextmanager
+def otel_span_with_status(tracer: Tracer, name: str, **attributes: Any) -> AbstractContextManager[Span]:
+    """
+    Context manager to create an OpenTelemetry span with status handling.
+
+    Args:
+        tracer: The tracer instance.
+        name (str): Name of the span.
+        **attributes: Attributes to set on the span.
+
+    Yields:
+        span: The created span.
+    """
+    with tracer.start_as_current_span(name) as span:
+        for key, value in attributes.items():
+            span.set_attribute(key, value)
+        try:
+            yield span
+        except Exception as exc:
+            span.record_exception(exc)
+            span.set_status(Status(StatusCode.ERROR, str(exc)))
+            raise

notifications_utils/config.py

@@ -0,0 +1,24 @@
+from os import getenv
+
+
+# Common configuration for applications
+class BaseConfig:
+    OTEL_EXPORT_TYPE = getenv("OTEL_EXPORT_TYPE", "none")
+    OTEL_COLLECTOR_ENDPOINT = getenv("OTEL_COLLECTOR_ENDPOINT", "localhost:4317")
+    OTEL_COLLECTOR_INSECURE = getenv("OTEL_COLLECTOR_INSECURE", "true").lower() in ("true", "1", "yes")
+
+    OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST = getenv(
+        "OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST", ".*"
+    )
+    OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE = getenv(
+        "OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE", ".*"
+    )
+
+    OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS = getenv(
+        "OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS", "authorization,.*cookie.*,.*session.*"
+    )
+
+    OTEL_PYTHON_FLASK_EXCLUDED_URLS = getenv("OTEL_PYTHON_FLASK_EXCLUDED_URLS", "_status(\\?.*)?$,/metrics")
+
+    OTEL_METRIC_EXPORT_INTERVAL = getenv("OTEL_METRIC_EXPORT_INTERVAL", "15s")
+    OTEL_METRIC_EXPORT_TIMEOUT = getenv("OTEL_METRIC_EXPORT_TIMEOUT", "30s")

notifications_utils/version.py

@@ -5,4 +5,4 @@
 # - `make version-minor` for new features
 # - `make version-patch` for bug fixes
 
-__version__ = "100.0.0"  # 6e2957149e9f88e6a4a21d64d0dd7f43
+__version__ = "100.1.0"  # 21ce083334b259ba846e018404fca793

pyproject.toml

@@ -31,8 +31,27 @@ dependencies = [
     "segno>=1.6.1",
     "smartypants>=2.0.1",
     "statsd>=4.0.1",
+    "opentelemetry-distro==0.54b1",
+    "opentelemetry-exporter-otlp==1.33.1",
+    "opentelemetry-processor-baggage==0.54b1",
 ]
 
+[project.optional-dependencies]
+otel-instrumentation = [
+    "opentelemetry-instrumentation-flask==0.54b1",
+    "opentelemetry-instrumentation-celery==0.54b1",
+    "opentelemetry-instrumentation-requests==0.54b1",
+    "opentelemetry-instrumentation-redis==0.54b1",
+    "opentelemetry-instrumentation-sqlalchemy==0.54b1",
+    "opentelemetry-instrumentation-botocore==0.54b1"
+]
+otel-flask = [ "opentelemetry-instrumentation-flask==0.54b1" ]
+otel-celery = [ "opentelemetry-instrumentation-celery==0.54b1" ]
+otel-requests = [ "opentelemetry-instrumentation-requests==0.54b1" ]
+otel-redis = [ "opentelemetry-instrumentation-redis==0.54b1" ]
+otel-sqlalchemy = [ "opentelemetry-instrumentation-sqlalchemy==0.54b1" ]
+otel-botocore = [ "opentelemetry-instrumentation-botocore==0.54b1" ]
+
 [project.urls]
 Homepage = "https://github.com/alphagov/notifications-utils"