vllm.distributed.ec_transfer.ec_connector.metrics

Classes:

• ECConnectorLogging –
• ECConnectorProm –

  Support for registering per-connector Prometheus metrics, and

• ECConnectorPromMetrics –

  A base class for per-connector Prometheus metric registration

• ECConnectorStats –

  Base class for EC Connector Stats, a container for transfer performance

ECConnectorLogging

Methods:

• log –

  Log transfer metrics periodically, similar to throughput logging

Source code in vllm/distributed/ec_transfer/ec_connector/metrics.py

class ECConnectorLogging:
    def __init__(self, ec_transfer_config: ECTransferConfig | None):
        # Instantiate the connector's stats class.
        if ec_transfer_config and ec_transfer_config.ec_connector:
            self.connector_cls = ECConnectorFactory.get_connector_class(
                ec_transfer_config
            )
        self.reset()

    def reset(self):
        self.transfer_stats_accumulator: ECConnectorStats | None = None

    def observe(self, transfer_stats_data: dict[str, Any]):
        # Should not be called when an ECConnector is not configured.
        assert self.connector_cls is not None
        # Called periodically when connector syncs with the scheduler.
        # Note that this is not the same as the logging interval.
        # We expect transfer_stats_data to be aggregated across all workers and
        # consist of observations from a single connector.
        transfer_stats = self.connector_cls.build_ec_connector_stats(
            transfer_stats_data
        )
        if transfer_stats is None:
            logger.warning_once(
                "The connector %s is collecting stats but "
                "does not implement the "
                "`build_ec_connector_stats` method. "
                "Stats will not be logged.",
                self.connector_cls,
            )
            return

        if self.transfer_stats_accumulator is None:
            self.transfer_stats_accumulator = transfer_stats
        else:
            # Accumulate last interval stats.
            self.transfer_stats_accumulator = self.transfer_stats_accumulator.aggregate(
                transfer_stats
            )

    def log(self, log_fn=logger.info):
        """Log transfer metrics periodically, similar to throughput logging"""
        if (
            self.transfer_stats_accumulator
            and not self.transfer_stats_accumulator.is_empty()
        ):
            # Produce a single cumulative stats object for the last time
            # interval from the recorded observations.
            xfer_metrics = self.transfer_stats_accumulator.reduce()
            xfer_metrics_str = ", ".join(f"{k}={v}" for k, v in xfer_metrics.items())
            log_fn("EC Transfer metrics: %s", xfer_metrics_str)

            # Reset metrics for next interval
            self.reset()

log(log_fn=logger.info)

Log transfer metrics periodically, similar to throughput logging

Source code in vllm/distributed/ec_transfer/ec_connector/metrics.py

def log(self, log_fn=logger.info):
    """Log transfer metrics periodically, similar to throughput logging"""
    if (
        self.transfer_stats_accumulator
        and not self.transfer_stats_accumulator.is_empty()
    ):
        # Produce a single cumulative stats object for the last time
        # interval from the recorded observations.
        xfer_metrics = self.transfer_stats_accumulator.reduce()
        xfer_metrics_str = ", ".join(f"{k}={v}" for k, v in xfer_metrics.items())
        log_fn("EC Transfer metrics: %s", xfer_metrics_str)

        # Reset metrics for next interval
        self.reset()

ECConnectorProm

Support for registering per-connector Prometheus metrics, and recording transfer statistics to those metrics. Uses ECConnectorBase.build_prom_metrics().

Source code in vllm/distributed/ec_transfer/ec_connector/metrics.py

class ECConnectorProm:
    """Support for registering per-connector Prometheus metrics, and
    recording transfer statistics to those metrics. Uses
    ECConnectorBase.build_prom_metrics().
    """

    _gauge_cls = Gauge
    _counter_cls = Counter
    _histogram_cls = Histogram

    def __init__(
        self,
        vllm_config: VllmConfig,
        labelnames: list[str],
        per_engine_labelvalues: dict[int, list[object]],
    ):
        self.prom_metrics: ECConnectorPromMetrics | None = None
        ec_transfer_config = vllm_config.ec_transfer_config
        if ec_transfer_config and ec_transfer_config.ec_connector:
            connector_cls = ECConnectorFactory.get_connector_class(ec_transfer_config)
            metric_types = {
                Gauge: self._gauge_cls,
                Counter: self._counter_cls,
                Histogram: self._histogram_cls,
            }
            self.prom_metrics = connector_cls.build_prom_metrics(
                vllm_config,
                metric_types,
                labelnames,
                per_engine_labelvalues,
            )

    def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
        if self.prom_metrics is None:
            return
        self.prom_metrics.observe(transfer_stats_data, engine_idx)

ECConnectorPromMetrics

A base class for per-connector Prometheus metric registration and recording.

Methods:

• observe –

  Record the supplied transfer statistics to Prometheus metrics. These

Source code in vllm/distributed/ec_transfer/ec_connector/metrics.py

class ECConnectorPromMetrics:
    """A base class for per-connector Prometheus metric registration
    and recording.
    """

    def __init__(
        self,
        vllm_config: VllmConfig,
        metric_types: dict[type[PromMetric], type[PromMetricT]],
        labelnames: list[str],
        per_engine_labelvalues: dict[int, list[object]],
    ):
        self._ec_transfer_config = vllm_config.ec_transfer_config
        self._gauge_cls = metric_types[Gauge]
        self._counter_cls = metric_types[Counter]
        self._histogram_cls = metric_types[Histogram]
        self._labelnames = labelnames
        self.per_engine_labelvalues = per_engine_labelvalues

    def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
        """Record the supplied transfer statistics to Prometheus metrics. These
        statistics are engine-specific, and should be recorded to a metric
        with the appropriate 'engine' label. These metric instances can be
        created using the create_metric_per_engine() helper method.
        """
        raise NotImplementedError

observe(transfer_stats_data, engine_idx=0)

Record the supplied transfer statistics to Prometheus metrics. These statistics are engine-specific, and should be recorded to a metric with the appropriate 'engine' label. These metric instances can be created using the create_metric_per_engine() helper method.

Source code in vllm/distributed/ec_transfer/ec_connector/metrics.py

def observe(self, transfer_stats_data: dict[str, Any], engine_idx: int = 0):
    """Record the supplied transfer statistics to Prometheus metrics. These
    statistics are engine-specific, and should be recorded to a metric
    with the appropriate 'engine' label. These metric instances can be
    created using the create_metric_per_engine() helper method.
    """
    raise NotImplementedError

ECConnectorStats dataclass

Base class for EC Connector Stats, a container for transfer performance metrics or otherwise important telemetry from the connector. All sub-classes need to be serializable as stats are sent from worker to logger process.

Methods:

• aggregate –

  Aggregate stats with another ECConnectorStats object.

• is_empty –

  Return True if the stats are empty.

• reduce –

  Reduce the observations collected during a time interval to one or

• reset –

  Reset the stats, clear the state.

Source code in vllm/distributed/ec_transfer/ec_connector/metrics.py

@dataclass
class ECConnectorStats:
    """Base class for EC Connector Stats, a container for transfer performance
    metrics or otherwise important telemetry from the connector.
    All sub-classes need to be serializable as stats are sent from worker to
    logger process.
    """

    data: dict[str, Any] = field(default_factory=dict)

    def reset(self):
        """Reset the stats, clear the state."""
        raise NotImplementedError

    def aggregate(self, other: "ECConnectorStats") -> "ECConnectorStats":
        """Aggregate stats with another `ECConnectorStats` object."""
        raise NotImplementedError

    def reduce(self) -> dict[str, int | float]:
        """Reduce the observations collected during a time interval to one or
        more representative values (eg avg/median/sum of the series).
        This is meant to be called by the logger to produce a summary of the
        stats for the last time interval.
        """
        raise NotImplementedError

    def is_empty(self) -> bool:
        """Return True if the stats are empty."""
        raise NotImplementedError

aggregate(other)

Aggregate stats with another ECConnectorStats object.

Source code in vllm/distributed/ec_transfer/ec_connector/metrics.py

def aggregate(self, other: "ECConnectorStats") -> "ECConnectorStats":
    """Aggregate stats with another `ECConnectorStats` object."""
    raise NotImplementedError

is_empty()

Return True if the stats are empty.

Source code in vllm/distributed/ec_transfer/ec_connector/metrics.py

def is_empty(self) -> bool:
    """Return True if the stats are empty."""
    raise NotImplementedError

reduce()

Reduce the observations collected during a time interval to one or more representative values (eg avg/median/sum of the series). This is meant to be called by the logger to produce a summary of the stats for the last time interval.

Source code in vllm/distributed/ec_transfer/ec_connector/metrics.py

def reduce(self) -> dict[str, int | float]:
    """Reduce the observations collected during a time interval to one or
    more representative values (eg avg/median/sum of the series).
    This is meant to be called by the logger to produce a summary of the
    stats for the last time interval.
    """
    raise NotImplementedError

reset()

Reset the stats, clear the state.

Source code in vllm/distributed/ec_transfer/ec_connector/metrics.py

def reset(self):
    """Reset the stats, clear the state."""
    raise NotImplementedError