Skip to main content

handler

Datadog metrics handler for sending custom metrics to Datadog's API.

This module provides a handler that integrates with Python's logging framework to send custom metrics to Datadog's Metrics API (/api/v2/series), with per-metric buffering and size management. It mirrors the logs handler but routes to a separate endpoint.

Module​

Functions​

emit_metric​

def emit_metric(    metric_name: str,    value: float,    *,    timestamp: int | None = None,    tags: list[str] | None = None,    metric_type: MetricIntakeType | None = None,) ‑> None:

Emit a custom metric point to Datadog via metrics_logger.

Preferred public API for emitting metrics. Builds the log record contract expected by DatadogMetricsHandler (metric name as the message; point timestamp/value and options carried on extra) so call sites do not hand-build it. The timestamp/value are passed via extra rather than as logging args so the record's message never depends on %-formatting.

Arguments

  • metric_name: The metric name. Accepts a MetricName member or a raw string.
  • value: The metric value.
  • timestamp: Unix timestamp (seconds) for the point. Defaults to now.
  • tags: Per-call tags applied to this metric's series.
  • metric_type: The Datadog intake type for this metric. If omitted, the handler defaults the series to GAUGE.

flush_datadog_metrics​

def flush_datadog_metrics() ‑> None:

Flush the Datadog metrics buffer.

Should be called after each task completes, mirroring flush_datadog_telemetry.

record_execution_time​

def record_execution_time(start: float, *, tags: list[str] | None = None) ‑> None:

Emit elapsed time since start as the EXECUTION_TIME_SECONDS gauge.

Shared emission core for execution-time telemetry. Used by track_execution_time (which supplies function:/class: tags) and by call sites that need a dynamic tag the decorator cannot express — e.g. a per-DAG-step step:<name>. Only active when config.settings.enable_execution_time_metrics is True, and never raises: telemetry must not break execution.

Arguments

  • start: time.perf_counter() captured before the timed work began.
  • tags: Tags applied to the emitted gauge series.

setup_datadog_metrics​

def setup_datadog_metrics(    dd_client_token: str | None = None,    dd_site: str | None = None,    hostname: str | None = None,    tags: list[str] | None = None,) ‑> None:

Setup Datadog metrics if credentials are available.

Mirrors setup_datadog_telemetry but routes to the Datadog Metrics API (/api/v2/series) rather than the Logs API. Must be called alongside setup_datadog_telemetry at pod startup.

This function is idempotent - calling it multiple times is safe.

Arguments

  • dd_client_token: The Datadog API key.
  • dd_site: The Datadog site (e.g. 'datadoghq.com', 'datadoghq.eu').
  • hostname: The host name attached to every metric series. Defaults to the system hostname.
  • tags: Static tags applied to every metric series (e.g. ['env:prod']).

Returns None

shutdown_datadog_metrics​

def shutdown_datadog_metrics() ‑> None:

Shutdown Datadog metrics and flush any pending series.

Should be called at pod shutdown, mirroring shutdown_datadog_telemetry.

track_execution_time​

def track_execution_time(func: F) ‑> F:

Decorator that emits function execution time as a Datadog gauge metric.

Supports both sync and async callables. Only active when config.settings.enable_execution_time_metrics is True. Emits the MetricName.EXECUTION_TIME_SECONDS gauge with the wrapped callable's class and function name attached as class:<X> / function:<Y> tags.

Arguments

  • func: The function to wrap.

Returns The wrapped function.

Classes​

DatadogMetricsHandler​

class DatadogMetricsHandler(    api_instance: MetricsApi, hostname: str, tags: list[str] | None = None,):

A logging handler that sends custom metrics to Datadog's Metrics API.

Buffers points per unique (metric_name, tags) combination, so each combination becomes its own MetricSeries. Each buffer is monitored independently and flushed when it approaches 90% of the 5 MB uncompressed limit, and again on flush()/close().

Call sites do not use this handler directly; they go through emit_metric, which forwards to metrics_logger:

metrics_logger.info(metric_name, extra={...})

where metric_name maps to record.msg and the point's timestamp/value and options are carried as record attributes via extra:

  • metric_timestamp / metric_value: the point. Carried on extra (not as logging args) so the message never depends on %-formatting.
  • tags: per-call tags appended to this series' tags.
  • metric_type: the Datadog intake type for the series. It is set per call site; calls that omit it default the series to GAUGE.

See: https://docs.datadoghq.com/api/latest/metrics/

Initialise the handler.

Arguments

  • api_instance: A configured Datadog MetricsApi instance.
  • hostname: Attached to every MetricSeries via resources so metrics are filterable by host in Datadog.
  • tags: Static tags applied to every series (e.g. ['env:prod']).

Ancestors​

Variables​

  • static MAX_SERIES_SIZE

Methods​

close​

def close(self) ‑> None:

Flush and close the handler.

emit​

def emit(self, record: logging.LogRecord) ‑> None:

Buffer a metric point.

Expects record.msg to be the metric name, with the point's timestamp and value carried as the metric_timestamp / metric_value attributes (set via extra=...). Per-call tags can be supplied via extra={"tags": [...]}. Records that do not match this shape are silently dropped. The timestamp/value are deliberately not passed as logging args so the record's message never depends on %-formatting and cannot break a foreign formatter.

Each unique (metric_name, tags) combination is buffered independently and flushed as its own MetricSeries when it approaches the size limit.

Arguments

  • record: The log record.

flush​

def flush(self) ‑> None:

Flush all series buffers.