Ensure we are not exhausting the sqlalchemy pool

Because of eventlet, notification messages can be consumed more quickly
that than they are wroted to the database, this have to effect to exhaust
the sqlalchemy connection pool and raise:

TimeoutError: QueuePool limit of size 5 overflow 10 reached, connection
timed out, timeout 30

The patch creates a eventlet pool to ensure that the pool never exhausts
and prints a warning message in the log to inform that perhaps the pool
size need to be increased.

Fixes bug #1243251

Change-Id: I5ccdc141c441db32fab361783a31a96015516878

ceilometer/storage/impl_sqlalchemy.py

@@ -19,6 +19,7 @@
 
 from __future__ import absolute_import
 import datetime
+import eventlet
 import operator
 import os
 import types
@@ -27,6 +28,7 @@ from sqlalchemy import and_
 from sqlalchemy import desc
 from sqlalchemy import func
 from sqlalchemy.orm import aliased
+from sqlalchemy import pool
 
 from ceilometer.openstack.common.db import exception as dbexc
 import ceilometer.openstack.common.db.sqlalchemy.session as sqlalchemy_session
@@ -174,6 +176,14 @@ class Connection(base.Connection):
             conf.database.connection = \
                 os.environ.get('CEILOMETER_TEST_SQL_URL', url)
 
+        session = sqlalchemy_session.get_session()
+        engine = session.get_bind()
+        if isinstance(engine.pool, pool.QueuePool):
+            poolsize = engine.pool.size() + engine.pool._max_overflow
+            self.pool = eventlet.GreenPool(poolsize)
+        else:
+            self.pool = None
+
     def upgrade(self):
         session = sqlalchemy_session.get_session()
         migration.db_sync(session.get_bind())
@@ -219,8 +229,17 @@ class Connection(base.Connection):
             setattr(obj, k, kwargs[k])
         return obj
 
+    def record_metering_data(self, data):
+        if self.pool:
+            if self.pool.waiting() > 0:
+                LOG.warn(_("Sqlalchemy connection pool is full, "
+                           "perhaps pool_size should be increased"))
+            self.pool.spawn(self._real_record_metering_data, data)
+        else:
+            self._real_record_metering_data(data)
+
     @classmethod
-    def record_metering_data(cls, data):
+    def _real_record_metering_data(cls, data):
         """Write the data to the backend storage system.
 
         :param data: a dictionary such as returned by