Connecting a Python app to Monium Traces over Unified Agent

pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp
pip install opentelemetry-instrumentation-flask  # For Flask auto-instrumentation
pip install python-dotenv  # For environment variables

# Basic OpenTelemetry settings
OTEL_SERVICE_NAME=my-service  # must be the same as x-monium-service in Unified Agent
OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317  # endpoint Unified Agent
OTEL_EXPORTER_OTLP_INSECURE=true

# Sampling settings (optional)
OTEL_TRACES_SAMPLER=traceidratio
OTEL_TRACES_SAMPLER_ARG=1.0

import os
from dotenv import load_dotenv
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource

# Loading environment variables from the `.env` file
load_dotenv()

def init_tracing():
    # Reading the configuration from the environment variables
    service_name = os.getenv("OTEL_SERVICE_NAME", "unknown-service")
    otlp_endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
    insecure = os.getenv("OTEL_EXPORTER_OTLP_INSECURE", "false").lower() == "true"
    
    # Setting the service metadata resource
    resource = Resource(attributes={
        "service.name": service_name
    })
    
    # Creating a trace provider
    provider = TracerProvider(resource=resource)
    
    # Configuring an exporter to send spans to Unified Agent
    otlp_exporter = OTLPSpanExporter(
        endpoint=otlp_endpoint,
        insecure=insecure
    )
    
    # Adding a batch span processor for efficient span sending
    provider.add_span_processor(BatchSpanProcessor(otlp_exporter))
    
    # Registering the provider globally
    trace.set_tracer_provider(provider)
    
    return provider

from flask import Flask
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from tracing import init_tracing

app = Flask(__name__)

# Initializing tracing
init_tracing()

# Flask auto-instrumentation
FlaskInstrumentor().instrument_app(app)

@app.route("/")
def hello():
    return "Hello World"

@app.route("/order/<int:order_id>")
def process_order(order_id):
    result = handle_order(order_id)
    return result

def handle_order(order_id):
    # Business logic for order processing
    return {"status": "processed", "order_id": order_id}

if __name__ == "__main__":
    app.run()

from flask import Flask
from opentelemetry import trace
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.trace import Status, StatusCode
from tracing import init_tracing

app = Flask(__name__)

# Initializing tracing
init_tracing()

# Flask auto-instrumentation
FlaskInstrumentor().instrument_app(app)

# Getting a tracer for the current module
tracer = trace.get_tracer(__name__)

@app.route("/")
def hello():
    return "Hello World"

@app.route("/order/<int:order_id>")
def process_order(order_id):
    result = handle_order(order_id)
    return result

def handle_order(order_id):
    # Creating a new span to track function execution
    # `start_as_current_span` automatically makes the span current in the context
    with tracer.start_as_current_span("handle_order") as span:
        # Adding attributes to the span for filtering and analysis
        span.set_attribute("order.id", order_id)
        span.set_attribute("order.type", "online")
        
        # Creating a nested span for a sub-operation
        # It automatically becomes a child of the current span
        with tracer.start_as_current_span("validate_payment"):
            payment_valid = validate_payment(order_id)
            # Attributes help clarify the outcome of your operation
            span.set_attribute("payment.valid", payment_valid)
            
            if not payment_valid:
                # Marking the span as failed
                span.set_status(Status(StatusCode.ERROR, "Payment validation failed"))
                return {"status": "error", "message": "Invalid payment"}
        
        with tracer.start_as_current_span("update_inventory"):
            # `add_event` records important events within the span
            span.add_event("Inventory update started")
            update_inventory(order_id)
            span.add_event("Inventory update completed")
        
        # Explicitly marking the span as successful
        span.set_status(Status(StatusCode.OK))
        return {"status": "processed", "order_id": order_id}

def validate_payment(order_id):
    # Simulating payment validation
    return True

def update_inventory(order_id):
    # Simulating inventory update
    pass

if __name__ == "__main__":
    app.run()

def fetch_user_data(user_id):
    with tracer.start_as_current_span("fetch_user_data") as span:
        span.set_attribute("user.id", user_id)
        
        try:
            # Calling an external API or database
            user_data = external_api_call(user_id)
            span.set_attribute("user.country", user_data.get("country"))
            span.set_status(Status(StatusCode.OK))
            return user_data
        except Exception as e:
            # `record_exception` automatically adds a stack trace
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            raise

# Ratio-based sampling
OTEL_TRACES_SAMPLER=traceidratio
OTEL_TRACES_SAMPLER_ARG=0.1  # 10% of traces

# For development, collect all traces
# OTEL_TRACES_SAMPLER=always_on

# To disable tracing
# OTEL_TRACES_SAMPLER=always_off

import os
from dotenv import load_dotenv
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased, ParentBased

load_dotenv()

def init_tracing():
    service_name = os.getenv("OTEL_SERVICE_NAME", "unknown-service")
    otlp_endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
    insecure = os.getenv("OTEL_EXPORTER_OTLP_INSECURE", "false").lower() == "true"
    
    # Reading the sampling parameter
    sampling_ratio = float(os.getenv("OTEL_TRACES_SAMPLER_ARG", "1.0"))
    
    resource = Resource(attributes={"service.name": service_name})
    
    # ParentBased preserves the parent span sampling decision
    # This is critical for correct distributed tracing
    sampler = ParentBased(root=TraceIdRatioBased(sampling_ratio))
    
    provider = TracerProvider(
        resource=resource,
        sampler=sampler
    )
    
    # Configuring an exporter to send spans to Unified Agent
    otlp_exporter = OTLPSpanExporter(
        endpoint=otlp_endpoint,
        insecure=insecure
    )
    
    provider.add_span_processor(BatchSpanProcessor(otlp_exporter))
    trace.set_tracer_provider(provider)
    
    return provider