Updated k8s stuff

also fixed spelling of amqp

.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,14 +24,6 @@ steps:
       - latest
       - ${DRONE_BUILD_NUMBER}
 
-- name: sentry
-  image: tormachris/drone-sentry
-  settings:
-    sentry_project: ${DRONE_REPO_NAME}
-    sentry_domain: sentry.kmlabz.com
-    sentry_token:
-      from_secret: SENTRY_TOKEN
-
 - name: ms-teams
   image: kuperiu/drone-teams
   settings:
@@ -78,8 +31,3 @@ steps:
       from_secret: TEAMS_WEBHOOK
   when:
     status: [ failure ]
-
-volumes:
-- name: cache
-  host:
-    path: "/tmp/cache"

Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.8-slim
+FROM python:3.9-slim
 
 ENV TZ Europe/Budapest
 RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
@@ -16,4 +16,4 @@ COPY ./src .
 
 EXPOSE 8080
 
-ENTRYPOINT ["gunicorn", "-b", "0.0.0.0:8080", "--workers", "1", "--threads", "1", "app:app"]
+ENTRYPOINT ["gunicorn", "-b", "0.0.0.0:8080", "app:app"]

k8s/configmap.yaml

@@ -6,16 +6,17 @@ metadata:
     app: input-service
   namespace: birbnetes
 data:
-  SENTRY_DSN: https://b181edf362e549f4967c6cd42629693d@sentry.kmlabz.com/3
+  SENTRY_DSN: https://fce9e078e1494bf4b959b45d0b435386@sentry.kmlabz.com/2
-  RELEASE_ID: luna-k8s
+  RELEASE_ID: kmlabz-k8s
   INPUT_SERVICE_RELEASEMODE: release
   INPUT_RABBITMQ_HOSTNAME: birb-rabbitmq
-  INPUT_RABBITMQ_EXCHANGE: "wave"
+  INPUT_RABBITMQ_EXCHANGE_META: "sample-meta"
-  INPUT_RABBITMQ_QUEUE: wave-ready
+  INPUT_RABBITMQ_EXCHANGE_CACHE: "sample-cache"
   INPUT_RABBITMQ_USERNAME: user
-  INPUT_RABBITMQ_PASSWORD: 1wZVQnP5vy
+  INPUT_RABBITMQ_PASSWORD: ZgCiSiSO8t
-  INPUT_POSTGRES_HOSTNAME: input-postgres
+  INFLUX_HOST: input-influx
-  INPUT_POSTGRES_USERNAME: input-service
+  INFLUX_PORT: "8086"
-  INPUT_POSTGRES_PASSWORD: input-service-supersecret
+  INFLUX_USERNAME: input-service
-  INPUT_POSTGRES_DB: input-service
+  INFLUX_PASSWORD: input-service-supersecret
-  INPUT_STORAGE_HOSTNAME: storage-service
+  INFLUX_DB: input-service
+  CACHE_REDIS_URL: "redis://input-redis:6379/0"

k8s/deployment.yaml

@@ -19,11 +19,34 @@ spec:
     spec:
       containers:
       - name: input-service
-        image: registry.kmlabz.com/birbnetesgit/input-service
+        image: registry.kmlabz.com/birbnetes/input-service
+        imagePullPolicy: Always
         envFrom:
           - configMapRef:
               name: input-service
         ports:
         - containerPort: 8080
+      - 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
+          - --reporter.type=grpc
       imagePullSecrets:
       - name: regcred

k8s/service.yaml

@@ -13,4 +13,4 @@ spec:
     protocol: TCP
   selector:
     app: input-service
-  type: ClusterIP
+  type: ClusterIP

requirements.txt

@@ -1,14 +1,21 @@
 sentry_sdk[flask]
-gunicorn
+gunicorn~=20.1.0
-Flask
+Flask~=2.0.1
-Flask-RESTful
+Flask-RESTful~=0.3.9
-requests
+requests~=2.26.0
 werkzeug
-sqlalchemy
+xeger~=0.3.5
-flask_sqlalchemy
+pika~=1.2.0
-xeger
+marshmallow~=3.13.0
-pika
-psycopg2-binary
-marshmallow
-marshmallow-sqlalchemy
 flask-marshmallow
+py-healthcheck
+Flask-InfluxDB
+tzdata
+tzlocal
+apscheduler~=3.7.0
+
+flask-redis~=0.4.0
+
+opentracing~=2.4.0
+jaeger-client
+Flask-Opentracing

src/app.py

@@ -1,15 +1,26 @@
 #!/usr/bin/env python3
-import logging
 from flask import Flask
 from flask_restful import Api
 import sentry_sdk
 from sentry_sdk.integrations.flask import FlaskIntegration
+from healthcheck import HealthCheck
 
-from config import *
+from redis_client import redis_client
-from db import db
+from config import Config
 from marshm import ma
-from rabbit_broker_instance import mq
+from influxus import influx_db
-from resources import SampleResource, SampleParameterResource
+from resources import SampleResource
+from healthchecks import amqp_connection_status
+
+import atexit
+
+from apscheduler.schedulers.background import BackgroundScheduler
+from magic_amqp import magic_amqp
+from error_handlers import register_all_error_handlers
+
+import jaeger_client
+import opentracing
+from flask_opentracing import FlaskTracing
 
 """
 Main Flask RESTful API
@@ -20,51 +31,66 @@ __copyright__ = "Copyright 2020, Birbnetes Team"
 __module_name__ = "app"
 __version__text__ = "1"
 
-if SENTRY_DSN:
+if Config.SENTRY_DSN:
     sentry_sdk.init(
-        dsn=SENTRY_DSN,
+        dsn=Config.SENTRY_DSN,
         integrations=[FlaskIntegration()],
-        traces_sample_rate=1.0,
+        traces_sample_rate=0.0,
         send_default_pii=True,
-        release=RELEASE_ID,
+        release=Config.RELEASE_ID,
-        environment=RELEASEMODE,
+        environment=Config.RELEASEMODE,
         _experiments={"auto_enabling_integrations": True}
     )
 
-
 app = Flask(__name__)
-app.config['SQLALCHEMY_DATABASE_URI'] = f"postgresql://{POSTGRES_USERNAME}:{POSTGRES_PASSWORD}@{POSTGRES_HOSTNAME}:5432/{POSTGRES_DB}"
+app.config.from_object(Config)
-app.config['RABBIT_USERNAME'] = RABBITMQ_USERNAME
-app.config['RABBIT_PASSWORD'] = RABBITMQ_PASSWORD
-app.config['RABBIT_HOST'] = RABBITMQ_HOST
-app.config['RABBIT_ROUTING_KEY'] = "feature"
-app.config['EXCHANGE_NAME'] = RABBITMQ_EXCHANGE
 
 api = Api(app)
-db.init_app(app)
+health = HealthCheck()
 ma.init_app(app)
-mq.init_app(app)
 
-with app.app_context():
+redis_client.init_app(app)
-    db.create_all()
 
-formatter = logging.Formatter(
+# ampq magic stuff
-    fmt="%(asctime)s - %(levelname)s - %(module)s - %(message)s"
+magic_amqp.init_app(app)
-)
 
-handler = logging.StreamHandler()
+ampq_loop_scheduler = BackgroundScheduler()
-handler.setFormatter(formatter)
+ampq_loop_scheduler.add_job(func=lambda: magic_amqp.loop(), trigger="interval", seconds=5)
+atexit.register(lambda: ampq_loop_scheduler.shutdown())
 
-logger = logging.getLogger(__name__)
+ampq_loop_scheduler.start()
-logger.setLevel(logging.DEBUG)
-logger.addHandler(handler)
 
-api.add_resource(SampleResource, "/sample")
+if Config.ENABLE_INFLUXDB:
-api.add_resource(SampleParameterResource, '/sample/<tag>')
+    influx_db.init_app(app)
 
-if __name__ == "__main__":
+
-    app.run(
+@app.before_first_request
-        debug=bool(DEBUG),
+def init_db():
-        host="0.0.0.0",
+    if Config.ENABLE_INFLUXDB:
-        port=int(PORT),
+        influx_db.database.create(Config.INFLUXDB_DATABASE)
-    )
+
+
+# Setup tracing
+def initialize_tracer():
+    app.logger.info("Initializing jaeger...")
+    jaeger_cfg = jaeger_client.Config(config={}, service_name='input-service', validate=True)
+    tracer = jaeger_cfg.initialize_tracer()
+    return tracer
+
+
+tracing = FlaskTracing(initialize_tracer, True, app)
+
+api.add_resource(SampleResource, "/input")
+
+health.add_check(amqp_connection_status)
+
+register_all_error_handlers(app)
+
+app.add_url_rule("/healthz", "healthcheck", view_func=lambda: health.run())
+
+if __name__ != '__main__':
+    import logging
+
+    gunicorn_logger = logging.getLogger('gunicorn.error')
+    app.logger.handlers = gunicorn_logger.handlers
+    app.logger.setLevel(gunicorn_logger.level)

src/config.py

@@ -1,35 +1,40 @@
 #!/usr/bin/env python3
 import os
 
-
 """
 Main Flask RESTful API
 """
 
-
 __author__ = "@tormakris"
 __copyright__ = "Copyright 2020, Birbnetes Team"
 __module_name__ = "app"
 __version__text__ = "1"
 
 
-PORT = os.environ.get("INPUT_SERVICE_PORT", 8080)
+class Config:
-DEBUG = os.environ.get("INPUT_SERVICE_DEBUG", True)
+    PORT = 8080
+    DEBUG = os.environ.get("INPUT_SERVICE_DEBUG", "true").lower() in ["true", "yes", "1"]
 
+    SENTRY_DSN = os.environ.get("SENTRY_DSN")
+    RELEASE_ID = os.environ.get("RELEASE_ID", "test")
+    RELEASEMODE = os.environ.get("INPUT_SERVICE_RELEASEMODE", "dev")
 
-SENTRY_DSN = os.environ.get("SENTRY_DSN")
+    REDIS_URL = os.environ['CACHE_REDIS_URL']
-RELEASE_ID = os.environ.get("RELEASE_ID", "test")
-RELEASEMODE = os.environ.get("INPUT_SERVICE_RELEASEMODE", "dev")
 
-RABBITMQ_HOST = os.getenv("INPUT_RABBITMQ_HOSTNAME", "localhost")
+    EXCHANGE_NAME_META = os.getenv("INPUT_RABBITMQ_EXCHANGE_META", "sample-meta")
-RABBITMQ_EXCHANGE = os.getenv("INPUT_RABBITMQ_EXCHANGE", "dev")
+    EXCHANGE_NAME_CACHE = os.getenv("INPUT_RABBITMQ_EXCHANGE_CACHE", "sample-cache")
-RABBITMQ_QUEUE = os.getenv("INPUT_RABBITMQ_QUEUE", "wave-extract")
-RABBITMQ_USERNAME = os.getenv("INPUT_RABBITMQ_USERNAME", "rabbitmq")
-RABBITMQ_PASSWORD = os.getenv("INPUT_RABBITMQ_PASSWORD", "rabbitmq")
 
-POSTGRES_HOSTNAME = os.getenv("INPUT_POSTGRES_HOSTNAME", "localhost")
+    FLASK_PIKA_PARAMS = {
-POSTGRES_USERNAME = os.getenv("INPUT_POSTGRES_USERNAME", "input-service")
+        'host': os.getenv("INPUT_RABBITMQ_HOSTNAME", "localhost"),
-POSTGRES_PASSWORD = os.getenv("INPUT_POSTGRES_PASSWORD", "input-service")
+        'username': os.getenv("INPUT_RABBITMQ_USERNAME", "rabbitmq"),
-POSTGRES_DB = os.getenv("INPUT_POSTGRES_DB", "input-service")
+        'password': os.getenv("INPUT_RABBITMQ_PASSWORD", "rabbitmq"),
+        'port': int(os.getenv("INPUT_RABBITMQ_PORT", 5672)),
+        'virtual_host': '/'
+    }
 
-STORAGE_HOSTNAME = os.getenv("INPUT_STORAGE_HOSTNAME", "localhost:8042")
+    ENABLE_INFLUXDB = os.environ.get("INPUT_ENABLE_INFLUX", "true").lower() in ["true", "yes", "1"]
+    INFLUXDB_HOST = os.getenv("INFLUX_HOST", "input-influx")
+    INFLUXDB_PORT = os.getenv("INFLUX_PORT", "8086")
+    INFLUXDB_USER = os.getenv("INFLUX_USERNAME", "input-service")
+    INFLUXDB_PASSWORD = os.getenv("INFLUX_PASSWORD", "input-service-supersecret")
+    INFLUXDB_DATABASE = os.getenv("INFLUX_DB", "input-service")

src/error_handlers.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python3
+
+
+def get_standard_error_handler(code: int):
+    def error_handler(err):
+        return {"msg": str(err)}, code
+
+    return error_handler
+
+
+# function to register all handlers
+
+
+def register_all_error_handlers(app):
+    error_codes_to_override = [404, 403, 401, 405, 400, 409, 422, 500]
+
+    for code in error_codes_to_override:
+        app.register_error_handler(code, get_standard_error_handler(code))

src/flask_rabbit_broker.py

@@ -1,102 +0,0 @@
-#!/usr/bin/env python3
-import uuid
-import pika
-
-"""
-Flask Rabbit Broker
-"""
-
-__author__ = '@tormakris'
-__copyright__ = "Copyright 2020, Birbnetes Team"
-__module_name__ = "flask_rabbit_broker"
-__version__text__ = "1"
-
-
-class FlaskRabbitBroker:
-    """Message Broker using RabbitMQ middleware"""
-
-    def __init__(self, app=None):
-        """
-        Create a new instance of Broker Rabbit by using
-        the given parameters to connect to RabbitMQ.
-        """
-        self.app = app
-        self.exchange_name = None
-        self.username = None
-        self.password = None
-        self.rabbitmq_host = None
-        self.routing_key = None
-        self.connection = None
-        self.channel = None
-        self.exchange = None
-        self.exchange_type = "fanout"
-
-    def init_app(self, app) -> None:
-        """
-        Init the broker with the current application context
-        :param app: application context
-        :return:
-        """
-        self.username = app.config.get('RABBIT_USERNAME')
-        self.password = app.config.get('RABBIT_PASSWORD')
-        self.rabbitmq_host = app.config.get('RABBIT_HOST')
-        self.exchange_name = app.config.get('EXCHANGE_NAME')
-        self.routing_key = app.config.get('RABBIT_ROUTING_KEY')
-        self.init_connection(timeout=5)
-        self.init_exchange()
-
-    def init_connection(self, timeout: int = 5) -> None:
-        """"
-        Init RabbitMQ connection
-        :param timeout: timeout of connection
-        :return:
-        """
-        credentials = pika.PlainCredentials(self.username, self.password)
-        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host=self.rabbitmq_host,
-                                                                            credentials=credentials,
-                                                                            heartbeat=0,
-                                                                            socket_timeout=timeout))
-
-    def close_connection(self) -> None:
-        self.connection.close()
-
-    def init_exchange(self) -> None:
-        """
-        Init the exchange use to send messages
-        :return:
-        """
-        channel = self.connection.channel()
-        try:
-            channel.exchange_declare(exchange=self.exchange_name,
-                                     exchange_type=self.exchange_type,
-                                     durable=True,
-                                     auto_delete=False)
-        finally:
-            channel.close()
-
-    def register_callback(self, callback) -> None:
-        """
-        Register a callback.
-        :param callback:
-        :return:
-        """
-        channel = self.connection.channel()
-        queue = channel.queue_declare(durable=True, auto_delete=False, exclusive=True,
-                                      queue=uuid.uuid4().urn.split(':')[2]).method.queue
-        channel.bind(exchange=self.exchange_name, queue=queue)
-        channel.basic_consume(queue=queue, on_message_callback=callback, auto_ack=True)
-
-    def send(self, message: str) -> None:
-        """
-        Sends a message to the declared exchange.
-        :param message:
-        :return:
-        """
-        channel = self.connection.channel()
-        try:
-            channel.basic_publish(
-                exchange=self.exchange_name,
-                routing_key=self.routing_key,
-                body=message.encode('utf-8'))
-        finally:
-            channel.close()

src/healthchecks.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python3
+from magic_amqp import magic_amqp
+
+"""
+Healthchek functions
+"""
+
+__author__ = "@tormakris"
+__copyright__ = "Copyright 2020, Birbnetes Team"
+__module_name__ = "healthchecks"
+__version__text__ = "1"
+
+
+def amqp_connection_status():
+    if magic_amqp.is_healthy():
+        result = True
+        text = "amqp connection is ok"
+    else:
+        result = False
+        text = "amqp connection is unhealthy"
+
+    return result, text

src/influxus.py

@@ -1,13 +1,15 @@
 #!/usr/bin/env python3
-from flask_sqlalchemy import SQLAlchemy
+
+from flask_influxdb import InfluxDB
+
 
 """
-Flask Restful endpoints
+Influx api
 """
 
 __author__ = '@tormakris'
 __copyright__ = "Copyright 2020, Birbnetes Team"
-__module_name__ = "db"
+__module_name__ = "influxus"
 __version__text__ = "1"
 
-db = SQLAlchemy()
+influx_db = InfluxDB()

src/magic_amqp.py

@@ -0,0 +1,142 @@
+from flask import Flask
+from threading import Lock
+import pika
+import pika.exceptions
+import json
+import time
+
+import opentracing
+from opentracing.ext import tags
+from opentracing.propagation import Format
+
+
+class MagicAMQP:
+    """
+    This is my pathetic attempt to make RabbitMQ connection in a Flask app reliable and performant.
+    """
+
+    def __init__(self, app: Flask = None):
+        self.app = app
+        if app:
+            self.init_app(app)
+
+        self._lock = Lock()
+        self._credentials = None
+
+    def init_app(self, app: Flask):
+        self.app = app
+        self.app.config.setdefault('FLASK_PIKA_PARAMS', {})
+        self.app.config.setdefault('EXCHANGE_NAME_META', None)
+        self.app.config.setdefault('EXCHANGE_NAME_CACHE', None)
+
+        self._credentials = pika.PlainCredentials(
+            app.config['FLASK_PIKA_PARAMS']['username'],
+            app.config['FLASK_PIKA_PARAMS']['password']
+        )
+
+        self._reconnect_ampq()
+
+    def _reconnect_ampq(self):
+        self._pika_connection = pika.BlockingConnection(
+            pika.ConnectionParameters(
+                host=self.app.config['FLASK_PIKA_PARAMS']['host'],
+                credentials=self._credentials,
+                heartbeat=10,
+                socket_timeout=5)
+        )
+        self._pika_channel = self._pika_connection.channel()
+        self._pika_channel.exchange_declare(
+            exchange=self.app.config['EXCHANGE_NAME_META'],
+            exchange_type='direct'
+        )
+        self._pika_channel.exchange_declare(
+            exchange=self.app.config['EXCHANGE_NAME_CACHE'],
+            exchange_type='direct'
+        )
+
+    def loop(self):
+        """
+        This method should be called periodically to keep up the connection
+        """
+        lock_start = time.time()
+        with self._lock:
+            lock_acquire_time = time.time() - lock_start
+            if lock_acquire_time >= 0.5:
+                self.app.logger.warning(f"Loop: Lock acquire took {lock_acquire_time:5f} sec")
+
+            try:
+                self._pika_connection.process_data_events(0)
+                # We won't attempt retry if this fail
+            except pika.exceptions.AMQPConnectionError as e:
+                self.app.logger.warning(f"Connection error during process loop: {e} (attempting reconnect)")
+                self._reconnect_ampq()
+
+            total_time = time.time() - lock_start
+            if total_time > 1:
+                self.app.logger.warning(f"Loop: Total loop took {total_time:5f} sec")
+
+    def _publish(self, exchange: str, payload=None):
+        """
+        Publish a simple json serialized message to the configured queue.
+        If the connection is broken, then this call will block until the connection is restored
+        """
+        span_tags = {tags.SPAN_KIND: tags.SPAN_KIND_PRODUCER}
+        with opentracing.tracer.start_active_span('magic_amqp.publish', tags=span_tags) as scope:
+            opentracing.tracer.inject(scope.span.context, Format.TEXT_MAP, payload)
+            lock_start = time.time()
+            with self._lock:
+                scope.span.log_kv({'event': 'lockAcquired'})
+                lock_acquire_time = time.time() - lock_start
+                if lock_acquire_time >= 0.2:
+                    self.app.logger.warning(f"Publish: Lock acquire took {lock_acquire_time:5f} sec")
+                tries = 0
+                while True:
+                    try:
+                        self._pika_channel.basic_publish(
+                            exchange=exchange,
+                            routing_key='feature',
+                            body=json.dumps(payload).encode('UTF-8')
+                        )
+                        self.app.logger.debug(f"Published: {payload}")
+                        break  # message sent successfully
+                    except pika.exceptions.AMQPConnectionError as e:
+                        scope.span.log_kv({'event': 'connectionError', 'error': str(e)})
+                        self.app.logger.warning(f"Connection error during publish: {e} (attempting reconnect)")
+
+                        if tries > 30:
+                            raise  # just give up
+
+                        while True:
+                            try:
+                                self._reconnect_ampq()
+                                break
+                            except pika.exceptions.AMQPConnectionError as e:
+                                self.app.logger.warning(
+                                    f"Connection error during reconnection: {e} (attempting reconnect)")
+                                tries += 1
+
+                                if tries > 30:
+                                    raise  # just give up
+
+                            if tries > 10:
+                                time.sleep(2)
+                total_time = time.time() - lock_start
+                if total_time > 0.4:
+                    self.app.logger.warning(f"Publish: Total publish took {total_time:5f} sec")
+
+    def publish_cache(self, payload=None):
+        return self._publish(self.app.config['EXCHANGE_NAME_CACHE'], payload)
+
+    def publish_meta(self, payload=None):
+        return self._publish(self.app.config['EXCHANGE_NAME_META'], payload)
+
+    def is_healthy(self) -> bool:
+        with self._lock:
+            if not self._pika_channel:
+                return False
+
+            return self._pika_channel.is_open and self._pika_connection.is_open
+
+
+# instance to be used in the flask app
+magic_amqp = MagicAMQP()

src/models.py

@@ -1,25 +0,0 @@
-#!/usr/bin/env python3
-from db import db
-from sqlalchemy.sql import func
-
-"""
-Flask Restful endpoints
-"""
-
-__author__ = '@tormakris'
-__copyright__ = "Copyright 2020, Birbnetes Team"
-__module_name__ = "models"
-__version__text__ = "1"
-
-
-class SampleMetadata(db.Model):
-    """
-    SQLAlchemy model of metadata entries
-    """
-    id = db.Column(db.Integer, primary_key=True, auto_increment=True)
-    timestamp = db.Column(db.TIMESTAMP, nullable=False, server_default=func.now())
-
-    device_id = db.Column(db.Integer, nullable=False)
-    device_date = db.Column(db.DateTime, nullable=False)
-
-    tag = db.Column(db.String(32), nullable=False, unique=True)

src/rabbit_broker_instance.py

@@ -1,14 +0,0 @@
-#!/usr/bin/env python3
-
-from flask_rabbit_broker import FlaskRabbitBroker
-
-"""
-Rabbit Broker instance
-"""
-
-__author__ = '@tormakris'
-__copyright__ = "Copyright 2020, Birbnetes Team"
-__module_name__ = "rabbit_broker_instance"
-__version__text__ = "1"
-
-mq = FlaskRabbitBroker()

src/rabbitmqqueue.py

@@ -1,18 +0,0 @@
-#!/usr/bin/env python3
-
-import pika
-from config import *
-
-"""
-Rabbitmq setup
-"""
-
-__author__ = '@tormakris'
-__copyright__ = "Copyright 2020, Birbnetes Team"
-__module_name__ = "endpoints"
-__version__text__ = "1"
-
-credentials = pika.PlainCredentials(RABBITMQ_USERNAME, RABBITMQ_PASSWORD)
-rabbitmq = pika.BlockingConnection(pika.ConnectionParameters(host=RABBITMQ_HOST, credentials=credentials))
-rabbitmq_channel = rabbitmq.channel()
-rabbitmq_channel.exchange_declare(exchange=RABBITMQ_EXCHANGE, exchange_type='fanout')

src/redis_client.py

@@ -0,0 +1,4 @@
+#!/usr/bin/env python3
+from flask_redis import FlaskRedis
+
+redis_client = FlaskRedis()

src/resources.py

@@ -1,16 +1,15 @@
 #!/usr/bin/env python3
-import logging
+import io
-import json
+from datetime import datetime
+import tzlocal
 from xeger import Xeger
 from flask_restful import Resource
-from flask import request
+from flask import request, current_app, abort
-import requests
+from magic_amqp import magic_amqp
-import pika
+from influxus import influx_db
-from db import db
+from schemas import SampleSchema
-from models import SampleMetadata
+from redis_client import redis_client
-from schemas import SampleSchema, SampleMetadataSchema
+import opentracing
-from config import *
-from rabbit_broker_instance import mq
 
 """
 Flask Restful endpoints
@@ -21,8 +20,6 @@ __copyright__ = "Copyright 2020, Birbnetes Team"
 __module_name__ = "endpoints"
 __version__text__ = "1"
 
-LOGGER = logging.getLogger(__name__)
-
 
 class SampleResource(Resource):
     """
@@ -31,83 +28,87 @@ class SampleResource(Resource):
     """
 
     sampleschema = SampleSchema(many=False)
-    samplemetadataschema = SampleMetadataSchema(many=True)
 
     def post(self):
         """
         Post request send to the endpoint
         :return:
         """
-        if 'file' not in request.files:
+        with opentracing.tracer.start_active_span('parseAndValidate'):
-            return {"err_msg": "no file found"}, 469
+            if 'file' not in request.files:
-        else:
+                return abort(400, "no file found")
-            soundfile = request.files['file']
+            else:
+                soundfile = request.files['file']
 
-        if 'description' not in request.form:
+            if 'description' not in request.form:
-            return {"err_msg": "no description found"}, 470
+                return abort(400, "no description found")
-        else:
+            else:
-            description = request.form.get("description")
+                description_raw = request.form.get("description")
 
-        if soundfile.content_type != 'audio/wave':
+            if soundfile.content_type != 'audio/wave':
-            LOGGER.info(
+                current_app.logger.info(f"Input file was not WAV.")
-                f"Input file was not WAV.")
+                return abort(415, 'Input file not a wave file.')
-            return {'err_msg': 'Input file not a wave file.'}, 415
+            try:
+                desc = self.sampleschema.loads(description_raw)
+            except Exception as e:
+                current_app.logger.exception(e)
+                return abort(417, 'Input JSON schema invalid')
 
-        try:
+        with opentracing.tracer.start_active_span('generateTag'):
-            desc = self.sampleschema.loads(description)
+            xeger = Xeger(limit=30)
-        except Exception as e:
+            while True:
-            LOGGER.exception(e)
+                generated_tag = xeger.xeger(r'^[a-zA-Z]+[0-9a-zA-Z_]*$')[:32]
-            return {'err_msg': 'Input JSON schema invalid'}, 417
+                if len(generated_tag) > 2:  # Ensure minimum length
+                    break
 
-        xeger = Xeger(limit=30)
+        with opentracing.tracer.start_active_span('publishMetaMessage'):
-        generated_tag = xeger.xeger(r'^[a-zA-Z]+[0-9a-zA-Z_]*$')[:32]
+            try:
+                magic_amqp.publish_meta(
+                    {
+                        'tag': generated_tag,
+                        'timestamp': datetime.now().isoformat(),
+                        'device_id': desc['device_id'],
+                        'device_date': desc['date'].isoformat()
+                    }
+                )
+            except Exception as e:
+                current_app.logger.exception(e)
+                return abort(500, f"AMQP Publish error: {str(e)}")
 
-        record = SampleMetadata(
+        with opentracing.tracer.start_active_span('readSampleToMemory'):
-            device_id=desc['device_id'],
+            buf = io.BytesIO()
-            device_date=desc['date'],
+            soundfile.save(buf)
-            tag=generated_tag)
+
-        try:
+        with opentracing.tracer.start_active_span('putToCache'):
-            db.session.add(record)
+            redis_client.set(generated_tag, buf.getbuffer())  # getbuffer is quick as it does not copy like getvalue
-            requests.post(
+
-                f"http://{STORAGE_HOSTNAME}/object",
+        # Announce only after the data is successfully committed
-                files={
+        with opentracing.tracer.start_active_span('publishInCacheMessage'):
-                    'description': (None, json.dumps({'tag': generated_tag}), 'application/json'),
+            try:
-                    'soundFile': (
+                magic_amqp.publish_cache({
-                        'wave.wav',
+                    'tag': generated_tag,
-                        soundfile,
+                    'mime_type': soundfile.mimetype
-                        soundfile.content_type,
+                })
-                        {'Content-Length': soundfile.content_length})}).raise_for_status()
+            except Exception as e:
-            mq.send(json.dumps({'tag': generated_tag}))
+                current_app.logger.exception(e)
-        except Exception as e:
+                return abort(500, f"AMQP Publish error: {str(e)}")
-            LOGGER.exception(e)
+
-            db.session.rollback()
+        # metrics
-            return {"err_msg": str(
+        if current_app.config['ENABLE_INFLUXDB']:
-                e), "hint": "DB or downstream service error"}, 569
+            with opentracing.tracer.start_active_span('influxdb.write_points'):
+                influx_db.write_points(
+                    [
+                        {
+                            'time': datetime.now(tz=tzlocal.get_localzone()),
+                            'measurement': 'cloudinput',
+                            'tags': {
+                                'device': desc['device_id']
+                            },
+                            'fields': {
+                                'bruh': 1.0
+                            }
+                        }
+                    ]
+                )
 
-        db.session.commit()
         return {"tag": generated_tag}, 200
-
-    def get(self):
-        """
-        Get all stored items
-        :return:
-        """
-        samples = SampleMetadata.query.all()
-        return self.samplemetadataschema.dump(list(samples)), 200
-
-
-class SampleParameterResource(Resource):
-    """
-    Sample endpoint with parameters
-    """
-
-    samplemetadataschema = SampleMetadataSchema(many=False)
-
-    def get(self, tag: str):
-        """
-        Get a specific item
-        :param tag:
-        :return:
-        """
-        sample = SampleMetadata.query.filter_by(tag=tag).first_or_404()
-        return self.samplemetadataschema.dump(sample), 200

src/schemas.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-from models import SampleMetadata
 from marshm import ma
 from marshmallow import fields
 
@@ -19,17 +18,8 @@ class SampleSchema(ma.Schema):
     """
     Parameters:
      - date (date)
-     - device_id (str)
+     - device_id (int)
     """
 
     date = fields.DateTime(required=True)
-    device_id = fields.Str(required=True)
+    device_id = fields.Integer(required=True)
-
-
-class SampleMetadataSchema(ma.SQLAlchemyAutoSchema):
-    """
-    Marshmallow schema generated
-    """
-    class Meta:
-        model = SampleMetadata
-