From d44e4021fe879230adb762972b9080d021176146 Mon Sep 17 00:00:00 2001 From: nico Date: Tue, 28 Apr 2020 14:11:49 +0200 Subject: code cleanup * split up api and metrics class * revamped file nameming to better resemble their function * update prometheus and influx files * fixed version regex to not break on - in version string --- api.py | 71 +++++++++++ ejabberdrpc.py | 373 --------------------------------------------------------- influx.py | 2 +- metrics.py | 322 +++++++++++++++++++++++++++++++++++++++++++++++++ prometheus.py | 2 +- 5 files changed, 395 insertions(+), 375 deletions(-) create mode 100644 api.py delete mode 100755 ejabberdrpc.py create mode 100644 metrics.py mode change 100755 => 100644 prometheus.py diff --git a/api.py b/api.py new file mode 100644 index 0000000..ccd63e4 --- /dev/null +++ b/api.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +import re + +from packaging import version + + +class EjabberdApi: + """ + class to interact with the ejabberd rest/ xmlrpc api + """ + def __init__(self, url, login=None, api: str = "rpc"): + # api variables + self._login = login + self._url = url + + if api == "rpc": + self.cmd = self._rpc + else: + import requests + + self.session = requests.Session() + self.cmd = self._rest + + @property + def _auth(self) -> (str, None): + if self._login is not None: + return f"{self._login['user']}@{self._login['server']}", self._login['password'] + return None + + @property + def verstring(self): + if self._login is not None: + ver_str = re.compile('([1-9][0-9.]*)') + status = self.cmd('status', {}) + + # matches + tmp = ver_str.findall(status)[0] + + # return parsed version string + return version.parse(tmp) + + return None + + def _rest(self, command: str, data) -> dict: + # add authentication header to the session obj + if self.session.auth is None: + self.session.auth = self._auth + + # post + r = self.session.post('/'.join([self._url, command]), json=data) + + # proceed if response is ok + if r.ok: + return r.json() + + return {} + + def _rpc(self, command: str, data): + from xmlrpc import client + + with client.ServerProxy(self._url) as server: + fn = getattr(server, command) + try: + if self._login is not None: + return fn(self._login, data) + return fn(data) + + except: + # this needs to be more specific + return {} diff --git a/ejabberdrpc.py b/ejabberdrpc.py deleted file mode 100755 index 6a34cd0..0000000 --- a/ejabberdrpc.py +++ /dev/null @@ -1,373 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- -import ipaddress -import re - -from packaging import version - -# rfc6052: IPv6 Addressing of IPv4/IPv6 Translators -nat64 = ipaddress.ip_network("64:ff9b::/96") - - -class EjabberdMetrics: - """ - class to fetch metrics per xmlrpc - """ - def __init__(self, url, login=None, api="rpc", muc_host: str = 'conference'): - self._login = login - self.muc_host = muc_host - if api == "rpc": - self.url = url - self._cmd = self._rpc - else: - import requests - - self._url = url - self.session = requests.Session() - self._cmd = self._rest - - @property - def _auth(self): - if self._login is not None: - return f"{self._login['user']}@{self._login['server']}", self._login['password'] - return None - - @property - def _verstring(self): - if self._login is not None: - ver_str = re.compile('([1-9][0-9.-]*)') - status = self._cmd('status', {}) - - # matches - tmp = ver_str.findall(status)[0] - - # return parsed version string - return version.parse(tmp) - - return None - - def _rest(self, command: str, data) -> dict: - # add authentication header to the session obj - if self.session.auth is None: - self.session.auth = self._auth - - # post - r = self.session.post('/'.join([self._url, command]), json=data) - - # proceed if response is ok - if r.ok: - return r.json() - - return {} - - def _rpc(self, command: str, data): - from xmlrpc import client - - with client.ServerProxy(self.url) as server: - fn = getattr(server, command) - try: - if self._login is not None: - return fn(self._login, data) - return fn(data) - except: - return {} - - def _client(self, resource): - clientmap = { - "Conv6ations for Sum7": ["Conversations with IPv6"], - "Conversations": [], - "Pix-Art Messenger": [], - "Gajim": ["gajim"], - "Psi+": [], - "jitsi": [], - "Dino": ["dino"], - "poezio": [], - "profanity": [], - "Xabber": [], - "ChatSecure": ["chatsecure"] - } - - for client, names in clientmap.items(): - for c in names: - if c in resource: - return client - if client in resource: - return client - return "other" - - @staticmethod - def _ipversion(ip): - addr = ipaddress.ip_address(ip) - if addr.version == 6: - if addr.ipv4_mapped: - return 4 - if addr in nat64: - return 4 - return addr.version - - def fetch_onlineuser(self): - tmp = self._cmd("connected_users_info", {}) - if "connected_users_info" not in tmp: - return tmp - data = [] - for c in tmp["connected_users_info"]: - if "session" not in c: - continue - user = {} - for attrs in c["session"]: - for k, v in attrs.items(): - user[k] = v - data.append(user) - return data - - def fetch_nodes(self): - result = self._cmd("list_cluster", {}) - if "nodes" not in result: - return result - data = [] - for node in result["nodes"]: - data.append(node["node"]) - return data - - def fetch_vhosts(self): - result = self._cmd("registered_vhosts", {}) - if "vhosts" not in result: - return result - data = [] - for vhost in result["vhosts"]: - data.append(vhost["vhost"]) - return data - - def fetch_s2s_in(self): - result = self._cmd("incoming_s2s_number", {}) - if "s2s_incoming" not in result: - return result - return result["s2s_incoming"] - - def fetch_s2s_out(self): - result = self._cmd("outgoing_s2s_number", {}) - if "s2s_outgoing" not in result: - return result - return result["s2s_outgoing"] - - def fetch_registered(self, vhost=None): - if vhost is None: - result = self._cmd("stats", {"name":"registeredusers"}) - if "stat" in result: - return result["stat"] - else: - result = self._cmd("stats_host", {"name":"registeredusers", "host": vhost}) - if "stat" in result: - return result["stat"] - - def fetch_muc(self, vhost=None): - host = "global" - if vhost is not None: - if self._verstring.major >= 19: - host = '.'.join([self.muc_host, vhost]) - else: - host = vhost - result = self._cmd("muc_online_rooms", {"host": host}) - if "rooms" in result: - return len(result["rooms"]) - return len(result) - - def update(self): - # nodes - self._nodes = self.fetch_nodes() - - # vhosts - self._vhosts = self.fetch_vhosts() - - # registered - if not hasattr(self, "_registered"): - self._registered = {} - self._registered[None] = self.fetch_registered() - - # muc - if not hasattr(self, "_muc"): - self._muc = {} - self._muc[None] = self.fetch_muc() - - # registered + muc - for vhost in self._vhosts: - self._registered[vhost] = self.fetch_registered(vhost) - self._muc[vhost] = self.fetch_muc(vhost) - - # online user - self._onlineuser = self.fetch_onlineuser() - - # s2s - self._s2s_in = self.fetch_s2s_in() - self._s2s_out = self.fetch_s2s_out() - - def get_online_by(self, by="node", parse=None, vhost=None, node=None): - parser = parse or (lambda a: a) - if not hasattr(self, "_onlineuser"): - self._onlineuser = self.fetch_onlineuser() - data = {} - - for conn in self._onlineuser: - if vhost is not None and vhost not in conn["jid"]: - continue - if node is not None and node != conn["node"]: - continue - if by not in conn: - continue - value = parser(conn[by]) - if value not in data: - data[value] = 1 - else: - data[value] += 1 - return data - - def get_online_by_node(self, vhost=None): - return self.get_online_by("node", vhost=vhost) - - def get_online_by_vhost(self, node=None): - return self.get_online_by("jid", parse=lambda jid: jid[jid.find("@")+1:jid.find("/")], node=node) - - def get_online_by_status(self, vhost=None, node=None): - return self.get_online_by("status", vhost=vhost, node=node) - - def get_online_by_connection(self, vhost=None, node=None): - return self.get_online_by("connection", vhost=vhost, node=node) - - def get_online_by_client(self, vhost=None, node=None): - return self.get_online_by("resource", parse=self._client, vhost=vhost, node=node) - - def get_online_by_ipversion(self, vhost=None, node=None): - return self.get_online_by("ip", parse=self._ipversion, vhost=vhost, node=node) - - def get_online_client_by(self, by="ip", parse=None, vhost=None, node=None): - parser = parse or self._ipversion - if not hasattr(self, "_onlineuser"): - self._onlineuser = self.fetch_onlineuser() - data = {} - - for conn in self._onlineuser: - client = "other" - if "resource" in conn: - client = self._client(conn["resource"]) - if client not in data: - data[client] = {} - if vhost is not None and vhost not in conn["jid"]: - continue - if node is not None and node != conn["node"]: - continue - if by not in conn: - continue - value = parser(conn[by]) - if value not in data[client]: - data[client][value] = 1 - else: - data[client][value] += 1 - return data - - def get_online_client_by_ipversion(self, vhost=None, node=None): - return self.get_online_client_by("ip", parse=self._ipversion, vhost=vhost, node=node) - - def get_registered(self, vhost=None): - if not hasattr(self, "_registered"): - self._registered = {} - if vhost not in self._registered: - self._registered[vhost] = self.fetch_registered(vhost) - return self._registered[vhost] - - def get_muc(self, vhost=None): - if not hasattr(self, "_muc"): - self._muc = {} - if vhost not in self._muc: - self._muc[vhost] = self.fetch_muc(vhost) - return self._muc[vhost] - - def get_vhosts(self): - if not hasattr(self, "_vhosts"): - self._vhosts = self.fetch_vhosts() - return self._vhosts - - def get_s2s_in(self): - if not hasattr(self, "_s2s_in"): - self._s2s_in = self.fetch_s2s_in() - return self._s2s_in - - def get_s2s_out(self): - if not hasattr(self, "_s2s_out"): - self._s2s_out = self.fetch_s2s_out() - return self._s2s_out - - def get_vhost_metrics(self, vhost): - data = { - "registered": self.get_registered(vhost), - "muc": self.get_muc(vhost), - "online_by_status": self.get_online_by_status(vhost), - "online_by_client": self.get_online_by_client(vhost), - "online_by_ipversion": self.get_online_by_ipversion(vhost), - "online_by_connection": self.get_online_by_connection(vhost), - "online_by_node": self.get_online_by_node(vhost) - } - - return data - - def get_nodes(self): - if not hasattr(self, "_nodes"): - self._nodes = self.fetch_nodes() - return self._nodes - - def get_node_metrics(self, node): - data = { - "online_by_status": self.get_online_by_status(node=node), - "online_by_client": self.get_online_by_client(node=node), - "online_by_ipversion": self.get_online_by_ipversion(node=node), - "online_by_connection": self.get_online_by_connection(node=node), - "online_by_vhost": self.get_online_by_vhost(node=node) - } - - return data - - def get_all(self): - data = { - "registered": self.get_registered(), - "muc": self.get_muc(), - "online_by_status": self.get_online_by_status(), - "online_by_client": self.get_online_by_client(), - "online_by_ipversion": self.get_online_by_ipversion(), - "online_by_connection": self.get_online_by_connection(), - "online_by_node": self.get_online_by_node(), - "online_by_vhost": self.get_online_by_vhost() - } - - vhosts = {} - for host in self.get_vhosts(): - vhosts[host] = self.get_vhost_metrics(host) - data["vhosts"] = vhosts - - nodes = {} - for node in self.get_nodes(): - nodes[node] = self.get_node_metrics(node) - - data["online_client_by_ipversion"] = self.get_online_client_by_ipversion() - data["nodes"] = nodes - - data["s2s_in"] = self.get_s2s_in() - data["s2s_out"] = self.get_s2s_out() - return data - - -if __name__ == "__main__": - import json - from config import Config - - # load config - config = Config() - - # credentials and parameters - url = config.get('url', default='http://localhost:5280/api') - login = config.get('login', default=None) - api = config.get('api', default='rest') - - # init handler - metrics = EjabberdMetrics(url, login, api) - - data = metrics.get_all() - print(json.dumps(data, indent=True)) diff --git a/influx.py b/influx.py index 59728ac..d0569d4 100644 --- a/influx.py +++ b/influx.py @@ -5,7 +5,7 @@ import time from influxdb import InfluxDBClient from config import Config -from ejabberdrpc import EjabberdMetrics +from metrics import EjabberdMetrics class Influx: diff --git a/metrics.py b/metrics.py new file mode 100644 index 0000000..abc57db --- /dev/null +++ b/metrics.py @@ -0,0 +1,322 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +import ipaddress + +from api import EjabberdApi + +# rfc6052: IPv6 Addressing of IPv4/IPv6 Translators +nat64 = ipaddress.ip_network("64:ff9b::/96") + + +class EjabberdMetrics: + """ + class to fetch metrics per xmlrpc + """ + def __init__(self, url, login=None, api="rpc", muc_host: str = 'conference'): + # init ejabberd api + self.api = EjabberdApi(url, login, api) + self._cmd = self.api.cmd + + # variables + self._verstring = self.api.verstring + self.muc_host = muc_host + + def _client(self, resource): + clientmap = { + "Conv6ations for Sum7": ["Conversations with IPv6"], + "Conversations": [], + "Pix-Art Messenger": [], + "Gajim": ["gajim"], + "Psi+": [], + "jitsi": [], + "Dino": ["dino"], + "poezio": [], + "profanity": [], + "Xabber": [], + "ChatSecure": ["chatsecure"] + } + + for client, names in clientmap.items(): + for c in names: + if c in resource: + return client + if client in resource: + return client + return "other" + + @staticmethod + def _ipversion(ip): + addr = ipaddress.ip_address(ip) + if addr.version == 6: + if addr.ipv4_mapped: + return 4 + if addr in nat64: + return 4 + return addr.version + + def fetch_onlineuser(self): + tmp = self._cmd("connected_users_info", {}) + if "connected_users_info" not in tmp: + return tmp + data = [] + for c in tmp["connected_users_info"]: + if "session" not in c: + continue + user = {} + for attrs in c["session"]: + for k, v in attrs.items(): + user[k] = v + data.append(user) + return data + + def fetch_nodes(self): + result = self._cmd("list_cluster", {}) + if "nodes" not in result: + return result + data = [] + for node in result["nodes"]: + data.append(node["node"]) + return data + + def fetch_vhosts(self): + result = self._cmd("registered_vhosts", {}) + if "vhosts" not in result: + return result + data = [] + for vhost in result["vhosts"]: + data.append(vhost["vhost"]) + return data + + def fetch_s2s_in(self): + result = self._cmd("incoming_s2s_number", {}) + if "s2s_incoming" not in result: + return result + return result["s2s_incoming"] + + def fetch_s2s_out(self): + result = self._cmd("outgoing_s2s_number", {}) + if "s2s_outgoing" not in result: + return result + return result["s2s_outgoing"] + + def fetch_registered(self, vhost=None): + if vhost is None: + result = self._cmd("stats", {"name":"registeredusers"}) + if "stat" in result: + return result["stat"] + else: + result = self._cmd("stats_host", {"name":"registeredusers", "host": vhost}) + if "stat" in result: + return result["stat"] + + def fetch_muc(self, vhost=None): + host = "global" + if vhost is not None: + if self._verstring.major >= 19: + host = '.'.join([self.muc_host, vhost]) + else: + host = vhost + result = self._cmd("muc_online_rooms", {"host": host}) + if "rooms" in result: + return len(result["rooms"]) + return len(result) + + def update(self): + # nodes + self._nodes = self.fetch_nodes() + + # vhosts + self._vhosts = self.fetch_vhosts() + + # registered + if not hasattr(self, "_registered"): + self._registered = {} + self._registered[None] = self.fetch_registered() + + # muc + if not hasattr(self, "_muc"): + self._muc = {} + self._muc[None] = self.fetch_muc() + + # registered + muc + for vhost in self._vhosts: + self._registered[vhost] = self.fetch_registered(vhost) + self._muc[vhost] = self.fetch_muc(vhost) + + # online user + self._onlineuser = self.fetch_onlineuser() + + # s2s + self._s2s_in = self.fetch_s2s_in() + self._s2s_out = self.fetch_s2s_out() + + def get_online_by(self, by="node", parse=None, vhost=None, node=None): + parser = parse or (lambda a: a) + if not hasattr(self, "_onlineuser"): + self._onlineuser = self.fetch_onlineuser() + data = {} + + for conn in self._onlineuser: + if vhost is not None and vhost not in conn["jid"]: + continue + if node is not None and node != conn["node"]: + continue + if by not in conn: + continue + value = parser(conn[by]) + if value not in data: + data[value] = 1 + else: + data[value] += 1 + return data + + def get_online_by_node(self, vhost=None): + return self.get_online_by("node", vhost=vhost) + + def get_online_by_vhost(self, node=None): + return self.get_online_by("jid", parse=lambda jid: jid[jid.find("@")+1:jid.find("/")], node=node) + + def get_online_by_status(self, vhost=None, node=None): + return self.get_online_by("status", vhost=vhost, node=node) + + def get_online_by_connection(self, vhost=None, node=None): + return self.get_online_by("connection", vhost=vhost, node=node) + + def get_online_by_client(self, vhost=None, node=None): + return self.get_online_by("resource", parse=self._client, vhost=vhost, node=node) + + def get_online_by_ipversion(self, vhost=None, node=None): + return self.get_online_by("ip", parse=self._ipversion, vhost=vhost, node=node) + + def get_online_client_by(self, by="ip", parse=None, vhost=None, node=None): + parser = parse or self._ipversion + if not hasattr(self, "_onlineuser"): + self._onlineuser = self.fetch_onlineuser() + data = {} + + for conn in self._onlineuser: + client = "other" + if "resource" in conn: + client = self._client(conn["resource"]) + if client not in data: + data[client] = {} + if vhost is not None and vhost not in conn["jid"]: + continue + if node is not None and node != conn["node"]: + continue + if by not in conn: + continue + value = parser(conn[by]) + if value not in data[client]: + data[client][value] = 1 + else: + data[client][value] += 1 + return data + + def get_online_client_by_ipversion(self, vhost=None, node=None): + return self.get_online_client_by("ip", parse=self._ipversion, vhost=vhost, node=node) + + def get_registered(self, vhost=None): + if not hasattr(self, "_registered"): + self._registered = {} + if vhost not in self._registered: + self._registered[vhost] = self.fetch_registered(vhost) + return self._registered[vhost] + + def get_muc(self, vhost=None): + if not hasattr(self, "_muc"): + self._muc = {} + if vhost not in self._muc: + self._muc[vhost] = self.fetch_muc(vhost) + return self._muc[vhost] + + def get_vhosts(self): + if not hasattr(self, "_vhosts"): + self._vhosts = self.fetch_vhosts() + return self._vhosts + + def get_s2s_in(self): + if not hasattr(self, "_s2s_in"): + self._s2s_in = self.fetch_s2s_in() + return self._s2s_in + + def get_s2s_out(self): + if not hasattr(self, "_s2s_out"): + self._s2s_out = self.fetch_s2s_out() + return self._s2s_out + + def get_vhost_metrics(self, vhost): + data = { + "registered": self.get_registered(vhost), + "muc": self.get_muc(vhost), + "online_by_status": self.get_online_by_status(vhost), + "online_by_client": self.get_online_by_client(vhost), + "online_by_ipversion": self.get_online_by_ipversion(vhost), + "online_by_connection": self.get_online_by_connection(vhost), + "online_by_node": self.get_online_by_node(vhost) + } + + return data + + def get_nodes(self): + if not hasattr(self, "_nodes"): + self._nodes = self.fetch_nodes() + return self._nodes + + def get_node_metrics(self, node): + data = { + "online_by_status": self.get_online_by_status(node=node), + "online_by_client": self.get_online_by_client(node=node), + "online_by_ipversion": self.get_online_by_ipversion(node=node), + "online_by_connection": self.get_online_by_connection(node=node), + "online_by_vhost": self.get_online_by_vhost(node=node) + } + + return data + + def get_all(self): + data = { + "registered": self.get_registered(), + "muc": self.get_muc(), + "online_by_status": self.get_online_by_status(), + "online_by_client": self.get_online_by_client(), + "online_by_ipversion": self.get_online_by_ipversion(), + "online_by_connection": self.get_online_by_connection(), + "online_by_node": self.get_online_by_node(), + "online_by_vhost": self.get_online_by_vhost() + } + + vhosts = {} + for host in self.get_vhosts(): + vhosts[host] = self.get_vhost_metrics(host) + data["vhosts"] = vhosts + + nodes = {} + for node in self.get_nodes(): + nodes[node] = self.get_node_metrics(node) + + data["online_client_by_ipversion"] = self.get_online_client_by_ipversion() + data["nodes"] = nodes + + data["s2s_in"] = self.get_s2s_in() + data["s2s_out"] = self.get_s2s_out() + return data + + +if __name__ == "__main__": + import json + from config import Config + + # load config + config = Config() + + # credentials and parameters + url = config.get('url', default='http://localhost:5280/api') + login = config.get('login', default=None) + api = config.get('api', default='rest') + + # init handler + metrics = EjabberdMetrics(url, login, api) + + data = metrics.get_all() + print(json.dumps(data, indent=True)) diff --git a/prometheus.py b/prometheus.py old mode 100755 new mode 100644 index 4aab1b3..4441c87 --- a/prometheus.py +++ b/prometheus.py @@ -8,7 +8,7 @@ from http.server import BaseHTTPRequestHandler, HTTPServer from socketserver import ThreadingMixIn from config import Config -from ejabberdrpc import EjabberdMetrics +from metrics import EjabberdMetrics class _ThreadingSimpleServer(ThreadingMixIn, HTTPServer): -- cgit v1.2.3-54-g00ecf