- rework the types to be internal, use built in types for reporting

Author: zzzeek

collectdconsole.conf

@@ -35,3 +35,26 @@ LoadPlugin python
     </Module>
 </Plugin>
 
+
+LoadPlugin write_graphite
+
+<Plugin write_graphite>
+    <Node "local">
+    Host localhost
+    Port "2003"
+    Protocol "tcp"
+    ReconnectInterval 0
+    LogSendErrors true
+    Prefix "collectd."
+    #Postfix "collectd"
+    StoreRates true
+    AlwaysAppendDS false
+    EscapeCharacter "_"
+    SeparateInstances false
+    PreserveSeparator false
+    DropDuplicateFields false
+    </Node>
+</Plugin>
+
+#LoadPlugin interface
+#LoadPlugin cpu

sqlalchemy_collectd/client/collector.py

@@ -31,6 +31,11 @@ def __init__(self, name):
         # doesn't include DBAPI implicit transactions
         self.transactions = set()
 
+        self.total_checkouts = 0
+        self.total_invalidated = 0
+        self.total_connects = 0
+        self.total_disconnects = 0
+
     @classmethod
     def collection_for_name(cls, name):
         cls.create_mutex.acquire()
@@ -107,11 +112,13 @@ def conn_ident(self, dbapi_connection):
     def _connect_evt(self, dbapi_conn, connection_rec):
         worker._check_threads_started()
         id_ = self.conn_ident(dbapi_conn)
+        self.collection_target.total_connects += 1
         self.connections.add(id_)
         self.checkedin.add(id_)
 
     def _checkout_evt(self, dbapi_conn, connection_rec, connection_proxy):
         id_ = self.conn_ident(dbapi_conn)
+        self.collection_target.total_checkouts += 1
         self.checkedin.remove(id_)
 
     def _checkin_evt(self, dbapi_conn, connection_rec):
@@ -120,6 +127,7 @@ def _checkin_evt(self, dbapi_conn, connection_rec):
 
     def _invalidate_evt(self, dbapi_conn, connection_rec):
         id_ = self.conn_ident(dbapi_conn)
+        self.collection_target.total_invalidated += 1
         self.invalidated.add(id_)
 
     def _reset_evt(self, dbapi_conn, connection_rec):
@@ -135,6 +143,7 @@ def _close_evt(self, dbapi_conn, connection_rec):
 
         try:
             self.connections.remove(id_)
+            self.collection_target.total_disconnects += 1
         except KeyError:
             self._warn_missing_connection(dbapi_conn)
 
@@ -165,6 +174,7 @@ def _close_detached_evt(self, dbapi_conn):
 
         try:
             self.connections.remove(id_)
+            self.collection_target.total_disconnects += 1
         except KeyError:
             self._warn_missing_connection(dbapi_conn)
 

sqlalchemy_collectd/client/internal_types.py

@@ -0,0 +1,26 @@
+from .. import protocol
+
+pool = protocol.Type(
+    "sqlalchemy_pool",
+    ("numpools", protocol.VALUE_GAUGE),
+    ("checkedout", protocol.VALUE_GAUGE),
+    ("checkedin", protocol.VALUE_GAUGE),
+    ("detached", protocol.VALUE_GAUGE),
+    ("invalidated", protocol.VALUE_GAUGE),
+    ("connections", protocol.VALUE_GAUGE),
+)
+
+transactions = protocol.Type(
+    "sqlalchemy_transactions",
+    ("commits", protocol.VALUE_DERIVE),
+    ("rollbacks", protocol.VALUE_DERIVE),
+    ("transactions", protocol.VALUE_DERIVE),
+)
+
+totals = protocol.Type(
+    "sqlalchemy_totals",
+    ("checkouts", protocol.VALUE_DERIVE),
+    ("invalidated", protocol.VALUE_DERIVE),
+    ("connects", protocol.VALUE_DERIVE),
+    ("disconnects", protocol.VALUE_DERIVE),
+)

sqlalchemy_collectd/client/sender.py

@@ -1,5 +1,5 @@
 from .. import protocol
-from .. import types
+from . import internal_types
 
 
 senders = []
@@ -29,7 +29,7 @@ def send(self, connection, collection_target, timestamp, interval, pid):
             sender(message_sender, connection, collection_target, timestamp)
 
 
-@sends(types.pool)
+@sends(internal_types.pool)
 def _send_pool(message_sender, connection, collection_target, timestamp):
     message_sender.send(
         connection, timestamp,
@@ -40,3 +40,15 @@ def _send_pool(message_sender, connection, collection_target, timestamp):
         collection_target.num_invalidated,
         collection_target.num_connections
     )
+
+
+@sends(internal_types.totals)
+def _send_connection_totals(
+        message_sender, connection, collection_target, timestamp):
+    message_sender.send(
+        connection, timestamp,
+        collection_target.total_checkouts,
+        collection_target.total_invalidated,
+        collection_target.total_connects,
+        collection_target.total_disconnects
+    )

sqlalchemy_collectd/protocol.py

@@ -93,6 +93,10 @@ def __init__(self, name, *db_template):
         for value_type in self._value_types:
             self._message_template += char.pack(value_type)
 
+    @property
+    def names(self):
+        return self._field_names
+
     def _encode_values(self, *values):
         """Encode a series of values according to the type template."""
 

sqlalchemy_collectd/server/aggregator.py

@@ -24,7 +24,10 @@ def get_stats_by_progname(self, bucket_name, timestamp, agg_func):
             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)]
+            yield (
+                hostname, progname, len(recs),
+                [agg_func(coll) for coll in zip(*recs)]
+            )
 
     def get_stats_by_hostname(self, bucket_name, timestamp, agg_func):
         bucket = self.buckets[bucket_name]
@@ -33,7 +36,7 @@ def get_stats_by_hostname(self, bucket_name, timestamp, agg_func):
             sorted(records), key=lambda rec: rec[0]
         ):
             recs = [records[key] for key in keys]
-            yield hostname, [agg_func(coll) for coll in zip(*recs)]
+            yield hostname, len(recs), [agg_func(coll) for coll in zip(*recs)]
 
 
 class TimeBucket(object):

sqlalchemy_collectd/server/receiver.py

@@ -1,28 +1,35 @@
 import collectd
 
 from .. import protocol
-from .. import types
 from . import aggregator
+from ..client import internal_types
 
 import logging
 log = logging.getLogger(__name__)
 
 
+summarizers = {}
+
+
+def summarizes(protocol_type):
+    def decorate(fn):
+        summarizers[protocol_type] = fn
+        return fn
+
+    return decorate
+
+
 class Receiver(object):
     def __init__(self, plugin="sqlalchemy"):
         self.plugin = plugin
-        self.types = types_ = [
-            types.pool,
-            types.checkouts,
-            types.commits,
-            types.rollbacks,
-            types.invalidated,
-            types.transactions
+        self.internal_types = [
+            internal_types.pool,
+            internal_types.totals
         ]
-        self.message_receiver = protocol.MessageReceiver(*types_)
+        self.message_receiver = protocol.MessageReceiver(*self.internal_types)
 
         self.aggregator = aggregator.Aggregator(
-            [type_.name for type_ in types_]
+            [type_.name for type_ in self.internal_types]
         )
 
     def receive(self, connection):
@@ -39,42 +46,72 @@ def receive(self, connection):
         )
 
     def summarize(self, timestamp):
-        for type_ in self.types:
-            self._summarize_for_type(type_, timestamp)
-
-    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):
+        for type_ in self.internal_types:
+            summarizer = summarizers.get(type_, None)
+            if summarizer:
+                summarizer(self, type_, timestamp)
+
+
+@summarizes(internal_types.pool)
+def _summarize_pool_stats(receiver, type_, timestamp):
+    values = collectd.Values(
+        type="count",
+        plugin=receiver.plugin,
+        time=timestamp,
+        interval=receiver.aggregator.interval
+    )
+    for hostname, progname, numrecs, stats in \
+            receiver.aggregator.get_stats_by_progname(
+                type_.name, timestamp, sum):
+        for name, value in zip(type_.names, stats):
             values.dispatch(
-                type_instance="sum", host=hostname, plugin_instance=progname,
-                values=stats
+                host=hostname, plugin_instance=progname,
+                type_instance=name,
+                values=[value]
             )
 
-        for hostname, stats in self.aggregator.get_stats_by_hostname(
-                type_.name, timestamp, sum):
+        values.dispatch(
+            host=hostname, plugin_instance=progname,
+            type_instance="numprocs", values=[numrecs])
+
+    for hostname, numrecs, stats in receiver.aggregator.get_stats_by_hostname(
+            type_.name, timestamp, sum):
+        for name, value in zip(type_.names, stats):
             values.dispatch(
-                type_instance="sum", host=hostname, plugin_instance="all",
-                values=stats
+                host=hostname, plugin_instance="host",
+                type_instance=name,
+                values=[value]
             )
+        values.dispatch(
+            host=hostname, plugin_instance="host",
+            type_instance="numprocs", values=[numrecs])
 
-        for hostname, progname, stats in self.aggregator.get_stats_by_progname(
-                type_.name, timestamp, aggregator.avg):
+
+@summarizes(internal_types.totals)
+def _summarize_totals(receiver, type_, timestamp):
+    values = collectd.Values(
+        type="derive",
+        plugin=receiver.plugin,
+        time=timestamp,
+        interval=receiver.aggregator.interval
+    )
+
+    for hostname, progname, numrecs, stats in \
+            receiver.aggregator.get_stats_by_progname(
+                type_.name, timestamp, sum):
+        for name, value in zip(type_.names, stats):
             values.dispatch(
-                type_instance="avg", host=hostname, plugin_instance=progname,
-                values=stats
+                host=hostname, plugin_instance=progname,
+                type_instance=name,
+                values=[value]
             )
 
-        for hostname, stats in self.aggregator.get_stats_by_hostname(
-                type_.name, timestamp, aggregator.avg):
+    for hostname, numrecs, stats in receiver.aggregator.get_stats_by_hostname(
+            type_.name, timestamp, sum):
+        for name, value in zip(type_.names, stats):
             values.dispatch(
-                type_instance="avg", host=hostname, plugin_instance="all",
-                values=stats
+                host=hostname, plugin_instance="host",
+                type_instance=name,
+                values=[value]
             )