spawn multiple workers in services

the collector service only ever grabs one connection to the
database at a time. this causes a massive backlog on the message
queue. enable ability to spawn multiple workers for notification
and collector services.

Change-Id: Ie85c4a6eace52c771c3eb8425839d253f8dd74cd
Closes-Bug: #1291054

1 files changed, 23 insertions, 0 deletions

diff --git a/ceilometer/service.py b/ceilometer/service.py
index f69ee3db..76c93010 100644
--- a/ceilometer/service.py
+++ b/ceilometer/service.py
@@ -28,6 +28,7 @@ from ceilometer.openstack.common import gettextutils
 from ceilometer.openstack.common.gettextutils import _  # noqa
 from ceilometer.openstack.common import log
 from ceilometer.openstack.common import rpc
+from ceilometer import utils
 
 
 OPTS = [
@@ -40,6 +41,12 @@ OPTS = [
                     deprecated_group="collector",
                     default=['database'],
                     help='Dispatcher to process data.'),
+    cfg.IntOpt('collector_workers',
+               help='Number of workers for collector service. The default '
+               'will be equal to the number of CPUs available.'),
+    cfg.IntOpt('notification_workers',
+               help='Number of workers for notification service. The default '
+               'will be equal to the number of CPUs available.'),
 ]
 cfg.CONF.register_opts(OPTS)
 
@@ -88,6 +95,11 @@ cfg.CONF.register_cli_opts(CLI_OPTIONS, group="service_credentials")
 LOG = log.getLogger(__name__)
 
 
+class WorkerException(Exception):
+    """Exception for errors relating to service workers
+    """
+
+
 class DispatchedService(object):
 
     DISPATCHER_NAMESPACE = 'ceilometer.dispatcher'
@@ -106,6 +118,17 @@ class DispatchedService(object):
                         self.DISPATCHER_NAMESPACE)
 
 
+def get_workers(name):
+    workers = (cfg.CONF.get('%s_workers' % name) or
+               utils.cpu_count())
+    if workers and workers < 1:
+        msg = (_("%(worker_name)s value of %(workers)s is invalid, "
+                 "must be greater than 0") %
+               {'worker_name': '%s_workers' % name, 'workers': str(workers)})
+        raise WorkerException(msg)
+    return workers
+
+
 def prepare_service(argv=None):
     gettextutils.install('ceilometer', lazy=True)
     rpc.set_defaults(control_exchange='ceilometer')