- work up client/server round trip

Author: zzzeek

collectdconsole.conf

@@ -0,0 +1,37 @@
+# A config file that lets you test collectd locally.
+# Run just like this:
+# collectd -f -C collectdconsole.conf
+# received messages will be dumped as JSON to stdout.
+
+TypesDB "/usr/share/collectd/types.db"
+TypesDB "/home/classic/dev/sqlalchemy-collectd/types.db"
+
+LoadPlugin logfile
+<Plugin logfile>
+       LogLevel info
+       File STDOUT
+       Timestamp true
+       PrintSeverity false
+</Plugin>
+
+LoadPlugin write_log
+<Plugin write_log>
+   Format JSON
+</Plugin>
+
+LoadPlugin python
+<Plugin python>
+    ModulePath "/home/classic/dev/sqlalchemy-collectd"
+    LogTraces true
+
+</Plugin>
+
+<Plugin python>
+    Import "sqlalchemy_collectd.server.plugin"
+
+    <Module "sqlalchemy_collectd.server.plugin">
+        listen "localhost" 25827
+
+    </Module>
+</Plugin>
+

sqlalchemy_collectd/client/plugin.py

@@ -26,26 +26,27 @@ def engine_created(self, engine):
         registering engine or connection pool events.
 
         """
+        start_plugin(engine)
 
-        # TODO: all this configurable
-        hostname = socket.gethostname()
-        progname = sys.argv[0]
 
-        collectd_hostname = "localhost"
-        collectd_port = 25826
+def start_plugin(engine):
+    # TODO: all this configurable
+    hostname = socket.gethostname()
+    progname = sys.argv[0]
 
-        sender_ = sender.Sender(hostname, progname)
-        collection_target = collector.CollectionTarget.collection_for_name(
-            progname)
-        collector.EngineCollector(collection_target, engine)
+    collectd_hostname = "localhost"
+    collectd_port = 25827
 
-        connection = protocol.ClientConnection.for_host_port(
-            collectd_hostname, collectd_port)
-
-        worker.add_target(
-            connection,
-            collection_target,
-            sender_)
+    sender_ = sender.Sender(hostname, progname)
+    collection_target = collector.CollectionTarget.collection_for_name(
+        progname)
+    collector.EngineCollector(collection_target, engine)
 
+    connection = protocol.ClientConnection.for_host_port(
+        collectd_hostname, collectd_port)
 
+    worker.add_target(
+        connection,
+        collection_target,
+        sender_)
 

sqlalchemy_collectd/protocol.py

@@ -171,6 +171,7 @@ def receive(self, buf):
             type_ = self._types[type_name]
         except KeyError:
             log.warn("Type %s not known, skipping", type_name)
+            return None
         else:
             result["type"] = type_
             result["values"] = {
@@ -202,7 +203,7 @@ def _unpack_long(self, type_, length, buf):
         return long_.unpack_from(buf, header.size)[0]
 
     def _unpack_string(self, type_, length, buf):
-        return buf[header.size:length - 1]
+        return buf[header.size:length - 1].decode('ascii')
 
     def _unpack_values(self, type_, length, buf):
         num = short.unpack_from(buf, header.size)[0]
@@ -261,4 +262,11 @@ def send(self, message):
 
 
 class ServerConnection(object):
-    pass
+    def __init__(self, host, port):
+        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.sock.bind((host, port))
+
+    def receive(self):
+        data, addr = self.sock.recvfrom(1024)
+        return data, addr
+

sqlalchemy_collectd/server/aggregator.py

@@ -0,0 +1,14 @@
+import collections
+
+
+class Aggregator(object):
+    def __init__(self):
+        # TOOD: configurable size
+        self.queue = collections.deque(maxlen=100000)
+
+    def put(self, message):
+        self.queue.appendleft(message)
+
+    def outgoing(self):
+        while self.queue:
+            yield self.queue.pop()

sqlalchemy_collectd/server/listener.py

@@ -0,0 +1,15 @@
+import threading
+
+
+def _receive(connection, aggregator, receiver):
+    while True:
+        receiver.receive(connection, aggregator)
+
+
+def listen(connection, aggregator, receiver):
+    listen_thread = threading.Thread(
+        target=_receive, args=(connection, aggregator, receiver))
+    listen_thread.daemon = True
+    listen_thread.start()
+
+

sqlalchemy_collectd/server/plugin.py

@@ -0,0 +1,63 @@
+from .. import __version__
+import sys
+import collectd
+from . import listener
+from . import receiver
+from . import aggregator
+from .. import protocol
+
+aggregator_ = None
+
+
+def _notice(msg):
+    collectd.notice("[sqlalchemy-collectd] %s" % msg)
+
+
+def get_config(config):
+    global aggregator_
+
+    _notice("sqlalchemy_collectd plugin version %s" % __version__)
+    _notice("Python version: %s" % sys.version)
+    start_plugin(config)
+
+
+def start_plugin(config):
+    global aggregator_
+
+    config_dict = {elem.key: tuple(elem.values) for elem in config.children}
+    host, port = config_dict.get("listen", ("localhost", 25827))
+
+    aggregator_ = aggregator.Aggregator()
+    receiver_ = receiver.Receiver()
+    connection = protocol.ServerConnection(host, int(port))
+
+    listener.listen(connection, aggregator_, receiver_)
+
+
+def _read_raw_struct_to_values(message):
+    # dispatch(
+    #    [type][, values][, plugin_instance]
+    # [, type_instance][, plugin][, host][, time][, interval]) -> None.
+
+    return collectd.Values(
+        type=message[protocol.TYPE_TYPE],
+        values=message[protocol.TYPE_VALUES],
+        plugin=message[protocol.TYPE_PLUGIN],
+        host=message[protocol.TYPE_HOST],
+        time=message[protocol.TYPE_TIME],
+        type_instance=message[protocol.TYPE_TYPE_INSTANCE],
+        plugin_instance=message[protocol.TYPE_PLUGIN_INSTANCE],
+        interval=message[protocol.TYPE_INTERVAL]
+    )
+
+_read_struct_to_values = _read_raw_struct_to_values
+
+
+def read(data=None):
+    for message in aggregator_.outgoing():
+        values = _read_struct_to_values(message)
+        values.dispatch()
+
+
+collectd.register_config(get_config)
+collectd.register_read(read)

sqlalchemy_collectd/server/receiver.py

@@ -0,0 +1,34 @@
+from .. import protocol
+from .. import types
+
+
+receivers = []
+
+
+def receives(protocol_type):
+    def decorate(fn):
+        receivers.append((protocol_type, fn))
+        return fn
+
+    return decorate
+
+
+class Receiver(object):
+    def __init__(self):
+        self.message_receiver = protocol.MessageReceiver(
+            types.pool,
+            types.checkouts,
+            types.commits,
+            types.rollbacks,
+            types.invalidated,
+            types.transactions
+        )
+
+    def receive(self, connection, aggregator):
+        data, host = connection.receive()
+        message = self.message_receiver.receive(data)
+        if message is not None:
+            message['host'] = host
+        # TODO: look up type-specific handler
+        # feed to aggregator per-type
+        aggregator.put(message)