Feature 11016: Service KPI Metric Based Scaling of VNF using exporter endpoint in...

[osm/NG-SA.git] / src / osm_ngsa / osm_mon / vim_connectors / azure.py

diff --git a/src/osm_ngsa/osm_mon/vim_connectors/azure.py b/src/osm_ngsa/osm_mon/vim_connectors/azure.py
index 23086c4..4cf20b7 100644 (file)
--- a/src/osm_ngsa/osm_mon/vim_connectors/azure.py
+++ b/src/osm_ngsa/osm_mon/vim_connectors/azure.py
@@ -14,11 +14,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #######################################################################################
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #######################################################################################

+import datetime

 import logging
 from typing import Dict, List
 
 from azure.identity import ClientSecretCredential
 from azure.mgmt.compute import ComputeManagementClient
 import logging
 from typing import Dict, List
 
 from azure.identity import ClientSecretCredential
 from azure.mgmt.compute import ComputeManagementClient

+from azure.mgmt.monitor import MonitorManagementClient

 from azure.profiles import ProfileDefinition
 from osm_mon.vim_connectors.base_vim import VIMConnector
 
 from azure.profiles import ProfileDefinition
 from osm_mon.vim_connectors.base_vim import VIMConnector
 


@@ -26,8 +28,39 @@ from osm_mon.vim_connectors.base_vim import VIMConnector
 log = logging.getLogger(__name__)
 
 
 log = logging.getLogger(__name__)
 
 

-class AzureCollector(VIMConnector):
+METRIC_MAPPINGS = {
+    "cpu_utilization": {
+        "metricname": "Percentage CPU",
+        "aggregation": "Average",
+    },
+    "average_memory_utilization": {
+        "metricname": "Available Memory Bytes",
+        "aggregation": "Average",
+    },
+    "disk_read_ops": {
+        "metricname": "Disk Read Operations/Sec",
+        "aggregation": "Average",
+    },
+    "disk_write_ops": {
+        "metricname": "Disk Write Operations/Sec",
+        "aggregation": "Average",
+    },
+    "disk_read_bytes": {
+        "metricname": "Disk Read Bytes",
+        "aggregation": "Total",
+    },
+    "disk_write_bytes": {
+        "metricname": "Disk Write Bytes",
+        "aggregation": "Total",
+    },
+    # "packets_in_dropped": {},
+    # "packets_out_dropped": {},
+    # "packets_received": {},
+    # "packets_sent": {},
+}
+

 
 

+class AzureCollector(VIMConnector):

     # Translate azure provisioning state to OSM provision state.
     # The first three ones are the transitional status once a user initiated
     # action has been requested. Once the operation is complete, it will
     # Translate azure provisioning state to OSM provision state.
     # The first three ones are the transitional status once a user initiated
     # action has been requested. Once the operation is complete, it will


@@ -103,8 +136,8 @@ class AzureCollector(VIMConnector):
     def __init__(self, vim_account: Dict):
         self.vim_account = vim_account
         self.reload_client = True
     def __init__(self, vim_account: Dict):
         self.vim_account = vim_account
         self.reload_client = True

-        logger = logging.getLogger("azure")
-        logger.setLevel(logging.ERROR)
+        self.vm_sizes = {}
+

         # Store config to create azure subscription later
         self._config = {
             "user": vim_account["vim_user"],
         # Store config to create azure subscription later
         self._config = {
             "user": vim_account["vim_user"],


@@ -128,6 +161,13 @@ class AzureCollector(VIMConnector):
             log.error("Azure resource_group is not specified at config")
             return
 
             log.error("Azure resource_group is not specified at config")
             return
 

+        # REGION_NAME
+        if "region_name" in config:
+            self.region = config.get("region_name")
+        else:
+            log.error("Azure region_name is not specified at config")
+            return
+

     def _reload_connection(self):
         if self.reload_client:
             log.debug("reloading azure client")
     def _reload_connection(self):
         if self.reload_client:
             log.debug("reloading azure client")


@@ -142,11 +182,25 @@ class AzureCollector(VIMConnector):
                     self._config["subscription_id"],
                     profile=self.AZURE_COMPUTE_MGMT_PROFILE,
                 )
                     self._config["subscription_id"],
                     profile=self.AZURE_COMPUTE_MGMT_PROFILE,
                 )

+                # create client
+                self.conn_monitor = MonitorManagementClient(
+                    self.credentials,
+                    self._config["subscription_id"],
+                )

                 # Set to client created
                 self.reload_client = False
             except Exception as e:
                 log.error(e)
 
                 # Set to client created
                 self.reload_client = False
             except Exception as e:
                 log.error(e)
 

+    def _get_region_vm_sizes(self):
+        if len(self.vm_sizes) == 0:
+            log.debug("getting VM sizes available in region")
+            try:
+                for size in self.conn_compute.virtual_machine_sizes.list(self.region):
+                    self.vm_sizes[size.name] = size
+            except Exception as e:
+                log.error(e)
+

     def collect_servers_status(self) -> List[Dict]:
         servers = []
         log.debug("collect_servers_status")
     def collect_servers_status(self) -> List[Dict]:
         servers = []
         log.debug("collect_servers_status")


@@ -171,7 +225,6 @@ class AzureCollector(VIMConnector):
                             status = self.power_state2osm.get(
                                 splitted_status[1], "OTHER"
                             )
                             status = self.power_state2osm.get(
                                 splitted_status[1], "OTHER"
                             )

-                # log.info(f'id: {id}, name: {name}, status: {status}')

                 vm = {
                     "id": id,
                     "name": name,
                 vm = {
                     "id": id,
                     "name": name,


@@ -180,6 +233,7 @@ class AzureCollector(VIMConnector):
                 servers.append(vm)
         except Exception as e:
             log.error(e)
                 servers.append(vm)
         except Exception as e:
             log.error(e)

+

         return servers
 
     def is_vim_ok(self) -> bool:
         return servers
 
     def is_vim_ok(self) -> bool:


@@ -191,3 +245,81 @@ class AzureCollector(VIMConnector):
         except Exception as e:
             log.error(e)
         return status
         except Exception as e:
             log.error(e)
         return status

+
+    def collect_metrics(self, metric_list: List[Dict]) -> List[Dict]:
+        log.debug("collect_metrics")
+        self._reload_connection()
+
+        metric_results = []
+        # VMs RAM cache for calculating "average_memory_utilization" metric
+        cache = {}
+        for metric in metric_list:
+            server = metric["vm_id"]
+            metric_name = metric["metric"]
+            metric_mapping = METRIC_MAPPINGS.get(metric_name)
+            if not metric_mapping:
+                continue
+            if metric_name == "average_memory_utilization" and len(cache) == 0:
+                # storing VMs RAM sizes in cache
+                self._get_region_vm_sizes()
+                try:
+                    for vm in self.conn_compute.virtual_machines.list(
+                        self.resource_group
+                    ):
+                        id = vm.id
+                        size_name = vm.hardware_profile.vm_size
+                        vm_size = self.vm_sizes.get(size_name)
+                        if vm_size:
+                            ram = vm_size.memory_in_mb
+                            cache[id] = ram
+                except Exception as e:
+                    log.error(e)
+            azure_metric_name = metric_mapping["metricname"]
+            azure_aggregation = metric_mapping["aggregation"]
+            end = datetime.datetime.now()
+            init = end - datetime.timedelta(minutes=5)
+            try:
+                metrics_data = self.conn_monitor.metrics.list(
+                    server,
+                    timespan="{}/{}".format(init, end),
+                    interval="PT1M",
+                    metricnames=azure_metric_name,
+                    aggregation=azure_aggregation,
+                )
+            except Exception as e:
+                log.error(e)
+                continue
+            total = 0
+            n_metrics = 0
+            for item in metrics_data.value:
+                log.info("{} ({})".format(item.name.localized_value, item.unit))
+                for timeserie in item.timeseries:
+                    for data in timeserie.data:
+                        if azure_aggregation == "Average":
+                            val = data.average
+                        elif azure_aggregation == "Total":
+                            val = data.total
+                        else:
+                            val = None
+                        log.info("{}: {}".format(data.time_stamp, val))
+                        if val is not None:
+                            total += val
+                            n_metrics += 1
+            if n_metrics > 0:
+                value = total / n_metrics
+                if metric_name == "average_memory_utilization":
+                    ram = cache.get(server)
+                    if ram:
+                        log.info(f"VM RAM = {ram}")
+                        value = ram - (value / 1048576)
+                    else:
+                        log.error(f"Not found RAM value for server {server}")
+                        value = None
+                if value is not None:
+                    log.info(f"value = {value}")
+                    metric["value"] = value
+                    metric_results.append(metric)
+            else:
+                log.info("No metric available")
+
+        return metric_results