proxystore.store.metrics

proxystore/store/metrics.py

Utilities for recording Store operation metrics.

See the Performance Guide to learn more about interacting with metrics recorded for Store operations.

KeyT module-attribute

KeyT = ConnectorKeyT | Sequence[ConnectorKeyT]

Key types supported by StoreMetrics.

DEFAULT_MAX_KEYS module-attribute

DEFAULT_MAX_KEYS = 10000

Default maximum number of keys to keep metrics for.

ProxyT module-attribute

ProxyT = Proxy[Any] | Sequence[Proxy[Any]]

Proxy types supported by StoreMetrics.

When a ProxyT is passed, the keys are extracted from the proxies.

TimeStats dataclass

TimeStats(
    count: int = 0,
    avg_time_ms: float = 0,
    min_time_ms: float = inf,
    max_time_ms: float = 0,
    last_time_ms: float = 0,
    last_timestamp: float = 0,
)

Tracks time statistics of a recurring event.

Attributes:

• count (int) –

  Number of times this event has occurred.

• avg_time_ms (float) –

  Average time in milliseconds of the event.

• min_time_ms (float) –

  Minimum time in milliseconds of all event occurrences.

• max_time_ms (float) –

  Maximum time in milliseconds of all event occurrences.

• last_time_ms (float) –

  Time in milliseconds of the most recent event occurrence.

• last_timestamp (float) –

  The UNIX timestamp (seconds) of when the last event time was recorded.

as_dict

as_dict() -> dict[str, Any]

Convert the dataclass to a dict.

Source code in proxystore/store/metrics.py

def as_dict(self) -> dict[str, Any]:
    """Convert the dataclass to a [`dict`][dict]."""
    return dataclasses.asdict(self)

Metrics dataclass

Metrics(
    attributes: dict[str, Any] = dict(),
    counters: dict[str, int] = dict(),
    times: dict[str, TimeStats] = dict(),
)

Records metrics and attributes for events.

Attributes:

• attributes (dict[str, Any]) –

  A mapping of attributes to their values.

• counters (dict[str, int]) –

  A mapping of counter names to the integer value of the counter.

• times (dict[str, TimeStats]) –

  A mapping of events to a summary of the statistics recorded over occurrences of that event.

as_dict

as_dict() -> dict[str, Any]

Convert the dataclass to a dict.

Source code in proxystore/store/metrics.py

def as_dict(self) -> dict[str, Any]:
    """Convert the dataclass to a [`dict`][dict]."""
    return dataclasses.asdict(self)

StoreMetrics

StoreMetrics(max_keys: int = DEFAULT_MAX_KEYS)

Record and query metrics on Store operations.

Metrics are kept per key for the max_keys most recently recorded keys (or sequences of keys). When a key is dropped, its times are still included in aggregate_times(), but get_metrics() returns None for it.

Parameters:

• max_keys (int, default: DEFAULT_MAX_KEYS ) –

  Maximum number of keys to keep metrics for.

Raises:

• ValueError –

  If max_keys is less than zero.

Source code in proxystore/store/metrics.py

def __init__(self, max_keys: int = DEFAULT_MAX_KEYS) -> None:
    if max_keys < 0:
        raise ValueError(f'max_keys must be >= 0. Got {max_keys}.')
    self.max_keys = max_keys
    # Metrics of each key, least recently recorded first.
    self._metrics: OrderedDict[Hashable, Metrics] = OrderedDict()
    # Times of the keys which were dropped.
    self._dropped_times: dict[str, TimeStats] = defaultdict(TimeStats)
    self._lock = threading.Lock()

add_attribute

add_attribute(name: str, key: KeyT, value: Any) -> None

Add an attribute associated with the key.

Parameters:

• name (str) –

  Name of attribute.

• key (KeyT) –

  Key to add attribute to.

• value (Any) –

  Attribute value.

Source code in proxystore/store/metrics.py

def add_attribute(self, name: str, key: KeyT, value: Any) -> None:
    """Add an attribute associated with the key.

    Args:
        name: Name of attribute.
        key: Key to add attribute to.
        value: Attribute value.
    """
    with self._lock:
        self._record(key).attributes[name] = value
        self._trim()

add_counter

add_counter(name: str, key: KeyT, value: int) -> None

Add to a counter.

Parameters:

• name (str) –

  Name of counter.

• key (KeyT) –

  Key associated with the counter.

• value (int) –

  Amount to increment counter by.

Source code in proxystore/store/metrics.py

def add_counter(self, name: str, key: KeyT, value: int) -> None:
    """Add to a counter.

    Args:
        name: Name of counter.
        key: Key associated with the counter.
        value: Amount to increment counter by.
    """
    with self._lock:
        counters = self._record(key).counters
        counters[name] = counters.get(name, 0) + value
        self._trim()

add_time

add_time(name: str, key: KeyT, time_ms: float) -> None

Record a new time for an event.

Parameters:

• name (str) –

  Event or operation the time is for.

• key (KeyT) –

  Key associated with the event.

• time_ms (float) –

  The time in milliseconds of the event.

Source code in proxystore/store/metrics.py

def add_time(self, name: str, key: KeyT, time_ms: float) -> None:
    """Record a new time for an event.

    Args:
        name: Event or operation the time is for.
        key: Key associated with the event.
        time_ms: The time in milliseconds of the event.
    """
    with self._lock:
        times = self._record(key).times
        if name not in times:
            times[name] = TimeStats()
        times[name].add_time(time_ms)
        self._trim()

aggregate_times

aggregate_times() -> dict[str, TimeStats]

Aggregate time statistics over all keys.

This includes keys which per-key metrics are no longer kept for.

Returns:

• dict[str, TimeStats] –

  Dictionary mapping event names to the time statistics aggregated for that event.

Source code in proxystore/store/metrics.py

def aggregate_times(self) -> dict[str, TimeStats]:
    """Aggregate time statistics over all keys.

    This includes keys which per-key metrics are no longer kept for.

    Returns:
        Dictionary mapping event names to the time statistics aggregated \
        for that event.
    """
    with self._lock:
        times: dict[str, TimeStats] = defaultdict(TimeStats)
        for name, stats in self._dropped_times.items():
            times[name] += stats
        for metrics in self._metrics.values():
            for name, stats in metrics.times.items():
                times[name] += stats
    return times

get_metrics

get_metrics(key_or_proxy: KeyT | ProxyT) -> Metrics | None

Get the metrics associated with a key.

Parameters:

• key_or_proxy (KeyT | ProxyT) –

  Key to get associated metrics. If a proxy or sequence of proxies, the key(s) will be extracted.

Returns:

• Metrics | None –

  Metrics associated with the key or None if no metrics are kept for the key.

Source code in proxystore/store/metrics.py

def get_metrics(self, key_or_proxy: KeyT | ProxyT) -> Metrics | None:
    """Get the metrics associated with a key.

    Args:
        key_or_proxy: Key to get associated metrics. If a proxy or
            sequence of proxies, the key(s) will be extracted.

    Returns:
        Metrics associated with the key or `None` if no metrics are \
        kept for the key.
    """
    key = _normalize_key(key_or_proxy)
    with self._lock:
        metrics = self._metrics.get(key)
        return None if metrics is None else copy.deepcopy(metrics)