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

97.62

/src/python/pants/backend/observability/opentelemetry/single_threaded_processor_test.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
import queue
from collections.abc import Mapping

from pants.backend.observability.opentelemetry.processor import (
    IncompleteWorkunit,
    Level,
    Processor,
    ProcessorContext,
    Workunit,
)
from pants.backend.observability.opentelemetry.single_threaded_processor import (
    SingleThreadedProcessor,
)
from pants.util.frozendict import FrozenDict


class CapturingProcessor(Processor):
    def __init__(self) -> None:
        self.initialize_called = False
        self.started_workunits: queue.Queue[IncompleteWorkunit] = queue.Queue()
        self.completed_workunits: queue.Queue[Workunit] = queue.Queue()
        self.finish_called = False

    def initialize(self) -> None:
        self.initialize_called = True

    def start_workunit(self, workunit: IncompleteWorkunit, *, context: ProcessorContext) -> None:
        self.started_workunits.put_nowait(workunit)

    def complete_workunit(self, workunit: Workunit, *, context: ProcessorContext) -> None:
        self.completed_workunits.put_nowait(workunit)

    def finish(
        self, timeout: datetime.timedelta | None = None, *, context: ProcessorContext
    ) -> None:
        self.finish_called = True


class MockProcessorContext(ProcessorContext):
    def get_metrics(self) -> Mapping[str, int]:
        return {}


def test_single_threaded_processor_roundtrip() -> None:
    context = MockProcessorContext()
    processor = CapturingProcessor()
    stp_processor = SingleThreadedProcessor(processor)

    stp_processor.initialize()
    assert processor.initialize_called

    start_time = datetime.datetime.now(datetime.UTC)
    incomplete_workunit = IncompleteWorkunit(
        name="test-span",
        span_id="SOME_SPAN_ID",
        parent_ids=("A_PARENT_SPAN_ID",),
        level=Level.INFO,
        description="This is where the span is described.",
        start_time=start_time,
    )
    stp_processor.start_workunit(workunit=incomplete_workunit, context=context)
    actual_incomplete_workunit = processor.started_workunits.get(timeout=0.250)
    assert actual_incomplete_workunit == incomplete_workunit

    start_time = datetime.datetime.now(datetime.UTC)
    workunit = Workunit(
        name=incomplete_workunit.name,
        span_id=incomplete_workunit.span_id,
        parent_ids=incomplete_workunit.parent_ids,
        level=incomplete_workunit.level,
        description=incomplete_workunit.description,
        start_time=incomplete_workunit.start_time,
        end_time=incomplete_workunit.start_time + datetime.timedelta(milliseconds=100),
        metadata=FrozenDict(),
    )
    stp_processor.complete_workunit(workunit=workunit, context=context)
    actual_workunit = processor.completed_workunits.get(timeout=0.250)
    assert actual_workunit == workunit

    stp_processor.finish(context=context)
    assert processor.finish_called

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	import queue	1✔
8	from collections.abc import Mapping	1✔
9
10	from pants.backend.observability.opentelemetry.processor import (	1✔
11	IncompleteWorkunit,
12	Level,
13	Processor,
14	ProcessorContext,
15	Workunit,
16	)
17	from pants.backend.observability.opentelemetry.single_threaded_processor import (	1✔
18	SingleThreadedProcessor,
19	)
20	from pants.util.frozendict import FrozenDict	1✔
21
22
23	class CapturingProcessor(Processor):	1✔
24	def __init__(self) -> None:	1✔
25	self.initialize_called = False	1✔
26	self.started_workunits: queue.Queue[IncompleteWorkunit] = queue.Queue()	1✔
27	self.completed_workunits: queue.Queue[Workunit] = queue.Queue()	1✔
28	self.finish_called = False	1✔
29
30	def initialize(self) -> None:	1✔
31	self.initialize_called = True	1✔
32
33	def start_workunit(self, workunit: IncompleteWorkunit, *, context: ProcessorContext) -> None:	1✔
34	self.started_workunits.put_nowait(workunit)	1✔
35
36	def complete_workunit(self, workunit: Workunit, *, context: ProcessorContext) -> None:	1✔
37	self.completed_workunits.put_nowait(workunit)	1✔
38
39	def finish(	1✔
40	self, timeout: datetime.timedelta \| None = None, *, context: ProcessorContext
41	) -> None:
42	self.finish_called = True	1✔
43
44
45	class MockProcessorContext(ProcessorContext):	1✔
46	def get_metrics(self) -> Mapping[str, int]:	1✔
NEW 47	return {}	×
48
49
50	def test_single_threaded_processor_roundtrip() -> None:	1✔
51	context = MockProcessorContext()	1✔
52	processor = CapturingProcessor()	1✔
53	stp_processor = SingleThreadedProcessor(processor)	1✔
54
55	stp_processor.initialize()	1✔
56	assert processor.initialize_called	1✔
57
58	start_time = datetime.datetime.now(datetime.UTC)	1✔
59	incomplete_workunit = IncompleteWorkunit(	1✔
60	name="test-span",
61	span_id="SOME_SPAN_ID",
62	parent_ids=("A_PARENT_SPAN_ID",),
63	level=Level.INFO,
64	description="This is where the span is described.",
65	start_time=start_time,
66	)
67	stp_processor.start_workunit(workunit=incomplete_workunit, context=context)	1✔
68	actual_incomplete_workunit = processor.started_workunits.get(timeout=0.250)	1✔
69	assert actual_incomplete_workunit == incomplete_workunit	1✔
70
71	start_time = datetime.datetime.now(datetime.UTC)	1✔
72	workunit = Workunit(	1✔
73	name=incomplete_workunit.name,
74	span_id=incomplete_workunit.span_id,
75	parent_ids=incomplete_workunit.parent_ids,
76	level=incomplete_workunit.level,
77	description=incomplete_workunit.description,
78	start_time=incomplete_workunit.start_time,
79	end_time=incomplete_workunit.start_time + datetime.timedelta(milliseconds=100),
80	metadata=FrozenDict(),
81	)
82	stp_processor.complete_workunit(workunit=workunit, context=context)	1✔
83	actual_workunit = processor.completed_workunits.get(timeout=0.250)	1✔
84	assert actual_workunit == workunit	1✔
85
86	stp_processor.finish(context=context)	1✔
87	assert processor.finish_called	1✔

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