File size: 2,183 Bytes
f3b3e7a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
"""
Telemetry stub for HuggingFace deployment with minimal dependencies.
Provides mock implementations when full OpenTelemetry is not available.
"""
import functools
from typing import Any, Callable, Optional
class MockTracer:
"""Mock tracer for when OpenTelemetry is not available."""
def start_as_current_span(self, name: str, **kwargs):
"""Mock span context manager."""
class MockSpan:
def __enter__(self):
return self
def __exit__(self, *args):
pass
def set_attribute(self, key: str, value: Any):
pass
def set_status(self, status: Any):
pass
return MockSpan()
class MockMeter:
"""Mock meter for when OpenTelemetry is not available."""
def create_counter(self, name: str, **kwargs):
"""Mock counter."""
class MockCounter:
def add(self, amount: int = 1, attributes: Optional[dict] = None):
pass
return MockCounter()
def create_histogram(self, name: str, **kwargs):
"""Mock histogram."""
class MockHistogram:
def record(self, amount: float, attributes: Optional[dict] = None):
pass
return MockHistogram()
def create_up_down_counter(self, name: str, **kwargs):
"""Mock up-down counter."""
class MockUpDownCounter:
def add(self, amount: int = 1, attributes: Optional[dict] = None):
pass
return MockUpDownCounter()
# Global mock instances
mock_tracer = MockTracer()
mock_meter = MockMeter()
def trace_method(span_name: Optional[str] = None) -> Callable:
"""Decorator for tracing methods - no-op version."""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
async def async_wrapper(*args, **kwargs):
return await func(*args, **kwargs)
@functools.wraps(func)
def sync_wrapper(*args, **kwargs):
return func(*args, **kwargs)
return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
return decorator
import asyncio |