Function fetch_prometheus_metric_timeseries

python/ray/_common/test_utils.py:519–530

(
    prom_addresses: List[str],
    result: PrometheusTimeseries,
    timeout=None,
)

def fetch_prometheus_metric_timeseries(
    prom_addresses: List[str],
    result: PrometheusTimeseries,
    timeout=None,
) -> Dict[str, List[Any]]:
    samples = fetch_prometheus_timeseries(
        prom_addresses, result, timeout=timeout
    ).metric_samples.values()
    samples_by_name = defaultdict(list)
    for sample in samples:
        samples_by_name[sample.name].append(sample)
    return samples_by_name