Example exporter for Smartoptics T-4900 / PacketLight PL1000

exporter.py

from prometheus_client import start_http_server
from prometheus_client.core import GaugeMetricFamily, CounterMetricFamily, REGISTRY

import math
import re
import requests
from urllib.parse import urlparse


class XpdrCollector(object):
    def __init__(self, target):
        self.target = target

    def collect(self):
        sys_html = requests.get(f'{self.target}/Pm_System.asp').text
        g = re.search(r'sfpConfDiag\("([^"]+)"\)', sys_html, re.MULTILINE)
        sys_performance = [x.split('#') for x in g.group(1).split('$')]
        g = re.search(r'oper[^(]+\(([^)]+)\)', sys_html, re.MULTILINE)
        sys_oper = [int(x) if x != 'ERROR' else None for x in g.group(1).replace('\r\n','').replace('\t','').replace(' ','').split(',')]
        g = re.search(r'admin[^(]+\(([^)]+)\)', sys_html, re.MULTILINE)
        sys_admin = [int(x) if x != 'ERROR' else None for x in g.group(1).replace('\r\n','').replace('\t','').replace(' ','').split(',')]

        sys_general_html = requests.get(f'{self.target}/Config_System_General.asp').text
        g = re.search(r'([0-9]+) °C', sys_general_html)
        temperature = int(g.group(1))

        panel_leds_html = requests.get(f'{self.target}/ProjectsConfig/PL1000_PanelLeds.asp').text
        g = re.search(r'DisplayLedTable\("([^"]+)"', panel_leds_html, re.MULTILINE)

        config_url = urlparse(requests.post(f'{self.target}/goform/formCfgLog', data={'get_config': '1'}).url)
        config = requests.get(f'{self.target}/{config_url.path}').text
        port_freqs = {}
        for m in re.finditer(r'sfpConfigXfpDesiredFreq\.([0-9]+)=([0-9]+)', config):
            port = int(m.group(1)) >> 16
            thz = float(m.group(2)) / 1000.0
            itu = int(thz*10 - 1900)
            nm = round(299792.458 / thz, 2)
            port_freqs[f'port{port}'] = {'thz': thz, 'itu': itu, 'nm': nm}

        color_names = {
                0: 'off',
                1: 'off',
                2: 'critical',
                3: 'warning',
                4: 'ok',
                5: 'ok',
                6: 'critical'
            }
        psu_status = {}
        port_status = {}
        for (idx, color, _, _) in ([int(y) for y in x.split('#')] for x in g.group(1).split('$') if x):
            slot = idx >> 24
            port = (idx >> 16) & 0xff
            path = idx & 0xff
            if port > 0 and port <= 8:
                port_status[f'port{port}'] = {'status': color_names[color] == 'ok'}
            if slot == 101:
                psu_status['psu1'] = color_names[color] == 'ok'
            if slot == 102:
                psu_status['psu2'] = color_names[color] == 'ok'

        for idx, pm in enumerate(sys_performance):
            if idx < 8:
                p = (int(pm[0]) >> 16) & 0xff
                port_status[f'port{p}']['lambda'] = int(pm[4])
                port_status[f'port{p}']['tx'] = int(pm[5])
                port_status[f'port{p}']['rx'] = int(pm[6])
                port_status[f'port{p}']['temp'] = int(pm[7])
                port_status[f'port{p}']['oper'] = sys_oper[p] == 1
                port_status[f'port{p}']['admin'] = sys_admin[p] != 2


        yield GaugeMetricFamily('xpdr_system_temperature_celsius', 'System temperature', value=temperature)
        psu_status_f = GaugeMetricFamily('xpdr_psu_status', 'PSU status', labels=['psu'])
        for k, v in psu_status.items():
            psu_status_f.add_metric([k], 1.0 if v else 0.0)
        yield psu_status_f
        port_oper_f = GaugeMetricFamily('xpdr_port_oper_status', 'Port operational status', labels=['port'])
        port_admin_f = GaugeMetricFamily('xpdr_port_admin_status', 'Port administrative status', labels=['port'])
        port_rx_f = GaugeMetricFamily('xpdr_port_rx_power_dbm', 'Port RX power', labels=['port'])
        port_tx_f = GaugeMetricFamily('xpdr_port_tx_power_dbm', 'Port TX power', labels=['port'])
        port_temp_f = GaugeMetricFamily('xpdr_port_temperature_celsius', 'Port SFP temperature', labels=['port'])
        port_lambda_f = GaugeMetricFamily('xpdr_port_wavelength_nm', 'Port wavelength', labels=['port'])
        port_d_lambda_f = GaugeMetricFamily('xpdr_port_desired_wavelength_nm', 'Desired wavelength', labels=['port'])
        port_d_thz_f = GaugeMetricFamily('xpdr_port_desired_wavelength_thz', 'Desired wavelength in THz', labels=['port'])
        port_d_itu_f = GaugeMetricFamily('xpdr_port_desired_wavelength_itu', 'Desired wavelength in ITU channel number', labels=['port'])
        for k, v in port_status.items():
            port_oper_f.add_metric([k], 1.0 if v['oper'] else 0.0)
            port_admin_f.add_metric([k], 1.0 if v['admin'] else 0.0)
            port_rx_f.add_metric([k], 10.0 * math.log10(float(v['rx']) / 10000.0) if v['rx'] > 0 else float('NaN'))
            port_tx_f.add_metric([k], 10.0 * math.log10(float(v['tx']) / 10000.0) if v['tx'] > 0 else float('NaN'))
            port_temp_f.add_metric([k], float(v['temp'] / 1000.0))
            port_lambda_f.add_metric([k], float(v['lambda']) / 100.0)
            if k in port_freqs:
                port_d_lambda_f.add_metric([k], port_freqs[k]['nm'])
                port_d_thz_f.add_metric([k], port_freqs[k]['thz'])
                port_d_itu_f.add_metric([k], port_freqs[k]['itu'])
        yield port_lambda_f
        yield port_tx_f
        yield port_rx_f
        yield port_temp_f
        yield port_admin_f
        yield port_oper_f
        yield port_d_lambda_f
        yield port_d_thz_f
        yield port_d_itu_f


if __name__ == '__main__':
    import sys
    import signal
    REGISTRY.register(XpdrCollector(sys.argv[2]))
    start_http_server(int(sys.argv[1]))
    signal.pause()