Collector

Metrics collection for performance monitoring.

Classes

class arshai.observability.metrics.collector.Metric(name, type, value, timestamp=<factory>, tags=<factory>, unit=None)

Bases: object

Individual metric data point.

name: str

type: MetricType

value: float

timestamp: datetime

tags: Dict[str, str]

unit: Optional[str] = None

__init__(name, type, value, timestamp=<factory>, tags=<factory>, unit=None)

class arshai.observability.metrics.collector.MetricType(value)

Bases: Enum

Types of metrics that can be collected.

COUNTER = 'counter'

GAUGE = 'gauge'

HISTOGRAM = 'histogram'

TIMER = 'timer'

class arshai.observability.metrics.collector.MetricsCollector(max_history=1000)

Bases: object

Thread-safe metrics collector for Arshai components.

Example

collector = MetricsCollector()

# Count events collector.increment(“api.calls”, tags={“endpoint”: “/process”})

# Record timing with collector.timer(“processing.time”):

# Do work pass

# Record value collector.gauge(“queue.size”, 42)

# Get metrics stats = collector.get_stats()

__init__(max_history=1000)

Initialize metrics collector.

Parameters:

max_history (int) – Maximum number of data points to keep per metric

increment(name, value=1, tags=None)

Increment a counter metric.

Parameters:

• name (str) – Metric name

• value (float) – Amount to increment

• tags (Optional[Dict[str, str]]) – Optional tags for the metric

decrement(name, value=1, tags=None)

Decrement a counter metric.

Parameters:

• name (str) – Metric name

• value (float) – Amount to decrement

• tags (Optional[Dict[str, str]]) – Optional tags for the metric

gauge(name, value, tags=None)

Set a gauge metric value.

Parameters:

• name (str) – Metric name

• value (float) – Current value

• tags (Optional[Dict[str, str]]) – Optional tags for the metric

histogram(name, value, tags=None)

Add a value to a histogram metric.

Parameters:

• name (str) – Metric name

• value (float) – Value to record

• tags (Optional[Dict[str, str]]) – Optional tags for the metric

timer(name, tags=None)

Context manager for timing operations.

Parameters:

• name (str) – Metric name

• tags (Optional[Dict[str, str]]) – Optional tags for the metric

Example

with collector.timer(“db.query”, tags={“query”: “select”}):

# Perform query pass

start_timer(name, tags=None)

Start a timer manually.

Parameters:

• name (str) – Metric name

• tags (Optional[Dict[str, str]]) – Optional tags for the metric

stop_timer(name, tags=None)

Stop a manually started timer.

Parameters:

• name (str) – Metric name

• tags (Optional[Dict[str, str]]) – Optional tags for the metric

Returns:

Elapsed time in seconds

get_stats()

Get current statistics for all metrics.

Return type:

Dict[str, Any]

Returns:

Dictionary with metric statistics

reset()

Reset all metrics.

class arshai.observability.metrics.collector.TimerContext(collector, name, tags=None)

Bases: object

Context manager for timing operations.

__init__(collector, name, tags=None)