localstack · giograno · Nov 15, 2024 · Nov 18, 2024 · Nov 18, 2024 · Nov 19, 2024
@@ -76,6 +76,7 @@ def __init__(self, service_manager: ServiceManager = None) -> None:
                 handlers.add_cors_response_headers,
                 handlers.log_response,
                 handlers.count_service_request,
+                handlers.user_agent_request,
                 metric_collector.update_metric_collection,
             ]
         )

@@ -35,6 +35,7 @@
 log_exception = logging.ExceptionLogger()
 log_response = logging.ResponseLogger()
 count_service_request = analytics.ServiceRequestCounter()
+user_agent_request = analytics.UserAgentCounter()
 handle_service_exception = service.ServiceExceptionSerializer()
 handle_internal_failure = fallback.InternalFailureHandler()
 serve_custom_service_request_handlers = chain.CompositeHandler()

@@ -1,16 +1,18 @@
 import logging
 import threading
-from typing import Optional
+from typing import Any, Optional
 
 from localstack import config
 from localstack.aws.api import RequestContext
 from localstack.aws.chain import HandlerChain
 from localstack.aws.client import parse_response
+from localstack.constants import INTERNAL_RESOURCE_PATH
 from localstack.http import Response
 from localstack.utils.analytics.service_request_aggregator import (
     ServiceRequestAggregator,
     ServiceRequestInfo,
 )
+from localstack.utils.analytics.usage import UsageSetCounter
 
 LOG = logging.getLogger(__name__)
 
@@ -67,3 +69,52 @@ def _get_err_type(self, context: RequestContext, response: Response) -> Optional
             if config.DEBUG_ANALYTICS:
                 LOG.exception("error parsing error response")
             return None
+
+
+class UsageCollectorFactory:
+    _collector_registry: dict[str, Any] = {}
+    """Registry for the different paths."""
+
+    NAMESPACE_PREFIX = "agent:"
+    """Namespace prefix to track usage of public endpoints (_localstack/ and _aws/)."""
+
+    @classmethod
+    def get_collector(cls, path: str):
+        namespace = f"{cls.NAMESPACE_PREFIX}{path}"
+        if namespace not in cls._collector_registry:
+            cls._collector_registry[namespace] = UsageSetCounter(namespace)
+        return cls._collector_registry[namespace]
+
+
+class UserAgentCounter:
+    """
+    This handler collects User-Agents analytics for the LocalStack public endpoints (the ones with a _localstack or a
+    _aws prefix).
+    """
+
+    def _record_usage(self, context: RequestContext) -> None:
+        request_path = context.request.path.lstrip("/")
+        user_agent = context.request.headers.get("User-Agent")
+        if not request_path or not user_agent:
+            return
+        # Skip the endpoints for the new API Gateway implementation
+        if "execute-api" in request_path:
+            return
+        # We only record the first segment in the path after the _internal/ or _aws/ prefix, as a path can have
+        #   potentially an infinite number of parameters.
+        recorded_path = request_path.split("/")[:2]
+        if len(recorded_path) < 2:
+            return
+        recorded_path = "/".join(recorded_path)
+        collector = UsageCollectorFactory.get_collector(recorded_path)
+        collector.record(user_agent)
+
+    def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):
+        if config.DISABLE_EVENTS:
+            return
+
+        path = context.request.path
+        if not (path.startswith(f"{INTERNAL_RESOURCE_PATH}/") or path.startswith("/_aws/")):
+            return
+
+        self._record_usage(context)
@@ -45,31 +45,33 @@ class OpenAPIRequestValidator(OpenAPIValidator):
     """
 
     def __call__(self, chain: HandlerChain, context: RequestContext, response: Response):
+        path = context.request.path
+        if not (path.startswith(f"{INTERNAL_RESOURCE_PATH}/") or path.startswith("/_aws/")):
+            return
+
         if not config.OPENAPI_VALIDATE_REQUEST:
             return
 
         hasattr(self, "open_apis") or self._load_specs()
-        path = context.request.path
 
-        if path.startswith(f"{INTERNAL_RESOURCE_PATH}/") or path.startswith("/_aws/"):
-            for openapi in self.open_apis:
-                try:
-                    openapi.validate_request(WerkzeugOpenAPIRequest(context.request))
-                    # We stop the handler at the first succeeded validation, as the other spec might not even specify
-                    #   this path.
-                    break
-                except RequestValidationError as e:
-                    # Note: in this handler we only check validation errors, e.g., wrong body, missing required.
-                    response.status_code = 400
-                    response.set_json({"error": "Bad Request", "message": str(e)})
-                    chain.stop()
-                except OpenAPIError:
-                    # Other errors can be raised when validating a request against the OpenAPI specification.
-                    #   The most common are: ServerNotFound, OperationNotFound, or PathNotFound.
-                    #   We explicitly do not check any other error but RequestValidationError ones.
-                    #   We shallow the exception to avoid excessive logging (e.g., a lot of ServerNotFound), as the only
-                    #   purpose of this handler is to check for request validation errors.
-                    pass
+        for openapi in self.open_apis:
+            try:
+                openapi.validate_request(WerkzeugOpenAPIRequest(context.request))
+                # We stop the handler at the first succeeded validation, as the other spec might not even specify
+                #   this path.
+                break
+            except RequestValidationError as e:
+                # Note: in this handler we only check validation errors, e.g., wrong body, missing required.
+                response.status_code = 400
+                response.set_json({"error": "Bad Request", "message": str(e)})
+                chain.stop()
+            except OpenAPIError:
+                # Other errors can be raised when validating a request against the OpenAPI specification.
+                #   The most common are: ServerNotFound, OperationNotFound, or PathNotFound.
+                #   We explicitly do not check any other error but RequestValidationError ones.
+                #   We shallow the exception to avoid excessive logging (e.g., a lot of ServerNotFound), as the only
+                #   purpose of this handler is to check for request validation errors.
+                pass
 
 
 class OpenAPIResponseValidator(OpenAPIValidator):
@@ -81,20 +83,21 @@ def __call__(self, chain: HandlerChain, context: RequestContext, response: Respo
 
         hasattr(self, "open_apis") or self._load_specs()
         path = context.request.path
+        if not (path.startswith(f"{INTERNAL_RESOURCE_PATH}/") or path.startswith("/_aws/")):
+            return
 
-        if path.startswith(f"{INTERNAL_RESOURCE_PATH}/") or path.startswith("/_aws/"):
-            for openapi in self.open_apis:
-                try:
-                    openapi.validate_response(
-                        WerkzeugOpenAPIRequest(context.request),
-                        WerkzeugOpenAPIResponse(response),
-                    )
-                    break
-                except ResponseValidationError as exc:
-                    LOG.error("Response validation failed for %s: %s", path, exc)
-                    response.status_code = 500
-                    response.set_json({"error": exc.__class__.__name__, "message": str(exc)})
-                    chain.terminate()
-                except OpenAPIError:
-                    # Same logic from the request validator applies here.
-                    pass
+        for openapi in self.open_apis:
+            try:
+                openapi.validate_response(
+                    WerkzeugOpenAPIRequest(context.request),
+                    WerkzeugOpenAPIResponse(response),
+                )
+                break
+            except ResponseValidationError as exc:
+                LOG.error("Response validation failed for %s: %s", path, exc)
+                response.status_code = 500
+                response.set_json({"error": exc.__class__.__name__, "message": str(exc)})
+                chain.terminate()
+            except OpenAPIError:
+                # Same logic from the request validator applies here.
+                pass