VictoriaMetrics as push gateway.

victoria_metrics_push_gw.py

from prometheus_client import Counter, start_http_server
from threading import Thread
import requests as re
import time

JOB_NAME = 'test'
INSTANCE = 'localhost'


def scrape_and_send(local_url: str, vm_url: str, scrape_interval: int):
    s = re.Session()
    scrape_cnt = Counter('self_scrapes_ok_total', 'some help')
    push_cnt = Counter('self_vm_push_ok_total', 'some help')
    vm_url_parametrized = f"{vm_url}?extra_label=job={JOB_NAME}&extra_label=instance={INSTANCE}"
    while True:
        try:
            time.sleep(scrape_interval)
            with s.get(local_url, timeout=3) as local_data:
                if local_data.status_code != 200:
                    print(f"bad status code, want 200, got {local_data.status_code}")
                    continue
                scrape_cnt.inc()
                with s.post(vm_url_parametrized, data=local_data.text, timeout=5) as vm_resp:
                    if vm_resp.status_code != 204:
                        print(f"bad status code, want 204, got {vm_resp.status_code}")
                        continue
                    push_cnt.inc()
        except Exception as e:
            print(f"error occurred: {e}")


def push_to_vm(local_url: str, vm_url: str, scrape_interval=None) -> Thread:
    if not scrape_interval:
        # scrape interval 10s by default
        scrape_interval = 10
    th = Thread(target=scrape_and_send, args=(local_url, vm_url, scrape_interval,))
    th.start()
    return th


push_to_vm("http://localhost:8000", "http://localhost:8428/api/v1/import/prometheus")
start_http_server(8000)

from prometheus_client.exposition import (
    default_handler,
    basic_auth_handler,
    generate_latest,
    CONTENT_TYPE_LATEST,
)

# url == "http://vika:8428/api/v1/import/prometheus"
def push_to_victoriametrics(url, job, registry, timeout=30, handler=default_handler):
    url = f"{url}?extra_label=job={job}"  # &extra_label=instance={INSTANCE}
    data = generate_latest(registry)
    handler(
        url=url,
        method="POST",
        timeout=timeout,
        headers=[("Content-Type", CONTENT_TYPE_LATEST)],
        data=data,
    )()