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