diff --git a/src/pipecat/utils/tracing/class_decorators.py b/src/pipecat/utils/tracing/class_decorators.py deleted file mode 100644 index 73dacbe51..000000000 --- a/src/pipecat/utils/tracing/class_decorators.py +++ /dev/null @@ -1,257 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# Portions Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Base OpenTelemetry tracing decorators and utilities for Pipecat. - -.. deprecated:: 0.0.103 - This module is unused and will be removed in a future release. - Service tracing is handled by the decorators in - :mod:`pipecat.utils.tracing.service_decorators`. - -This module provides class and method level tracing capabilities -similar to the original NVIDIA implementation. -""" - -import asyncio -import contextlib -import enum -import functools -import inspect -import warnings -from typing import Callable, Optional, TypeVar - -warnings.warn( - "pipecat.utils.tracing.class_decorators is deprecated and will be removed in a future " - "release. Use pipecat.utils.tracing.service_decorators instead.", - DeprecationWarning, - stacklevel=2, -) - -from pipecat.utils.tracing.setup import is_tracing_available - -# Import OpenTelemetry if available -if is_tracing_available(): - import opentelemetry.trace - from opentelemetry import metrics, trace - -# Type variables for better typing support -T = TypeVar("T") -C = TypeVar("C", bound=type) - - -class AttachmentStrategy(enum.Enum): - """Controls how spans are attached to the trace hierarchy. - - Parameters: - CHILD: Attached to class span if no parent, otherwise to parent. - LINK: Attached to class span with link to parent. - NONE: Always attached to class span regardless of context. - """ - - CHILD = enum.auto() - LINK = enum.auto() - NONE = enum.auto() - - -class Traceable: - """Base class for objects that can be traced with OpenTelemetry. - - Provides the foundational tracing capabilities used by @traced methods. - """ - - def __init__(self, name: str, **kwargs): - """Initialize a traceable object. - - Args: - name: Name of the traceable object for the span. - **kwargs: Additional arguments passed to parent class. - """ - super().__init__(**kwargs) - - if not is_tracing_available(): - self._tracer = self._meter = self._parent_span_id = self._span = None - return - - self._tracer = trace.get_tracer("pipecat") - self._meter = metrics.get_meter("pipecat") - self._parent_span_id = trace.get_current_span().get_span_context().span_id - self._span = self._tracer.start_span(name) - self._span.end() - - @property - def meter(self): - """Get the OpenTelemetry meter instance. - - Returns: - The OpenTelemetry meter instance for this object. - """ - return self._meter - - -@contextlib.contextmanager -def __traced_context_manager( - self: Traceable, func: Callable, name: str | None, attachment_strategy: AttachmentStrategy -): - """Internal context manager for the traced decorator. - - Args: - self: The Traceable instance. - func: The function being traced. - name: Custom span name or None to use function name. - attachment_strategy: How to attach this span to the trace hierarchy. - - Raises: - RuntimeError: If used in a class not inheriting from Traceable. - """ - if not isinstance(self, Traceable): - raise RuntimeError( - "@traced annotation can only be used in classes inheriting from Traceable" - ) - - stack = contextlib.ExitStack() - try: - current_span = trace.get_current_span() - is_span_class_parent_span = current_span.get_span_context().span_id == self._parent_span_id - match attachment_strategy: - case AttachmentStrategy.CHILD if not is_span_class_parent_span: - stack.enter_context( - self._tracer.start_as_current_span(func.__name__ if name is None else name) # type: ignore - ) - case AttachmentStrategy.LINK: - if is_span_class_parent_span: - link = trace.Link(self._span.get_span_context()) # type: ignore - else: - link = trace.Link(current_span.get_span_context()) - stack.enter_context( - opentelemetry.trace.use_span(span=self._span, end_on_exit=False) # type: ignore - ) - stack.enter_context( - self._tracer.start_as_current_span( # type: ignore - func.__name__ if name is None else name, links=[link] - ) - ) - case AttachmentStrategy.NONE | AttachmentStrategy.CHILD: - stack.enter_context( - opentelemetry.trace.use_span(span=self._span, end_on_exit=False) # type: ignore - ) - stack.enter_context( - self._tracer.start_as_current_span(func.__name__ if name is None else name) # type: ignore - ) - yield - finally: - stack.close() - - -def __traced_decorator(func, name, attachment_strategy: AttachmentStrategy): - """Implementation of the traced decorator. - - Args: - func: The function to trace. - name: Custom span name. - attachment_strategy: How to attach this span. - - Returns: - The wrapped function with tracing capabilities. - """ - - @functools.wraps(func) - async def coroutine_wrapper(self: Traceable, *args, **kwargs): - exception = None - with __traced_context_manager(self, func, name, attachment_strategy): - try: - return await func(self, *args, **kwargs) - except asyncio.CancelledError as e: - exception = e - if exception: - raise exception - - @functools.wraps(func) - async def generator_wrapper(self: Traceable, *args, **kwargs): - exception = None - with __traced_context_manager(self, func, name, attachment_strategy): - try: - async for v in func(self, *args, **kwargs): - yield v - except asyncio.CancelledError as e: - exception = e - if exception: - raise exception - - if inspect.iscoroutinefunction(func): - return coroutine_wrapper - if inspect.isasyncgenfunction(func): - return generator_wrapper - - raise ValueError("@traced annotation can only be used on async or async generator functions") - - -def traced( - func: Optional[Callable] = None, - *, - name: Optional[str] = None, - attachment_strategy: AttachmentStrategy = AttachmentStrategy.CHILD, -) -> Callable: - """Add tracing to an async function in a Traceable class. - - Args: - func: The async function to trace. - name: Custom span name. Defaults to function name. - attachment_strategy: How to attach this span (CHILD, LINK, NONE). - - Returns: - Wrapped async function with tracing. - - Raises: - RuntimeError: If used in a class not inheriting from Traceable. - ValueError: If used on a non-async function. - """ - if not is_tracing_available(): - # Just return the original function or a simple decorator - def decorator(f): - return f - - return decorator if func is None else func - - if func is not None: - return __traced_decorator(func, name=name, attachment_strategy=attachment_strategy) - else: - return functools.partial( - __traced_decorator, name=name, attachment_strategy=attachment_strategy - ) - - -def traceable(cls: C) -> C: - """Make a class traceable for OpenTelemetry. - - Creates a new class that inherits from both the original class - and Traceable, enabling tracing for class methods. - - Args: - cls: The class to make traceable. - - Returns: - A new class with tracing capabilities. - """ - if not is_tracing_available(): - return cls - - @functools.wraps(cls, updated=()) - class TracedClass(cls, Traceable): - def __init__(self, *args, **kwargs): - """Initialize the traced class instance. - - Args: - *args: Positional arguments passed to parent classes. - **kwargs: Keyword arguments passed to parent classes. - """ - cls.__init__(self, *args, **kwargs) - if hasattr(self, "name"): - Traceable.__init__(self, self.name) - else: - Traceable.__init__(self, cls.__name__) - - return TracedClass diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index d87955755..256626e34 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -100,11 +100,6 @@ def _get_parent_service_context(self): if not is_tracing_available(): return None - # TODO: Remove this block and delete class_decorators.py once Traceable is removed. - # Legacy: support for classes inheriting from Traceable (currently unused, deprecated). - if hasattr(self, "_span") and self._span: - return trace.set_span_in_context(self._span) - # Use the conversation context set by TurnTraceObserver via TracingContext. tracing_ctx = getattr(self, "_tracing_context", None) conversation_context = tracing_ctx.get_conversation_context() if tracing_ctx else None