radio_exporter/src/PrometheusServer.py

import socket
import signal
from ipaddress import ip_address
from http.server import HTTPServer, BaseHTTPRequestHandler
from selectors import DefaultSelector, EVENT_READ

from src import Collector


class PromServer(HTTPServer):
    def __init__(self, address):
        if ip_address(address[0]).version == 6:
            self.address_family = socket.AF_INET6
        super().__init__(address, self.RequestHandler)

    class RequestHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            if self.path != "/metrics":
                return self.send_error(404)

            metrics = [x.format_prom() for x in self.server.collector.collect()]

            output = "\n".join(metrics) + "\n"
            self.send_response(200)
            self.end_headers()
            self.wfile.write(output.encode())

    def signal_handler(self, signum, frame):
        print("Signal handler called with signal", signum)
        self.interrupt_write.send(b"\0")

    def serve_forever(self):
        sel = DefaultSelector()
        sel.register(self.interrupt_read, EVENT_READ)
        sel.register(self, EVENT_READ)
        while True:
            for key, _ in sel.select():
                if key.fileobj == self.interrupt_read:
                    self.interrupt_read.recv(1)
                    return
                if key.fileobj == self:
                    self.handle_request()

    def start(self):
        self.collector = Collector.Collector()
        self.interrupt_read, self.interrupt_write = socket.socketpair()
        signal.signal(signal.SIGINT, self.signal_handler)
        self.serve_forever()