diff --git a/.changelog/4739.added b/.changelog/4739.added new file mode 100644 index 0000000000..c3e7255a45 --- /dev/null +++ b/.changelog/4739.added @@ -0,0 +1 @@ +`opentelemetry-instrumentation-pymemcache`: add database semconv stability migration support diff --git a/instrumentation/README.md b/instrumentation/README.md index 823b1929b2..6466bde4dc 100644 --- a/instrumentation/README.md +++ b/instrumentation/README.md @@ -33,7 +33,7 @@ | [opentelemetry-instrumentation-pika](./opentelemetry-instrumentation-pika) | pika >= 0.12.0 | No | development | [opentelemetry-instrumentation-psycopg](./opentelemetry-instrumentation-psycopg) | psycopg >= 3.1.0 | No | migration | [opentelemetry-instrumentation-psycopg2](./opentelemetry-instrumentation-psycopg2) | psycopg2 >= 2.7.3.1,psycopg2-binary >= 2.7.3.1 | No | migration -| [opentelemetry-instrumentation-pymemcache](./opentelemetry-instrumentation-pymemcache) | pymemcache >= 1.3.5, < 5 | No | development +| [opentelemetry-instrumentation-pymemcache](./opentelemetry-instrumentation-pymemcache) | pymemcache >= 1.3.5, < 5 | No | migration | [opentelemetry-instrumentation-pymongo](./opentelemetry-instrumentation-pymongo) | pymongo >= 3.1, < 5.0 | No | development | [opentelemetry-instrumentation-pymssql](./opentelemetry-instrumentation-pymssql) | pymssql >= 2.1.5, < 3 | No | migration | [opentelemetry-instrumentation-pymysql](./opentelemetry-instrumentation-pymysql) | PyMySQL < 2 | No | migration diff --git a/instrumentation/opentelemetry-instrumentation-pymemcache/src/opentelemetry/instrumentation/pymemcache/__init__.py b/instrumentation/opentelemetry-instrumentation-pymemcache/src/opentelemetry/instrumentation/pymemcache/__init__.py index e748475dc0..c433c88e73 100644 --- a/instrumentation/opentelemetry-instrumentation-pymemcache/src/opentelemetry/instrumentation/pymemcache/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-pymemcache/src/opentelemetry/instrumentation/pymemcache/__init__.py @@ -32,20 +32,26 @@ import pymemcache from wrapt import wrap_function_wrapper as _wrap +from opentelemetry.instrumentation._semconv import ( + _get_schema_url_for_signal_types, + _OpenTelemetrySemanticConventionStability, + _OpenTelemetryStabilitySignalType, + _set_db_statement, + _set_db_system, + _set_http_net_peer_name_client, + _set_http_peer_port_client, + _set_net_transport, +) from opentelemetry.instrumentation.instrumentor import BaseInstrumentor from opentelemetry.instrumentation.pymemcache.package import _instruments from opentelemetry.instrumentation.pymemcache.version import __version__ from opentelemetry.instrumentation.utils import unwrap -from opentelemetry.semconv._incubating.attributes.db_attributes import ( - DB_STATEMENT, - DB_SYSTEM, -) from opentelemetry.semconv._incubating.attributes.net_attributes import ( - NET_PEER_NAME, - NET_PEER_PORT, - NET_TRANSPORT, NetTransportValues, ) +from opentelemetry.semconv.attributes.network_attributes import ( + NetworkTransportValues, +) from opentelemetry.trace import SpanKind, get_tracer logger = logging.getLogger(__name__) @@ -77,23 +83,24 @@ ] -def _set_connection_attributes(span, instance): - if not span.is_recording(): - return - for key, value in _get_address_attributes(instance).items(): - span.set_attribute(key, value) - - def _with_tracer_wrapper(func): """Helper for providing tracer for wrapper functions.""" - def _with_tracer(tracer, cmd): + def _with_tracer(tracer, cmd, sem_conv_opt_in_mode): def wrapper(wrapped, instance, args, kwargs): # prevent double wrapping if hasattr(wrapped, "__wrapped__"): return wrapped(*args, **kwargs) - return func(tracer, cmd, wrapped, instance, args, kwargs) + return func( + tracer, + cmd, + sem_conv_opt_in_mode, + wrapped, + instance, + args, + kwargs, + ) return wrapper @@ -101,7 +108,9 @@ def wrapper(wrapped, instance, args, kwargs): @_with_tracer_wrapper -def _wrap_cmd(tracer, cmd, wrapped, instance, args, kwargs): +def _wrap_cmd( + tracer, cmd, sem_conv_opt_in_mode, wrapped, instance, args, kwargs +): with tracer.start_as_current_span( cmd, kind=SpanKind.CLIENT, attributes={} ) as span: @@ -113,9 +122,10 @@ def _wrap_cmd(tracer, cmd, wrapped, instance, args, kwargs): vals = _get_query_string(args[0]) query = f"{cmd}{' ' if vals else ''}{vals}" - span.set_attribute(DB_STATEMENT, query) - _set_connection_attributes(span, instance) + attrs = _get_address_attributes(instance, sem_conv_opt_in_mode) + _set_db_statement(attrs, query, sem_conv_opt_in_mode) + span.set_attributes(attrs) except Exception as ex: # pylint: disable=broad-except logger.warning( "Failed to set attributes for pymemcache span %s", str(ex) @@ -148,22 +158,37 @@ def _get_query_string(arg): return keys -def _get_address_attributes(instance): +def _get_address_attributes(instance, sem_conv_opt_in_mode): """Attempt to get host and port from Client instance.""" address_attributes = {} - address_attributes[DB_SYSTEM] = "memcached" - + _set_db_system(address_attributes, "memcached", sem_conv_opt_in_mode) # client.base.Client contains server attribute which is either a host/port tuple, or unix socket path string # https://github.com/pinterest/pymemcache/blob/f02ddf73a28c09256589b8afbb3ee50f1171cac7/pymemcache/client/base.py#L228 if hasattr(instance, "server"): if isinstance(instance.server, tuple): host, port = instance.server - address_attributes[NET_PEER_NAME] = host - address_attributes[NET_PEER_PORT] = port - address_attributes[NET_TRANSPORT] = NetTransportValues.IP_TCP.value + _set_http_net_peer_name_client( + address_attributes, host, sem_conv_opt_in_mode + ) + _set_http_peer_port_client( + address_attributes, port, sem_conv_opt_in_mode + ) + _set_net_transport( + address_attributes, + NetTransportValues.IP_TCP.value, + NetworkTransportValues.TCP.value, + sem_conv_opt_in_mode, + ) elif isinstance(instance.server, str): - address_attributes[NET_PEER_NAME] = instance.server - address_attributes[NET_TRANSPORT] = NetTransportValues.OTHER.value + _set_http_net_peer_name_client( + address_attributes, instance.server, sem_conv_opt_in_mode + ) + _set_net_transport( + address_attributes, + NetTransportValues.OTHER.value, + NetworkTransportValues.PIPE.value, + sem_conv_opt_in_mode, + ) return address_attributes @@ -175,19 +200,26 @@ def instrumentation_dependencies(self) -> Collection[str]: return _instruments def _instrument(self, **kwargs): + _OpenTelemetrySemanticConventionStability._initialize() + sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode( + _OpenTelemetryStabilitySignalType.DATABASE, + ) + tracer_provider = kwargs.get("tracer_provider") tracer = get_tracer( __name__, __version__, tracer_provider, - schema_url="https://opentelemetry.io/schemas/1.11.0", + schema_url=_get_schema_url_for_signal_types( + [_OpenTelemetryStabilitySignalType.DATABASE] + ), ) for cmd in COMMANDS: _wrap( "pymemcache.client.base", f"Client.{cmd}", - _wrap_cmd(tracer, cmd), + _wrap_cmd(tracer, cmd, sem_conv_opt_in_mode), ) def _uninstrument(self, **kwargs): diff --git a/instrumentation/opentelemetry-instrumentation-pymemcache/src/opentelemetry/instrumentation/pymemcache/package.py b/instrumentation/opentelemetry-instrumentation-pymemcache/src/opentelemetry/instrumentation/pymemcache/package.py index f931558a26..78eb2b32ac 100644 --- a/instrumentation/opentelemetry-instrumentation-pymemcache/src/opentelemetry/instrumentation/pymemcache/package.py +++ b/instrumentation/opentelemetry-instrumentation-pymemcache/src/opentelemetry/instrumentation/pymemcache/package.py @@ -3,3 +3,4 @@ _instruments = ("pymemcache >= 1.3.5, < 5",) +_semconv_status = "migration" diff --git a/instrumentation/opentelemetry-instrumentation-pymemcache/tests/test_pymemcache.py b/instrumentation/opentelemetry-instrumentation-pymemcache/tests/test_pymemcache.py index 92fd117d6b..5f0feda89f 100644 --- a/instrumentation/opentelemetry-instrumentation-pymemcache/tests/test_pymemcache.py +++ b/instrumentation/opentelemetry-instrumentation-pymemcache/tests/test_pymemcache.py @@ -15,6 +15,9 @@ ) from opentelemetry import trace as trace_api +from opentelemetry.instrumentation._semconv import ( + _OpenTelemetrySemanticConventionStability, +) from opentelemetry.instrumentation.pymemcache import PymemcacheInstrumentor from opentelemetry.semconv._incubating.attributes.db_attributes import ( DB_STATEMENT, @@ -24,6 +27,14 @@ NET_PEER_NAME, NET_PEER_PORT, ) +from opentelemetry.semconv.attributes.db_attributes import ( + DB_QUERY_TEXT, + DB_SYSTEM_NAME, +) +from opentelemetry.semconv.attributes.server_attributes import ( + SERVER_ADDRESS, + SERVER_PORT, +) from opentelemetry.test.test_base import TestBase from opentelemetry.trace import get_tracer @@ -475,6 +486,85 @@ def test_stats(self): self.check_spans(spans, 1, ["stats"]) + def test_new_semconv(self): + PymemcacheInstrumentor().uninstrument() + with mock.patch.dict( + "os.environ", + {"OTEL_SEMCONV_STABILITY_OPT_IN": "database"}, + ): + _OpenTelemetrySemanticConventionStability._initialized = False + PymemcacheInstrumentor().instrument() + client = self.make_client([b"STORED\r\n"]) + client.set(b"key", b"value", noreply=False) + + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 1) + span = spans[0] + + self.assertEqual(span.attributes[DB_SYSTEM_NAME], "memcached") + self.assertEqual(span.attributes[DB_QUERY_TEXT], "set key") + self.assertEqual(span.attributes[SERVER_ADDRESS], TEST_HOST) + self.assertEqual(span.attributes[SERVER_PORT], TEST_PORT) + self.assertNotIn(DB_SYSTEM, span.attributes) + self.assertNotIn(DB_STATEMENT, span.attributes) + self.assertNotIn(NET_PEER_NAME, span.attributes) + self.assertNotIn(NET_PEER_PORT, span.attributes) + self.assertEqual( + span.instrumentation_scope.schema_url, + "https://opentelemetry.io/schemas/1.25.0", + ) + _OpenTelemetrySemanticConventionStability._initialized = False + + def test_dup_semconv(self): + PymemcacheInstrumentor().uninstrument() + with mock.patch.dict( + "os.environ", + {"OTEL_SEMCONV_STABILITY_OPT_IN": "database/dup"}, + ): + _OpenTelemetrySemanticConventionStability._initialized = False + PymemcacheInstrumentor().instrument() + client = self.make_client([b"STORED\r\n"]) + client.set(b"key", b"value", noreply=False) + + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 1) + span = spans[0] + + self.assertEqual(span.attributes[DB_SYSTEM_NAME], "memcached") + self.assertEqual(span.attributes[DB_QUERY_TEXT], "set key") + self.assertEqual(span.attributes[SERVER_ADDRESS], TEST_HOST) + self.assertEqual(span.attributes[SERVER_PORT], TEST_PORT) + self.assertEqual(span.attributes[DB_SYSTEM], "memcached") + self.assertEqual(span.attributes[DB_STATEMENT], "set key") + self.assertEqual(span.attributes[NET_PEER_NAME], TEST_HOST) + self.assertEqual(span.attributes[NET_PEER_PORT], TEST_PORT) + self.assertEqual( + span.instrumentation_scope.schema_url, + "https://opentelemetry.io/schemas/1.25.0", + ) + _OpenTelemetrySemanticConventionStability._initialized = False + + def test_default_semconv(self): + client = self.make_client([b"STORED\r\n"]) + client.set(b"key", b"value", noreply=False) + + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(spans), 1) + span = spans[0] + + self.assertEqual(span.attributes[DB_SYSTEM], "memcached") + self.assertEqual(span.attributes[DB_STATEMENT], "set key") + self.assertEqual(span.attributes[NET_PEER_NAME], TEST_HOST) + self.assertEqual(span.attributes[NET_PEER_PORT], TEST_PORT) + self.assertNotIn(DB_SYSTEM_NAME, span.attributes) + self.assertNotIn(DB_QUERY_TEXT, span.attributes) + self.assertNotIn(SERVER_ADDRESS, span.attributes) + self.assertNotIn(SERVER_PORT, span.attributes) + self.assertEqual( + span.instrumentation_scope.schema_url, + "https://opentelemetry.io/schemas/1.11.0", + ) + def test_uninstrumented(self): PymemcacheInstrumentor().uninstrument()