From 2b6f9137908513c7a6b0c74a82dd3a25680084ad Mon Sep 17 00:00:00 2001 From: Tom de Bruijn Date: Sat, 4 Feb 2023 15:22:23 +0100 Subject: [PATCH 1/4] Add Redis instrumentation query sanitization (#1572) * Add Redis instrumentation query sanitization Add a query sanitizer to the Redis instrumentation. This can be disabled with the `sanitize_query = False` config option. Given the query `SET key value`, the sanitized query becomes `SET ? ?`. Both the keys and values are sanitized, as both can contain PII data. The Redis queries are sanitized by default. This changes the default behavior of this instrumentation. Previously it reported unsanitized Redis queries. This was previously discussed in the previous implementation of this PR in PR #1571 Closes #1548 * Update Redis sanitize_query option documentation Changes suggested in https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1572#discussion_r1066069584 * Remove uninstrument & instrument from test setup The Redis test that performs the tests with the default options, doesn't need to uninstrument and then instrument the instrumentor. This commit removes the unnecessary setup code. The setup code is already present at the top of the file. * Fix code style formatting * Update Redis functional tests - Update the sanitizer to also account for a max `db.statement` attribute value length. No longer than 1000 characters. - Update the functional tests to assume the queries are sanitized by default. - Add new tests that test the behavior with sanitization turned off. Only for the tests in the first test class. I don't think it's needed to duplicate this test for the clustered and async setup combinations. * Test Redis unsanitized queries by default Change the Redis functional tests so that they test the unsanitized query by default, and test the sanitized query results in the separate test functions. This is a partial revert of the previous commit 8d56c2f72e12c7d7dc4ef25a9fe6a69ea685a6d8 * Fix formatting issue in Redis utils * Disable Redis query sanitization by default Update the Redis instrumentation library to not change the default behavior for the Redis instrumentation. This can be enabled at a later time when the spec discussion about this topic has concluded. https://github.com/open-telemetry/opentelemetry-specification/issues/3104 * Fix pylint issue Remove else statement. * Update changelog about Redis query sanitization default [ci skip] Co-authored-by: Srikanth Chekuri * Fix potential error on Redis args being 0 Check the length of the args array and return an empty string if there are no args. That way it won't cause an IndexError if the args array is empty and it tries to fetch the first element, which should be the Redis command. --------- Co-authored-by: Srikanth Chekuri --- CHANGELOG.md | 1 + .../instrumentation/redis/__init__.py | 11 ++- .../instrumentation/redis/util.py | 24 +++++- .../tests/test_redis.py | 34 ++++++++ .../tests/redis/test_redis_functional.py | 80 +++++++++++++++++++ 5 files changed, 144 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ee5ece78bf..99b702799e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `opentelemetry-instrumentation-redis` Add `sanitize_query` config option to allow query sanitization. ([#1572](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1572)) - `opentelemetry-instrumentation-celery` Record exceptions as events on the span. ([#1573](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1573)) - Add metric instrumentation for urllib diff --git a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py index b85c2336b0..0f18639bd2 100644 --- a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py @@ -64,6 +64,8 @@ async def redis_get(): response_hook (Callable) - a function with extra user-defined logic to be performed after performing the request this function signature is: def response_hook(span: Span, instance: redis.connection.Connection, response) -> None +sanitize_query (Boolean) - default False, enable the Redis query sanitization + for example: .. code: python @@ -139,9 +141,11 @@ def _instrument( tracer, request_hook: _RequestHookT = None, response_hook: _ResponseHookT = None, + sanitize_query: bool = False, ): def _traced_execute_command(func, instance, args, kwargs): - query = _format_command_args(args) + query = _format_command_args(args, sanitize_query) + if len(args) > 0 and args[0]: name = args[0] else: @@ -169,7 +173,9 @@ def _traced_execute_pipeline(func, instance, args, kwargs): ) cmds = [ - _format_command_args(c.args if hasattr(c, "args") else c[0]) + _format_command_args( + c.args if hasattr(c, "args") else c[0], sanitize_query + ) for c in command_stack ] resource = "\n".join(cmds) @@ -281,6 +287,7 @@ def _instrument(self, **kwargs): tracer, request_hook=kwargs.get("request_hook"), response_hook=kwargs.get("response_hook"), + sanitize_query=kwargs.get("sanitize_query", False), ) def _uninstrument(self, **kwargs): diff --git a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py index fdc5cb5fd6..1eadaba718 100644 --- a/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py +++ b/instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py @@ -48,11 +48,27 @@ def _extract_conn_attributes(conn_kwargs): return attributes -def _format_command_args(args): - """Format command arguments and trim them as needed""" - value_max_len = 100 - value_too_long_mark = "..." +def _format_command_args(args, sanitize_query): + """Format and sanitize command arguments, and trim them as needed""" cmd_max_len = 1000 + value_too_long_mark = "..." + if sanitize_query: + # Sanitized query format: "COMMAND ? ?" + args_length = len(args) + if args_length > 0: + out = [str(args[0])] + ["?"] * (args_length - 1) + out_str = " ".join(out) + + if len(out_str) > cmd_max_len: + out_str = ( + out_str[: cmd_max_len - len(value_too_long_mark)] + + value_too_long_mark + ) + else: + out_str = "" + return out_str + + value_max_len = 100 length = 0 out = [] for arg in args: diff --git a/instrumentation/opentelemetry-instrumentation-redis/tests/test_redis.py b/instrumentation/opentelemetry-instrumentation-redis/tests/test_redis.py index 1ae1690efa..1c64fcb13e 100644 --- a/instrumentation/opentelemetry-instrumentation-redis/tests/test_redis.py +++ b/instrumentation/opentelemetry-instrumentation-redis/tests/test_redis.py @@ -148,6 +148,40 @@ def request_hook(span, conn, args, kwargs): span = spans[0] self.assertEqual(span.attributes.get(custom_attribute_name), "GET") + def test_query_sanitizer_enabled(self): + redis_client = redis.Redis() + connection = redis.connection.Connection() + redis_client.connection = connection + + RedisInstrumentor().uninstrument() + RedisInstrumentor().instrument( + tracer_provider=self.tracer_provider, + sanitize_query=True, + ) + + with mock.patch.object(redis_client, "connection"): + redis_client.set("key", "value") + + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 1) + + span = spans[0] + self.assertEqual(span.attributes.get("db.statement"), "SET ? ?") + + def test_query_sanitizer_disabled(self): + redis_client = redis.Redis() + connection = redis.connection.Connection() + redis_client.connection = connection + + with mock.patch.object(redis_client, "connection"): + redis_client.set("key", "value") + + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 1) + + span = spans[0] + self.assertEqual(span.attributes.get("db.statement"), "SET key value") + def test_no_op_tracer_provider(self): RedisInstrumentor().uninstrument() tracer_provider = trace.NoOpTracerProvider() diff --git a/tests/opentelemetry-docker-tests/tests/redis/test_redis_functional.py b/tests/opentelemetry-docker-tests/tests/redis/test_redis_functional.py index db82ec489c..675a37fa9f 100644 --- a/tests/opentelemetry-docker-tests/tests/redis/test_redis_functional.py +++ b/tests/opentelemetry-docker-tests/tests/redis/test_redis_functional.py @@ -45,6 +45,27 @@ def _check_span(self, span, name): ) self.assertEqual(span.attributes[SpanAttributes.NET_PEER_PORT], 6379) + def test_long_command_sanitized(self): + RedisInstrumentor().uninstrument() + RedisInstrumentor().instrument( + tracer_provider=self.tracer_provider, sanitize_query=True + ) + + self.redis_client.mget(*range(2000)) + + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 1) + span = spans[0] + self._check_span(span, "MGET") + self.assertTrue( + span.attributes.get(SpanAttributes.DB_STATEMENT).startswith( + "MGET ? ? ? ?" + ) + ) + self.assertTrue( + span.attributes.get(SpanAttributes.DB_STATEMENT).endswith("...") + ) + def test_long_command(self): self.redis_client.mget(*range(1000)) @@ -61,6 +82,22 @@ def test_long_command(self): span.attributes.get(SpanAttributes.DB_STATEMENT).endswith("...") ) + def test_basics_sanitized(self): + RedisInstrumentor().uninstrument() + RedisInstrumentor().instrument( + tracer_provider=self.tracer_provider, sanitize_query=True + ) + + self.assertIsNone(self.redis_client.get("cheese")) + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 1) + span = spans[0] + self._check_span(span, "GET") + self.assertEqual( + span.attributes.get(SpanAttributes.DB_STATEMENT), "GET ?" + ) + self.assertEqual(span.attributes.get("db.redis.args_length"), 2) + def test_basics(self): self.assertIsNone(self.redis_client.get("cheese")) spans = self.memory_exporter.get_finished_spans() @@ -72,6 +109,28 @@ def test_basics(self): ) self.assertEqual(span.attributes.get("db.redis.args_length"), 2) + def test_pipeline_traced_sanitized(self): + RedisInstrumentor().uninstrument() + RedisInstrumentor().instrument( + tracer_provider=self.tracer_provider, sanitize_query=True + ) + + with self.redis_client.pipeline(transaction=False) as pipeline: + pipeline.set("blah", 32) + pipeline.rpush("foo", "éé") + pipeline.hgetall("xxx") + pipeline.execute() + + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 1) + span = spans[0] + self._check_span(span, "SET RPUSH HGETALL") + self.assertEqual( + span.attributes.get(SpanAttributes.DB_STATEMENT), + "SET ? ?\nRPUSH ? ?\nHGETALL ?", + ) + self.assertEqual(span.attributes.get("db.redis.pipeline_length"), 3) + def test_pipeline_traced(self): with self.redis_client.pipeline(transaction=False) as pipeline: pipeline.set("blah", 32) @@ -89,6 +148,27 @@ def test_pipeline_traced(self): ) self.assertEqual(span.attributes.get("db.redis.pipeline_length"), 3) + def test_pipeline_immediate_sanitized(self): + RedisInstrumentor().uninstrument() + RedisInstrumentor().instrument( + tracer_provider=self.tracer_provider, sanitize_query=True + ) + + with self.redis_client.pipeline() as pipeline: + pipeline.set("a", 1) + pipeline.immediate_execute_command("SET", "b", 2) + pipeline.execute() + + spans = self.memory_exporter.get_finished_spans() + # expecting two separate spans here, rather than a + # single span for the whole pipeline + self.assertEqual(len(spans), 2) + span = spans[0] + self._check_span(span, "SET") + self.assertEqual( + span.attributes.get(SpanAttributes.DB_STATEMENT), "SET ? ?" + ) + def test_pipeline_immediate(self): with self.redis_client.pipeline() as pipeline: pipeline.set("a", 1) From a217682a394a5c51959fc89883c49b0d88d32440 Mon Sep 17 00:00:00 2001 From: avzis <107620508+avzis@users.noreply.github.com> Date: Sat, 4 Feb 2023 21:41:30 +0200 Subject: [PATCH 2/4] Add a test for aws lambda using NoOpTracerProvider (#1602) * add a test for aws lambda using NoOpTracerProvider * fix import * add assert for spans --------- Co-authored-by: Srikanth Chekuri --- .../tests/test_aws_lambda_instrumentation_manual.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-aws-lambda/tests/test_aws_lambda_instrumentation_manual.py b/instrumentation/opentelemetry-instrumentation-aws-lambda/tests/test_aws_lambda_instrumentation_manual.py index f22d50489a..fd3c9f88ca 100644 --- a/instrumentation/opentelemetry-instrumentation-aws-lambda/tests/test_aws_lambda_instrumentation_manual.py +++ b/instrumentation/opentelemetry-instrumentation-aws-lambda/tests/test_aws_lambda_instrumentation_manual.py @@ -38,7 +38,7 @@ from opentelemetry.semconv.resource import ResourceAttributes from opentelemetry.semconv.trace import SpanAttributes from opentelemetry.test.test_base import TestBase -from opentelemetry.trace import SpanKind +from opentelemetry.trace import NoOpTracerProvider, SpanKind from opentelemetry.trace.propagation.tracecontext import ( TraceContextTextMapPropagator, ) @@ -413,3 +413,12 @@ def test_uninstrument(self): mock_execute_lambda(MOCK_LAMBDA_API_GATEWAY_HTTP_API_EVENT) spans = self.memory_exporter.get_finished_spans() self.assertEqual(len(spans), 0) + + def test_no_op_tracer_provider(self): + tracer_provider = NoOpTracerProvider() + AwsLambdaInstrumentor().instrument(tracer_provider=tracer_provider) + + mock_execute_lambda(MOCK_LAMBDA_API_GATEWAY_HTTP_API_EVENT) + spans = self.memory_exporter.get_finished_spans() + assert spans is not None + self.assertEqual(len(spans), 0) From 2519223a5ce884ba7f52f6fe584bfc59946144c6 Mon Sep 17 00:00:00 2001 From: Akochavi <121871419+Akochavi@users.noreply.github.com> Date: Sat, 4 Feb 2023 22:55:41 +0200 Subject: [PATCH 3/4] Audit and test opentelemetry-instrumentation-aiopg NoOpTracerProvider (#1606) * Audit and test opentelemetry-instrumentation-aiopg NoOpTracerProvider * Change line length --------- Co-authored-by: Shalev Roda <65566801+shalevr@users.noreply.github.com> Co-authored-by: Srikanth Chekuri --- .../tests/test_aiopg_integration.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/instrumentation/opentelemetry-instrumentation-aiopg/tests/test_aiopg_integration.py b/instrumentation/opentelemetry-instrumentation-aiopg/tests/test_aiopg_integration.py index b5a248f488..c6d00a51fe 100644 --- a/instrumentation/opentelemetry-instrumentation-aiopg/tests/test_aiopg_integration.py +++ b/instrumentation/opentelemetry-instrumentation-aiopg/tests/test_aiopg_integration.py @@ -222,6 +222,19 @@ def test_instrument_connection_after_instrument(self): spans_list = self.memory_exporter.get_finished_spans() self.assertEqual(len(spans_list), 1) + def test_no_op_tracer_provider(self): + cnx = async_call(aiopg.connect(database="test")) + AiopgInstrumentor().instrument_connection( + cnx, tracer_provider=trace_api.NoOpTracerProvider() + ) + + cursor = async_call(cnx.cursor()) + query = "SELECT * FROM test" + async_call(cursor.execute(query)) + + spans_list = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans_list), 0) + def test_custom_tracer_provider_instrument_connection(self): resource = resources.Resource.create( {"service.name": "db-test-service"} From 66ceef5fe11ac91270902f2cfda69e698a311822 Mon Sep 17 00:00:00 2001 From: Phillip Verheyden Date: Sun, 5 Feb 2023 00:24:21 -0600 Subject: [PATCH 4/4] Support aio_pika 8.x (#1481) * Support aio_pika 8 - Fix tests for new shape of the AbstractConnection class - Run tests against aio_pika 7 and 8 * Update CHANGELOG.md --------- Co-authored-by: Srikanth Chekuri --- CHANGELOG.md | 5 +- instrumentation/README.md | 2 +- .../pyproject.toml | 2 +- .../instrumentation/aio_pika/package.py | 2 +- .../instrumentation/aio_pika/span_builder.py | 36 +++++++--- .../tests/consts.py | 6 +- .../tests/test_callback_decorator.py | 54 ++++++++++++-- .../tests/test_publish_decorator.py | 70 +++++++++++++++++-- .../tests/test_span_builder.py | 2 +- .../instrumentation/bootstrap_gen.py | 2 +- tox.ini | 9 +++ 11 files changed, 158 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 99b702799e..57b92e590c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,9 +16,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ([#1553](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1553)) - `opentelemetry/sdk/extension/aws` Implement [`aws.ecs.*`](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/resource/semantic_conventions/cloud_provider/aws/ecs.md) and [`aws.logs.*`](https://opentelemetry.io/docs/reference/specification/resource/semantic_conventions/cloud_provider/aws/logs/) resource attributes in the `AwsEcsResourceDetector` detector when the ECS Metadata v4 is available ([#1212](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1212)) +- `opentelemetry-instrumentation-aio-pika` Support `aio_pika` 8.x + ([#1481](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1481)) - `opentelemetry-instrumentation-aws-lambda` Flush `MeterProvider` at end of function invocation. ([#1613](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1613)) -- Fix aiohttp bug with unset `trace_configs` ([#1592](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1592)) +- Fix aiohttp bug with unset `trace_configs` + ([#1592](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1592)) - `opentelemetry-instrumentation-django` Allow explicit `excluded_urls` configuration through `instrument()` ([#1618](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1618)) diff --git a/instrumentation/README.md b/instrumentation/README.md index a269b09397..b1482a0227 100644 --- a/instrumentation/README.md +++ b/instrumentation/README.md @@ -1,7 +1,7 @@ | Instrumentation | Supported Packages | Metrics support | | --------------- | ------------------ | --------------- | -| [opentelemetry-instrumentation-aio-pika](./opentelemetry-instrumentation-aio-pika) | aio_pika ~= 7.2.0 | No +| [opentelemetry-instrumentation-aio-pika](./opentelemetry-instrumentation-aio-pika) | aio_pika >= 7.2.0, < 9.0.0 | No | [opentelemetry-instrumentation-aiohttp-client](./opentelemetry-instrumentation-aiohttp-client) | aiohttp ~= 3.0 | No | [opentelemetry-instrumentation-aiopg](./opentelemetry-instrumentation-aiopg) | aiopg >= 0.13.0, < 2.0.0 | No | [opentelemetry-instrumentation-asgi](./opentelemetry-instrumentation-asgi) | asgiref ~= 3.0 | No diff --git a/instrumentation/opentelemetry-instrumentation-aio-pika/pyproject.toml b/instrumentation/opentelemetry-instrumentation-aio-pika/pyproject.toml index 4511254a70..994642e22a 100644 --- a/instrumentation/opentelemetry-instrumentation-aio-pika/pyproject.toml +++ b/instrumentation/opentelemetry-instrumentation-aio-pika/pyproject.toml @@ -31,7 +31,7 @@ dependencies = [ [project.optional-dependencies] instruments = [ - "aio_pika ~= 7.2.0", + "aio_pika >= 7.2.0, < 9.0.0", ] test = [ "opentelemetry-instrumentation-aio-pika[instruments]", diff --git a/instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/package.py b/instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/package.py index 6c7ed74ea4..285e9f99cb 100644 --- a/instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/package.py +++ b/instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/package.py @@ -13,4 +13,4 @@ # limitations under the License. from typing import Collection -_instruments: Collection[str] = ("aio_pika ~= 7.2.0",) +_instruments: Collection[str] = ("aio_pika >= 7.2.0, < 9.0.0",) diff --git a/instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/span_builder.py b/instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/span_builder.py index a61209e0ce..056f3dab25 100644 --- a/instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/span_builder.py +++ b/instrumentation/opentelemetry-instrumentation-aio-pika/src/opentelemetry/instrumentation/aio_pika/span_builder.py @@ -24,7 +24,7 @@ ) from opentelemetry.trace import Span, SpanKind, Tracer -_DEFAULT_ATTRIBUTES = {SpanAttributes.MESSAGING_SYSTEM: 'rabbitmq'} +_DEFAULT_ATTRIBUTES = {SpanAttributes.MESSAGING_SYSTEM: "rabbitmq"} class SpanBuilder: @@ -49,18 +49,30 @@ def set_destination(self, destination: str): self._attributes[SpanAttributes.MESSAGING_DESTINATION] = destination def set_channel(self, channel: AbstractChannel): - url = channel.connection.connection.url - self._attributes.update({ - SpanAttributes.NET_PEER_NAME: url.host, - SpanAttributes.NET_PEER_PORT: url.port - }) + connection = channel.connection + if getattr(connection, "connection", None): + # aio_rmq 7 + url = connection.connection.url + else: + # aio_rmq 8 + url = connection.url + self._attributes.update( + { + SpanAttributes.NET_PEER_NAME: url.host, + SpanAttributes.NET_PEER_PORT: url.port, + } + ) def set_message(self, message: AbstractMessage): properties = message.properties if properties.message_id: - self._attributes[SpanAttributes.MESSAGING_MESSAGE_ID] = properties.message_id + self._attributes[ + SpanAttributes.MESSAGING_MESSAGE_ID + ] = properties.message_id if properties.correlation_id: - self._attributes[SpanAttributes.MESSAGING_CONVERSATION_ID] = properties.correlation_id + self._attributes[ + SpanAttributes.MESSAGING_CONVERSATION_ID + ] = properties.correlation_id def build(self) -> Optional[Span]: if not is_instrumentation_enabled(): @@ -69,9 +81,11 @@ def build(self) -> Optional[Span]: self._attributes[SpanAttributes.MESSAGING_OPERATION] = self._operation.value else: self._attributes[SpanAttributes.MESSAGING_TEMP_DESTINATION] = True - span = self._tracer.start_span(self._generate_span_name(), kind=self._kind, attributes=self._attributes) + span = self._tracer.start_span( + self._generate_span_name(), kind=self._kind, attributes=self._attributes + ) return span def _generate_span_name(self) -> str: - operation_value = self._operation.value if self._operation else 'send' - return f'{self._destination} {operation_value}' + operation_value = self._operation.value if self._operation else "send" + return f"{self._destination} {operation_value}" diff --git a/instrumentation/opentelemetry-instrumentation-aio-pika/tests/consts.py b/instrumentation/opentelemetry-instrumentation-aio-pika/tests/consts.py index ada7080192..7e34c73c63 100644 --- a/instrumentation/opentelemetry-instrumentation-aio-pika/tests/consts.py +++ b/instrumentation/opentelemetry-instrumentation-aio-pika/tests/consts.py @@ -15,8 +15,10 @@ SERVER_URL = URL( f"amqp://{SERVER_USER}:{SERVER_PASS}@{SERVER_HOST}:{SERVER_PORT}/" ) -CONNECTION = Namespace(connection=Namespace(url=SERVER_URL)) -CHANNEL = Namespace(connection=CONNECTION, loop=None) +CONNECTION_7 = Namespace(connection=Namespace(url=SERVER_URL)) +CONNECTION_8 = Namespace(url=SERVER_URL) +CHANNEL_7 = Namespace(connection=CONNECTION_7, loop=None) +CHANNEL_8 = Namespace(connection=CONNECTION_8, loop=None) MESSAGE = Namespace( properties=Namespace( message_id=MESSAGE_ID, correlation_id=CORRELATION_ID, headers={} diff --git a/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_callback_decorator.py b/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_callback_decorator.py index 70883c116c..7c9288a657 100644 --- a/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_callback_decorator.py +++ b/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_callback_decorator.py @@ -12,9 +12,9 @@ # See the License for the specific language governing permissions and # limitations under the License. import asyncio -from unittest import TestCase, mock +from unittest import TestCase, mock, skipIf -from aio_pika import Queue +from aio_pika import Queue, version_info from opentelemetry.instrumentation.aio_pika.callback_decorator import ( CallbackDecorator, @@ -23,7 +23,8 @@ from opentelemetry.trace import SpanKind, get_tracer from .consts import ( - CHANNEL, + CHANNEL_7, + CHANNEL_8, CORRELATION_ID, EXCHANGE_NAME, MESSAGE, @@ -35,7 +36,8 @@ ) -class TestInstrumentedQueue(TestCase): +@skipIf(version_info >= (8, 0), "Only for aio_pika 7") +class TestInstrumentedQueueAioRmq7(TestCase): EXPECTED_ATTRIBUTES = { SpanAttributes.MESSAGING_SYSTEM: MESSAGING_SYSTEM, SpanAttributes.MESSAGING_DESTINATION: EXCHANGE_NAME, @@ -52,7 +54,7 @@ def setUp(self): asyncio.set_event_loop(self.loop) def test_get_callback_span(self): - queue = Queue(CHANNEL, QUEUE_NAME, False, False, False, None) + queue = Queue(CHANNEL_7, QUEUE_NAME, False, False, False, None) tracer = mock.MagicMock() CallbackDecorator(tracer, queue)._get_span(MESSAGE) tracer.start_span.assert_called_once_with( @@ -62,7 +64,47 @@ def test_get_callback_span(self): ) def test_decorate_callback(self): - queue = Queue(CHANNEL, QUEUE_NAME, False, False, False, None) + queue = Queue(CHANNEL_7, QUEUE_NAME, False, False, False, None) + callback = mock.MagicMock(return_value=asyncio.sleep(0)) + with mock.patch.object( + CallbackDecorator, "_get_span" + ) as mocked_get_callback_span: + callback_decorator = CallbackDecorator(self.tracer, queue) + decorated_callback = callback_decorator.decorate(callback) + self.loop.run_until_complete(decorated_callback(MESSAGE)) + mocked_get_callback_span.assert_called_once() + callback.assert_called_once_with(MESSAGE) + + +@skipIf(version_info <= (8, 0), "Only for aio_pika 8") +class TestInstrumentedQueueAioRmq8(TestCase): + EXPECTED_ATTRIBUTES = { + SpanAttributes.MESSAGING_SYSTEM: MESSAGING_SYSTEM, + SpanAttributes.MESSAGING_DESTINATION: EXCHANGE_NAME, + SpanAttributes.NET_PEER_NAME: SERVER_HOST, + SpanAttributes.NET_PEER_PORT: SERVER_PORT, + SpanAttributes.MESSAGING_MESSAGE_ID: MESSAGE_ID, + SpanAttributes.MESSAGING_CONVERSATION_ID: CORRELATION_ID, + SpanAttributes.MESSAGING_OPERATION: "receive", + } + + def setUp(self): + self.tracer = get_tracer(__name__) + self.loop = asyncio.new_event_loop() + asyncio.set_event_loop(self.loop) + + def test_get_callback_span(self): + queue = Queue(CHANNEL_8, QUEUE_NAME, False, False, False, None) + tracer = mock.MagicMock() + CallbackDecorator(tracer, queue)._get_span(MESSAGE) + tracer.start_span.assert_called_once_with( + f"{EXCHANGE_NAME} receive", + kind=SpanKind.CONSUMER, + attributes=self.EXPECTED_ATTRIBUTES, + ) + + def test_decorate_callback(self): + queue = Queue(CHANNEL_8, QUEUE_NAME, False, False, False, None) callback = mock.MagicMock(return_value=asyncio.sleep(0)) with mock.patch.object( CallbackDecorator, "_get_span" diff --git a/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_publish_decorator.py b/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_publish_decorator.py index 80dfa3182b..e5586b9a00 100644 --- a/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_publish_decorator.py +++ b/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_publish_decorator.py @@ -13,9 +13,9 @@ # limitations under the License. import asyncio from typing import Type -from unittest import TestCase, mock +from unittest import TestCase, mock, skipIf -from aio_pika import Exchange, RobustExchange +from aio_pika import Exchange, RobustExchange, version_info from opentelemetry.instrumentation.aio_pika.publish_decorator import ( PublishDecorator, @@ -24,8 +24,10 @@ from opentelemetry.trace import SpanKind, get_tracer from .consts import ( - CHANNEL, - CONNECTION, + CHANNEL_7, + CHANNEL_8, + CONNECTION_7, + CONNECTION_8, CORRELATION_ID, EXCHANGE_NAME, MESSAGE, @@ -37,7 +39,8 @@ ) -class TestInstrumentedExchange(TestCase): +@skipIf(version_info >= (8, 0), "Only for aio_pika 7") +class TestInstrumentedExchangeAioRmq7(TestCase): EXPECTED_ATTRIBUTES = { SpanAttributes.MESSAGING_SYSTEM: MESSAGING_SYSTEM, SpanAttributes.MESSAGING_DESTINATION: f"{EXCHANGE_NAME},{ROUTING_KEY}", @@ -54,7 +57,7 @@ def setUp(self): asyncio.set_event_loop(self.loop) def test_get_publish_span(self): - exchange = Exchange(CONNECTION, CHANNEL, EXCHANGE_NAME) + exchange = Exchange(CONNECTION_7, CHANNEL_7, EXCHANGE_NAME) tracer = mock.MagicMock() PublishDecorator(tracer, exchange)._get_publish_span( MESSAGE, ROUTING_KEY @@ -66,7 +69,60 @@ def test_get_publish_span(self): ) def _test_publish(self, exchange_type: Type[Exchange]): - exchange = exchange_type(CONNECTION, CHANNEL, EXCHANGE_NAME) + exchange = exchange_type(CONNECTION_7, CHANNEL_7, EXCHANGE_NAME) + with mock.patch.object( + PublishDecorator, "_get_publish_span" + ) as mock_get_publish_span: + with mock.patch.object( + Exchange, "publish", return_value=asyncio.sleep(0) + ) as mock_publish: + decorated_publish = PublishDecorator( + self.tracer, exchange + ).decorate(mock_publish) + self.loop.run_until_complete( + decorated_publish(MESSAGE, ROUTING_KEY) + ) + mock_publish.assert_called_once() + mock_get_publish_span.assert_called_once() + + def test_publish(self): + self._test_publish(Exchange) + + def test_robust_publish(self): + self._test_publish(RobustExchange) + + +@skipIf(version_info <= (8, 0), "Only for aio_pika 8") +class TestInstrumentedExchangeAioRmq8(TestCase): + EXPECTED_ATTRIBUTES = { + SpanAttributes.MESSAGING_SYSTEM: MESSAGING_SYSTEM, + SpanAttributes.MESSAGING_DESTINATION: f"{EXCHANGE_NAME},{ROUTING_KEY}", + SpanAttributes.NET_PEER_NAME: SERVER_HOST, + SpanAttributes.NET_PEER_PORT: SERVER_PORT, + SpanAttributes.MESSAGING_MESSAGE_ID: MESSAGE_ID, + SpanAttributes.MESSAGING_CONVERSATION_ID: CORRELATION_ID, + SpanAttributes.MESSAGING_TEMP_DESTINATION: True, + } + + def setUp(self): + self.tracer = get_tracer(__name__) + self.loop = asyncio.new_event_loop() + asyncio.set_event_loop(self.loop) + + def test_get_publish_span(self): + exchange = Exchange(CHANNEL_8, EXCHANGE_NAME) + tracer = mock.MagicMock() + PublishDecorator(tracer, exchange)._get_publish_span( + MESSAGE, ROUTING_KEY + ) + tracer.start_span.assert_called_once_with( + f"{EXCHANGE_NAME},{ROUTING_KEY} send", + kind=SpanKind.PRODUCER, + attributes=self.EXPECTED_ATTRIBUTES, + ) + + def _test_publish(self, exchange_type: Type[Exchange]): + exchange = exchange_type(CONNECTION_8, CHANNEL_8, EXCHANGE_NAME) with mock.patch.object( PublishDecorator, "_get_publish_span" ) as mock_get_publish_span: diff --git a/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_span_builder.py b/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_span_builder.py index 5f87d53846..a4a1d8ec8b 100644 --- a/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_span_builder.py +++ b/instrumentation/opentelemetry-instrumentation-aio-pika/tests/test_span_builder.py @@ -21,6 +21,6 @@ class TestBuilder(TestCase): def test_build(self): builder = SpanBuilder(get_tracer(__name__)) builder.set_as_consumer() - builder.set_destination('destination') + builder.set_destination("destination") span = builder.build() self.assertTrue(isinstance(span, Span)) diff --git a/opentelemetry-instrumentation/src/opentelemetry/instrumentation/bootstrap_gen.py b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/bootstrap_gen.py index 36fda70ab1..20c5a0b725 100644 --- a/opentelemetry-instrumentation/src/opentelemetry/instrumentation/bootstrap_gen.py +++ b/opentelemetry-instrumentation/src/opentelemetry/instrumentation/bootstrap_gen.py @@ -17,7 +17,7 @@ libraries = { "aio_pika": { - "library": "aio_pika ~= 7.2.0", + "library": "aio_pika >= 7.2.0, < 9.0.0", "instrumentation": "opentelemetry-instrumentation-aio-pika==0.37b0.dev", }, "aiohttp": { diff --git a/tox.ini b/tox.ini index 162656a4d0..e51515f906 100644 --- a/tox.ini +++ b/tox.ini @@ -206,6 +206,10 @@ envlist = py3{7,8,9,10,11}-test-instrumentation-pika{0,1} pypy3-test-instrumentation-pika{0,1} + ; opentelemetry-instrumentation-aio-pika + py3{7,8,9,10,11}-test-instrumentation-aio-pika{7,8} + pypy3-test-instrumentation-aio-pika{7,8} + ; opentelemetry-instrumentation-kafka-python py3{7,8,9,10,11}-test-instrumentation-kafka-python pypy3-test-instrumentation-kafka-python @@ -250,6 +254,8 @@ deps = sqlalchemy14: sqlalchemy~=1.4 pika0: pika>=0.12.0,<1.0.0 pika1: pika>=1.0.0 + aio-pika7: aio_pika~=7.2.0 + aio-pika8: aio_pika>=8.0.0,<9.0.0 pymemcache135: pymemcache ==1.3.5 pymemcache200: pymemcache >2.0.0,<3.0.0 pymemcache300: pymemcache >3.0.0,<3.4.2 @@ -296,6 +302,7 @@ changedir = test-instrumentation-logging: instrumentation/opentelemetry-instrumentation-logging/tests test-instrumentation-mysql: instrumentation/opentelemetry-instrumentation-mysql/tests test-instrumentation-pika{0,1}: instrumentation/opentelemetry-instrumentation-pika/tests + test-instrumentation-aio-pika{7,8}: instrumentation/opentelemetry-instrumentation-aio-pika/tests test-instrumentation-psycopg2: instrumentation/opentelemetry-instrumentation-psycopg2/tests test-instrumentation-pymemcache{135,200,300,342}: instrumentation/opentelemetry-instrumentation-pymemcache/tests test-instrumentation-pymongo: instrumentation/opentelemetry-instrumentation-pymongo/tests @@ -337,6 +344,8 @@ commands_pre = pika{0,1}: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-pika[test] + aio-pika{7,8}: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-aio-pika[test] + kafka-python: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-kafka-python[test] confluent-kafka: pip install {toxinidir}/instrumentation/opentelemetry-instrumentation-confluent-kafka[test]