README.md

@@ -4,7 +4,7 @@ Playing with IoT
 
 [![Build Status](https://travis-ci.com/daniviga/bite.svg?branch=master)](https://travis-ci.com/daniviga/bite)
 ![AGPLv3](./docs/.badges/agpl3.svg)
-![Python 3.11](./docs/.badges/python.svg)
+![Python 3.9](./docs/.badges/python.svg)
 ![MQTT](./docs/.badges/mqtt.svg)
 ![Moby](./docs/.badges/moby.svg)
 ![docker-compose 3.7+](./docs/.badges/docker-compose.svg)
@@ -13,6 +13,13 @@ This project is for educational purposes only. It does not implement any
 authentication and/or encryption protocol, so it is not suitable for real
 production.
 
+![Application Schema](./docs/application_chart.png)
+
+### Future implementations
+
+- Broker HA via [VerneMQ clustering](https://docs.vernemq.com/clustering/introduction)
+- Stream analytics via [Apache Spark](https://spark.apache.org/)
+
 ## Installation
 
 ### Requirements
@@ -29,10 +36,8 @@ The application stack is composed by the following components:
 - [Django](https://www.djangoproject.com/) with
 [Django REST framework](https://www.django-rest-framework.org/)
 web application (running via `gunicorn` in production mode)
-  - `dispatcher` custom daemon to dump telemetry into the Kafka queue
-  - `handler` custom daemon to dump telemetry into the timeseries database from the Kafka queue
+  - `mqtt-to-db` custom daemon to dump telemetry into the timeseries database
   - telemetry payload is stored as json object (via PostgreSQL JSON data type)
-- [Kafka](https://kafka.apache.org/) broker
 - [Timescale](https://www.timescale.com/) DB,
 a [PostgreSQL](https://www.postgresql.org/) database with a timeseries extension
 - [Mosquitto](https://mosquitto.org/) MQTT broker (see alternatives below)

bite/bite/settings.py

@@ -151,10 +151,6 @@ STATIC_URL = '/static/'
 
 STATIC_ROOT = '/srv/appdata/bite/static'
 
-REST_FRAMEWORK = {
-    'DEFAULT_AUTHENTICATION_CLASSES': []
-}
-
 SKIP_WHITELIST = True
 
 MQTT_BROKER = {
@@ -162,11 +158,6 @@ MQTT_BROKER = {
     'PORT': '1883',
 }
 
-KAFKA_BROKER = {
-    'HOST': 'kafka',
-    'PORT': '9092',
-}
-
 # If no local_settings.py is availble in the current folder let's try to
 # load it from the application root
 try:

bite/telemetry/management/commands/handler.py

@@ -1,76 +0,0 @@
-# -*- coding: utf-8 -*-
-# vim: tabstop=4 shiftwidth=4 softtabstop=4
-#
-# BITE - A Basic/IoT/Example
-# Copyright (C) 2020-2021 Daniele Viganò <daniele@vigano.me>
-#
-# BITE is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-#
-# BITE is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU Affero General Public License for more details.
-#
-# You should have received a copy of the GNU Affero General Public License
-# along with this program.  If not, see <http://www.gnu.org/licenses/>.
-
-import json
-import time
-from kafka import KafkaConsumer
-from kafka.errors import NoBrokersAvailable
-
-from django.conf import settings
-from django.core.management.base import BaseCommand
-from django.core.exceptions import ObjectDoesNotExist
-
-from api.models import Device
-from telemetry.models import Telemetry
-
-
-class Command(BaseCommand):
-    help = 'MQTT to DB deamon'
-
-    KAFKA_HOST = settings.KAFKA_BROKER['HOST']
-    KAFKA_PORT = int(settings.KAFKA_BROKER['PORT'])
-
-    def get_device(self, serial):
-        try:
-            return Device.objects.get(serial=serial)
-        except ObjectDoesNotExist:
-            return None
-
-    def store_telemetry(self, transport, message):
-        Telemetry.objects.create(
-            transport=transport,
-            device=self.get_device(message["device"]),
-            clock=message["clock"],
-            payload=message["payload"]
-        )
-
-    def handle(self, *args, **options):
-        while True:
-            try:
-                consumer = KafkaConsumer(
-                    "telemetry",
-                    bootstrap_servers='{}:{}'.format(
-                        self.KAFKA_HOST, self.KAFKA_PORT
-                    ),
-                    group_id="handler",
-                    value_deserializer=lambda m: json.loads(m.decode('utf8')),
-                )
-                break
-            except NoBrokersAvailable:
-                self.stdout.write(
-                    self.style.WARNING('WARNING: Kafka broker not available'))
-                time.sleep(5)
-
-        self.stdout.write(self.style.SUCCESS('INFO: Kafka broker subscribed'))
-        for message in consumer:
-            self.store_telemetry(
-                message.value["transport"],
-                message.value["body"]
-            )
-        consumer.unsuscribe()

bite/telemetry/management/commands/mqtt-to-db.py

@@ -22,27 +22,23 @@ import asyncio
 import json
 import time
 import paho.mqtt.client as mqtt
-from kafka import KafkaProducer
-from kafka.errors import NoBrokersAvailable
 from asgiref.sync import sync_to_async
-from aiomqtt import Client
+from asyncio_mqtt import Client
 
 from django.conf import settings
 from django.core.management.base import BaseCommand
 from django.core.exceptions import ObjectDoesNotExist
 
 from api.models import Device
+from telemetry.models import Telemetry
+
+MQTT_HOST = settings.MQTT_BROKER['HOST']
+MQTT_PORT = int(settings.MQTT_BROKER['PORT'])
 
 
 class Command(BaseCommand):
     help = 'MQTT to DB deamon'
 
-    MQTT_HOST = settings.MQTT_BROKER['HOST']
-    MQTT_PORT = int(settings.MQTT_BROKER['PORT'])
-    KAFKA_HOST = settings.KAFKA_BROKER['HOST']
-    KAFKA_PORT = int(settings.KAFKA_BROKER['PORT'])
-    producer = None
-
     @sync_to_async
     def get_device(self, serial):
         try:
@@ -51,23 +47,24 @@ class Command(BaseCommand):
             return None
 
     @sync_to_async
-    def dispatch(self, message):
-        self.producer.send(
-            'telemetry', {"transport": 'mqtt',
-                          "body": message}
+    def store_telemetry(self, device, payload):
+        Telemetry.objects.create(
+            device=device,
+            transport='mqtt',
+            clock=payload['clock'],
+            payload=payload['payload']
         )
 
     async def mqtt_broker(self):
-        async with Client(self.MQTT_HOST, port=self.MQTT_PORT) as client:
+        async with Client(MQTT_HOST, port=MQTT_PORT) as client:
             # use shared subscription for HA/balancing
             await client.subscribe("$share/telemetry/#")
-            async with client.messages() as messages:
+            async with client.unfiltered_messages() as messages:
                 async for message in messages:
+                    payload = json.loads(message.payload.decode('utf-8'))
                     device = await self.get_device(message.topic)
                     if device is not None:
-                        message_body = json.loads(
-                            message.payload.decode('utf-8'))
-                        await self.dispatch(message_body)
+                        await self.store_telemetry(device, payload)
                     else:
                         self.stdout.write(
                             self.style.ERROR(
@@ -77,28 +74,13 @@ class Command(BaseCommand):
         client = mqtt.Client()
         while True:
             try:
-                client.connect(self.MQTT_HOST, self.MQTT_PORT)
+                client.connect(MQTT_HOST, MQTT_PORT)
                 break
             except (socket.gaierror, ConnectionRefusedError):
                 self.stdout.write(
-                    self.style.WARNING('WARNING: MQTT broker not available'))
+                    self.style.WARNING('WARNING: Broker not available'))
                 time.sleep(5)
 
-        while True:
-            try:
-                self.producer = KafkaProducer(
-                    bootstrap_servers='{}:{}'.format(
-                        self.KAFKA_HOST, self.KAFKA_PORT
-                    ),
-                    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
-                    retries=5
-                )
-                break
-            except NoBrokersAvailable:
-                self.stdout.write(
-                    self.style.WARNING('WARNING: Kafka broker not available'))
-                time.sleep(5)
-
-        self.stdout.write(self.style.SUCCESS('INFO: Brokers subscribed'))
+        self.stdout.write(self.style.SUCCESS('INFO: Broker subscribed'))
         client.disconnect()
         asyncio.run(self.mqtt_broker())

docker/django/Dockerfile

@@ -17,20 +17,20 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-FROM python:3.11-alpine AS builder
+FROM python:3.9-alpine AS builder
 RUN apk update && apk add gcc musl-dev postgresql-dev \
         && pip install psycopg2-binary
 
 # ---
 
-FROM python:3.11-alpine
+FROM python:3.9-alpine
 ENV PYTHONUNBUFFERED 1
 ENV DJANGO_SETTINGS_MODULE "bite.settings"
 
 RUN apk update && apk add --no-cache postgresql-libs \
-        && wget https://github.com/jwilder/dockerize/releases/download/v0.7.0/dockerize-alpine-linux-amd64-v0.7.0.tar.gz -qO- \
+        && wget https://github.com/jwilder/dockerize/releases/download/v0.6.1/dockerize-alpine-linux-amd64-v0.6.1.tar.gz -qO- \
         | tar -xz -C /usr/local/bin
-COPY --from=builder /usr/local/lib/python3.11/site-packages/ /usr/local/lib/python3.11/site-packages/
+COPY --from=builder /usr/local/lib/python3.9/site-packages/ /usr/local/lib/python3.9/site-packages/
 COPY --chown=1000:1000 requirements.txt /srv/app/bite/requirements.txt
 RUN pip3 install --no-cache-dir -r /srv/app/bite/requirements.txt
 

docker/docker-compose.dev.yml

@@ -36,10 +36,6 @@ services:
     ports:
         - "${CUSTOM_DOCKER_IP:-0.0.0.0}:8000:8000"
 
-  kafka:
-    ports:
-        - "${CUSTOM_DOCKER_IP:-0.0.0.0}:9092:9092"
-
   data-migration:
     volumes:
       - ../bite:/srv/app/bite
@@ -48,10 +44,6 @@ services:
     volumes:
       - ../bite:/srv/app/bite
 
-  dispatcher:
-    volumes:
-      - ../bite:/srv/app/bite
-
-  handler:
+  mqtt-to-db:
     volumes:
       - ../bite:/srv/app/bite

docker/docker-compose.prod.yml

@@ -29,10 +29,6 @@ services:
     volumes:
       - ./django/production.py.sample:/srv/app/bite/bite/production.py
 
-  dispatcher:
-    volumes:
-      - ./django/production.py.sample:/srv/app/bite/bite/production.py
-
-  handler:
+  mqtt-to-db:
     volumes:
       - ./django/production.py.sample:/srv/app/bite/bite/production.py

docker/docker-compose.yml

@@ -62,30 +62,6 @@ services:
     ports:
       - "${CUSTOM_DOCKER_IP:-0.0.0.0}:1883:1883"
 
-  zookeeper:
-    image: confluentinc/cp-zookeeper:latest
-    networks:
-      - net
-    environment:
-      ZOOKEEPER_CLIENT_PORT: 2181
-      ZOOKEEPER_TICK_TIME: 2000
-    ports:
-      - 22181:2181
-
-  kafka:
-    image: confluentinc/cp-kafka:latest
-    depends_on:
-      - zookeeper
-    networks:
-      - net
-    environment:
-      KAFKA_BROKER_ID: 1
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
-      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-
   ingress:
     <<: *service_default
     image: nginx:stable-alpine
@@ -128,21 +104,13 @@ services:
       - staticdata:/srv/appdata/bite/static
     command: ["python3", "manage.py", "collectstatic", "--noinput"]
 
-  dispatcher:
+  mqtt-to-db:
     <<: *service_default
     image: daniviga/bite
-    command: ["python3", "manage.py", "dispatcher"]
-    networks:
-      - net
-    depends_on:
-      - broker
-
-  handler:
-    <<: *service_default
-    image: daniviga/bite
-    command: ["python3", "manage.py", "handler"]
+    command: ["python3", "manage.py", "mqtt-to-db"]
     networks:
       - net
     depends_on:
       - data-migration
       - timescale
+      - broker

docker/ntpd/Dockerfile

@@ -17,9 +17,9 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-FROM alpine:3.18
+FROM alpine:3.15
 
-RUN apk add --no-cache chrony && \
+RUN apk update && apk add chrony && \
     chown -R chrony:chrony /var/lib/chrony
 COPY ./chrony.conf /etc/chrony/chrony.conf
 

docker/simulator/Dockerfile

@@ -17,7 +17,7 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-FROM python:3.11-alpine
+FROM python:3.9-alpine
 
 RUN pip3 install urllib3 paho-mqtt
 COPY ./device_simulator.py /opt/bite/device_simulator.py

docker/simulator/device_simulator.py

@@ -94,7 +94,7 @@ def main():
     parser.add_argument('-s', '--serial',
                         default=os.environ.get('IOT_SERIAL'),
                         help='IoT device serial number')
-    parser.add_argument('-d', '--delay', metavar='s', type=float,
+    parser.add_argument('-d', '--delay', metavar='s', type=int,
                         default=os.environ.get('IOT_DELAY', 10),
                         help='Delay between requests')
     args = parser.parse_args()

docs/.badges/python.svg

@@ -1 +1 @@
-<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="93.0" height="20"><linearGradient id="smooth" x2="0" y2="100%"><stop offset="0" stop-color="#bbb" stop-opacity=".1"/><stop offset="1" stop-opacity=".1"/></linearGradient><clipPath id="round"><rect width="93.0" height="20" rx="3" fill="#fff"/></clipPath><g clip-path="url(#round)"><rect width="65.5" height="20" fill="#555"/><rect x="65.5" width="27.5" height="20" fill="#007ec6"/><rect width="93.0" height="20" fill="url(#smooth)"/></g><g fill="#fff" text-anchor="middle" font-family="DejaVu Sans,Verdana,Geneva,sans-serif" font-size="110"><image x="5" y="3" width="14" height="14" xlink:href="data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCAxMDAgMTAwIj4KICA8ZGVmcz4KICAgIDxsaW5lYXJHcmFkaWVudCBpZD0icHlZZWxsb3ciIGdyYWRpZW50VHJhbnNmb3JtPSJyb3RhdGUoNDUpIj4KICAgICAgPHN0b3Agc3RvcC1jb2xvcj0iI2ZlNSIgb2Zmc2V0PSIwLjYiLz4KICAgICAgPHN0b3Agc3RvcC1jb2xvcj0iI2RhMSIgb2Zmc2V0PSIxIi8+CiAgICA8L2xpbmVhckdyYWRpZW50PgogICAgPGxpbmVhckdyYWRpZW50IGlkPSJweUJsdWUiIGdyYWRpZW50VHJhbnNmb3JtPSJyb3RhdGUoNDUpIj4KICAgICAgPHN0b3Agc3RvcC1jb2xvcj0iIzY5ZiIgb2Zmc2V0PSIwLjQiLz4KICAgICAgPHN0b3Agc3RvcC1jb2xvcj0iIzQ2OCIgb2Zmc2V0PSIxIi8+CiAgICA8L2xpbmVhckdyYWRpZW50PgogIDwvZGVmcz4KCiAgPHBhdGggZD0iTTI3LDE2YzAtNyw5LTEzLDI0LTEzYzE1LDAsMjMsNiwyMywxM2wwLDIyYzAsNy01LDEyLTExLDEybC0yNCwwYy04LDAtMTQsNi0xNCwxNWwwLDEwbC05LDBjLTgsMC0xMy05LTEzLTI0YzAtMTQsNS0yMywxMy0yM2wzNSwwbDAtM2wtMjQsMGwwLTlsMCwweiBNODgsNTB2MSIgZmlsbD0idXJsKCNweUJsdWUpIi8+CiAgPHBhdGggZD0iTTc0LDg3YzAsNy04LDEzLTIzLDEzYy0xNSwwLTI0LTYtMjQtMTNsMC0yMmMwLTcsNi0xMiwxMi0xMmwyNCwwYzgsMCwxNC03LDE0LTE1bDAtMTBsOSwwYzcsMCwxMyw5LDEzLDIzYzAsMTUtNiwyNC0xMywyNGwtMzUsMGwwLDNsMjMsMGwwLDlsMCwweiBNMTQwLDUwdjEiIGZpbGw9InVybCgjcHlZZWxsb3cpIi8+CgogIDxjaXJjbGUgcj0iNCIgY3g9IjY0IiBjeT0iODgiIGZpbGw9IiNGRkYiLz4KICA8Y2lyY2xlIHI9IjQiIGN4PSIzNyIgY3k9IjE1IiBmaWxsPSIjRkZGIi8+Cjwvc3ZnPgo="/><text x="422.5" y="150" fill="#010101" fill-opacity=".3" transform="scale(0.1)" textLength="385.0" lengthAdjust="spacing">python</text><text x="422.5" y="140" transform="scale(0.1)" textLength="385.0" lengthAdjust="spacing">python</text><text x="782.5" y="150" fill="#010101" fill-opacity=".3" transform="scale(0.1)" textLength="175.0" lengthAdjust="spacing">3.11</text><text x="782.5" y="140" transform="scale(0.1)" textLength="175.0" lengthAdjust="spacing">3.11</text></g></svg>
+<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="93.0" height="20"><linearGradient id="smooth" x2="0" y2="100%"><stop offset="0" stop-color="#bbb" stop-opacity=".1"/><stop offset="1" stop-opacity=".1"/></linearGradient><clipPath id="round"><rect width="93.0" height="20" rx="3" fill="#fff"/></clipPath><g clip-path="url(#round)"><rect width="65.5" height="20" fill="#555"/><rect x="65.5" width="27.5" height="20" fill="#007ec6"/><rect width="93.0" height="20" fill="url(#smooth)"/></g><g fill="#fff" text-anchor="middle" font-family="DejaVu Sans,Verdana,Geneva,sans-serif" font-size="110"><image x="5" y="3" width="14" height="14" xlink:href="data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCAxMDAgMTAwIj4KICA8ZGVmcz4KICAgIDxsaW5lYXJHcmFkaWVudCBpZD0icHlZZWxsb3ciIGdyYWRpZW50VHJhbnNmb3JtPSJyb3RhdGUoNDUpIj4KICAgICAgPHN0b3Agc3RvcC1jb2xvcj0iI2ZlNSIgb2Zmc2V0PSIwLjYiLz4KICAgICAgPHN0b3Agc3RvcC1jb2xvcj0iI2RhMSIgb2Zmc2V0PSIxIi8+CiAgICA8L2xpbmVhckdyYWRpZW50PgogICAgPGxpbmVhckdyYWRpZW50IGlkPSJweUJsdWUiIGdyYWRpZW50VHJhbnNmb3JtPSJyb3RhdGUoNDUpIj4KICAgICAgPHN0b3Agc3RvcC1jb2xvcj0iIzY5ZiIgb2Zmc2V0PSIwLjQiLz4KICAgICAgPHN0b3Agc3RvcC1jb2xvcj0iIzQ2OCIgb2Zmc2V0PSIxIi8+CiAgICA8L2xpbmVhckdyYWRpZW50PgogIDwvZGVmcz4KCiAgPHBhdGggZD0iTTI3LDE2YzAtNyw5LTEzLDI0LTEzYzE1LDAsMjMsNiwyMywxM2wwLDIyYzAsNy01LDEyLTExLDEybC0yNCwwYy04LDAtMTQsNi0xNCwxNWwwLDEwbC05LDBjLTgsMC0xMy05LTEzLTI0YzAtMTQsNS0yMywxMy0yM2wzNSwwbDAtM2wtMjQsMGwwLTlsMCwweiBNODgsNTB2MSIgZmlsbD0idXJsKCNweUJsdWUpIi8+CiAgPHBhdGggZD0iTTc0LDg3YzAsNy04LDEzLTIzLDEzYy0xNSwwLTI0LTYtMjQtMTNsMC0yMmMwLTcsNi0xMiwxMi0xMmwyNCwwYzgsMCwxNC03LDE0LTE1bDAtMTBsOSwwYzcsMCwxMyw5LDEzLDIzYzAsMTUtNiwyNC0xMywyNGwtMzUsMGwwLDNsMjMsMGwwLDlsMCwweiBNMTQwLDUwdjEiIGZpbGw9InVybCgjcHlZZWxsb3cpIi8+CgogIDxjaXJjbGUgcj0iNCIgY3g9IjY0IiBjeT0iODgiIGZpbGw9IiNGRkYiLz4KICA8Y2lyY2xlIHI9IjQiIGN4PSIzNyIgY3k9IjE1IiBmaWxsPSIjRkZGIi8+Cjwvc3ZnPgo="/><text x="422.5" y="150" fill="#010101" fill-opacity=".3" transform="scale(0.1)" textLength="385.0" lengthAdjust="spacing">python</text><text x="422.5" y="140" transform="scale(0.1)" textLength="385.0" lengthAdjust="spacing">python</text><text x="782.5" y="150" fill="#010101" fill-opacity=".3" transform="scale(0.1)" textLength="175.0" lengthAdjust="spacing">3.9</text><text x="782.5" y="140" transform="scale(0.1)" textLength="175.0" lengthAdjust="spacing">3.9</text></g></svg>

requirements-dev.txt

@@ -4,4 +4,3 @@ ipython
 flake8
 pyinstrument
 django-debug-toolbar
-urllib3

requirements.txt

@@ -4,8 +4,7 @@ djangorestframework
 django-health-check
 psycopg2-binary
 paho-mqtt
-kafka-python
-aiomqtt
+asyncio-mqtt
 PyYAML
 uritemplate
 pygments