Re-use rpc transport

Solum is currently not re-using the rpc transport, this
will cause connections to leak. This patch introduces
a commonly shared rpc pattern for openstack services
that ensures that the transport is always re-used.

Change-Id: Ibdfb4fa4711b10f66ae4c01b2a10b850d0c42ca1

solum/common/rpc/service.py

@@ -15,13 +15,12 @@
 """Common RPC service and API tools for Solum."""
 
 import eventlet
-from oslo_config import cfg
 import oslo_messaging as messaging
-from oslo_messaging.rpc import dispatcher
 from oslo_serialization import jsonutils
 
 import solum.common.context
 from solum import objects
+from solum import rpc
 
 
 # NOTE(paulczar):
@@ -65,14 +64,9 @@ class Service(object):
 
     def __init__(self, topic, server, handlers):
         serializer = RequestContextSerializer(JsonPayloadSerializer())
-        transport = messaging.get_rpc_transport(cfg.CONF)
         # TODO(asalkeld) add support for version='x.y'
         target = messaging.Target(topic=topic, server=server)
-        access_policy = dispatcher.DefaultRPCAccessPolicy
-        self._server = messaging.get_rpc_server(transport, target, handlers,
-                                                executor='threading',
-                                                serializer=serializer,
-                                                access_policy=access_policy)
+        self._server = rpc.get_server(target, handlers, serializer)
 
     def serve(self):
         objects.load()
@@ -81,16 +75,14 @@ class Service(object):
 
 
 class API(object):
-    def __init__(self, transport=None, context=None, topic=None):
+    def __init__(self, context=None, topic=None):
         serializer = RequestContextSerializer(JsonPayloadSerializer())
-        if transport is None:
-            transport = messaging.get_rpc_transport(cfg.CONF)
         self._context = context
         if topic is None:
             topic = ''
         target = messaging.Target(topic=topic)
-        self._client = messaging.RPCClient(transport, target,
-                                           serializer=serializer)
+        self._client = rpc.get_client(target,
+                                      serializer=serializer)
 
     def _call(self, method, *args, **kwargs):
         return self._client.call(self._context, method, *args, **kwargs)

solum/conductor/api.py

@@ -20,10 +20,10 @@ from solum.common.rpc import service
 
 
 class API(service.API):
-    def __init__(self, transport=None, context=None):
+    def __init__(self, context=None):
         cfg.CONF.import_opt('topic', 'solum.conductor.config',
                             group='conductor')
-        super(API, self).__init__(transport, context,
+        super(API, self).__init__(context,
                                   topic=cfg.CONF.conductor.topic)
 
     def build_job_update(self, build_id, status, description, created_image_id,

solum/deployer/api.py

@@ -20,10 +20,10 @@ from solum.common.rpc import service
 
 
 class API(service.API):
-    def __init__(self, transport=None, context=None):
+    def __init__(self, context=None):
         cfg.CONF.import_opt('topic', 'solum.deployer.config',
                             group='deployer')
-        super(API, self).__init__(transport, context,
+        super(API, self).__init__(context,
                                   topic=cfg.CONF.deployer.topic)
 
     def deploy(self, assembly_id, image_loc, image_name, ports):

solum/rpc.py

@@ -0,0 +1,56 @@
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+from oslo_config import cfg
+import oslo_messaging as messaging
+from oslo_messaging.rpc import dispatcher
+
+
+CONF = cfg.CONF
+TRANSPORT = None
+
+
+def init():
+    global TRANSPORT
+    if TRANSPORT is None:
+        TRANSPORT = create_transport(get_transport_url())
+
+
+def get_transport_url(url_str=None):
+    return messaging.TransportURL.parse(CONF, url_str)
+
+
+def get_client(target, serializer):
+    if TRANSPORT is None:
+        init()
+    return messaging.RPCClient(
+        TRANSPORT,
+        target,
+        serializer=serializer,
+    )
+
+
+def get_server(target, endpoints, serializer):
+    if TRANSPORT is None:
+        init()
+    access_policy = dispatcher.DefaultRPCAccessPolicy
+    return messaging.get_rpc_server(
+        TRANSPORT,
+        target,
+        endpoints,
+        executor='threading',
+        serializer=serializer,
+        access_policy=access_policy,
+    )
+
+
+def create_transport(url):
+    return messaging.get_rpc_transport(CONF, url=url)

solum/worker/api.py

@@ -20,10 +20,10 @@ from solum.common.rpc import service
 
 
 class API(service.API):
-    def __init__(self, transport=None, context=None):
+    def __init__(self, context=None):
         cfg.CONF.import_opt('topic', 'solum.worker.config',
                             group='worker')
-        super(API, self).__init__(transport, context,
+        super(API, self).__init__(context,
                                   topic=cfg.CONF.worker.topic)
 
     def build_app(self, verb, build_id, git_info, ports, name, base_image_id,