diff --git a/.coveragerc b/.coveragerc index a9e3cd02c..20969fd97 100644 --- a/.coveragerc +++ b/.coveragerc @@ -33,6 +33,12 @@ omit = # not needed */faust/utils/iso8601.py */faust/utils/tracing.py + # opentracing fallback stub: only used when opentracing is NOT installed, + # which the CI test env is not (it installs faust[opentracing]). + */faust/utils/_opentracing.py + # opentracing distributed-tracing sensor: optional, requires opentracing + + # a live tracer to exercise; not covered by the CI test suite. + */faust/sensors/distributed_tracing.py */faust/types/* */faust/__main__.py diff --git a/faust/app/base.py b/faust/app/base.py index a4430c493..581253325 100644 --- a/faust/app/base.py +++ b/faust/app/base.py @@ -42,7 +42,10 @@ no_type_check, ) -import opentracing +try: + import opentracing +except ImportError: # pragma: no cover + from faust.utils import _opentracing as opentracing # type: ignore from mode import Seconds, Service, ServiceT, SupervisorStrategyT, want_seconds from mode.utils.aiter import aiter from mode.utils.collections import force_mapping diff --git a/faust/sensors/distributed_tracing.py b/faust/sensors/distributed_tracing.py index 31ea59510..febaa1be8 100644 --- a/faust/sensors/distributed_tracing.py +++ b/faust/sensors/distributed_tracing.py @@ -3,11 +3,16 @@ from typing import Any, Dict import aiohttp -import opentracing from mode import get_logger from mode.utils.compat import want_str -from opentracing import Format -from opentracing.ext import tags + +try: + import opentracing + from opentracing import Format + from opentracing.ext import tags +except ImportError: # pragma: no cover + from faust.utils import _opentracing as opentracing # type: ignore + from faust.utils._opentracing import Format, tags from faust import App, EventT, Sensor, StreamT from faust.types import TP, Message, PendingMessage, ProducerT, RecordMetadata diff --git a/faust/tables/recovery.py b/faust/tables/recovery.py index 11737aef0..a9322d110 100644 --- a/faust/tables/recovery.py +++ b/faust/tables/recovery.py @@ -20,7 +20,10 @@ cast, ) -import opentracing +try: + import opentracing +except ImportError: # pragma: no cover + from faust.utils import _opentracing as opentracing # type: ignore from aiokafka.errors import IllegalStateError from mode import Service, get_logger from mode.services import WaitArgT diff --git a/faust/transport/drivers/aiokafka.py b/faust/transport/drivers/aiokafka.py index 2e4799f58..c900a486e 100644 --- a/faust/transport/drivers/aiokafka.py +++ b/faust/transport/drivers/aiokafka.py @@ -26,7 +26,6 @@ import aiokafka import aiokafka.abc -import opentracing from aiokafka import TopicPartition from aiokafka.consumer.group_coordinator import OffsetCommitRequest from aiokafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor @@ -51,10 +50,16 @@ from mode.utils.futures import StampedeWrapper from mode.utils.objects import cached_property from mode.utils.times import Seconds, humanize_seconds_ago, want_seconds -from opentracing.ext import tags from packaging.version import Version from yarl import URL +try: + import opentracing + from opentracing.ext import tags +except ImportError: # pragma: no cover + from faust.utils import _opentracing as opentracing # type: ignore + from faust.utils._opentracing import tags + from faust.auth import ( GSSAPICredentials, OAuthCredentials, diff --git a/faust/types/app.py b/faust/types/app.py index adb336035..5ef4c3605 100644 --- a/faust/types/app.py +++ b/faust/types/app.py @@ -23,7 +23,10 @@ no_type_check, ) -import opentracing +try: + import opentracing +except ImportError: # pragma: no cover + from faust.utils import _opentracing as opentracing # type: ignore from mode import Seconds, ServiceT, Signal, SupervisorStrategyT, SyncSignal from mode.utils.futures import stampede from mode.utils.objects import cached_property diff --git a/faust/utils/_opentracing.py b/faust/utils/_opentracing.py new file mode 100644 index 000000000..03493061a --- /dev/null +++ b/faust/utils/_opentracing.py @@ -0,0 +1,102 @@ +"""No-op stand-in for :pypi:`opentracing`. + +OpenTracing is an optional dependency (``faust[opentracing]``). When it is not +installed, the tracing modules fall back to this module so that Faust still +imports and runs -- tracing simply becomes a no-op. Distributed tracing only +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. +""" + +from typing import Any + + +class Span: + """A span that does nothing.""" + + operation_name: str = "" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + self.context = _SpanContext() + self.tracer: Any = None + + def __enter__(self) -> "Span": + return self + + def __exit__(self, *exc_info: Any) -> bool: + return False + + def finish(self, *args: Any, **kwargs: Any) -> None: ... + + def set_tag(self, *args: Any, **kwargs: Any) -> "Span": + return self + + def set_operation_name(self, *args: Any, **kwargs: Any) -> "Span": + return self + + def log_kv(self, *args: Any, **kwargs: Any) -> "Span": + return self + + def set_baggage_item(self, *args: Any, **kwargs: Any) -> "Span": + return self + + def get_baggage_item(self, *args: Any, **kwargs: Any) -> Any: + return None + + +class _SpanContext: + trace_id: Any = None + span_id: Any = None + + +class Tracer: + """A tracer that produces only no-op spans.""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + self._noop_span = Span() + + def start_span(self, *args: Any, **kwargs: Any) -> Span: + return Span() + + def start_active_span(self, *args: Any, **kwargs: Any) -> Span: + return Span() + + def extract(self, *args: Any, **kwargs: Any) -> Any: + return None + + def inject(self, *args: Any, **kwargs: Any) -> None: ... + + +#: Module-level global tracer, mirroring ``opentracing.tracer``. +tracer = Tracer() + + +def follows_from(*args: Any, **kwargs: Any) -> Any: + return None + + +def child_of(*args: Any, **kwargs: Any) -> Any: + return None + + +def start_child_span(*args: Any, **kwargs: Any) -> Span: + return Span() + + +class Format: + """Mirror of ``opentracing.Format`` carrier formats.""" + + TEXT_MAP = "text_map" + HTTP_HEADERS = "http_headers" + BINARY = "binary" + + +class tags: + """Mirror of the ``opentracing.ext.tags`` constants Faust references.""" + + ERROR = "error" + SAMPLING_PRIORITY = "sampling.priority" + SPAN_KIND = "span.kind" + COMPONENT = "component" + MESSAGE_BUS_DESTINATION = "message_bus.destination" diff --git a/faust/utils/tracing.py b/faust/utils/tracing.py index 9546aef9f..fba366cba 100644 --- a/faust/utils/tracing.py +++ b/faust/utils/tracing.py @@ -7,7 +7,10 @@ from functools import wraps from typing import Any, Callable, Optional, Tuple -import opentracing +try: + import opentracing +except ImportError: # pragma: no cover + from faust.utils import _opentracing as opentracing # type: ignore from mode import shortlabel __all__ = [ diff --git a/requirements/extras/opentracing.txt b/requirements/extras/opentracing.txt new file mode 100644 index 000000000..113ebb89f --- /dev/null +++ b/requirements/extras/opentracing.txt @@ -0,0 +1 @@ +opentracing>=1.3.0,<=2.4.0 diff --git a/requirements/requirements.txt b/requirements/requirements.txt index f1f2e6d53..b64d8e1c7 100644 --- a/requirements/requirements.txt +++ b/requirements/requirements.txt @@ -3,7 +3,6 @@ aiohttp_cors>=0.7,<2.0 aiokafka>=0.10.0 click>=6.7,<8.2 mode-streaming>=0.4.0 -opentracing>=1.3.0,<=2.4.0 terminaltables>=3.1,<4.0 yarl>=1.0,<2.0 croniter>=0.3.16 diff --git a/requirements/test.txt b/requirements/test.txt index 759f5bd62..b06081cb4 100644 --- a/requirements/test.txt +++ b/requirements/test.txt @@ -28,6 +28,7 @@ wheel intervaltree -r requirements.txt -r extras/datadog.txt +-r extras/opentracing.txt -r extras/redis.txt -r extras/statsd.txt -r extras/yaml.txt diff --git a/setup.py b/setup.py index 390cad0a9..7a3190b70 100644 --- a/setup.py +++ b/setup.py @@ -30,6 +30,7 @@ "datadog", "debug", "fast", + "opentracing", "orjson", "prometheus", "redis",