diff --git a/tests/test_tracing.py b/tests/test_tracing.py index 089183cc..5f7b2c5b 100644 --- a/tests/test_tracing.py +++ b/tests/test_tracing.py @@ -13,14 +13,28 @@ # limitations under the License. +from types import SimpleNamespace + import pytest from opentelemetry import trace as trace_api from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( OTLPSpanExporter, ) from opentelemetry.sdk import trace as trace_sdk -from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter as OTelInMemorySpanExporter, +) +from opentelemetry.util._once import Once +from veadk.agent import Agent +from veadk.tracing.telemetry import ( + opentelemetry_tracer as opentelemetry_tracer_module, +) +from veadk.tracing.telemetry import telemetry as telemetry_module +from veadk.tracing.telemetry.exporters import ( + apmplus_exporter as apmplus_exporter_module, +) from veadk.tracing.telemetry.exporters.apmplus_exporter import ( APMPlusExporter, APMPlusExporterConfig, @@ -49,13 +63,7 @@ def init_exporters(): ) ) - apmplus_exporter = APMPlusExporter( - config=APMPlusExporterConfig( - endpoint="http://localhost:8000", - app_key="test_app_key", - service_name="test_service_name", - ) - ) + apmplus_exporter = init_apmplus_exporter() tls_exporter = TLSExporter( config=TLSExporterConfig( @@ -69,6 +77,16 @@ def init_exporters(): return [cozeloop_exporter, apmplus_exporter, tls_exporter] +def init_apmplus_exporter(): + return APMPlusExporter( + config=APMPlusExporterConfig( + endpoint="http://localhost:8000", + app_key="test_app_key", + service_name="test_service_name", + ) + ) + + def gen_span_processor(endpoint: str): otlp_exporter = OTLPSpanExporter( endpoint=endpoint, @@ -77,8 +95,127 @@ def gen_span_processor(endpoint: str): return span_processor +@pytest.fixture +def fresh_global_tracer_provider(monkeypatch): + """Give each test an isolated OpenTelemetry global provider.""" + monkeypatch.setattr(trace_api, "_TRACER_PROVIDER", None) + monkeypatch.setattr(trace_api, "_TRACER_PROVIDER_SET_ONCE", Once()) + + yield + + tracer_provider = trace_api.get_tracer_provider() + if isinstance(tracer_provider, trace_sdk.TracerProvider): + tracer_provider.shutdown() + + +@pytest.fixture +def controlled_apmplus_exporter(monkeypatch): + """Provide an APMPlus exporter without network or global meter side effects.""" + constructed_exporters = [] + monkeypatch.setattr(telemetry_module, "meter_uploader", None) + + class ControlledAPMPlusExporter(APMPlusExporter): + def __init__(self): + super().__init__( + config=APMPlusExporterConfig( + endpoint="http://localhost:8000", + app_key="test_app_key", + service_name="test_service_name", + ) + ) + + def model_post_init(self, context): + self._exporter = OTelInMemorySpanExporter() + self.processor = SimpleSpanProcessor(self._exporter) + self.meter_uploader = object() + constructed_exporters.append(self) + + monkeypatch.setattr( + apmplus_exporter_module, + "APMPlusExporter", + ControlledAPMPlusExporter, + ) + monkeypatch.setattr( + opentelemetry_tracer_module, + "APMPlusExporter", + ControlledAPMPlusExporter, + ) + return ControlledAPMPlusExporter, constructed_exporters + + +@pytest.mark.parametrize( + "enable_apmplus", + [False, True], + ids=["env-disabled", "env-enabled"], +) +@pytest.mark.parametrize( + "manual_exporter", + [False, True], + ids=["no-manual-exporter", "manual-exporter"], +) +def test_apmplus_preconfigured_provider_matrix( + fresh_global_tracer_provider, + controlled_apmplus_exporter, + monkeypatch, + enable_apmplus, + manual_exporter, +): + """A preconfigured provider owns traces; env exporter retains metrics.""" + 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) + + tracers = [] + if manual_exporter: + tracers.append(OpentelemetryTracer(exporters=[controlled_exporter_class()])) + + agent = SimpleNamespace(tracers=tracers) + Agent._prepare_tracers(agent) + + 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 + if not should_create_tracer: + assert span_processors == () + return + + tracer = agent.tracers[0] + 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 + ) + assert len(span_processors) == 1 # VeADK in-memory processor only + assert tracer.apmplus_managed_externally is True + expected_meter_uploader = ( + constructed_exporters[-1].meter_uploader if enable_apmplus else None + ) + assert telemetry_module.meter_uploader is expected_meter_uploader + + +def test_tracing_registers_apmplus_without_global_provider( + fresh_global_tracer_provider, +): + apmplus_exporter = init_apmplus_exporter() + + tracer = OpentelemetryTracer(exporters=[apmplus_exporter]) + tracer_provider = trace_api.get_tracer_provider() + span_processors = tracer_provider._active_span_processor._span_processors + + assert apmplus_exporter in tracer.exporters + assert apmplus_exporter.processor in span_processors + + @pytest.mark.asyncio -async def test_tracing(): +async def test_tracing(fresh_global_tracer_provider): exporters = init_exporters() tracer = OpentelemetryTracer(exporters=exporters) @@ -88,7 +225,7 @@ async def test_tracing(): @pytest.mark.asyncio -async def test_tracing_with_global_provider(): +async def test_tracing_with_global_provider(fresh_global_tracer_provider): exporters = init_exporters() # set global tracer provider before init OpentelemetryTracer trace_api.set_tracer_provider(trace_sdk.TracerProvider()) @@ -98,11 +235,11 @@ async def test_tracing_with_global_provider(): # tracer = OpentelemetryTracer(exporters=exporters) - assert len(tracer.exporters) == 4 # with extra 1 built-in exporters + assert len(tracer.exporters) == 3 # APMPlus is managed by the existing provider @pytest.mark.asyncio -async def test_tracing_with_apmplus_global_provider(): +async def test_tracing_with_apmplus_global_provider(fresh_global_tracer_provider): exporters = init_exporters() # add apmplus exporter to global tracer provider before init OpentelemetryTracer trace_api.set_tracer_provider(trace_sdk.TracerProvider()) diff --git a/veadk/tracing/telemetry/opentelemetry_tracer.py b/veadk/tracing/telemetry/opentelemetry_tracer.py index ae16bc54..276111cb 100644 --- a/veadk/tracing/telemetry/opentelemetry_tracer.py +++ b/veadk/tracing/telemetry/opentelemetry_tracer.py @@ -22,7 +22,6 @@ from opentelemetry.sdk import trace as trace_sdk from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import SpanLimits, TracerProvider -from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor from pydantic import BaseModel, ConfigDict, Field, field_validator from typing_extensions import override @@ -160,28 +159,31 @@ def _init_global_tracer_provider(self) -> None: for each exporter, and ensures proper resource attribution. It also handles duplicate exporter detection and in-memory span collection setup. """ - # set provider anyway, then get global provider - 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 + global_tracer_provider = trace_api.get_tracer_provider() + have_global_tracer_provider = not isinstance( + global_tracer_provider, trace_api.ProxyTracerProvider ) - if have_apmplus_exporter: + if not have_global_tracer_provider: + provider = trace_sdk.TracerProvider( + span_limits=SpanLimits(max_attributes=4096) + ) + trace_api.set_tracer_provider(provider) + global_tracer_provider = trace_api.get_tracer_provider() + + global_tracer_provider: TracerProvider + 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 @@ -230,6 +232,11 @@ def _init_global_tracer_provider(self) -> None: init_global_meter_uploader_from_exporters(self.exporters) + @property + def apmplus_managed_externally(self) -> bool: + """Whether a global provider existed before this tracer was initialized.""" + return self._apmplus_managed_externally + @property def trace_file_path(self) -> str: """Get the file path of the most recent trace dump.