Created
October 17, 2025 22:28
-
-
Save jasonnerothin/6c86b57a45ac949f731f56d39e1d17ed to your computer and use it in GitHub Desktop.
threadsafe oltp tracing module
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import os | |
| import threading | |
| from typing import Optional | |
| from opentelemetry import trace | |
| from opentelemetry.sdk.trace import TracerProvider | |
| from opentelemetry.sdk.trace.export import BatchSpanProcessor | |
| from opentelemetry.sdk.resources import Resource | |
| from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter | |
| # Configuration | |
| OTEL_COLLECTOR_URL = os.environ.get( | |
| "OTEL_COLLECTOR_BASE_URL", | |
| "http://localhost:4318" | |
| ) | |
| # Thread-safe globals | |
| _tracer_provider: Optional[TracerProvider] = None | |
| _tracer_lock = threading.Lock() | |
| _initialized = False | |
| def initialize_traces(service_name: str, exporter_endpoint: Optional[str] = None): | |
| """ | |
| Initialize OpenTelemetry tracing with OTLP HTTP exporter. | |
| Thread-safe initialization - only initializes once. | |
| Args: | |
| service_name: Name of the service for telemetry | |
| exporter_endpoint: Optional override for collector endpoint | |
| """ | |
| global _tracer_provider, _initialized | |
| with _tracer_lock: | |
| if _initialized: | |
| return | |
| # Define service metadata | |
| resource = Resource.create({ | |
| "service.name": service_name, | |
| }) | |
| # Create tracer provider | |
| _tracer_provider = TracerProvider(resource=resource) | |
| # Configure OTLP exporter | |
| endpoint = exporter_endpoint or OTEL_COLLECTOR_URL | |
| span_exporter = OTLPSpanExporter( | |
| endpoint=f"{endpoint}/v1/traces", | |
| timeout=5, | |
| ) | |
| # Add batch processor | |
| span_processor = BatchSpanProcessor(span_exporter) | |
| _tracer_provider.add_span_processor(span_processor) | |
| # Set as global tracer provider | |
| trace.set_tracer_provider(_tracer_provider) | |
| _initialized = True | |
| print(f"OpenTelemetry Traces initialized for {service_name}. Endpoint: {endpoint}/v1/traces") | |
| def get_tracer(name: str = __name__): | |
| """ | |
| Get a tracer instance. Must call initialize_traces() first. | |
| Args: | |
| name: Name of the tracer (usually __name__ of the module) | |
| Returns: | |
| Tracer instance | |
| """ | |
| if not _initialized: | |
| raise RuntimeError("Traces not initialized. Call initialize_traces() first.") | |
| return trace.get_tracer(name) | |
| def create_span(tracer, span_name: str, attributes: Optional[dict] = None): | |
| """ | |
| Context manager for creating a span with optional attributes. | |
| Args: | |
| tracer: Tracer instance from get_tracer() | |
| span_name: Name of the span | |
| attributes: Optional dictionary of span attributes | |
| Returns: | |
| Span context manager | |
| Example: | |
| tracer = get_tracer(__name__) | |
| with create_span(tracer, "process_data", {"record_count": 100}): | |
| # Your code here | |
| pass | |
| """ | |
| span = tracer.start_span(span_name) | |
| if attributes: | |
| for key, value in attributes.items(): | |
| span.set_attribute(key, value) | |
| return span | |
| def shutdown_traces(): | |
| """ | |
| Gracefully shutdown the tracer provider and flush all pending spans. | |
| Should be called before application exit. | |
| """ | |
| global _tracer_provider, _initialized | |
| with _tracer_lock: | |
| if _tracer_provider and _initialized: | |
| _tracer_provider.shutdown() | |
| _initialized = False | |
| print("OpenTelemetry Traces shutdown complete.") | |
| # Example usage functions | |
| def record_event(span, event_name: str, attributes: Optional[dict] = None): | |
| """ | |
| Add an event to the current span. | |
| Args: | |
| span: Active span object | |
| event_name: Name of the event | |
| attributes: Optional event attributes | |
| """ | |
| if attributes: | |
| span.add_event(event_name, attributes) | |
| else: | |
| span.add_event(event_name) | |
| def set_span_attributes(span, attributes: dict): | |
| """ | |
| Set multiple attributes on a span. | |
| Args: | |
| span: Active span object | |
| attributes: Dictionary of attributes to set | |
| """ | |
| for key, value in attributes.items(): | |
| span.set_attribute(key, value) | |
| def record_exception(span, exception: Exception): | |
| """ | |
| Record an exception in the current span. | |
| Args: | |
| span: Active span object | |
| exception: Exception to record | |
| """ | |
| span.record_exception(exception) | |
| span.set_status(trace.Status(trace.StatusCode.ERROR, str(exception))) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment