Telemetry

Overview

Install Dependencies

To use BOSA Telemetry, install telemetry dependencies.

pip install bosa-core-binary[telemetry]

BOSA Core provides a unified telemetry solution that use OpenTelemetry for observability and Sentry for error tracking. This integration enables you to monitor, debug, and optimize your applications. This integration helps you observe, debug, and optimize your applications with minimal configuration.

Integration Guide

This guide will help you integrate BOSA SDK Telemetry into your application in step by step. We'll start by configuring what telemetry options you'd like to enable and proceed to initializing it in your app.

Choose Your Telemetry Setup

Firstly, decide how you want to monitor your app:

Using Sentry Only

Use this setup for the simplest option, where you only need error monitoring without full trace instrumentation.

from bosa_core.telemetry.initializer import TelemetryConfig
from bosa_core.telemetry.sentry.initializer import SentryConfig

config = TelemetryConfig(
    sentry_config=SentryConfig(
        dsn="your-sentry-dsn",
        environment="your-env",
        release="1.0.0"
        # open_telemetry_config
    )
)

You can safely ignore the open_telemetry_config field inside SentryConfig. The Sentry SDK will be initialized independently without any additional tracing setup.

Using OTel with External Exporter

Choose this setup if your team already uses observability platforms like Jaeger, Zipkin, or Grafana Tempo to manage and explore distributed traces.

from bosa_core.telemetry.initializer import TelemetryConfig
from bosa_core.telemetry.opentelemetry.initializer import OpenTelemetryConfig
from bosa_core.telemetry.initializer import init_telemetry

config = TelemetryConfig(
    otel_config=OpenTelemetryConfig(
        endpoint="localhost:4317",
        attributes={"service.name": "your-service-name"}
    )
)

Using OTel + Sentry

Use this integration plan when you want to capture both distributed tracing and rich error diagnostics in one view. OpenTelemetry handles span-level trace propagation, while Sentry acts as the monitoring platform—collecting performance data and exceptions.

from bosa_core.telemetry.initializer import TelemetryConfig
from bosa_core.telemetry.opentelemetry.initializer import OpenTelemetryConfig
from bosa_core.telemetry.sentry.initializer import SentryConfig
from bosa_core.telemetry.initializer import init_telemetry

otel_cfg = OpenTelemetryConfig(
    endpoint="localhost:4317",
    attributes={"service.name": "your-service-name"}
)

sentry_cfg = SentryConfig(
    dsn="your-sentry-dsn",
    environment="your-env",
    release="2.0.0",
    open_telemetry_config=otel_cfg
)

config = TelemetryConfig(
    sentry_config=sentry_cfg
)

Initialize Telemetry

Once you've choose and complete the configuration needed, initialize telemetry:

from bosa_core.telemetry.initializer import init_telemetry

## Initialize your config
init_telemetry(config)

## Your fastAPI app

@app.get("/ping")
async def ping():
    return {"message": "pong"}

How to Use

For complete installation instructions and setup details, please refer to our Example. Currently BOSA Core Telemetry supports this following implementation patterns:

1. Telemetry with OTel and Sentry

2. Telemetry with Sentry Only

3. Telemetry with OTel and External Exporter

Sentry Integration

Sentry is an error monitoring and performance tracking service that helps identify and fix issues in applications. In BOSA Core, Sentry can be used in two ways:

1. Standalone: Captures uncaught exceptions and performance metrics

2. With OpenTelemetry: Sentry will acts as an OTLP trace receiver for distributed tracing

Under the hood, BOSA is using sentry SDK, you can use the same functionality from those SDK. for the documentation you can visit here

For this integration we can use sentry cloud service or using our GDP Labs sentry service in here (you need to contract our DSO team for access it)

OpenTelemetry Integration

OpenTelemetry, also known as OTel is an open-source observability framework that provides a standardized way to collect, process, and export telemetry data (traces, metrics, and logs) from your applications. BOSA Core telemetry uses OpenTelemetry's auto-instrumentation for FastAPI and LangChain operations.

The telemetry data can be exported in two ways:

1. Through an external OpenTelemetry backend (like Jaeger, Zipkin, or any OTLP-compatible system) by configuring the endpoint and port

2. Through Sentry integration, where traces are automatically forwarded to Sentry's performance monitoring system, providing additional error context and performance insights

Currently, BOSA Core Telemetry provides instrumentation support specifically for FastAPI and LangChain. Other web frameworks (like Flask or Django), background workers (e.g., Celery tasks, cron jobs) , and CLI scripts are not yet supported out of the box.

This is the example if you want to add more instrumentation in your code, for example how to add instrumentation with HTTPX

1. install the dependencies open telemetry instrumentation https

pip install opentelemetry-instrumentation-httpx

1. Initiliaze the instrumentation

import httpx
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

HTTPXClientInstrumentor().instrument()

This is the list of instrumentation supported by open telemetry SDK: open telemetry instrumentation list

Custom Functions instrumentation

BOSA also provide custom functions instrumentation. this instrumentation can provide your functions to be instrumented. we currently support:

• Class functions (CustomClass.function)

• Module functions (module.function)

• Static methods (@staticmethod)

• Class methods (@classmethod)

This is the example of how to use Custom functions instrumentation, you have a functions.py with path module that you want to instrument. this is the code:

def sync_function(...):
            ...

async def async_function(...):
...

and then you have this custom_class.py with path module.classes.custom_class that you want to instrument, this is the code:

class CustomClass:

            def method(self, ...):
                ...

            @classmethod
            def class_method(cls, ...):
                ...

            @staticmethod
            def static_method(...):
                ...

            async def async_method(self, ...):
                ...

            @classmethod
            async def async_class_method(cls, ...):
                ...

            @staticmethod
            async def async_static_method(...):
                ...

This is the example how you can instrument that class & function:

from module import functions
from module.classes.custom_class import CustomClass
from bosa_core.telemetry.opentelemetry.instrument.functions import BOSAFunctionsInstrumentor

# Instrument all functions in the module
BOSAFunctionsInstrumentor().instrument(
    methods=[
        functions.sync_function, functions.async_function,
        CustomClass.method, CustomClass.class_method,
        CustomClass.static_method, CustomClass.async_method,
        CustomClass.async_class_method, CustomClass.async_static_method
])

obj = CustomClass()
# Call the methods
obj.method(...)
CustomClass.class_method(...)
CustomClass.static_method(...)

await obj.async_method(...)
await CustomClass.async_class_method(...)
await CustomClass.async_static_method(...)

functions.sync_function(...)
await functions.async_function(...)


# Uninstrument all functions in the module
BOSAFunctionsInstrumentor().uninstrument()

From that example, the methods params in function instrument will be instrumented, the additional data that will be instrumented is only for input or output from that functions. this is the example of additional data:

input.args[0]", "val1"
input.kwargs.arg2", "val2"
"output", "sync-val1-val2"

But, we have limitation for using this, the limitation is if you want to instrument a function you need to include the module for that function call. For example you can see in the example above, we have 2 function in functions.py file, sync_function and asyns_function . If we want to intrumented that function you need to call the function in your current implementation to become like this

functions.sync_function(...)
await functions.async_function(...)

Trace Sampler

The Trace Sampler is a global setting that applies to all OpenTelemetry exporters. It must be configured during the first call of init_telemetry(...) and cannot be changed by calling init_telemetry(...) again.

BOSA Telemetry supports custom trace samplers for OpenTelemetry, allowing you to define your own sampling logic. To create one, you must implement the Sampler abstract class provided by the OpenTelemetry SDK. Below is the example of custom Sampler:

from typing import Sequence, Optional

from opentelemetry.context import Context
from opentelemetry.sdk.trace.sampling import (
    ParentBased,
    Sampler,
    SamplingResult,
    TraceIdRatioBased,
)
from opentelemetry.trace import Link, SpanKind
from opentelemetry.trace.span import TraceState
from opentelemetry.util.types import Attributes

NO_SAMPLING_ENDPOINTS: set[str] = {"/health", "not_important"}
IMPORTANT_ENDPOINTS: set[str] = {"/important"}

DEFAULT_SAMPLING_RATE: float = 0.1
IMPORTANT_SAMPLING_RATE: float = 1.0
NO_SAMPLING_RATE: float = 0.0

class OtelSampler(Sampler):

    def should_sample(
        self,
        parent_context: Optional[Context],
        trace_id: int,
        name: str,
        kind: SpanKind = None,
        attributes: Attributes = None,
        links: Sequence[Link] = None,
        trace_state: TraceState = None,
    ) -> SamplingResult:
        trace_sample_rate = DEFAULT_SAMPLING_RATE

        routes = attributes.get("http.route") if attributes else None
        if routes in NO_SAMPLING_ENDPOINTS:
            trace_sample_rate = NO_SAMPLING_RATE
        elif routes in IMPORTANT_ENDPOINTS:
            trace_sample_rate = IMPORTANT_SAMPLING_RATE

        sampler = ParentBased(TraceIdRatioBased(trace_sample_rate))
        return sampler.should_sample(
            parent_context, trace_id, name, kind, attributes, links, trace_state
        )

    def get_description(self) -> str:
        return "OtelSampler"

To use the custom sampler, provide an instance of it to OpenTelemetryConfig

from bosa_core.telemetry.initializer import TelemetryConfig
from bosa_core.telemetry.opentelemetry.initializer import OpenTelemetryConfig
from bosa_core.telemetry.initializer import init_telemetry

config = TelemetryConfig(
    otel_config=OpenTelemetryConfig(
        trace_sampler=OtelSampler(),
        ...
    )
)
init_telemetry(config)

API Reference

TelemetryConfig

TelemetryConfig is a wrapper configuration object that holds your telemetry setup. Pass this to init_telemetry().

Parameters:

• sentry_config: SentryConfig | None = None — optional. Enables Sentry error tracking and performance monitoring.

• otel_config: OpenTelemetryConfig | None = None — optional. Enables OpenTelemetry tracing to external exporters or Sentry.

Use either, both, or none (no-op).

OpenTelemetryConfig

Parameters:

• endpoint: str — the hostname (include port) of your OTLP-compatible trace exporter (e.g., Jaeger, Tempo).

• trace_sampler: Sampler — the OpenTelemetry Sampler to sampling traces.

• headers: dict[str, str] — the header sent to connect to OpenTelemetry exporter.

• attributes: dict[str, str] — a dictionary of resource-level tags (e.g., "service.name": "bosa-api").

• use_grpc: bool — if True, wil connect using GRPC to exporter, otherwise will use HTTP.

• fastapi_config: FastAPIConfig | None — optional. Enables automatic FastAPI instrumentation.

• use_langchain: bool = True — if enabled, automatically instruments LangChain-based pipelines.

• use_httpx: bool = True — if enabled, automatically instruments HTTPX library.

• use_requests: bool = True — if enabled, automatically instruments python Requests library.

SentryConfig

Parameters:

• dsn: str — your Sentry project DSN.

• environment: str — deployment environment (e.g., "production").

• release: str — application version tag.

• profiles_sample_rate: float | None — optional, Sentry performance sample rate.

• send_default_pii: bool | None — whether to send PII (e.g., usernames, emails) to Sentry.

• traces_sampler — a callable to control dynamic sampling.

• open_telemetry_config: OpenTelemetryConfig | None — if provided, links Sentry to OpenTelemetry spans.

• **kwargs — other parameters supported by sentry_sdk.init().

init_telemetry(config: TelemetryConfig)

Initializes the appropriate telemetry layers according to the provided config.

• If sentry_config is present, initializes Sentry (either use sentry only or with telemetry).

• If otel_config is present, sets the global tracer (Only supply it while using external exporter).