diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f9524dce..e28af3f6a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,6 +53,14 @@ 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 failed when `opentracing` was not installed. The no-op + stand-in Faust falls back to gave its spans no tracer, but + `traced_from_parent_span` starts a child span from `parent.tracer`, so + `on_partitions_revoked` and `on_partitions_assigned` both raised + `AttributeError`. Because `on_rebalance_start()` had already run, the app was + left mid-rebalance rather than crashing outright, surfacing as agents that + timed out and stalled. The stand-in now matches the real library across the + surface Faust uses, and a new parity test guards it. ### 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..1eccdec4a 100644 --- a/faust/utils/_opentracing.py +++ b/faust/utils/_opentracing.py @@ -6,7 +6,10 @@ 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. +Only the surface Faust touches is implemented, but across it this must stay +substitutable: callers reach through what they are handed (``parent.tracer``), +and absences count too -- the consumer thread reads ``operation_name`` and +treats ``AttributeError`` as "not a real span". """ from typing import Any, Literal @@ -15,16 +18,15 @@ class Span: """A span that does nothing.""" - operation_name: str = "" - - def __init__(self, *args: Any, **kwargs: Any) -> None: + def __init__(self, tracer: Any, *args: Any, **kwargs: Any) -> None: self.context = _SpanContext() - self.tracer: Any = None + self.tracer: Any = tracer def __enter__(self) -> "Span": return self def __exit__(self, *exc_info: Any) -> Literal[False]: + self.finish() return False def finish(self, *args: Any, **kwargs: Any) -> None: ... @@ -54,13 +56,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 +79,10 @@ 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) class Format: diff --git a/tests/unit/utils/test_opentracing_shim.py b/tests/unit/utils/test_opentracing_shim.py new file mode 100644 index 000000000..df68a2fa2 --- /dev/null +++ b/tests/unit/utils/test_opentracing_shim.py @@ -0,0 +1,168 @@ +"""Parity tests for the no-op :pypi:`opentracing` stand-in. + +Three kinds of check, none sufficient alone: surface comparison catches a name +appearing or disappearing, value comparison catches a mirrored constant +drifting, and functional calls catch a name that exists but holds the wrong +value -- the bug that broke rebalancing, where ``Span.tracer`` existed and was +``None``. + +These exercise the shim's objects directly, so they do not prove the +``except ImportError`` fallback wiring itself. +""" + +import opentracing +import pytest +from opentracing.ext import tags as real_tags + +from faust.utils import _opentracing as shim +from faust.utils.tracing import current_span, set_current_span, traced_from_parent_span + +SPAN_SURFACE = frozenset( + { + "tracer", + "context", + "finish", + "set_tag", + "set_operation_name", + "log_kv", + "__enter__", + "__exit__", + } +) + +SPAN_ABSENT = frozenset({"operation_name"}) + +# ``start_active_span`` is excluded: the shim omits it by design. +TRACER_SURFACE = frozenset({"start_span", "extract", "inject", "_noop_span"}) + +# ``tags`` is excluded: the real one lives at ``opentracing.ext.tags``. +MODULE_SURFACE = frozenset( + { + "tracer", + "follows_from", + "child_of", + "start_child_span", + "Format", + "Span", + "Tracer", + } +) + +MODULES = [pytest.param(opentracing, id="real"), pytest.param(shim, id="shim")] + +SURFACES = [ + pytest.param( + lambda m: m.Tracer().start_span(operation_name="p"), SPAN_SURFACE, id="span" + ), + pytest.param(lambda m: m.Tracer(), TRACER_SURFACE, id="tracer"), + pytest.param(lambda m: m, MODULE_SURFACE, id="module"), +] + +CONSTANTS = [ + pytest.param( + real_tags, + shim.tags, + ( + "ERROR", + "SAMPLING_PRIORITY", + "SPAN_KIND", + "COMPONENT", + "MESSAGE_BUS_DESTINATION", + ), + id="tags", + ), + pytest.param( + opentracing.Format, + shim.Format, + ("TEXT_MAP", "HTTP_HEADERS", "BINARY"), + id="format", + ), +] + + +def noop_span(module): + return module.Tracer().start_span(operation_name="parity-probe") + + +@pytest.fixture(autouse=True) +def restore_current_span(): + yield + set_current_span(None) + + +@pytest.mark.parametrize("get, names", SURFACES) +def test_surface_matches_real_library(get, names): + real, fake = get(opentracing), get(shim) + for name in names: + assert hasattr(real, name) == hasattr(fake, name), name + + +@pytest.mark.parametrize("real, fake, names", CONSTANTS) +def test_constant_values_match_real_library(real, fake, names): + for name in names: + assert getattr(real, name) == getattr(fake, name), name + + +@pytest.mark.parametrize("module", MODULES) +def test_start_span_returns_the_tracers_own_usable_noop_span(module): + tracer = module.Tracer() + span = tracer.start_span(operation_name="x") + assert span is tracer._noop_span + assert span.tracer.start_span(operation_name="child") is not None + + +@pytest.mark.parametrize("module", MODULES) +def test_span_hides_operation_name(module): + for name in SPAN_ABSENT: + assert not hasattr(noop_span(module), name) + + +@pytest.mark.parametrize("module", MODULES) +def test_start_child_span_returns_a_span_with_a_tracer(module): + assert module.start_child_span(noop_span(module), "child-op").tracer is not None + + +@pytest.mark.parametrize("module", MODULES) +def test_traced_from_parent_span_runs_with_a_noop_parent(module): + """Both rebalance callbacks hand this decorator a no-op span.""" + + @traced_from_parent_span(noop_span(module)) + def work(value): + return value * 2 + + assert work(21) == 42 + + +@pytest.mark.parametrize("module", MODULES) +async def test_traced_from_parent_span_awaits_a_coroutine(module): + """Production awaits, routing through ``corowrapped`` and ``_restore_span``.""" + parent = noop_span(module) + + @traced_from_parent_span(parent) + async def work(value): + return value * 2 + + assert await work(21) == 42 + assert current_span() is parent + + +def test_shim_exit_finishes_the_span(): + finished = [] + + class RecordingSpan(shim.Span): + def finish(self, *args, **kwargs): + finished.append(True) + + with RecordingSpan(tracer=shim.Tracer()): + pass + + assert finished == [True] + + +def test_shim_span_requires_a_tracer(): + with pytest.raises(TypeError): + shim.Span() + + +def test_shim_hides_start_active_span(): + assert not hasattr(shim.Tracer, "start_active_span")