feat(Telemetry): Emit CLI usage events from an API middleware

api/app/settings/common.py

@@ -397,6 +397,7 @@
     "django.contrib.messages.middleware.MessageMiddleware",
     "django.middleware.clickjacking.XFrameOptionsMiddleware",
     "simple_history.middleware.HistoryRequestMiddleware",
+    "telemetry.middleware.CLIUsageLoggerMiddleware",
     "telemetry.middleware.MCPUsageLoggerMiddleware",  # Must come last!
 ]
 

api/projects/views.py

@@ -48,6 +48,7 @@
     ProjectRetrieveSerializer,
     ProjectUpdateSerializer,
 )
+from telemetry.spans import set_span_attribute
 from users.models import FFAdminUser
 
 
@@ -81,6 +82,17 @@ def get_serializer_class(self):  # type: ignore[no-untyped-def]
 
     pagination_class = None
 
+    def list(self, request, *args, **kwargs):  # type: ignore[no-untyped-def]
+        response = super().list(request, *args, **kwargs)
+
+        if response.data and (
+            organisation_id := request.query_params.get("organisation")
+        ):
+            # get_queryset validates the ID and restricts results to permitted projects.
+            set_span_attribute("organisation.id", int(organisation_id))
+
+        return response
+
     def get_serializer_context(self):  # type: ignore[no-untyped-def]
         return super().get_serializer_context()
 

api/telemetry/middleware.py

@@ -1,4 +1,6 @@
+import re
 from collections.abc import Callable
+from typing import cast
 
 import structlog
 from django.http.request import HttpRequest
@@ -7,6 +9,29 @@
 
 from telemetry.spans import get_span_attribute
 
+CLI_USER_AGENT_PATTERN = re.compile(
+    r"^flagsmith-cli/(?P<version>.+) \((?P<os>[^/()]+)/(?P<arch>[^/()]+)\)$"
+)
+
+
+def _get_organisation_id_from_request_context(
+    request: HttpRequest,
+) -> int | None:
+    # Set by the permission layer for organisations the user belongs to
+    if isinstance(
+        organisation_id := get_span_attribute("organisation.id"),
+        int,
+    ):
+        return organisation_id
+
+    if getattr(request.user, "is_master_api_key_user", False):
+        from api_keys.user import APIKeyUser
+
+        assert isinstance(request.user, APIKeyUser)
+        return cast(int, request.user.key.organisation_id)
+
+    return None
+
 
 class MCPUsageLoggerMiddleware:
     """Emit telemetry events for MCP usage"""
@@ -45,8 +70,9 @@ def _get_organisation_id(self, request: HttpRequest) -> int | None:
         """Obtain the organisation ID from the request context."""
         from organisations.models import Organisation
 
-        # Set by the permission layer for organisations the user belongs to
-        if isinstance(organisation_id := get_span_attribute("organisation.id"), int):
+        if (
+            organisation_id := _get_organisation_id_from_request_context(request)
+        ) is not None:
             return organisation_id
 
         assert request.user.is_authenticated  # NOTE: protected upstream
@@ -57,3 +83,45 @@ def _get_organisation_id(self, request: HttpRequest) -> int | None:
             Organisation.MultipleObjectsReturned,  # Don't guess
         ):
             return None
+
+
+class CLIUsageLoggerMiddleware:
+    """Emit telemetry events for Flagsmith CLI usage."""
+
+    def __init__(
+        self,
+        get_response: Callable[[HttpRequest], HttpResponse],
+    ) -> None:
+        self.get_response = get_response
+
+    def __call__(self, request: HttpRequest) -> HttpResponse:
+        response = self.get_response(request)
+
+        match = CLI_USER_AGENT_PATTERN.fullmatch(request.headers.get("User-Agent", ""))
+        if match is None:
+            return response
+
+        if not request.user or not request.user.is_authenticated:
+            return response
+
+        event = {
+            "cli__version": match.group("version"),
+            "cli__os": match.group("os"),
+            "cli__arch": match.group("arch"),
+            "status": "error" if response.status_code >= 400 else "success",
+        }
+
+        if (
+            organisation_id := _get_organisation_id_from_request_context(request)
+        ) is not None:
+            event["organisation__id"] = organisation_id
+
+        if not getattr(request.user, "is_master_api_key_user", False):
+            from users.models import FFAdminUser
+
+            assert isinstance(request.user, FFAdminUser)
+            event["amplitude__user_id"] = str(request.user.uuid)
+
+        structlog.get_logger("cli").info("request.made", **event)
+
+        return response