X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=policy_module%2Fosm_policy_module%2Fcore%2Fagent.py;h=cdd5dfc9eb8b8b24b8307a20fdfa39daa5b0015c;hb=4515b8f6f5e3548884828eb8d96ac750e37821e3;hp=60da337074706dea4d87624ba2da8927cb7fbeb0;hpb=2f1be6ba8722211068fb678bc1048283f275f0af;p=osm%2FMON.git

diff --git a/policy_module/osm_policy_module/core/agent.py b/policy_module/osm_policy_module/core/agent.py
index 60da337..cdd5dfc 100644
--- a/policy_module/osm_policy_module/core/agent.py
+++ b/policy_module/osm_policy_module/core/agent.py
@@ -25,15 +25,12 @@ import json
 import logging
 from typing import Dict, List
 
-import peewee
 import yaml
-
 from kafka import KafkaConsumer
-from osm_policy_module.core.config import Config
-from osm_policy_module.common.lcm_client import LcmClient
-
 from osm_policy_module.common.alarm_config import AlarmConfig
+from osm_policy_module.common.lcm_client import LcmClient
 from osm_policy_module.common.mon_client import MonClient
+from osm_policy_module.core.config import Config
 from osm_policy_module.core.database import ScalingRecord, ScalingAlarm
 
 log = logging.getLogger(__name__)
@@ -48,11 +45,10 @@ class PolicyModuleAgent:
 
         # Initialize Kafka consumer
         log.info("Connecting to Kafka server at %s", kafka_server)
-        # TODO: Add logic to handle deduplication of messages when using group_id.
-        # See: https://stackoverflow.com/a/29836412
         consumer = KafkaConsumer(bootstrap_servers=kafka_server,
                                  key_deserializer=bytes.decode,
-                                 value_deserializer=bytes.decode)
+                                 value_deserializer=bytes.decode,
+                                 group_id="pm-consumer")
         consumer.subscribe(['lcm_pm', 'alarm_response'])
 
         for message in consumer:
@@ -95,14 +91,28 @@ class PolicyModuleAgent:
                 if message.key == 'notify_alarm':
                     content = json.loads(message.value)
                     alarm_id = content['notify_details']['alarm_uuid']
-                    log.info("Received alarm notification for alarm %s", alarm_id)
+                    metric_name = content['notify_details']['metric_name']
+                    operation = content['notify_details']['operation']
+                    threshold = content['notify_details']['threshold_value']
+                    vdu_name = content['notify_details']['vdu_name']
+                    vnf_member_index = content['notify_details']['vnf_member_index']
+                    ns_id = content['notify_details']['ns_id']
+                    log.info(
+                        "Received alarm notification for alarm %s, \
+                        metric %s, \
+                        operation %s, \
+                        threshold %s, \
+                        vdu_name %s, \
+                        vnf_member_index %s, \
+                        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()
                         lcm_client = LcmClient()
                         log.info("Sending scaling action message for ns: %s", alarm_id)
                         lcm_client.scale(alarm.scaling_record.nsr_id, alarm.scaling_record.name, alarm.action)
                     except ScalingAlarm.DoesNotExist:
-                        log.info("There is no action configured for alarm %.", alarm_id)
+                        log.info("There is no action configured for alarm %s.", alarm_id)
             except Exception:
                 log.exception("Error consuming message: ")