prometheus stats cli tool

src/capport/stats.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import ipaddress
+import sys
+import typing
+import time
+
+import trio
+
+import capport.utils.ipneigh
+import capport.utils.nft_set
+from . import cptypes
+
+
+def print_metric(name: str, mtype: str, value, *, now: typing.Optional[int]=None, help: typing.Optional[str]=None):
+    # no labels in our names for now, always print help and type
+    if help:
+        print(f'# HELP {name} {help}')
+    print(f'# TYPE {name} {mtype}')
+    if now:
+        print(f'{name} {value} {now}')
+    else:
+        print(f'{name} {value}')
+
+
+async def amain(client_ifname: str):
+    ns = capport.utils.nft_set.NftSet()
+    captive_allowed_entries: typing.Set[cptypes.MacAddress] = {
+        entry['mac']
+        for entry in ns.list()
+    }
+    seen_allowed_entries: typing.Set[cptypes.MacAddress] = set()
+    total_ipv4 = 0
+    total_ipv6 = 0
+    unique_clients = set()
+    unique_ipv4 = set()
+    unique_ipv6 = set()
+    async with capport.utils.ipneigh.connect() as ipn:
+        ipn.ip.strict_check = True
+        async for (mac, addr) in ipn.dump_neighbors(client_ifname):
+            if mac in captive_allowed_entries:
+                seen_allowed_entries.add(mac)
+            unique_clients.add(mac)
+            if isinstance(addr, ipaddress.IPv4Address):
+                total_ipv4 += 1
+                unique_ipv4.add(mac)
+            else:
+                total_ipv6 += 1
+                unique_ipv6.add(mac)
+    print_metric('capport_allowed_macs', 'gauge', len(captive_allowed_entries), help='Number of allowed client mac addresses')
+    print_metric('capport_allowed_neigh_macs', 'gauge', len(seen_allowed_entries), help='Number of allowed client mac addresses seen in neighbor cache')
+    print_metric('capport_unique', 'gauge', len(unique_clients), help='Number of clients (mac addresses) in client network seen in neighbor cache')
+    print_metric('capport_unique_ipv4', 'gauge', len(unique_ipv4), help='Number of IPv4 clients (unique per mac) in client network seen in neighbor cache')
+    print_metric('capport_unique_ipv6', 'gauge', len(unique_ipv6), help='Number of IPv4 clients (unique per mac) in client network seen in neighbor cache')
+    print_metric('capport_total_ipv4', 'gauge', total_ipv4, help='Number of IPv4 addresses seen in neighbor cache')
+    print_metric('capport_total_ipv6', 'gauge', total_ipv6, help='Number of IPv6 addresses seen in neighbor cache')
+
+
+def main():
+    assert len(sys.argv) == 2, "Need name of client interface as argument"
+    trio.run(amain, sys.argv[1])