DTA Observability

A lightweight wrapper around OpenTelemetry for Python applications.

Overview

DTA Observability simplifies the use of OpenTelemetry by providing a streamlined interface for instrumentation. It handles configuration of tracing, metrics, and logging with minimal setup.

Features

• Single function initialization of all telemetry components
• Automatic instrumentation for Flask, FastAPI, Celery, and other frameworks
• Structured logging with trace context correlation
• Function decoration for easy span creation
• System and application metrics collection
• Support for OTLP, GCP Cloud, and console exporters
• Optional trace-only none exporter for log correlation without trace export
• Automatic resource detection
• Configuration via parameters or environment variables

Installation

pip install dta-observability

Or with Poetry:

poetry add dta-observability

Basic Usage

import dta_observability
from dta_observability import get_logger, traced

# Initialize telemetry
dta_observability.init_telemetry(
    service_name="my-service",
    service_version="1.0.0",
    otlp_endpoint="http://otel-collector:4317",
    exporter_type="otlp",  # Options: "otlp", "console", "gcp"
)

# Get a logger
logger = get_logger("my-service")

# Use the traced decorator
@traced(name="my_function")
def my_function():
    logger.info("Doing work")
    return "result"

Framework Integration

Flask

from flask import Flask
import dta_observability

app = Flask(__name__)

dta_observability.init_telemetry(
    service_name="flask-service",
    flask_app=app
)

Flask Audit Logging

Flask audit logging works with any Flask application, not just DTA-based ones, through two optional resolver callbacks passed to init_telemetry(). auth_resolver maps the current request to an (identity, auth_type) pair, and context_resolver returns a dictionary of extra fields to attach to the audit record. If neither resolver is configured, audit logging still emits complete records using the WSGI identity defaults (REMOTE_USER/AUTH_TYPE) and request.remote_addr for the client IP.

from flask import Flask, g

import dta_observability

app = Flask(__name__)


def resolve_auth(request):
    user = getattr(g, "user", None)
    if user is None:
        return None, None
    return str(user.id), "session"


def resolve_context(request):
    return {
        "tenant": getattr(g, "tenant_id", None),
    }


dta_observability.init_telemetry(
    service_name="flask-service",
    flask_app=app,
    auth_resolver=resolve_auth,
    context_resolver=resolve_context,
)

Audit Log Format

Every audit record — Flask or FastAPI — follows the same schema (audit_schema: 2), emitted as structured extra fields on the audit logger:

{
  "audit_schema": 2,
  "logger": "audit",
  "path": "/api/orders/42",
  "method": "GET",
  "user_id": "user-123",
  "client_ip": "203.0.113.7",
  "client_ip_source": "socket",
  "status_code": 200,
  "outcome": "success",
  "duration_ms": 14,
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "user_agent": "Mozilla/5.0",
  "auth_type": "session",
  "host": "api.example.com",
  "referer": "https://example.com/orders",
  "session_id": null,
  "context": {
    "tenant": "acme"
  }
}

outcome is derived from status_code: success (< 400, INFO), denied (401/403, WARNING), client_error (other 4xx, WARNING), error (5xx, ERROR), unknown (no status — e.g. a propagated exception, ERROR). trace_id, session_id, and context are null when unavailable; they are never omitted from the record.

FastAPI

from fastapi import FastAPI
import dta_observability

app = FastAPI()

dta_observability.init_telemetry(
    service_name="fastapi-service",
    fastapi_app=app
)

Celery

from celery import Celery
import dta_observability

app = Celery("tasks")

dta_observability.init_telemetry(
    service_name="worker-service",
    celery_app=app
)

GCP Integration

When using the GCP exporter type:

1. For traces: Uses Cloud Trace exporter
2. For metrics: Uses Cloud Monitoring exporter with workload.googleapis.com prefix
3. For logs: Uses GCP log format when log_format is set to "gcp", sending logs to stdout in the proper format

To use GCP integration:

dta_observability.init_telemetry(
    service_name="my-gcp-service",
    exporter_type="gcp",  # Uses GCP exporters for all signal types
    log_format="gcp"      # Formats logs for GCP
)

When log_format is set to "gcp", all logs will be formatted for Google Cloud Logging and sent to stdout, while metrics and traces will use their respective GCP exporters.

If you want GCP-formatted logs with trace/span correlation fields but do not want to export traces, set traces_exporter_type="none" and keep enable_traces=True:

dta_observability.init_telemetry(
    service_name="my-gcp-service",
    exporter_type="gcp",
    log_format="gcp",
    traces_exporter_type="none",
    enable_traces=True,
    enable_logs=True,
    enable_metrics=False,
)

This keeps spans active in-process for log correlation, but does not export them to Cloud Trace or to the console.

OTLP Integration

OTLP exporters send telemetry to an OpenTelemetry Collector:

dta_observability.init_telemetry(
    service_name="my-otlp-service",
    exporter_type="otlp",
    otlp_endpoint="http://otel-collector:4317"
)

Configuration

Configuration options available in init_telemetry():

Environment variables can also be prefixed with OTEL_ or DTA_ (e.g., DTA_SERVICE_NAME).

Examples

The examples directory contains sample applications demonstrating usage with different frameworks.