diff --git a/tests/test_tracing.py b/tests/test_tracing.py index 5f7b2c5be..1365b280f 100644 --- a/tests/test_tracing.py +++ b/tests/test_tracing.py @@ -148,26 +148,34 @@ def model_post_init(self, context): [False, True], ids=["env-disabled", "env-enabled"], ) +@pytest.mark.parametrize( + "provider_preconfigured", + [False, True], + ids=["provider-proxy", "provider-preconfigured"], +) @pytest.mark.parametrize( "manual_exporter", [False, True], ids=["no-manual-exporter", "manual-exporter"], ) -def test_apmplus_preconfigured_provider_matrix( +def test_apmplus_enable_provider_and_manual_exporter_matrix( fresh_global_tracer_provider, controlled_apmplus_exporter, monkeypatch, enable_apmplus, + provider_preconfigured, manual_exporter, ): - """A preconfigured provider owns traces; env exporter retains metrics.""" + """Validate env, global provider, and explicit exporter independently.""" controlled_exporter_class, constructed_exporters = controlled_apmplus_exporter monkeypatch.setenv("ENABLE_APMPLUS", str(enable_apmplus).lower()) monkeypatch.setenv("ENABLE_COZELOOP", "false") monkeypatch.setenv("ENABLE_TLS", "false") - tracer_provider = trace_sdk.TracerProvider() - trace_api.set_tracer_provider(tracer_provider) + initial_provider = trace_api.get_tracer_provider() + if provider_preconfigured: + initial_provider = trace_sdk.TracerProvider() + trace_api.set_tracer_provider(initial_provider) tracers = [] if manual_exporter: @@ -178,29 +186,72 @@ def test_apmplus_preconfigured_provider_matrix( should_create_tracer = manual_exporter or enable_apmplus assert len(agent.tracers) == int(should_create_tracer) - assert trace_api.get_tracer_provider() is tracer_provider - assert len(constructed_exporters) == int(manual_exporter) + int(enable_apmplus) - span_processors = tracer_provider._active_span_processor._span_processors + final_provider = trace_api.get_tracer_provider() if not should_create_tracer: - assert span_processors == () + assert final_provider is initial_provider + assert constructed_exporters == [] + assert telemetry_module.meter_uploader is None return tracer = agent.tracers[0] + assert isinstance(final_provider, trace_sdk.TracerProvider) + if provider_preconfigured: + assert final_provider is initial_provider + else: + assert final_provider is not initial_provider + + should_register_apmplus = not provider_preconfigured and ( + manual_exporter or enable_apmplus + ) + should_construct_apmplus = manual_exporter or enable_apmplus + assert len(constructed_exporters) == int(should_construct_apmplus) assert sum( isinstance(exporter, controlled_exporter_class) for exporter in tracer.exporters - ) == int(enable_apmplus) - assert all( - exporter.processor not in span_processors for exporter in constructed_exporters + ) == int(should_construct_apmplus) + + span_processors = final_provider._active_span_processor._span_processors + registered_apmplus_processors = sum( + any(processor is exporter.processor for processor in span_processors) + for exporter in constructed_exporters ) - assert len(span_processors) == 1 # VeADK in-memory processor only - assert tracer.apmplus_managed_externally is True + assert registered_apmplus_processors == int(should_register_apmplus) + assert len(span_processors) == 1 + int(should_register_apmplus) + assert tracer.apmplus_managed_externally is provider_preconfigured expected_meter_uploader = ( - constructed_exporters[-1].meter_uploader if enable_apmplus else None + constructed_exporters[0].meter_uploader if should_construct_apmplus else None ) assert telemetry_module.meter_uploader is expected_meter_uploader +def test_add_exporter_registers_after_tracer_initialization( + fresh_global_tracer_provider, +): + tracer = OpentelemetryTracer() + exporter = init_exporters()[1] + + tracer.add_exporter(exporter) + + tracer_provider = trace_api.get_tracer_provider() + span_processors = tracer_provider._active_span_processor._span_processors + assert exporter in tracer.exporters + assert exporter.processor in span_processors + assert exporter.processor in tracer._processors + + +def test_add_exporter_is_idempotent(fresh_global_tracer_provider): + tracer = OpentelemetryTracer() + exporter = init_exporters()[1] + + tracer.add_exporter(exporter) + tracer.add_exporter(exporter) + + tracer_provider = trace_api.get_tracer_provider() + span_processors = tracer_provider._active_span_processor._span_processors + assert sum(processor is exporter.processor for processor in span_processors) == 1 + assert sum(processor is exporter.processor for processor in tracer._processors) == 1 + + def test_tracing_registers_apmplus_without_global_provider( fresh_global_tracer_provider, ): @@ -214,6 +265,66 @@ def test_tracing_registers_apmplus_without_global_provider( assert apmplus_exporter.processor in span_processors +def test_tracing_skips_apmplus_for_preconfigured_provider( + fresh_global_tracer_provider, +): + tracer_provider = trace_sdk.TracerProvider() + trace_api.set_tracer_provider(tracer_provider) + apmplus_exporter = init_apmplus_exporter() + + tracer = OpentelemetryTracer(exporters=[apmplus_exporter]) + global_tracer_provider = trace_api.get_tracer_provider() + span_processors = global_tracer_provider._active_span_processor._span_processors + + assert global_tracer_provider is tracer_provider + assert apmplus_exporter in tracer.exporters + assert apmplus_exporter.processor not in span_processors + assert len(span_processors) == 1 # VeADK in-memory processor only + assert telemetry_module.meter_uploader is apmplus_exporter.meter_uploader + + +def test_add_exporter_skips_apmplus_for_preconfigured_provider( + fresh_global_tracer_provider, +): + tracer_provider = trace_sdk.TracerProvider() + trace_api.set_tracer_provider(tracer_provider) + tracer = OpentelemetryTracer() + apmplus_exporter = init_apmplus_exporter() + + tracer.add_exporter(apmplus_exporter) + + span_processors = tracer_provider._active_span_processor._span_processors + assert apmplus_exporter in tracer.exporters + assert apmplus_exporter.processor not in span_processors + assert len(span_processors) == 1 # VeADK in-memory processor only + assert telemetry_module.meter_uploader is apmplus_exporter.meter_uploader + + +def test_agent_env_keeps_metrics_only_apmplus_for_preconfigured_provider( + fresh_global_tracer_provider, + controlled_apmplus_exporter, + monkeypatch, +): + controlled_exporter_class, constructed_exporters = controlled_apmplus_exporter + tracer_provider = trace_sdk.TracerProvider() + trace_api.set_tracer_provider(tracer_provider) + tracer = OpentelemetryTracer() + monkeypatch.setenv("ENABLE_APMPLUS", "true") + monkeypatch.setenv("ENABLE_COZELOOP", "false") + monkeypatch.setenv("ENABLE_TLS", "false") + + Agent._prepare_tracers(SimpleNamespace(tracers=[tracer])) + + span_processors = tracer_provider._active_span_processor._span_processors + assert len(constructed_exporters) == 1 + exporter = constructed_exporters[0] + assert isinstance(exporter, controlled_exporter_class) + assert exporter in tracer.exporters + assert exporter.processor not in span_processors + assert telemetry_module.meter_uploader is exporter.meter_uploader + assert len(span_processors) == 1 # VeADK in-memory processor only + + @pytest.mark.asyncio async def test_tracing(fresh_global_tracer_provider): exporters = init_exporters() @@ -232,10 +343,10 @@ async def test_tracing_with_global_provider(fresh_global_tracer_provider): tracer_provider = trace_api.get_tracer_provider() tracer_provider.add_span_processor(gen_span_processor("http://localhost:8000")) trace_api.set_tracer_provider(tracer_provider) - # tracer = OpentelemetryTracer(exporters=exporters) - assert len(tracer.exporters) == 3 # APMPlus is managed by the existing provider + # APMPlus is retained for metrics but its span processor is not registered. + assert len(tracer.exporters) == 4 @pytest.mark.asyncio @@ -249,5 +360,5 @@ async def test_tracing_with_apmplus_global_provider(fresh_global_tracer_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 is retained for metrics but its span processor is not registered. + assert len(tracer.exporters) == 4 # with extra 1 built-in exporters diff --git a/veadk/agent.py b/veadk/agent.py index fd83dd305..d5e0e2145 100644 --- a/veadk/agent.py +++ b/veadk/agent.py @@ -671,17 +671,17 @@ 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 + self.tracers[0].add_exporter(APMPlusExporter()) # type: ignore 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 + self.tracers[0].add_exporter(CozeloopExporter()) # type: ignore 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 + self.tracers[0].add_exporter(TLSExporter()) # type: ignore logger.info("Enable TLS exporter by env.") logger.debug( diff --git a/veadk/tracing/telemetry/exporters/base_exporter.py b/veadk/tracing/telemetry/exporters/base_exporter.py index 8ec2ac072..9fe0d7e61 100644 --- a/veadk/tracing/telemetry/exporters/base_exporter.py +++ b/veadk/tracing/telemetry/exporters/base_exporter.py @@ -12,7 +12,8 @@ # See the License for the specific language governing permissions and # limitations under the License. -from opentelemetry.sdk.trace import SpanProcessor +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 @@ -32,8 +33,27 @@ class BaseExporter(BaseModel): headers: dict = Field(default_factory=dict) _exporter: SpanExporter | None = None + _registered_provider: TracerProvider | None = None processor: SpanProcessor | None = None + def register(self, provider: TracerProvider) -> bool: + """Register this exporter's processor with a tracer provider once. + + Returns: + Whether the processor was newly registered with ``provider``. + """ + if self.processor is None or self._registered_provider is provider: + return False + + if self.resource_attributes: + provider._resource = provider._resource.merge( + Resource.create(self.resource_attributes) + ) + + provider.add_span_processor(self.processor) + self._registered_provider = provider + return True + def export(self) -> None: """Force export of telemetry data.""" pass diff --git a/veadk/tracing/telemetry/opentelemetry_tracer.py b/veadk/tracing/telemetry/opentelemetry_tracer.py index 276111cb9..282ee63b2 100644 --- a/veadk/tracing/telemetry/opentelemetry_tracer.py +++ b/veadk/tracing/telemetry/opentelemetry_tracer.py @@ -20,7 +20,6 @@ from opentelemetry import trace as trace_api from opentelemetry.sdk import trace as trace_sdk -from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import SpanLimits, TracerProvider from pydantic import BaseModel, ConfigDict, Field, field_validator from typing_extensions import override @@ -36,21 +35,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. @@ -172,39 +156,11 @@ def _init_global_tracer_provider(self) -> None: global_tracer_provider = trace_api.get_tracer_provider() global_tracer_provider: TracerProvider + self._global_tracer_provider = global_tracer_provider self._apmplus_managed_externally = have_global_tracer_provider - if self._apmplus_managed_externally: - exporter_count = len(self.exporters) - self.exporters = [ - e for e in self.exporters if not isinstance(e, APMPlusExporter) - ] - if len(self.exporters) != exporter_count: - logger.info( - "Reuse existing global TracerProvider and skip registering " - "APMPlusExporter." - ) - 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) - - logger.debug( - f"Add span processor for exporter `{exporter.__class__.__name__}` to OpentelemetryTracer." - ) - else: - logger.error( - f"Add span processor for exporter `{exporter.__class__.__name__}` to OpentelemetryTracer failed." - ) + self._register_exporter(exporter) self._inmemory_exporter = InMemoryExporter() if self._inmemory_exporter.processor: @@ -232,11 +188,46 @@ def _init_global_tracer_provider(self) -> None: init_global_meter_uploader_from_exporters(self.exporters) + def _register_exporter(self, exporter: BaseExporter) -> bool: + if isinstance(exporter, APMPlusExporter) and self.apmplus_managed_externally: + logger.info( + "Reuse existing global TracerProvider and skip registering " + "APMPlusExporter span processor." + ) + return False + + if exporter.register(self._global_tracer_provider): + self._processors.append(exporter.processor) + logger.debug( + f"Add span processor for exporter `{exporter.__class__.__name__}` to OpentelemetryTracer." + ) + return True + + if exporter.processor is None: + logger.error( + f"Add span processor for exporter `{exporter.__class__.__name__}` to OpentelemetryTracer failed." + ) + return False + @property def apmplus_managed_externally(self) -> bool: """Whether a global provider existed before this tracer was initialized.""" return self._apmplus_managed_externally + def add_exporter(self, exporter: BaseExporter) -> bool: + """Add an exporter and immediately register it with the global provider.""" + if not any(existing is exporter for existing in self.exporters): + self.exporters.append(exporter) + + registered = self._register_exporter(exporter) + + from veadk.tracing.telemetry.telemetry import ( + init_global_meter_uploader_from_exporters, + ) + + init_global_meter_uploader_from_exporters(self.exporters) + return registered + @property def trace_file_path(self) -> str: """Get the file path of the most recent trace dump.