- dev

Author: zzzeek

sqlalchemy.conf

@@ -32,11 +32,11 @@ LoadPlugin "network"
 	#
 	# <host>/<plugin>-<program-name>/<type>-<process pid>
 	#
-	# hostname/sqlalchemy-nova/checkouts-12345
-	# hostname/sqlalchemy-nova/checkouts-5839
-	# hostname/sqlalchemy-nova/checkouts-9905
-	# hostname/sqlalchemy-neutron/checkouts-19385
-	# hostname/sqlalchemy-neutron/checkouts-6991
+	# hostname/sqlalchemy-nova/sqlalchemy_pool-12345
+	# hostname/sqlalchemy-nova/sqlalchemy_pool-5839
+	# hostname/sqlalchemy-nova/sqlalchemy_pool-9905
+	# hostname/sqlalchemy-neutron/sqlalchemy_pool-19385
+	# hostname/sqlalchemy-neutron/sqlalchemy_pool-6991
 	#
 	# to:
 	#

sqlalchemy_collectd/collectd.py

@@ -2,6 +2,7 @@
 import socket
 import logging
 import threading
+import os
 
 log = logging.getLogger(__name__)
 
@@ -61,6 +62,12 @@ def __init__(self, name, *db_template):
                 ("midterm", VALUE_GAUGE),
                 ("longterm", VALUE_GAUGE)
             )
+
+        note: for all the great effort here in working up types,
+        collectd aggregation plugin doesn't support more than one dsvalue
+        at a time in a record so all this flexibility is all a waste of
+        time :(
+
         """
         self.name = name
         self._value_types = [value_type for dsname, value_type in db_template]
@@ -127,15 +134,39 @@ def send(self, connection, timestamp, *values):
 
 
 class Connection(object):
+    connections = {}
+    create_mutex = threading.Lock()
+
     def __init__(self, host="localhost", port=25826):
         self.host = host
         self.port = port
-        self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self._mutex = threading.Lock()
+        self.socket = None
+        self.pid = None
+
+    def _check_connect(self):
+        if self.socket is None or self.pid != os.getpid():
+            self.pid = os.getpid()
+            self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+
+    @classmethod
+    def for_host_port(cls, host, port):
+        cls.create_mutex.acquire()
+        try:
+            key = (host, port)
+            if key not in cls.connections:
+                cls.connections[key] = connection = Connection(host, port)
+                return connection
+            else:
+                return cls.connections[key]
+
+        finally:
+            cls.create_mutex.release()
 
     def send(self, message):
         self._mutex.acquire()
         try:
+            self._check_connect()
             log.debug("sending: %r", message)
             self.socket.sendto(message, (self.host, self.port))
         except IOError:

sqlalchemy_collectd/collector.py

@@ -1,47 +1,111 @@
 from sqlalchemy import event
 import threading
+import weakref
+import logging
+from . import worker
 
 
-class Collector(object):
-    collectors = {}
-    create_mutex = threading.Mutex()
+class CollectionTarget(object):
+    targets = {}
+    create_mutex = threading.Lock()
 
     def __init__(self, name):
         self.name = name
 
+        self.collectors = weakref.WeakSet()
+
         # 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()
 
+        # identifiers for connections where we've seen begin().
+        # doesn't include DBAPI implicit transactions
+        self.transactions = set()
+
     @classmethod
-    def collector_for_name(cls, name):
+    def collection_for_name(cls, name):
         cls.create_mutex.acquire()
         try:
-            if name not in cls.collectors:
-                cls.collectors[name] = collector = Collector(name)
-                return collector
+            if name not in cls.targets:
+                cls.targets[name] = collection = CollectionTarget(name)
+                return collection
             else:
-                return cls.collectors[name]
+                return cls.targets[name]
         finally:
             cls.create_mutex.release()
 
+    @property
+    def num_pools(self):
+        return len(self.collectors)
+
+    @property
+    def num_checkedout(self):
+        checkedout = self.connections.\
+            difference(self.detached).\
+            difference(self.invalidated).\
+            difference(self.checkedin)
+        return len(checkedout)
+
+    @property
+    def num_checkedin(self):
+        return len(self.checkedin)
+
+    @property
+    def num_detached(self):
+        return len(self.detached)
+
+    @property
+    def num_invalidated(self):
+        return len(self.invalidated)
+
+    @property
+    def num_connections(self):
+        return len(self.connections)
+
+    @property
+    def num_transactions(self):
+        return len(self.transactions)
+
+
+class EngineCollector(object):
+
+    def __init__(self, collection_target, engine):
+        self.collection_target = collection_target
+        self.engine = engine
+        collection_target.collectors.add(self)
+
+        eng = 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)
+
+        self.connections = collection_target.connections
+        self.checkedin = collection_target.checkedin
+        self.transactions = collection_target.transactions
+        self.invalidated = collection_target.invalidated
+        self.detached = collection_target.detached
+        self.logger = logging.getLogger("%s.%s" % (__name__, eng.logging_name))
+
     def conn_ident(self, dbapi_connection):
         return id(dbapi_connection)
 
     def _connect_evt(self, dbapi_conn, connection_rec):
+        worker._check_threads_started()
         id_ = self.conn_ident(dbapi_conn)
         self.connections.add(id_)
         self.checkedin.add(id_)
@@ -69,12 +133,23 @@ def _close_evt(self, dbapi_conn, connection_rec):
         self.invalidated.discard(id_)
         self.checkedin.discard(id_)
 
-        if not self.connections.discard(id_):
+        try:
+            self.connections.remove(id_)
+        except KeyError:
             self._warn_missing_connection(dbapi_conn)
 
         # this shouldn't be there
-        if self.detached.discard(id_):
+        if id_ in self.detached:
             self._warn("shouldn't have detached")
+        self.detached.discard(id_)
+
+    def _warn_missing_connection(self, dbapi_conn):
+        self._warn(
+            "connection %s was closed but not part of "
+            "total connections" % dbapi_conn)
+
+    def _warn(self, msg):
+        self.logger.warn(msg)
 
     def _detach_evt(self, dbapi_conn, connection_rec):
         id_ = self.conn_ident(dbapi_conn)
@@ -83,23 +158,15 @@ def _detach_evt(self, dbapi_conn, connection_rec):
     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)
+        try:
+            self.connections.remove(id_)
+        except KeyError:
+            self._warn_missing_connection(dbapi_conn)
+
+
 

sqlalchemy_collectd/plugin.py

@@ -1,5 +1,14 @@
+import os
+import socket
+import sys
+
 from sqlalchemy.engine import CreateEnginePlugin
 
+from . import collectd
+from . import sender
+from . import worker
+from . import collector
+
 
 class Plugin(CreateEnginePlugin):
     def __init__(self, url, kwargs):
@@ -19,4 +28,25 @@ def engine_created(self, engine):
 
         """
 
+        # TODO: all this configurable
+        hostname = socket.gethostname()
+        progname = sys.argv[0]
+
+        collectd_hostname = "localhost"
+        collectd_port = 25826
+
+        sender_ = sender.Sender(hostname, progname)
+        collection_target = collector.CollectionTarget.collection_for_name(
+            progname)
+        collector.EngineCollector(collection_target, engine)
+
+        connection = collectd.Connection.for_host_port(
+            collectd_hostname, collectd_port)
+
+        worker.add_target(
+            connection,
+            collection_target,
+            sender_)
+
+
 

sqlalchemy_collectd/sender.py

@@ -0,0 +1,62 @@
+from . import collectd
+
+
+senders = []
+
+
+def sends(name, type_):
+    collectd_type = collectd.Type(
+        name, ("value", type_)
+    )
+
+    def decorate(fn):
+        senders.append((collectd_type, fn))
+        return fn
+
+    return decorate
+
+
+class Sender(object):
+    def __init__(self, hostname, stats_name):
+        self.hostname = hostname
+        self.stats_name = stats_name
+
+    def send(self, connection, collection_target, timestamp, pid):
+        for collectd_type, sender in senders:
+            message_sender = collectd.MessageSender(
+                collectd_type, self.hostname, "sqlalchemy",
+                plugin_instance=self.stats_name, type_instance=str(pid)
+            )
+            value = sender(collection_target)
+            message_sender.send(connection, timestamp, value)
+
+
+@sends("sqlalchemy_numpools", collectd.VALUE_GAUGE)
+def _numpools(collection_target):
+    return collection_target.num_pools
+
+
+@sends("sqlalchemy_checkedout", collectd.VALUE_GAUGE)
+def _checkedout(collection_target):
+    return collection_target.num_checkedout
+
+
+@sends("sqlalchemy_checkedin", collectd.VALUE_GAUGE)
+def _checkedin(collection_target):
+    return collection_target.num_checkedin
+
+
+@sends("sqlalchemy_detached", collectd.VALUE_GAUGE)
+def _detached(collection_target):
+    return collection_target.num_detached
+
+
+@sends("sqlalchemy_invalidated", collectd.VALUE_GAUGE)
+def _invalidated(collection_target):
+    return collection_target.num_invalidated
+
+
+@sends("sqlalchemy_connections", collectd.VALUE_GAUGE)
+def _connections(collection_target):
+    return collection_target.num_connections
+