fix: fix exporter register

Author: feng-95

tests/test_tracing.py

@@ -112,5 +112,6 @@ async def test_tracing_with_apmplus_global_provider():
     # init OpentelemetryTracer
     tracer = OpentelemetryTracer(exporters=exporters)
 
-    # apmplus exporter won't init again, so there are cozeloop, tls, in_memory exporter
-    assert len(tracer.exporters) == 3  # with extra 1 built-in exporters
+    # apmplus exporter won't init again, so there are cozeloop, tls, and in_memory exporter
+    # Note: APMPlusExporter still exists in the list but isn't registered with the tracer provider
+    assert len(tracer.exporters) == 4  # with extra 1 built-in exporters

veadk/agent.py

@@ -287,17 +287,23 @@ def _prepare_tracers(self):
         if enable_apmplus_tracer and not any(
             isinstance(e, APMPlusExporter) for e in exporters
         ):
-            self.tracers[0].exporters.append(APMPlusExporter())  # type: ignore
+            exporter = APMPlusExporter()
+            self.tracers[0].exporters.append(exporter)  # type: ignore
+            exporter.register()
             logger.info("Enable APMPlus exporter by env.")
 
         if enable_cozeloop_tracer and not any(
             isinstance(e, CozeloopExporter) for e in exporters
         ):
-            self.tracers[0].exporters.append(CozeloopExporter())  # type: ignore
+            exporter = CozeloopExporter()
+            self.tracers[0].exporters.append(exporter)  # type: ignore
+            exporter.register()
             logger.info("Enable CozeLoop exporter by env.")
 
         if enable_tls_tracer and not any(isinstance(e, TLSExporter) for e in exporters):
-            self.tracers[0].exporters.append(TLSExporter())  # type: ignore
+            exporter = TLSExporter()
+            self.tracers[0].exporters.append(exporter)  # type: ignore
+            exporter.register()
             logger.info("Enable TLS exporter by env.")
 
         logger.debug(

veadk/tracing/telemetry/exporters/apmplus_exporter.py

@@ -535,6 +535,39 @@ def model_post_init(self, context: Any) -> None:
             resource_attributes=self.resource_attributes,
         )
 
+    @override
+    def register(self) -> None:
+        """Register the APMPlus exporter with the global tracer provider.
+
+        This method ensures compatibility with automatic instrumentation by
+        checking if there are any existing APMPlus span processors before
+        registering the exporter. If there are existing ones, it skips
+        registration to avoid conflicts.
+        """
+        # Check for existing APMPlus span processors to ensure compatibility with automatic instrumentation
+        from opentelemetry import trace
+        from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor
+        
+        # Get the global tracer provider
+        tracer_provider = trace.get_tracer_provider()
+        
+        # Check if there are any existing span processors that are APMPlus related
+        span_processors = tracer_provider._active_span_processor._span_processors
+
+        # TODO: Compatible with one-agent collector
+        has_existing_apmplus = any(
+            isinstance(p, (BatchSpanProcessor, SimpleSpanProcessor))
+            and hasattr(p.span_exporter, "_endpoint")
+            and "apmplus" in p.span_exporter._endpoint
+            for p in span_processors
+        )
+        
+        if not has_existing_apmplus:
+            # Call the base class register method to handle resource attributes and processor registration
+            super().register()
+        else:
+            logger.info("APMPlusExporter: Skipping registration as APMPlus processor already exists.")
+    
     @override
     def export(self) -> None:
         """Force immediate export of pending telemetry data to APMPlus.

veadk/tracing/telemetry/exporters/base_exporter.py

@@ -12,10 +12,26 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from opentelemetry.sdk.trace import SpanProcessor
+from opentelemetry import trace
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import SpanProcessor, TracerProvider
 from opentelemetry.sdk.trace.export import SpanExporter
 from pydantic import BaseModel, ConfigDict, Field
 
+def _update_resource_attributions(
+    provider: TracerProvider, resource_attributes: dict
+) -> None:
+    """Update the resource attributes of a TracerProvider instance.
+
+    This function merges new resource attributes with the existing ones in the
+    provider, allowing dynamic configuration of telemetry metadata.
+
+    Args:
+        provider: The TracerProvider instance to update
+        resource_attributes: Dictionary of attributes to merge with existing resources
+    """
+    provider._resource = provider._resource.merge(Resource.create(resource_attributes))
+
 
 class BaseExporter(BaseModel):
     """Abstract base class for OpenTelemetry span exporters in VeADK tracing system.
@@ -33,6 +49,28 @@ class BaseExporter(BaseModel):
 
     _exporter: SpanExporter | None = None
     processor: SpanProcessor | None = None
+    _registered: bool = False
+
+    def register(self) -> None:
+        """Register the exporter with the global tracer provider.
+
+        This method will automatically get the global tracer provider
+        and register the exporter's span processor with it.
+        The registration Dprocess will only be executed once.
+        """
+        if self._registered:
+            return
+
+        tracer_provider = trace.get_tracer_provider()
+        # Update resource attributes if any
+        if self.resource_attributes:
+            _update_resource_attributions(tracer_provider, self.resource_attributes)
+        
+        # Add processor to tracer provider if exists
+        if self.processor:
+            tracer_provider.add_span_processor(self.processor)
+            
+        self._registered = True
 
     def export(self) -> None:
         """Force export of telemetry data."""

veadk/tracing/telemetry/exporters/inmemory_exporter.py

@@ -14,6 +14,7 @@
 
 from typing import Sequence
 
+from opentelemetry import trace
 from opentelemetry.context import (
     _SUPPRESS_INSTRUMENTATION_KEY,
     attach,
@@ -246,3 +247,18 @@ def __init__(self, name: str = "inmemory_exporter") -> None:
 
         self._exporter = _InMemoryExporter()
         self.processor = _InMemorySpanProcessor(self._exporter)
+
+    @override
+    def register(self) -> None:
+        """Register the in-memory exporter with the global tracer provider.
+
+        Ensures the in-memory exporter's processor is added at the beginning
+        of the span processors list to record all spans.
+        """
+        tracer_provider = trace.get_tracer_provider()
+
+        # Ensure the in-memory exporter processor is added at index 0
+        # because we use this to record all spans
+        tracer_provider._active_span_processor._span_processors = (
+            self.processor,
+        ) + tracer_provider._active_span_processor._span_processors

veadk/tracing/telemetry/opentelemetry_tracer.py

@@ -27,7 +27,6 @@
 from typing_extensions import override
 
 from veadk.tracing.base_tracer import BaseTracer
-from veadk.tracing.telemetry.exporters.apmplus_exporter import APMPlusExporter
 from veadk.tracing.telemetry.exporters.base_exporter import BaseExporter
 from veadk.tracing.telemetry.exporters.inmemory_exporter import InMemoryExporter
 from veadk.utils.logger import get_logger
@@ -37,21 +36,6 @@
 logger = get_logger(__name__)
 
 
-def _update_resource_attributions(
-    provider: TracerProvider, resource_attributes: dict
-) -> None:
-    """Update the resource attributes of a TracerProvider instance.
-
-    This function merges new resource attributes with the existing ones in the
-    provider, allowing dynamic configuration of telemetry metadata.
-
-    Args:
-        provider: The TracerProvider instance to update
-        resource_attributes: Dictionary of attributes to merge with existing resources
-    """
-    provider._resource = provider._resource.merge(Resource.create(resource_attributes))
-
-
 class OpentelemetryTracer(BaseModel, BaseTracer):
     """OpenTelemetry-based tracer implementation for comprehensive agent observability.
 
@@ -152,41 +136,27 @@ def _init_global_tracer_provider(self) -> None:
         duplicate exporter detection and in-memory span collection setup.
         """
         # set provider anyway, then get global provider
+        # set if not exist
         trace_api.set_tracer_provider(
             trace_sdk.TracerProvider(
                 span_limits=SpanLimits(
                     max_attributes=4096,
                 )
             )
         )
-        global_tracer_provider: TracerProvider = trace_api.get_tracer_provider()  # type: ignore
-
-        span_processors = global_tracer_provider._active_span_processor._span_processors
-        have_apmplus_exporter = any(
-            isinstance(p, (BatchSpanProcessor, SimpleSpanProcessor))
-            and hasattr(p.span_exporter, "_endpoint")
-            and "apmplus" in p.span_exporter._endpoint
-            for p in span_processors
-        )
 
-        if have_apmplus_exporter:
-            self.exporters = [
-                e for e in self.exporters if not isinstance(e, APMPlusExporter)
-            ]
+        # add in-memory exporter to exporters list
+        self._inmemory_exporter = InMemoryExporter()
+        self.exporters.append(self._inmemory_exporter)
 
+        # Call each exporter's register method to register with the global tracer provider
+        # This allows each exporter to handle its own configuration and registration logic
         for exporter in self.exporters:
-            processor = exporter.processor
-            resource_attributes = exporter.resource_attributes
-
-            if resource_attributes:
-                _update_resource_attributions(
-                    global_tracer_provider, resource_attributes
-                )
-
-            if processor:
-                global_tracer_provider.add_span_processor(processor)
-                self._processors.append(processor)
-
+            exporter.register()
+            
+            # Add processor to internal list if exists
+            if exporter.processor:
+                self._processors.append(exporter.processor)
                 logger.debug(
                     f"Add span processor for exporter `{exporter.__class__.__name__}` to OpentelemetryTracer."
                 )
@@ -195,21 +165,6 @@ def _init_global_tracer_provider(self) -> None:
                     f"Add span processor for exporter `{exporter.__class__.__name__}` to OpentelemetryTracer failed."
                 )
 
-        self._inmemory_exporter = InMemoryExporter()
-        if self._inmemory_exporter.processor:
-            # make sure the in memory exporter processor is added at index 0
-            # because we use this to record all spans
-            global_tracer_provider._active_span_processor._span_processors = (
-                self._inmemory_exporter.processor,
-            ) + global_tracer_provider._active_span_processor._span_processors
-
-            self._processors.append(self._inmemory_exporter.processor)
-            self.exporters.append(self._inmemory_exporter)
-        else:
-            logger.warning(
-                "InMemoryExporter processor is not initialized, cannot add to OpentelemetryTracer."
-            )
-
         logger.info(
             f"Init OpentelemetryTracer with {len(self._processors)} exporter(s)."
         )