tessl/pypi-opencensus

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

—

Pending

Overview

Eval results

Files

Metrics and Statistics

Name: tessl/pypi-opencensus
Author: tessl

Comprehensive metrics collection, aggregation, and export functionality for monitoring application performance, resource usage, and business metrics. Provides measures, views, aggregations, and export capabilities.

Capabilities

Measures and Measurements

Define what to measure and record individual measurements for later aggregation and analysis.

class BaseMeasure:
    """
    Base class for all measures.
    
    Parameters:
    - name: str, measure name/identifier
    - description: str, human-readable description
    - unit: str, measurement unit (e.g., "ms", "bytes", "1")
    """
    def __init__(self, name, description, unit=None): ...

    @property
    def name(self):
        """str: Measure name"""

    @property
    def description(self):
        """str: Measure description"""

    @property
    def unit(self):
        """str: Measurement unit"""

class MeasureInt(BaseMeasure):
    """
    Integer-valued measure for counting and discrete values.
    
    Parameters:
    - name: str, measure name
    - description: str, measure description  
    - unit: str, measurement unit
    """
    def __init__(self, name, description, unit=None): ...

class MeasureFloat(BaseMeasure):
    """
    Float-valued measure for continuous values and rates.
    
    Parameters:
    - name: str, measure name
    - description: str, measure description
    - unit: str, measurement unit
    """
    def __init__(self, name, description, unit=None): ...

class Measurement:
    """
    Individual measurement combining measure and value.
    
    Parameters:
    - measure: BaseMeasure, what is being measured
    - value: int/float, measured value
    """
    def __init__(self, measure, value): ...

    @property
    def measure(self):
        """BaseMeasure: What is being measured"""

    @property  
    def value(self):
        """int/float: Measured value"""

class MeasurementInt(Measurement):
    """Integer measurement."""
    def __init__(self, measure, value): ...

class MeasurementFloat(Measurement):
    """Float measurement."""
    def __init__(self, measure, value): ...

Views and Aggregations

Define how measurements are aggregated over time and exported as metrics.

class View:
    """
    Defines aggregation and tag keys for measurements.
    
    Parameters:
    - name: str, view name/identifier
    - description: str, human-readable description
    - columns: list, tag keys for grouping/filtering
    - measure: BaseMeasure, measure to aggregate
    - aggregation: Aggregation, how to aggregate measurements
    """
    def __init__(self, name, description, columns, measure, aggregation): ...

    @property
    def name(self):
        """str: View name"""

    @property
    def description(self):
        """str: View description"""

    @property
    def columns(self):
        """list: Tag keys for grouping"""

    @property
    def measure(self):
        """BaseMeasure: Measure being aggregated"""

    @property
    def aggregation(self):
        """Aggregation: Aggregation method"""

    def new_aggregation_data(self):
        """Create new aggregation data instance."""

    def get_metric_descriptor(self):
        """Get metric descriptor for this view."""

class SumAggregation:
    """
    Sum aggregation - adds all measurement values.
    
    Parameters:
    - sum: int/float, initial sum value
    """
    def __init__(self, sum=None): ...

    def new_aggregation_data(self, measure):
        """
        Create new sum aggregation data.
        
        Parameters:
        - measure: BaseMeasure, measure being aggregated
        
        Returns:
        SumAggregationData: Aggregation data instance
        """

    def get_metric_type(self, measure):
        """
        Get metric type for this aggregation.
        
        Parameters:
        - measure: BaseMeasure, measure being aggregated
        
        Returns:
        MetricDescriptorType: Appropriate metric type
        """

class CountAggregation:
    """
    Count aggregation - counts number of measurements.
    
    Parameters:
    - count: int, initial count value
    """
    def __init__(self, count=0): ...

    def new_aggregation_data(self, measure=None):
        """Create new count aggregation data."""

    def get_metric_type(self, measure):
        """Get metric type for count aggregation."""

class DistributionAggregation:
    """
    Distribution/histogram aggregation with configurable buckets.
    
    Parameters:
    - boundaries: list, bucket boundary values
    """
    def __init__(self, boundaries=None): ...

    def new_aggregation_data(self, measure=None):
        """Create new distribution aggregation data."""

    def get_metric_type(self, measure):
        """Get metric type for distribution aggregation."""

class LastValueAggregation:
    """
    Last value aggregation - keeps most recent measurement.
    
    Parameters:
    - value: int/float, initial value
    """
    def __init__(self, value=0): ...

    def new_aggregation_data(self, measure):
        """Create new last value aggregation data."""

    def get_metric_type(self, measure):
        """Get metric type for last value aggregation."""

Stats Recording and Management

Core runtime classes that handle measurement recording, view management, and data collection workflows.

class StatsRecorder:
    """
    Records measurements against registered views.
    
    The main interface for recording measurements in the stats system.
    """
    def __init__(self): ...
    
    def new_measurement_map(self):
        """
        Create a new measurement map for recording multiple measurements.
        
        Returns:
        MeasurementMap: New measurement map instance
        """

class MeasurementMap:
    """
    Collection of measurements that can be recorded atomically.
    
    Parameters:
    - measure_to_view_map: MeasureToViewMap, view mapping
    - attachments: dict, contextual information (optional)
    """
    def __init__(self, measure_to_view_map, attachments=None): ...
    
    def measure_int_put(self, measure, value):
        """
        Associate integer measure with value.
        
        Parameters:
        - measure: MeasureInt, integer measure
        - value: int, measurement value
        """
    
    def measure_float_put(self, measure, value):
        """
        Associate float measure with value.
        
        Parameters:
        - measure: MeasureFloat, float measure
        - value: float, measurement value
        """
    
    def measure_put_attachment(self, key, value):
        """
        Add contextual attachment information.
        
        Parameters:
        - key: str, attachment key
        - value: str, attachment value
        """
    
    def record(self, tags=None):
        """
        Record all measurements in this map.
        
        Parameters:
        - tags: TagMap, tags to associate with measurements (optional)
        """

class ViewManager:
    """
    Manages view registration and data collection.
    
    Central registry for views and their associated data collection.
    """
    def __init__(self): ...
    
    def register_view(self, view):
        """
        Register a view for data collection.
        
        Parameters:
        - view: View, view to register
        """
    
    def get_view(self, view_name):
        """
        Get view data by name.
        
        Parameters:
        - view_name: str, name of the view
        
        Returns:
        ViewData: View data if found, None otherwise
        """
    
    def get_all_exported_views(self):
        """
        Get all views marked for export.
        
        Returns:
        list: List of ViewData objects for exported views
        """
    
    def register_exporter(self, exporter):
        """
        Register a stats exporter.
        
        Parameters:
        - exporter: StatsExporter, exporter to register
        """
    
    def unregister_exporter(self, exporter):
        """
        Unregister a stats exporter.
        
        Parameters:
        - exporter: StatsExporter, exporter to unregister
        """

class ViewData:
    """
    Contains aggregated data for a specific view.
    
    Parameters:
    - view: View, associated view
    - start_time: datetime, data collection start time
    - end_time: datetime, data collection end time (optional)
    """
    def __init__(self, view, start_time, end_time=None): ...
    
    def start(self):
        """Set start time for data collection."""
    
    def end(self):
        """Set end time for data collection."""
    
    def record(self, context, value, timestamp, attachments=None):
        """
        Record a measurement value.
        
        Parameters:
        - context: tag context for the measurement
        - value: int/float, measured value
        - timestamp: datetime, measurement timestamp
        - attachments: dict, contextual attachments (optional)
        """
    
    @property
    def view(self):
        """View: Associated view definition"""
    
    @property
    def start_time(self):
        """datetime: Data collection start time"""
    
    @property
    def end_time(self):
        """datetime: Data collection end time"""

class BucketBoundaries:
    """
    Defines histogram bucket boundaries for distribution aggregations.
    
    Parameters:
    - boundaries: list, bucket boundary values in ascending order
    """
    def __init__(self, boundaries=None): ...
    
    def is_valid_boundaries(self, boundaries):
        """
        Validate that boundaries are in ascending order.
        
        Parameters:
        - boundaries: list, boundary values to validate
        
        Returns:
        bool: True if boundaries are valid
        """
    
    @property
    def boundaries(self):
        """list: Bucket boundary values"""

Global Stats Interface

Singleton instance providing access to the stats recording and view management system.

class _Stats:
    """
    Global stats instance providing access to recording and view management.
    
    This is the main entry point for the OpenCensus stats system.
    """
    def __init__(self): ...
    
    def get_metrics(self):
        """
        Get all metrics from registered views.
        
        Returns:
        iterator: Iterator over Metric objects
        """
    
    @property
    def stats_recorder(self):
        """StatsRecorder: Global stats recorder instance"""
    
    @property
    def view_manager(self):
        """ViewManager: Global view manager instance"""

# Global stats instance - the main interface for OpenCensus stats
stats = _Stats()
"""_Stats: Global stats instance for recording measurements and managing views"""

Convenient global interface for recording measurements and managing views.

```python { .api }
class _Stats:
    """Internal stats implementation providing recorder and view manager."""
    
    @property
    def stats_recorder(self):
        """StatsRecorder: Global measurement recorder"""

    @property
    def view_manager(self):
        """ViewManager: Global view registry"""

# Global stats instance
stats = _Stats()

Metric Export Types

Structured data types for exporting metrics to various backends and monitoring systems.

class LabelKey:
    """
    Label key with description for metric metadata.
    
    Parameters:
    - key: str, label key name
    - description: str, label description
    """
    def __init__(self, key, description): ...

    @property
    def key(self):
        """str: Label key name"""

    @property
    def description(self):
        """str: Label description"""

class LabelValue:
    """
    Label value for metric labeling.
    
    Parameters:
    - value: str, label value (None for missing labels)
    """
    def __init__(self, value=None): ...

    @property
    def value(self):
        """str: Label value"""

    def __eq__(self, other):
        """Compare label values for equality."""

    def __hash__(self):
        """Hash function for use in sets/dicts."""

class Metric:
    """
    Collection of time series data with metadata.
    
    Parameters:
    - descriptor: MetricDescriptor, metric type and schema
    - time_series: list, list of TimeSeries objects
    """
    def __init__(self, descriptor, time_series): ...

    @property
    def descriptor(self):
        """MetricDescriptor: Metric metadata"""

    @property
    def time_series(self):
        """list: Time series data points"""

class MetricDescriptor:
    """
    Defines metric type and schema.
    
    Parameters:
    - name: str, metric name
    - description: str, metric description
    - unit: str, measurement unit
    - type_: MetricDescriptorType, metric type
    - label_keys: list, available label keys
    """
    def __init__(self, name, description, unit, type_, label_keys): ...

    @property
    def name(self):
        """str: Metric name"""

    @property
    def description(self):
        """str: Metric description"""

    @property
    def unit(self):
        """str: Measurement unit"""

    @property
    def type(self):
        """MetricDescriptorType: Metric type"""

    @property
    def label_keys(self):
        """list: Available label keys"""

class MetricDescriptorType:
    """Enumeration of metric types."""
    GAUGE_INT64 = 1
    GAUGE_DOUBLE = 2
    GAUGE_DISTRIBUTION = 3
    CUMULATIVE_INT64 = 4
    CUMULATIVE_DOUBLE = 5
    CUMULATIVE_DISTRIBUTION = 6
    SUMMARY = 7

    @classmethod
    def to_type_class(cls, metric_descriptor_type):
        """
        Convert metric type to appropriate value class.
        
        Parameters:
        - metric_descriptor_type: int, metric type constant
        
        Returns:
        type: Appropriate value class
        """

class TimeSeries:
    """
    Time-varying values collection with labels.
    
    Parameters:
    - label_values: list, label values for this series
    - points: list, time-ordered measurement points
    - start_timestamp: str, series start time
    """
    def __init__(self, label_values, points, start_timestamp): ...

    @property
    def label_values(self):
        """list: Label values"""

    @property
    def points(self):
        """list: Measurement points"""

    @property
    def start_timestamp(self):
        """str: Series start time"""

    def check_points_type(self, type_class):
        """Validate that all points are of expected type."""

class Point:
    """
    Timestamped measurement point.
    
    Parameters:
    - value: Value, measurement value
    - timestamp: time, measurement time
    """
    def __init__(self, value, timestamp): ...

    @property
    def value(self):
        """Value: Measurement value"""

    @property
    def timestamp(self):
        """time: Measurement time"""

Export Infrastructure

Manage metric producers and coordinate metric export to monitoring backends.

class MetricProducer:
    """Base class for metric production."""
    
    def get_metrics(self):
        """
        Get current metrics.
        
        Returns:
        iterator: Iterator of Metric objects
        """

class MetricProducerManager:
    """
    Container for managing multiple metric producers.
    
    Parameters:
    - metric_producers: list, initial list of producers
    """
    def __init__(self, metric_producers=None): ...

    def add(self, metric_producer):
        """
        Add metric producer.
        
        Parameters:
        - metric_producer: MetricProducer, producer to add
        """

    def remove(self, metric_producer):
        """
        Remove metric producer.
        
        Parameters:
        - metric_producer: MetricProducer, producer to remove
        """

    def get_all(self):
        """
        Get all registered producers.
        
        Returns:
        set: Set of MetricProducer instances
        """

class StatsExporter:
    """Base class for stats exporters."""
    
    def on_register_view(self, view):
        """
        Called when view is registered.
        
        Parameters:
        - view: View, newly registered view
        """

    def emit(self, view_datas):
        """
        Export view data.
        
        Parameters:
        - view_datas: list, list of ViewData objects
        """

Metric Transport

Periodic export of metrics with background processing and error handling.

class PeriodicMetricTask:
    """
    Periodic metric export task.
    
    Parameters:
    - interval: float, export interval in seconds
    - function: callable, export function
    - args: tuple, function arguments
    - kwargs: dict, function keyword arguments
    - name: str, task name
    """
    def __init__(self, interval=None, function=None, args=None, kwargs=None, name=None): ...

    def run(self):
        """Start periodic metric export."""

    def close(self):
        """Stop periodic export and cleanup."""

def get_exporter_thread(metric_producers, exporter, interval=None):
    """
    Get running metric export task.
    
    Parameters:
    - metric_producers: list, metric producers to export
    - exporter: Exporter, export destination
    - interval: float, export interval in seconds
    
    Returns:
    PeriodicMetricTask: Running export task
    """

class TransportError(Exception):
    """Exception raised for transport-related errors."""

Usage Examples

Basic Measurements

from opencensus.stats.stats import stats
from opencensus.stats.measure import MeasureInt, MeasureFloat
from opencensus.stats.view import View
from opencensus.stats.aggregation import CountAggregation, SumAggregation

# Create measures
request_count = MeasureInt("requests", "number of requests", "1")
request_latency = MeasureFloat("request_latency", "request latency", "ms")

# Create views
request_count_view = View(
    "request_count_total",
    "total number of requests", 
    [],  # no grouping tags
    request_count,
    CountAggregation()
)

latency_sum_view = View(
    "request_latency_sum",
    "sum of request latencies",
    ["method", "status"],  # group by HTTP method and status
    request_latency,
    SumAggregation()
)

# Register views
from opencensus.stats.stats import stats
stats.view_manager.register_view(request_count_view)
stats.view_manager.register_view(latency_sum_view)

# Record measurements
measurement_map = stats.stats_recorder.new_measurement_map()
measurement_map.measure_int_put(request_count, 1)
measurement_map.measure_float_put(request_latency, 45.2)
measurement_map.record()

Tagged Measurements

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

# Create tag keys and values
method_key = TagKey("method")
status_key = TagKey("status")

method_value = TagValue("GET")
status_value = TagValue("200")

# Create tag map
tag_map = TagMap()
tag_map.insert(method_key, method_value)
tag_map.insert(status_key, status_value)

# Record measurement with tags
measurement_map = stats.stats_recorder.new_measurement_map()
measurement_map.measure_int_put(request_count, 1)
measurement_map.record(tag_map)

Distribution Metrics

from opencensus.stats.aggregation import DistributionAggregation

# Create distribution with custom bucket boundaries
latency_distribution = DistributionAggregation([
    0, 10, 50, 100, 200, 500, 1000, 2000, 5000
])

latency_dist_view = View(
    "request_latency_distribution",
    "distribution of request latencies",
    ["service", "endpoint"],
    request_latency,
    latency_distribution
)

stats.view_manager.register_view(latency_dist_view)

# Record measurements - they'll be bucketed automatically
for latency in [15, 45, 120, 350, 1200]:
    measurement_map = stats.stats_recorder.new_measurement_map()
    measurement_map.measure_float_put(request_latency, latency)
    measurement_map.record(tag_map)

Install with Tessl CLI