feat: Add command to monitor the load test

This doesn't work in the current form, but is getting close.

platform_plugin_aspects/management/commands/monitor_load_test_tracking.py

@@ -1,8 +1,12 @@
 """
 Monitors the load test tracking script and saves output for later analysis.
 """
+import csv
+import io
+import json
 import logging
 import datetime
+import uuid
 from pprint import pprint
 from time import sleep
 from typing import Any
@@ -22,42 +26,77 @@
 
 class Monitor:
     def __init__(self, sleep_time: float):
-        # TODO: Eventually I'd like to store the stats data in Clickhouse with a sink
-        # self.sink = BaseSink()
+        self.run_id = str(uuid.uuid4())[:6]
         self.ch_url = settings.EVENT_SINK_CLICKHOUSE_BACKEND_CONFIG["url"]
-
         self.ch_auth = ClickHouseAuth(
             settings.EVENT_SINK_CLICKHOUSE_BACKEND_CONFIG["username"],
             settings.EVENT_SINK_CLICKHOUSE_BACKEND_CONFIG["password"],
         )
         self.ch_database = settings.EVENT_SINK_CLICKHOUSE_BACKEND_CONFIG["database"]
         self.ch_xapi_database = settings.EVENT_SINK_CLICKHOUSE_BACKEND_CONFIG.get("xapi_database", "xapi")
         self.ch_xapi_table = settings.EVENT_SINK_CLICKHOUSE_BACKEND_CONFIG.get("xapi_table", "xapi_events_all")
-        self.ch_stats_database = settings.EVENT_SINK_CLICKHOUSE_BACKEND_CONFIG.get("stats_database", "load_test")
         self.ch_stats_table = settings.EVENT_SINK_CLICKHOUSE_BACKEND_CONFIG.get("stats_table", "load_test_stats")
         self.ch_timeout_secs = settings.EVENT_SINK_CLICKHOUSE_BACKEND_CONFIG[
             "timeout_secs"
         ]
         self.sleep_time = sleep_time
 
     def run(self) -> None:
+        collect_redis_bus = True
+        collect_celery = True
+        collect_kafka_bus = False
+
         while True:
             # TODO: Need to pass in which stats to collect, or infer them from config
-            self.get_clickhouse_stats()
-            # self.get_celery_stats()
-            self.get_redis_bus_stats()
+            current_stats = {"clickhouse": self.get_clickhouse_stats()}
+            if collect_redis_bus:
+                current_stats["redis_bus"] = self.get_redis_bus_stats()
+            if collect_celery:
+                current_stats["celery"] = self.get_celery_stats()
+            # if collect_kafka_bus:
+                # current_stats["kafka_bus"] = self.get_kafka_stats()
+
+            self.store_stats(current_stats)
 
             # TODO: This should eventually be sleep time - however long it's been since the last run
             # so it stays on a steady cadence if the stats start taking longer. It should also
             # error if stats collection starts taking longer than sleep time.
             sleep(self.sleep_time)
 
+    def store_stats(self, current_stats: dict) -> None:
+        stats = json.dumps(current_stats)
+
+        # TODO: There must be a way to parameterize this
+        insert = f"""INSERT INTO {self.ch_stats_table} (run_id, stats) FORMAT CSV"""
+
+        output = io.StringIO()
+        writer = csv.writer(output, quoting=csv.QUOTE_NONNUMERIC)
+        writer.writerow((self.run_id, stats))
+
+        response = requests.post(
+            url=self.ch_url,
+            auth=self.ch_auth,
+            params={
+                "database": self.ch_database,
+                "query": insert
+            },
+            data=output.getvalue().encode("utf-8"),
+            timeout=self.ch_timeout_secs
+        )
+
+        response.raise_for_status()
+        print(datetime.datetime.now())
+
     def get_clickhouse_stats(self):
         select = f"""
-            SELECT count(*) as ttl_count, max(emission_time) as most_recent,
-                date_diff('second', max(emission_time), now())
+            SELECT
+                count(*) as ttl_count,
+                max(emission_time) as most_recent,
+                date_diff('second', max(emission_time), now()) as lag_seconds
             FROM {self.ch_xapi_table}
+            FORMAT JSON
         """
+
         response = requests.post(
             url=self.ch_url,
             auth=self.ch_auth,
@@ -67,19 +106,29 @@ def get_clickhouse_stats(self):
             },
             timeout=self.ch_timeout_secs
         )
-        print(response.text)
+
         response.raise_for_status()
+        resp = response.json()["data"][0]
+        print(f"Clickhouse lag seconds: {resp['lag_seconds']}")
 
-        return response
+        return {
+            "total_rows": resp["ttl_count"],
+            "most_recent_event": resp["most_recent"],
+            "lag_seconds": resp["lag_seconds"],
+        }
 
     def get_celery_stats(self):
-        # TODO: Still trying to find the right way to access these queues
-        r = redis.Redis(host="redis", port=6379, db=1)
-        if r.llen("edx.lms.core.default"):
-            raise Exception("yey")
+        # TODO: Get this information from config
+        r = redis.Redis(host="redis", port=6379, db=0)
+        lms_queue = r.llen("edx.lms.core.default")
+        cms_queue = r.llen("edx.cms.core.default")
+
+        print(f"Celery queues: LMS {lms_queue}, CMS {cms_queue}")
 
-        if r.llen("edx.cms.core.default"):
-            raise Exception("yey1111")
+        return {
+            "lms_queue_length": lms_queue,
+            "cms_queue_length": cms_queue,
+        }
 
     def get_redis_bus_stats(self):
         # TODO: Get this information from config
@@ -115,11 +164,15 @@ def get_redis_bus_stats(self):
         consumer_stats = {
             "total_events": info["length"],
             "consumers": [
-                {"name": x[1], "processing": x[7], "queue_length": len(x[9])}
+                {"name": str(x[1]), "processing": x[7], "queue_length": len(x[9])}
                 for x in all_consumers
             ]
         }
 
+        total_queue = sum([x["queue_length"] for x in consumer_stats["consumers"]])
+
+        print(f"Redis bus total queue length: {total_queue}")
+
         return consumer_stats