Function fetch_prometheus_timeseries

python/ray/_common/test_utils.py:482–500

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

def fetch_prometheus_timeseries(
    prom_addreses: List[str],
    result: PrometheusTimeseries,
    timeout=None,
) -> PrometheusTimeseries:
    components_dict, metric_descriptors, metric_samples = fetch_prometheus(
        prom_addreses, timeout=timeout
    )
    for address, components in components_dict.items():
        if address not in result.components_dict:
            result.components_dict[address] = set()
        result.components_dict[address].update(components)
    result.metric_descriptors.update(metric_descriptors)
    for sample in metric_samples:
        # udpate sample to the latest value
        result.metric_samples[
            frozenset(list(sample.labels.items()) + [("_metric_name_", sample.name)])
        ] = sample
    return result


def fetch_prometheus_metrics(prom_addresses: List[str]) -> Dict[str, List[Any]]: