upload to docker hub

.drone.yml

@@ -3,16 +3,6 @@ type: docker
 name: default
 
 steps:
-  - name: static_analysis
-    image: "python:3.8"
-    commands:
-      - pip3 install --cache-dir='./.pipcache' pylint bandit mccabe
-      - pip3 install --cache-dir='./.pipcache' -r requirements.txt
-      - find . -name "*.py" -exec python3 -m py_compile '{}' \;
-      - find . -name "*.py" -exec pylint '{}' + || if [ $? -eq 1 ]; then echo "you fail"; fi
-      - find . -name "*.py" -exec python3 -m mccabe --min 3 '{}' + || if [ $? -eq 1 ]; then echo "you fail"; fi
-      - bandit -r . + || if [ $? -eq 1 ]; then echo "you fail"; fi
-
   - name: code-analysis
     image: aosapps/drone-sonar-plugin
     settings:
@@ -34,13 +24,17 @@ steps:
         - latest
         - ${DRONE_BUILD_NUMBER}
 
-  - name: sentry
-    image: tormachris/drone-sentry
+  - name: dockerhub
+    image: plugins/docker
     settings:
-      sentry_project: ${DRONE_REPO_NAME}
-      sentry_domain: sentry.kmlabz.com
-      sentry_token:
-        from_secret: SENTRY_TOKEN
+      repo: birbnetes/${DRONE_REPO_NAME}
+      username:
+        from_secret: DOCKERHUB_USER
+      password:
+        from_secret: DOCKERHUB_PASSWORD
+      tags:
+        - latest
+        - ${DRONE_BUILD_NUMBER}
 
   - name: ms-teams
     image: kuperiu/drone-teams

k8s/configmap.yaml

@@ -6,17 +6,17 @@ metadata:
     app: guard-service
   namespace: birbnetes
 data:
-  SENTRY_DSN: "https://80d27db8c74f4556a19a1bf0180b373f@sentry.kmlabz.com/23"
-  RELEASE_ID: birb-k8s
   GUARD_CLIENT_ID: guard-b50d97f6-29f9-4de5-a96e-f9b69ca69f7f
-  GUARD_SERVICE_RELEASEMODE: release
+  GUARD_MQTT_HOSTNAME: activemq
+  GUARD_MQTT_PASSWORD: de4d2182
+  GUARD_MQTT_PORT: "1883"
+  GUARD_MQTT_TOPIC: command
+  GUARD_MQTT_USERNAME: birbnetes
+  GUARD_RABBITMQ_EXCHANGE: output
   GUARD_RABBITMQ_HOSTNAME: birb-rabbitmq
-  GUARD_RABBITMQ_EXCHANGE: "sample"
+  GUARD_RABBITMQ_PASSWORD: ZgCiSiSO8t
   GUARD_RABBITMQ_USERNAME: user
-  GUARD_RABBITMQ_PASSWORD: 1wZVQnP5vy
-  GUARD_MQTT_HOSTNAME: guard-postgres
-  GUARD_MQTT_PORT: 1883
-  GUARD_MQTT_USERNAME: guard-service
-  GUARD_MQTT_PASSWORD: guard-service-supersecret
-  GUARD_MQTT_TOPIC: guard-service
-  INPUT_SVC_HOSTNAME: input-service
+  GUARD_SERVICE_RELEASEMODE: release
+  SAMPLE_SVC_HOSTNAME: sample-service
+  RELEASE_ID: kmlabz-k8s
+  SENTRY_DSN: https://1d01460ec3094d5bb6c4d78c0a028b08@glitchtip.kmlabz.com/7

k8s/deployment.yaml

@@ -24,5 +24,26 @@ spec:
           envFrom:
             - configMapRef:
                 name: guard-service
+        - name: jaeger-agent
+          image: jaegertracing/jaeger-agent:latest
+          imagePullPolicy: IfNotPresent
+          ports:
+          - containerPort: 5775
+            name: zk-compact-trft
+            protocol: UDP
+          - containerPort: 5778
+            name: config-rest
+            protocol: TCP
+          - containerPort: 6831
+            name: jg-compact-trft
+            protocol: UDP
+          - containerPort: 6832
+            name: jg-binary-trft
+            protocol: UDP
+          - containerPort: 14271
+            name: admin-http
+            protocol: TCP
+          args:
+            - --reporter.grpc.host-port=dns:///woolsey.tormakristof.eu:14250
       imagePullSecrets:
-        - name: regcred
+        - name: regcred

requirements.txt

@@ -1,4 +1,8 @@
 sentry_sdk
 pika
 requests
-paho-mqtt
+paho-mqtt
+
+opentracing~=2.4.0
+jaeger-client
+requests-opentracing

src/app.py

@@ -8,6 +8,12 @@ import pika
 import requests
 from sentry_sdk.integrations.logging import LoggingIntegration
 
+import jaeger_client
+import opentracing
+from opentracing.ext import tags
+from opentracing.propagation import Format
+from requests_opentracing import SessionTracing
+
 import config
 import uuid
 from mqtt_helper import MQTT
@@ -21,38 +27,26 @@ __copyright__ = "Copyright 2020, Birbnetes Team"
 __module_name__ = "app"
 __version__text__ = "1"
 
-if config.SENTRY_DSN:
-    sentry_logging = LoggingIntegration(
-        level=logging.DEBUG,  # Capture info and above as breadcrumbs
-        event_level=logging.ERROR  # Send errors as events
-    )
-    sentry_sdk.init(
-        dsn=config.SENTRY_DSN,
-        send_default_pii=True,
-        integrations=[sentry_logging],
-        traces_sample_rate=1.0,
-        release=config.RELEASE_ID,
-        environment=config.RELEASEMODE,
-        _experiments={"auto_enabling_integrations": True}
-    )
-
 
 def setup_rabbit(mqtt_: MQTT) -> None:
-    logging.info("Connecting to RabbitMQ")
+    logging.info("Connecting to RabbitMQ...")
     credentials = pika.PlainCredentials(config.RABBIT_USERNAME, config.RABBIT_PASSWORD)
-    connection = pika.BlockingConnection(pika.ConnectionParameters(host=config.RABBIT_HOSTNAME,
-                                                                   credentials=credentials,
-                                                                   heartbeat=10,
-                                                                   socket_timeout=5))
-    channel = connection.channel()
-    channel.exchange_declare(exchange=config.RABBIT_EXCHANGE,
-                             exchange_type='fanout')
-    queue = channel.queue_declare(durable=True, auto_delete=True, queue=uuid.uuid4().urn.split(':')[2],
-                                  exclusive=True).method.queue
-    channel.queue_bind(exchange=config.RABBIT_EXCHANGE, queue=queue)
-    channel.basic_consume(queue=queue, on_message_callback=on_message_creator(mqtt_), auto_ack=True)
-    logging.debug("Starting consumption")
-    channel.start_consuming()
+    while True:
+        connection = pika.BlockingConnection(pika.ConnectionParameters(host=config.RABBIT_HOSTNAME,
+                                                                       credentials=credentials,
+                                                                       heartbeat=30,
+                                                                       socket_timeout=45))
+        channel = connection.channel()
+        channel.exchange_declare(exchange=config.RABBIT_EXCHANGE, exchange_type='fanout')
+        queue = channel.queue_declare(durable=True, auto_delete=True, queue=uuid.uuid4().urn.split(':')[2],
+                                      exclusive=True).method.queue
+        channel.queue_bind(exchange=config.RABBIT_EXCHANGE, queue=queue)
+        channel.basic_consume(queue=queue, on_message_callback=on_message_creator(mqtt_), auto_ack=False)
+        logging.debug("Starting consumption...")
+        try:
+            channel.start_consuming()  # this automagically responds to heartbeats
+        except pika.exceptions.AMQPConnectionError as e:
+            logging.warning(f"AMQP Error happened: {e}; Reconnecting...")
 
 
 def on_message_creator(mqtt_: MQTT):
@@ -60,28 +54,107 @@ def on_message_creator(mqtt_: MQTT):
     This generator is used, so that the mqtt object can be injected just when the callback is registered
     """
 
-    def on_message(channel, method_frame, header_frame, body):
-        msg_json = json.loads(body)
-        if 'probability' not in msg_json:
-            logging.error("Malformed message from broker")
-        if msg_json['probability'] > 0.5:
-            r = requests.get(f"http://{config.INPUT_HOSTNAME}/sample/{msg_json['tag']}")
-            r.raise_for_status()
-            if 'device_id' not in r.json():
-                logging.error("Input-service response invalid")
-            logging.info(f"Sending alert command to device {r.json()['device_id']}")
-            mqtt_.publish(subtopic=r.json()['device_id'],
-                          message=json.dumps({"command": "doAlert"}))
+    requests_session = SessionTracing(propagate=True)
+
+    def on_message(
+            channel: pika.channel.Channel,
+            method: pika.spec.Basic.Deliver,
+            properties: pika.spec.BasicProperties,
+            body: bytes
+    ):
+        try:
+            msg_json = json.loads(body)
+        except (json.JSONDecodeError, UnicodeDecodeError) as e:
+            logging.error(f"Malformed message from classifier: {e}")
+            channel.basic_ack(delivery_tag=method.delivery_tag)
+            return
+
+        span_ctx = opentracing.tracer.extract(Format.TEXT_MAP, msg_json)
+        span_tags = {tags.SPAN_KIND: tags.SPAN_KIND_CONSUMER}
+
+        with opentracing.tracer.start_active_span(
+                'handleMessage', finish_on_close=True, child_of=span_ctx, tags=span_tags
+        ) as scope:
+
+            if ('probability' not in msg_json) or ('class' not in msg_json):
+                logging.error("Malformed message from classifier: Missing fields")
+                channel.basic_ack(delivery_tag=method.delivery_tag)
+                return
+
+            # TODO: strurnus should not be hardcoded here
+            if (msg_json['class'] == 'sturnus') and (msg_json['probability'] > config.TRIGGER_LEVEL):
+                scope.span.log_kv({'event': 'decisionMade', 'alerting': True})
+                try:
+                    r = requests_session.get(
+                        f"http://{config.SAMPLE_SVC_HOSTNAME}/sample/{msg_json['tag']}",
+                        timeout=config.INPUT_TIMEOUT
+                    )
+                except requests.exceptions.Timeout:
+                    logging.error(f"Input-service timed out! (Timeout: {config.INPUT_TIMEOUT} sec)")
+                    channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
+                    return
+
+                if r.status_code != 200:
+                    logging.error(f"Input-service status code is not 200: {r.status_code}")
+                    channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
+                    return
+
+                if 'device_id' not in r.json():
+                    logging.error("Input-service response invalid")
+                    channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
+                    return
+
+                logging.info(f"Sending alert command to device {r.json()['device_id']}...")
+                with opentracing.tracer.start_active_span(
+                        'publishAlert',
+                        tags={
+                            tags.SPAN_KIND: tags.SPAN_KIND_PRODUCER,
+                            "device_id": r.json()['device_id']
+                        }
+                ):
+                    mqtt_.publish(
+                        subtopic=r.json()['device_id'],
+                        message=json.dumps({"command": "doAlert"})
+                    )
+
+            else:
+                scope.span.log_kv({'event': 'decisionMade', 'alerting': False})
+                logging.debug(f"Probability is either bellow trigger level, or not the target class. Nothing to do.")
+
+        # This concludes the job
+        channel.basic_ack(delivery_tag=method.delivery_tag)
 
     return on_message
 
 
-if __name__ == "__main__":
-    logging.basicConfig(stream=sys.stdout, format="%(asctime)s - %(name)s [%(levelname)s]: %(message)s",
-                        level=logging.DEBUG if '--debug' in sys.argv else logging.INFO)
-    logging.info("Guard service starting")
+def main():
+    logging.basicConfig(
+        stream=sys.stdout,
+        format="%(asctime)s - %(name)s [%(levelname)s]: %(message)s",
+        level=config.LOG_LEVEL
+    )
+    if config.SENTRY_DSN:
+        sentry_logging = LoggingIntegration(
+            level=logging.DEBUG,  # Capture info and above as breadcrumbs
+            event_level=logging.ERROR  # Send errors as events
+        )
+        sentry_sdk.init(
+            dsn=config.SENTRY_DSN,
+            send_default_pii=True,
+            integrations=[sentry_logging],
+            traces_sample_rate=0.0,
+            release=config.RELEASE_ID,
+            environment=config.RELEASEMODE,
+            _experiments={"auto_enabling_integrations": True}
+        )
+    jaeger_client.Config(config={}, service_name='guard-service', validate=True).initialize_tracer()
+    logging.info("Guard service starting...")
     mqtt = MQTT()
     mqtt.topic = config.MQTT_TOPIC
     mqtt.connect()
     mqtt.client.loop_start()  # Start MQTT event loop on a different thread
     setup_rabbit(mqtt)
+
+
+if __name__ == "__main__":
+    main()

src/config.py

@@ -1,5 +1,7 @@
 #!/usr/bin/env python3
 import os
+import sys
+import logging
 
 
 """
@@ -29,4 +31,12 @@ MQTT_USERNAME = os.getenv("GUARD_MQTT_USERNAME", "guard-service")
 MQTT_PASSWORD = os.getenv("GUARD_MQTT_PASSWORD", "guard-service")
 MQTT_TOPIC = os.getenv("GUARD_MQTT_TOPIC", "guard-service")
 
-INPUT_HOSTNAME = os.getenv("INPUT_SVC_HOSTNAME", "input-service")
+SAMPLE_SVC_HOSTNAME = os.getenv("SAMPLE_SVC_HOSTNAME", "input-service")
+INPUT_TIMEOUT = int(os.environ.get("INPUT_SVC_TIMEOUT", 5))
+TRIGGER_LEVEL = float(os.environ.get("TRIGGER_LEVEL", 0.51))
+
+LOG_LEVEL = logging.DEBUG if (
+                                     '--debug' in sys.argv
+                             ) or (
+                                     os.environ.get('DEBUG', '0').lower() in ['yes', 'true', '1']
+                             ) else logging.INFO