exporter.py

"""
SPDX-FileCopyrightText: 2022-present Intel Corporation
SPDX-FileCopyrightText: 2020-present Open Networking Foundation <info@opennetworking.org>
SPDX-License-Identifier: Apache-2.0
"""

import os
import time
from prometheus_client import start_http_server, Gauge, REGISTRY, PROCESS_COLLECTOR, PLATFORM_COLLECTOR
import requests
import json

REGISTRY.unregister(PROCESS_COLLECTOR)
REGISTRY.unregister(PLATFORM_COLLECTOR)
# Unlike process and platform_collector gc_collector registers itself as three different collectors that have no corresponding public named variable. 
REGISTRY.unregister(REGISTRY._names_to_collectors['python_gc_objects_collected_total'])
#REGISTRY.unregister(REGISTRY._names_to_collectors['python_gc_uncollectable_objects_sum'])
#REGISTRY.unregister(REGISTRY._names_to_collectors['python_gc_collected_objects_sum'])

class Metrics:
    def __init__(self, app_port=3333, polling_interval_seconds=5):
        self.app_port = app_port
        self.polling_interval_seconds = polling_interval_seconds

        self.num_configured_devices = Gauge("num_configured_devices", "Number of configured devices")
        self.num_reachable_devices = Gauge("num_reachable_devices", "Number of reachable devices")

    def run_metrics_loop(self):
        while True:
            self.fetch()
            time.sleep(self.polling_interval_seconds)

    def fetch(self):
        resp = requests.get(url=f"http://localhost:{self.app_port}/devices")
        self.num_configured_devices.set(len(json.loads(resp.text)))

        resp = requests.get(url=f"http://localhost:{self.app_port}/devices/reachable")
        self.num_reachable_devices.set(len(json.loads(resp.text)))

def main():
    polling_interval_seconds = int(os.getenv("POLLING_INTERVAL_SECONDS", "5"))
    app_port = int(os.getenv("APP_PORT", "3333"))
    exporter_port = int(os.getenv("EXPORTER_PORT", "4444"))

    app_metrics = Metrics(
        app_port=app_port,
        polling_interval_seconds=polling_interval_seconds
    )
    start_http_server(exporter_port)
    app_metrics.run_metrics_loop()

if __name__ == "__main__":
    main()