prometheus-to-influx.py

#!/usr/bin/env python3

import influxdb
import requests
import json

PROMETHEUS_SERVER = ""
BEGIN_TIMESTAMP = 0
END_TIMESTAMP = 0

INFLUXDB_DATABASE = "prometheus_dump"

session = requests.Session()

def needed(series):
    return series.startswith("netdata_")

def get_series():
    print("Getting series")
    req = session.get(f"{PROMETHEUS_SERVER}/api/v1/label/__name__/values")
    req.raise_for_status()
    data = req.json()
    if data["status"] != "success":
        print("Unable to get series")
        exit()
    return data["data"]

def query_range(series: str, start: int, end: int, step: int):
    params = {
        "query": series,
        "start": start,
        "end": end,
        "step": step
    }
    req = session.get(f"{PROMETHEUS_SERVER}/api/v1/query_range", params = params)
    req.raise_for_status()
    data = req.json()
    if data["status"] != "success":
        print("Unable to query_range")
        exit()
    return data["data"]

def dump_all_data(series: str, start: int, end: int):
    result = []
    for time in range(start, end, 3600):
        print(f"Querying {series} from {time} to {time+3600-1}")
        yield query_range(series, time, time+3600-1, 1)

def chunks(l, n):
    """Yield successive n-sized chunks from l."""
    for i in range(0, len(l), n):
        yield l[i:i + n]

def add_data_to_influxdb(datas, client):
    datas = datas["result"]
    data_to_be_written = []
    for data in datas:
        measurement = data["metric"]["__name__"]
        tags = {
            "instance": data["metric"]["instance"],
            "chart": data["metric"]["chart"],
            "dimension": data["metric"]["dimension"],
            "family": data["metric"]["family"]
        }
        
        for value in data["values"]:
            data_to_be_written.append({
                "measurement": measurement,
                "tags": tags,
                "time": value[0] * 1000000000,
                "fields": {
                    "value": value[1]
                }
            })
    print(f"{len(data_to_be_written)} entries")
    for chunk in chunks(data_to_be_written, 50000):
        print("Writing chunk to influxdb")
        client.write_points(chunk)

client = influxdb.InfluxDBClient()
client.drop_database(INFLUXDB_DATABASE)
client.create_database(INFLUXDB_DATABASE)
client.switch_database(INFLUXDB_DATABASE)

series = get_series()
print(f"{len(series)} series available")
for serie in series:
    if not needed(serie):
        continue

    with open(f"{serie}.json", "w") as f:
        print(f"Getting series {serie}")
        for data in dump_all_data(serie, BEGIN_TIMESTAMP, END_TIMESTAMP):
            add_data_to_influxdb(data, client)