- aggreagation can be generic for now

Author: zzzeek

sqlalchemy_collectd/client/sender.py

@@ -14,14 +14,15 @@ def decorate(fn):
 
 
 class Sender(object):
-    def __init__(self, hostname, stats_name):
+    def __init__(self, hostname, stats_name, plugin="sqlalchemy"):
         self.hostname = hostname
         self.stats_name = stats_name
+        self.plugin = plugin
 
     def send(self, connection, collection_target, timestamp, interval, pid):
         for protocol_type, sender in senders:
             message_sender = protocol.MessageSender(
-                protocol_type, self.hostname, "sqlalchemy",
+                protocol_type, self.hostname, self.plugin,
                 plugin_instance=self.stats_name, type_instance=str(pid),
                 interval=interval
             )

sqlalchemy_collectd/server/aggregator.py

@@ -6,36 +6,31 @@ def avg(values):
 
 
 class Aggregator(object):
-    def __init__(self, interval=10):
+    def __init__(self, bucket_names, interval=10):
         self.interval = interval
-        self.pool_stats = TimeBucket(4, interval)
+        self.buckets = {name: TimeBucket(4, interval) for name in bucket_names}
 
-    def set_pool_stats(
-            self, hostname, progname, pid, timestamp, numpools,
+    def set_stats(
+            self, bucket_name, hostname, progname, pid, timestamp, numpools,
             checkedout, checkedin, detached, invalidated, total):
 
-        records = self.pool_stats.get_data(timestamp)
+        bucket = self.buckets[bucket_name]
+        records = bucket.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):
+    def get_stats_by_progname(self, bucket_name, timestamp, agg_func):
+        bucket = self.buckets[bucket_name]
         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):
+    def get_stats_by_hostname(self, bucket_name, timestamp, agg_func):
+        bucket = self.buckets[bucket_name]
         records = bucket.get_data(timestamp)
         for hostname, keys in itertools.groupby(
             sorted(records), key=lambda rec: rec[0]

sqlalchemy_collectd/server/listener.py

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

sqlalchemy_collectd/server/plugin.py

@@ -3,14 +3,12 @@
 import collectd
 from . import listener
 from . import receiver
-from . import aggregator
 from .. import protocol
 import time
 import logging
 
 log = logging.getLogger(__name__)
 
-aggregator_ = None
 receiver_ = None
 
 
@@ -28,22 +26,20 @@ 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))
 
-    aggregator_ = aggregator.Aggregator()
     receiver_ = receiver.Receiver()
     connection = protocol.ServerConnection(host, int(port))
 
-    listener.listen(connection, aggregator_, receiver_)
+    listener.listen(connection, receiver_)
 
 
 def read(data=None):
     now = time.time()
-    receiver_.summarize(aggregator_, now)
+    receiver_.summarize(now)
 
 
 collectd.register_config(get_config)

sqlalchemy_collectd/server/receiver.py

@@ -7,98 +7,74 @@
 import logging
 log = logging.getLogger(__name__)
 
-receivers = {}
-summarizers = []
-
-
-def receives(protocol_type):
-    def decorate(fn):
-        receivers[protocol_type.name] = fn
-        return fn
-
-    return decorate
-
-
-def summarizes(protocol_type):
-    def decorate(fn):
-        summarizers.append(fn)
-        return fn
-
-    return decorate
-
 
 class Receiver(object):
-    def __init__(self):
-        self.message_receiver = protocol.MessageReceiver(
+    def __init__(self, plugin="sqlalchemy"):
+        self.plugin = plugin
+        self.types = types_ = [
             types.pool,
             types.checkouts,
             types.commits,
             types.rollbacks,
             types.invalidated,
             types.transactions
+        ]
+        self.message_receiver = protocol.MessageReceiver(*types_)
+
+        self.aggregator = aggregator.Aggregator(
+            [type_.name for type_ in types_]
         )
 
-    def receive(self, connection, aggregator_):
+    def receive(self, connection):
         data, host = connection.receive()
         message = self.message_receiver.receive(data)
-        type_ = message[protocol.TYPE_TYPE]
+        type_name = 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
+        self.aggregator.set_stats(
+            type_name, host, progname, pid, timestamp, *values
         )
 
-    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
-        )
+    def summarize(self, timestamp):
+        for type_ in self.types:
+            self._summarize_for_type(type_, timestamp)
 
-    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
+    def _summarize_for_type(self, type_, timestamp):
+        values = collectd.Values(
+            type=type_.name,
+            plugin=self.plugin,
+            time=timestamp,
+            interval=self.aggregator.interval
         )
+        for hostname, progname, stats in \
+                self.aggregator.get_stats_by_progname(
+                    type_.name, timestamp, sum):
+            values.dispatch(
+                type_instance="sum", host=hostname, plugin_instance=progname,
+                values=stats
+            )
+
+        for hostname, stats in self.aggregator.get_stats_by_hostname(
+                type_.name, timestamp, sum):
+            values.dispatch(
+                type_instance="sum", host=hostname, plugin_instance="all",
+                values=stats
+            )
+
+        for hostname, progname, stats in self.aggregator.get_stats_by_progname(
+                type_.name, timestamp, aggregator.avg):
+            values.dispatch(
+                type_instance="avg", host=hostname, plugin_instance=progname,
+                values=stats
+            )
+
+        for hostname, stats in self.aggregator.get_stats_by_hostname(
+                type_.name, timestamp, aggregator.avg):
+            values.dispatch(
+                type_instance="avg", host=hostname, plugin_instance="all",
+                values=stats
+            )