Redis Http,Database semconv opt in

Author: tammy-baylis-swi

instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/__init__.py

@@ -152,6 +152,12 @@ def response_hook(span, instance, response):
 from wrapt import wrap_function_wrapper
 
 from opentelemetry import trace
+from opentelemetry.instrumentation._semconv import (
+    _get_schema_url_for_signal_types,
+    _OpenTelemetrySemanticConventionStability,
+    _OpenTelemetryStabilitySignalType,
+    _set_db_statement,
+)
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.instrumentation.redis.package import _instruments
 from opentelemetry.instrumentation.redis.util import (
@@ -167,9 +173,6 @@ def response_hook(span, instance, response):
     is_instrumentation_enabled,
     unwrap,
 )
-from opentelemetry.semconv._incubating.attributes.db_attributes import (
-    DB_STATEMENT,
-)
 from opentelemetry.trace import (
     StatusCode,
     Tracer,
@@ -222,6 +225,15 @@ def _traced_execute_factory(
     request_hook: RequestHook | None = None,
     response_hook: ResponseHook | None = None,
 ):
+    # Get semconv opt-in modes for database and HTTP signal types
+    _OpenTelemetrySemanticConventionStability._initialize()
+    db_sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
+        _OpenTelemetryStabilitySignalType.DATABASE
+    )
+    http_sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
+        _OpenTelemetryStabilitySignalType.HTTP
+    )
+
     def _traced_execute_command(
         func: Callable[..., R],
         instance: RedisInstance,
@@ -237,9 +249,20 @@ def _traced_execute_command(
             name, kind=trace.SpanKind.CLIENT
         ) as span:
             if span.is_recording():
-                span.set_attribute(DB_STATEMENT, query)
-                _set_connection_attributes(span, instance)
-                span.set_attribute("db.redis.args_length", len(args))
+                span_attrs = {}
+                _set_db_statement(span_attrs, query, db_sem_conv_opt_in_mode)
+                span_attrs["db.redis.args_length"] = len(args)
+
+                # Set all DB attributes
+                for key, value in span_attrs.items():
+                    span.set_attribute(key, value)
+
+                _set_connection_attributes(
+                    span,
+                    instance,
+                    db_sem_conv_opt_in_mode,
+                    http_sem_conv_opt_in_mode,
+                )
                 if span.name == "redis.create_index":
                     _add_create_attributes(span, args)
             if callable(request_hook):
@@ -260,6 +283,15 @@ def _traced_execute_pipeline_factory(
     request_hook: RequestHook | None = None,
     response_hook: ResponseHook | None = None,
 ):
+    # Get semconv opt-in modes for database and HTTP signal types
+    _OpenTelemetrySemanticConventionStability._initialize()
+    db_sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
+        _OpenTelemetryStabilitySignalType.DATABASE
+    )
+    http_sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
+        _OpenTelemetryStabilitySignalType.HTTP
+    )
+
     def _traced_execute_pipeline(
         func: Callable[..., R],
         instance: PipelineInstance,
@@ -279,10 +311,21 @@ def _traced_execute_pipeline(
             span_name, kind=trace.SpanKind.CLIENT
         ) as span:
             if span.is_recording():
-                span.set_attribute(DB_STATEMENT, resource)
-                _set_connection_attributes(span, instance)
-                span.set_attribute(
-                    "db.redis.pipeline_length", len(command_stack)
+                span_attrs = {}
+                _set_db_statement(
+                    span_attrs, resource, db_sem_conv_opt_in_mode
+                )
+                span_attrs["db.redis.pipeline_length"] = len(command_stack)
+
+                # Set all DB attributes
+                for key, value in span_attrs.items():
+                    span.set_attribute(key, value)
+
+                _set_connection_attributes(
+                    span,
+                    instance,
+                    db_sem_conv_opt_in_mode,
+                    http_sem_conv_opt_in_mode,
                 )
 
             response = None
@@ -308,6 +351,15 @@ def _async_traced_execute_factory(
     request_hook: RequestHook | None = None,
     response_hook: ResponseHook | None = None,
 ):
+    # Get semconv opt-in modes for database and HTTP signal types
+    _OpenTelemetrySemanticConventionStability._initialize()
+    db_sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
+        _OpenTelemetryStabilitySignalType.DATABASE
+    )
+    http_sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
+        _OpenTelemetryStabilitySignalType.HTTP
+    )
+
     async def _async_traced_execute_command(
         func: Callable[..., Awaitable[R]],
         instance: AsyncRedisInstance,
@@ -324,9 +376,20 @@ async def _async_traced_execute_command(
             name, kind=trace.SpanKind.CLIENT
         ) as span:
             if span.is_recording():
-                span.set_attribute(DB_STATEMENT, query)
-                _set_connection_attributes(span, instance)
-                span.set_attribute("db.redis.args_length", len(args))
+                span_attrs = {}
+                _set_db_statement(span_attrs, query, db_sem_conv_opt_in_mode)
+                span_attrs["db.redis.args_length"] = len(args)
+
+                # Set all DB attributes
+                for key, value in span_attrs.items():
+                    span.set_attribute(key, value)
+
+                _set_connection_attributes(
+                    span,
+                    instance,
+                    db_sem_conv_opt_in_mode,
+                    http_sem_conv_opt_in_mode,
+                )
             if callable(request_hook):
                 request_hook(span, instance, args, kwargs)
             response = await func(*args, **kwargs)
@@ -342,6 +405,15 @@ def _async_traced_execute_pipeline_factory(
     request_hook: RequestHook | None = None,
     response_hook: ResponseHook | None = None,
 ):
+    # Get semconv opt-in modes for database and HTTP signal types
+    _OpenTelemetrySemanticConventionStability._initialize()
+    db_sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
+        _OpenTelemetryStabilitySignalType.DATABASE
+    )
+    http_sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
+        _OpenTelemetryStabilitySignalType.HTTP
+    )
+
     async def _async_traced_execute_pipeline(
         func: Callable[..., Awaitable[R]],
         instance: AsyncPipelineInstance,
@@ -363,10 +435,21 @@ async def _async_traced_execute_pipeline(
             span_name, kind=trace.SpanKind.CLIENT
         ) as span:
             if span.is_recording():
-                span.set_attribute(DB_STATEMENT, resource)
-                _set_connection_attributes(span, instance)
-                span.set_attribute(
-                    "db.redis.pipeline_length", len(command_stack)
+                span_attrs = {}
+                _set_db_statement(
+                    span_attrs, resource, db_sem_conv_opt_in_mode
+                )
+                span_attrs["db.redis.pipeline_length"] = len(command_stack)
+
+                # Set all DB attributes
+                for key, value in span_attrs.items():
+                    span.set_attribute(key, value)
+
+                _set_connection_attributes(
+                    span,
+                    instance,
+                    db_sem_conv_opt_in_mode,
+                    http_sem_conv_opt_in_mode,
                 )
 
             response = None
@@ -540,12 +623,20 @@ def _pipeline_wrapper(func, instance, args, kwargs):
 class RedisInstrumentor(BaseInstrumentor):
     @staticmethod
     def _get_tracer(**kwargs):
+        # Initialize semantic conventions opt-in if needed
+        _OpenTelemetrySemanticConventionStability._initialize()
+        # Redis instrumentation supports both DATABASE and HTTP signal types
+        signal_types = [
+            _OpenTelemetryStabilitySignalType.DATABASE,
+            _OpenTelemetryStabilitySignalType.HTTP,
+        ]
+
         tracer_provider = kwargs.get("tracer_provider")
         return get_tracer(
             __name__,
             __version__,
             tracer_provider=tracer_provider,
-            schema_url="https://opentelemetry.io/schemas/1.11.0",
+            schema_url=_get_schema_url_for_signal_types(signal_types),
         )
 
     def instrument(

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

@@ -14,3 +14,5 @@
 
 
 _instruments = ("redis >= 2.6",)
+
+_semconv_status = "migration"

instrumentation/opentelemetry-instrumentation-redis/src/opentelemetry/instrumentation/redis/util.py

@@ -20,13 +20,15 @@
 
 from typing import TYPE_CHECKING, Any
 
+from opentelemetry.instrumentation._semconv import (
+    _set_db_system,
+    _set_http_net_peer_name_client,
+    _set_http_peer_port_client,
+)
 from opentelemetry.semconv._incubating.attributes.db_attributes import (
     DB_REDIS_DATABASE_INDEX,
-    DB_SYSTEM,
 )
 from opentelemetry.semconv._incubating.attributes.net_attributes import (
-    NET_PEER_NAME,
-    NET_PEER_PORT,
     NET_TRANSPORT,
 )
 from opentelemetry.semconv.trace import (
@@ -46,19 +48,33 @@
 _FIELD_TYPES = ["NUMERIC", "TEXT", "GEO", "TAG", "VECTOR"]
 
 
-def _extract_conn_attributes(conn_kwargs):
+def _extract_conn_attributes(
+    conn_kwargs, db_sem_conv_opt_in_mode, http_sem_conv_opt_in_mode
+):
     """Transform redis conn info into dict"""
-    attributes = {
-        DB_SYSTEM: DbSystemValues.REDIS.value,
-    }
+    attributes = {}
+    _set_db_system(
+        attributes, DbSystemValues.REDIS.value, db_sem_conv_opt_in_mode
+    )
+
     db = conn_kwargs.get("db", 0)
     attributes[DB_REDIS_DATABASE_INDEX] = db
     if "path" in conn_kwargs:
-        attributes[NET_PEER_NAME] = conn_kwargs.get("path", "")
+        _set_http_net_peer_name_client(
+            attributes, conn_kwargs.get("path", ""), http_sem_conv_opt_in_mode
+        )
         attributes[NET_TRANSPORT] = NetTransportValues.OTHER.value
     else:
-        attributes[NET_PEER_NAME] = conn_kwargs.get("host", "localhost")
-        attributes[NET_PEER_PORT] = conn_kwargs.get("port", 6379)
+        _set_http_net_peer_name_client(
+            attributes,
+            conn_kwargs.get("host", "localhost"),
+            http_sem_conv_opt_in_mode,
+        )
+        _set_http_peer_port_client(
+            attributes,
+            conn_kwargs.get("port", 6379),
+            http_sem_conv_opt_in_mode,
+        )
         attributes[NET_TRANSPORT] = NetTransportValues.IP_TCP.value
 
     return attributes
@@ -99,12 +115,17 @@ def _value_or_none(values, n):
 
 
 def _set_connection_attributes(
-    span: Span, conn: RedisInstance | AsyncRedisInstance
+    span: Span,
+    conn: RedisInstance | AsyncRedisInstance,
+    db_sem_conv_opt_in_mode,
+    http_sem_conv_opt_in_mode,
 ) -> None:
     if not span.is_recording() or not hasattr(conn, "connection_pool"):
         return
     for key, value in _extract_conn_attributes(
-        conn.connection_pool.connection_kwargs
+        conn.connection_pool.connection_kwargs,
+        db_sem_conv_opt_in_mode,
+        http_sem_conv_opt_in_mode,
     ).items():
         span.set_attribute(key, value)