osm_policy_module/core/agent.py

   1 # -*- coding: utf-8 -*-
   2
   3 # Copyright 2018 Whitestack, LLC
   4 # *************************************************************
   5
   6 # This file is part of OSM Monitoring module
   7 # All Rights Reserved to Whitestack, LLC
   8
   9 # Licensed under the Apache License, Version 2.0 (the "License"); you may
  10 # not use this file except in compliance with the License. You may obtain
  11 # a copy of the License at
  12
  13 #         http://www.apache.org/licenses/LICENSE-2.0
  14
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  17 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  18 # License for the specific language governing permissions and limitations
  19 # under the License.
  20
  21 # For those usages not covered by the Apache License, Version 2.0 please
  22 # contact: bdiaz@whitestack.com or glavado@whitestack.com
  23 ##
  24 import json
  25 import logging
  26 import threading
  27 from json import JSONDecodeError
  28
  29 import yaml
  30 from kafka import KafkaConsumer
  31
  32 from osm_policy_module.common.db_client import DbClient
  33 from osm_policy_module.common.lcm_client import LcmClient
  34 from osm_policy_module.common.mon_client import MonClient
  35 from osm_policy_module.core import database
  36 from osm_policy_module.core.config import Config
  37 from osm_policy_module.core.database import ScalingRecord, ScalingAlarm
  38
  39 log = logging.getLogger(__name__)
  40
  41
  42 class PolicyModuleAgent:
  43     def __init__(self):
  44         cfg = Config.instance()
  45         self.db_client = DbClient()
  46         self.mon_client = MonClient()
  47         self.kafka_server = '{}:{}'.format(cfg.OSMPOL_MESSAGE_HOST,
  48                                            cfg.OSMPOL_MESSAGE_PORT)
  49
  50     def run(self):
  51         consumer = KafkaConsumer(bootstrap_servers=self.kafka_server,
  52                                  key_deserializer=bytes.decode,
  53                                  value_deserializer=bytes.decode,
  54                                  group_id='pol-consumer')
  55         consumer.subscribe(["ns", "alarm_response"])
  56
  57         for message in consumer:
  58             t = threading.Thread(target=self._process_msg, args=(message.topic, message.key, message.value,))
  59             t.start()
  60
  61     def _process_msg(self, topic, key, msg):
  62         try:
  63             # Check for ns instantiation
  64             if key == 'instantiated':
  65                 try:
  66                     content = json.loads(msg)
  67                 except JSONDecodeError:
  68                     content = yaml.safe_load(msg)
  69                 log.info("Message arrived with topic: %s, key: %s, msg: %s", topic, key, content)
  70                 nslcmop_id = content['nslcmop_id']
  71                 nslcmop = self.db_client.get_nslcmop(nslcmop_id)
  72                 if nslcmop['operationState'] == 'COMPLETED' or nslcmop['operationState'] == 'PARTIALLY_COMPLETED':
  73                     nsr_id = nslcmop['nsInstanceId']
  74                     log.info("Configuring scaling groups for network service with nsr_id: %s", nsr_id)
  75                     self._configure_scaling_groups(nsr_id)
  76                 else:
  77                     log.info(
  78                         "Network service is not in COMPLETED or PARTIALLY_COMPLETED state. "
  79                         "Current state is %s. Skipping...",
  80                         nslcmop['operationState'])
  81
  82             if key == 'notify_alarm':
  83                 try:
  84                     content = json.loads(msg)
  85                 except JSONDecodeError:
  86                     content = yaml.safe_load(msg)
  87                 log.info("Message arrived with topic: %s, key: %s, msg: %s", topic, key, content)
  88                 alarm_id = content['notify_details']['alarm_uuid']
  89                 metric_name = content['notify_details']['metric_name']
  90                 operation = content['notify_details']['operation']
  91                 threshold = content['notify_details']['threshold_value']
  92                 vdu_name = content['notify_details']['vdu_name']
  93                 vnf_member_index = content['notify_details']['vnf_member_index']
  94                 ns_id = content['notify_details']['ns_id']
  95                 log.info(
  96                     "Received alarm notification for alarm %s, \
  97                     metric %s, \
  98                     operation %s, \
  99                     threshold %s, \
 100                     vdu_name %s, \
 101                     vnf_member_index %s, \
 102                     ns_id %s ",
 103                     alarm_id, metric_name, operation, threshold, vdu_name, vnf_member_index, ns_id)
 104                 try:
 105                     alarm = ScalingAlarm.select().where(ScalingAlarm.alarm_id == alarm_id).get()
 106                     lcm_client = LcmClient()
 107                     log.info("Sending scaling action message for ns: %s", alarm_id)
 108                     lcm_client.scale(alarm.scaling_record.nsr_id, alarm.scaling_record.name, alarm.vnf_member_index,
 109                                      alarm.action)
 110                 except ScalingAlarm.DoesNotExist:
 111                     log.info("There is no action configured for alarm %s.", alarm_id)
 112         except Exception:
 113             log.exception("Error consuming message: ")
 114
 115     def _configure_scaling_groups(self, nsr_id: str):
 116         # TODO(diazb): Check for alarm creation on exception and clean resources if needed.
 117         with database.db.atomic():
 118             vnfrs = self.db_client.get_vnfrs(nsr_id)
 119             log.info("Checking %s vnfrs...", len(vnfrs))
 120             for vnfr in vnfrs:
 121                 vnfd = self.db_client.get_vnfd(vnfr['vnfd-id'])
 122                 log.info("Looking for vnfd %s", vnfr['vnfd-id'])
 123                 scaling_groups = vnfd['scaling-group-descriptor']
 124                 vnf_monitoring_params = vnfd['monitoring-param']
 125                 for scaling_group in scaling_groups:
 126                     log.info("Creating scaling record in DB...")
 127                     scaling_record = ScalingRecord.create(
 128                         nsr_id=nsr_id,
 129                         name=scaling_group['name'],
 130                         content=json.dumps(scaling_group)
 131                     )
 132                     log.info("Created scaling record in DB : nsr_id=%s, name=%s, content=%s",
 133                              scaling_record.nsr_id,
 134                              scaling_record.name,
 135                              scaling_record.content)
 136                     for scaling_policy in scaling_group['scaling-policy']:
 137                         for vdur in vnfd['vdu']:
 138                             vdu_monitoring_params = vdur['monitoring-param']
 139                             for scaling_criteria in scaling_policy['scaling-criteria']:
 140                                 vnf_monitoring_param = next(
 141                                     filter(lambda param: param['id'] == scaling_criteria['vnf-monitoring-param-ref'],
 142                                            vnf_monitoring_params))
 143                                 # TODO: Add support for non-nfvi metrics
 144                                 vdu_monitoring_param = next(
 145                                     filter(
 146                                         lambda param: param['id'] == vnf_monitoring_param['vdu-monitoring-param-ref'],
 147                                         vdu_monitoring_params))
 148                                 alarm_uuid = self.mon_client.create_alarm(
 149                                     metric_name=vdu_monitoring_param['nfvi-metric'],
 150                                     ns_id=nsr_id,
 151                                     vdu_name=vdur['name'],
 152                                     vnf_member_index=vnfr['member-vnf-index-ref'],
 153                                     threshold=scaling_criteria['scale-in-threshold'],
 154                                     operation=scaling_criteria['scale-in-relational-operation'],
 155                                     statistic=vnf_monitoring_param['aggregation-type']
 156                                 )
 157                                 ScalingAlarm.create(
 158                                     alarm_id=alarm_uuid,
 159                                     action='scale_in',
 160                                     vnf_member_index=int(vnfr['member-vnf-index-ref']),
 161                                     vdu_name=vdur['name'],
 162                                     scaling_record=scaling_record
 163                                 )
 164                                 alarm_uuid = self.mon_client.create_alarm(
 165                                     metric_name=vdu_monitoring_param['nfvi-metric'],
 166                                     ns_id=nsr_id,
 167                                     vdu_name=vdur['name'],
 168                                     vnf_member_index=vnfr['member-vnf-index-ref'],
 169                                     threshold=scaling_criteria['scale-out-threshold'],
 170                                     operation=scaling_criteria['scale-out-relational-operation'],
 171                                     statistic=vnf_monitoring_param['aggregation-type']
 172                                 )
 173                                 ScalingAlarm.create(
 174                                     alarm_id=alarm_uuid,
 175                                     action='scale_out',
 176                                     vnf_member_index=int(vnfr['member-vnf-index-ref']),
 177                                     vdu_name=vdur['name'],
 178                                     scaling_record=scaling_record
 179                                 )