add collectd plugin for collecting running transcoder stats

ansible/roles/monitoring-client/tasks/collectd.yml

@@ -17,6 +17,15 @@
       - install
       - config
 
+  - name: install pip packages for systemd exporter
+    pip:
+      name:
+        - pydbus
+        - collectd
+      state: present
+    when: transcoder_streams is defined or transcoder_lounges is defined
+    tags: install
+
   - name: install collectd perl packages
     apt: name={{ packages|flatten(levels=1) }} state=latest
     vars:
@@ -72,6 +81,13 @@
     when: haproxy is defined and haproxy
     tags: install
 
+  - name: add systemd unit collectd plugin
+    template:
+      src: collectd/plugins/TranscodingUnits.py
+      dest: /opt/voc/collectd/plugins/TranscodingUnits.py
+    when: transcoder_streams is defined or transcoder_lounges is defined
+    tags: install
+
   - name: adding entries for curl_json php to types.db
     lineinfile: dest=/opt/voc/collectd/types.db
                 regexp="{{ item.regex }}"

ansible/roles/monitoring-client/tasks/main.yml

@@ -6,7 +6,7 @@
   # Install packages
   - name: install packages needed to monitor host
     apt:
-      name: ['nagios-plugins-basic', 'collectd']
+      name: ['nagios-plugins-basic', 'collectd', 'python-pip']
       state: latest
     tags: apt
 

ansible/roles/monitoring-client/templates/collectd/collectd.client.conf.j2

@@ -39,7 +39,6 @@ LoadPlugin ipmi
   Globals true
 </LoadPlugin>
 
-LoadPlugin "logfile"
 <Plugin "logfile">
   LogLevel "info"
   File "/var/log/collectd.log"
@@ -96,6 +95,12 @@ LoadPlugin "logfile"
   Import NginxHls
   Import NginxDash
 {% endif %}
+{% if transcoder_streams is defined or transcoding_lounge is defined %}
+  Import "TranscodingUnits"
+  <Module TranscodingUnits>
+    UnitRegex "transcode_(h264|vpx|audio)@[a-zA-Z0-9]{2,2}\.service" 
+  </Module>
+{% endif %}
 </Plugin>
 
 {% if ansible_virtualization_role == 'host' and libvirtd_bin.stdout != '' %}

ansible/roles/monitoring-client/templates/collectd/plugins/TranscodingUnits.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from pydbus import SystemBus
+from gi.repository import GLib
+import collectd
+import re
+import time
+
+DBUS_SYSTEMD = '.systemd1'
+
+class NoSuchUnitException(Exception):
+    pass
+
+class SystemdCollector:
+    """
+    Collector class to collect statistics about systemd units
+    """
+
+    def __init__(self):
+        self.__monitored_units = []
+        self.__monitored_regexes = []
+
+        self.__dbus = SystemBus()
+        self.__systemd = self.__dbus.get(DBUS_SYSTEMD)
+
+    def add_unit(self, unit):
+        """
+        add unit to scraping
+
+        :param unit: str name of a systemd unit
+        """
+        try:
+            unit_path = self.__systemd.GetUnit(unit)
+            if unit_path not in self.__monitored_units:
+                self.__monitored_units.append(unit_path)
+        except GLib.Error as e:
+            if 'NoSuchUnit' in e.message:
+                raise NoSuchUnit('No Such Unit {}'.format(unit))
+
+    def add_units_regex(self, regex):
+        """
+        add all units to scraping, which names match the regex
+
+        :param regex: str regex
+        """
+        if regex not in self.__monitored_regexes:
+            self.__monitored_regexes.append(regex)
+
+    def collectd_config_callback(self, config):
+        """
+        collectd plugin api compatile configuration callback
+
+        :param config: str configuration key
+        """
+        for sub_conf in config.children:
+            if sub_conf.key == 'UnitName':
+                self.add_unit(sub_conf.values[0])
+            elif sub_conf.key == 'UnitRegex':
+                self.add_units_regex(sub_conf.values[0])
+
+    def collect_unit_stats(self):
+        units = []
+        units.extend(self.__monitored_units)
+
+        for unit_tuple in self.__systemd.ListUnits():
+            for regex in self.__monitored_regexes:
+                if re.match(regex, unit_tuple[0]) is not None:
+                    try:
+                        unit_path = self.__systemd.GetUnit(unit_tuple[0])
+                        if unit_path not in units:
+                            units.append(unit_path)
+                    except GLib.Error:
+                        pass
+
+        stats = {}
+
+        for unit_path in units:
+            unit = self.__dbus.get(DBUS_SYSTEMD, unit_path)
+
+            enabled = unit.UnitFileState == 'enabled'
+            active = unit.ActiveState == 'active'
+
+            uptime = int(time.time() * 1000000) - unit.ActiveEnterTimestamp if active else 0
+            downtime = int(time.time() * 1000000) - unit.InactiveEnterTimestamp if not active else 0
+
+            stats[unit.Id] = {
+                    "enabled": 1 if enabled else 0,
+                    "active": 1 if active else 0,
+                    "uptime": uptime,
+                    "downtime": downtime,
+            }
+
+        return stats
+
+    def collectd_read_callback(self, data=None):
+        """
+        collectd compatible read callback
+        """
+        for unit_name, stats in self.collect_unit_stats().items():
+            for stat_name, value in stats.items():
+                vl = collectd.Values(plugin='systemd_units', type='gauge', type_instance="{}/{}".format(unit_name, stat_name), values=[value] )
+                vl.dispatch()
+
+try:
+    collector = SystemdCollector()
+
+    collectd.register_read(collector.collectd_read_callback)
+    collectd.register_config(collector.collectd_config_callback)
+except (ImportError, AttributeError):
+    collector = SystemdCollector()
+
+    collector.add_units_regex('transcode_(h264|vpx|audio)@[a-zA-Z0-9]{2,2}\.service')
+
+    print(collector.collect_unit_stats())
+