osm_mon/collector/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 # For those usages not covered by the Apache License, Version 2.0 please
  21 # contact: bdiaz@whitestack.com or glavado@whitestack.com
  22 ##
  23
  24 # This version uses a ProcessThreadPoolExecutor to limit the number of processes launched
  25
  26 import logging
  27 from typing import List
  28 import concurrent.futures
  29 import time
  30 import keystoneauth1.exceptions
  31
  32 from osm_mon.collector.infra_collectors.onos import OnosInfraCollector
  33 from osm_mon.collector.infra_collectors.openstack import OpenstackInfraCollector
  34 from osm_mon.collector.infra_collectors.vio import VIOInfraCollector
  35 from osm_mon.collector.infra_collectors.vmware import VMwareInfraCollector
  36 from osm_mon.collector.metric import Metric
  37 from osm_mon.collector.vnf_collectors.juju import VCACollector
  38 from osm_mon.collector.vnf_collectors.openstack import OpenstackCollector
  39 from osm_mon.collector.vnf_collectors.vio import VIOCollector
  40 from osm_mon.collector.vnf_collectors.vmware import VMwareCollector
  41 from osm_mon.core.common_db import CommonDbClient
  42 from osm_mon.core.config import Config
  43
  44 log = logging.getLogger(__name__)
  45
  46 VIM_COLLECTORS = {
  47     "openstack": OpenstackCollector,
  48     "vmware": VMwareCollector,
  49     "vio": VIOCollector,
  50 }
  51 VIM_INFRA_COLLECTORS = {
  52     "openstack": OpenstackInfraCollector,
  53     "vmware": VMwareInfraCollector,
  54     "vio": VIOInfraCollector,
  55 }
  56 SDN_INFRA_COLLECTORS = {"onosof": OnosInfraCollector, "onos_vpls": OnosInfraCollector}
  57
  58
  59 class CollectorService:
  60     def __init__(self, config: Config):
  61         self.conf = config
  62         self.common_db = CommonDbClient(self.conf)
  63         return
  64
  65     # static methods to be executed in the Processes
  66     @staticmethod
  67     def _get_vim_type(conf: Config, vim_account_id: str) -> str:
  68         common_db = CommonDbClient(conf)
  69         vim_account = common_db.get_vim_account(vim_account_id)
  70         vim_type = vim_account["vim_type"]
  71         if "config" in vim_account and "vim_type" in vim_account["config"]:
  72             vim_type = vim_account["config"]["vim_type"].lower()
  73             if vim_type == "vio" and "vrops_site" not in vim_account["config"]:
  74                 vim_type = "openstack"
  75         return vim_type
  76
  77     @staticmethod
  78     def _collect_vim_metrics(conf: Config, vnfr: dict, vim_account_id: str):
  79         # TODO(diazb) Add support for aws
  80         metrics = []
  81         vim_type = CollectorService._get_vim_type(conf, vim_account_id)
  82         log.debug("vim type.....{}".format(vim_type))
  83         if vim_type in VIM_COLLECTORS:
  84             collector = VIM_COLLECTORS[vim_type](conf, vim_account_id)
  85             metrics = collector.collect(vnfr)
  86             log.debug("Collecting vim metrics.....{}".format(metrics))
  87         else:
  88             log.debug("vimtype %s is not supported.", vim_type)
  89         return metrics
  90
  91     @staticmethod
  92     def _collect_vca_metrics(conf: Config, vnfr: dict):
  93         metrics = []
  94         vca_collector = VCACollector(conf)
  95         metrics = vca_collector.collect(vnfr)
  96         log.debug("Collecting vca metrics.....{}".format(metrics))
  97         return metrics
  98
  99     @staticmethod
 100     def _collect_vim_infra_metrics(conf: Config, vim_account_id: str):
 101         log.info("Collecting vim infra metrics")
 102         metrics = []
 103         vim_type = CollectorService._get_vim_type(conf, vim_account_id)
 104         if vim_type in VIM_INFRA_COLLECTORS:
 105             collector = VIM_INFRA_COLLECTORS[vim_type](conf, vim_account_id)
 106             metrics = collector.collect()
 107             log.debug("Collecting vim infra metrics.....{}".format(metrics))
 108         else:
 109             log.debug("vimtype %s is not supported.", vim_type)
 110         return metrics
 111
 112     @staticmethod
 113     def _collect_sdnc_infra_metrics(conf: Config, sdnc_id: str):
 114         log.info("Collecting sdnc metrics")
 115         metrics = []
 116         common_db = CommonDbClient(conf)
 117         sdn_type = common_db.get_sdnc(sdnc_id)["type"]
 118         if sdn_type in SDN_INFRA_COLLECTORS:
 119             collector = SDN_INFRA_COLLECTORS[sdn_type](conf, sdnc_id)
 120             metrics = collector.collect()
 121             log.debug("Collecting sdnc metrics.....{}".format(metrics))
 122         else:
 123             log.debug("sdn_type %s is not supported.", sdn_type)
 124         return metrics
 125
 126     @staticmethod
 127     def _stop_process_pool(executor):
 128         log.info("Shutting down process pool")
 129         try:
 130             log.debug("Stopping residual processes in the process pool")
 131             for pid, process in executor._processes.items():
 132                 if process.is_alive():
 133                     process.terminate()
 134         except Exception as e:
 135             log.info("Exception during process termination")
 136             log.debug("Exception %s" % (e))
 137
 138         try:
 139             # Shutting down executor
 140             log.debug("Shutting down process pool executor")
 141             executor.shutdown()
 142         except RuntimeError as e:
 143             log.info("RuntimeError in shutting down executer")
 144             log.debug("RuntimeError %s" % (e))
 145         return
 146
 147     def collect_metrics(self) -> List[Metric]:
 148         vnfrs = self.common_db.get_vnfrs()
 149         metrics = []
 150
 151         start_time = time.time()
 152         # Starting executor pool with pool size process_pool_size. Default process_pool_size is 20
 153         with concurrent.futures.ProcessPoolExecutor(
 154             self.conf.get("collector", "process_pool_size")
 155         ) as executor:
 156             log.info(
 157                 "Started metric collector process pool with pool size %s"
 158                 % (self.conf.get("collector", "process_pool_size"))
 159             )
 160             futures = []
 161             for vnfr in vnfrs:
 162                 nsr_id = vnfr["nsr-id-ref"]
 163                 vnf_member_index = vnfr["member-vnf-index-ref"]
 164                 vim_account_id = self.common_db.get_vim_account_id(
 165                     nsr_id, vnf_member_index
 166                 )
 167                 futures.append(
 168                     executor.submit(
 169                         CollectorService._collect_vim_metrics,
 170                         self.conf,
 171                         vnfr,
 172                         vim_account_id,
 173                     )
 174                 )
 175                 futures.append(
 176                     executor.submit(
 177                         CollectorService._collect_vca_metrics, self.conf, vnfr
 178                     )
 179                 )
 180
 181             vims = self.common_db.get_vim_accounts()
 182             for vim in vims:
 183                 futures.append(
 184                     executor.submit(
 185                         CollectorService._collect_vim_infra_metrics,
 186                         self.conf,
 187                         vim["_id"],
 188                     )
 189                 )
 190
 191             sdncs = self.common_db.get_sdncs()
 192             for sdnc in sdncs:
 193                 futures.append(
 194                     executor.submit(
 195                         CollectorService._collect_sdnc_infra_metrics,
 196                         self.conf,
 197                         sdnc["_id"],
 198                     )
 199                 )
 200
 201             try:
 202                 # Wait for future calls to complete till process_execution_timeout. Default is 50 seconds
 203                 for future in concurrent.futures.as_completed(
 204                     futures, self.conf.get("collector", "process_execution_timeout")
 205                 ):
 206                     try:
 207                         result = future.result(
 208                             timeout=int(
 209                                 self.conf.get("collector", "process_execution_timeout")
 210                             )
 211                         )
 212                         metrics.extend(result)
 213                         log.debug("result = %s" % (result))
 214                     except keystoneauth1.exceptions.connection.ConnectionError as e:
 215                         log.info("Keystone connection error during metric collection")
 216                         log.debug("Keystone connection error exception %s" % (e))
 217             except concurrent.futures.TimeoutError as e:
 218                 # Some processes have not completed due to timeout error
 219                 log.info(
 220                     "Some processes have not finished due to TimeoutError exception"
 221                 )
 222                 log.debug("concurrent.futures.TimeoutError exception %s" % (e))
 223
 224             # Shutting down process pool executor
 225             CollectorService._stop_process_pool(executor)
 226
 227         end_time = time.time()
 228         log.info("Collection completed in %s seconds", end_time - start_time)
 229
 230         return metrics