Skip to content

Instantly share code, notes, and snippets.

@sorz
Last active April 14, 2022 08:48
Show Gist options
  • Save sorz/c5115fbef0aa45585d2408e946c90fa0 to your computer and use it in GitHub Desktop.
Save sorz/c5115fbef0aa45585d2408e946c90fa0 to your computer and use it in GitHub Desktop.
Turn APC UPS status `apcaccess status` to OpenMetrics (prometheus) exporter with zero-dependency Python script & systemd socket.
#!/usr/bin/env python3
import sys
import socket
from subprocess import Popen, PIPE
def main():
sock = socket.fromfd(3, socket.AF_INET, socket.SOCK_STREAM)
while True:
conn, addr = sock.accept()
print("Connection from", addr)
try:
handle_client(conn)
except Exception as e:
print(e, file=sys.stderr)
def handle_client(sock):
proc = Popen(
['/usr/bin/apcaccess', 'status'],
stdout=PIPE,
text=True,
env=dict(),
)
outs, _ = proc.communicate(timeout=5)
kvs = (l.split(':', 1) for l in outs.split('\n') if ':' in l)
kvs = { k.strip(): v.strip() for k, v in kvs }
f = sock.makefile('rw', encoding='utf-8')
f.write("HTTP/1.0 200 OK\r\n")
f.write("Server: apcmetrics\r\n")
f.write("Content-Type: application/openmetrics-text; version=1.0.0; charset=utf-8\r\n\r\n")
def parse_float(value):
if ' ' in value:
value, _unit = value.split(' ', 1)
return float(value)
def parse_int(value):
return int(parse_float(value))
def metric(name, value_fn, typ, help, unit=None):
value = value_fn(kvs[name])
path = f"apcmetrics_{name.lower()}"
if unit is not None:
path += f"_{unit}"
f.write(f"# HELP {path} {help}\n")
f.write(f"# TYPE {path} {typ}\n")
if unit is not None:
f.write(f"# UNIT {path} {unit}\n")
f.write(f"{path}{{host=\"{kvs['HOSTNAME']}\"}} {value}\n")
metric('LINEV', parse_float, 'gauge',
'Current input line voltage', 'volt')
metric('LOADPCT', parse_float, 'gauge',
'Percentage of UPS load capacity used as estimated by UPS', 'precent')
metric('BCHARGE', parse_float, 'gauge',
'Current battery capacity charge percentage', 'precent')
metric('TIMELEFT', parse_float, 'gauge',
'Remaining runtime left on battery as estimated by the UPS', 'minutes')
metric('BATTV', parse_float, 'gauge',
'Current battery voltage', 'volts')
metric('NUMXFERS', parse_int, 'gauge',
'Number of transfers to battery since apcupsd startup')
metric('TONBATT', parse_int, 'gauge',
'Seconds currently on battery', 'seconds')
metric('CUMONBATT', parse_int, 'gauge',
'Cumulative seconds on battery since apcupsd startup', 'seconds')
f.write("# EOF")
f.flush()
sock.shutdown(socket.SHUT_RDWR)
sock.close()
if __name__ == '__main__':
main()
[Unit]
Description=APC USP OpenMetrics explorer
[Service]
Type=simple
DynamicUser=yes
WorkingDirectory=/opt/apcmetrics
ExecStart=/usr/bin/python apcmetrics.py
Environment=PYTHONUNBUFFERED=1
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelModules=true
ProtectControlGroups=true
ProtectKernelTunables=true
ProtectProc=invisible
ProtectHostname=true
RestrictNamespaces=true
SystemCallArchitectures=native
ProtectClock=true
ProtectKernelLogs=true
MemoryDenyWriteExecute=true
PrivateUsers=true
RestrictAddressFamilies=AF_INET
RestrictRealtime=true
LockPersonality=true
SystemCallFilter=@system-service @basic-io @network-io @io-event @aio @file-system
SystemCallFilter=~@resources @privileged
CapabilityBoundingSet=
[Unit]
Description=Listen on http for APC UPS metrics.
[Socket]
ListenStream=127.0.0.80:3060
Accept=no
[Install]
WantedBy=sockets.target
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment