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 # Map to store vim ids and corresponding vim session objects
  59 vim_sess_map = {}
  60
  61
  62 # Invoked from process executor to initialize the vim session map
  63 def init_session(session_map: dict):
  64     global vim_sess_map
  65     vim_sess_map = session_map
  66
  67
  68 class CollectorService:
  69     def __init__(self, config: Config):
  70         self.conf = config
  71         self.common_db = CommonDbClient(self.conf)
  72         return
  73
  74     # static methods to be executed in the Processes
  75     @staticmethod
  76     def _get_vim_type(conf: Config, vim_account_id: str) -> str:
  77         common_db = CommonDbClient(conf)
  78         vim_account = common_db.get_vim_account(vim_account_id)
  79         vim_type = vim_account["vim_type"]
  80         if "config" in vim_account and "vim_type" in vim_account["config"]:
  81             vim_type = vim_account["config"]["vim_type"].lower()
  82             if vim_type == "vio" and "vrops_site" not in vim_account["config"]:
  83                 vim_type = "openstack"
  84         return vim_type
  85
  86     @staticmethod
  87     def _collect_vim_metrics(conf: Config, vnfr: dict, vim_account_id: str):
  88         # TODO(diazb) Add support for aws
  89         metrics = []
  90         vim_type = CollectorService._get_vim_type(conf, vim_account_id)
  91         log.debug("vim type.....{}".format(vim_type))
  92         if vim_type in VIM_COLLECTORS:
  93             collector = VIM_COLLECTORS[vim_type](conf, vim_account_id, vim_sess_map[vim_account_id])
  94             metrics = collector.collect(vnfr)
  95             log.debug("Collecting vim metrics.....{}".format(metrics))
  96         else:
  97             log.debug("vimtype %s is not supported.", vim_type)
  98         return metrics
  99
 100     @staticmethod
 101     def _collect_vca_metrics(conf: Config, vnfr: dict):
 102         metrics = []
 103         vca_collector = VCACollector(conf)
 104         metrics = vca_collector.collect(vnfr)
 105         log.debug("Collecting vca metrics.....{}".format(metrics))
 106         return metrics
 107
 108     @staticmethod
 109     def _collect_vim_infra_metrics(conf: Config, vim_account_id: str):
 110         log.info("Collecting vim infra metrics")
 111         metrics = []
 112         vim_type = CollectorService._get_vim_type(conf, vim_account_id)
 113         if vim_type in VIM_INFRA_COLLECTORS:
 114             collector = VIM_INFRA_COLLECTORS[vim_type](conf, vim_account_id)
 115             metrics = collector.collect()
 116             log.debug("Collecting vim infra metrics.....{}".format(metrics))
 117         else:
 118             log.debug("vimtype %s is not supported.", vim_type)
 119         return metrics
 120
 121     @staticmethod
 122     def _collect_sdnc_infra_metrics(conf: Config, sdnc_id: str):
 123         log.info("Collecting sdnc metrics")
 124         metrics = []
 125         common_db = CommonDbClient(conf)
 126         sdn_type = common_db.get_sdnc(sdnc_id)["type"]
 127         if sdn_type in SDN_INFRA_COLLECTORS:
 128             collector = SDN_INFRA_COLLECTORS[sdn_type](conf, sdnc_id)
 129             metrics = collector.collect()
 130             log.debug("Collecting sdnc metrics.....{}".format(metrics))
 131         else:
 132             log.debug("sdn_type %s is not supported.", sdn_type)
 133         return metrics
 134
 135     @staticmethod
 136     def _stop_process_pool(executor):
 137         log.info("Shutting down process pool")
 138         try:
 139             log.debug("Stopping residual processes in the process pool")
 140             for pid, process in executor._processes.items():
 141                 if process.is_alive():
 142                     process.terminate()
 143         except Exception as e:
 144             log.info("Exception during process termination")
 145             log.debug("Exception %s" % (e))
 146
 147         try:
 148             # Shutting down executor
 149             log.debug("Shutting down process pool executor")
 150             executor.shutdown()
 151         except RuntimeError as e:
 152             log.info("RuntimeError in shutting down executer")
 153             log.debug("RuntimeError %s" % (e))
 154         return
 155
 156     def collect_metrics(self) -> List[Metric]:
 157         vnfrs = self.common_db.get_vnfrs()
 158         metrics = []
 159
 160         # Get all vim ids regiestered in osm and create their corresponding vim session objects
 161         # Vim ids and their corresponding session objects are stored in vim-session-map
 162         # It optimizes the number of authentication tokens created in vim for metric colleciton
 163         vim_sess_map.clear()
 164         vims = self.common_db.get_vim_accounts()
 165         for vim in vims:
 166             vim_type = CollectorService._get_vim_type(self.conf, vim["_id"])
 167             if vim_type in VIM_INFRA_COLLECTORS:
 168                 collector = VIM_INFRA_COLLECTORS[vim_type](self.conf, vim["_id"])
 169                 vim_sess = collector.vim_session if vim_type == "openstack" else None
 170                 # Populate the vim session map with vim ids and corresponding session objects
 171                 # vim session objects are stopred only for vim type openstack
 172                 if vim_sess:
 173                     vim_sess_map[vim["_id"]] = vim_sess
 174
 175         start_time = time.time()
 176         # Starting executor pool with pool size process_pool_size. Default process_pool_size is 20
 177         # init_session is called to assign the session map to the gloabal vim session map variable
 178         with concurrent.futures.ProcessPoolExecutor(
 179             self.conf.get("collector", "process_pool_size"), initializer=init_session, initargs=(vim_sess_map,)
 180         ) as executor:
 181             log.info(
 182                 "Started metric collector process pool with pool size %s"
 183                 % (self.conf.get("collector", "process_pool_size"))
 184             )
 185             futures = []
 186             for vnfr in vnfrs:
 187                 nsr_id = vnfr["nsr-id-ref"]
 188                 vnf_member_index = vnfr["member-vnf-index-ref"]
 189                 vim_account_id = self.common_db.get_vim_account_id(
 190                     nsr_id, vnf_member_index
 191                 )
 192                 futures.append(
 193                     executor.submit(
 194                         CollectorService._collect_vim_metrics,
 195                         self.conf,
 196                         vnfr,
 197                         vim_account_id,
 198                     )
 199                 )
 200                 futures.append(
 201                     executor.submit(
 202                         CollectorService._collect_vca_metrics, self.conf, vnfr
 203                     )
 204                 )
 205
 206             for vim in vims:
 207                 futures.append(
 208                     executor.submit(
 209                         CollectorService._collect_vim_infra_metrics,
 210                         self.conf,
 211                         vim["_id"],
 212                     )
 213                 )
 214
 215             sdncs = self.common_db.get_sdncs()
 216             for sdnc in sdncs:
 217                 futures.append(
 218                     executor.submit(
 219                         CollectorService._collect_sdnc_infra_metrics,
 220                         self.conf,
 221                         sdnc["_id"],
 222                     )
 223                 )
 224
 225             try:
 226                 # Wait for future calls to complete till process_execution_timeout. Default is 50 seconds
 227                 for future in concurrent.futures.as_completed(
 228                     futures, self.conf.get("collector", "process_execution_timeout")
 229                 ):
 230                     try:
 231                         result = future.result(
 232                             timeout=int(
 233                                 self.conf.get("collector", "process_execution_timeout")
 234                             )
 235                         )
 236                         metrics.extend(result)
 237                         log.debug("result = %s" % (result))
 238                     except keystoneauth1.exceptions.connection.ConnectionError as e:
 239                         log.info("Keystone connection error during metric collection")
 240                         log.debug("Keystone connection error exception %s" % (e))
 241             except concurrent.futures.TimeoutError as e:
 242                 # Some processes have not completed due to timeout error
 243                 log.info(
 244                     "Some processes have not finished due to TimeoutError exception"
 245                 )
 246                 log.debug("concurrent.futures.TimeoutError exception %s" % (e))
 247
 248             # Shutting down process pool executor
 249             CollectorService._stop_process_pool(executor)
 250
 251         end_time = time.time()
 252         log.info("Collection completed in %s seconds", end_time - start_time)
 253
 254         return metrics