upload_results_to_prometheus.py

# Parse the output of a duplicati run and upload to prometheus.  Assumes duplicati is running in kubernetes and the gateway is as well.
# Also requires that duplicati is run with the following commands:
#   --run-script-after=/scripts/upload_results_to_prometheus.sh --run-script-result-output-format=json

import os
import prometheus_client
import json

json_filename = os.environ.get('DUPLICATI__RESULTFILE', '/tmp/latest_duplicati_resultfile.json')

if json_filename is None:
    raise Exception(f'Failed to get $DUPLICATI__RESULTFILE! ({json_filename})')

contents = {}
with open(json_filename, 'rt') as fp:
    contents = json.load(fp)


data = contents.get('Data', {})

prefix = "duplicati"

def do_flatten(obj, local_prefix, result_list):

    if isinstance(obj, dict):
        for k, v in obj.items():
            key = f'{local_prefix}_{k}'.lower()

            do_flatten(v, f'{key}', result_list)

    elif isinstance(obj, list):
        return

    elif isinstance(obj, (int, str)) or obj is None:
        result_list.append((local_prefix, obj))
    else:
        raise Exception(f"Unhandled type: {type(obj)}")

flattened = []
do_flatten(contents, local_prefix=prefix, result_list=flattened)

prom_registry = prometheus_client.CollectorRegistry()

for item in flattened:
    name, value = item
    g = prometheus_client.Gauge(name, name, registry=prom_registry)
    g.set_to_current_time()

prometheus_client.push_to_gateway('prometheus-pushgateway.prometheus.svc.cluster.local.:9091', job="batch", registry=prom_registry)

print(f'Done sending {len(flattened):,} metrics.')