Replaces direct use of aiokafka with osm_common message bus in agent and
lcmclient
Changes config handling to comply with the way it is handled in other modules,
by using a config file and overriding it with env vars.
Adds unit tests for message_bus_client.
Mon client remains using aiokafka directly, as there is no support yet for
auto_offset_reset configuration in osm_common.
Change-Id: I99615287cc934ce310105e86544a6bfe26bc0673
Signed-off-by: Benjamin Diaz <bdiaz@whitestack.com>
diff --git a/osm_policy_module/common/common_db_client.py b/osm_policy_module/common/common_db_client.py
index 2be3693..5731155 100644
--- a/osm_policy_module/common/common_db_client.py
+++ b/osm_policy_module/common/common_db_client.py
@@ -21,18 +21,21 @@
# For those usages not covered by the Apache License, Version 2.0 please
# contact: bdiaz@whitestack.com or glavado@whitestack.com
##
-from osm_common import dbmongo
+from osm_common import dbmongo, dbmemory
from osm_policy_module.core.config import Config
from osm_policy_module.core.exceptions import VdurNotFound
class CommonDbClient:
- def __init__(self):
- cfg = Config.instance()
- self.common_db = dbmongo.DbMongo()
- self.common_db.db_connect({'uri': cfg.OSMPOL_DATABASE_URI,
- 'name': 'osm'})
+ def __init__(self, config: Config):
+ if config.get('database', 'driver') == "mongo":
+ self.common_db = dbmongo.DbMongo()
+ elif config.get('database', 'driver') == "memory":
+ self.common_db = dbmemory.DbMemory()
+ else:
+ raise Exception("Unknown database driver {}".format(config.get('section', 'driver')))
+ self.common_db.db_connect(config.get("database"))
def get_vnfr(self, nsr_id: str, member_index: int):
vnfr = self.common_db.get_one("vnfrs",
@@ -65,3 +68,6 @@
return vdur
raise VdurNotFound('vdur not found for nsr-id %s, member_index %s and vdur_name %s', nsr_id, member_index,
vdur_name)
+
+ def create_nslcmop(self, nslcmop):
+ self.common_db.create("nslcmops", nslcmop)
diff --git a/osm_policy_module/common/lcm_client.py b/osm_policy_module/common/lcm_client.py
index 05378d7..2efa241 100644
--- a/osm_policy_module/common/lcm_client.py
+++ b/osm_policy_module/common/lcm_client.py
@@ -28,22 +28,17 @@
import time
import uuid
-from aiokafka import AIOKafkaProducer
-from osm_common import dbmongo
-
+from osm_policy_module.common.common_db_client import CommonDbClient
+from osm_policy_module.common.message_bus_client import MessageBusClient
from osm_policy_module.core.config import Config
log = logging.getLogger(__name__)
class LcmClient:
- def __init__(self, loop=None):
- cfg = Config.instance()
- self.kafka_server = '{}:{}'.format(cfg.OSMPOL_MESSAGE_HOST,
- cfg.OSMPOL_MESSAGE_PORT)
- self.common_db = dbmongo.DbMongo()
- self.common_db.db_connect({'uri': cfg.OSMPOL_DATABASE_URI,
- 'name': 'osm'})
+ def __init__(self, config: Config, loop=None):
+ self.db_client = CommonDbClient(config)
+ self.msg_bus = MessageBusClient(config)
if not loop:
loop = asyncio.get_event_loop()
self.loop = loop
@@ -51,19 +46,9 @@
async def scale(self, nsr_id: str, scaling_group_name: str, vnf_member_index: int, action: str):
log.debug("scale %s %s %s %s", nsr_id, scaling_group_name, vnf_member_index, action)
nslcmop = self._generate_nslcmop(nsr_id, scaling_group_name, vnf_member_index, action)
- self.common_db.create("nslcmops", nslcmop)
+ self.db_client.create_nslcmop(nslcmop)
log.info("Sending scale action message: %s", json.dumps(nslcmop))
- producer = AIOKafkaProducer(loop=self.loop,
- bootstrap_servers=self.kafka_server,
- key_serializer=str.encode,
- value_serializer=str.encode)
- await producer.start()
- try:
- # Produce message
- await producer.send_and_wait("ns", key="scale", value=json.dumps(nslcmop))
- finally:
- # Wait for all pending messages to be delivered or expire.
- await producer.stop()
+ await self.msg_bus.aiowrite("ns", "scale", nslcmop)
def _generate_nslcmop(self, nsr_id: str, scaling_group_name: str, vnf_member_index: int, action: str):
log.debug("_generate_nslcmop %s %s %s %s", nsr_id, scaling_group_name, vnf_member_index, action)
diff --git a/osm_policy_module/common/message_bus_client.py b/osm_policy_module/common/message_bus_client.py
new file mode 100644
index 0000000..ea5095d
--- /dev/null
+++ b/osm_policy_module/common/message_bus_client.py
@@ -0,0 +1,72 @@
+# -*- coding: utf-8 -*-
+
+# Copyright 2018 Whitestack, LLC
+# *************************************************************
+
+# This file is part of OSM Monitoring module
+# All Rights Reserved to Whitestack, LLC
+
+# 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.
+
+# For those usages not covered by the Apache License, Version 2.0 please
+# contact: bdiaz@whitestack.com or glavado@whitestack.com
+##
+import asyncio
+from typing import List, Callable
+
+from osm_common import msgkafka, msglocal
+
+from osm_policy_module.core.config import Config
+
+
+class MessageBusClient:
+ def __init__(self, config: Config, loop=None):
+ if config.get('message', 'driver') == "local":
+ self.msg_bus = msglocal.MsgLocal()
+ elif config.get('message', 'driver') == "kafka":
+ self.msg_bus = msgkafka.MsgKafka()
+ else:
+ raise Exception("Unknown message bug driver {}".format(config.get('section', 'driver')))
+ self.msg_bus.connect(config.get('message'))
+ if not loop:
+ loop = asyncio.get_event_loop()
+ self.loop = loop
+
+ async def aioread(self, topics: List[str], callback: Callable = None, **kwargs):
+ """
+ Retrieves messages continuously from bus and executes callback for each message consumed.
+ :param topics: List of message bus topics to consume from.
+ :param callback: Async callback function to be called for each message received.
+ :param kwargs: Keyword arguments to be passed to callback function.
+ :return: None
+ """
+ await self.msg_bus.aioread(topics, self.loop, aiocallback=callback, **kwargs)
+
+ async def aiowrite(self, topic: str, key: str, msg: dict):
+ """
+ Writes message to bus.
+ :param topic: Topic to write to.
+ :param key: Key to write to.
+ :param msg: Dictionary containing message to be written.
+ :return: None
+ """
+ await self.msg_bus.aiowrite(topic, key, msg, self.loop)
+
+ async def aioread_once(self, topic: str):
+ """
+ Retrieves last message from bus.
+ :param topic: topic to retrieve message from.
+ :return: tuple(topic, key, message)
+ """
+ result = await self.msg_bus.aioread(topic, self.loop)
+ return result
diff --git a/osm_policy_module/common/mon_client.py b/osm_policy_module/common/mon_client.py
index 5124ed5..76a6f52 100644
--- a/osm_policy_module/common/mon_client.py
+++ b/osm_policy_module/common/mon_client.py
@@ -36,10 +36,9 @@
class MonClient:
- def __init__(self, loop=None):
- cfg = Config.instance()
- self.kafka_server = '{}:{}'.format(cfg.OSMPOL_MESSAGE_HOST,
- cfg.OSMPOL_MESSAGE_PORT)
+ def __init__(self, config: Config, loop=None):
+ self.kafka_server = '{}:{}'.format(config.get('message', 'host'),
+ config.get('message', 'port'))
if not loop:
loop = asyncio.get_event_loop()
self.loop = loop