Fix gh --paginate mode; cache workflow files downloading

Author: dimikot

docker/ci-scaler/Dockerfile

@@ -11,7 +11,11 @@ ENV DEBIAN_FRONTEND=noninteractive
 RUN true \
     && apt-get update -y \
     && apt-get install -y --no-install-recommends \
-      awscli jq gh rsync python3 python3-yaml rsyslog systemctl tzdata gosu less mc git curl wget pv psmisc unzip vim nano telnet net-tools apt-transport-https ca-certificates locales gnupg lsb-release \
+      awscli jq rsync python3 python3-yaml rsyslog systemctl tzdata gosu less mc git curl wget pv psmisc unzip vim nano telnet net-tools apt-transport-https ca-certificates locales gnupg lsb-release \
+    && curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | gpg --dearmor -o /usr/share/keyrings/githubcli-archive-keyring.gpg \
+    && echo "deb [signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main" > /etc/apt/sources.list.d/github-cli.list \
+    && apt-get update -y \
+    && apt-get install -y --no-install-recommends gh \
     && useradd -m guest -s /bin/bash \
     && mkdir ~guest/.ssh && chmod 700 ~guest/.ssh \
     && chown -R guest:guest ~guest

docker/ci-scaler/guest/scaler/api_gh.py

@@ -2,8 +2,11 @@
 import hashlib
 import json
 import os
+import random
+import shlex
 import subprocess
 import time
+import traceback
 import yaml
 from helpers import Runner, RateLimits, check_output
 from typing import Any, cast
@@ -20,26 +23,35 @@ def gh_api(
     *args: str,
     input: Any | None = None,
 ) -> Any:
-    args = ("-H", "Accept: application/vnd.github.v3+json", *args)
-    res = gh(
+    args = (
         "api",
         *(["--input=-"] if input else []),
+        "-H",
+        "Accept: application/vnd.github.v3+json",
         *args,
-        input=json.dumps(input) if input is not None else None,
     )
-    return json.loads(res.strip()) if res != "" else None
+    res = None
+    try:
+        res = gh(
+            *args,
+            input=json.dumps(input) if input is not None else None,
+        ).strip()
+        return json.loads(res) if res else None
+    except Exception:
+        with open(f"/tmp/gh_api_error.{random.randint(0, 9)}.txt", "w") as f:
+            f.write(f"$ gh {shlex.join(args).strip()}\n")
+            f.write(f"{traceback.format_exc().rstrip()}\n")
+            f.write(f"{res}\n")
+        raise
 
 
 def gh_fetch_runners(
     *,
     repository: str,
 ) -> list[Runner]:
-    res = gh_api(f"repos/{repository}/actions/runners", "--paginate")
-    if not isinstance(res, dict):
-        raise ValueError(f"gh api returned a non-object: {res}")
-    runners = cast(dict[str, Any], res).get("runners")
-    if not isinstance(runners, list):
-        raise ValueError(f'gh api returned a response with no "runners" key: {res}')
+    res = gh_api(f"repos/{repository}/actions/runners", "--paginate", "--slurp")
+    if not isinstance(res, list) or not res:
+        raise ValueError(f"gh api returned a non-list of pages: {res}")
     return [
         Runner(
             id=str(runner["id"]),
@@ -53,7 +65,8 @@ def gh_fetch_runners(
             ],
             loaded_at=int(time.time()),
         )
-        for runner in cast(list[dict[str, Any]], runners)
+        for page in cast(list[dict[str, Any]], res)
+        for runner in cast(list[dict[str, Any]], page["runners"])
     ]
 
 
@@ -114,7 +127,6 @@ def gh_webhook_ensure_absent(
 def gh_webhook_get_id(*, repository: str, url: str) -> str | None:
     ids: list[str] = gh_api(
         f"/repos/{repository}/hooks",
-        "--paginate",
         "--jq",
         f"[.[] | select(.config.url=={json.dumps(url)}) | .id]",
     )

docker/ci-scaler/guest/scaler/handler_webhooks.py

@@ -30,7 +30,8 @@
 
 
 DUPLICATED_EVENTS_TTL = 3600
-JOB_TIMING_TTL = 7200  # 2 hours to track job timing data
+JOB_TIMING_TTL = 3600 * 2
+WORKFLOW_TTL = 60 * 5
 WORKFLOW_RUN_EVENT = "workflow_run"
 WORKFLOW_JOB_EVENT = "workflow_job"
 IGNORE_KEYS = [
@@ -78,6 +79,7 @@ def __init__(self, *, domain: str, asg_specs: list[AsgSpec]):
             ttl=DUPLICATED_EVENTS_TTL
         )
         self.job_timings = ExpiringDict[int, JobTiming](ttl=JOB_TIMING_TTL)
+        self.workflows = ExpiringDict[str, dict[str, Any]](ttl=WORKFLOW_TTL)
         this = self
 
         class RequestHandler(PostJsonHttpRequestHandler):
@@ -225,15 +227,24 @@ def handle(
 
             head_sha = str(run_payload["head_sha"])
             path = str(run_payload["path"])
-            message = f"{repository}: downloading {os.path.basename(path)} and parsing jobs list..."
+            message = f"{repository}{event_key}: downloading {os.path.basename(path)} and parsing jobs list"
             try:
-                workflow = gh_fetch_workflow(
-                    repository=repository,
-                    sha=head_sha,
-                    path=path,
-                )
+                cache_key = f"{repository}:{path}"
+                workflow = self.workflows.get(cache_key, None)
+                if not workflow:
+                    workflow = gh_fetch_workflow(
+                        repository=repository,
+                        sha=head_sha,
+                        path=path,
+                    )
+                    self.workflows[cache_key] = workflow
+                else:
+                    message += f" (cached)"
                 labels = gh_predict_workflow_labels(workflow=workflow)
-                log(f"{message} " + " ".join([f"{k}:+{v}" for k, v in labels.items()]))
+                log(
+                    f"{message}... "
+                    + " ".join([f"{k}:+{v}" for k, v in labels.items()])
+                )
             except Exception as e:
                 return handler.send_error(500, f"{message} failed: {e}")
 

docker/ci-scaler/guest/scaler/main.py

@@ -69,7 +69,7 @@ def main():
     parser.add_argument(
         "--poll-interval-sec",
         type=int,
-        default=10,
+        default=30,
         help="poll for the list of runners that often; it also determines the interval for publishing CloudWatch metrics",
     )
     parser.add_argument(