System test, updated monitor interface

Author: lpicou

README.md

@@ -75,12 +75,32 @@ A Web Interface is available to monitor which frames are pending for which devic
 ![sw arch](images/arch.png)
 
 
+# Sizing
+
+For the docker deployment, container should have the following sizing :
+
+- CPU : 1 Core
+- RAM : 256 MB
+- Network : 1 Mbps
+- Disk space : No disk usage. Everything is stored in-memory
+
 # Test
 
+Tests dependencies are listed in the `requirements.txt` file under `tests` folder.
 Tests should always be run inside the `tests` folder for proper initialization.
 
+
+# Limitation
+
+Stress tests shows the following :
+    - At very high load (thousands of different devEUI over a minute) : 
+      - Monitor interface can fail because the `frame_buffer` object is being rendered to front-end and modified in same time.
+      - Self-hosted MQTT broker can crash (mosquitto process)
+    - For a very high number of pending fragment (100k different devEUI) : Monitor interface can take up to 1 mn to fully load, because everything is displayed in the front-end.
+
+
 # Roadmap
 
 - HTTP Input/Output
 - Do some modularization to allow interfacing multiple decoders
-- Tests
+- Multi-stage docker to remove NPM dependencies (which add few hundreds of MB to the image size)

src/app/lib/validate_config.py

@@ -64,7 +64,7 @@
         "type": "dict",
         "schema": {
             "max_chunks": {"type": "integer", "min": 1, "required": True},
-            "timeout": {"type": "integer", "min": 1, "required": True},
+            "timeout": {"type": "float", "min": 0.005, "required": True}, # min is 20s
             "lns": {"type": "string", "allowed": ["ttn", "loriot"], "required": True},
         },
     },

src/app/main.py

@@ -144,8 +144,10 @@ def on_mqtt_message(client, userdata, message: mqtt.MQTTMessage) -> None:
     # We don't care about non fragmented frames
     if frame["fPort"] in {FRAGMENT_FPORT, LAST_FRAGMENT_FPORT}:
         with frame_lock:
+
             if frame["devEUI"] in frame_buffer:
                 frame_buffer[frame["devEUI"]].append(frame)
+                logger.info(f"Received fragment from {frame['devEUI']}")
             else:
                 frame_buffer[frame["devEUI"]] = [frame]
 
@@ -172,7 +174,7 @@ def parse_mqtt(message: mqtt.MQTTMessage):
             exit(1)
 
     if not frame:
-        logger.debug("Could not decode the MQTT received.")
+        logger.debug("Could not decode the MQTT received. Check that LNS is the right one in config.yaml.")
         raise InvalidFrame
 
     return frame
@@ -208,13 +210,14 @@ def receive_http_chunk():
 def monitor_buffer():
     table_data = copy.deepcopy(frame_buffer)
 
+
     for frame_list in table_data.values():
         for frame in frame_list:
             frame["received_time_str"] = datetime.datetime.fromtimestamp(frame["received_time"]).strftime("%Y-%m-%d %H:%M:%S") # type: ignore
             frame["raw_hex"] =  frame["raw"].hex() # type: ignore
 
 
-    result = render_template("monitor.html",data=table_data)
+    result = render_template("monitor.html",data=table_data, timeout=get_timeout(), max_chunk=get_max_chunk())
     return result, 200
 
 ######### OUTPUT #########
@@ -224,8 +227,9 @@ def monitor_buffer():
 
 # MQTT : Publish
 def send_mqtt_message(devEUI: str, frame: dict):
-    logger.debug(f"Publishing decoded frame for DevEUI {devEUI} to MQTT")
-    res = client_mqtt_output.publish(config["output"]["mqtt"]["topic"]+f"/{devEUI}", json.dumps(frame))
+    topic = config["output"]["mqtt"]["topic"]+f"/{devEUI}"
+    logger.info(f"Publishing decoded frame for DevEUI {devEUI} to MQTT on topic {topic}")
+    res = client_mqtt_output.publish(topic, json.dumps(frame))
     if res.rc == mqtt.MQTT_ERR_SUCCESS:
         return True
     else:
@@ -306,7 +310,6 @@ def init_timeout_checker():
 
 def init_output():
     global client_mqtt_output
-    print(config)
     if config["output"]["mqtt"]["enable"] == True:
         try:
             status_code = client_mqtt_output.connect(config["output"]["mqtt"]["host"], config["output"]["mqtt"]["port"])

src/app/templates/monitor.html

@@ -12,6 +12,8 @@
 
     <h2>List of pending fragments</h2>
     <p>If the table is empty, it means that there are no pending frames to be re-assembled.</p>
+    <p>Timeout configured : {{ timeout }} hours</p>
+    <p>Max Chunk configured : {{ max_chunk }}</p>
 
     <button id="refreshBtn">Refresh Page</button>
 

tests/benchmark.py

@@ -0,0 +1,80 @@
+# File used to benchmark app with a lot of requests
+import paho.mqtt.client as mqtt
+from paho.mqtt.enums import CallbackAPIVersion
+import json
+import time
+from threading import Thread
+
+
+
+def on_message(client, userdata, message: mqtt.MQTTMessage) -> None:
+    decoded = json.loads(message.payload)
+    print("message, ", decoded)
+
+    pass
+    
+
+DEVEUI = "4200000000000000"
+client_mqtt = mqtt.Client(callback_api_version=CallbackAPIVersion.VERSION2)
+client_mqtt.on_message = on_message
+status_code = client_mqtt.connect("localhost", 1883)
+client_mqtt.subscribe("output/topic/#")
+client_mqtt.loop_start()
+
+
+msg1 = json.load(open("payload/payload_ttn_fragment1.json"))
+last = json.load(open("payload/payload_ttn_fragment_last.json"))
+
+
+def scenario1():
+    # SCENARIO 1 : Lot of proper request
+
+    FRAME_NUMBER = 100
+    for i in range(FRAME_NUMBER):
+        # Change devEUI dynamically
+
+        msg1["end_device_ids"]["dev_eui"] = "1Dev" +str(i) 
+        last["end_device_ids"]["dev_eui"] = "1Dev" +str(i) 
+
+        res = client_mqtt.publish(payload=json.dumps(msg1), topic="test")
+        res.wait_for_publish()
+
+
+        res = client_mqtt.publish(payload=json.dumps(last), topic="test")
+        res.wait_for_publish()
+
+
+def scenario2():
+    # SCENARIO 2 : lot of pending fragments
+
+    FRAME_NUMBER = 10 *1000
+    for i in range(FRAME_NUMBER):
+        # Change devEUI dynamically
+
+        msg1["end_device_ids"]["dev_eui"] = "2Dev" +str(i) 
+
+        res = client_mqtt.publish(payload=json.dumps(msg1), topic="test")
+        res.wait_for_publish()
+
+def scenario3():
+
+    FRAME_NUMBER = 100
+    THREAD_COUNT = 10
+    def spam(dev_prefix):
+        for i in range(FRAME_NUMBER):
+            msg1["end_device_ids"]["dev_eui"] = f"{dev_prefix}{i}"
+            client_mqtt.publish(payload=json.dumps(msg1), topic="test").wait_for_publish()
+
+    threads = [Thread(target=spam, args=(f"concurrent{i}_",)) for i in range(THREAD_COUNT)]
+
+    for t in threads: t.start()
+    for t in threads: t.join()
+
+
+
+scenario3()
+
+
+# Finish sending
+time.sleep(1)
+print("Benchmark done")

tests/conftest.py

@@ -148,3 +148,6 @@ def start_system(monkeymodule):
     time.sleep(1)
 
     yield mosquitto_process
+
+    with pytest.raises(SystemExit):
+        main.shutdown(mosquitto_process)

tests/system/test_system.py

@@ -1,8 +1,13 @@
+import json
+import time
 import pytest
 import requests
 
 # Project import
 import main
+import paho.mqtt.client as mqtt
+from paho.mqtt.enums import CallbackAPIVersion
+from tests.static import EXAMPLE_CONFIG, wait_until, DATAFORMAT2_DECODED
 
 @pytest.mark.usefixtures("start_system")
 class TestSystem:
@@ -12,6 +17,56 @@ def test_system_monitor(self,start_system):
         assert requests.get(f'http://localhost:{main.config["input"]["http"]["port"]}/monitor').status_code == 200
 
 
+    def test_frame_accepted(self,start_system):
+        """Check that a TTN frame is accepted and seen in monitor page"""
+        
+        frame_received = None
+
+        def on_message(client, userdata, message: mqtt.MQTTMessage) -> None:
+            nonlocal frame_received
+            decoded = json.loads(message.payload)
+
+            # It means we recieved a true decoded frame
+            if "preset_id" in decoded["data"]:
+                frame_received = decoded
+            pass
+            
+        DEVEUI = "4200000000000000"
+        client_mqtt = mqtt.Client(callback_api_version=CallbackAPIVersion.VERSION2)
+        client_mqtt.on_message = on_message
+        status_code = client_mqtt.connect("localhost", EXAMPLE_CONFIG["output"]["mqtt"]["port"])
+        client_mqtt.subscribe(f"output/{DEVEUI}")
+        client_mqtt.loop_start()
+
+
+        assert status_code == mqtt.MQTT_ERR_SUCCESS
+
+
+        assert DEVEUI not in main.frame_buffer
+
+        # Fragment 1
+        raw_msg = json.load(open("payload/payload_ttn_fragment1.json"))
+        client_mqtt.publish(payload=json.dumps(raw_msg), topic=EXAMPLE_CONFIG["input"]["mqtt"]["topic"])
+
+        
+        assert wait_until(lambda: DEVEUI in main.frame_buffer)
+        assert len(main.frame_buffer[DEVEUI]) == 1
+
+        # Fragment 2
+        raw_msg = json.load(open("payload/payload_ttn_fragment_last.json"))
+        client_mqtt.publish(payload=json.dumps(raw_msg), topic=EXAMPLE_CONFIG["input"]["mqtt"]["topic"])
+        
+
+        # No longer fragmented
+        assert wait_until(lambda: DEVEUI not in main.frame_buffer)
+
+        # We received the answer, decoded
+        assert wait_until(lambda: frame_received == DATAFORMAT2_DECODED)
+
+
+ 
+        
+
 
     def test_stop_system(self,start_system):
         """Check that launch func launch every services"""