helper(scalingo-log): add status code 5xx to search

app/commands.py

@@ -15,7 +15,11 @@
 from app.domain.regulations import compute_regulation_for_user
 from app.domain.vehicle import find_vehicle
 from app.helpers.oauth.models import ThirdPartyApiKey
-from app.helpers.scalingo_logs import get_long_requests, get_user_requests
+from app.helpers.scalingo_logs import (
+    get_long_requests,
+    get_user_requests,
+    get_status_5xx_requests_all,
+)
 from app.helpers.xml.greco import temp_write_greco_xml
 from app.models.controller_control import ControllerControl
 from app.models.user import User
@@ -315,3 +319,18 @@ def log_long_requests():
 @click.argument("user_id", required=True)
 def log_user_requests(user_id):
     get_user_requests(user_id=user_id, nb_results=20, nb_lines=1000000)
+
+
+@app.cli.command("log_status_5xx_requests_all", with_appcontext=False)
+@click.argument("nb_lines", required=True, type=int)
+@click.option(
+    "--source",
+    required=True,
+    type=click.Choice(["scalingo", "file"]),
+    help="'scalingo' or 'file'.",
+)
+@click.option("--file_path", required=False, type=str, help="File path.")
+def log_status_5xx_requests_all(nb_lines, source, file_path):
+    get_status_5xx_requests_all(
+        nb_lines=nb_lines, source_type=source, file_path=file_path
+    )

app/helpers/scalingo_logs.py

@@ -12,12 +12,30 @@ def extract_endpoint_from_log(log):
     return endpoint_match.group(1) if endpoint_match else ""
 
 
+def extract_path_from_log(log):
+    path_pattern = r'path="([^"]+)"'
+    path_match = re.search(path_pattern, log)
+    return path_match.group(1) if path_match else ""
+
+
+def extract_referer_from_log(log):
+    referer_pattern = r'referer="([^"]+)"'
+    referer_match = re.search(referer_pattern, log)
+    return referer_match.group(1) if referer_match else ""
+
+
 def extract_time_from_log(log):
     time_pattern = r"time=(\d+)ms"
     time_match = re.search(time_pattern, log)
     return int(time_match.group(1)) if time_match else None
 
 
+def extract_duration_from_log(log):
+    time_pattern = r"duration=([\d.]+)s"
+    time_match = re.search(time_pattern, log)
+    return float(time_match.group(1)) if time_match else None
+
+
 def extract_user_from_log(log):
     username_pattern = r"user=([\w\s]+) user_id"
     userid_pattern = r"user_id=([\d]+)"
@@ -31,6 +49,18 @@ def extract_user_from_log(log):
     return (username, userid)
 
 
+def extract_status_code_from_log(log):
+    status_code_pattern = r"status=(5\d{2})"
+    status_code_match = re.search(status_code_pattern, log)
+    return status_code_match.group(1) if status_code_match else ""
+
+
+def extract_request_id_from_log(log):
+    request_id_pattern = r"request_id=([^ ]+)"
+    request_id_match = re.search(request_id_pattern, log)
+    return request_id_match.group(1) if request_id_match else ""
+
+
 def get_long_requests(nb_results=20, nb_lines=10000):
     cmd = f"scalingo --region osc-fr1 --app mobilic-api logs --lines {nb_lines} | grep '\[web-' | awk 'NF'"
 
@@ -68,3 +98,43 @@ def get_user_requests(user_id, nb_results=20, nb_lines=10000):
 
     for l in lines[:nb_results]:
         print(l)
+
+
+def get_status_5xx_requests_all(nb_lines, source_type, file_path=None):
+    if source_type == "scalingo":
+        cmd = f"scalingo --region osc-fr1 --app mobilic-api logs --lines {nb_lines} | grep 'status=5' || true"
+    elif source_type == "file" and file_path:
+        cmd = f"cat {file_path} | grep 'status=5'"
+    else:
+        raise ValueError(
+            "source_type must be 'scalingo' or 'file' with a valid file_path."
+        )
+
+    lines = [
+        l.decode("utf-8")
+        for l in subprocess.check_output(cmd, shell=True).splitlines()
+    ]
+
+    lines_sorted = sorted(lines, key=lambda x: x[:19])
+
+    for l in lines_sorted:
+        ts = extract_timestamp_from_log(l)
+        status_code = extract_status_code_from_log(l)
+        duration = extract_duration_from_log(l)
+        path = extract_path_from_log(l)
+        referer = extract_referer_from_log(l)
+        request_id = extract_request_id_from_log(l)
+
+        duration_str = f"{duration}s" if duration is not None else "N/A"
+        print(
+            " - ".join(
+                [
+                    ts,
+                    f"status={status_code}",
+                    f"duration={duration_str}",
+                    f"path={path}",
+                    f"referer={referer}",
+                    f"id={request_id}",
+                ]
+            )
+        )