mstar.observability.recorders

mstar.observability.recorders#

Dependency-free accumulators for the families in catalog.

A process other than the API server keeps one recorder per family and sends snapshot() on a message the API server already receives; the API server exports the latest snapshot of each family. Snapshots are cumulative, never deltas, so a lost or reordered message cannot corrupt a counter.

Not thread-safe: update and snapshot a recorder from one thread (an EdgeTransfers hands latest to another thread whole).

Functions

merge(family, snapshots)

One snapshot of family summing the series of snapshots that share label values: several processes' views of one deployment (the replicas of a graph node on different workers, say).

Classes

Counter(family)

EdgeTransfers(transport)

One tensor manager's transfer families (TRANSFER_BYTES, TRANSFERS, TRANSFER_SECONDS).

Gauge(family)

Histogram(family)

Fixed buckets from the family.

MetricSnapshot(name, label_values[, values, ...])

One family's series, columnar so the wire codec encodes it typed.
class mstar.observability.recorders.Counter(family)[source]#

Bases: Gauge

Parameters:

family (Family)

inc(labels, amount=1.0)[source]#
Parameters:
Return type:

None

class mstar.observability.recorders.EdgeTransfers(transport)[source]#

Bases: object

One tensor manager’s transfer families (TRANSFER_BYTES, TRANSFERS, TRANSFER_SECONDS). transport is the manager’s protocol.

record only appends to five columns; snapshot folds them into one row per (src, dst) once per status interval, so an edge read creates no object and touches no per-edge state.

latest is the last publish(): the data worker publishes from its own thread and the API server reads it from another, swapping whole lists.

Parameters:

transport (str)

publish()[source]#
Return type:

None

record(src, dst, nbytes, tensors, seconds)[source]#
Parameters:
Return type:

None

snapshot()[source]#
Return type:

list[MetricSnapshot]

class mstar.observability.recorders.Gauge(family)[source]#

Bases: object

Parameters:

family (Family)

set(labels, value)[source]#
Parameters:
Return type:

None

snapshot()[source]#
Return type:

MetricSnapshot

class mstar.observability.recorders.Histogram(family)[source]#

Bases: object

Fixed buckets from the family. observe bumps one bucket; the cumulative counts Prometheus wants are built at snapshot.

Parameters:

family (Family)

observe(labels, value)[source]#
Parameters:
Return type:

None

snapshot()[source]#
Return type:

MetricSnapshot

class mstar.observability.recorders.MetricSnapshot(name, label_values, values=<factory>, bucket_counts=<factory>, sums=<factory>)[source]#

Bases: object

One family’s series, columnar so the wire codec encodes it typed.

Parameters:
bucket_counts: list[list[int]]#

per series, cumulative counts per bucket, +Inf last (so the last is the observation count), and the sum of observations.

Type:

Histogram

label_values: list[list[str]]#

One row per series, in the family’s label order.

name: str#
sums: list[float]#
values: list[float]#

one value per series.

Type:

Counter or gauge

mstar.observability.recorders.merge(family, snapshots)[source]#

One snapshot of family summing the series of snapshots that share label values: several processes’ views of one deployment (the replicas of a graph node on different workers, say).

Parameters:
Return type:

MetricSnapshot