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 = Union[ConnectorKeyT, Sequence[ConnectorKeyT]]

Key types supported by StoreMetrics.

ProxyT module-attribute

ProxyT = Union[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 = math.inf,
    max_time_ms: float = 0,
    last_time_ms: float = 0,
    last_timestamp: float = 0,
)

Tracks time statistics of a reoccuring event.

Attributes:

• count (int) –

  Number of times this event as 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()

Record and query metrics on Store operations.

Source code in proxystore/store/metrics.py

def __init__(self) -> None:
    self._metrics: dict[int, Metrics] = defaultdict(Metrics)

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.
    """
    self._metrics[_hash_key(key)].attributes[name] = value

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.
    """
    counters = self._metrics[_hash_key(key)].counters
    if name in counters:
        counters[name] += value
    else:
        counters[name] = value

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.
    """
    times = self._metrics[_hash_key(key)].times
    if name not in times:
        times[name] = TimeStats()
    times[name].add_time(time_ms)

aggregate_times()

aggregate_times() -> dict[str, TimeStats]

Aggregate time statistics over all keys.

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.

    Returns:
        Dictionary mapping event names to the time statistics aggregated \
        for that event.
    """
    times: dict[str, TimeStats] = defaultdict(TimeStats)
    for metrics in self._metrics.values():
        for key, value in metrics.times.items():
            times[key] += value
    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 the key does not exist.

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 the key does not \
        exist.
    """
    key_hash = _hash_key(key_or_proxy)
    if key_hash in self._metrics:
        return copy.deepcopy(self._metrics[key_hash])
    return None