Skip to main content

telemetry

Datadog telemetry for Bitfount.

Split into two pipelines:

  • bitfount.telemetry.logs — structured log events sent to the Datadog Logs API via telemetry_logger.
  • bitfount.telemetry.metrics — custom metrics sent to the Datadog Metrics API via metrics_logger / emit_metric.

The full public surface is re-exported here for convenience.

Module​

Submodules​

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.

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.

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

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_metrics​

def shutdown_datadog_metrics() ‑> None:

Shutdown Datadog metrics and flush any pending series.

Should be called at pod shutdown, mirroring shutdown_datadog_telemetry.

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.

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​

AlgorithmProgressEvent​

class AlgorithmProgressEvent(**data: Any):

Emitted at algorithm lifecycle boundaries (run/epoch start and end).

Fired automatically by AlgorithmProgress for every algorithm via the BaseAlgorithmHook infrastructure — individual algorithm classes do not need any changes.

step is one of "run_start", "run_end", "epoch_start", or "epoch_end".

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static algorithm_name : str
  • static current_epoch : int | None
  • static datetime : str
  • static max_epochs : int | None
  • static model_config
  • static step : str
  • static step_info : str | None
  • static task_context : str

BatchProgressEvent​

class BatchProgressEvent(**data: Any):

Emitted by the orchestrator at the start of each protocol batch.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static batch_number : int
  • static model_config
  • static protocol_name : str

DatadogLogsHandler​

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

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.

Ancestors​

Variables​

  • static BUFFER_PERCENTAGE
  • static MAX_BUFFER_SIZE

Methods​

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.

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.

DatasetConnectedEvent​

class DatasetConnectedEvent(**data: Any):

Emitted once per datasource when a pod successfully connects it.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static connection_datetime : str
  • static dataset_name : str
  • static datasource_type : str
  • static model_config

EHRQueryCompleteEvent​

class EHRQueryCompleteEvent(**data: Any):

Emitted after an EHR patient query batch completes.

Fires once per protocol batch in EHR-enabled algorithms. Provides visibility into the per-batch EHR query phase, which is otherwise the longest silent stretch during task execution.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static config_type : str
  • static model_config
  • static patients_queried : int
  • static records_found : int

EHRScreeningBatchEvent​

class EHRScreeningBatchEvent(**data: Any):

Emitted after each EHR screening page is queried and eligibility-matched.

Fires at the end of trial eligibility matching, after all eligible patients have been identified and before the CSV is output.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static model_config
  • static page_eligible : int
  • static page_number : int
  • static page_size : int
  • static project_id : str | None
  • static task_id : str
  • static total_eligible : int

EHRSessionInitialisedEvent​

class EHRSessionInitialisedEvent(**data: Any):

Emitted after a NextGen or FHIR R4 EHR session is set up.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static config_type : str
  • static model_config

EHRTokenRefreshedEvent​

class EHRTokenRefreshedEvent(**data: Any):

Emitted after a FHIR client access token is refreshed.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static config_type : str
  • static model_config

FileMultiSeriesReducedEvent​

class FileMultiSeriesReducedEvent(**data: Any):

Emitted when multiple series match filters and are reduced to one.

Provides context on the reduction so that future capabilities can be built to handle multi-series output (e.g. returning multiple rows per file).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static datasource_type : str
  • static file_name : str
  • static laterality : str
  • static model_config
  • static rows_after : int
  • static rows_before : int
  • static series_protocol : str

FileMultiSeriesSkippedEvent​

class FileMultiSeriesSkippedEvent(**data: Any):

Emitted when a file is skipped due to unresolvable multi-series ambiguity.

Supplements the generic task_skip_file telemetry with structured context that will inform future multi-series handling capabilities.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static datasource_type : str
  • static file_name : str
  • static laterality_configured : str | None
  • static model_config
  • static series_count : int
  • static series_protocol_configured : str | None

FilteringCompleteEvent​

class FilteringCompleteEvent(**data: Any):

Emitted after RecordFilterAlgorithm.setup_run() finishes selecting files.

Fires once per task, before the first batch begins. Provides visibility into how many files were considered and how many passed the filter criteria.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static files_selected : int
  • static model_config
  • static total_files : int

GroupingSummaryEvent​

class GroupingSummaryEvent(**data: Any):

Aggregate Datadog event for grouping-aware batching behaviour.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static batch_size : int | None
  • static cached_files : int | None
  • static group_count : int | None
  • static include_non_new_group_files : bool | None
  • static maybe_final : bool | None
  • static model_config
  • static oversized_cohort_size : int | None
  • static selected_files : int | None
  • static step : str

MetricName​

class MetricName(*args, **kwds):

Canonical names for all Datadog custom metrics.

Inherits from str so values serialise directly as metric names. The intake type for each metric is set at the call site (see emit_metric), defaulting to GAUGE when omitted.

Ancestors​

Variables​

  • static EHR_REQUEST_COUNT
  • static EHR_RESPONSE_BODY_SIZE_BYTES
  • static EHR_RESPONSE_TIME_SECONDS
  • static EXECUTION_TIME_SECONDS

MissingColumnEvent​

class MissingColumnEvent(**data: Any):

Emitted when a required column is absent from a file (e.g. DICOM pixel data).

file_name is passed through PII redaction before logging because file names in medical imaging datasets may encode patient identifiers. column_name is a static code constant (e.g. "Pixel Data") and carries no PII.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static column_name : str
  • static file_name : str
  • static model_config

PatientEligibilityEvent​

class PatientEligibilityEvent(**data: Any):

Emitted by the orchestrator after eligible patient counts are calculated.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static datasource : str
  • static datetime : str
  • static eligible_count : int
  • static model_config
  • static project_id : str

SchemaGenerationEndEvent​

class SchemaGenerationEndEvent(**data: Any):

Emitted when the Prefect schema worker finishes generating a schema.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static dataset_id : str
  • static datetime : str
  • static model_config

SchemaGenerationStartEvent​

class SchemaGenerationStartEvent(**data: Any):

Emitted when the Prefect schema worker begins generating a schema.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static dataset_id : str
  • static datetime : str
  • static model_config

SchemaInitiationSource​

class SchemaInitiationSource(*args, **kwds):

Source that triggered schema generation.

Inherits from str so values serialise directly as JSON strings.

Ancestors​

Variables​

  • static PREFECT_SCHEMA_WORKER

SchemaUploadSuccessEvent​

class SchemaUploadSuccessEvent(**data: Any):

Emitted after a full schema is successfully uploaded to the Hub.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static dataset_id : str
  • static datetime : str
  • static model_config
  • static number_of_records : int

TaskAcceptedByPodEvent​

class TaskAcceptedByPodEvent(**data: Any):

Emitted by the modeller when a pod accepts a task request.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static app_version : str
  • static datetime : str
  • static model_config
  • static pod_identifier : str
  • static task_id : str

TaskAcceptedEvent​

class TaskAcceptedEvent(**data: Any):

Emitted by the worker when it accepts and begins a task.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static app_version : str
  • static datasource_type : str
  • static datetime : str
  • static model_config
  • static modeller_username : str
  • static pod_name : str
  • static task_id : str

TaskCompleteEvent​

class TaskCompleteEvent(**data: Any):

Emitted by the worker when a task finishes successfully.

Symmetric counterpart to TaskAcceptedEvent — together they bracket the full task execution window in Datadog logs.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static datetime : str
  • static model_config
  • static modeller_username : str
  • static pod_name : str
  • static task_id : str

TaskCompleteTimeoutEvent​

class TaskCompleteTimeoutEvent(**data: Any):

Emitted when the worker times out waiting for TASK_COMPLETE from the modeller.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static datetime : str
  • static model_config
  • static modeller_username : str
  • static pod_name : str
  • static task_id : str

TaskErrorEvent​

class TaskErrorEvent(**data: Any):

Emitted by the worker when an unhandled exception aborts a task.

stacktrace_frames contains only the frame strings from traceback.format_tb() — the exception message and .args are deliberately excluded to avoid leaking PII (patient names, IDs, etc. that may appear in exception messages).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static error_type : str
  • static model_config
  • static modeller_username : str
  • static pod_name : str
  • static stacktrace_frames : str
  • static task_id : str

TelemetryEventName​

class TelemetryEventName(*args, **kwds):

Canonical names for all Datadog telemetry events.

Inherits from str so values serialise directly as JSON strings.

Ancestors​

Variables​

  • static ALGORITHM_PROGRESS
  • static BATCH_PROGRESS
  • static DATASET_CONNECTED
  • static EHR_QUERY_COMPLETE
  • static EHR_SCREENING_BATCH
  • static EHR_SESSION_INITIALISED
  • static EHR_TOKEN_REFRESHED
  • static FILE_MULTI_SERIES_REDUCED
  • static FILE_MULTI_SERIES_SKIPPED
  • static FILTERING_COMPLETE
  • static GROUPING_SUMMARY
  • static MISSING_COLUMN
  • static PATIENT_ELIGIBILITY
  • static SCHEMA_GENERATION_END
  • static SCHEMA_GENERATION_START
  • static SCHEMA_UPLOAD_SUCCESS
  • static TASK_ACCEPTED
  • static TASK_ACCEPTED_BY_POD
  • static TASK_COMPLETE
  • static TASK_COMPLETE_TIMEOUT
  • static TASK_ERROR
  • static TRIAL_FILTER_SUMMARY
  • static WORKER_STARTUP_PHASE

TrialFilterSummaryEvent​

class TrialFilterSummaryEvent(**data: Any):

Aggregate Datadog event for trial filter outcomes within a batch.

rows_missing counts rows with missing inputs required by the filter. It is not mutually exclusive with rows_matched or rows_failed.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static filter_name : str
  • static matching_column_count : int | None
  • static missing_columns : list[str] | None
  • static model_config
  • static rows_failed : int
  • static rows_matched : int
  • static rows_missing : int
  • static step : str

WorkerStartupPhaseEvent​

class WorkerStartupPhaseEvent(**data: Any):

Emitted after each phase of worker startup, with how long that phase took.

Together these bracket the window between the worker reporting "Configuring task" and its first exchange with the modeller — protocol deserialisation and model download, protocol pickling, spawning the child interpreter, child setup, and algorithm initialisation. That window is otherwise silent, and a message arriving from the modeller during it is only tolerated for handler_register_grace_period seconds, so the split between these phases is what decides whether a task starts or stalls.

process is "parent" (the pod) or "child" (the spawned worker).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static duration_seconds : float
  • static model_config
  • static phase : str
  • static pod_name : str | None
  • static process : str
  • static task_id : str