Delete stale metrics from pushgateway

delete-stale-metrics-from-pushgateway.py

from datetime import datetime, timedelta
import os
import requests
from prometheus_client.parser import text_string_to_metric_families


PUSHGATEWAY_METRICS_URL = "https://xxx.com/metrics"
PUSH_TIME_METRICS_NAME = "push_time_seconds"


def is_metrics_stale(timestamp: float, retention_period_days: int = 7) -> bool:
    """Return True if the metrics is stale.

    Args:
        timestamp (float): epoch timestamp in prom format. ex: 1701841866.00605802

    Returns:
        bool: True if the stale than RETENTION_PERIOD_DAYS(days).
    """
    return datetime.fromtimestamp(timestamp) < datetime.now() - timedelta(days=retention_period_days)


if __name__ == "__main__":
    with requests.session() as session:
        metrics_list_response = session.get(PUSHGATEWAY_METRICS_URL)
        metrics_list_response.raise_for_status()
        for family in text_string_to_metric_families(metrics_list_response.text):
            for sample in family.samples:
                if sample.name.startswith(PUSH_TIME_METRICS_NAME):
                    if is_metrics_stale(sample.value):
                        url = f"{PUSHGATEWAY_METRICS_URL}/job/{sample.labels['job']}/instance/{sample.labels['instance']}"
                        res = session.delete(url)
                        res.raise_for_status()
                        print(f"DELETE {url}")