HYBIM-513 Push down attributes to Weaviate (#247)

Author: JWinermaSplunk

instrumentation-genai/opentelemetry-instrumentation-weaviate/src/opentelemetry/instrumentation/weaviate/wrapper.py

@@ -40,6 +40,13 @@
     parse_url_to_host_port,
 )
 
+# GenAI context attributes to propagate from parent agent/workflow spans
+_GENAI_PROPAGATE_ATTRS = (
+    "gen_ai.agent.name",
+    "gen_ai.agent.id",
+    "gen_ai.workflow.name",
+)
+
 
 # Context variable for passing connection info within operation call stacks
 _connection_host_context: ContextVar[Optional[str]] = ContextVar(
@@ -163,6 +170,10 @@ def _create_db_span(
             wrapped, instance, args, kwargs, module_name, function_name
         )
 
+        # Capture parent span before entering the new span so we can propagate
+        # gen_ai agent/workflow attributes set by LangChain, CrewAI, LlamaIndex, etc.
+        parent_span = trace.get_current_span()
+
         with self.tracer.start_as_current_span(name, kind=SpanKind.CLIENT) as span:
             span.set_attribute(DbAttributes.DB_SYSTEM_NAME, "weaviate")
 
@@ -176,6 +187,8 @@ def _create_db_span(
             if connection_port is not None:
                 span.set_attribute(ServerAttributes.SERVER_PORT, connection_port)
 
+            self._propagate_genai_context(span, parent_span)
+
             # Record operation duration as a metric
             start_time = time.perf_counter()
             try:
@@ -196,14 +209,9 @@ def _create_db_span(
                 if connection_port is not None:
                     metric_attributes[ServerAttributes.SERVER_PORT] = connection_port
 
-                # Record the duration metric with span context to link metric to trace
-                try:
-                    context = trace.set_span_in_context(span)
-                except (TypeError, ValueError, AttributeError):
-                    context = None
-
+                # Record the duration metric (automatically linked to current span context)
                 self.duration_histogram.record(
-                    duration_ms, attributes=metric_attributes, context=context
+                    duration_ms, attributes=metric_attributes
                 )
 
             # Extract documents from similarity search operations
@@ -247,6 +255,29 @@ def _create_db_span(
 
         return return_value
 
+    def _propagate_genai_context(self, span: Any, parent_span: Any) -> None:
+        """Copy gen_ai agent/workflow attributes from the parent span into the Weaviate span.
+
+        This allows Weaviate spans to inherit context (e.g. gen_ai.agent.name,
+        gen_ai.workflow.name) set by upstream instrumentations such as LangChain,
+        CrewAI, or LlamaIndex, without taking a dependency on splunk-otel-util-genai.
+        """
+        try:
+            if parent_span is None or not parent_span.is_recording():
+                return
+            parent_attrs = getattr(parent_span, "_attributes", None) or getattr(
+                parent_span, "attributes", None
+            )
+            if not parent_attrs:
+                return
+            for attr in _GENAI_PROPAGATE_ATTRS:
+                val = parent_attrs.get(attr)
+                if val is not None:
+                    span.set_attribute(attr, val)
+        except Exception as e:
+            if Config.exception_logger:
+                Config.exception_logger(e)
+
     def _is_similarity_search(self) -> bool:
         """Check if the operation is a similarity search or retrieval operation that returns documents."""
         function_name = self.wrap_properties.get("function", "")

instrumentation-genai/opentelemetry-instrumentation-weaviate/tests/test_span_attributes.py

@@ -524,6 +524,48 @@ def mock_graphql_func(*args, **kwargs):
             assert event.name == "weaviate.document"
 
 
+class TestGenAIContextPropagation:
+    """Test that gen_ai agent/workflow attributes are propagated from parent spans."""
+
+    def test_propagates_agent_and_workflow_attrs_from_parent(
+        self, tracer_provider, span_exporter
+    ):
+        """Test that gen_ai.agent.name, gen_ai.agent.id, and gen_ai.workflow.name
+        are copied from an active parent span into the Weaviate span."""
+        trace.set_tracer_provider(tracer_provider)
+        tracer = tracer_provider.get_tracer(__name__)
+        mock_histogram = MagicMock()
+
+        wrapper = _WeaviateOperationWrapper(
+            tracer=tracer,
+            duration_histogram=mock_histogram,
+            wrap_properties={
+                "module": "weaviate.collections.queries.fetch_objects",
+                "function": "fetch_objects",
+                "span_name": "collections.query.fetch_objects",
+            },
+        )
+
+        def mock_op(*args, **kwargs):
+            return MagicMock(objects=[])
+
+        with tracer.start_as_current_span("invoke_agent my_agent") as parent_span:
+            parent_span.set_attribute("gen_ai.agent.name", "my_agent")
+            parent_span.set_attribute("gen_ai.agent.id", "agent-123")
+            parent_span.set_attribute("gen_ai.workflow.name", "my_workflow")
+
+            wrapper(mock_op, None, (), {})
+
+        spans = span_exporter.get_finished_spans()
+        weaviate_span = next((s for s in spans if "fetch_objects" in s.name), None)
+        assert weaviate_span is not None
+
+        attrs = dict(weaviate_span.attributes or {})
+        assert attrs.get("gen_ai.agent.name") == "my_agent"
+        assert attrs.get("gen_ai.agent.id") == "agent-123"
+        assert attrs.get("gen_ai.workflow.name") == "my_workflow"
+
+
 @pytest.mark.integration
 @pytest.mark.skipif(
     not WEAVIATE_AVAILABLE or WEAVIATE_VERSION < 4,