add metric collector

README.md

@@ -40,7 +40,7 @@ source bin/activate
 
 # clone codebase and install
 git clone /~https://github.com/wmo-im/wis2-gdc.git
-cd wis2-gdc
+cd wis2-gdc-management
 python3 setup.py install
 ```
 
@@ -97,6 +97,34 @@ To adjust service ports, edit [`docker-compose.override.yml`](docker-compose.ove
 
 The [`Makefile`](Makefile) in the root directory provides options to manage the Docker Compose setup.
 
+```bash
+# build all images
+make build
+
+# build all images (no cache)
+make force-build
+
+# start all containers
+make up
+
+# start all containers in dev mode
+make dev
+
+# view all container logs in realtime
+make logs
+
+# login to the wis2-gdc-management container
+make login
+
+# restart all containers
+make restart
+
+# shutdown all containers
+make down
+
+# remove all volumes
+make rm
+```
 
 ## Development
 

docker-compose.override.yml

@@ -26,7 +26,7 @@ services:
       - 1884:1884  # websockets
   wis2-gdc-metrics-collector:
     ports:
-      - 8001:8001
+      - 8006:8006
   wis2-gdc-api:
     ports:
       - 80:80

wis2-gdc-metrics-collector/Dockerfile

@@ -24,8 +24,6 @@ FROM python:3-alpine
 COPY . /app
 RUN pip install -r /app/requirements.txt
 
-EXPOSE 8001
-
 ENV PYTHONUNBUFFERED="true"
 
 ENTRYPOINT [ "python3","-u","/app/metrics_collector.py" ]

wis2-gdc-metrics-collector/metrics_collector.py

@@ -22,117 +22,125 @@
 import json
 import logging
 import os
+import sys
 from urllib.parse import urlparse
 
 import paho.mqtt.client as mqtt_client
-import prometheus_client
-from prometheus_client import Counter, Gauge, Info, start_http_server
-
+from prometheus_client import (
+    Counter, Gauge, Info, start_http_server, REGISTRY, GC_COLLECTOR,
+    PLATFORM_COLLECTOR, PROCESS_COLLECTOR
+)
 
-prometheus_client.REGISTRY.unregister(prometheus_client.GC_COLLECTOR)
-prometheus_client.REGISTRY.unregister(prometheus_client.PLATFORM_COLLECTOR)
-prometheus_client.REGISTRY.unregister(prometheus_client.PROCESS_COLLECTOR)
-
-LOGGER = logging.getLogger(__name__)
+REGISTRY.unregister(GC_COLLECTOR)
+REGISTRY.unregister(PLATFORM_COLLECTOR)
+REGISTRY.unregister(PROCESS_COLLECTOR)
 
 API_URL = os.environ['WIS2_GDC_API_URL']
 BROKER_URL = os.environ['WIS2_GDC_BROKER_URL']
 CENTRE_ID = os.environ['WIS2_GDC_CENTRE_ID']
-GB = os.environ['WIS2_GDC_GB']
+GB = urlparse(os.environ['WIS2_GDC_GB'])
 GB_TOPIC = os.environ['WIS2_GDC_GB_TOPIC']
-HTTP_PORT = 8001
+HTTP_PORT = 8006
+LOGGING_LEVEL = os.environ['WIS2_GDC_LOGGING_LEVEL']
 
+logging.basicConfig(stream=sys.stdout)
+LOGGER = logging.getLogger(__name__)
+LOGGER.setLevel(LOGGING_LEVEL)
 
 # sets metrics as per /~https://github.com/wmo-im/wis2-metric-hierarchy/blob/main/metric-hierarchy/gdc.csv  # noqa
 
-metric_info = Info(
+METRIC_INFO = Info(
     'wis2_gdc_metrics',
     'WIS2 GDC metrics'
 )
 
-metric_passed_total = Counter(
+METRIC_PASSED_TOTAL = Counter(
     'wmo_wis2_gdc_passed_total',
     'Number of metadata records passed validation',
     ['centre_id', 'report_by']
 )
 
-metric_failed_total = Counter(
+METRIC_FAILED_TOTAL = Counter(
     'wmo_wis2_gdc_failed_total',
     'Number of metadata records failed validation',
     ['centre_id', 'report_by']
 )
 
-metric_core_total = Counter(
+METRIC_CORE_TOTAL = Counter(
     'wmo_wis2_gdc_core_total',
     'Number of core metadata records',
     ['centre_id', 'report_by']
 )
 
-metric_recommended_total = Counter(
+METRIC_RECOMMENDED_TOTAL = Counter(
     'wmo_wis2_gdc_recommendedcore_total',
     'Number of recommended metadata records',
     ['centre_id', 'report_by']
 )
 
-metric_kpi_percentage_total = Counter(
+METRIC_KPI_PERCENTAGE_TOTAL = Counter(
     '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']
 )
 
-metric_kpi_percentage_average = Gauge(
+METRIC_KPI_PERCENTAGE_AVERAGE = Gauge(
     'wmo_wis2_gdc_kpi_percentage_average',
     'Average KPI percentage',
     ['centre_id', 'report_by']
 )
 
-metric_kpi_percentage_over80_total = Counter(
+METRIC_KPI_PERCENTAGE_OVER80_TOTAL = Counter(
     '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 = Counter(
     'wmo_wis2_gdc_search_total',
     'Number of search requests (during last monitoring period)',
     ['centre_id', 'report_by']
 )
 
-metric_search_terms = Gauge(
+METRIC_SEARCH_TERMS = Gauge(
     'wmo_wis2_gdc_search_terms',
     'Most popular search terms (e.g. top=1 to top=5)',
     ['top', 'centre_id', 'report_by']
 )
 
-metric_info.info({
+METRIC_INFO.info({
     'centre-id': CENTRE_ID,
     'url': API_URL,
-    'subscribed-to': f'{GB}/{GB_TOPIC}'
+    'subscribed-to': f'{GB.scheme}://{GB.hostname}:{GB.port} (topic: {GB_TOPIC})'  # noqa
 })
 
 
-def collect_metrics():
+def collect_metrics() -> None:
     """
     Subscribe to MQTT wis2-gdc/metrics and collect metrics
 
     :returns: `None`
     """
 
     def _sub_connect(client, userdata, flags, rc):
-        client.subscribe('wis2-gdc/metrics')
+        LOGGER.info('Subscribing to topic wis2-gdc/metrics/#')
+        client.subscribe('wis2-gdc/metrics/#', qos=0)
 
-    def _sub_collect(client, userdata, msg):
-        topic = json.loads(msg.topic)
+    def _sub_message(client, userdata, msg):
+        LOGGER.debug('Processing message')
+        topic = msg.topic
         labels = json.loads(msg.payload)
+        LOGGER.debug(f'Topic: {topic}')
+        LOGGER.debug(f'Labels: {labels}')
 
         if topic == 'wis2-gdc/metrics/passed_total':
-            metric_passed_total.labels(*labels).inc()
+            METRIC_PASSED_TOTAL.labels(*labels).inc()
         if topic == 'wis2-gdc/metrics/failed_total':
-            metric_failed_total.labels(*labels).inc()
+            METRIC_FAILED_TOTAL.labels(*labels).inc()
         elif topic == 'wis2-gdc/metrics/core_total':
-            metric_core_total.labels(*labels).inc()
+            METRIC_CORE_TOTAL.labels(*labels).inc()
         elif topic == 'wis2-gdc/metrics/recommended_total':
-            metric_recommended_total.labels(*labels).inc()
+            METRIC_RECOMMENDED_TOTAL.labels(*labels).inc()
 
     url = urlparse(BROKER_URL)
 
@@ -142,15 +150,16 @@ def _sub_collect(client, userdata, msg):
         LOGGER.info('Setting up MQTT client')
         client = mqtt_client.Client(client_id)
         client.on_connect = _sub_connect
-        client.on_message = _sub_collect
+        client.on_message = _sub_message
         client.username_pw_set(url.username, url.password)
+        LOGGER.info(f'Connecting to {url.hostname}')
         client.connect(url.hostname, url.port)
         client.loop_forever()
     except Exception as err:
         LOGGER.error(err)
 
 
 if __name__ == '__main__':
-    print(f'Starting metrics collector server on port {HTTP_PORT}')
+    LOGGER.info(f'Starting metrics collector server on port {HTTP_PORT}')
     start_http_server(HTTP_PORT)
     collect_metrics()

wis2-gdc.env

@@ -1,14 +1,16 @@
+export WIS2_GDC_LOGGING_LEVEL=DEBUG
 export WIS2_GDC_API_URL=http://localhost
 export WIS2_GDC_API_URL_DOCKER=http://wis2-gdc-api
 export WIS2_GDC_BACKEND_TYPE=Elasticsearch
 export WIS2_GDC_BACKEND_CONNECTION=http://elasticsearch:9200/wis2-discovery-metadata
-export WIS2_GDC_BROKER_URL=mqtt://wis2-gdc:wis2-gdc@mosquitto:1883
+export WIS2_GDC_BROKER_URL=mqtt://wis2-gdc:wis2-gdc@wis2-gdc-broker:1883
 export WIS2_GDC_CENTRE_ID=ca-eccc-msc-global-discovery-catalogue
-export WIS2_GDC_GB=mqtts://everyone:everyone@globalbroker.meteo.fr:8883
+#export WIS2_GDC_GB=mqtts://everyone:everyone@globalbroker.meteo.fr:8883
+export WIS2_GDC_GB=$WIS2_GDC_BROKER_URL
 export WIS2_GDC_GB_TOPIC=origin/a/wis2/+/metadata/#
 export WIS2_GDC_METADATA_ARCHIVE_ZIPFILE=/data/wis2-gdc-archive.zip
 export WIS2_GDC_PUBLISH_REPORTS=true
-export WIS2_GDC_REJECT_ON_FAILING_ETS=true
+export WIS2_GDC_REJECT_ON_FAILING_ETS=false
 export WIS2_GDC_RUN_KPI=true
 
 # global broker links