marathon-apps-collectd-plugin/collectd_opentsdb_plugin.py at master · scrapinghub/marathon-apps-collectd-plugin · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import os
import potsdb
import collectd


class OpenTSDBExportPlugin:

    def __init__(self):
        self.metrics = None

    def configure_callback(self, conf):
        for node in conf.children:
            key = node.key.lower()
            value = node.values[0]
            if key == 'host':
                self._opentsdb_host = value
            elif key == 'port':
                self._opentsdb_port = value
        if not self._opentsdb_host or not self._opentsdb_port:
            raise Exception("OpenTSDB export plugin is not configured")
        collectd.info("Configured OpenTSDB export plugin (%s:%s)" %
                      (self._opentsdb_host, self._opentsdb_port))

    def init_callback(self):
        self.metrics = potsdb.Client(host=self._opentsdb_host,
                                     port=self._opentsdb_port,
                                     mps=100, check_host=False)
        collectd.info("Initialized OpenTSDB export plugin.")

    def write_callback(self, vl):
        if not self.metrics:
            return
        metric_name = 'kumo.{}'.format(vl.type)
        if vl.type_instance:
            metric_name += '.{}'.format(vl.type_instance)

        tags = {'timestamp': vl.time}
        if vl.plugin_instance.startswith('kumo.'):
            project, spider, job = vl.plugin_instance.split('.')[1:]
            tags.update({
                'project': project,
                'spider': '{}/{}'.format(project, spider),
                'job': '{}/{}/{}'.format(project, spider, job),
            })

        for value in vl.values:
            if isinstance(value, (float, int)):
                collectd.info('W %s=%s (%s)' % (metric_name, value, tags))
                self.metrics.send(metric_name, value, **tags)

    def shutdown_callback(self):
        if self.metrics:
            self.metrics.wait()
        collectd.info("Shutdown-ed OpenTSDB export plugin.")


if __name__ == '__main__':
    print("OpenTSDB export plugin is called as a python script")
else:
    try:
        plugin = OpenTSDBExportPlugin()
        collectd.register_config(plugin.configure_callback)
        collectd.register_init(plugin.init_callback)
        collectd.register_write(plugin.write_callback, name='write_opentsdb')
        collectd.register_shutdown(plugin.shutdown_callback)
        print("OpenTSDB export plugin is registered.")
    except Exception as ex:
        print("OpenTSDB export plugin exception: %s" % ex)