diff --git a/nucliadb_telemetry/pyproject.toml b/nucliadb_telemetry/pyproject.toml index a5b60c830f..e8ec5d9dec 100644 --- a/nucliadb_telemetry/pyproject.toml +++ b/nucliadb_telemetry/pyproject.toml @@ -36,10 +36,9 @@ dependencies = [ # dependencies but that can run into problems when installing editable # dependencies. Keeping the duplication for now. otel = [ - "opentelemetry-sdk>=1.21.0,<1.40", # Restricted to <1.40 because jaeger exporter is not compatible with newer versions - "opentelemetry-api>=1.21.0,<1.40", - "opentelemetry-proto>=1.21.0,<1.40", - "opentelemetry-exporter-jaeger-thrift>=1.21.0", + "opentelemetry-sdk>=1.21.0", + "opentelemetry-api>=1.21.0", + "opentelemetry-proto>=1.21.0", "opentelemetry-exporter-otlp-proto-grpc>=1.21.0", "opentelemetry-propagator-b3>=1.21.0", "opentelemetry-instrumentation-fastapi>=0.42b0", @@ -57,10 +56,9 @@ grpc = [ "grpcio-reflection>=1.71.0", # Include OpenTelemetry deps - "opentelemetry-sdk>=1.21.0,<1.40", # Restricted to <1.40 because jaeger exporter is not compatible with newer versions - "opentelemetry-api>=1.21.0,<1.40", - "opentelemetry-proto>=1.21.0,<1.40", - "opentelemetry-exporter-jaeger-thrift>=1.21.0", + "opentelemetry-sdk>=1.21.0", + "opentelemetry-api>=1.21.0", + "opentelemetry-proto>=1.21.0", "opentelemetry-exporter-otlp-proto-grpc>=1.21.0", "opentelemetry-propagator-b3>=1.21.0", "opentelemetry-instrumentation-fastapi>=0.42b0", @@ -73,10 +71,9 @@ nats = [ "PyNaCl", # Include OpenTelemetry deps - "opentelemetry-sdk>=1.21.0,<1.40", # Restricted to <1.40 because jaeger exporter is not compatible with newer versions - "opentelemetry-api>=1.21.0,<1.40", - "opentelemetry-proto>=1.21.0,<1.40", - "opentelemetry-exporter-jaeger-thrift>=1.21.0", + "opentelemetry-sdk>=1.21.0", + "opentelemetry-api>=1.21.0", + "opentelemetry-proto>=1.21.0", "opentelemetry-exporter-otlp-proto-grpc>=1.21.0", "opentelemetry-propagator-b3>=1.21.0", "opentelemetry-instrumentation-fastapi>=0.42b0", @@ -88,10 +85,9 @@ fastapi = [ "fastapi", # Include OpenTelemetry deps - "opentelemetry-sdk>=1.21.0,<1.40", # Restricted to <1.40 because jaeger exporter is not compatible with newer versions - "opentelemetry-api>=1.21.0,<1.40", - "opentelemetry-proto>=1.21.0,<1.40", - "opentelemetry-exporter-jaeger-thrift>=1.21.0", + "opentelemetry-sdk>=1.21.0", + "opentelemetry-api>=1.21.0", + "opentelemetry-proto>=1.21.0", "opentelemetry-exporter-otlp-proto-grpc>=1.21.0", "opentelemetry-propagator-b3>=1.21.0", "opentelemetry-instrumentation-fastapi>=0.42b0", @@ -101,10 +97,9 @@ fastapi = [ ] all = [ # Include everything - "opentelemetry-sdk>=1.21.0,<1.40", # Restricted to <1.40 because jaeger exporter is not compatible with newer versions - "opentelemetry-api>=1.21.0,<1.40", - "opentelemetry-proto>=1.21.0,<1.40", - "opentelemetry-exporter-jaeger-thrift>=1.21.0", + "opentelemetry-sdk>=1.21.0", + "opentelemetry-api>=1.21.0", + "opentelemetry-proto>=1.21.0", "opentelemetry-exporter-otlp-proto-grpc>=1.21.0", "opentelemetry-propagator-b3>=1.21.0", "opentelemetry-instrumentation-fastapi>=0.42b0", diff --git a/nucliadb_telemetry/src/nucliadb_telemetry/batch_span.py b/nucliadb_telemetry/src/nucliadb_telemetry/batch_span.py deleted file mode 100644 index 851ce6225d..0000000000 --- a/nucliadb_telemetry/src/nucliadb_telemetry/batch_span.py +++ /dev/null @@ -1,309 +0,0 @@ -# Copyright 2025 Bosutech XXI S.L. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -import asyncio -import time - -from opentelemetry.context import ( - _SUPPRESS_INSTRUMENTATION_KEY, - Context, - attach, - detach, - set_value, -) -from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor -from opentelemetry.sdk.trace.export import SpanExporter - -from nucliadb_telemetry import logger - - -class _FlushRequest: - """Represents a request for the BatchSpanProcessor to flush spans.""" - - __slots__ = ["event", "num_spans"] - - def __init__(self): - self.event = asyncio.Event() - self.num_spans = 0 - - -class BatchSpanProcessor(SpanProcessor): - """Batch span processor implementation. - `BatchSpanProcessor` is an implementation of `SpanProcessor` that - batches ended spans and pushes them to the configured `SpanExporter`. - `BatchSpanProcessor` is configurable with the following environment - variables which correspond to constructor parameters: - - :envvar:`OTEL_BSP_SCHEDULE_DELAY` - - :envvar:`OTEL_BSP_MAX_QUEUE_SIZE` - - :envvar:`OTEL_BSP_MAX_EXPORT_BATCH_SIZE` - - :envvar:`OTEL_BSP_EXPORT_TIMEOUT` - """ - - def __init__( - self, - span_exporter: SpanExporter, - max_queue_size: int = 2048, - schedule_delay_millis: int = 5000, - max_export_batch_size: int = 512, - export_timeout_millis: int = 30000, - ): - if max_queue_size <= 0: - raise ValueError("max_queue_size must be a positive integer.") - - if schedule_delay_millis <= 0: - raise ValueError("schedule_delay_millis must be positive.") - - if max_export_batch_size <= 0: - raise ValueError("max_export_batch_size must be a positive integer.") - - if max_export_batch_size > max_queue_size: - raise ValueError("max_export_batch_size must be less than or equal to max_queue_size.") - - self.span_exporter = span_exporter - self.queue = asyncio.Queue(maxsize=max_queue_size) # type: asyncio.Queue[Span] - self.worker_task: asyncio.Task = asyncio.create_task( - self.worker(), name="OtelBatchSpanProcessor" - ) - self.condition = asyncio.Condition() - self._flush_request: _FlushRequest | None = None - self.schedule_delay_millis = schedule_delay_millis - self.max_export_batch_size = max_export_batch_size - self.max_queue_size = max_queue_size - self.export_timeout_millis = export_timeout_millis - self.done = False - # flag that indicates that spans are being dropped - self._spans_dropped = False - # precallocated list to send spans to exporter - self.spans_list: list[Span | None] = [None] * self.max_export_batch_size - self.notify_tasks: set[asyncio.Task[None]] = set() - - def on_start(self, span: Span, parent_context: Context | None = None) -> None: - pass - - def on_end(self, span: ReadableSpan) -> None: - if self.done: - logger.warning("Already shutdown, dropping span.") - return - if not span.context.trace_flags.sampled: - return - if self.queue.full(): - if not self._spans_dropped: - logger.warning("Queue is full, likely spans will be dropped.") - self._spans_dropped = True - return - - try: - self.queue.put_nowait(span) # type: ignore[arg-type] - except asyncio.QueueFull: - if not self._spans_dropped: - logger.warning(f"Queue is full. Queue size : {self.queue.qsize()}") - self._spans_dropped = True - return - except Exception as e: - logger.exception(e) - - if self.queue.qsize() >= self.max_export_batch_size: - task = asyncio.create_task(self.notify()) - self.notify_tasks.add(task) - task.add_done_callback(self.notify_tasks.discard) - - async def notify(self): - async with self.condition: - self.condition.notify() - - async def notify_all(self): - async with self.condition: - self.condition.notify_all() - - async def worker(self): - try: - logger.info("Batch telemetry event loop started") - await self._worker() - except Exception as e: - logger.exception(e) - raise e - - async def _worker(self) -> None: - timeout = self.schedule_delay_millis / 1e3 - flush_request: _FlushRequest | None = None - while not self.done: - logger.debug("Waiting condition") - async with self.condition: - if self.done: - # done flag may have changed, avoid waiting - break - logger.debug(f"{self.queue.qsize()} spans on queue") - flush_request = self._get_and_unset_flush_request() - if self.queue.qsize() < self.max_export_batch_size and flush_request is None: - try: - await asyncio.wait_for(self.condition.wait(), timeout) - except asyncio.TimeoutError: - pass - flush_request = self._get_and_unset_flush_request() - if not self.queue: - # spurious notification, let's wait again, reset timeout - timeout = self.schedule_delay_millis / 1e3 - self._notify_flush_request_finished(flush_request) - flush_request = None - continue - if self.done: - # missing spans will be sent when calling flush - break - - self._spans_dropped = False - - # subtract the duration of this export call to the next timeout - start = time.perf_counter_ns() - try: - await asyncio.wait_for(self._export(flush_request), timeout) - except asyncio.TimeoutError: - logger.warning("Took to much time to export, network problem ahead") - end = time.perf_counter_ns() - duration = (end - start) / 1e9 - timeout = self.schedule_delay_millis / 1e3 - duration - - self._notify_flush_request_finished(flush_request) - flush_request = None - - # there might have been a new flush request while export was running - # and before the done flag switched to true - async with self.condition: - shutdown_flush_request = self._get_and_unset_flush_request() - - # be sure that all spans are sent - await self._drain_queue() - self._notify_flush_request_finished(flush_request) - self._notify_flush_request_finished(shutdown_flush_request) - - def _get_and_unset_flush_request( - self, - ) -> _FlushRequest | None: - """Returns the current flush request and makes it invisible to the - worker thread for subsequent calls. - """ - flush_request = self._flush_request - self._flush_request = None - if flush_request is not None: - flush_request.num_spans = self.queue.qsize() - return flush_request - - @staticmethod - def _notify_flush_request_finished( - flush_request: _FlushRequest | None, - ): - """Notifies the flush initiator(s) waiting on the given request/event - that the flush operation was finished. - """ - if flush_request is not None: - flush_request.event.set() - - def _get_or_create_flush_request(self) -> _FlushRequest: - """Either returns the current active flush event or creates a new one. - The flush event will be visible and read by the worker thread before an - export operation starts. Callers of a flush operation may wait on the - returned event to be notified when the flush/export operation was - finished. - This method is not thread-safe, i.e. callers need to take care about - synchronization/locking. - """ - if self._flush_request is None: - self._flush_request = _FlushRequest() - return self._flush_request - - async def _export(self, flush_request: _FlushRequest | None): - """Exports spans considering the given flush_request. - In case of a given flush_requests spans are exported in batches until - the number of exported spans reached or exceeded the number of spans in - the flush request. - In no flush_request was given at most max_export_batch_size spans are - exported. - """ - if not flush_request: - await self._export_batch() - return - - num_spans = flush_request.num_spans - while self.queue.qsize(): - num_exported = await self._export_batch() - num_spans -= num_exported - - if num_spans <= 0: - break - - async def _export_batch(self) -> int: - """Exports at most max_export_batch_size spans and returns the number of - exported spans. - """ - idx = 0 - # currently only a single thread acts as consumer, so queue.pop() will - # not raise an exception - while idx < self.max_export_batch_size and self.queue.qsize(): - self.spans_list[idx] = self.queue.get_nowait() - idx += 1 - token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True)) - try: - # Ignore type b/c the Optional[None]+slicing is too "clever" - # for mypy - await self.span_exporter.async_export(self.spans_list[:idx]) # type: ignore - except asyncio.CancelledError: - logger.exception("Task was canceled while exporting Span batch") - except Exception: # pylint: disable=broad-except - logger.exception("Exception while exporting Span batch)") - detach(token) - - # clean up list - for index in range(idx): - self.spans_list[index] = None - return idx - - async def _drain_queue(self): - """Export all elements until queue is empty. - Can only be called from the worker thread context because it invokes - `export` that is not thread safe. - """ - while self.queue.qsize(): - await self._export_batch() - - async def async_force_flush(self, timeout_millis: int | None = None) -> bool: - if timeout_millis is None: - timeout_millis = self.export_timeout_millis - - if self.done: - logger.warning("Already shutdown, ignoring call to async_force_flush().") - return True - - async with self.condition: - flush_request = self._get_or_create_flush_request() - # signal the worker task to flush and wait for it to finish - self.condition.notify_all() - - if flush_request.num_spans == 0: - return True - - # wait for token to be processed - ret = await asyncio.wait_for(flush_request.event.wait(), timeout_millis) - if not ret: - logger.warning("Timeout was exceeded in async_force_flush().") - return False - return ret - - def shutdown(self) -> None: - # signal the worker thread to finish and then wait for it - self.done = True - self.span_exporter.shutdown() - try: - self.worker_task.cancel() - except RuntimeError: - pass diff --git a/nucliadb_telemetry/src/nucliadb_telemetry/jaeger.py b/nucliadb_telemetry/src/nucliadb_telemetry/jaeger.py deleted file mode 100644 index 03102f9239..0000000000 --- a/nucliadb_telemetry/src/nucliadb_telemetry/jaeger.py +++ /dev/null @@ -1,183 +0,0 @@ -# Copyright 2025 Bosutech XXI S.L. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -import asyncio -import math -import socket -from asyncio import CancelledError, Future -from functools import partial - -from opentelemetry.exporter.jaeger.thrift import JaegerExporter -from opentelemetry.exporter.jaeger.thrift.gen.agent import Agent -from opentelemetry.exporter.jaeger.thrift.gen.jaeger import Collector -from opentelemetry.exporter.jaeger.thrift.translate import ThriftTranslator, Translate -from opentelemetry.sdk.resources import SERVICE_NAME -from opentelemetry.sdk.trace import Span -from opentelemetry.sdk.trace.export import SpanExportResult -from thrift.protocol import TCompactProtocol # type: ignore[import-untyped] -from thrift.transport import TTransport # type: ignore[import-untyped] - -from nucliadb_telemetry import logger - -UDP_PACKET_MAX_LENGTH = 65000 - - -class JaegerExporterAsync(JaegerExporter): - def __init__(self, **kwags): - super().__init__(**kwags) - self._agent_client = AgentClientUDPAsync( - host_name=self.agent_host_name, - port=self.agent_port, - split_oversized_batches=self.udp_split_oversized_batches, - ) - - async def async_export(self, spans: list[Span]) -> SpanExportResult: - # Populate service_name from first span - # We restrict any SpanProcessor to be only associated with a single - # TracerProvider, so it is safe to assume that all Spans in a single - # batch all originate from one TracerProvider (and in turn have all - # the same service.name) - if len(spans) == 0: - return SpanExportResult.SUCCESS - if spans: - service_name = spans[0].resource.attributes.get(SERVICE_NAME) - if service_name: - self.service_name = service_name - translator = Translate(spans) - thrift_translator = ThriftTranslator(self._max_tag_value_length) - jaeger_spans = translator._translate(thrift_translator) - batch = Collector.Batch( - spans=jaeger_spans, - process=Collector.Process(serviceName=self.service_name), - ) - if self._collector_http_client is not None: - raise Exception("Not supported on asyncio") - # self._collector_http_client.submit(batch) - else: - await self._agent_client.emit(batch) - - return SpanExportResult.SUCCESS - - -class JaegerClientProtocol: - def __init__(self, message: bytes, on_con_lost: Future): - self.message = message - self.on_con_lost = on_con_lost - self.transport = None - - def error_received(self, exc): - logger.exception("Error received from Jaeger", exc_info=exc) - if not self.on_con_lost.done(): - self.on_con_lost.set_result(False) - - def connection_lost(self, exc): - logger.exception("Connection lost with Jaeger", exc_info=exc) - if not self.on_con_lost.done(): - self.on_con_lost.set_result(True) - - def connection_made(self, transport): - self.transport = transport - self.transport.sendto(self.message) - - -class AgentClientUDPAsync: - """Implement a UDP client to agent. - - Args: - host_name: The host name of the Jaeger server. - port: The port of the Jaeger server. - max_packet_size: Maximum size of UDP packet. - client: Class for creating new client objects for agencies. - split_oversized_batches: Re-emit oversized batches in smaller chunks. - """ - - def __init__( - self, - host_name, - port, - max_packet_size=UDP_PACKET_MAX_LENGTH, - client=Agent.Client, - split_oversized_batches=False, - ): - self.host_name = host_name - self.port = port - self.max_packet_size = max_packet_size - self.buffer = TTransport.TMemoryBuffer() - self.client = client(iprot=TCompactProtocol.TCompactProtocol(trans=self.buffer)) - self.split_oversized_batches = split_oversized_batches - self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - self._sock.setblocking(False) - self._addr = (host_name, int(port)) - - async def emit(self, batch: Collector.Batch): - """ - Args: - batch: Object to emit Jaeger spans. - """ - - # pylint: disable=protected-access - self.client._seqid = 0 - # truncate and reset the position of BytesIO object - self.buffer._buffer.truncate(0) - self.buffer._buffer.seek(0) - self.client.emitBatch(batch) - buff = self.buffer.getvalue() - if len(buff) > self.max_packet_size: - if self.split_oversized_batches and batch.spans and len(batch.spans) > 1: - packets = math.ceil(len(buff) / self.max_packet_size) - div = math.ceil(len(batch.spans) / packets) - for packet in range(packets): - start = packet * div - end = (packet + 1) * div - if start < len(batch.spans): - await self.emit( - Collector.Batch( - process=batch.process, - spans=batch.spans[start:end], - ) - ) - else: - logger.warning( - "Data exceeds the max UDP packet size; size %r, max %r", - len(buff), - self.max_packet_size, - ) - return - - loop = asyncio.get_running_loop() - on_con_lost = loop.create_future() - - send_to = partial(self._sendto, buff, on_con_lost) - loop.add_writer(self._sock.fileno(), send_to) - try: - await on_con_lost - except CancelledError: - logger.warning("Emission to Jaeger was cancelled") - except Exception: - logger.exception("Exception on sending to jaeger", stack_info=True) - finally: - loop.remove_writer(self._sock.fileno()) - - def _sendto(self, buff, on_con_lost: asyncio.Future): - try: - self._sock.sendto(buff, self._addr) - except (BlockingIOError, InterruptedError): - return - except OSError as exc: - on_con_lost.set_exception(exc) - except Exception as exc: - on_con_lost.set_exception(exc) - else: - on_con_lost.set_result(True) diff --git a/nucliadb_telemetry/src/nucliadb_telemetry/settings.py b/nucliadb_telemetry/src/nucliadb_telemetry/settings.py index 857caafa4f..a8f90a801a 100644 --- a/nucliadb_telemetry/src/nucliadb_telemetry/settings.py +++ b/nucliadb_telemetry/src/nucliadb_telemetry/settings.py @@ -20,12 +20,6 @@ class TelemetrySettings(BaseSettings): - jaeger_agent_host: str = "localhost" - jaeger_agent_port: int = 6831 - jaeger_enabled: bool = False - jaeger_query_port: int = 16686 - jaeger_query_host: str = "jaeger.observability.svc.cluster.local" - otlp_collector_endpoint: str | None = None observe_garbage_collector: bool = pydantic.Field( @@ -54,7 +48,7 @@ class TelemetrySettings(BaseSettings): ) def tracing_enabled(self) -> bool: - return self.jaeger_enabled or self.otlp_collector_endpoint is not None + return self.otlp_collector_endpoint is not None telemetry_settings = TelemetrySettings() diff --git a/nucliadb_telemetry/src/nucliadb_telemetry/tests/telemetry.py b/nucliadb_telemetry/src/nucliadb_telemetry/tests/telemetry.py index 0d337fdf29..a34daada81 100644 --- a/nucliadb_telemetry/src/nucliadb_telemetry/tests/telemetry.py +++ b/nucliadb_telemetry/src/nucliadb_telemetry/tests/telemetry.py @@ -96,19 +96,8 @@ async def jaeger_server(): @pytest.fixture(scope="function") async def set_telemetry_settings(jaeger_server: Jaeger): - if os.environ.get("TEST_JAEGER"): - telemetry_settings.jaeger_enabled = True - telemetry_settings.jaeger_agent_host = "127.0.0.1" - telemetry_settings.jaeger_agent_port = jaeger_server.get_port() - telemetry_settings.jaeger_query_port = jaeger_server.get_http_port() - print("Testing telemetry with Jaeger protocol (deprecated)") - else: - otlp_port = jaeger_server.get_port("4317/tcp") - telemetry_settings.otlp_collector_endpoint = f"127.0.0.1:{otlp_port}" - - telemetry_settings.jaeger_agent_host = "127.0.0.1" - telemetry_settings.jaeger_query_port = jaeger_server.get_http_port() - print("Testing telemetry with OTLP protocol") + otlp_port = jaeger_server.get_port("4317/tcp") + telemetry_settings.otlp_collector_endpoint = f"127.0.0.1:{otlp_port}" @pytest.fixture(scope="function") diff --git a/nucliadb_telemetry/src/nucliadb_telemetry/tracerprovider.py b/nucliadb_telemetry/src/nucliadb_telemetry/tracerprovider.py deleted file mode 100644 index 7a744dd25e..0000000000 --- a/nucliadb_telemetry/src/nucliadb_telemetry/tracerprovider.py +++ /dev/null @@ -1,93 +0,0 @@ -# Copyright 2025 Bosutech XXI S.L. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -import asyncio -import time - -from opentelemetry.context import Context -from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor, TracerProvider - - -class AsyncMultiSpanProcessor(SpanProcessor): - """Implementation of class:`SpanProcessor` that forwards all received - events to a list of span processors sequentially. - - The underlying span processors are called in sequential order as they were - added. - """ - - def __init__(self): - # use a tuple to avoid race conditions when adding a new span and - # iterating through it on "on_start" and "on_end". - self._span_processors = () - self._lock = asyncio.Lock() - - async def async_add_span_processor(self, span_processor: SpanProcessor) -> None: - """Adds a SpanProcessor to the list handled by this instance.""" - async with self._lock: - self._span_processors += (span_processor,) - - def on_start( - self, - span: Span, - parent_context: Context | None = None, - ) -> None: - for sp in self._span_processors: - sp.on_start(span, parent_context=parent_context) - - def on_end(self, span: ReadableSpan) -> None: - for sp in self._span_processors: - sp.on_end(span) - - def shutdown(self) -> None: - """Sequentially shuts down all underlying span processors.""" - for sp in self._span_processors: - sp.shutdown() - - async def async_force_flush(self, timeout_millis: int = 30000) -> bool: - """Sequentially calls async_force_flush on all underlying - :class:`SpanProcessor` - - Args: - timeout_millis: The maximum amount of time over all span processors - to wait for spans to be exported. In case the first n span - processors exceeded the timeout followup span processors will be - skipped. - - Returns: - True if all span processors flushed their spans within the - given timeout, False otherwise. - """ - deadline_ns = time.perf_counter_ns() + timeout_millis * 1000000 - for sp in self._span_processors: - current_time_ns = time.perf_counter_ns() - if current_time_ns >= deadline_ns: - return False - - if not await sp.async_force_flush((deadline_ns - current_time_ns) // 1000000): - return False - - return True - - -class AsyncTracerProvider(TracerProvider): - initialized: bool = False - _active_span_processor: AsyncMultiSpanProcessor # type: ignore[assignment] - - async def async_add_span_processor(self, span_processor: SpanProcessor) -> None: - await self._active_span_processor.async_add_span_processor(span_processor) - - async def async_force_flush(self, timeout_millis: int = 30000) -> bool: - return await self._active_span_processor.async_force_flush(timeout_millis) diff --git a/nucliadb_telemetry/src/nucliadb_telemetry/utils.py b/nucliadb_telemetry/src/nucliadb_telemetry/utils.py index 84f2798bee..ececd1ebc3 100644 --- a/nucliadb_telemetry/src/nucliadb_telemetry/utils.py +++ b/nucliadb_telemetry/src/nucliadb_telemetry/utils.py @@ -13,10 +13,6 @@ # limitations under the License. # -import asyncio -import os -from typing import cast - from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.propagate import set_global_textmap from opentelemetry.propagators.b3 import B3MultiFormat @@ -24,22 +20,16 @@ from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor -import nucliadb_telemetry.batch_span -from nucliadb_telemetry.jaeger import JaegerExporterAsync from nucliadb_telemetry.settings import telemetry_settings -from nucliadb_telemetry.tracerprovider import ( - AsyncMultiSpanProcessor, - AsyncTracerProvider, -) from .context import set_info_on_span set_info_on_span # b/w compatible import -GLOBAL_PROVIDER: dict[str, AsyncTracerProvider | TracerProvider] = {} +GLOBAL_PROVIDER: dict[str, TracerProvider] = {} -def get_telemetry(service_name: str | None = None) -> AsyncTracerProvider | TracerProvider | None: +def get_telemetry(service_name: str | None = None) -> TracerProvider | None: if service_name is None: return None if service_name not in GLOBAL_PROVIDER and service_name is not None: @@ -50,19 +40,13 @@ def get_telemetry(service_name: str | None = None) -> AsyncTracerProvider | Trac return GLOBAL_PROVIDER.get(service_name) -def create_telemetry(service_name: str) -> AsyncTracerProvider | TracerProvider | None: +def create_telemetry(service_name: str) -> TracerProvider | None: if not telemetry_settings.tracing_enabled(): return None resource = Resource.create({SERVICE_NAME: service_name}) - if telemetry_settings.otlp_collector_endpoint is not None: - tracer_provider = TracerProvider(resource=resource) - tracer_provider.initialized = False # type: ignore - else: - tracer_provider = AsyncTracerProvider( - active_span_processor=AsyncMultiSpanProcessor(), # type: ignore - resource=resource, - ) + tracer_provider = TracerProvider(resource=resource) + tracer_provider.initialized = False # type: ignore return tracer_provider @@ -71,18 +55,13 @@ async def clean_telemetry(service_name: str): if service_name in GLOBAL_PROVIDER and service_name: tracer_provider = GLOBAL_PROVIDER[service_name] - if isinstance(tracer_provider, AsyncTracerProvider): - await tracer_provider.async_force_flush() - # Without this sleep, async_force_flush fails on exporting pending spans - await asyncio.sleep(0) - else: - tracer_provider.force_flush() + tracer_provider.force_flush() tracer_provider.shutdown() del GLOBAL_PROVIDER[service_name] -async def init_telemetry(tracer_provider: AsyncTracerProvider | TracerProvider | None = None): +async def init_telemetry(tracer_provider: TracerProvider | None = None): if tracer_provider is None: return @@ -94,36 +73,11 @@ async def init_telemetry(tracer_provider: AsyncTracerProvider | TracerProvider | exporter = OTLPSpanExporter(endpoint=telemetry_settings.otlp_collector_endpoint, insecure=True) span_processor = BatchSpanProcessor(exporter) tracer_provider.add_span_processor(span_processor) - else: - # Fallback to create a JaegerExporter - jaeger_exporter = JaegerExporterAsync( - # configure agent - agent_host_name=telemetry_settings.jaeger_agent_host, - agent_port=telemetry_settings.jaeger_agent_port, - # optional: configure also collector - # collector_endpoint='http://localhost:14268/api/traces?format=jaeger.thrift', - # username=xxxx, # optional - # password=xxxx, # optional - # max_tag_value_length=None # optional - ) - schedule_delay_millis = int(os.environ.get("OTEL_BSP_SCHEDULE_DELAY", 5000)) - max_queue_size = int(os.environ.get("OTEL_BSP_MAX_QUEUE_SIZE", 2048)) - max_export_batch_size = int(os.environ.get("OTEL_BSP_MAX_EXPORT_BATCH_SIZE", 512)) - export_timeout_millis = int(os.environ.get("OTEL_BSP_EXPORT_TIMEOUT", 30000)) - jaeger_span_processor = nucliadb_telemetry.batch_span.BatchSpanProcessor( - jaeger_exporter, - schedule_delay_millis=schedule_delay_millis, - max_queue_size=max_queue_size, - max_export_batch_size=max_export_batch_size, - export_timeout_millis=export_timeout_millis, - ) - tracer_provider = cast(AsyncTracerProvider, tracer_provider) - await tracer_provider.async_add_span_processor(jaeger_span_processor) tracer_provider.initialized = True # type: ignore -async def setup_telemetry(service_name: str) -> AsyncTracerProvider | TracerProvider | None: +async def setup_telemetry(service_name: str) -> TracerProvider | None: """ Setup telemetry for a service if it is enabled """ diff --git a/nucliadb_telemetry/tests/integration/test_telemetry.py b/nucliadb_telemetry/tests/integration/test_telemetry.py index a11fbf54cf..d06eb0e975 100644 --- a/nucliadb_telemetry/tests/integration/test_telemetry.py +++ b/nucliadb_telemetry/tests/integration/test_telemetry.py @@ -20,7 +20,6 @@ from nucliadb_telemetry import grpc_metrics from nucliadb_telemetry.jetstream import msg_consume_time_histo -from nucliadb_telemetry.settings import telemetry_settings from nucliadb_telemetry.tests.telemetry import Greeter @@ -44,7 +43,7 @@ def debug_spans(spans): print(json.dumps(spans, indent=4)) -async def test_telemetry_dict(http_service: AsyncClient, greeter: Greeter): +async def test_telemetry_dict(http_service: AsyncClient, greeter: Greeter, jaeger_server): resp = await http_service.get( "http://test/", headers={ @@ -72,7 +71,7 @@ async def test_telemetry_dict(http_service: AsyncClient, greeter: Greeter): client = AsyncClient() for _ in range(10): resp = await client.get( - f"http://localhost:{telemetry_settings.jaeger_query_port}/api/traces/f13dc5318bf3bef64a0a5ea607db93a1", + f"http://localhost:{jaeger_server.get_http_port()}/api/traces/f13dc5318bf3bef64a0a5ea607db93a1", headers={"Accept": "application/json"}, ) if resp.status_code != 200 or len(resp.json()["data"][0]["spans"]) < expected_spans: diff --git a/uv.lock b/uv.lock index fd0d927d3a..dc8d7004e9 100644 --- a/uv.lock +++ b/uv.lock @@ -2249,7 +2249,6 @@ all = [ { name = "grpcio-tools" }, { name = "nats-py", extra = ["nkeys"] }, { name = "opentelemetry-api" }, - { name = "opentelemetry-exporter-jaeger-thrift" }, { name = "opentelemetry-exporter-otlp-proto-grpc" }, { name = "opentelemetry-instrumentation-aiohttp-client" }, { name = "opentelemetry-instrumentation-fastapi" }, @@ -2263,7 +2262,6 @@ all = [ fastapi = [ { name = "fastapi" }, { name = "opentelemetry-api" }, - { name = "opentelemetry-exporter-jaeger-thrift" }, { name = "opentelemetry-exporter-otlp-proto-grpc" }, { name = "opentelemetry-instrumentation-aiohttp-client" }, { name = "opentelemetry-instrumentation-fastapi" }, @@ -2282,7 +2280,6 @@ grpc = [ { name = "grpcio-testing" }, { name = "grpcio-tools" }, { name = "opentelemetry-api" }, - { name = "opentelemetry-exporter-jaeger-thrift" }, { name = "opentelemetry-exporter-otlp-proto-grpc" }, { name = "opentelemetry-instrumentation-aiohttp-client" }, { name = "opentelemetry-instrumentation-fastapi" }, @@ -2295,7 +2292,6 @@ grpc = [ nats = [ { name = "nats-py", extra = ["nkeys"] }, { name = "opentelemetry-api" }, - { name = "opentelemetry-exporter-jaeger-thrift" }, { name = "opentelemetry-exporter-otlp-proto-grpc" }, { name = "opentelemetry-instrumentation-aiohttp-client" }, { name = "opentelemetry-instrumentation-fastapi" }, @@ -2308,7 +2304,6 @@ nats = [ ] otel = [ { name = "opentelemetry-api" }, - { name = "opentelemetry-exporter-jaeger-thrift" }, { name = "opentelemetry-exporter-otlp-proto-grpc" }, { name = "opentelemetry-instrumentation-aiohttp-client" }, { name = "opentelemetry-instrumentation-fastapi" }, @@ -2339,16 +2334,11 @@ requires-dist = [ { name = "grpcio-tools", marker = "extra == 'grpc'", specifier = ">=1.71.0" }, { name = "nats-py", extras = ["nkeys"], marker = "extra == 'all'", specifier = ">=2.5.0" }, { name = "nats-py", extras = ["nkeys"], marker = "extra == 'nats'", specifier = ">=2.5.0" }, - { name = "opentelemetry-api", marker = "extra == 'all'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-api", marker = "extra == 'fastapi'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-api", marker = "extra == 'grpc'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-api", marker = "extra == 'nats'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-api", marker = "extra == 'otel'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-exporter-jaeger-thrift", marker = "extra == 'all'", specifier = ">=1.21.0" }, - { name = "opentelemetry-exporter-jaeger-thrift", marker = "extra == 'fastapi'", specifier = ">=1.21.0" }, - { name = "opentelemetry-exporter-jaeger-thrift", marker = "extra == 'grpc'", specifier = ">=1.21.0" }, - { name = "opentelemetry-exporter-jaeger-thrift", marker = "extra == 'nats'", specifier = ">=1.21.0" }, - { name = "opentelemetry-exporter-jaeger-thrift", marker = "extra == 'otel'", specifier = ">=1.21.0" }, + { name = "opentelemetry-api", marker = "extra == 'all'", specifier = ">=1.21.0" }, + { name = "opentelemetry-api", marker = "extra == 'fastapi'", specifier = ">=1.21.0" }, + { name = "opentelemetry-api", marker = "extra == 'grpc'", specifier = ">=1.21.0" }, + { name = "opentelemetry-api", marker = "extra == 'nats'", specifier = ">=1.21.0" }, + { name = "opentelemetry-api", marker = "extra == 'otel'", specifier = ">=1.21.0" }, { name = "opentelemetry-exporter-otlp-proto-grpc", marker = "extra == 'all'", specifier = ">=1.21.0" }, { name = "opentelemetry-exporter-otlp-proto-grpc", marker = "extra == 'fastapi'", specifier = ">=1.21.0" }, { name = "opentelemetry-exporter-otlp-proto-grpc", marker = "extra == 'grpc'", specifier = ">=1.21.0" }, @@ -2374,16 +2364,16 @@ requires-dist = [ { name = "opentelemetry-propagator-b3", marker = "extra == 'grpc'", specifier = ">=1.21.0" }, { name = "opentelemetry-propagator-b3", marker = "extra == 'nats'", specifier = ">=1.21.0" }, { name = "opentelemetry-propagator-b3", marker = "extra == 'otel'", specifier = ">=1.21.0" }, - { name = "opentelemetry-proto", marker = "extra == 'all'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-proto", marker = "extra == 'fastapi'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-proto", marker = "extra == 'grpc'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-proto", marker = "extra == 'nats'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-proto", marker = "extra == 'otel'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-sdk", marker = "extra == 'all'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-sdk", marker = "extra == 'fastapi'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-sdk", marker = "extra == 'grpc'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-sdk", marker = "extra == 'nats'", specifier = ">=1.21.0,<1.40" }, - { name = "opentelemetry-sdk", marker = "extra == 'otel'", specifier = ">=1.21.0,<1.40" }, + { name = "opentelemetry-proto", marker = "extra == 'all'", specifier = ">=1.21.0" }, + { name = "opentelemetry-proto", marker = "extra == 'fastapi'", specifier = ">=1.21.0" }, + { name = "opentelemetry-proto", marker = "extra == 'grpc'", specifier = ">=1.21.0" }, + { name = "opentelemetry-proto", marker = "extra == 'nats'", specifier = ">=1.21.0" }, + { name = "opentelemetry-proto", marker = "extra == 'otel'", specifier = ">=1.21.0" }, + { name = "opentelemetry-sdk", marker = "extra == 'all'", specifier = ">=1.21.0" }, + { name = "opentelemetry-sdk", marker = "extra == 'fastapi'", specifier = ">=1.21.0" }, + { name = "opentelemetry-sdk", marker = "extra == 'grpc'", specifier = ">=1.21.0" }, + { name = "opentelemetry-sdk", marker = "extra == 'nats'", specifier = ">=1.21.0" }, + { name = "opentelemetry-sdk", marker = "extra == 'otel'", specifier = ">=1.21.0" }, { name = "opentelemetry-semantic-conventions", marker = "extra == 'all'", specifier = ">=0.42b0" }, { name = "opentelemetry-semantic-conventions", marker = "extra == 'fastapi'", specifier = ">=0.42b0" }, { name = "opentelemetry-semantic-conventions", marker = "extra == 'grpc'", specifier = ">=0.42b0" }, @@ -2597,20 +2587,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ae/a2/d86e01c28300bd41bab8f18afd613676e2bd63515417b77636fc1add426f/opentelemetry_api-1.38.0-py3-none-any.whl", hash = "sha256:2891b0197f47124454ab9f0cf58f3be33faca394457ac3e09daba13ff50aa582", size = 65947, upload-time = "2025-10-16T08:35:30.23Z" }, ] -[[package]] -name = "opentelemetry-exporter-jaeger-thrift" -version = "1.21.0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "opentelemetry-api" }, - { name = "opentelemetry-sdk" }, - { name = "thrift" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/c7/ff/560d975ce09ec123dce1b7a97877bad1a5b2424c5c8e31e55169d4cd85e6/opentelemetry_exporter_jaeger_thrift-1.21.0.tar.gz", hash = "sha256:41119bc7e5602cec83dd7d7060f061ecbc91de231272e8f515b07ef9a4b6e41c", size = 27722, upload-time = "2023-11-07T23:16:26.522Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/81/2b/c6ff9604f5b8469e1fabe2b6ee50c36dadf0f572a430494d8dd5ce37d99f/opentelemetry_exporter_jaeger_thrift-1.21.0-py3-none-any.whl", hash = "sha256:4364b8dfa6965707c72c43d85942b1491982b7d44f0123d593513e8bedafa9e2", size = 35929, upload-time = "2023-11-07T23:15:52.341Z" }, -] - [[package]] name = "opentelemetry-exporter-otlp-proto-common" version = "1.38.0" @@ -4164,15 +4140,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/51/da/545b75d420bb23b5d494b0517757b351963e974e79933f01e05c929f20a6/starlette-0.49.1-py3-none-any.whl", hash = "sha256:d92ce9f07e4a3caa3ac13a79523bd18e3bc0042bb8ff2d759a8e7dd0e1859875", size = 74175, upload-time = "2025-10-28T17:34:09.13Z" }, ] -[[package]] -name = "thrift" -version = "0.21.0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "six" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/33/1c/418058b4750176b638ab60b4d5a554a2969dcd2363ae458519d0f747adff/thrift-0.21.0.tar.gz", hash = "sha256:5e6f7c50f936ebfa23e924229afc95eb219f8c8e5a83202dd4a391244803e402", size = 62509, upload-time = "2024-09-21T13:14:19.273Z" } - [[package]] name = "tomli" version = "2.2.1"