+
+ 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')):
+ result = future.result(timeout=int(self.conf.get('collector',
+ 'process_execution_timeout')))
+ metrics.extend(result)
+ log.debug('result = %s' % (result))
+ 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)
+