25565075335

Committed 08 May 2026 03:47PM UTC coverage: 92.787% (-0.1%) from 92.887%

Build # 25565075335

Build Type

push

github

Committed by

web-flow

Commit Message

add OpenTelemetry backend for work unit reporting (#23284)

# Overview

Add a new `pants.backend.observability.opentelemetry` backend to report
work unit tracing to OpenTelemetry. The backend is based on
[shoalsoft-pants-opentelemetry-plugin](https://github.com/shoalsoft/shoalsoft-pants-opentelemetry-plugin)
with unnecessary compatibility code and "shoalsoft" branding removed.

Notes:
- This backend only reports Pants engine work units to OpenTelemetry; it
does not report tracing data for Pants rule code or Rust code.
- This backend does not support gRPC export due to fork safety issues
with the gRPC C library and Python. See
https://github.com/shoalsoft/shoalsoft-pants-opentelemetry-plugin/issues/84
and https://github.com/grpc/grpc/blob/master/doc/fork_support.md for
additional details.

# Lockfile

```
    Lockfile diff: 3rdparty/python/user_reqs.lock [python-default]

    ==                    Upgraded dependencies                     ==

      anyio                          4.12.1       -->   4.13.0
      certifi                        2026.1.4     -->   2026.4.22
      charset-normalizer             3.4.4        -->   3.4.7
      click                          8.3.1        -->   8.3.2
      cross-web                      0.4.1        -->   0.6.0
      cryptography                   46.0.5       -->   46.0.7
      graphql-core                   3.2.7        -->   3.2.8
      idna                           3.11         -->   3.12
      librt                          0.8.1        -->   0.9.0
      pydantic                       2.12.5       -->   2.13.3
      pydantic-core                  2.41.5       -->   2.46.3
      pygments                       2.19.2       -->   2.20.0
      pyjwt                          2.11.0       -->   2.12.1
      python-dotenv                  1.2.1        -->   1.2.2
      python-multipart               0.0.22       -->   0.0.26
      ujson                          5.11.0       -->   5.12.0

    ==                   ... (continued)

Coverage Stats

564 of 740 new or added lines in 12 files covered. (76.22%)

1 existing line in 1 file now uncovered.

92944 of 100169 relevant lines covered (92.79%)

4.02 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

50.0

/src/python/pants/backend/observability/opentelemetry/workunit_handler.py

# Copyright 2026 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import datetime
from typing import Any, Mapping

from pants.backend.observability.opentelemetry.processor import (
    IncompleteWorkunit,
    Level,
    Processor,
    ProcessorContext,
    Workunit,
)
from pants.engine.internals.native_engine import all_counter_names
from pants.engine.internals.scheduler import Workunit as RawWorkunit
from pants.engine.streaming_workunit_handler import StreamingWorkunitContext, WorkunitsCallback
from pants.util.frozendict import FrozenDict


class _TelemetryContext(ProcessorContext):
    def __init__(self, pants_context: StreamingWorkunitContext) -> None:
        self._pants_context = pants_context

    def get_metrics(self) -> Mapping[str, int]:
        metric_names = all_counter_names()
        metrics = self._pants_context.get_metrics()
        for metric_name in metric_names:
            if metric_name not in metrics:
                metrics[metric_name] = 0
        return metrics


class TelemetryWorkunitsCallback(WorkunitsCallback):
    def __init__(
        self,
        processor: Processor,
        *,
        finish_timeout: datetime.timedelta,
        async_completion: bool,
    ) -> None:
        self.processor: Processor = processor
        self.finish_timeout = finish_timeout
        self.async_completion = async_completion

    @property
    def can_finish_async(self) -> bool:
        return self.async_completion

    def _convert_time(self, seconds: int, nanoseconds: int) -> datetime.datetime:
        t = datetime.datetime(year=1970, month=1, day=1, tzinfo=datetime.UTC)
        t = t + datetime.timedelta(seconds=seconds, microseconds=nanoseconds // 1000)
        return t

    def _convert_incomplete_workunit(self, raw_workunit: RawWorkunit) -> IncompleteWorkunit:
        return IncompleteWorkunit(
            name=raw_workunit["name"],
            span_id=raw_workunit["span_id"],
            parent_ids=tuple(raw_workunit["parent_ids"]),
            level=Level(raw_workunit["level"]),
            description=raw_workunit.get("description"),
            start_time=self._convert_time(raw_workunit["start_secs"], raw_workunit["start_nanos"]),
        )

    def _convert_completed_workunit(self, raw_workunit: RawWorkunit) -> Workunit:
        start_time = self._convert_time(raw_workunit["start_secs"], raw_workunit["start_nanos"])
        end_time = start_time + datetime.timedelta(
            seconds=raw_workunit["duration_secs"],
            microseconds=raw_workunit["duration_nanos"] // 1000,
        )
        return Workunit(
            name=raw_workunit["name"],
            span_id=raw_workunit["span_id"],
            parent_ids=tuple(raw_workunit["parent_ids"]),
            level=Level(raw_workunit["level"]),
            description=raw_workunit.get("description"),
            start_time=start_time,
            end_time=end_time,
            metadata=FrozenDict.deep_freeze(raw_workunit.get("metadata", {})),
        )

    def __call__(
        self,
        *,
        completed_workunits: tuple[RawWorkunit, ...],
        started_workunits: tuple[RawWorkunit, ...],
        context: StreamingWorkunitContext,
        finished: bool = False,
        **kwargs: Any,
    ) -> None:
        telemetry_context = _TelemetryContext(context)

        for started_workunit in started_workunits:
            self.processor.start_workunit(
                self._convert_incomplete_workunit(started_workunit), context=telemetry_context
            )

        for completed_workunit in completed_workunits:
            self.processor.complete_workunit(
                self._convert_completed_workunit(completed_workunit), context=telemetry_context
            )

        if finished:
            self.processor.finish(timeout=self.finish_timeout, context=telemetry_context)

1	# Copyright 2026 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	1✔
5
6	import datetime	1✔
7	from typing import Any, Mapping	1✔
8
9	from pants.backend.observability.opentelemetry.processor import (	1✔
10	IncompleteWorkunit,
11	Level,
12	Processor,
13	ProcessorContext,
14	Workunit,
15	)
16	from pants.engine.internals.native_engine import all_counter_names	1✔
17	from pants.engine.internals.scheduler import Workunit as RawWorkunit	1✔
18	from pants.engine.streaming_workunit_handler import StreamingWorkunitContext, WorkunitsCallback	1✔
19	from pants.util.frozendict import FrozenDict	1✔
20
21
22	class _TelemetryContext(ProcessorContext):	1✔
23	def __init__(self, pants_context: StreamingWorkunitContext) -> None:	1✔
NEW 24	self._pants_context = pants_context	×
25
26	def get_metrics(self) -> Mapping[str, int]:	1✔
NEW 27	metric_names = all_counter_names()	×
NEW 28	metrics = self._pants_context.get_metrics()	×
NEW 29	for metric_name in metric_names:	×
NEW 30	if metric_name not in metrics:	×
NEW 31	metrics[metric_name] = 0	×
NEW 32	return metrics	×
33
34
35	class TelemetryWorkunitsCallback(WorkunitsCallback):	1✔
36	def __init__(	1✔
37	self,
38	processor: Processor,
39	*,
40	finish_timeout: datetime.timedelta,
41	async_completion: bool,
42	) -> None:
43	self.processor: Processor = processor	1✔
44	self.finish_timeout = finish_timeout	1✔
45	self.async_completion = async_completion	1✔
46
47	@property	1✔
48	def can_finish_async(self) -> bool:	1✔
NEW 49	return self.async_completion	×
50
51	def _convert_time(self, seconds: int, nanoseconds: int) -> datetime.datetime:	1✔
NEW 52	t = datetime.datetime(year=1970, month=1, day=1, tzinfo=datetime.UTC)	×
NEW 53	t = t + datetime.timedelta(seconds=seconds, microseconds=nanoseconds // 1000)	×
NEW 54	return t	×
55
56	def _convert_incomplete_workunit(self, raw_workunit: RawWorkunit) -> IncompleteWorkunit:	1✔
NEW 57	return IncompleteWorkunit(	×
58	name=raw_workunit["name"],
59	span_id=raw_workunit["span_id"],
60	parent_ids=tuple(raw_workunit["parent_ids"]),
61	level=Level(raw_workunit["level"]),
62	description=raw_workunit.get("description"),
63	start_time=self._convert_time(raw_workunit["start_secs"], raw_workunit["start_nanos"]),
64	)
65
66	def _convert_completed_workunit(self, raw_workunit: RawWorkunit) -> Workunit:	1✔
NEW 67	start_time = self._convert_time(raw_workunit["start_secs"], raw_workunit["start_nanos"])	×
NEW 68	end_time = start_time + datetime.timedelta(	×
69	seconds=raw_workunit["duration_secs"],
70	microseconds=raw_workunit["duration_nanos"] // 1000,
71	)
NEW 72	return Workunit(	×
73	name=raw_workunit["name"],
74	span_id=raw_workunit["span_id"],
75	parent_ids=tuple(raw_workunit["parent_ids"]),
76	level=Level(raw_workunit["level"]),
77	description=raw_workunit.get("description"),
78	start_time=start_time,
79	end_time=end_time,
80	metadata=FrozenDict.deep_freeze(raw_workunit.get("metadata", {})),
81	)
82
83	def __call__(	1✔
84	self,
85	*,
86	completed_workunits: tuple[RawWorkunit, ...],
87	started_workunits: tuple[RawWorkunit, ...],
88	context: StreamingWorkunitContext,
89	finished: bool = False,
90	**kwargs: Any,
91	) -> None:
NEW 92	telemetry_context = _TelemetryContext(context)	×
93
NEW 94	for started_workunit in started_workunits:	×
NEW 95	self.processor.start_workunit(	×
96	self._convert_incomplete_workunit(started_workunit), context=telemetry_context
97	)
98
NEW 99	for completed_workunit in completed_workunits:	×
NEW 100	self.processor.complete_workunit(	×
101	self._convert_completed_workunit(completed_workunit), context=telemetry_context
102	)
103
NEW 104	if finished:	×
NEW 105	self.processor.finish(timeout=self.finish_timeout, context=telemetry_context)	×

pantsbuild / pants / 25565075335

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous