|
4 | 4 | import re |
5 | 5 | import uuid |
6 | 6 | from datetime import timedelta |
7 | | -from typing import cast |
| 7 | +from typing import Any, cast |
8 | 8 | from urllib.request import urlopen |
9 | 9 |
|
10 | 10 | import pytest |
|
14 | 14 | from temporalio.runtime import ( |
15 | 15 | LogForwardingConfig, |
16 | 16 | LoggingConfig, |
| 17 | + OpenTelemetryConfig, |
17 | 18 | PrometheusConfig, |
18 | 19 | Runtime, |
19 | 20 | TelemetryConfig, |
@@ -269,6 +270,107 @@ async def check_metrics() -> None: |
269 | 270 | await assert_eventually(check_metrics) |
270 | 271 |
|
271 | 272 |
|
| 273 | +async def test_opentelemetry_histogram_bucket_overrides(client: Client): |
| 274 | + # Set up an OpenTelemetry configuration with custom histogram bucket overrides |
| 275 | + import threading |
| 276 | + from http.server import BaseHTTPRequestHandler, HTTPServer |
| 277 | + |
| 278 | + from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import ( |
| 279 | + ExportMetricsServiceRequest, |
| 280 | + ExportMetricsServiceResponse, |
| 281 | + ) |
| 282 | + |
| 283 | + special_value = float(1234.5678) |
| 284 | + histogram_overrides = { |
| 285 | + "temporal_long_request_latency": [special_value / 2, special_value], |
| 286 | + "custom_histogram": [special_value / 2, special_value], |
| 287 | + } |
| 288 | + |
| 289 | + captured: dict[str, list[float]] = {} |
| 290 | + lock = threading.Lock() |
| 291 | + |
| 292 | + class Handler(BaseHTTPRequestHandler): |
| 293 | + def log_message(self, format: str, *args: Any): |
| 294 | + pass # silence default stderr logging |
| 295 | + |
| 296 | + def do_POST(self): |
| 297 | + length = int(self.headers.get("Content-Length", "0")) |
| 298 | + req = ExportMetricsServiceRequest() |
| 299 | + req.ParseFromString(self.rfile.read(length)) |
| 300 | + with lock: |
| 301 | + for rm in req.resource_metrics: |
| 302 | + for sm in rm.scope_metrics: |
| 303 | + for m in sm.metrics: |
| 304 | + if m.HasField("histogram"): |
| 305 | + for dp in m.histogram.data_points: |
| 306 | + captured[m.name] = list(dp.explicit_bounds) |
| 307 | + body = ExportMetricsServiceResponse().SerializeToString() |
| 308 | + self.send_response(200) |
| 309 | + self.send_header("Content-Type", "application/x-protobuf") |
| 310 | + self.send_header("Content-Length", str(len(body))) |
| 311 | + self.end_headers() |
| 312 | + self.wfile.write(body) |
| 313 | + |
| 314 | + otel_port = find_free_port() |
| 315 | + server = HTTPServer(("127.0.0.1", otel_port), Handler) |
| 316 | + thread = threading.Thread(target=server.serve_forever, daemon=True) |
| 317 | + thread.start() |
| 318 | + try: |
| 319 | + runtime = Runtime( |
| 320 | + telemetry=TelemetryConfig( |
| 321 | + metrics=OpenTelemetryConfig( |
| 322 | + url=f"http://127.0.0.1:{otel_port}/v1/metrics", |
| 323 | + http=True, |
| 324 | + metric_periodicity=timedelta(milliseconds=100), |
| 325 | + durations_as_seconds=False, |
| 326 | + histogram_bucket_overrides=histogram_overrides, |
| 327 | + ), |
| 328 | + ), |
| 329 | + ) |
| 330 | + |
| 331 | + # Create and record to a custom histogram |
| 332 | + custom_histogram = runtime.metric_meter.create_histogram( |
| 333 | + "custom_histogram", "Custom histogram", "ms" |
| 334 | + ) |
| 335 | + custom_histogram.record(600) |
| 336 | + |
| 337 | + # Run a workflow so built-in histograms (e.g. temporal_long_request_latency) |
| 338 | + # are recorded and exported. |
| 339 | + client_with_overrides = await Client.connect( |
| 340 | + client.service_client.config.target_host, |
| 341 | + namespace=client.namespace, |
| 342 | + runtime=runtime, |
| 343 | + ) |
| 344 | + task_queue = f"task-queue-{uuid.uuid4()}" |
| 345 | + async with Worker( |
| 346 | + client_with_overrides, |
| 347 | + task_queue=task_queue, |
| 348 | + workflows=[HelloWorkflow], |
| 349 | + ): |
| 350 | + assert "Hello, World!" == await client_with_overrides.execute_workflow( |
| 351 | + HelloWorkflow.run, |
| 352 | + "World", |
| 353 | + id=f"workflow-{uuid.uuid4()}", |
| 354 | + task_queue=task_queue, |
| 355 | + ) |
| 356 | + |
| 357 | + async def check_metrics() -> None: |
| 358 | + with lock: |
| 359 | + snapshot = dict(captured) |
| 360 | + for key, buckets in histogram_overrides.items(): |
| 361 | + assert key in snapshot, ( |
| 362 | + f"Missing {key} in captured metrics: {list(snapshot)}" |
| 363 | + ) |
| 364 | + assert snapshot[key] == pytest.approx(buckets), ( |
| 365 | + f"Bucket mismatch for {key}: got {snapshot[key]} expected {buckets}" |
| 366 | + ) |
| 367 | + |
| 368 | + await assert_eventually(check_metrics) |
| 369 | + finally: |
| 370 | + server.shutdown() |
| 371 | + server.server_close() |
| 372 | + |
| 373 | + |
272 | 374 | def test_runtime_options_invalid_heartbeat() -> None: |
273 | 375 | with pytest.raises(ValueError): |
274 | 376 | Runtime( |
|
0 commit comments