Skip to main content

handler

Datadog logging handler for sending logs to Datadog's API.

This module provides a handler that integrates with Python's logging framework to send logs to Datadog, with automatic batching and size management.

For file-related skip events in Datadog: set config.settings.enable_skipped_file_telemetry True and provide dd_client_token and dd_site. Skip payloads are sanitized (UTF-8 safe) before sending.

Every record is enriched on the way out with the run context (bitfount.telemetry.context) and, when the record carries an exception, its stack frames (bitfount.telemetry.errors). Doing it here rather than in each event model means an event type never has to remember to carry identity, and telemetry_logger.error(event, exc_info=exc) is all a caller needs for a traceback to reach Datadog.

Module​

Functions​

flush_datadog_telemetry​

def flush_datadog_telemetry() ‑> None:

Flush the Datadog telemetry buffer.

This should be called to ensure all buffered logs are sent to Datadog.

setup_datadog_telemetry​

def setup_datadog_telemetry(    dd_client_token: str | None = None,    dd_site: str | None = None,    service: str = 'pod',    hostname: str | None = None,    tags: list[str] | None = None,    log_level: str = 'INFO',) ‑> None:

Setup Datadog telemetry logging if credentials are available.

If credentials are not provided, the telemetry logger will exist but have no handlers, meaning all telemetry logs will be silently dropped.

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

Arguments

  • dd_client_token: The Datadog client token.
  • dd_site: The Datadog site to use (e.g., 'datadoghq.com', 'datadoghq.eu').
  • service: The service to use for the logs.
  • hostname: The hostname to use for the logs. Defaults to system hostname.
  • tags: The tags to use for the logs.
  • log_level: The log level for the Datadog handler (e.g., 'INFO', 'DEBUG').

Returns None

shutdown_datadog_telemetry​

def shutdown_datadog_telemetry() ‑> None:

Shutdown Datadog telemetry logging and flush any pending logs.

This should be called during application shutdown to ensure all buffered logs are sent to Datadog.

Classes​

DatadogLogsHandler​

class DatadogLogsHandler(    api_instance: LogsApi,    source: str,    hostname: str,    service: str,    tags: list[str] | None = None,    capacity: int = 1000,    flush_interval_seconds: float = 0.0,):

A MemoryHandler that sends logs to Datadog.

This handler automatically flushes when the buffer approaches a 5MB limit. These numbers are taken from Datadog's Logs API documentation: https://docs.datadoghq.com/api/latest/logs/

We piggyback off Python's MemoryHandler class since we do not want to implement our own buffer management system, including what happens when there are records left in the buffer when the handler is closed.

Initialize the DatadogMemoryHandler.

Arguments

  • api_instance: The Datadog API instance.
  • source: The source of the logs.
  • hostname: The hostname of the logs.
  • service: The service of the logs.
  • tags: The tags of the logs.
  • capacity: The capacity of the buffer, defaulted to 1000 according to Datadog's documentation.
  • flush_interval_seconds: How often to flush the buffer regardless of how full it is. Defaults to no periodic flush; the production path is setup_datadog_telemetry, which passes the configured interval.

Ancestors​

Variables​

  • static BUFFER_PERCENTAGE
  • static MAX_BUFFER_SIZE

Methods​

close​

def close(self) ‑> None:

Stop the periodic flush, then flush and close the handler.

The flusher is signalled but not waited for: logging.shutdown calls this while holding the handler, and a flush already in flight cannot finish until that caller returns. shutdown_datadog_* stops the flusher properly first, so the orderly path loses nothing.

emit​

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

Emit a record to the buffer.

Note that we immediately build the HTTPLogItem object and add it to a separate buffer, rather than waiting for the flush operation to do so.

Arguments

  • record: The log record.

flush​

def flush(self) ‑> None:

Flush the buffer by sending all records to Datadog in a single request.

Override the parent's flush method to flush records instead of calling emit() per record.

shouldFlush​

def shouldFlush(self, record: logging.LogRecord, item_size: int = 0) ‑> bool:

Determine if we should flush the buffer.

Arguments

  • record: The log record.
  • item_size: The size of the item to add to the buffer.

Returns True if we should flush the buffer, False otherwise.