upload to docker hub

.drone.yml

@@ -3,30 +3,6 @@ type: docker
 name: default
 
 steps:
-  - name: restore-cache-with-filesystem
-    image: meltwater/drone-cache
-    settings:
-      backend: "filesystem"
-      restore: true
-      cache_key: "{{ .Repo.Name }}"
-      archive_format: "gzip"
-      filesystem_cache_root: "/tmp/cache"
-      mount:
-        - '.pipcache'
-    volumes:
-      - name: cache
-        path: /tmp/cache
-
-  - 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:
@@ -35,21 +11,6 @@ steps:
       sonar_token:
         from_secret: SONAR_CODE
 
-  - name: rebuild-cache-with-filesystem
-    image: meltwater/drone-cache:dev
-    pull: true
-    settings:
-      backend: "filesystem"
-      rebuild: true
-      cache_key: "{{ .Repo.Name }}"
-      archive_format: "gzip"
-      filesystem_cache_root: "/tmp/cache"
-      mount:
-        - '.pipcache'
-    volumes:
-      - name: cache
-        path: /tmp/cache
-
   - name: kaniko
     image: banzaicloud/drone-kaniko
     settings:
@@ -63,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
@@ -78,8 +43,3 @@ steps:
         from_secret: TEAMS_WEBHOOK
     when:
       status: [ failure ]
-
-volumes:
-  - name: cache
-    host:
-      path: "/tmp/cache"

.idea/.gitignore

@@ -1,8 +0,0 @@
-# Default ignored files
-/shelf/
-/workspace.xml
-# Datasource local storage ignored files
-/dataSources/
-/dataSources.local.xml
-# Editor-based HTTP Client requests
-/httpRequests/

.idea/guard-service.iml

@@ -1,15 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<module type="PYTHON_MODULE" version="4">
-  <component name="NewModuleRootManager">
-    <content url="file://$MODULE_DIR$">
-      <sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" />
-      <sourceFolder url="file://$MODULE_DIR$/k8s" type="java-resource" />
-      <excludeFolder url="file://$MODULE_DIR$/venv" />
-    </content>
-    <orderEntry type="jdk" jdkName="Python 3.8 (guard-service)" jdkType="Python SDK" />
-    <orderEntry type="sourceFolder" forTests="false" />
-  </component>
-  <component name="TestRunnerService">
-    <option name="PROJECT_TEST_RUNNER" value="pytest" />
-  </component>
-</module>

.idea/inspectionProfiles/profiles_settings.xml

@@ -1,6 +0,0 @@
-<component name="InspectionProjectProfileManager">
-  <settings>
-    <option name="USE_PROJECT_PROFILE" value="false" />
-    <version value="1.0" />
-  </settings>
-</component>

.idea/misc.xml

@@ -1,7 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project version="4">
-  <component name="JavaScriptSettings">
-    <option name="languageLevel" value="ES6" />
-  </component>
-  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.8 (guard-service)" project-jdk-type="Python SDK" />
-</project>

.idea/modules.xml

@@ -1,8 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project version="4">
-  <component name="ProjectModuleManager">
-    <modules>
-      <module fileurl="file://$PROJECT_DIR$/.idea/guard-service.iml" filepath="$PROJECT_DIR$/.idea/guard-service.iml" />
-    </modules>
-  </component>
-</project>

.idea/vcs.xml

@@ -1,6 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project version="4">
-  <component name="VcsDirectoryMappings">
-    <mapping directory="$PROJECT_DIR$" vcs="Git" />
-  </component>
-</project>

docker-compose.yml

@@ -6,7 +6,7 @@ networks:
 
 services:
   rabbitmq:
-    image: "rabbitmq:3"
+    image: "rabbitmq:3-management"
     hostname: "test-rabbitmq"
     environment:
       RABBITMQ_ERLANG_COOKIE: "akjahsfvbkueasnvfjkhsga"
@@ -24,6 +24,8 @@ services:
     restart: always
     networks:
       - guard
+    ports:
+      - "127.0.0.1:1883:1883"
 
   guard-service:
     image: registry.kmlabz.com/birbnetes/guard-service
@@ -31,6 +33,7 @@ services:
     depends_on:
       - activemq
       - rabbitmq
+      - input-service
     environment:
       SENTRY_DSN: "https://5c925c5a898b4b7d82ee51cd544de623@sentry.kmlabz.com/3"
       GUARD_RABBITMQ_HOSTNAME: rabbitmq

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

@@ -19,9 +19,31 @@ spec:
     spec:
       containers:
         - name: guard-service
-          image: registry.kmlabz.com/birbnetesgit/guard-service
+          image: registry.kmlabz.com/birbnetes/guard-service
+          imagePullPolicy: Always
           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,59 +27,134 @@ __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],
-        release=config.RELEASE_ID,
-        environment=config.RELEASEMODE
-    )
 
-
-def setup_rabbit() -> None:
-    logging.info("Connecting to RabbitMQ")
+def setup_rabbit(mqtt_: MQTT) -> None:
+    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=0,
-                                                                   socket_timeout=5))
-    channel = connection.channel()
-    channel.exchange_declare(exchange=config.RABBIT_EXCHANGE,
-                             exchange_type='fanout',
-                             durable=True,
-                             auto_delete=False)
-    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, 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(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"}))
+def on_message_creator(mqtt_: MQTT):
+    """
+    This generator is used, so that the mqtt object can be injected just when the callback is registered
+    """
+
+    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()
-    setup_rabbit()
+    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

src/mqtt_helper.py

@@ -19,7 +19,7 @@ class MQTT:
     """
 
     def __init__(self, host=config.MQTT_HOSTNAME, port=config.MQTT_PORT, client_id=config.CLIENT_ID, qos=2,
-                 retain=True):
+                 retain=False, username=config.MQTT_USERNAME, password=config.MQTT_PASSWORD):
         """
         Init variables
         :param host:
@@ -35,6 +35,8 @@ class MQTT:
         self._topic = None
         self.qos = qos
         self.retain = retain
+        self.username = username
+        self.password = password
 
     def get_topic(self) -> str:
         """
@@ -61,6 +63,7 @@ class MQTT:
         logging.info("Connecting to MQTT")
         self.client = mqtt.Client(client_id=self.client_id, clean_session=True, userdata=None, protocol=mqtt.MQTTv311,
                                   transport="tcp")
+        self.client.username_pw_set(username=self.username,password=self.password)
         self.port = int(self.port)
         self.client.connect(host=self.host, port=self.port, keepalive=60)