Refactor scaling group config and adds partial support for vnf metrics
[osm/POL.git] / osm_policy_module / core / agent.py
index 24f1004..8a1a31c 100644 (file)
@@ -35,7 +35,7 @@ from osm_policy_module.common.lcm_client import LcmClient
 from osm_policy_module.common.mon_client import MonClient
 from osm_policy_module.core import database
 from osm_policy_module.core.config import Config
-from osm_policy_module.core.database import ScalingGroup, ScalingAlarm, ScalingPolicy, ScalingCriteria
+from osm_policy_module.core.database import ScalingGroup, ScalingAlarm, ScalingPolicy, ScalingCriteria, DatabaseManager
 
 log = logging.getLogger(__name__)
 
@@ -53,6 +53,7 @@ class PolicyModuleAgent:
         self.lcm_client = LcmClient(loop=self.loop)
         self.kafka_server = '{}:{}'.format(cfg.OSMPOL_MESSAGE_HOST,
                                            cfg.OSMPOL_MESSAGE_PORT)
+        self.database_manager = DatabaseManager()
 
     def run(self):
         self.loop.run_until_complete(self.start())
@@ -112,7 +113,7 @@ class PolicyModuleAgent:
             ns_id %s ",
             alarm_id, metric_name, operation, threshold, vdu_name, vnf_member_index, ns_id)
         try:
-            alarm = ScalingAlarm.select().where(ScalingAlarm.alarm_id == alarm_id).get()
+            alarm = self.database_manager.get_alarm(alarm_id)
             delta = datetime.datetime.now() - alarm.scaling_criteria.scaling_policy.last_scale
             log.debug("last_scale: %s", alarm.scaling_criteria.scaling_policy.last_scale)
             log.debug("now: %s", datetime.datetime.now())
@@ -121,7 +122,7 @@ class PolicyModuleAgent:
                 log.info("Time between last scale and now is less than cooldown time. Skipping.")
                 return
             log.info("Sending scaling action message for ns: %s", alarm_id)
-            await self.lcm_client.scale(alarm.scaling_criteria.scaling_policy.scaling_group.nsr_id,
+            await self.lcm_client.scale(ns_id,
                                         alarm.scaling_criteria.scaling_policy.scaling_group.name,
                                         alarm.vnf_member_index,
                                         alarm.action)
@@ -148,158 +149,164 @@ class PolicyModuleAgent:
         log.debug("_configure_scaling_groups: %s", nsr_id)
         # TODO: Add support for non-nfvi metrics
         alarms_created = []
-        with database.db.atomic():
+        with database.db.atomic() as tx:
             try:
-                with database.db.atomic():
-                    vnfrs = self.db_client.get_vnfrs(nsr_id)
-                    log.info("Found %s vnfrs", len(vnfrs))
-                    for vnfr in vnfrs:
-                        vnfd = self.db_client.get_vnfd(vnfr['vnfd-id'])
-                        log.info("Looking for vnfd %s", vnfr['vnfd-id'])
-                        scaling_groups = vnfd['scaling-group-descriptor']
-                        vnf_monitoring_params = vnfd['monitoring-param']
-                        for scaling_group in scaling_groups:
+                vnfrs = self.db_client.get_vnfrs(nsr_id)
+                for vnfr in vnfrs:
+                    log.info("Processing vnfr: %s", vnfr)
+                    vnfd = self.db_client.get_vnfd(vnfr['vnfd-id'])
+                    log.info("Looking for vnfd %s", vnfr['vnfd-id'])
+                    if 'scaling-group-descriptor' not in vnfd:
+                        continue
+                    scaling_groups = vnfd['scaling-group-descriptor']
+                    vnf_monitoring_params = vnfd['monitoring-param']
+                    for scaling_group in scaling_groups:
+                        try:
+                            scaling_group_record = ScalingGroup.select().where(
+                                ScalingGroup.nsr_id == nsr_id,
+                                ScalingGroup.vnf_member_index == int(vnfr['member-vnf-index-ref']),
+                                ScalingGroup.name == scaling_group['name']
+                            ).get()
+                            log.info("Found existing scaling group record in DB...")
+                        except ScalingGroup.DoesNotExist:
+                            log.info("Creating scaling group record in DB...")
+                            scaling_group_record = ScalingGroup.create(
+                                nsr_id=nsr_id,
+                                vnf_member_index=vnfr['member-vnf-index-ref'],
+                                name=scaling_group['name'],
+                                content=json.dumps(scaling_group)
+                            )
+                            log.info(
+                                "Created scaling group record in DB : nsr_id=%s, vnf_member_index=%s, name=%s",
+                                scaling_group_record.nsr_id,
+                                scaling_group_record.vnf_member_index,
+                                scaling_group_record.name)
+                        for scaling_policy in scaling_group['scaling-policy']:
+                            if scaling_policy['scaling-type'] != 'automatic':
+                                continue
                             try:
-                                scaling_group_record = ScalingGroup.select().where(
-                                    ScalingGroup.nsr_id == nsr_id,
-                                    ScalingGroup.vnf_member_index == int(vnfr['member-vnf-index-ref']),
-                                    ScalingGroup.name == scaling_group['name']
+                                scaling_policy_record = ScalingPolicy.select().join(ScalingGroup).where(
+                                    ScalingPolicy.name == scaling_policy['name'],
+                                    ScalingGroup.id == scaling_group_record.id
                                 ).get()
-                                log.info("Found existing scaling group record in DB...")
-                            except ScalingGroup.DoesNotExist:
-                                log.info("Creating scaling group record in DB...")
-                                scaling_group_record = ScalingGroup.create(
+                                log.info("Found existing scaling policy record in DB...")
+                            except ScalingPolicy.DoesNotExist:
+                                log.info("Creating scaling policy record in DB...")
+                                scaling_policy_record = ScalingPolicy.create(
                                     nsr_id=nsr_id,
-                                    vnf_member_index=vnfr['member-vnf-index-ref'],
-                                    name=scaling_group['name'],
-                                    content=json.dumps(scaling_group)
+                                    name=scaling_policy['name'],
+                                    cooldown_time=scaling_policy['cooldown-time'],
+                                    scaling_group=scaling_group_record
                                 )
-                                log.info(
-                                    "Created scaling group record in DB : nsr_id=%s, vnf_member_index=%s, name=%s",
-                                    scaling_group_record.nsr_id,
-                                    scaling_group_record.vnf_member_index,
-                                    scaling_group_record.name)
-                            for scaling_policy in scaling_group['scaling-policy']:
-                                if scaling_policy['scaling-type'] != 'automatic':
-                                    continue
+                                log.info("Created scaling policy record in DB : name=%s, scaling_group.name=%s",
+                                         scaling_policy_record.name,
+                                         scaling_policy_record.scaling_group.name)
+
+                            for scaling_criteria in scaling_policy['scaling-criteria']:
                                 try:
-                                    scaling_policy_record = ScalingPolicy.select().join(ScalingGroup).where(
-                                        ScalingPolicy.name == scaling_policy['name'],
-                                        ScalingGroup.id == scaling_group_record.id
+                                    scaling_criteria_record = ScalingCriteria.select().join(ScalingPolicy).where(
+                                        ScalingPolicy.id == scaling_policy_record.id,
+                                        ScalingCriteria.name == scaling_criteria['name']
                                     ).get()
-                                    log.info("Found existing scaling policy record in DB...")
-                                except ScalingPolicy.DoesNotExist:
-                                    log.info("Creating scaling policy record in DB...")
-                                    scaling_policy_record = ScalingPolicy.create(
+                                    log.info("Found existing scaling criteria record in DB...")
+                                except ScalingCriteria.DoesNotExist:
+                                    log.info("Creating scaling criteria record in DB...")
+                                    scaling_criteria_record = ScalingCriteria.create(
                                         nsr_id=nsr_id,
-                                        name=scaling_policy['name'],
-                                        cooldown_time=scaling_policy['cooldown-time'],
-                                        scaling_group=scaling_group_record
+                                        name=scaling_criteria['name'],
+                                        scaling_policy=scaling_policy_record
                                     )
-                                    log.info("Created scaling policy record in DB : name=%s, scaling_group.name=%s",
-                                             scaling_policy_record.name,
-                                             scaling_policy_record.scaling_group.name)
-
-                                for scaling_criteria in scaling_policy['scaling-criteria']:
-                                    try:
-                                        scaling_criteria_record = ScalingCriteria.select().join(ScalingPolicy).where(
-                                            ScalingPolicy.id == scaling_policy_record.id,
-                                            ScalingCriteria.name == scaling_criteria['name']
-                                        ).get()
-                                        log.info("Found existing scaling criteria record in DB...")
-                                    except ScalingCriteria.DoesNotExist:
-                                        log.info("Creating scaling criteria record in DB...")
-                                        scaling_criteria_record = ScalingCriteria.create(
-                                            nsr_id=nsr_id,
-                                            name=scaling_criteria['name'],
-                                            scaling_policy=scaling_policy_record
-                                        )
-                                        log.info(
-                                            "Created scaling criteria record in DB : name=%s, scaling_policy.name=%s",
-                                            scaling_criteria_record.name,
-                                            scaling_criteria_record.scaling_policy.name)
+                                    log.info(
+                                        "Created scaling criteria record in DB : name=%s, scaling_policy.name=%s",
+                                        scaling_criteria_record.name,
+                                        scaling_criteria_record.scaling_policy.name)
 
-                                    for vdu_ref in scaling_group['vdu']:
-                                        vnf_monitoring_param = next(
-                                            filter(
-                                                lambda param: param['id'] == scaling_criteria[
-                                                    'vnf-monitoring-param-ref'
-                                                ],
-                                                vnf_monitoring_params)
+                                vnf_monitoring_param = next(
+                                    filter(
+                                        lambda param: param['id'] == scaling_criteria[
+                                            'vnf-monitoring-param-ref'
+                                        ],
+                                        vnf_monitoring_params)
+                                )
+                                if 'vdu-monitoring-param' in vnf_monitoring_param:
+                                    vdurs = list(
+                                        filter(
+                                            lambda vdur: vdur['vdu-id-ref'] == vnf_monitoring_param
+                                            ['vdu-monitoring-param']
+                                            ['vdu-ref'],
+                                            vnfr['vdur']
                                         )
-                                        if vdu_ref['vdu-id-ref'] != vnf_monitoring_param['vdu-monitoring-param'][
-                                            'vdu-ref'
-                                        ]:
-                                            continue
-                                        vdu = next(
-                                            filter(lambda vdu: vdu['id'] == vdu_ref['vdu-id-ref'], vnfd['vdu'])
+                                    )
+                                elif 'vdu-metric' in vnf_monitoring_param:
+                                    vdurs = list(
+                                        filter(
+                                            lambda vdur: vdur['vdu-id-ref'] == vnf_monitoring_param
+                                            ['vdu-metric']
+                                            ['vdu-ref'],
+                                            vnfr['vdur']
                                         )
-                                        vdu_monitoring_params = vdu['monitoring-param']
-                                        vdu_monitoring_param = next(
-                                            filter(
-                                                lambda param: param['id'] == vnf_monitoring_param[
-                                                    'vdu-monitoring-param'
-                                                ][
-                                                    'vdu-monitoring-param-ref'
-                                                ],
-                                                vdu_monitoring_params))
-                                        vdurs = list(
-                                            filter(lambda vdur: vdur['vdu-id-ref'] == vnf_monitoring_param[
-                                                'vdu-monitoring-param'
-                                            ][
-                                                'vdu-ref'
-                                            ], vnfr['vdur']))
-                                        for vdur in vdurs:
-                                            try:
-                                                (ScalingAlarm.select()
-                                                 .join(ScalingCriteria)
-                                                 .join(ScalingPolicy)
-                                                 .join(ScalingGroup)
-                                                 .where(
-                                                    ScalingAlarm.vdu_name == vdur['name'],
-                                                    ScalingCriteria.name == scaling_criteria['name'],
-                                                    ScalingPolicy.name == scaling_policy['name'],
-                                                    ScalingGroup.nsr_id == nsr_id
-                                                ).get())
-                                                log.debug("vdu %s already has an alarm configured", vdur['name'])
-                                                continue
-                                            except ScalingAlarm.DoesNotExist:
-                                                pass
-                                            alarm_uuid = await self.mon_client.create_alarm(
-                                                metric_name=vdu_monitoring_param['nfvi-metric'],
-                                                ns_id=nsr_id,
-                                                vdu_name=vdur['name'],
-                                                vnf_member_index=vnfr['member-vnf-index-ref'],
-                                                threshold=scaling_criteria['scale-in-threshold'],
-                                                operation=scaling_criteria['scale-in-relational-operation'],
-                                                statistic=vnf_monitoring_param['aggregation-type']
-                                            )
-                                            ScalingAlarm.create(
-                                                alarm_id=alarm_uuid,
-                                                action='scale_in',
-                                                vnf_member_index=int(vnfr['member-vnf-index-ref']),
-                                                vdu_name=vdur['name'],
-                                                scaling_criteria=scaling_criteria_record
-                                            )
-                                            alarm_uuid = await self.mon_client.create_alarm(
-                                                metric_name=vdu_monitoring_param['nfvi-metric'],
-                                                ns_id=nsr_id,
-                                                vdu_name=vdur['name'],
-                                                vnf_member_index=vnfr['member-vnf-index-ref'],
-                                                threshold=scaling_criteria['scale-out-threshold'],
-                                                operation=scaling_criteria['scale-out-relational-operation'],
-                                                statistic=vnf_monitoring_param['aggregation-type']
-                                            )
-                                            ScalingAlarm.create(
-                                                alarm_id=alarm_uuid,
-                                                action='scale_out',
-                                                vnf_member_index=int(vnfr['member-vnf-index-ref']),
-                                                vdu_name=vdur['name'],
-                                                scaling_criteria=scaling_criteria_record
-                                            )
+                                    )
+                                elif 'vnf-metric' in vnf_monitoring_param:
+                                    log.warning("vnf-metric is not currently supported.")
+                                    continue
+                                else:
+                                    log.warning(
+                                        "Scaling criteria is referring to a vnf-monitoring-param that does not "
+                                        "contain a reference to a vdu or vnf metric.")
+                                    continue
+                                for vdur in vdurs:
+                                    log.info("Creating alarm for vdur %s ", vdur)
+                                    try:
+                                        (ScalingAlarm.select()
+                                         .join(ScalingCriteria)
+                                         .join(ScalingPolicy)
+                                         .join(ScalingGroup)
+                                         .where(
+                                            ScalingAlarm.vdu_name == vdur['name'],
+                                            ScalingCriteria.name == scaling_criteria['name'],
+                                            ScalingPolicy.name == scaling_policy['name'],
+                                            ScalingGroup.nsr_id == nsr_id
+                                        ).get())
+                                        log.debug("vdu %s already has an alarm configured", vdur['name'])
+                                        continue
+                                    except ScalingAlarm.DoesNotExist:
+                                        pass
+                                    alarm_uuid = await self.mon_client.create_alarm(
+                                        metric_name=vnf_monitoring_param['id'],
+                                        ns_id=nsr_id,
+                                        vdu_name=vdur['name'],
+                                        vnf_member_index=vnfr['member-vnf-index-ref'],
+                                        threshold=scaling_criteria['scale-in-threshold'],
+                                        operation=scaling_criteria['scale-in-relational-operation'],
+                                        statistic=vnf_monitoring_param['aggregation-type']
+                                    )
+                                    ScalingAlarm.create(
+                                        alarm_id=alarm_uuid,
+                                        action='scale_in',
+                                        vnf_member_index=int(vnfr['member-vnf-index-ref']),
+                                        vdu_name=vdur['name'],
+                                        scaling_criteria=scaling_criteria_record
+                                    )
+                                    alarm_uuid = await self.mon_client.create_alarm(
+                                        metric_name=vnf_monitoring_param['id'],
+                                        ns_id=nsr_id,
+                                        vdu_name=vdur['name'],
+                                        vnf_member_index=vnfr['member-vnf-index-ref'],
+                                        threshold=scaling_criteria['scale-out-threshold'],
+                                        operation=scaling_criteria['scale-out-relational-operation'],
+                                        statistic=vnf_monitoring_param['aggregation-type']
+                                    )
+                                    ScalingAlarm.create(
+                                        alarm_id=alarm_uuid,
+                                        action='scale_out',
+                                        vnf_member_index=int(vnfr['member-vnf-index-ref']),
+                                        vdu_name=vdur['name'],
+                                        scaling_criteria=scaling_criteria_record
+                                    )
 
             except Exception as e:
                 log.exception("Error configuring scaling groups:")
+                tx.rollback()
                 if len(alarms_created) > 0:
                     log.info("Cleaning alarm resources in MON")
                     for alarm in alarms_created: