opentelemetry-instrumentation-pymemcache: add semconv stability migration support

.changelog/4739.added

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-pymemcache`: add database semconv stability migration support

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

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,31 +83,34 @@
 ]
 
 
-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
 
     return _with_tracer
 
 
 @_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):

instrumentation/opentelemetry-instrumentation-pymemcache/src/opentelemetry/instrumentation/pymemcache/package.py

@@ -3,3 +3,4 @@
 
 
 _instruments = ("pymemcache >= 1.3.5, < 5",)
+_semconv_status = "migration"

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()