- work on collection

Author: zzzeek

sqlalchemy_collectd/collector.py

@@ -0,0 +1,105 @@
+from sqlalchemy import event
+import threading
+
+
+class Collector(object):
+    collectors = {}
+    create_mutex = threading.Mutex()
+
+    def __init__(self, name):
+        self.name = name
+
+        # all identifiers for known DBAPI connections
+        self.connections = set()
+
+        # identifers for connections that have not been checked out
+        # or were checked in
+        self.checkedin = set()
+
+        # identifiers for connections where we've seen begin().
+        # doesn't include DBAPI implicit transactions
+        self.transactions = set()
+
+        # note these are prior to being closed and/or discarded
+        self.invalidated = set()
+
+        # detached connections.
+        self.detached = set()
+
+    @classmethod
+    def collector_for_name(cls, name):
+        cls.create_mutex.acquire()
+        try:
+            if name not in cls.collectors:
+                cls.collectors[name] = collector = Collector(name)
+                return collector
+            else:
+                return cls.collectors[name]
+        finally:
+            cls.create_mutex.release()
+
+    def conn_ident(self, dbapi_connection):
+        return id(dbapi_connection)
+
+    def _connect_evt(self, dbapi_conn, connection_rec):
+        id_ = self.conn_ident(dbapi_conn)
+        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.checkedin.remove(id_)
+
+    def _checkin_evt(self, dbapi_conn, connection_rec):
+        id_ = self.conn_ident(dbapi_conn)
+        self.checkedin.add(id_)
+
+    def _invalidate_evt(self, dbapi_conn, connection_rec):
+        id_ = self.conn_ident(dbapi_conn)
+        self.invalidated.add(id_)
+
+    def _reset_evt(self, dbapi_conn, connection_rec):
+        id_ = self.conn_ident(dbapi_conn)
+        # may or may not have been part of "transactions"
+        self.transactions.discard(id_)
+
+    def _close_evt(self, dbapi_conn, connection_rec):
+        id_ = self.conn_ident(dbapi_conn)
+        self.transactions.discard(id_)
+        self.invalidated.discard(id_)
+        self.checkedin.discard(id_)
+
+        if not self.connections.discard(id_):
+            self._warn_missing_connection(dbapi_conn)
+
+        # this shouldn't be there
+        if self.detached.discard(id_):
+            self._warn("shouldn't have detached")
+
+    def _detach_evt(self, dbapi_conn, connection_rec):
+        id_ = self.conn_ident(dbapi_conn)
+        self.detached.add(id_)
+
+    def _close_detached_evt(self, dbapi_conn):
+        id_ = self.conn_ident(dbapi_conn)
+
+        if not self.connections.discard(id_):
+            self._warn_missing_connection(dbapi_conn)
+
+        self.transactions.discard(id_)
+        self.invalidated.discard(id_)
+        self.checkedin.discard(id_)
+        self.detached.discard(id_)
+
+    def add_engine(self, sqlalchemy_engine):
+        eng = sqlalchemy_engine
+        event.listen(eng, "connect", self._connect_evt)
+        event.listen(eng, "checkout", self._checkout_evt)
+        event.listen(eng, "checkin", self._checkin_evt)
+        event.listen(eng, "invalidate", self._invalidate_evt)
+        event.listen(eng, "soft_invalidate", self._invalidate_evt)
+        event.listen(eng, "reset", self._reset_evt)
+        event.listen(eng, "close", self._close_evt)
+        event.listen(eng, "detach", self._detach_evt)
+        event.listen(eng, "close_detached", self._close_detached_evt)
+

sqlalchemy_collectd/plugin.py

@@ -0,0 +1,22 @@
+from sqlalchemy.engine import CreateEnginePlugin
+
+
+class Plugin(CreateEnginePlugin):
+    def __init__(self, url, kwargs):
+        self.url = url
+
+    def handle_dialect_kwargs(self, dialect_cls, dialect_args):
+        """parse and modify dialect kwargs"""
+
+    def handle_pool_kwargs(self, pool_cls, pool_args):
+        """parse and modify pool kwargs"""
+
+    def engine_created(self, engine):
+        """Receive the :class:`.Engine` object when it is fully constructed.
+
+        The plugin may make additional changes to the engine, such as
+        registering engine or connection pool events.
+
+        """
+
+

sqlalchemy_collectd/tests/test_collectd.py

@@ -14,8 +14,12 @@ def test_encode_type_values(self):
         )
 
         self.assertEqual(
-            b'\x00\x06\x00\x18\x00\x02\x01\x02\xc9v\xbe\x9f\x1a\xcf9'
-            b'@\x00\x00\x00\x00\x00\x00\x01\xc2',
+            b'\x00\x06'  # TYPE_VALUES
+            b'\x00\x18'  # part length
+            b'\x00\x02'  # number of values
+            b'\x01\x02'  # dstype codes GAUGE, DERIVE
+            b'\xc9v\xbe\x9f\x1a\xcf9@'  # 8 bytes for 25.809
+            b'\x00\x00\x00\x00\x00\x00\x01\xc2',  # 8 bytes for 450
             type_.encode_values(25.809, 450)
         )
 
@@ -36,9 +40,23 @@ def test_message_construct(self):
         sender.send(connection, 1517607042.95968, 25.809, 450)
 
         self.assertEqual(
-            [
-                mock.call(b'\x00\x00\x00\rsomehost\x00\x00\x01\x00\x0c\x00\x00\x00\x00Zt\xd8\x82\x00\x02\x00\x0fsomeplugin\x00\x00\x03\x00\x17someplugininstance\x00\x00\x04\x00\x0cmy_type\x00\x00\x07\x00\x0c\x00\x00\x00\x00\x00\x00\x00\n\x00\x05\x00\x15sometypeinstance\x00\x00\x06\x00\x18\x00\x02\x01\x02\xc9v\xbe\x9f\x1a\xcf9@\x00\x00\x00\x00\x00\x00\x01\xc2')
-            ],
+            [mock.call(
+                b'\x00\x00\x00\rsomehost\x00'  # TYPE_HOST
+                b'\x00\x01\x00\x0c\x00\x00\x00\x00Zt\xd8\x82'  # TYPE_TIME
+                b'\x00\x02\x00\x0fsomeplugin\x00'  # TYPE_PLUGIN
+                # TYPE_PLUGIN_INSTANCE
+                b'\x00\x03\x00\x17someplugininstance\x00'
+                b'\x00\x04\x00\x0cmy_type\x00'  # TYPE_TYPE
+                # TYPE_TIMESTAMP
+                b'\x00\x07\x00\x0c\x00\x00\x00\x00\x00\x00\x00\n'
+                b'\x00\x05\x00\x15sometypeinstance\x00'  # TYPE_TYPE_INSTANCE
+                b'\x00\x06'  # TYPE_VALUES
+                b'\x00\x18'  # part length
+                b'\x00\x02'  # number of values
+                b'\x01\x02'  # dstype codes GAUGE, DERIVE
+                b'\xc9v\xbe\x9f\x1a\xcf9@'  # 8 bytes for 25.809
+                b'\x00\x00\x00\x00\x00\x00\x01\xc2'  # 8 bytes for 450
+            )],
             connection.send.mock_calls
         )
 

types.db

@@ -1,2 +1,9 @@
-sqlalchemy_pool numpools:GAUGE:0:U, checkedout:GAUGE:0:U, checkedin:GAUGE:0:U, detached:GAUGE:0:U, total:GAUGE:0:U
+sqlalchemy_pool numpools:GAUGE:0:U, checkedout:GAUGE:0:U, checkedin:GAUGE:0:U, detached:GAUGE:0:U, invalidated:GAUGE:0:U, total:GAUGE:0:U
+sqlalchemy_checkouts count:DERIVE:0:U
+sqlalchemy_commits count:DERIVE:0:U
+sqlalchemy_rollbacks count:DERIVE:0:U
+sqlalchemy_invalidated count:DERIVE:0:U
+sqlalchemy_transactions count:GAUGE:0:U
+
+