- 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 scale(self, nsr_id: str, scaling_group_name: str, vnf_member_index: int, action: str):
- 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()
-
- def _generate_nslcmop(self, nsr_id: str, scaling_group_name: str, vnf_member_index: int, action: str):
+ """
+ Client to communicate with LCM through the message bus.
+ """
+
+ 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
+
+ async def scale(self, nsr_id: str, scaling_group_name: str, vnf_member_index: str, action: str):
+ """
+ Sends scaling action to LCM through the message bus.
+
+ :param nsr_id: Network service record id
+ :param scaling_group_name: Scaling group name
+ :param vnf_member_index: VNF member index
+ :param action: Scaling action to be executed. Valid values: scale_in, scale_out
+ :return:
+ """
+ log.debug("scale %s %s %s %s", nsr_id, scaling_group_name, vnf_member_index, action)
+ nsr = self.db_client.get_nsr(nsr_id)
+ nslcmop = self._generate_nslcmop(nsr_id, scaling_group_name, vnf_member_index, action, nsr['_admin'])
+ 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: str, action: str, admin: dict):
+ """
+ Builds scaling nslcmop.
+
+ :param nsr_id: Network service record id
+ :param scaling_group_name: Scaling group name
+ :param vnf_member_index: VNF member index
+ :param action: Scaling action to be executed. Valid values: scale_in, scale_out
+ :param admin: Dict corresponding to the _admin section of the nsr. Required keys: projects_read, projects_write.
+ :return:
+ """
+ log.debug("_generate_nslcmop %s %s %s %s %s", nsr_id, scaling_group_name, vnf_member_index, action, admin)