- aggregration up for pool stats

Author: zzzeek

sqlalchemy_collectd/server/aggregator.py

@@ -1,17 +1,47 @@
-import collections
+import itertools
 
 
-class Aggregator(object):
-    def __init__(self):
-        # TOOD: configurable size
-        self.queue = collections.deque(maxlen=100000)
+def avg(values):
+    return sum(values) / len(values)
 
-    def put(self, message):
-        self.queue.appendleft(message)
 
-    def outgoing(self):
-        while self.queue:
-            yield self.queue.pop()
+class Aggregator(object):
+    def __init__(self, interval=10):
+        self.interval = interval
+        self.pool_stats = TimeBucket(4, interval)
+
+    def set_pool_stats(
+            self, hostname, progname, pid, timestamp, numpools,
+            checkedout, checkedin, detached, invalidated, total):
+
+        records = self.pool_stats.get_data(timestamp)
+        records[(hostname, progname, pid)] = (
+            numpools, checkedout, checkedin, detached, invalidated, total
+        )
+
+    def get_pool_stats_by_progname(self, timestamp, agg_func):
+        return self._get_stats_by_progname(
+            self.pool_stats, timestamp, agg_func)
+
+    def get_pool_stats_by_hostname(self, timestamp, agg_func):
+        return self._get_stats_by_hostname(
+            self.pool_stats, timestamp, agg_func)
+
+    def _get_stats_by_progname(self, bucket, timestamp, agg_func):
+        records = bucket.get_data(timestamp)
+        for (hostname, progname), keys in itertools.groupby(
+            sorted(records), key=lambda rec: (rec[0], rec[1])
+        ):
+            recs = [records[key] for key in keys]
+            yield hostname, progname, [agg_func(coll) for coll in zip(*recs)]
+
+    def _get_stats_by_hostname(self, bucket, timestamp, agg_func):
+        records = bucket.get_data(timestamp)
+        for hostname, keys in itertools.groupby(
+            sorted(records), key=lambda rec: rec[0]
+        ):
+            recs = [records[key] for key in keys]
+            yield hostname, [agg_func(coll) for coll in zip(*recs)]
 
 
 class TimeBucket(object):
@@ -61,6 +91,7 @@ def __init__(self, num_buckets, interval):
         self.interval = interval
 
     def _get_bucket(self, timestamp):
+        timestamp = int(timestamp)
         slot = (timestamp // self.interval)
         bucket_num = slot % self.num_buckets
         bucket = self.buckets[bucket_num]

sqlalchemy_collectd/server/plugin.py

@@ -5,10 +5,16 @@
 from . import receiver
 from . import aggregator
 from .. import protocol
+import time
+import logging
+
+log = logging.getLogger(__name__)
 
 aggregator_ = None
+receiver_ = None
 
 
+# TODO: merge collectd.notice w/ python logging?
 def _notice(msg):
     collectd.notice("[sqlalchemy-collectd] %s" % msg)
 
@@ -23,6 +29,7 @@ def get_config(config):
 
 def start_plugin(config):
     global aggregator_
+    global receiver_
 
     config_dict = {elem.key: tuple(elem.values) for elem in config.children}
     host, port = config_dict.get("listen", ("localhost", 25827))
@@ -34,29 +41,9 @@ def start_plugin(config):
     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()
+    now = time.time()
+    receiver_.summarize(aggregator_, now)
 
 
 collectd.register_config(get_config)

sqlalchemy_collectd/server/receiver.py

@@ -1,13 +1,27 @@
+import collectd
+
 from .. import protocol
 from .. import types
+from . import aggregator
 
+import logging
+log = logging.getLogger(__name__)
 
-receivers = []
+receivers = {}
+summarizers = []
 
 
 def receives(protocol_type):
     def decorate(fn):
-        receivers.append((protocol_type, fn))
+        receivers[protocol_type.name] = fn
+        return fn
+
+    return decorate
+
+
+def summarizes(protocol_type):
+    def decorate(fn):
+        summarizers.append(fn)
         return fn
 
     return decorate
@@ -24,11 +38,67 @@ def __init__(self):
             types.transactions
         )
 
-    def receive(self, connection, aggregator):
+    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)
+        type_ = message[protocol.TYPE_TYPE]
+        timestamp = message[protocol.TYPE_TIME]
+        host = message[protocol.TYPE_HOST]
+        progname = message[protocol.TYPE_PLUGIN_INSTANCE]
+        values = message[protocol.TYPE_VALUES]
+        pid = message[protocol.TYPE_TYPE_INSTANCE]
+        try:
+            receiver = receivers[type_]
+        except KeyError:
+            log.warn("Don't understand message type %s, skipping" % type_)
+        else:
+            receiver(
+                message, timestamp, host, progname, pid, values, aggregator_)
+
+    def summarize(self, aggregator_, timestamp):
+        for summarizer in summarizers:
+            summarizer(aggregator_, timestamp)
+
+
+@receives(types.pool)
+def _receive_pool(
+        message, timestamp, host, progname, pid, values, aggregator_):
+    aggregator_.set_pool_stats(host, progname, pid, timestamp, *values)
+
+
+@summarizes(types.pool)
+def _summarize_pool(aggregator_, timestamp):
+    values = collectd.Values(
+        type=types.pool.name,
+        plugin="sqlalchemy",
+        time=timestamp,
+        interval=aggregator_.interval
+    )
+    for hostname, progname, stats in \
+            aggregator_.get_pool_stats_by_progname(timestamp, sum):
+        values.dispatch(
+            type_instance="sum", host=hostname, plugin_instance=progname,
+            values=stats
+        )
+
+    for hostname, stats in aggregator_.get_pool_stats_by_hostname(
+            timestamp, sum):
+        values.dispatch(
+            type_instance="sum", host=hostname, plugin_instance="all",
+            values=stats
+        )
+
+    for hostname, progname, stats in aggregator_.get_pool_stats_by_progname(
+            timestamp, aggregator.avg):
+        values.dispatch(
+            type_instance="avg", host=hostname, plugin_instance=progname,
+            values=stats
+        )
+
+    for hostname, stats in aggregator_.get_pool_stats_by_hostname(
+            timestamp, aggregator.avg):
+        values.dispatch(
+            type_instance="avg", host=hostname, plugin_instance="all",
+            values=stats
+        )
+