upload to docker hub

.dockerignore

@@ -0,0 +1,140 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+#Pycharm
+.idea/
+
+*.md
+.gitignore
+.git/
+*.yml
+contrib/*
+postman/*
+*.wav

.drone.yml

@@ -0,0 +1,45 @@
+kind: pipeline
+type: docker
+name: default
+
+steps:
+  - name: code-analysis
+    image: aosapps/drone-sonar-plugin
+    settings:
+      sonar_host:
+        from_secret: SONAR_HOST
+      sonar_token:
+        from_secret: SONAR_CODE
+
+  - name: kaniko
+    image: banzaicloud/drone-kaniko
+    settings:
+      registry: registry.kmlabz.com
+      repo: birbnetes/${DRONE_REPO_NAME}
+      username:
+        from_secret: DOCKER_USERNAME
+      password:
+        from_secret: DOCKER_PASSWORD
+      tags:
+        - latest
+        - ${DRONE_BUILD_NUMBER}
+
+  - name: dockerhub
+    image: plugins/docker
+    settings:
+      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
+    settings:
+      webhook:
+        from_secret: TEAMS_WEBHOOK
+    when:
+      status: [ failure ]

.gitignore

@@ -129,3 +129,4 @@ dmypy.json
 # Pyre type checker
 .pyre/
 
+.idea/

Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.8-slim
+
+ENV TZ Europe/Budapest
+RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
+
+WORKDIR /app
+
+ARG RELEASE_ID
+ENV RELEASE_ID ${RELEASE_ID:-""}
+
+COPY requirements.txt ./
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY ./src .
+
+ENTRYPOINT ["python3", "app.py"]

README.md

@@ -1,3 +1,5 @@
 # guard-service
 
 Service responsible for triggering alerts.
+
+[![Build Status](https://drone.kmlabz.com/api/badges/birbnetes/guard-service/status.svg)](https://drone.kmlabz.com/birbnetes/guard-service)

docker-compose.yml

@@ -0,0 +1,55 @@
+version: '3.4'
+
+networks:
+  guard:
+    external: false
+
+services:
+  rabbitmq:
+    image: "rabbitmq:3-management"
+    hostname: "test-rabbitmq"
+    environment:
+      RABBITMQ_ERLANG_COOKIE: "akjahsfvbkueasnvfjkhsga"
+      RABBITMQ_DEFAULT_USER: "rabbitmq"
+      RABBITMQ_DEFAULT_PASS: "rabbitmq"
+      RABBITMQ_DEFAULT_VHOST: "/"
+    networks:
+      - guard
+    ports:
+      - "127.0.0.1:15672:15672"
+      - "127.0.0.1:5672:5672"
+
+  activemq:
+    image: registry.kmlabz.com/birbnetes/activemq-artemis
+    restart: always
+    networks:
+      - guard
+    ports:
+      - "127.0.0.1:1883:1883"
+
+  guard-service:
+    image: registry.kmlabz.com/birbnetes/guard-service
+    restart: always
+    depends_on:
+      - activemq
+      - rabbitmq
+      - input-service
+    environment:
+      SENTRY_DSN: "https://5c925c5a898b4b7d82ee51cd544de623@sentry.kmlabz.com/3"
+      GUARD_RABBITMQ_HOSTNAME: rabbitmq
+      GUARD_RABBITMQ_USERNAME: rabbitmq
+      GUARD_RABBITMQ_PASSWORD: rabbitmq
+      GUARD_RABBITMQ_EXCHANGE: "sample"
+      GUARD_MQTT_HOSTNAME: activemq
+      GUARD_MQTT_PORT: "1883"
+      GUARD_MQTT_USERNAME: artemis
+      GUARD_MQTT_PASSWORD: artemis
+      GUARD_MQTT_TOPIC: "command"
+    networks:
+      - guard
+    ports:
+    - "127.0.0.1:8080:8080"
+
+  input-service:
+    image: mockserver/mockserver
+    command: -logLevel DEBUG -serverPort 80

k8s/configmap.yaml

@@ -0,0 +1,22 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: guard-service
+  labels:
+    app: guard-service
+  namespace: birbnetes
+data:
+  GUARD_CLIENT_ID: guard-b50d97f6-29f9-4de5-a96e-f9b69ca69f7f
+  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_PASSWORD: ZgCiSiSO8t
+  GUARD_RABBITMQ_USERNAME: user
+  GUARD_SERVICE_RELEASEMODE: release
+  SAMPLE_SVC_HOSTNAME: sample-service
+  RELEASE_ID: kmlabz-k8s
+  SENTRY_DSN: https://1d01460ec3094d5bb6c4d78c0a028b08@glitchtip.kmlabz.com/7

k8s/deployment.yaml

@@ -0,0 +1,49 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: guard-service
+  namespace: birbnetes
+  labels:
+    app: guard-service
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: guard-service
+  strategy:
+    type: Recreate
+  template:
+    metadata:
+      labels:
+        app: guard-service
+    spec:
+      containers:
+        - name: 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

requirements.txt

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

src/app.py

@@ -0,0 +1,160 @@
+#!/usr/bin/env python3
+import sys
+import json
+import logging
+
+import sentry_sdk
+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
+
+"""
+Main entry point
+"""
+
+__author__ = "@tormakris"
+__copyright__ = "Copyright 2020, Birbnetes Team"
+__module_name__ = "app"
+__version__text__ = "1"
+
+
+def setup_rabbit(mqtt_: MQTT) -> None:
+    logging.info("Connecting to RabbitMQ...")
+    credentials = pika.PlainCredentials(config.RABBIT_USERNAME, config.RABBIT_PASSWORD)
+    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):
+    """
+    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
+
+
+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

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+import os
+import sys
+import logging
+
+
+"""
+Configuration variables
+"""
+
+
+__author__ = "@tormakris"
+__copyright__ = "Copyright 2020, Birbnetes Team"
+__module_name__ = "config"
+__version__text__ = "1"
+
+CLIENT_ID = os.environ.get("GUARD_CLIENT_ID", "guard-service")
+
+SENTRY_DSN = os.environ.get("SENTRY_DSN")
+RELEASE_ID = os.environ.get("RELEASE_ID", "test")
+RELEASEMODE = os.environ.get("GUARD_SERVICE_RELEASEMODE", "dev")
+
+RABBIT_HOSTNAME = os.getenv("GUARD_RABBITMQ_HOSTNAME", "localhost")
+RABBIT_USERNAME = os.getenv("GUARD_RABBITMQ_USERNAME", "guard-service")
+RABBIT_PASSWORD = os.getenv("GUARD_RABBITMQ_PASSWORD", "guard-service")
+RABBIT_EXCHANGE = os.getenv("GUARD_RABBITMQ_EXCHANGE", "guard-service")
+
+MQTT_HOSTNAME = os.getenv("GUARD_MQTT_HOSTNAME", "localhost")
+MQTT_PORT = os.getenv("GUARD_MQTT_PORT", "1883")
+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")
+
+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

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+import logging
+import paho.mqtt.client as mqtt
+import config
+
+"""
+MQTT class
+"""
+
+__author__ = "@tormakris"
+__copyright__ = "Copyright 2020, Birbnetes Team"
+__module_name__ = "mqtt"
+__version__text__ = "1"
+
+
+class MQTT:
+    """
+    MQTT class used to make sending mqtt messages nice and simple
+    """
+
+    def __init__(self, host=config.MQTT_HOSTNAME, port=config.MQTT_PORT, client_id=config.CLIENT_ID, qos=2,
+                 retain=False, username=config.MQTT_USERNAME, password=config.MQTT_PASSWORD):
+        """
+        Init variables
+        :param host:
+        :param port:
+        :param client_id:
+        :param qos:
+        :param retain:
+        """
+        self.client = None
+        self.host = host
+        self.port = port
+        self.client_id = client_id
+        self._topic = None
+        self.qos = qos
+        self.retain = retain
+        self.username = username
+        self.password = password
+
+    def get_topic(self) -> str:
+        """
+        Set topic
+        :return:
+        """
+        return self._topic
+
+    def set_topic(self, value: str) -> None:
+        """
+        Get current topic
+        :param value:
+        :return:
+        """
+        self._topic = value
+
+    topic = property(get_topic, set_topic)
+
+    def connect(self) -> None:
+        """
+        Setup client and connect to broker
+        :return:
+        """
+        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)
+
+    def publish(self, message: str, subtopic: str = "") -> None:
+        """
+        Publish a message
+        :param message:
+        :param subtopic:
+        :return:
+        """
+        logging.debug(f"MQTT Topic: {self.topic}/{subtopic} Message: {message} QOS: {self.qos} Retain: {self.retain}")
+        if subtopic:
+            self.client.publish(f"{self.topic}/{subtopic}", message, qos=self.qos, retain=self.retain)
+        else:
+            self.client.publish(self.topic, message, qos=self.qos, retain=self.retain)