Add trace-per-test option (#34)

Trace-per-run is not always great e.g. it's confusing to have a
sequence of tests which are normally independent, and then the layout
changes when xdist is involved as you get concurrent spans (one for
each worker).

This also causes issues in the more bare-bone / simplistic frontends /
final collectors (e.g. OpenTelemetry Desktop) as they might not have
the most complex folding and filtering features. OTD for instance has
little to no ability to manipulate spans, but it is possible to select
individual traces and see just that trace's spans (in fact that's the
only mode).

As a result, generating a separate trace per test per run allows
easier classification, observation, and manipulation of the trace. I
would also assume in the long term it allows comparing the traces of
the same test in order to get insight into their evolution, something
which is more difficult with trace-per-run.

Finally an other issue with trace-per-run, mostly in long suites, is
that the trace remains incomplete (tools complain of missing parent
spans) until the entire run has completed, making observation during
run more difficult.

Author: xmo-odoo

src/pytest_opentelemetry/instrumentation.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Dict, Generator, Optional, Union
+from typing import Any, Dict, Iterator, Optional, Union
 
 import pytest
 from _pytest.config import Config
@@ -23,20 +23,12 @@
 tracer = trace.get_tracer('pytest-opentelemetry')
 
 
-class OpenTelemetryPlugin:
-    """A pytest plugin which produces OpenTelemetry spans around test sessions and
-    individual test runs."""
+class PerTestOpenTelemetryPlugin:
+    """base logic for all otel pytest integration"""
 
     @property
-    def session_name(self):
-        # Lazy initialise session name
-        if not hasattr(self, '_session_name'):
-            self._session_name = os.environ.get('PYTEST_RUN_NAME', 'test run')
-        return self._session_name
-
-    @session_name.setter
-    def session_name(self, name):
-        self._session_name = name
+    def item_parent(self) -> Union[str, None]:
+        return self.trace_parent
 
     @classmethod
     def get_trace_parent(cls, config: Config) -> Optional[Context]:
@@ -71,22 +63,7 @@ def pytest_configure(self, config: Config) -> None:
         configurator.resource_detectors.append(OTELResourceDetector())
         configurator.configure()
 
-    def pytest_sessionstart(self, session: Session) -> None:
-        self.session_span = tracer.start_span(
-            self.session_name,
-            context=self.trace_parent,
-            attributes={
-                "pytest.span_type": "run",
-            },
-        )
-        self.has_error = False
-
     def pytest_sessionfinish(self, session: Session) -> None:
-        self.session_span.set_status(
-            StatusCode.ERROR if self.has_error else StatusCode.OK
-        )
-
-        self.session_span.end()
         self.try_force_flush()
 
     def _attributes_from_item(self, item: Item) -> Dict[str, Union[str, int]]:
@@ -103,17 +80,16 @@ def _attributes_from_item(self, item: Item) -> Dict[str, Union[str, int]]:
         return attributes
 
     @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_protocol(self, item: Item) -> Generator[None, None, None]:
-        context = trace.set_span_in_context(self.session_span)
+    def pytest_runtest_protocol(self, item: Item) -> Iterator[None]:
         with tracer.start_as_current_span(
             item.nodeid,
             attributes=self._attributes_from_item(item),
-            context=context,
+            context=self.item_parent,
         ):
             yield
 
     @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_setup(self, item: Item) -> Generator[None, None, None]:
+    def pytest_runtest_setup(self, item: Item) -> Iterator[None]:
         with tracer.start_as_current_span(
             f'{item.nodeid}::setup',
             attributes=self._attributes_from_item(item),
@@ -145,23 +121,23 @@ def _name_from_fixturedef(self, fixturedef: FixtureDef, request: FixtureRequest)
     @pytest.hookimpl(hookwrapper=True)
     def pytest_fixture_setup(
         self, fixturedef: FixtureDef, request: FixtureRequest
-    ) -> Generator[None, None, None]:
+    ) -> Iterator[None]:
         with tracer.start_as_current_span(
             name=f'{self._name_from_fixturedef(fixturedef, request)} setup',
             attributes=self._attributes_from_fixturedef(fixturedef),
         ):
             yield
 
     @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_call(self, item: Item) -> Generator[None, None, None]:
+    def pytest_runtest_call(self, item: Item) -> Iterator[None]:
         with tracer.start_as_current_span(
             name=f'{item.nodeid}::call',
             attributes=self._attributes_from_item(item),
         ):
             yield
 
     @pytest.hookimpl(hookwrapper=True)
-    def pytest_runtest_teardown(self, item: Item) -> Generator[None, None, None]:
+    def pytest_runtest_teardown(self, item: Item) -> Iterator[None]:
         with tracer.start_as_current_span(
             name=f'{item.nodeid}::teardown',
             attributes=self._attributes_from_item(item),
@@ -188,7 +164,7 @@ def pytest_runtest_teardown(self, item: Item) -> Generator[None, None, None]:
     @pytest.hookimpl(hookwrapper=True)
     def pytest_fixture_post_finalizer(
         self, fixturedef: FixtureDef, request: SubRequest
-    ) -> Generator[None, None, None]:
+    ) -> Iterator[None]:
         """When the span for a fixture teardown is created by
         pytest_runtest_teardown or a previous pytest_fixture_post_finalizer, we
         need to update the name and attributes now that we know which fixture it
@@ -250,10 +226,52 @@ def pytest_runtest_logreport(self, report: TestReport) -> None:
 
         has_error = report.outcome == 'failed'
         status_code = StatusCode.ERROR if has_error else StatusCode.OK
-        self.has_error |= has_error
         trace.get_current_span().set_status(status_code)
 
 
+class OpenTelemetryPlugin(PerTestOpenTelemetryPlugin):
+    """A pytest plugin which produces OpenTelemetry spans around test sessions and
+    individual test runs."""
+
+    @property
+    def session_name(self):
+        # Lazy initialise session name
+        if not hasattr(self, '_session_name'):
+            self._session_name = os.environ.get('PYTEST_RUN_NAME', 'test run')
+        return self._session_name
+
+    @session_name.setter
+    def session_name(self, name):
+        self._session_name = name
+
+    @property
+    def item_parent(self) -> Union[str, None]:
+        context = trace.set_span_in_context(self.session_span)
+        return context
+
+    def pytest_sessionstart(self, session: Session) -> None:
+        self.session_span = tracer.start_span(
+            self.session_name,
+            context=self.trace_parent,
+            attributes={
+                "pytest.span_type": "run",
+            },
+        )
+        self.has_error = False
+
+    def pytest_sessionfinish(self, session: Session) -> None:
+        self.session_span.set_status(
+            StatusCode.ERROR if self.has_error else StatusCode.OK
+        )
+
+        self.session_span.end()
+        super().pytest_sessionfinish(session)
+
+    def pytest_runtest_logreport(self, report: TestReport) -> None:
+        super().pytest_runtest_logreport(report)
+        self.has_error |= report.when == 'call' and report.outcome == 'failed'
+
+
 try:
     from xdist.workermanage import WorkerController  # pylint: disable=unused-import
 except ImportError:  # pragma: no cover

src/pytest_opentelemetry/plugin.py

@@ -26,16 +26,25 @@ def pytest_addoption(parser: Parser) -> None:
             'trace.  If it is omitted, this test run will start a new trace.'
         ),
     )
+    group.addoption(
+        "--trace-per-test",
+        action="store_true",
+        default=False,
+        help="Creates a separate trace per test instead of a trace for the test run",
+    )
 
 
 def pytest_configure(config: Config) -> None:
     # pylint: disable=import-outside-toplevel
     from pytest_opentelemetry.instrumentation import (
         OpenTelemetryPlugin,
+        PerTestOpenTelemetryPlugin,
         XdistOpenTelemetryPlugin,
     )
 
-    if config.pluginmanager.has_plugin("xdist"):
+    if config.getvalue('--trace-per-test'):
+        config.pluginmanager.register(PerTestOpenTelemetryPlugin())
+    elif config.pluginmanager.has_plugin("xdist"):
         config.pluginmanager.register(XdistOpenTelemetryPlugin())
     else:
         config.pluginmanager.register(OpenTelemetryPlugin())

tests/test_sessions.py

@@ -1,13 +1,15 @@
 import os
 from contextlib import contextmanager
-from typing import Dict, Generator, Optional
+from typing import Dict, Generator, List, Optional
 from unittest.mock import Mock, patch
 
+import pytest
 from _pytest.pytester import Pytester
 from opentelemetry import trace
 
 from pytest_opentelemetry.instrumentation import (
     OpenTelemetryPlugin,
+    PerTestOpenTelemetryPlugin,
     XdistOpenTelemetryPlugin,
 )
 
@@ -187,13 +189,50 @@ def test_two():
     result.assert_outcomes(passed=2)
 
 
+@pytest.mark.parametrize(
+    'args',
+    [
+        pytest.param([], id="default"),
+        pytest.param(['-n', '2'], id="xdist"),
+        pytest.param(['-p', 'no:xdist'], id='no:xdist'),
+    ],
+)
+def test_trace_per_test(
+    pytester: Pytester, span_recorder: SpanRecorder, args: List[str]
+) -> None:
+    pytester.makepyfile(
+        """
+        from opentelemetry import trace
+
+        def test_one():
+            span = trace.get_current_span()
+            assert span.context.trace_id == 0x1234567890abcdef1234567890abcdef
+
+        def test_two():
+            span = trace.get_current_span()
+            assert span.context.trace_id == 0x1234567890abcdef1234567890abcdef
+    """
+    )
+    result = pytester.runpytest_subprocess(
+        '--trace-per-test',
+        *args,
+        '--trace-parent',
+        '00-1234567890abcdef1234567890abcdef-fedcba0987654321-01',
+    )
+    result.assert_outcomes(passed=2)
+
+
 @patch.object(trace, 'get_tracer_provider')
 def test_force_flush_with_supported_provider(mock_get_tracer_provider):
     provider = Mock()
     provider.force_flush = Mock(return_value=None)
     mock_get_tracer_provider.return_value = provider
 
-    for plugin in OpenTelemetryPlugin, XdistOpenTelemetryPlugin:
+    for plugin in (
+        OpenTelemetryPlugin,
+        XdistOpenTelemetryPlugin,
+        PerTestOpenTelemetryPlugin,
+    ):
         assert plugin.try_force_flush() is True
 
 
@@ -202,5 +241,9 @@ def test_force_flush_with_unsupported_provider(mock_get_tracer_provider):
     provider = Mock(spec=trace.ProxyTracerProvider)
     mock_get_tracer_provider.return_value = provider
 
-    for plugin in OpenTelemetryPlugin, XdistOpenTelemetryPlugin:
+    for plugin in (
+        OpenTelemetryPlugin,
+        XdistOpenTelemetryPlugin,
+        PerTestOpenTelemetryPlugin,
+    ):
         assert plugin.try_force_flush() is False

tests/test_spans.py

@@ -1,5 +1,7 @@
 from collections import Counter
+from typing import List
 
+import pytest
 from _pytest.pytester import Pytester
 from opentelemetry.trace import SpanKind
 
@@ -412,8 +414,17 @@ def test_one(session_scoped: int, module_scoped: int, function_scoped: int):
     assert function_scoped.parent.span_id == setup.context.span_id
 
 
+@pytest.mark.parametrize(
+    'args',
+    [
+        pytest.param([], id='trace-per-suite'),
+        pytest.param(['--trace-per-test'], id='trace-per-test'),
+    ],
+)
 def test_spans_from_fixutres_used_multiple_times(
-    pytester: Pytester, span_recorder: SpanRecorder
+    pytester: Pytester,
+    span_recorder: SpanRecorder,
+    args: List[str],
 ) -> None:
     pytester.makepyfile(
         """
@@ -456,7 +467,7 @@ def test_c(self, class_scoped: int, function_scoped: int):
 
     """
     )
-    pytester.runpytest().assert_outcomes(passed=5)
+    pytester.runpytest(*args).assert_outcomes(passed=5)
     spans = Counter(span.name for span in span_recorder.finished_spans())
 
     assert spans['session_scoped setup'] == 1