- build out connection model to suit types.db

Author: zzzeek

README.rst

@@ -12,11 +12,84 @@ hostname/pid-12345/checkouts
 
 host  /  plugin   / plugininstance  type / typeinstance
 
-hostname/sqlalchemy-nova/checkouts-12345    derive value=value
-hostname/sqlalchemy-nova/transactions-12345  derive value=value
-hostname/sqlalchemy-nova/checkedout-12345   gauge
-hostname/sqlalchemy-nova/pooled-12345   gauge
-hostname/sqlalchemy-nova/connected-12345  gauge
+# use real types, requires custom types.db w/ awkward configuration
+hostname/sqlalchemy-nova/checkouts-12345
+hostname/sqlalchemy-nova/transactions-12345
+hostname/sqlalchemy-nova/checkedout-12345
+hostname/sqlalchemy-nova/pooled-12345
+hostname/sqlalchemy-nova/connected-12345
+
+# another one w/ a "wide" value
+
+# key:
+# host / plugin-plugininstance / type-typeinstance
+
+# we will use this layout, "configured-name" is part of the
+# sqlalchemy url conf, e.g  mysql://nova:pw@hostname/nova?plugin=sqlalchemy-collectd&collectd_name=nova
+
+# host / sqlalchemy-<configured-name> / <sqlalchemy custom type>-<process id>
+
+
+# for the values, when we see a "gauge" thats a number we report directly,
+# when we see a "derived", collectd can turn that into a rate over time, e.g.
+# checkouts / sec  transactions / sec
+
+# so say nova has process ids 123 and 4567, neutron has procss ids 9856 and 786,
+
+hostname/sqlalchemy-nova/sqlalchemy_pool-123   numpools=gauge checkedout=gauge overflow=gauge pooled=gauge total=gauge
+hostname/sqlalchemy-nova/sqlalchemy_transactions-123   count=gauge
+hostname/sqlalchemy-nova/sqlalchemy_checkouts-123   count=derived
+hostname/sqlalchemy-nova/sqlalchemy_commits-123   count=derived
+hostname/sqlalchemy-nova/sqlalchemy_rollbacks-123   count=derived
+hostname/sqlalchemy-nova/sqlalchemy_invalidated-123   count=derived
+
+
+hostname/sqlalchemy-neutron/sqlalchemy_pool-9865   numpools=gauge checkedout=gauge overflow=gauge pooled=gauge total=gauge
+hostname/sqlalchemy-neutron/sqlalchemy_checkouts-9865   count=derived
+hostname/sqlalchemy-neutron/sqlalchemy_transactions-9865   count=derived
+hostname/sqlalchemy-neutron/sqlalchemy_pool-786   numpools=gauge checkedout=gauge overflow=gauge pooled=gauge total=gauge
+hostname/sqlalchemy-neutron/sqlalchemy_checkouts-786   count=derived
+hostname/sqlalchemy-neutron/sqlalchemy_transactions-786   count=derived
+
+
+# then we're going to use the <Plugin "aggregation"> to flatten out the
+# process id part of it so that we get the stats per program / database URL
+# overall, then a second <Plugin "aggregation"> will further take that
+# and get us the stats per hostname, that's what will be sent to graphite
+# out of the box.
+
+
+
+hostname/sqlalchemy-nova/sqlalchemy_checkouts-12345   count=derived
+hostname/sqlalchemy-nova/sqlalchemy_transactions-12345   count=derived
+
+
+
+# alternate - but no way to aggregate b.c. can't aggregate on regexp
+hostname/sqlalchemy-nova-12345/derive-checkouts
+hostname/sqlalchemy-nova-12345/derive-transactions
+hostname/sqlalchemy-nova-12345/gauge-checkedout
+hostname/sqlalchemy-nova-12345/gauge-pooled
+hostname/sqlalchemy-nova-12345/gauge-connected
+
+# alternate #2
+hostname/sqlalchemy-12345/derive-nova-checkouts    value
+hostname/sqlalchemy-12345/derive-nova-transactions value
+hostname/sqlalchemy-12345/gauge-nova-checkedout  value
+hostname/sqlalchemy-12345/gauge-nova-pooled      value
+hostname/sqlalchemy-12345/gauge-nova-connected   value
+
+# with #2, aggregate on host, type instance, gives us:
+hostname/sqlalchemy/derive-nova-checkouts    value
+hostname/sqlalchemy/derive-nova-transactions value
+hostname/sqlalchemy/gauge-nova-checkedout  value
+hostname/sqlalchemy/gauge-nova-pooled      value
+hostname/sqlalchemy/gauge-nova-connected   value
+
+# but! still can't aggregate across nova/neutron/etc
+
+
+
 
 
 <Plugin "aggregation">

sqlalchemy_collectd/connection.py

@@ -1,18 +1,16 @@
-import collections
 import struct
 import socket
 import logging
 import threading
-import time
 
 log = logging.getLogger(__name__)
 
 DEFAULT_INTERVAL = 10
 MAX_PACKET_SIZE = 1024
 
-VALUE_COUNTER  = 0
-VALUE_GAUGE    = 1
-VALUE_DERIVE   = 2
+VALUE_COUNTER = 0
+VALUE_GAUGE = 1
+VALUE_DERIVE = 2
 VALUE_ABSOLUTE = 3
 
 # https://git.octo.it/?p=collectd.git;a=blob;hb=master;f=src/network.h
@@ -25,74 +23,106 @@
 TYPE_VALUES = 0x0006
 TYPE_INTERVAL = 0x0007
 
+_value_formats = {
+    VALUE_COUNTER: "!Q",
+    VALUE_GAUGE: "<d",
+    VALUE_DERIVE: "!q",
+    VALUE_ABSOLUTE: "!Q"
+}
+
+
+class Type(object):
+    """Represents a collectd type and its data template.
+
+    Here, we are encoding what we need to know about a collectd type that we'd
+    be targeting to send to a collectd server.A type  includes a name and a set
+    of values and types that go along with it. These names and value
+    definitions are a fixed thing on the collectd server  side, which are
+    listed in a file called types.db.   Additional custom types can be  added
+    by specifying additional type files in the collectd configuration.
+
+    .. seealso::
+
+        collectd's default types.db:
+        https://github.com/collectd/collectd/blob/master/src/types.db
+
+    """
+
+    __slots__ = 'name', '_value_types', '_value_formats', '_message_template'
+
+    def __init__(self, name, *db_template):
+        """Contruct a new Type.
+
+        E.g. to represent the "load" type in collectd's types.db::
+
+            load = Type(
+                "load",
+                ("shortterm", VALUE_GAUGE),
+                ("midterm", VALUE_GAUGE),
+                ("longterm", VALUE_GAUGE)
+            )
+        """
+        self.name = name
+        self._value_types = [value_type for dsname, value_type in db_template]
+        self._value_formats = [
+            _value_formats[value_type] for value_type in self._value_types]
+        self._message_template = struct.pack(
+            "!HHH", TYPE_VALUES, 6 + (9 * len(db_template)),
+            len(db_template))
+        for value_type in self._value_types:
+            self._message_template += struct.pack("B", value_type)
+
+    def encode_values(self, *values):
+        """Encode a series of values according to the type template."""
+
+        msg = self._message_template
+        for format_, dsvalue in zip(self._value_formats, values):
+            msg += struct.pack(format_, dsvalue)
+
+        return msg
+
 
 class MessageSender(object):
+    """Represents all the fields necessary to send a message."""
+
+    __slots__ = (
+        'type', 'host', 'plugin', 'plugin_instance', 'type_instance',
+        'interval', '_host_message_part', '_remainder_message_parts'
+    )
+
     def __init__(
-        self, type, host, plugin="sqlalchemy", plugin_instance=None,
+        self, type, host, plugin, plugin_instance=None,
             type_instance=None, interval=DEFAULT_INTERVAL):
 
-    # TODO: send template just like in types.db and fix to that
-
         self.type = type
         self.host = host
         self.plugin = plugin
         self.plugin_instance = plugin_instance
         self.type_instance = type_instance
         self.interval = interval
-        self._queue = collections.deque()
-
-    def _header(self, timestamp):
-        buf = b''
-        buf += struct.pack("!HH", TYPE_HOST, 5 + len(self.host)) + self.host + b"\0"
-        buf += struct.pack("!HHq", TYPE_TIME, 12, timestamp)
-        buf += struct.pack("!HH", TYPE_PLUGIN, 5 + len(self.plugin)) + self.plugin + b"\0"
-        buf += struct.pack("!HH", TYPE_PLUGIN_INSTANCE, 5 + len(self.plugin_instance)) + self.plugin_instance + b"\0"
-        buf += struct.pack("!HH", TYPE_TYPE, 5 + len(self.type)) + self.type + b"\0"
-        buf += struct.pack("!HHq", TYPE_INTERVAL, 12, self.interval)
-        buf += struct.pack("!HH", TYPE_TYPE_INSTANCE, 5 + len(self.type_instance)) + self.type_instance + b'\0'
-
-        return buf
-
-    def _gauge(self, dsname, dsvalue):
-        buf = b''
-        buf += struct.pack("!HHH", TYPE_VALUES, 15, 1)
-        buf += struct.pack("<Bd", VALUE_GAUGE, dsvalue)
-
-        return buf
-
-    def _derive(self, dsname, dsvalue):
-        buf = b''
-        buf += struct.pack("!HHH", TYPE_VALUES, 15, 1)
-        buf += struct.pack("!Bq", VALUE_DERIVE, dsvalue)
-
-        return buf
-
-    def queue_stat(self, *values):
-        # TODO TODO
-        pass
 
-    def queue_gauge(self, name, value):
-        self._queue.append((VALUE_GAUGE, time.time(), name, value))
+        self._host_message_part = self._pack_string(TYPE_HOST, self.host)
+        self._remainder_message_parts = (
+            self._pack_string(TYPE_PLUGIN, self.plugin) +
+            self._pack_string(TYPE_PLUGIN_INSTANCE, self.plugin_instance) +
+            self._pack_string(TYPE_TYPE, self.type.name) +
+            struct.pack("!HHq", TYPE_INTERVAL, 12, self.interval) +
+            self._pack_string(TYPE_TYPE_INSTANCE, self.type_instance)
+        )
 
-    def queue_derive(self, name, value):
-        self._queue.append((VALUE_DERIVE, time.time(), name, value))
+    def _pack_string(self, typecode, value):
+        return struct.pack("!HH", typecode, 5 + len(value)) + value + b"\0"
 
-    def flush(self, connection):
-        now = time.time()
-        too_old = now - self.interval
-        header = self._header(now)
+    def send(self, connection, timestamp, *values):
+        """Send a message on a connection."""
 
-        while self._queue:
-            type_, timestamp, name, value = self._queue.popleft()
-            if timestamp < too_old:
-                continue
+        header = self._host_message_part + \
+            struct.pack("!HHq", TYPE_TIME, 12, timestamp) + \
+            self._remainder_message_parts
 
-            if type_ == VALUE_GAUGE:
-                element = self._gauge(name, value)
-            else:
-                element = self._derive(name, value)
+        payload = self.type.encode_values(*values)
 
-            connection.send(header + element)
+        connection.send(header + payload)
 
 
 class Connection(object):
@@ -105,7 +135,7 @@ def __init__(self, host="localhost", port=25826):
     def send(self, message):
         self._mutex.acquire()
         try:
-            log.debug("sending: %s", message)
+            log.debug("sending: %r", message)
             self.socket.sendto(message, (self.host, self.port))
         except IOError:
             log.error("Error in socket.sendto", exc_info=True)

sqlalchemy_collectd/worker.py

@@ -0,0 +1,4 @@
+_WORKER_THREAD = None
+
+# set up uber-global-singleton thread for all connections
+# and everything

types.db

@@ -0,0 +1,2 @@
+sqlalchemy_pool numpools:GAUGE:0:U, checkedout:GAUGE:0:U, checkedin:GAUGE:0:U, detached:GAUGE:0:U, total:GAUGE:0:U
+