osm_policy_module/autoscaling/service.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 asyncio
  25 import datetime
  26 import json
  27 import logging
  28
  29 from osm_policy_module.common.common_db_client import CommonDbClient
  30 from osm_policy_module.common.lcm_client import LcmClient
  31 from osm_policy_module.common.mon_client import MonClient
  32 from osm_policy_module.core import database
  33 from osm_policy_module.core.config import Config
  34 from osm_policy_module.core.database import ScalingGroup, ScalingAlarm, ScalingPolicy, ScalingCriteria
  35 from osm_policy_module.core.exceptions import VdurNotFound
  36 from osm_policy_module.utils.vnfd import VnfdUtils
  37
  38 log = logging.getLogger(__name__)
  39
  40
  41 class Service:
  42
  43     def __init__(self, config: Config, loop=None):
  44         self.conf = config
  45         if not loop:
  46             loop = asyncio.get_event_loop()
  47         self.loop = loop
  48         self.db_client = CommonDbClient(config)
  49         self.mon_client = MonClient(config, loop=self.loop)
  50         self.lcm_client = LcmClient(config, loop=self.loop)
  51
  52     async def configure_scaling_groups(self, nsr_id: str):
  53         log.debug("_configure_scaling_groups: %s", nsr_id)
  54         alarms_created = []
  55         database.db.connect()
  56         with database.db.atomic() as tx:
  57             try:
  58                 vnfrs = self.db_client.get_vnfrs(nsr_id)
  59                 for vnfr in vnfrs:
  60                     log.info("Processing vnfr: %s", vnfr)
  61                     vnfd = self.db_client.get_vnfd(vnfr['vnfd-id'])
  62                     log.info("Looking for vnfd %s", vnfr['vnfd-id'])
  63                     if 'scaling-group-descriptor' not in vnfd:
  64                         continue
  65                     scaling_groups = vnfd['scaling-group-descriptor']
  66                     vnf_monitoring_params = vnfd['monitoring-param']
  67                     for scaling_group in scaling_groups:
  68                         try:
  69                             scaling_group_record = ScalingGroup.select().where(
  70                                 ScalingGroup.nsr_id == nsr_id,
  71                                 ScalingGroup.vnf_member_index == int(vnfr['member-vnf-index-ref']),
  72                                 ScalingGroup.name == scaling_group['name']
  73                             ).get()
  74                             log.info("Found existing scaling group record in DB...")
  75                         except ScalingGroup.DoesNotExist:
  76                             log.info("Creating scaling group record in DB...")
  77                             scaling_group_record = ScalingGroup.create(
  78                                 nsr_id=nsr_id,
  79                                 vnf_member_index=vnfr['member-vnf-index-ref'],
  80                                 name=scaling_group['name'],
  81                                 content=json.dumps(scaling_group)
  82                             )
  83                             log.info(
  84                                 "Created scaling group record in DB : nsr_id=%s, vnf_member_index=%s, name=%s",
  85                                 scaling_group_record.nsr_id,
  86                                 scaling_group_record.vnf_member_index,
  87                                 scaling_group_record.name)
  88                         for scaling_policy in scaling_group['scaling-policy']:
  89                             if scaling_policy['scaling-type'] != 'automatic':
  90                                 continue
  91                             try:
  92                                 scaling_policy_record = ScalingPolicy.select().join(ScalingGroup).where(
  93                                     ScalingPolicy.name == scaling_policy['name'],
  94                                     ScalingGroup.id == scaling_group_record.id
  95                                 ).get()
  96                                 log.info("Found existing scaling policy record in DB...")
  97                             except ScalingPolicy.DoesNotExist:
  98                                 log.info("Creating scaling policy record in DB...")
  99                                 scaling_policy_record = ScalingPolicy.create(
 100                                     nsr_id=nsr_id,
 101                                     name=scaling_policy['name'],
 102                                     cooldown_time=scaling_policy['cooldown-time'],
 103                                     scaling_group=scaling_group_record
 104                                 )
 105                                 log.info("Created scaling policy record in DB : name=%s, scaling_group.name=%s",
 106                                          scaling_policy_record.name,
 107                                          scaling_policy_record.scaling_group.name)
 108
 109                             for scaling_criteria in scaling_policy['scaling-criteria']:
 110                                 try:
 111                                     scaling_criteria_record = ScalingCriteria.select().join(ScalingPolicy).where(
 112                                         ScalingPolicy.id == scaling_policy_record.id,
 113                                         ScalingCriteria.name == scaling_criteria['name']
 114                                     ).get()
 115                                     log.info("Found existing scaling criteria record in DB...")
 116                                 except ScalingCriteria.DoesNotExist:
 117                                     log.info("Creating scaling criteria record in DB...")
 118                                     scaling_criteria_record = ScalingCriteria.create(
 119                                         nsr_id=nsr_id,
 120                                         name=scaling_criteria['name'],
 121                                         scaling_policy=scaling_policy_record
 122                                     )
 123                                     log.info(
 124                                         "Created scaling criteria record in DB : name=%s, scaling_policy.name=%s",
 125                                         scaling_criteria_record.name,
 126                                         scaling_criteria_record.scaling_policy.name)
 127
 128                                 vnf_monitoring_param = next(
 129                                     filter(
 130                                         lambda param: param['id'] == scaling_criteria[
 131                                             'vnf-monitoring-param-ref'
 132                                         ],
 133                                         vnf_monitoring_params)
 134                                 )
 135                                 if 'vdu-monitoring-param' in vnf_monitoring_param:
 136                                     vdurs = list(
 137                                         filter(
 138                                             lambda vdur: vdur['vdu-id-ref'] == vnf_monitoring_param
 139                                             ['vdu-monitoring-param']
 140                                             ['vdu-ref'],
 141                                             vnfr['vdur']
 142                                         )
 143                                     )
 144                                 elif 'vdu-metric' in vnf_monitoring_param:
 145                                     vdurs = list(
 146                                         filter(
 147                                             lambda vdur: vdur['vdu-id-ref'] == vnf_monitoring_param
 148                                             ['vdu-metric']
 149                                             ['vdu-ref'],
 150                                             vnfr['vdur']
 151                                         )
 152                                     )
 153                                 elif 'vnf-metric' in vnf_monitoring_param:
 154                                     vdu = VnfdUtils.get_mgmt_vdu(vnfd)
 155                                     vdurs = list(
 156                                         filter(
 157                                             lambda vdur: vdur['vdu-id-ref'] == vdu['id'],
 158                                             vnfr['vdur']
 159                                         )
 160                                     )
 161                                 else:
 162                                     log.warning(
 163                                         "Scaling criteria is referring to a vnf-monitoring-param that does not "
 164                                         "contain a reference to a vdu or vnf metric.")
 165                                     continue
 166                                 for vdur in vdurs:
 167                                     log.info("Creating alarm for vdur %s ", vdur)
 168                                     try:
 169                                         (ScalingAlarm.select()
 170                                          .join(ScalingCriteria)
 171                                          .join(ScalingPolicy)
 172                                          .join(ScalingGroup)
 173                                          .where(
 174                                             ScalingAlarm.vdu_name == vdur['name'],
 175                                             ScalingCriteria.name == scaling_criteria['name'],
 176                                             ScalingPolicy.name == scaling_policy['name'],
 177                                             ScalingGroup.nsr_id == nsr_id
 178                                         ).get())
 179                                         log.debug("vdu %s already has an alarm configured", vdur['name'])
 180                                         continue
 181                                     except ScalingAlarm.DoesNotExist:
 182                                         pass
 183                                     alarm_uuid = await self.mon_client.create_alarm(
 184                                         metric_name=vnf_monitoring_param['id'],
 185                                         ns_id=nsr_id,
 186                                         vdu_name=vdur['name'],
 187                                         vnf_member_index=vnfr['member-vnf-index-ref'],
 188                                         threshold=scaling_criteria['scale-in-threshold'],
 189                                         operation=scaling_criteria['scale-in-relational-operation'],
 190                                         statistic=vnf_monitoring_param['aggregation-type']
 191                                     )
 192                                     alarm = ScalingAlarm.create(
 193                                         alarm_uuid=alarm_uuid,
 194                                         action='scale_in',
 195                                         vnf_member_index=int(vnfr['member-vnf-index-ref']),
 196                                         vdu_name=vdur['name'],
 197                                         scaling_criteria=scaling_criteria_record
 198                                     )
 199                                     alarms_created.append(alarm)
 200                                     alarm_uuid = await self.mon_client.create_alarm(
 201                                         metric_name=vnf_monitoring_param['id'],
 202                                         ns_id=nsr_id,
 203                                         vdu_name=vdur['name'],
 204                                         vnf_member_index=vnfr['member-vnf-index-ref'],
 205                                         threshold=scaling_criteria['scale-out-threshold'],
 206                                         operation=scaling_criteria['scale-out-relational-operation'],
 207                                         statistic=vnf_monitoring_param['aggregation-type']
 208                                     )
 209                                     alarm = ScalingAlarm.create(
 210                                         alarm_uuid=alarm_uuid,
 211                                         action='scale_out',
 212                                         vnf_member_index=int(vnfr['member-vnf-index-ref']),
 213                                         vdu_name=vdur['name'],
 214                                         scaling_criteria=scaling_criteria_record
 215                                     )
 216                                     alarms_created.append(alarm)
 217
 218             except Exception as e:
 219                 log.exception("Error configuring scaling groups:")
 220                 tx.rollback()
 221                 if len(alarms_created) > 0:
 222                     log.info("Cleaning alarm resources in MON")
 223                     for alarm in alarms_created:
 224                         await self.mon_client.delete_alarm(alarm.scaling_criteria.scaling_policy.scaling_group.nsr_id,
 225                                                            alarm.vnf_member_index,
 226                                                            alarm.vdu_name,
 227                                                            alarm.alarm_uuid)
 228                 raise e
 229         database.db.close()
 230
 231     async def delete_scaling_groups(self, nsr_id: str):
 232         database.db.connect()
 233         with database.db.atomic() as tx:
 234             try:
 235                 for scaling_group in ScalingGroup.select().where(ScalingGroup.nsr_id == nsr_id):
 236                     for scaling_policy in scaling_group.scaling_policies:
 237                         for scaling_criteria in scaling_policy.scaling_criterias:
 238                             for alarm in scaling_criteria.scaling_alarms:
 239                                 try:
 240                                     await self.mon_client.delete_alarm(
 241                                         alarm.scaling_criteria.scaling_policy.scaling_group.nsr_id,
 242                                         alarm.vnf_member_index,
 243                                         alarm.vdu_name,
 244                                         alarm.alarm_uuid)
 245                                 except ValueError:
 246                                     log.exception("Error deleting alarm in MON %s", alarm.alarm_uuid)
 247                                 alarm.delete_instance()
 248                             scaling_criteria.delete_instance()
 249                         scaling_policy.delete_instance()
 250                     scaling_group.delete_instance()
 251
 252             except Exception as e:
 253                 log.exception("Error deleting scaling groups and alarms:")
 254                 tx.rollback()
 255                 raise e
 256         database.db.close()
 257
 258     async def delete_orphaned_alarms(self, nsr_id):
 259         database.db.connect()
 260         with database.db.atomic() as tx:
 261             try:
 262                 for scaling_group in ScalingGroup.select().where(ScalingGroup.nsr_id == nsr_id):
 263                     for scaling_policy in scaling_group.scaling_policies:
 264                         for scaling_criteria in scaling_policy.scaling_criterias:
 265                             for alarm in scaling_criteria.scaling_alarms:
 266                                 try:
 267                                     self.db_client.get_vdur(nsr_id, alarm.vnf_member_index, alarm.vdu_name)
 268                                 except VdurNotFound:
 269                                     log.info("Deleting orphaned alarm %s", alarm.alarm_uuid)
 270                                     try:
 271                                         await self.mon_client.delete_alarm(
 272                                             alarm.scaling_criteria.scaling_policy.scaling_group.nsr_id,
 273                                             alarm.vnf_member_index,
 274                                             alarm.vdu_name,
 275                                             alarm.alarm_uuid)
 276                                     except ValueError:
 277                                         log.exception("Error deleting alarm in MON %s", alarm.alarm_uuid)
 278                                     alarm.delete_instance()
 279
 280             except Exception as e:
 281                 log.exception("Error deleting orphaned alarms:")
 282                 tx.rollback()
 283                 raise e
 284         database.db.close()
 285
 286     async def scale(self, alarm):
 287         database.db.connect()
 288         with database.db.atomic():
 289             delta = datetime.datetime.now() - alarm.scaling_criteria.scaling_policy.last_scale
 290             if delta.total_seconds() > alarm.scaling_criteria.scaling_policy.cooldown_time:
 291                 log.info("Sending scaling action message for ns: %s",
 292                          alarm.scaling_criteria.scaling_policy.scaling_group.nsr_id)
 293                 await self.lcm_client.scale(alarm.scaling_criteria.scaling_policy.scaling_group.nsr_id,
 294                                             alarm.scaling_criteria.scaling_policy.scaling_group.name,
 295                                             alarm.vnf_member_index,
 296                                             alarm.action)
 297                 alarm.scaling_criteria.scaling_policy.last_scale = datetime.datetime.now()
 298                 alarm.scaling_criteria.scaling_policy.save()
 299             else:
 300                 log.info("Time between last scale and now is less than cooldown time. Skipping.")
 301         database.db.close()
 302
 303     def get_alarm(self, alarm_uuid: str):
 304         database.db.connect()
 305         with database.db.atomic():
 306             alarm = ScalingAlarm.select().where(ScalingAlarm.alarm_uuid == alarm_uuid).get()
 307         database.db.close()
 308         return alarm
 309
 310     def get_nslcmop(self, nslcmop_id):
 311         return self.db_client.get_nslcmop(nslcmop_id)