X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_mon%2Fcollector%2Fcollector.py;h=ad166e4154f82a3267e9b52d6cf9af9df0d8b9d9;hb=a2eeb474200b8f9ebcaee6fa68fe52b6e1a5e337;hp=9bf395334e3212dbf423f98c10a0b9fb7cf7c69a;hpb=4da146638bc3838270fa41c9f9fb91961f726c97;p=osm%2FMON.git

diff --git a/osm_mon/collector/collector.py b/osm_mon/collector/collector.py
index 9bf3953..ad166e4 100644
--- a/osm_mon/collector/collector.py
+++ b/osm_mon/collector/collector.py
@@ -20,99 +20,41 @@
 # For those usages not covered by the Apache License, Version 2.0 please
 # contact: bdiaz@whitestack.com or glavado@whitestack.com
 ##
-import json
 import logging
-import random
-import uuid
-from collections import Iterable
+import time
 
-from kafka import KafkaProducer, KafkaConsumer
-from osm_common import dbmongo
-from prometheus_client.core import GaugeMetricFamily
-
-from osm_mon.core.settings import Config
+from osm_mon.collector.backends.prometheus import PrometheusBackend
+from osm_mon.collector.service import CollectorService
+from osm_mon.core.config import Config
 
 log = logging.getLogger(__name__)
 
-
-class MonCollector:
-    def __init__(self):
-        cfg = Config.instance()
-        self.kafka_server = cfg.BROKER_URI
-        self.common_db_host = cfg.MONGO_URI.split(':')[0]
-        self.common_db_port = cfg.MONGO_URI.split(':')[1]
-        self.common_db = dbmongo.DbMongo()
-        self.common_db.db_connect({'host': self.common_db_host, 'port': int(self.common_db_port), 'name': 'osm'})
-        self.producer = KafkaProducer(bootstrap_servers=self.kafka_server,
-                                      key_serializer=str.encode,
-                                      value_serializer=str.encode)
-        self.consumer = KafkaConsumer(bootstrap_servers=self.kafka_server,
-                                      key_deserializer=bytes.decode,
-                                      value_deserializer=bytes.decode,
-                                      consumer_timeout_ms=10000,
-                                      group_id='mon-collector-' + str(uuid.uuid4()))
-        self.consumer.subscribe(['metric_response'])
-
-    def collect_metrics(self) -> Iterable:
-        # TODO(diazb): Remove dependencies on prometheus_client
-        log.debug("collect_metrics")
-        metrics = {}
-        vnfrs = self.common_db.get_list('vnfrs')
-        for vnfr in vnfrs:
-            nsr_id = vnfr['nsr-id-ref']
-            vnfd = self.common_db.get_one('vnfds', {"_id": vnfr['vnfd-id']})
-            payloads = self._generate_metric_data_payloads(vnfr, vnfd)
-            for payload in payloads:
-                cor_id = payload['correlation_id']
-                metric_name = payload['metric_name']
-                vnf_member_index = payload['vnf_member_index']
-                vdu_name = payload['vdu_name']
-                self.producer.send(topic='metric_request', key='read_metric_data_request',
-                                   value=json.dumps(payload))
-                self.producer.flush()
-                for message in self.consumer:
-                    if message.key == 'read_metric_data_response':
-                        content = json.loads(message.value)
-                        if content['correlation_id'] == cor_id:
-                            if len(content['metrics_data']['metrics_series']):
-                                metric_reading = content['metrics_data']['metrics_series'][-1]
-                                if metric_name not in metrics.keys():
-                                    metrics[metric_name] = GaugeMetricFamily(
-                                        metric_name,
-                                        'OSM metric',
-                                        labels=['ns_id', 'vnf_member_index', 'vdu_name']
-                                    )
-                                metrics[metric_name].add_metric([nsr_id, vnf_member_index, vdu_name],
-                                                                metric_reading)
-                            break
-        return metrics.values()
-
-    @staticmethod
-    def _generate_metric_data_payloads(vnfr: dict, vnfd: dict) -> list:
-        log.debug('_generate_metric_data_payloads')
-        payloads = []
-        nsr_id = vnfr['nsr-id-ref']
-        for vdur in vnfr['vdur']:
-            # This avoids errors when vdur records have not been completely filled
-            if 'name' not in vdur:
-                continue
-            vdu = next(
-                filter(lambda vdu: vdu['id'] == vdur['vdu-id-ref'], vnfd['vdu'])
-            )
-            if 'monitoring-param' in vdu:
-                for param in vdu['monitoring-param']:
-                    metric_name = param['nfvi-metric']
-                    vnf_member_index = vnfr['member-vnf-index-ref']
-                    vdu_name = vdur['name']
-                    cor_id = random.randint(1, 10e7)
-                    payload = {
-                        'correlation_id': cor_id,
-                        'metric_name': metric_name,
-                        'ns_id': nsr_id,
-                        'vnf_member_index': vnf_member_index,
-                        'vdu_name': vdu_name,
-                        'collection_period': 1,
-                        'collection_unit': 'DAY',
-                    }
-                    payloads.append(payload)
-        return payloads
+METRIC_BACKENDS = [
+    PrometheusBackend
+]
+
+
+class Collector:
+    def __init__(self, config: Config):
+        self.conf = config
+        self.service = CollectorService(config)
+        self.backends = []
+        self._init_backends()
+
+    def collect_forever(self):
+        log.debug('collect_forever')
+        while True:
+            try:
+                self.collect_metrics()
+                time.sleep(int(self.conf.get('collector', 'interval')))
+            except Exception:
+                log.exception("Error collecting metrics")
+
+    def collect_metrics(self):
+        metrics = self.service.collect_metrics()
+        for backend in self.backends:
+            backend.handle(metrics)
+
+    def _init_backends(self):
+        for backend in METRIC_BACKENDS:
+            self.backends.append(backend())