Add api count metrics, add get_user_id method

Author: nss10

gen3workflow/app.py

@@ -1,18 +1,21 @@
 from fastapi import FastAPI
+from fastapi.security import HTTPAuthorizationCredentials
 import httpx
 from importlib.metadata import version
 import os
+import time
 
 from cdislogging import get_logger
 from gen3authz.client.arborist.async_client import ArboristClient
-
+from fastapi import Request, HTTPException
 from gen3workflow import logger
 from gen3workflow.config import config
 from gen3workflow.metrics import Metrics
 from gen3workflow.routes.ga4gh_tes import router as ga4gh_tes_router
 from gen3workflow.routes.s3 import router as s3_router
 from gen3workflow.routes.storage import router as storage_router
 from gen3workflow.routes.system import router as system_router
+from gen3workflow.auth import Auth
 
 
 def get_app(httpx_client=None) -> FastAPI:
@@ -65,6 +68,39 @@ def get_app(httpx_client=None) -> FastAPI:
     if app.metrics.enabled:
         app.include_router("/metrics", app.metrics.get_asgi_app())
 
+    @app.middleware("http")
+    async def middleware_log_response_and_api_metric(
+        request: Request, call_next
+    ) -> None:
+        """
+        This FastAPI middleware effectively allows pre and post logic to a request.
+
+        We are using this to log the response consistently across defined endpoints (including execution time).
+
+        Args:
+            request (Request): the incoming HTTP request
+            call_next (Callable): function to call (this is handled by FastAPI's middleware support)
+        """
+        start_time = time.perf_counter()
+        response = await call_next(request)
+        response_time_seconds = time.perf_counter() - start_time
+
+        path = request.url.path
+        method = request.method
+        if path not in config["ENDPOINTS_WITH_METRICS"]:
+            return response
+
+        # TODO: Add user_id to this metric
+        metrics = app.metrics
+        metrics.add_create_task_api_interaction(
+            method=method,
+            path=path,
+            response_time_seconds=response_time_seconds,
+            status_code=response.status_code,
+        )
+
+        return response
+
     return app
 
 

gen3workflow/auth.py

@@ -1,3 +1,4 @@
+from typing import Union
 from authutils.token.fastapi import access_token
 from fastapi import HTTPException, Security
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
@@ -65,6 +66,13 @@ async def get_token_claims(self) -> dict:
 
         return token_claims
 
+    async def get_user_id(self) -> Union[str, None]:
+        try:
+            token_claims = await self.get_token_claims()
+        except Exception:
+            return None
+        return token_claims.get("sub")
+
     async def authorize(
         self,
         method: str,

gen3workflow/config-default.yaml

@@ -61,3 +61,4 @@ TASK_IMAGE_WHITELIST: []
 
 ENABLE_PROMETHEUS_METRICS: false
 PROMETHEUS_MULTIPROC_DIR: /var/tmp/prometheus_metrics
+ENDPOINTS_WITH_METRICS: [ /ga4gh/tes/v1/tasks ]

gen3workflow/config.py

@@ -63,6 +63,10 @@ def validate_top_level_configs(self) -> None:
                 "TES_SERVER_URL": {"type": "string"},
                 "ENABLE_PROMETHEUS_METRICS": {"type": "boolean"},
                 "PROMETHEUS_MULTIPROC_DIR": {"type": "string"},
+                "ENDPOINTS_WITH_METRICS": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                },
             },
         }
         validate(instance=self, schema=schema)

gen3workflow/metrics.py

@@ -10,3 +10,12 @@ def __init__(self, prometheus_dir: str, enabled: bool = True) -> None:
         super().__init__(
             prometheus_dir=config["PROMETHEUS_MULTIPROC_DIR"], enabled=enabled
         )
+
+    def add_create_task_api_interaction(
+        self,
+        **kwargs: Dict[str, Any],
+    ) -> None:
+        """
+        Add a metric for create_task API interactions
+        """
+        self.increment_counter(name="create_task_api_interaction", labels=kwargs)

gen3workflow/routes/ga4gh_tes.py

@@ -32,11 +32,7 @@ async def get_request_body(request: Request):
 
 @router.get("/service-info", status_code=HTTP_200_OK)
 async def service_info(request: Request, auth=Depends(Auth)) -> dict:
-    try:
-        token_claims = await auth.get_token_claims()
-    except Exception:
-        token_claims = {}
-    user_id = token_claims.get("sub")
+    user_id = await auth.get_user_id()
     logger.info(f"User '{user_id}' getting TES service info")
 
     url = f"{config['TES_SERVER_URL']}/service-info"
@@ -164,11 +160,7 @@ def apply_view_to_task(view: str, task: dict) -> dict:
 
 @router.get("/tasks", status_code=HTTP_200_OK)
 async def list_tasks(request: Request, auth=Depends(Auth)) -> dict:
-    try:
-        token_claims = await auth.get_token_claims()
-    except Exception:
-        token_claims = {}
-    user_id = token_claims.get("sub")
+    user_id = await auth.get_user_id()
     logger.info(f"User '{user_id}' listing TES tasks")
 
     supported_params = {
@@ -231,11 +223,7 @@ async def list_tasks(request: Request, auth=Depends(Auth)) -> dict:
 
 @router.get("/tasks/{task_id}", status_code=HTTP_200_OK)
 async def get_task(request: Request, task_id: str, auth=Depends(Auth)) -> dict:
-    try:
-        token_claims = await auth.get_token_claims()
-    except Exception:
-        token_claims = {}
-    user_id = token_claims.get("sub")
+    user_id = await auth.get_user_id()
     logger.info(f"User '{user_id}' getting TES task '{task_id}'")
 
     supported_params = {"view"}
@@ -268,11 +256,7 @@ async def get_task(request: Request, task_id: str, auth=Depends(Auth)) -> dict:
 
 @router.post("/tasks/{task_id}:cancel", status_code=HTTP_200_OK)
 async def cancel_task(request: Request, task_id: str, auth=Depends(Auth)) -> dict:
-    try:
-        token_claims = await auth.get_token_claims()
-    except Exception:
-        token_claims = {}
-    user_id = token_claims.get("sub")
+    user_id = await auth.get_user_id()
     logger.info(f"User '{user_id}' canceling TES task '{task_id}'")
 
     # check if this user has access to delete this task

gen3workflow/routes/s3.py

@@ -86,8 +86,7 @@ async def s3_endpoint(path: str, request: Request):
     await auth.authorize("create", ["/services/workflow/gen3-workflow/tasks"])
 
     # get the name of the user's bucket and ensure the user is making a call to their own bucket
-    token_claims = await auth.get_token_claims()
-    user_id = token_claims.get("sub")
+    user_id = await auth.get_user_id()
     logger.info(f"Incoming S3 request from user '{user_id}': '{request.method} {path}'")
     user_bucket = aws_utils.get_safe_name_from_hostname(user_id)
     request_bucket = path.split("?")[0].split("/")[0]

gen3workflow/routes/storage.py

@@ -10,8 +10,7 @@
 
 @router.get("/info", status_code=HTTP_200_OK)
 async def get_storage_info(request: Request, auth=Depends(Auth)) -> dict:
-    token_claims = await auth.get_token_claims()
-    user_id = token_claims.get("sub")
+    user_id = await auth.get_user_id()
     logger.info(f"User '{user_id}' getting their own storage info")
     bucket_name, bucket_prefix, bucket_region = aws_utils.create_user_bucket(user_id)
     return {