Initial revision.

This project started as an internal tool at Mochi Media.

Author: ralphm

AUTHORS

@@ -0,0 +1 @@
+Ralph Meijer <[email protected]>

LICENSE

@@ -0,0 +1,20 @@
+Copyright (c) 2012-2013 Mochi Media, Inc.
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

README.md

@@ -0,0 +1,28 @@
+Vör
+---
+
+What is this?
+=============
+
+Vör is a set of services for polling metrics to be sent to Graphite.
+
+
+Requirements
+============
+
+ - Twisted
+ - simplejson
+
+
+Copyright and Warranty
+======================
+
+The code in this distribution started as an internal tool at Mochi Media and is
+made available under the MIT License. See the included LICENSE file for details.
+
+
+Name
+====
+
+In Norse mythology, Vör is a goddess associated with wisdom. She is described
+as "wise and inquiring, so that nothing can be concealed from her".

doc/examples/poller.tac

@@ -0,0 +1,20 @@
+from twisted.application import service
+from twisted.internet import reactor
+
+from vor.elasticsearch import ElasticSearchNodeStatsGraphiteService
+from vor.elasticsearch import ElasticSearchHealthGraphiteService
+from vor.graphite import GraphiteClientFactory
+
+application = service.Application("Test")
+
+collector = ElasticSearchNodeStatsGraphiteService('http://localhost:9200/')
+collector.setServiceParent(application)
+
+factory = GraphiteClientFactory(collector)
+reactor.connectTCP('localhost', 2003, factory)
+
+collector = ElasticSearchHealthGraphiteService('http://localhost:9200/')
+collector.setServiceParent(application)
+
+factory = GraphiteClientFactory(collector)
+reactor.connectTCP('localhost', 2003, factory)

vor/__init__.py

@@ -0,0 +1,114 @@
+"""
+ElasticSearch stats poller that sends statistics to Graphite.
+"""
+
+import simplejson
+
+from twisted.application import service
+from twisted.internet import task
+from twisted.python import log
+from twisted.web.client import getPage
+
+class BaseElasticSearchGraphiteService(service.Service):
+    """
+    Base service for polling ElasticSearch stats to send to Graphite.
+
+    @ivar protocol: The Graphite protocol.
+    @type protocol: L{vor.graphite.GraphiteLineProtocol}
+    """
+    interval = 5
+    suffixes = ()
+    API = None
+
+    def __init__(self, baseURL):
+        """
+        @param baseURL: Base URL of the ElasticSearch API, including trailing
+            slash.
+        """
+        self.endpoint = baseURL + self.API
+        self.protocol = None
+
+
+    def _flattenValue(self, data, value, prefix, key, timestamp):
+        key = key.replace(' ', '')
+        flatKey = "%s.%s" % (prefix, key)
+        if hasattr(value, 'upper'):
+            # Skip strings unless they have suffixed counterparts with a
+            # metric.
+            for suffix in self.suffixes:
+                suffixedKey = key + suffix
+                if suffixedKey in data:
+                    self.sendMetric(flatKey, data[suffixedKey], timestamp)
+                    break
+        elif hasattr(value, 'iteritems'):
+            # Dicts are flattened.
+            self._flattenDict(value, flatKey, timestamp)
+        elif hasattr(value, 'index'):
+            # Give each item in a sequence their own index and flatten the
+            # value.
+            index = 0
+            for item in value:
+                self._flattenValue(None, item, flatKey, '%d' % index,
+                                   timestamp)
+                index += 1
+        else:
+            # A regular metric.
+            self.sendMetric(flatKey, value, timestamp)
+
+
+    def _flattenDict(self, data, prefix, timestamp):
+        if 'timestamp' in data:
+            timestamp = data['timestamp'] / 1000.
+
+        for key, value in data.iteritems():
+            if (key.endswith(self.suffixes) or
+                key == 'timestamp'):
+                continue
+            else:
+                self._flattenValue(data, value, prefix, key, timestamp)
+
+
+    def sendMetric(self, path, value, timestamp):
+        if self.protocol:
+            self.protocol.sendMetric(path, float(value), timestamp)
+
+
+    def collectStats(self):
+        d = getPage(self.endpoint)
+        d.addCallback(simplejson.loads)
+        d.addCallback(self.flatten)
+        d.addErrback(log.err)
+        return d
+
+
+    def startService(self):
+        self.call = task.LoopingCall(self.collectStats)
+        self.call.start(self.interval, now=False)
+
+
+    def stopService(self):
+        self.call.stop()
+
+
+
+class ElasticSearchNodeStatsGraphiteService(BaseElasticSearchGraphiteService):
+    """
+    Service that polls ElasticSearch node stats and sends them to Graphite.
+
+    @ivar protocol: The Graphite protocol.
+    @type protocol: L{vor.graphite.GraphiteLineProtocol}
+    """
+
+    suffixes = ('_in_bytes', '_in_millis')
+    API = '_cluster/nodes/stats?all=1'
+
+    def flatten(self, data):
+        for node in data['nodes'].itervalues():
+            name = node['name']
+            timestamp = node['timestamp']
+
+            prefix = 'es.nodes.'+ name
+            if ('attributes' in node and
+                node['attributes'].get('data', 'false') != 'true'):
+                del node['indices']
+            self._flattenDict(node, prefix, timestamp)

vor/elasticsearch.py

@@ -0,0 +1,40 @@
+from twisted.internet import protocol
+from twisted.protocols.basic import LineReceiver
+
+class GraphiteLineProtocol(LineReceiver):
+    delimiter = '\n'
+    service = None
+
+    def connectionMade(self):
+        self.service.protocol = self
+        print self.transport
+
+
+    def connectionLost(self, reason):
+        self.service.protocol = None
+
+
+    def lineReceived(self, line):
+        pass
+
+
+    def sendMetric(self, path, value, timestamp):
+        """
+        Send a single metric.
+        """
+        self.sendLine('%s %s %s' % (path, value, timestamp))
+
+
+
+class GraphiteClientFactory(protocol.ReconnectingClientFactory):
+    protocol = GraphiteLineProtocol
+
+    def __init__(self, service):
+        self.service = service
+
+
+    def buildProtocol(self, addr):
+        self.resetDelay()
+        p = self.protocol()
+        p.service = self.service
+        return p