From 40edb13762a7c46b23681aca62158fd249ff9b29 Mon Sep 17 00:00:00 2001 From: nico Date: Thu, 24 Oct 2019 20:31:49 +0200 Subject: [PATCH] final touches # added + add method to escape whitespaces in keys and values # changed * init client handler only once and not every time * don't try to create database on every run just on the first one # removed - vhosts are listed first and after that the respective nodes are listed --- .gitignore | 6 +- .../linux-systemd/ejabberd-influxdb.service | 14 +++ ejabberdrpc.py | 6 +- influx.py | 97 +++++++++++++++++++ 4 files changed, 121 insertions(+), 2 deletions(-) create mode 100644 contrib/init/linux-systemd/ejabberd-influxdb.service create mode 100644 influx.py diff --git a/.gitignore b/.gitignore index 31a4ad7..263cfcc 100644 --- a/.gitignore +++ b/.gitignore @@ -108,4 +108,8 @@ dmypy.json .pyre/ # Pycharm -.idea/ \ No newline at end of file +.idea/ +.venv/ + +# config +config.json diff --git a/contrib/init/linux-systemd/ejabberd-influxdb.service b/contrib/init/linux-systemd/ejabberd-influxdb.service new file mode 100644 index 0000000..bade80f --- /dev/null +++ b/contrib/init/linux-systemd/ejabberd-influxdb.service @@ -0,0 +1,14 @@ +[Unit] +Description=ejabberd2influxdb + +[Service] +Type=simple +User=nobody +Group=nobody +ExecStart=/opt/ejabberd-metrics/influx.py +Restart=always +RestartSec=5s +Environment=PATH=/usr/bin:/usr/local/bin + +[Install] +WantedBy=multi-user.target diff --git a/ejabberdrpc.py b/ejabberdrpc.py index b2cc908..cee769b 100755 --- a/ejabberdrpc.py +++ b/ejabberdrpc.py @@ -109,7 +109,11 @@ class EjabberdMetrics(): def fetch_muc(self, vhost=None): host = "global" if vhost is not None: - host = "conference." + vhost + version = self._cmd("status", {}) + if "19.09" in version: + host = "conference." + vhost + else: + host = vhost result = self._cmd("muc_online_rooms", {"host": host}) if "rooms" in result: return len(result["rooms"]) diff --git a/influx.py b/influx.py new file mode 100644 index 0000000..fcf2f41 --- /dev/null +++ b/influx.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +import json +import time + +from influxdb import InfluxDBClient + +from ejabberdrpc import EjabberdMetrics + + +class Influx: + def __init__(self, metrics, client): + self._metrics = metrics + self.client = client + + @staticmethod + def _timestamp(): + return int(time.time() * 1000) + + @staticmethod + def _rmspace(key: str = None, value=None): + try: + key = key.replace(" ", "\ ") + value = value.replace(" ","\ ") + except (TypeError, AttributeError): + pass + + return key, value + + def _parse(self, name, key, value, ts, tags=None): + output = name + + # check if tags is a dict + if isinstance(tags, dict): + + # create tag_key=tag_value pairs for all elements and append them to name + for k, v in tags.items(): + output += ",{}={}".format(*self._rmspace(k, v)) + + # append key=value to name + output += ' {}={}i {}'.format(*self._rmspace(key, value), ts) + return output + + def writeMetrics(self): + name = "ejabberd" + data = list() + + # global values + cur_ts = self._timestamp() + data.append("{m} s2s_in={v}i {ts}".format(m=name, v=self._metrics.get_s2s_in(), ts=cur_ts)) + data.append("{m} s2s_out={v}i {ts}".format(m=name, v=self._metrics.get_s2s_out(), ts=cur_ts)) + + # vhost values + for vhost in self._metrics.get_vhosts(): + cur_ts = self._timestamp() + data.append("{m},vhost={vh} registered={v}i {ts}".format(m=name, vh= vhost, v=self._metrics.get_registered(vhost),ts=cur_ts)) + data.append("{m},vhost={vh} muc={v}i {ts}".format(m=name, vh= vhost, v=self._metrics.get_muc(vhost), ts=cur_ts)) + + # vhost statistics on their repsective node + for node in self._metrics.get_nodes(): + cur_ts = self._timestamp() + for k, v in self._metrics.get_online_by_status(node=node, vhost=vhost).items(): + data.append(self._parse("ejabberd_online_status", k, v, cur_ts, {"node": node, "vhost": vhost})) + for k, v in self._metrics.get_online_by_client(node=node, vhost=vhost).items(): + data.append(self._parse("ejabberd_online_client", k, v, cur_ts, {"node": node, "vhost": vhost})) + for k, v in self._metrics.get_online_by_ipversion(node=node, vhost=vhost).items(): + data.append(self._parse("ejabberd_online_ipversion", k, v, cur_ts, {"node": node, "vhost": vhost})) + for k, v in self._metrics.get_online_by_connection(node=node, vhost=vhost).items(): + data.append(self._parse("ejabberd_online_connection", k, v, cur_ts, {"node": node, "vhost": vhost})) + for cl, ipv in self._metrics.get_online_client_by_ipversion(node=node, vhost=vhost).items(): + for k, v in ipv.items(): + data.append(self._parse("ejabberd_online_client_ipversion", k, v, cur_ts, {"vhost": vhost, "node": node, "ipversion": k, "client": cl})) + + # write output to database + self.client.write_points(data, database='custom', time_precision='ms', batch_size=10000, protocol='line') + +if __name__ == "__main__": + with open("config.json", "r", encoding="utf-8") as f: + login = json.load(f) + + # init global handler + metrics = EjabberdMetrics("http://localhost:4560", login) + client = InfluxDBClient(host='localhost', port=8086, database='custom') + + # create database only once + client.create_database('custom') + + # init influx class + influx = Influx(metrics, client) + + while True: + # TODO this will fail when the connection drops try except maybe? + metrics.update() + influx.writeMetrics() + + time.sleep(10)