# For those usages not covered by the Apache License, Version 2.0 please
# contact: bdiaz@whitestack.com or glavado@whitestack.com
##
+import asyncio
import datetime
import json
import logging
import time
import uuid
-from kafka import KafkaProducer
-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):
- cfg = Config.instance()
- self.kafka_server = '{}:{}'.format(cfg.OSMPOL_MESSAGE_HOST,
- cfg.OSMPOL_MESSAGE_PORT)
- self.producer = KafkaProducer(bootstrap_servers=self.kafka_server,
- key_serializer=str.encode,
- value_serializer=str.encode)
- self.common_db = dbmongo.DbMongo()
- self.common_db.db_connect({'host': cfg.OSMPOL_DATABASE_HOST,
- 'port': int(cfg.OSMPOL_DATABASE_PORT),
- '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
- def scale(self, nsr_id: str, scaling_group_name: str, vnf_member_index: int, action: str):
+ async def scale(self, nsr_id: str, scaling_group_name: str, vnf_member_index: str, 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)
- log.info("Sending scale action message: %s", json.dumps(nslcmop))
- self.producer.send(topic='ns', key='scale', value=json.dumps(nslcmop))
- self.producer.flush()
+ self.db_client.create_nslcmop(nslcmop)
+ log.debug("Sending scale action message: %s", json.dumps(nslcmop))
+ 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):
+ def _generate_nslcmop(self, nsr_id: str, scaling_group_name: str, vnf_member_index: str, action: str):
+ log.debug("_generate_nslcmop %s %s %s %s", nsr_id, scaling_group_name, vnf_member_index, action)
_id = str(uuid.uuid4())
now = time.time()
params = {
"scaleVnfType": action.upper(),
"scaleByStepData": {
"scaling-group-descriptor": scaling_group_name,
- "member-vnf-index": str(vnf_member_index)
+ "member-vnf-index": vnf_member_index
}
},
"scaleTime": "{}Z".format(datetime.datetime.utcnow().isoformat())