- def _get_vim_type(self, vim_account_id: str) -> str:
- common_db = CommonDbClient(self.conf)
- vim_account = common_db.get_vim_account(vim_account_id)
- vim_type = vim_account['vim_type']
- if 'config' in vim_account and 'vim_type' in vim_account['config']:
- vim_type = vim_account['config']['vim_type'].lower()
- if vim_type == 'vio' and 'vrops_site' not in vim_account['config']:
- vim_type = 'openstack'
- return vim_type
+ start_time = time.time()
+ # Starting executor pool with pool size process_pool_size. Default process_pool_size is 20
+ with concurrent.futures.ProcessPoolExecutor(
+ self.conf.get("collector", "process_pool_size")
+ ) as executor:
+ log.info(
+ "Started metric collector process pool with pool size %s"
+ % (self.conf.get("collector", "process_pool_size"))
+ )
+ futures = []
+ for vnfr in vnfrs:
+ nsr_id = vnfr["nsr-id-ref"]
+ vnf_member_index = vnfr["member-vnf-index-ref"]
+ vim_account_id = self.common_db.get_vim_account_id(
+ nsr_id, vnf_member_index
+ )
+ futures.append(
+ executor.submit(
+ CollectorService._collect_vim_metrics,
+ self.conf,
+ vnfr,
+ vim_account_id,
+ )
+ )
+ futures.append(
+ executor.submit(
+ CollectorService._collect_vca_metrics, self.conf, vnfr
+ )
+ )
+
+ vims = self.common_db.get_vim_accounts()
+ for vim in vims:
+ futures.append(
+ executor.submit(
+ CollectorService._collect_vim_infra_metrics,
+ self.conf,
+ vim["_id"],
+ )
+ )
+
+ sdncs = self.common_db.get_sdncs()
+ for sdnc in sdncs:
+ futures.append(
+ executor.submit(
+ CollectorService._collect_sdnc_infra_metrics,
+ self.conf,
+ sdnc["_id"],
+ )
+ )
+
+ try:
+ # Wait for future calls to complete till process_execution_timeout. Default is 50 seconds
+ for future in concurrent.futures.as_completed(
+ futures, self.conf.get("collector", "process_execution_timeout")
+ ):
+ try:
+ result = future.result(
+ timeout=int(
+ self.conf.get("collector", "process_execution_timeout")
+ )
+ )
+ metrics.extend(result)
+ log.debug("result = %s" % (result))
+ except keystoneauth1.exceptions.connection.ConnectTimeout as e:
+ log.info("Keystone connection timeout during metric collection")
+ log.debug("Keystone connection timeout exception %s" % (e))
+ except concurrent.futures.TimeoutError as e:
+ # Some processes have not completed due to timeout error
+ log.info(
+ "Some processes have not finished due to TimeoutError exception"
+ )
+ log.debug("concurrent.futures.TimeoutError exception %s" % (e))
+
+ # Shutting down process pool executor
+ CollectorService._stop_process_pool(executor)
+
+ end_time = time.time()
+ log.info("Collection completed in %s seconds", end_time - start_time)
+
+ return metrics