update internal message bus structure

wis2-gdc-management/wis2_gdc/registrar.py

@@ -23,6 +23,7 @@
 import json
 import logging
 from pathlib import Path
+from typing import Union
 
 import click
 import requests
@@ -137,24 +138,43 @@ def register(self, metadata: dict) -> None:
                 LOGGER.info('Publishing KPI report to broker')
                 self.broker.pub(topic, json.dumps(kpi_results))
 
+                kpi_labels = [self.metadata['id']] + centre_id_labels
+
+                self._process_record_metric(
+                    self.metadata['id'], 'kpi_percentage_total',
+                    kpi_labels, kpi_results['summary']['percentage'])
+
     def _process_record_metric(self, identifier: str, metric_name: str,
-                               labels: list) -> None:
+                               labels: list,
+                               value: Union[str, int, float] = None) -> None:
         """
         Helper function to process record metric
 
         :param identifier: identifier of metadata record
         :param metric_name: `str` of name of metric
         :param labels: `list` of labels to apply
+        :param value: optional value(s) to set
 
         :returns: `None`
         """
 
-        if self.backend.exists(identifier):
-            LOGGER.debug('Record exists; publishing metric')
-        else:
-            LOGGER.debug('Record does not exist; not publishing metric')
+        publish_metric = True
+
+        message_payload = {
+            'labels': labels
+        }
+
+        if value is not None:
+            message_payload['value'] = value
+
+        if self.backend.exists(identifier) and len(labels) == 2:
+            LOGGER.debug('Record exists; not publishing metric')
+            publish_metric = False
+
+        if publish_metric:
+            LOGGER.debug('Record does not exist; publishing metric')
             self.broker.pub(f'wis2-gdc/metrics/{metric_name}',
-                            json.dumps(labels))
+                            json.dumps(message_payload))
 
     def _run_ets(self) -> dict:
         """

wis2-gdc-metrics-collector/metrics_collector.py

@@ -78,7 +78,7 @@
     ['centre_id', 'report_by']
 )
 
-METRIC_KPI_PERCENTAGE_TOTAL = Counter(
+METRIC_KPI_PERCENTAGE_TOTAL = Gauge(
     'wmo_wis2_gdc_kpi_percentage_total',
     'KPI percentage for a single metadata record (metadata_id equals WCMP2 id)',  # noqa
     ['metadata_id', 'centre_id', 'report_by']
@@ -90,13 +90,13 @@
     ['centre_id', 'report_by']
 )
 
-METRIC_KPI_PERCENTAGE_OVER80_TOTAL = Counter(
+METRIC_KPI_PERCENTAGE_OVER80_TOTAL = Gauge(
     'wmo_wis2_gdc_kpi_percentage_over80_total',
     'Number of metadata records with KPI percentage over 80',
     ['centre_id', 'report_by']
 )
 
-METRIC_SEARCH_TOTAL = Counter(
+METRIC_SEARCH_TOTAL = Gauge(
     'wmo_wis2_gdc_search_total',
     'Number of search requests (during last monitoring period)',
     ['centre_id', 'report_by']
@@ -129,9 +129,12 @@ def _sub_connect(client, userdata, flags, rc):
     def _sub_message(client, userdata, msg):
         LOGGER.debug('Processing message')
         topic = msg.topic
-        labels = json.loads(msg.payload)
+        payload = json.loads(msg.payload)
+        labels = payload['labels']
+        value = payload.get('value')
         LOGGER.debug(f'Topic: {topic}')
-        LOGGER.debug(f'Labels: {labels}')
+        LOGGER.debug(f"Labels: {payload['labels']}")
+        LOGGER.debug(f"Value: {payload.get('labels')}")
 
         if topic == 'wis2-gdc/metrics/passed_total':
             METRIC_PASSED_TOTAL.labels(*labels).inc()
@@ -141,6 +144,8 @@ def _sub_message(client, userdata, msg):
             METRIC_CORE_TOTAL.labels(*labels).inc()
         elif topic == 'wis2-gdc/metrics/recommended_total':
             METRIC_RECOMMENDED_TOTAL.labels(*labels).inc()
+        elif topic == 'wis2-gdc/metrics/kpi_percentage_total':
+            METRIC_KPI_PERCENTAGE_TOTAL.labels(*labels).set(value)
 
     url = urlparse(BROKER_URL)