tessl/pypi-opentelemetry-api

OpenTelemetry Python API providing core abstractions for distributed tracing, metrics collection, and context propagation

—

Pending

Overview

Eval results

Files

Logging and Events

Name: tessl/pypi-opentelemetry-api
Author: tessl

Comprehensive structured logging and event recording capabilities with OpenTelemetry integration. The logging API provides severity-based log record emission with trace correlation, while the events API enables structured event recording with rich attributes and context integration.

Capabilities

Logger Creation and Management

Get loggers for structured log record emission with proper instrumentation scope identification.

def get_logger(
    name: str,
    instrumenting_library_version: Optional[str] = None,
    logger_provider: Optional[LoggerProvider] = None,
    schema_url: Optional[str] = None,
    attributes: Optional[Attributes] = None,
) -> Logger:
    """
    Returns a Logger for use by the given instrumentation library.
    
    Parameters:
    - name: The name of the instrumentation scope
    - instrumenting_library_version: Optional version of the instrumentation library
    - logger_provider: Optional specific LoggerProvider to use
    - schema_url: Optional Schema URL of the emitted telemetry
    - attributes: Optional attributes of the emitted telemetry
    
    Returns:
    Logger instance for emitting log records
    """

def get_logger_provider() -> LoggerProvider:
    """Gets the current global LoggerProvider object."""

def set_logger_provider(logger_provider: LoggerProvider) -> None:
    """
    Sets the current global LoggerProvider object.
    This can only be done once, a warning will be logged if any further attempt is made.
    """

Log Record Structure and Emission

Create and emit structured log records with severity levels and rich attributes.

class Logger(ABC):
    """Abstract base class for loggers."""
    
    def emit(self, log_record: LogRecord) -> None:
        """
        Emit a LogRecord.
        
        Parameters:
        - log_record: The LogRecord to emit
        """

class LogRecord:
    """A LogRecord instance represents an occurrence of an event."""
    
    def __init__(
        self,
        timestamp: Optional[int] = None,
        observed_timestamp: Optional[int] = None,
        trace_id: Optional[int] = None,
        span_id: Optional[int] = None,
        trace_flags: Optional[TraceFlags] = None,
        severity_text: Optional[str] = None,
        severity_number: Optional[SeverityNumber] = None,
        body: Optional[Any] = None,
        resource: Optional[Resource] = None,
        attributes: Optional[Attributes] = None,
    ) -> None:
        """
        Create a new LogRecord.
        
        Parameters:
        - timestamp: Time when the log record occurred
        - observed_timestamp: Time when the log record was observed
        - trace_id: Trace ID for correlation
        - span_id: Span ID for correlation
        - trace_flags: Trace flags for correlation
        - severity_text: Severity level as text
        - severity_number: Severity level as number
        - body: The log record body/message
        - resource: Resource information
        - attributes: Additional attributes
        """
    
    @property
    def timestamp(self) -> Optional[int]:
        """Returns the timestamp of the log record."""
    
    @property
    def observed_timestamp(self) -> Optional[int]:
        """Returns the observed timestamp of the log record."""
    
    @property
    def trace_id(self) -> Optional[int]:
        """Returns the trace ID for correlation."""
    
    @property  
    def span_id(self) -> Optional[int]:
        """Returns the span ID for correlation."""
    
    @property
    def trace_flags(self) -> Optional[TraceFlags]:
        """Returns the trace flags for correlation."""
    
    @property
    def severity_text(self) -> Optional[str]:
        """Returns the severity text."""
    
    @property
    def severity_number(self) -> Optional[SeverityNumber]:
        """Returns the severity number."""
    
    @property
    def body(self) -> Optional[Any]:
        """Returns the log record body."""
    
    @property
    def attributes(self) -> Optional[Attributes]:
        """Returns the log record attributes."""

Severity Levels

Standard severity levels for log records following OpenTelemetry specification.

class SeverityNumber(Enum):
    """Severity levels for log records."""
    
    TRACE = 1      # Trace level (most verbose)
    TRACE2 = 2
    TRACE3 = 3
    TRACE4 = 4
    DEBUG = 5      # Debug level
    DEBUG2 = 6
    DEBUG3 = 7
    DEBUG4 = 8
    INFO = 9       # Info level
    INFO2 = 10
    INFO3 = 11
    INFO4 = 12
    WARN = 13      # Warning level
    WARN2 = 14
    WARN3 = 15
    WARN4 = 16
    ERROR = 17     # Error level
    ERROR2 = 18
    ERROR3 = 19
    ERROR4 = 20
    FATAL = 21     # Fatal level (least verbose)
    FATAL2 = 22
    FATAL3 = 23
    FATAL4 = 24

Logger Provider Implementations

Provider classes for different deployment scenarios.

class LoggerProvider(ABC):
    """Abstract base class for logger providers."""
    
    def get_logger(
        self,
        name: str,
        instrumenting_library_version: Optional[str] = None,
        schema_url: Optional[str] = None,
        attributes: Optional[Attributes] = None,
    ) -> Logger:
        """Returns a Logger for use by the given instrumentation library."""

class NoOpLoggerProvider(LoggerProvider):
    """The default LoggerProvider, used when no implementation is available."""
    
    def get_logger(
        self,
        name: str,
        instrumenting_library_version: Optional[str] = None,
        schema_url: Optional[str] = None,
        attributes: Optional[Attributes] = None,
    ) -> Logger:
        """Returns a no-op logger."""

class NoOpLogger(Logger):
    """No-op implementation of Logger."""
    
    def emit(self, log_record: LogRecord) -> None:
        """No-op emit implementation."""

Event Logger System

Structured event recording with enhanced semantics and type safety.

def get_event_logger_provider() -> EventLoggerProvider:
    """Gets the current global EventLoggerProvider object."""

def set_event_logger_provider(event_logger_provider: EventLoggerProvider) -> None:
    """
    Sets the current global EventLoggerProvider object.
    This can only be done once, a warning will be logged if any further attempt is made.
    """

def get_event_logger(
    name: str,
    instrumenting_library_version: Optional[str] = None,
    event_logger_provider: Optional[EventLoggerProvider] = None,
    schema_url: Optional[str] = None,
    attributes: Optional[Attributes] = None,
) -> EventLogger:
    """
    Returns an EventLogger for use by the given instrumentation library.
    
    Parameters:
    - name: The name of the instrumentation scope
    - instrumenting_library_version: Optional version of the instrumentation library
    - event_logger_provider: Optional specific EventLoggerProvider to use
    - schema_url: Optional Schema URL of the emitted telemetry
    - attributes: Optional attributes of the emitted telemetry
    
    Returns:
    EventLogger instance for emitting events
    """

Event Structure and Operations

Create and emit structured events with rich semantic information.

class EventLogger(ABC):
    """Abstract base class for event loggers."""
    
    def emit(self, event: Event) -> None:
        """
        Emit an Event.
        
        Parameters:
        - event: The Event to emit
        """

class Event(LogRecord):
    """An Event is a specialized LogRecord with additional semantics."""
    
    def __init__(
        self,
        name: str,
        timestamp: Optional[int] = None,
        attributes: Optional[Attributes] = None,
        **kwargs,
    ) -> None:
        """
        Create a new Event.
        
        Parameters:
        - name: The name of the event
        - timestamp: Time when the event occurred
        - attributes: Event attributes
        - kwargs: Additional LogRecord parameters
        """
    
    @property
    def name(self) -> str:
        """Returns the event name."""

Event Logger Provider Implementations

Provider classes for event logger creation and management.

class EventLoggerProvider(ABC):
    """Abstract base class for event logger providers."""
    
    def get_event_logger(
        self,
        name: str,
        instrumenting_library_version: Optional[str] = None,
        schema_url: Optional[str] = None,
        attributes: Optional[Attributes] = None,
    ) -> EventLogger:
        """Returns an EventLogger for use by the given instrumentation library."""

class NoOpEventLoggerProvider(EventLoggerProvider):
    """The default EventLoggerProvider, used when no implementation is available."""
    
    def get_event_logger(
        self,
        name: str,
        instrumenting_library_version: Optional[str] = None,
        schema_url: Optional[str] = None,
        attributes: Optional[Attributes] = None,
    ) -> EventLogger:
        """Returns a no-op event logger."""

class ProxyEventLoggerProvider(EventLoggerProvider):
    """Proxy event logger provider for late binding of real providers."""
    
    def get_event_logger(
        self,
        name: str,
        instrumenting_library_version: Optional[str] = None,
        schema_url: Optional[str] = None,
        attributes: Optional[Attributes] = None,
    ) -> EventLogger:
        """Returns an event logger from the real provider or proxy event logger."""

class NoOpEventLogger(EventLogger):
    """No-op implementation of EventLogger."""
    
    def emit(self, event: Event) -> None:
        """No-op emit implementation."""

class ProxyEventLogger(EventLogger):
    """Proxy event logger for late binding of real event loggers."""
    
    def emit(self, event: Event) -> None:
        """Emit using real event logger or no-op."""

Usage Examples

Basic Structured Logging

from opentelemetry._logs import get_logger, LogRecord, SeverityNumber
from opentelemetry import trace
import time

# Get a logger
logger = get_logger(__name__)

# Create and emit log records
def log_user_action(user_id: str, action: str):
    # Get current trace context for correlation
    current_span = trace.get_current_span()
    span_context = current_span.get_span_context()
    
    log_record = LogRecord(
        timestamp=int(time.time() * 1_000_000_000),  # nanoseconds
        severity_text="INFO",
        severity_number=SeverityNumber.INFO,
        body=f"User {user_id} performed action: {action}",
        attributes={
            "user.id": user_id,
            "action": action,
            "service.name": "user-service",
        },
        trace_id=span_context.trace_id if span_context.is_valid else None,
        span_id=span_context.span_id if span_context.is_valid else None,
        trace_flags=span_context.trace_flags if span_context.is_valid else None,
    )
    
    logger.emit(log_record)

# Usage
log_user_action("12345", "login")

Event-Based Logging

from opentelemetry._events import get_event_logger, Event
from opentelemetry import trace, baggage
import time

# Get an event logger
event_logger = get_event_logger(__name__)

def record_purchase_event(user_id: str, product_id: str, amount: float):
    """Record a structured purchase event."""
    
    # Create event with rich attributes
    event = Event(
        name="user.purchase",
        timestamp=int(time.time() * 1_000_000_000),
        attributes={
            "user.id": user_id,
            "product.id": product_id,
            "purchase.amount": amount,
            "currency": "USD",
            "event.category": "commerce",
        }
    )
    
    # Emit the event
    event_logger.emit(event)

def record_system_event(event_name: str, **attributes):
    """Record a generic system event."""
    event = Event(
        name=event_name,
        attributes=attributes
    )
    event_logger.emit(event)

# Usage examples
record_purchase_event("12345", "prod-789", 29.99)
record_system_event("service.startup", service="user-service", version="1.2.3")

Logging with Trace Correlation

from opentelemetry import trace
from opentelemetry._logs import get_logger, LogRecord, SeverityNumber

logger = get_logger(__name__)
tracer = trace.get_tracer(__name__)

def process_order_with_logging(order_id: str):
    """Process an order with comprehensive logging."""
    
    with tracer.start_as_current_span("process-order") as span:
        span.set_attribute("order.id", order_id)
        
        try:
            # Log start of processing
            log_info("Order processing started", {
                "order.id": order_id,
                "stage": "start"
            })
            
            # Simulate processing steps
            validate_order(order_id)
            log_info("Order validated", {"order.id": order_id, "stage": "validation"})
            
            charge_payment(order_id)
            log_info("Payment charged", {"order.id": order_id, "stage": "payment"})
            
            ship_order(order_id)
            log_info("Order shipped", {"order.id": order_id, "stage": "shipping"})
            
            # Log successful completion
            log_info("Order processing completed", {
                "order.id": order_id,
                "stage": "complete",
                "result": "success"
            })
            
        except Exception as e:
            # Log error with exception details
            log_error(f"Order processing failed: {e}", {
                "order.id": order_id,
                "error.type": type(e).__name__,
                "error.message": str(e)
            })
            span.record_exception(e)
            raise

def log_info(message: str, attributes: dict = None):
    """Helper function for info logging with trace correlation."""
    current_span = trace.get_current_span()
    span_context = current_span.get_span_context()
    
    log_record = LogRecord(
        timestamp=int(time.time() * 1_000_000_000),
        severity_text="INFO",
        severity_number=SeverityNumber.INFO,
        body=message,
        attributes=attributes or {},
        trace_id=span_context.trace_id if span_context.is_valid else None,
        span_id=span_context.span_id if span_context.is_valid else None,
        trace_flags=span_context.trace_flags if span_context.is_valid else None,
    )
    
    logger.emit(log_record)

def log_error(message: str, attributes: dict = None):
    """Helper function for error logging with trace correlation."""
    current_span = trace.get_current_span()
    span_context = current_span.get_span_context()
    
    log_record = LogRecord(
        timestamp=int(time.time() * 1_000_000_000),
        severity_text="ERROR", 
        severity_number=SeverityNumber.ERROR,
        body=message,
        attributes=attributes or {},
        trace_id=span_context.trace_id if span_context.is_valid else None,
        span_id=span_context.span_id if span_context.is_valid else None,
        trace_flags=span_context.trace_flags if span_context.is_valid else None,
    )
    
    logger.emit(log_record)

Structured Logging with Baggage Context

from opentelemetry import baggage, context
from opentelemetry._logs import get_logger, LogRecord, SeverityNumber
import time

logger = get_logger(__name__)

def enhanced_logging_example():
    """Demonstrate logging with baggage context."""
    
    # Set baggage context
    ctx = baggage.set_baggage("user.id", "12345")
    ctx = baggage.set_baggage("request.id", "req-abc-123", ctx)
    ctx = baggage.set_baggage("service.version", "1.2.3", ctx)
    
    token = context.attach(ctx)
    
    try:
        # Log with baggage automatically included
        log_with_baggage("Processing user request", {
            "operation": "user.profile.update",
            "timestamp": time.time()
        })
        
        # Simulate some processing
        process_user_update()
        
        log_with_baggage("User request completed successfully", {
            "operation": "user.profile.update",
            "result": "success"
        })
        
    except Exception as e:
        log_with_baggage(f"User request failed: {e}", {
            "operation": "user.profile.update", 
            "result": "error",
            "error.type": type(e).__name__
        }, SeverityNumber.ERROR)
        
    finally:
        context.detach(token)

def log_with_baggage(message: str, attributes: dict = None, severity: SeverityNumber = SeverityNumber.INFO):
    """Log with automatic baggage inclusion."""
    
    # Get all baggage and include in attributes
    all_baggage = baggage.get_all()
    combined_attributes = dict(all_baggage)
    
    if attributes:
        combined_attributes.update(attributes)
    
    log_record = LogRecord(
        timestamp=int(time.time() * 1_000_000_000),
        severity_text=severity.name,
        severity_number=severity,
        body=message,
        attributes=combined_attributes,
    )
    
    logger.emit(log_record)

Custom Event Types

from opentelemetry._events import get_event_logger, Event
from opentelemetry import trace
import time
from typing import Dict, Any

event_logger = get_event_logger(__name__)

class SecurityEvent(Event):
    """Custom event type for security-related events."""
    
    def __init__(self, event_type: str, user_id: str, source_ip: str, **kwargs):
        super().__init__(
            name=f"security.{event_type}",
            attributes={
                "security.event_type": event_type,
                "user.id": user_id,
                "source.ip": source_ip,
                "event.category": "security",
                **kwargs
            }
        )

class BusinessEvent(Event):
    """Custom event type for business-related events."""
    
    def __init__(self, event_type: str, entity_id: str, **kwargs):
        super().__init__(
            name=f"business.{event_type}",
            attributes={
                "business.event_type": event_type,
                "entity.id": entity_id,
                "event.category": "business",
                **kwargs
            }
        )

# Usage examples
def record_security_events():
    # Failed login attempt
    security_event = SecurityEvent(
        event_type="login_failed",
        user_id="user123",
        source_ip="192.168.1.100",
        reason="invalid_password",
        attempt_count=3
    )
    event_logger.emit(security_event)
    
    # Successful login
    security_event = SecurityEvent(
        event_type="login_success",
        user_id="user123", 
        source_ip="192.168.1.100",
        session_id="sess-abc-456"
    )
    event_logger.emit(security_event)

def record_business_events():
    # Order placed
    business_event = BusinessEvent(
        event_type="order_placed",
        entity_id="order-789",
        customer_id="cust-123",
        order_total=99.99,
        currency="USD"
    )
    event_logger.emit(business_event)
    
    # Payment processed
    business_event = BusinessEvent(
        event_type="payment_processed",
        entity_id="payment-456",
        order_id="order-789",
        amount=99.99,
        payment_method="credit_card"
    )
    event_logger.emit(business_event)

Integration with Standard Python Logging

import logging
from opentelemetry._logs import get_logger, LogRecord, SeverityNumber
from opentelemetry import trace
import time

# Custom handler to bridge Python logging to OpenTelemetry
class OpenTelemetryLogHandler(logging.Handler):
    def __init__(self):
        super().__init__()
        self.otel_logger = get_logger(__name__)
    
    def emit(self, record: logging.LogRecord):
        # Convert Python log level to OpenTelemetry severity
        severity_map = {
            logging.DEBUG: SeverityNumber.DEBUG,
            logging.INFO: SeverityNumber.INFO,
            logging.WARNING: SeverityNumber.WARN,
            logging.ERROR: SeverityNumber.ERROR,
            logging.CRITICAL: SeverityNumber.FATAL,
        }
        
        # Get current trace context
        current_span = trace.get_current_span()
        span_context = current_span.get_span_context()
        
        # Create OpenTelemetry log record
        otel_record = LogRecord(
            timestamp=int(record.created * 1_000_000_000),  # Convert to nanoseconds
            severity_text=record.levelname,
            severity_number=severity_map.get(record.levelno, SeverityNumber.INFO),
            body=record.getMessage(),
            attributes={
                "python.logger.name": record.name,
                "python.module": record.module,
                "python.function": record.funcName,
                "python.lineno": record.lineno,
            },
            trace_id=span_context.trace_id if span_context.is_valid else None,
            span_id=span_context.span_id if span_context.is_valid else None,
            trace_flags=span_context.trace_flags if span_context.is_valid else None,
        )
        
        self.otel_logger.emit(otel_record)

# Set up Python logging with OpenTelemetry handler
def setup_logging():
    # Create and configure handler
    handler = OpenTelemetryLogHandler()
    handler.setLevel(logging.INFO)
    
    # Add to root logger
    root_logger = logging.getLogger()
    root_logger.addHandler(handler)
    root_logger.setLevel(logging.INFO)

# Usage
setup_logging()

# Now standard Python logging will be captured by OpenTelemetry
logger = logging.getLogger(__name__)

def example_function():
    tracer = trace.get_tracer(__name__)
    with tracer.start_as_current_span("example-operation"):
        logger.info("Starting operation")
        
        try:
            # Some operation that might fail
            result = risky_operation()
            logger.info(f"Operation completed successfully: {result}")
            
        except Exception as e:
            logger.error(f"Operation failed: {e}", exc_info=True)
            raise

Install with Tessl CLI

npx tessl i tessl/pypi-opentelemetry-api

docs