diff --git a/CHANGELOG.md b/CHANGELOG.md index 07382368b82a9493f37b1b71934baf60d8fa2f3e..6ca1434df4ff94aaf503623fa6b36e28d195f878 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,2 +1,4 @@ # v2.9.15 +* add `trace_future` + diff --git a/docs/index.rst b/docs/index.rst index 4c0f60fcc5a8ed14818e7f1578ca6df0474445ad..73a6aae6b8801b86cf339ad9507896bd2d5b7848 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -30,6 +30,7 @@ Visit the project's page at GitHub_! exceptions processes cassandra + opentracing Indices and tables diff --git a/docs/opentracing.rst b/docs/opentracing.rst new file mode 100644 index 0000000000000000000000000000000000000000..8cf8cd5c64dc5bed331457472f54e7ae3c643197 --- /dev/null +++ b/docs/opentracing.rst @@ -0,0 +1,10 @@ +**This module is available only if you have opentracing installed** + +OpenTracing +=========== + +trace_future +------------ + +.. autofunction:: satella.opentracing.trace_future + diff --git a/satella/__init__.py b/satella/__init__.py index de249f240c377da9c233aa21c1115d473262db6f..392a9ae7eeb962da32b29ef8488544215a24264b 100644 --- a/satella/__init__.py +++ b/satella/__init__.py @@ -1 +1 @@ -__version__ = '2.9.15_a1' +__version__ = '2.9.15_a2' diff --git a/satella/opentracing/__init__.py b/satella/opentracing/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..5d41143440c3ecc449707b13b13ae22aecf683b3 --- /dev/null +++ b/satella/opentracing/__init__.py @@ -0,0 +1,3 @@ +from .trace import trace_future + +__all__ = ['trace_future'] diff --git a/satella/opentracing/trace.py b/satella/opentracing/trace.py new file mode 100644 index 0000000000000000000000000000000000000000..81be1cd7fd03f7473b384b054449cfdd2119a14f --- /dev/null +++ b/satella/opentracing/trace.py @@ -0,0 +1,37 @@ +import typing as tp +from concurrent.futures import Future +try: + from opentracing import Span +except ImportError: + class Span: + pass + +try: + from cassandra.cluster import ResponseFuture +except ImportError: + class ResponseFuture: + pass + + +def trace_future(future: tp.Union[ResponseFuture, Future], span: Span): + """ + Install a handler that will close a span upon a future completing, attaching the exception + contents if the future ends with an exception. + + :param future: can be either a normal Future or a Cassandra's ResponseFuture + :param span: span to close + """ + if isinstance(future, ResponseFuture): + def close_exception(exc): + Span._on_error(span, type(exc), exc, '<unavailable>') + span.finish() + + future.add_callback(span.finish) + future.add_errback(close_exception) + else: + def close_future(fut): + exc = fut.exception() + if exc is not None: + Span._on_error(span, type(exc), exc, '<unavailable>') + span.finish() + future.add_done_callback(close_future) diff --git a/setup.py b/setup.py index e43515d9fc94adc33873c23d7492919b13ed0303..729eeac1c4e8084a887043a2019d2ef670c5881c 100644 --- a/setup.py +++ b/setup.py @@ -15,6 +15,7 @@ setup(keywords=['ha', 'high availability', 'scalable', 'scalability', 'server', 'YAMLSource': ['pyyaml'], 'TOMLSource': ['toml'], 'FasterJSONSource': ['ujson'], - 'cassandra': ['cassandra-driver'] + 'cassandra': ['cassandra-driver'], + 'opentracing': ['opentracing'] } )