From 9a9d96d4def7f2f446cdc466680af05834bf001b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 16:18:14 +0000 Subject: [PATCH] Fix Kafka rebalancing when opentracing is not installed Both rebalance callbacks trace their work from the span `_start_span_from_rebalancing()` returns, which is a no-op span whenever no tracer is configured. The no-op stand-in Faust falls back to when the `faust[opentracing]` extra is missing gave its spans no tracer, while `traced_from_parent_span()` starts its child span from `parent.tracer`, so every rebalance raised: AttributeError: 'NoneType' object has no attribute 'start_span' `on_rebalance_start()` has already run by then, and `on_rebalance_end()` is only reached through recovery -- downstream of the callback that raised -- so the app was left with `rebalancing` stuck true. The stand-in's spans now carry the tracer that made them, as the real library's do, and `Tracer.start_span()` returns that tracer's own no-op span rather than a fresh tracerless one. The same drift hid a second divergence: the stand-in defined `Span.operation_name`, which `AIOKafkaConsumerThread` reads to tell a real span from a no-op one, catching the `AttributeError` the real library raises. Defining it sent no-op spans down the real-span path, where they failed on `_real_finish`. Dropping it restores the sentinel. Two smaller parity fixes go with those: `Span.__exit__` now calls `finish()`, which is what the driver's lazy spans rebind and what the real library calls on exit, and `start_child_span()` goes through the parent's tracer instead of returning an unrelated span. Finally, `traced_from_parent_span()` no longer assumes a parent span has a tracer. A span is whatever the configured tracer hands back, and tracing is instrumentation: it now runs the wrapped function untraced rather than breaking its caller. `Tracer.start_active_span()` is dropped. Faust never calls it, and the real one returns a `Scope` wrapping the span rather than the span itself, so the stand-in's version would have mis-served any caller. Tests run the stand-in side by side with the real library, so a divergence fails as a parametrization rather than resting on a claim about what the real library does, and cover both rebalance callbacks with the stand-in substituted in. Eleven of them fail before this change. Fixes #786 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Sejq9tYzzeqWeZAgA3EB6s --- CHANGELOG.md | 17 +++ faust/utils/_opentracing.py | 48 +++++-- faust/utils/tracing.py | 30 ++-- tests/unit/utils/test_opentracing.py | 205 +++++++++++++++++++++++++++ 4 files changed, 275 insertions(+), 25 deletions(-) create mode 100644 tests/unit/utils/test_opentracing.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f9524dce..a5be4a508 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,6 +53,23 @@ resumes the Keep a Changelog format. - `faust[aerospike]` installed nothing: `requirements/extras/aerospike.txt` shipped without the matching `BUNDLES` entry in `setup.py`, despite being advertised in the README. A new test guards both directions of that mapping. +- Every Kafka rebalance raised `AttributeError: 'NoneType' object has no + attribute 'start_span'` when the `faust[opentracing]` extra was not installed + (#786). Both rebalance callbacks trace their work from the span + `_start_span_from_rebalancing()` returns, which is a no-op span whenever no + tracer is configured — but the no-op stand-in Faust falls back to gave its + spans no tracer, while `traced_from_parent_span()` starts its child span from + `parent.tracer`. `on_rebalance_start()` has already run by then, and + `on_rebalance_end()` is only reached through recovery, downstream of the + callback that raised, so the app was left with `rebalancing` stuck true. + The stand-in's spans now carry the tracer that made them, as the real + library's do, and `traced_from_parent_span()` no longer assumes a parent span + has a tracer: it runs the wrapped function untraced rather than breaking its + caller. The same drift hid a second divergence — the stand-in defined + `Span.operation_name`, which the aiokafka consumer thread reads to tell a + real span from a no-op one (catching the `AttributeError` the real library + raises), so no-op spans took the real-span path and failed on `_real_finish`. + New tests check the stand-in side by side with the real library. ### Changed - The `examples/fastapi/` directory is now `examples/fastapi_project/`. The old diff --git a/faust/utils/_opentracing.py b/faust/utils/_opentracing.py index 16c617bc8..15b722e23 100644 --- a/faust/utils/_opentracing.py +++ b/faust/utils/_opentracing.py @@ -6,25 +6,45 @@ does real work when an ``app.tracer`` is configured (or the ``TracingSensor`` is used), which requires the real ``opentracing`` package to be installed. -This intentionally implements only the small surface Faust touches. +This intentionally implements only the small surface Faust touches, but over +that surface it must stay *substitutable* for the real library: Faust reaches +through the objects it is handed (``parent.tracer.start_span(...)``), so a +plausible-looking attribute holding the wrong value breaks callers that never +mention this module. Absences count as surface too -- the aiokafka consumer +thread reads ``Span.operation_name`` and treats the resulting +:exc:`AttributeError` as "this is not a real span", which only works because +:class:`opentracing.Span` does not define that attribute either. + +``Tracer.start_active_span`` is deliberately *not* provided: Faust never calls +it, and the real one returns a ``Scope`` wrapping the span rather than the span +itself, so a stand-in returning a :class:`Span` would mis-serve anyone who did. """ from typing import Any, Literal class Span: - """A span that does nothing.""" + """A span that does nothing. - operation_name: str = "" + Mirrors :class:`opentracing.Span`, including its ``(tracer, context)`` + signature: every span knows the tracer that made it, so callers can start + a child span from any span they are given. + """ - def __init__(self, *args: Any, **kwargs: Any) -> None: - self.context = _SpanContext() - self.tracer: Any = None + def __init__(self, tracer: Any, context: Any = None) -> None: + self.tracer: Any = tracer + self.context: Any = _SpanContext() if context is None else context def __enter__(self) -> "Span": return self def __exit__(self, *exc_info: Any) -> Literal[False]: + # ``finish`` rather than ``pass``: subclasses (and the lazy-span + # rebinding in the aiokafka driver) override ``finish``, and the real + # library calls it on exit. + if exc_info and exc_info[0] is not None: + self.set_tag(tags.ERROR, True) + self.finish() return False def finish(self, *args: Any, **kwargs: Any) -> None: ... @@ -54,13 +74,10 @@ class Tracer: """A tracer that produces only no-op spans.""" def __init__(self, *args: Any, **kwargs: Any) -> None: - self._noop_span = Span() + self._noop_span = Span(tracer=self) def start_span(self, *args: Any, **kwargs: Any) -> Span: - return Span() - - def start_active_span(self, *args: Any, **kwargs: Any) -> Span: - return Span() + return self._noop_span def extract(self, *args: Any, **kwargs: Any) -> Any: return None @@ -80,8 +97,13 @@ def child_of(*args: Any, **kwargs: Any) -> Any: return None -def start_child_span(*args: Any, **kwargs: Any) -> Span: - return Span() +def start_child_span( + parent_span: Span, operation_name: Any = None, *args: Any, **kwargs: Any +) -> Span: + return parent_span.tracer.start_span( + operation_name=operation_name, + child_of=parent_span.context, + ) class Format: diff --git a/faust/utils/tracing.py b/faust/utils/tracing.py index faa2bc3dd..cf2a91a21 100644 --- a/faust/utils/tracing.py +++ b/faust/utils/tracing.py @@ -83,18 +83,24 @@ def _inner(*args: Any, **kwargs: Any) -> Any: parent = parent_span if parent is None: parent = current_span() - if parent is not None: - child = parent.tracer.start_span( - operation_name=operation_name, - child_of=parent, - tags={**extra_context, **more_context}, - ) - if callback is not None: - callback(child) - on_exit = (_restore_span, (parent, child)) - set_current_span(child) - return call_with_trace(child, fun, on_exit, *args, **kwargs) - return fun(*args, **kwargs) + # A child span can only come from the parent's own tracer, and + # not every span has one: ``getattr`` covers both "no parent at + # all" and "parent span carrying no tracer". Tracing is + # instrumentation, so fall back to calling the wrapped function + # untraced instead of failing its caller. + tracer = getattr(parent, "tracer", None) + if tracer is None: + return fun(*args, **kwargs) + child = tracer.start_span( + operation_name=operation_name, + child_of=parent, + tags={**extra_context, **more_context}, + ) + if callback is not None: + callback(child) + on_exit = (_restore_span, (parent, child)) + set_current_span(child) + return call_with_trace(child, fun, on_exit, *args, **kwargs) return _inner diff --git a/tests/unit/utils/test_opentracing.py b/tests/unit/utils/test_opentracing.py new file mode 100644 index 000000000..874db82f5 --- /dev/null +++ b/tests/unit/utils/test_opentracing.py @@ -0,0 +1,205 @@ +"""The no-op stand-in used when :pypi:`opentracing` is not installed. + +``requirements/test.txt`` pulls in the ``opentracing`` extra, so the real +library is always importable here and the ``except ImportError`` fallback in +``faust.utils.tracing`` never fires by accident. These tests therefore reach +for the stand-in explicitly, and check it two ways: side by side with the real +library, so a divergence shows up as a failing parametrization rather than as +a claim about what the real library does; and substituted for it, so the code +paths that broke without the extra (#786) are actually walked. +""" + +import opentracing +import pytest +from opentracing.ext import tags as real_tags + +from faust.types import TP +from faust.utils import _opentracing as shim +from faust.utils.tracing import ( + current_span, + noop_span, + set_current_span, + traced_from_parent_span, +) +from tests.helpers import AsyncMock + +#: Both implementations, so every parity check runs against the real library +#: too: a test that only the stand-in passes proves nothing about parity. +MODULES = [pytest.param(opentracing, id="real"), pytest.param(shim, id="shim")] + + +def a_noop_span(module): + """Build a no-op span the way ``faust.utils.tracing.noop_span`` does.""" + return module.Tracer()._noop_span + + +@pytest.fixture(autouse=True) +def _reset_current_span(): + # ``traced_from_parent_span`` leaves the parent span current. + yield + set_current_span(None) + + +@pytest.fixture() +def without_opentracing(monkeypatch): + """Make Faust build no-op spans as it does without the extra installed.""" + monkeypatch.setattr("faust.utils.tracing.opentracing", shim) + + +class TestSpanParity: + @pytest.mark.parametrize("module", MODULES) + def test_noop_span_carries_the_tracer_that_made_it(self, module): + # The bug behind #786: ``traced_from_parent_span`` starts its child + # from ``parent.tracer``, so a span whose tracer is None cannot be a + # parent -- and every rebalance passes a no-op span as one. + span = a_noop_span(module) + assert span.tracer is not None + assert span.tracer.start_span(operation_name="child", child_of=span) + + @pytest.mark.parametrize("module", MODULES) + def test_start_span_returns_the_tracers_noop_span(self, module): + tracer = module.Tracer() + assert tracer.start_span(operation_name="x") is tracer._noop_span + + @pytest.mark.parametrize("module", MODULES) + def test_noop_span_has_no_operation_name(self, module): + # ``AIOKafkaConsumerThread`` reads ``span.operation_name`` and catches + # AttributeError to mean "not a real span"; defining the attribute + # sends no-op spans down the real-span path, which then reaches for + # ``_real_finish`` and fails. + assert not hasattr(a_noop_span(module), "operation_name") + + @pytest.mark.parametrize("module", MODULES) + def test_span_exit_finishes_the_span(self, module): + # The driver's lazy spans work by rebinding ``finish``, which only + # runs if leaving the span calls it. + finished = [] + + class RecordingSpan(module.Span): + def finish(self, *args, **kwargs): + finished.append(True) + + tracer = module.Tracer() + with RecordingSpan(tracer, a_noop_span(module).context): + pass + + assert finished == [True] + + @pytest.mark.parametrize("module", MODULES) + def test_start_child_span_hangs_off_the_parents_tracer(self, module): + parent = a_noop_span(module) + child = module.start_child_span(parent, "child-op") + assert child.tracer is not None + assert child.tracer is parent.tracer + + @pytest.mark.parametrize( + "real, fake, names", + [ + pytest.param( + opentracing.Format, + shim.Format, + ("TEXT_MAP", "HTTP_HEADERS", "BINARY"), + id="Format", + ), + pytest.param( + real_tags, + shim.tags, + ( + "ERROR", + "SAMPLING_PRIORITY", + "SPAN_KIND", + "COMPONENT", + "MESSAGE_BUS_DESTINATION", + ), + id="tags", + ), + ], + ) + def test_mirrored_constants_match(self, real, fake, names): + assert {n: getattr(fake, n) for n in names} == { + n: getattr(real, n) for n in names + } + + +class TestTracedFromParentSpan: + @pytest.mark.parametrize("module", MODULES) + def test_traces_a_sync_function_from_a_noop_parent(self, module): + @traced_from_parent_span(a_noop_span(module)) + def double(x): + return x * 2 + + assert double(21) == 42 + + @pytest.mark.parametrize("module", MODULES) + @pytest.mark.asyncio + async def test_traces_a_coroutine_from_a_noop_parent(self, module): + parent = a_noop_span(module) + + @traced_from_parent_span(parent) + async def double(x): + return x * 2 + + assert await double(21) == 42 + assert current_span() is parent + + @pytest.mark.parametrize("module", MODULES) + def test_propagates_the_wrapped_functions_error(self, module): + @traced_from_parent_span(a_noop_span(module)) + def raiser(): + raise ValueError("boom") + + with pytest.raises(ValueError, match="boom"): + raiser() + + def test_runs_untraced_when_the_parent_has_no_tracer(self): + # Not reachable through the stand-in any more, but a span is whatever + # the configured tracer hands back: instrumentation must not be the + # thing that breaks the call. + class TracerlessSpan(shim.Span): + tracer = None + + @traced_from_parent_span(TracerlessSpan(tracer=None)) + def double(x): + return x * 2 + + assert double(21) == 42 + + def test_runs_untraced_with_no_parent_span_at_all(self): + @traced_from_parent_span() + def double(x): + return x * 2 + + assert double(21) == 42 + + +class TestRebalanceWithoutOpentracing: + """Regression tests for #786. + + Both rebalance callbacks trace their work from the span + ``_start_span_from_rebalancing`` returns, which is a no-op span whenever no + tracer is configured -- so without the extra installed *every* rebalance + raised ``AttributeError`` and left the app mid-rebalance. + """ + + def test_noop_span_comes_from_the_stand_in(self, *, without_opentracing): + assert isinstance(noop_span(), shim.Span) + + @pytest.mark.asyncio + async def test_on_partitions_revoked(self, *, app, without_opentracing): + consumer = app.consumer + consumer._on_partitions_revoked = AsyncMock(name="_on_partitions_revoked") + revoked = {TP("foo", 0)} + + await consumer.on_partitions_revoked(revoked) + + consumer._on_partitions_revoked.assert_called_once_with(revoked) + + @pytest.mark.asyncio + async def test_on_partitions_assigned(self, *, app, without_opentracing): + consumer = app.consumer + consumer._on_partitions_assigned = AsyncMock(name="_on_partitions_assigned") + assigned = {TP("foo", 0)} + + await consumer.on_partitions_assigned(assigned) + + consumer._on_partitions_assigned.assert_called_once_with(assigned, 0)