Setup the telemetry system. Args: service_name: Name of the OpenTelemetry service project_id: Project ID to include in resource attributes exporter_endpoint: Endpoint for the span exporter metrics_endpoint: Endpoint for the metrics exporter max_queue
(
service_name: str = "agentops",
project_id: Optional[str] = None,
exporter_endpoint: str = "https://otlp.agentops.ai/v1/traces",
metrics_endpoint: str = "https://otlp.agentops.ai/v1/metrics",
max_queue_size: int = 512,
max_wait_time: int = 5000,
export_flush_interval: int = 1000,
jwt_provider: Optional[Callable[[], Optional[str]]] = None,
)
| 76 | |
| 77 | |
| 78 | def setup_telemetry( |
| 79 | service_name: str = "agentops", |
| 80 | project_id: Optional[str] = None, |
| 81 | exporter_endpoint: str = "https://otlp.agentops.ai/v1/traces", |
| 82 | metrics_endpoint: str = "https://otlp.agentops.ai/v1/metrics", |
| 83 | max_queue_size: int = 512, |
| 84 | max_wait_time: int = 5000, |
| 85 | export_flush_interval: int = 1000, |
| 86 | jwt_provider: Optional[Callable[[], Optional[str]]] = None, |
| 87 | ) -> tuple[TracerProvider, MeterProvider]: |
| 88 | """ |
| 89 | Setup the telemetry system. |
| 90 | |
| 91 | Args: |
| 92 | service_name: Name of the OpenTelemetry service |
| 93 | project_id: Project ID to include in resource attributes |
| 94 | exporter_endpoint: Endpoint for the span exporter |
| 95 | metrics_endpoint: Endpoint for the metrics exporter |
| 96 | max_queue_size: Maximum number of spans to queue before forcing a flush |
| 97 | max_wait_time: Maximum time in milliseconds to wait before flushing |
| 98 | export_flush_interval: Time interval in milliseconds between automatic exports of telemetry data |
| 99 | jwt_provider: Function that returns the current JWT token |
| 100 | |
| 101 | Returns: |
| 102 | Tuple of (TracerProvider, MeterProvider) |
| 103 | """ |
| 104 | # Build resource attributes |
| 105 | resource_attrs = get_global_resource_attributes( |
| 106 | service_name=service_name, |
| 107 | project_id=project_id, |
| 108 | ) |
| 109 | |
| 110 | resource = Resource(resource_attrs) |
| 111 | provider = TracerProvider(resource=resource) |
| 112 | |
| 113 | # Set as global provider |
| 114 | trace.set_tracer_provider(provider) |
| 115 | |
| 116 | # Create exporter with dynamic JWT support |
| 117 | exporter = AuthenticatedOTLPExporter(endpoint=exporter_endpoint, jwt_provider=jwt_provider) |
| 118 | |
| 119 | # Regular processor for normal spans and immediate export |
| 120 | processor = BatchSpanProcessor( |
| 121 | exporter, |
| 122 | max_export_batch_size=max_queue_size, |
| 123 | schedule_delay_millis=export_flush_interval, |
| 124 | ) |
| 125 | provider.add_span_processor(processor) |
| 126 | internal_processor = InternalSpanProcessor() # Catches spans for AgentOps on-terminal printing |
| 127 | provider.add_span_processor(internal_processor) |
| 128 | |
| 129 | # Setup metrics with JWT provider |
| 130 | def get_metrics_headers(): |
| 131 | token = jwt_provider() if jwt_provider else None |
| 132 | return {"Authorization": f"Bearer {token}"} if token else {} |
| 133 | |
| 134 | metric_exporter = OTLPMetricExporter(endpoint=metrics_endpoint, headers=get_metrics_headers()) |
| 135 |
no test coverage detected
searching dependent graphs…