tessl/pypi-opencensus

A comprehensive observability framework providing distributed tracing, metrics collection, and statistics gathering capabilities for Python applications.

—

Pending

Overview

Eval results

Files

Tags and Context Management

Name: tessl/pypi-opencensus
Author: tessl

Tag management and context propagation for organizing and filtering observability data across service boundaries and execution contexts. Provides key-value labeling system and runtime context management.

Capabilities

Tag Components

Core building blocks for creating and managing key-value labels used throughout the OpenCensus ecosystem.

class Tag:
    """
    Tag representing a key-value pair.
    
    Parameters:
    - key: TagKey, tag key
    - value: TagValue, tag value
    """
    def __init__(self, key, value): ...
    
    @property
    def key(self):
        """TagKey: Tag key"""
    
    @property
    def value(self):
        """TagValue: Tag value"""

class TagKey:
    """
    String subclass for tag keys with validation.
    
    Parameters:
    - name: str, tag key name (must be valid)
    """
    def __init__(self, name): ...

class TagValue:
    """
    String subclass for tag values with validation.
    
    Parameters:
    - value: str, tag value (must be valid)
    """
    def __init__(self, value): ...

Tag Management

Ordered collection of tags with insertion, update, and query operations for managing key-value metadata.

class TagMap:
    """
    Ordered dictionary of tags for metadata management.
    
    Parameters:
    - tags: dict, initial tag mapping (TagKey -> TagValue)
    """
    def __init__(self, tags=None): ...

    def insert(self, key, value):
        """
        Insert new tag or update existing tag.
        
        Parameters:
        - key: TagKey, tag key
        - value: TagValue, tag value
        
        Returns:
        TagMap: New TagMap with inserted tag
        """

    def delete(self, key):
        """
        Remove tag by key.
        
        Parameters:
        - key: TagKey, tag key to remove
        
        Returns:
        TagMap: New TagMap without the specified tag
        """

    def update(self, key, value):
        """
        Update existing tag value.
        
        Parameters:
        - key: TagKey, existing tag key
        - value: TagValue, new tag value
        
        Returns:
        TagMap: New TagMap with updated tag
        """

    def tag_key_exists(self, key):
        """
        Check if tag key exists.
        
        Parameters:
        - key: TagKey, tag key to check
        
        Returns:
        bool: True if key exists
        """

    def get_value(self, key):
        """
        Get tag value by key.
        
        Parameters:
        - key: TagKey, tag key
        
        Returns:
        TagValue: Tag value or None if not found
        """

Runtime Context

Thread-local context management for propagating tags across function calls and async operations.

TagContext = RuntimeContext.register_slot('tag_context', None)
"""
Runtime context slot for tag context propagation.

Usage:
- TagContext.get() - Get current tag context
- TagContext.set(tag_map) - Set tag context for current scope
- TagContext.clear() - Clear tag context
"""

Tag Validation

Validation functions for ensuring tag keys and values meet OpenCensus requirements and standards.

def is_legal_chars(value):
    """
    Check if string contains only legal ASCII characters (32-126).
    
    Parameters:
    - value: str, string to validate
    
    Returns:
    bool: True if all characters are in legal range
    """

def is_valid_tag_name(name):
    """
    Validate tag key name according to OpenCensus rules.
    
    Parameters:
    - name: str, tag key name to validate
    
    Returns:
    bool: True if name is valid
    """

def is_valid_tag_value(value):
    """
    Validate tag value according to OpenCensus rules.
    
    Parameters:
    - value: str, tag value to validate
    
    Returns:
    bool: True if value is valid
    """

Tag Propagation

Binary serialization for propagating tags across process and network boundaries in distributed systems.

binary_serializer = BinarySerializer()
"""
Binary serialization instance for tag propagation.

Methods:
- to_byte_array(tag_map) - Serialize TagMap to bytes
- from_byte_array(byte_array) - Deserialize bytes to TagMap
"""

Usage Examples

Basic Tag Operations

from opencensus.tags import TagMap, TagKey, TagValue, Tag

# Create tag keys and values
service_key = TagKey("service")
version_key = TagKey("version")
environment_key = TagKey("environment")

service_value = TagValue("user-service")
version_value = TagValue("1.2.3")
environment_value = TagValue("production")

# Create tag map
tag_map = TagMap()
tag_map = tag_map.insert(service_key, service_value)
tag_map = tag_map.insert(version_key, version_value)
tag_map = tag_map.insert(environment_key, environment_value)

# Query tags
if tag_map.tag_key_exists(service_key):
    service = tag_map.get_value(service_key)
    print(f"Service: {service}")

# Update tag
tag_map = tag_map.update(version_key, TagValue("1.2.4"))

# Remove tag
tag_map = tag_map.delete(environment_key)

Context Management

from opencensus.tags import TagContext, TagMap, TagKey, TagValue

# Create tag map for context
tag_map = TagMap()
tag_map = tag_map.insert(TagKey("user_id"), TagValue("12345"))
tag_map = tag_map.insert(TagKey("session_id"), TagValue("abc-123"))

# Set context for current thread/execution
TagContext.set(tag_map)

def process_request():
    # Tags are automatically available in this context
    current_tags = TagContext.get()
    
    if current_tags:
        user_id = current_tags.get_value(TagKey("user_id"))
        print(f"Processing request for user: {user_id}")
    
    # Add additional context-specific tags
    local_tags = current_tags.insert(TagKey("operation"), TagValue("process_order"))
    
    with TagContext(local_tags):
        # This scope has both original and new tags
        process_order()

def process_order():
    # Access all tags from context
    tags = TagContext.get()
    operation = tags.get_value(TagKey("operation"))
    print(f"Executing operation: {operation}")

# Call function - tags are propagated automatically
process_request()

# Clear context when done
TagContext.clear()

Tag Validation

from opencensus.tags.validation import is_valid_tag_name, is_valid_tag_value, is_legal_chars

# Validate tag components before use
def create_safe_tag(key_name, value_str):
    if not is_valid_tag_name(key_name):
        raise ValueError(f"Invalid tag key: {key_name}")
    
    if not is_valid_tag_value(value_str):
        raise ValueError(f"Invalid tag value: {value_str}")
    
    return TagKey(key_name), TagValue(value_str)

# Test validation
try:
    key, value = create_safe_tag("service.name", "user-service")
    print("Valid tag created")
except ValueError as e:
    print(f"Tag validation failed: {e}")

# Check character legality
test_strings = ["valid_tag", "tag with spaces", "tag\twith\ttabs", "tág_with_unicode"]
for s in test_strings:
    if is_legal_chars(s):
        print(f"'{s}' - Legal characters")
    else:
        print(f"'{s}' - Contains illegal characters")

Integration with Stats and Tracing

from opencensus.tags import TagMap, TagKey, TagValue, TagContext
from opencensus.stats import Stats
from opencensus.stats.measure import MeasureInt
from opencensus.stats.view import View
from opencensus.stats.aggregation import CountAggregation

# Create tagged measurements
request_count = MeasureInt("requests", "number of requests", "1")

# Create view with tag keys for grouping
request_view = View(
    "request_count_by_service",
    "requests grouped by service and method",
    [TagKey("service"), TagKey("method")],  # group by these tags
    request_count,
    CountAggregation()
)

stats = Stats()
stats.view_manager.register_view(request_view)

# Record measurements with tags
service_tags = TagMap()
service_tags = service_tags.insert(TagKey("service"), TagValue("api-gateway"))
service_tags = service_tags.insert(TagKey("method"), TagValue("GET"))

measurement_map = stats.stats_recorder.new_measurement_map()
measurement_map.measure_int_put(request_count, 1)
measurement_map.record(service_tags)

# Tags are automatically included in traces when context is set
TagContext.set(service_tags)

from opencensus.trace.tracer import Tracer
tracer = Tracer()

with tracer.span('handle_request') as span:
    # Span automatically includes tags from context
    span.add_attribute('custom_attr', 'value')
    # Process request...

Cross-Service Tag Propagation

from opencensus.tags.propagation import binary_serializer
from opencensus.tags import TagMap, TagKey, TagValue
import requests

# Sending service
def make_request_with_tags():
    # Create tags for current operation
    tags = TagMap()
    tags = tags.insert(TagKey("trace_id"), TagValue("abc-123"))
    tags = tags.insert(TagKey("service"), TagValue("frontend"))
    tags = tags.insert(TagKey("user_id"), TagValue("12345"))
    
    # Serialize tags for transmission
    tag_bytes = binary_serializer.to_byte_array(tags)
    
    # Send in HTTP header
    headers = {
        'X-OpenCensus-Tags': tag_bytes.hex(),
        'Content-Type': 'application/json'
    }
    
    response = requests.post('http://backend/api/process', 
                           json={'data': 'payload'}, 
                           headers=headers)
    return response

# Receiving service
def handle_request(request):
    # Extract tags from header
    tag_header = request.headers.get('X-OpenCensus-Tags')
    
    if tag_header:
        try:
            tag_bytes = bytes.fromhex(tag_header)
            received_tags = binary_serializer.from_byte_array(tag_bytes)
            
            # Set context for this request
            TagContext.set(received_tags)
            
            # Add service-specific tags
            local_tags = received_tags.insert(TagKey("service"), TagValue("backend"))
            TagContext.set(local_tags)
            
            # Process with full tag context
            return process_request()
            
        except Exception as e:
            print(f"Failed to deserialize tags: {e}")
            # Continue without tags
            return process_request()
    else:
        return process_request()

Install with Tessl CLI