Add load test with celery monitor

README.md

@@ -57,6 +57,8 @@ VS Code can do type checking and type inference. However, it requires setting up
 
 ### Load testing
 
+First set the env var IS_LOAD_TEST to true in docker-compose.yml
+
 Locust is built into the dev dependencies. To run with Locust run
 `docker compose -f docker-compose.yml -f docker-compose.locust.yml up`
 

apps/event_ingest/api.py

@@ -26,6 +26,7 @@ router = Router(auth=event_auth)
 
 class EventIngestOut(Schema):
     event_id: str
+    task_id: Optional[str] = None  # For debug purposes only
 
 
 class EnvelopeIngestOut(Schema):
@@ -84,8 +85,11 @@ async def event_store(
         project_id=project_id,
         payload=issue_event_class(**payload.dict()),
     )
-    await async_call_celery_task(ingest_event, issue_event.dict())
-    return {"event_id": payload.event_id.hex}
+    task_result = await async_call_celery_task(ingest_event, issue_event.dict())
+    result = {"event_id": payload.event_id.hex}
+    if settings.IS_LOAD_TEST:
+        result["task_id"] = task_result.task_id
+    return result
 
 
 @router.post("/{project_id}/envelope/", response=EnvelopeIngestOut)

docker-compose.yml

@@ -7,6 +7,8 @@ x-environment: &default-environment
   DEBUG: "true"
   EMAIL_BACKEND: "django.core.mail.backends.console.EmailBackend"
   ENABLE_OBSERVABILITY_API: "true"
+  # GLITCHTIP_ENABLE_NEW_ISSUES: "true"
+  # IS_LOAD_TEST: "true"
   CELERY_WORKER_CONCURRENCY: 1
   PYTHONBREAKPOINT: "ipdb.set_trace"
 
@@ -38,11 +40,3 @@ services:
     depends_on: *default-depends_on
     volumes: *default-volumes
     environment: *default-environment
-  #flower:
-  #  build: .
-  #  command: bin/run-flower.sh
-  #  depends_on: *default-depends_on
-  #  volumes: *default-volumes
-  #  environment: *default-environment
-  #  ports:
-  #    - 5555:5555

glitchtip/api/api.py

@@ -36,9 +36,10 @@ if settings.GLITCHTIP_ENABLE_NEW_ISSUES:
     from apps.event_ingest.embed_api import router as embed_router
     from apps.issue_events.api import router as issue_events_router
 
-    api.add_router("v2", event_ingest_router)
-    api.add_router("v2", issue_events_router)
-    api.add_router("v2embed", embed_router)
+    # Remove the x to override old urls
+    api.add_router("x", event_ingest_router)
+    api.add_router("x0", issue_events_router)
+    api.add_router("xembed", embed_router)
 
 
 @api.exception_handler(ThrottleException)

glitchtip/settings.py

@@ -33,6 +33,7 @@ env = environ.FileAwareEnv(
     AZURE_ACCOUNT_KEY=(str, None),
     AZURE_CONTAINER=(str, None),
     AZURE_URL_EXPIRATION_SECS=(int, None),
+    IS_LOAD_TEST=(bool, False),
     GS_BUCKET_NAME=(str, None),
     GS_PROJECT_ID=(str, None),
     DEBUG=(bool, False),
@@ -455,6 +456,10 @@ if CELERY_BROKER_URL.startswith("sentinel"):
     CELERY_BROKER_TRANSPORT_OPTIONS["master_name"] = env.str(
         "CELERY_BROKER_MASTER_NAME", "mymaster"
     )
+IS_LOAD_TEST = env("IS_LOAD_TEST")
+# GlitchTip doesn't require a celery result backend
+if IS_LOAD_TEST:
+    CELERY_RESULT_BACKEND = REDIS_URL
 if socket_timeout := env.int("CELERY_BROKER_SOCKET_TIMEOUT", None):
     CELERY_BROKER_TRANSPORT_OPTIONS["socket_timeout"] = socket_timeout
 if broker_sentinel_password := env.str("CELERY_BROKER_SENTINEL_KWARGS_PASSWORD", None):

locustfile.py

@@ -1,15 +1,108 @@
+import datetime
+import os
+import time
+
+from celery.result import AsyncResult
 from locust import HttpUser, between, task
 
 from events.test_data.event_generator import generate_random_event
+from glitchtip.celery import app
+
+
+class CeleryClient:
+    """
+    CeleryClient is a wrapper around the Celery client.
+    It proxies any function calls and fires the *request* event when they finish,
+    so that the calls get recorded in Locust.
+    """
+
+    def __init__(self, request_event):
+        self.client = app
+        self.task_timeout = 60
+        self._request_event = request_event
+
+    def send_task(self, name, args=None, kwargs=None, queue=None):
+        options = {}
+        if queue:
+            options["queue"] = queue
+
+        request_meta = {
+            "request_type": "celery",
+            "response_length": 0,
+            "name": name,
+            "start_time": time.time(),
+            "response": None,
+            "context": {},
+            "exception": None,
+        }
+        t0 = datetime.datetime.now()
+        try:
+            async_result = self.client.send_task(
+                name, args=args, kwargs=kwargs, **options
+            )
+            result = async_result.get(self.task_timeout)  # blocking
+            request_meta["response"] = result
+            t1 = async_result.date_done
+        except Exception as e:
+            t1 = None
+            request_meta["exception"] = e
+
+        request_meta["response_time"] = (
+            None if not t1 else (t1 - t0).total_seconds() * 1000
+        )
+        self._request_event.fire(
+            **request_meta
+        )  # this is what makes the request actually get logged in Locust
+        return request_meta["response"]
+
+    def monitor_task(self, task_name, task_id):
+        """Monitor and record a known task by id"""
+        request_meta = {
+            "request_type": "celery",
+            "response_length": 0,
+            "name": task_name,
+            "start_time": time.time(),
+            "response": None,
+            "context": {},
+            "exception": None,
+        }
+        t0 = datetime.datetime.now()
+        try:
+            async_result = AsyncResult(task_id)
+            async_result.wait(self.task_timeout)
+            t1 = async_result.date_done
+        except Exception as e:
+            t1 = None
+            request_meta["exception"] = e
+        request_meta["response_time"] = (
+            None if not t1 else (t1 - t0).total_seconds() * 1000
+        )
+        self._request_event.fire(
+            **request_meta
+        )  # this is what makes the request actually get logged in Locust
+        return request_meta["response"]
 
 
 class WebsiteUser(HttpUser):
     wait_time = between(1.0, 2.0)
 
+    def __init__(self, environment):
+        super().__init__(environment)
+        self.celery_client = CeleryClient(
+            request_event=environment.events.request,
+        )
+
     @task
     def send_event(self):
         project_id = "1"
-        dsn = "2ed2762c07a04261bec95b197f500626"
+        dsn = ""
         event_url = f"/api/{project_id}/store/?sentry_key={dsn}"
         event = generate_random_event(True)
-        self.client.post(event_url, json=event)
+        res = self.client.post(event_url, json=event)
+        if os.environ.get("GLITCHTIP_ENABLE_NEW_ISSUES"):
+            task_id = res.json().get("task_id")
+            self.celery_client.monitor_task("ingest_event", task_id)
+
+    # @task
+    # def test_debug_task(self):
+    #     self.celery_client.send_task("glitchtip.celery.debug_task")