Update

[ghstack-poisoned]

Author: huydhn

aws/lambda/gha-log-uploader/README.md

@@ -31,13 +31,29 @@ Payload:
 `conclusion` is optional. A malformed payload raises, which means Lambda retries
 twice and then DLQs it.
 
-## What it does not do
+## Classification
 
-It does not ping the log classifier. An S3 `ObjectCreated` notification on the
-`log/` prefix invokes `call-log-classifier`, which invokes `log_classifier`. That
-split is deliberate: the old lambda called the classifier with an untimed
+After a log is stored, `log_classifier` is invoked with `InvocationType: "Event"`
+and the result is not awaited. `github-status-test` called it with an untimed
 `urlopen` and blocked until classification finished, which is what produced its
-274s/344s/400s/900s duration tails.
+274s/344s/400s/900s duration tails — the fix is the invocation type, not a
+separate function.
+
+It is reached through `lambda:InvokeFunction` rather than its public function URL
+(`AuthType: NONE`), so the path from here to classification never crosses a
+public endpoint. `log_classifier` builds on `lambda_http` with only the
+`apigw_http` feature, so `classifier_payload()` reproduces an API Gateway HTTP API
+v2.0 request. Verified against the deployed function: a payload with no `job_id`
+returns its 400 "no job id provided" branch, and a non-numeric one fails inside
+its `parse::<usize>()`, which together show both the envelope and the query
+string are read from a direct invoke.
+
+A failed handoff is logged and reported as `classified: false`, not raised.
+Raising would make Lambda retry the whole function, re-downloading a
+multi-megabyte log from GitHub to retry something that takes milliseconds; the
+log itself is already safe in S3.
+
+## What it does not do
 
 It does not archive raw webhook payloads. Nothing read them —
 `clickhouse-replicator-s3` has no `SUPPORTED_PATHS` entry for `workflow_job/`,
@@ -88,8 +104,10 @@ Not done by CI. Needed before the deploy workflow can run.
 1. Create the function: python3.12, x86_64, handler `lambda_function.lambda_handler`.
    512 MB and a 60s timeout are plenty — the old function averaged 200ms and its
    long tail was the classifier ping this one does not make.
-2. Give its execution role `s3:PutObject` on `arn:aws:s3:::ossci-raw-job-status/log/*`
-   plus the usual CloudWatch Logs permissions.
+2. Give its execution role `s3:PutObject` on `arn:aws:s3:::ossci-raw-job-status/log/*`,
+   `lambda:InvokeFunction` on
+   `arn:aws:lambda:us-east-1:308535385114:function:log_classifier`, plus the
+   usual CloudWatch Logs permissions.
 3. Set the env vars above. Prefer fresh credentials over copying
    `github-status-test`'s, whose PATs sit in plaintext env vars and are due for
    rotation.
@@ -105,7 +123,9 @@ Not done by CI. Needed before the deploy workflow can run.
    `OUR_AWS_ACCESS_KEY_ID` before granting.
 6. Create the `gha_workflow_gha-log-uploader-lambda` IAM role the deploy workflow
    assumes, mirroring `gha_workflow_github-status-test-lambda`.
-7. Wire the classifier notification — see `../call-log-classifier/README.md`.
+7. Nothing to wire for classification: this function invokes `log_classifier`
+   directly, so there is no S3 notification to add. `keep-going-call-log-classifier`
+   still covers the separate `temp_logs/` prefix on `gha-artifacts`.
 
 ## Deployment
 

aws/lambda/gha-log-uploader/lambda_function.py

@@ -1,20 +1,22 @@
 # Copyright (c) 2019-present, Facebook, Inc.
 
-"""Download a completed GitHub Actions job log and archive it to S3.
+"""Download a completed GitHub Actions job log, archive it to S3, and classify it.
 
 Invoked asynchronously (``InvocationType: "Event"``) by the PyTorch bot's
 ``workflow_job`` handler in torchci, and by torchci's backfill route. There is no
 API Gateway integration and no Lambda function URL: the only way in is
 ``lambda:InvokeFunction``, which is IAM-authenticated.
 
-Classification is *not* triggered from here. An S3 ObjectCreated notification on
-the ``log/`` prefix drives it, so any path that lands a log gets classified and
-this function never blocks on the classifier finishing.
+The classifier is invoked the same way, asynchronously, so this function never
+blocks on classification finishing. ``github-status-test`` called it with an
+untimed ``urlopen`` and waited, which is what produced its 274s/344s/400s/900s
+duration tails -- the fix is the invocation type, not a separate function.
 """
 
 import base64
 import contextlib
 import gzip
+import json
 import os
 import random
 import time
@@ -26,11 +28,13 @@
 
 
 s3 = boto3.resource("s3")
+lambda_client = boto3.client("lambda")
 GITHUB_TOKENS = os.environ.get("GITHUB_TOKENS")
 GITHUB_APP_ID = os.environ.get("GITHUB_APP_ID")
 # Base64-encoded PEM, the same encoding torchci uses for its app key
 GITHUB_APP_PRIVATE_KEY = os.environ.get("GITHUB_APP_PRIVATE_KEY")
 BUCKET_NAME = "ossci-raw-job-status"
+LOG_CLASSIFIER_FUNCTION = "log_classifier"
 
 GITHUB_API_URL = "https://api.github.com"
 # Installation tokens last an hour. Refresh early so a warm invocation never
@@ -131,6 +135,66 @@ def log_object_path(full_name, job_id):
     return f"log/{full_name}/{job_id}"
 
 
+def classifier_payload(full_name, job_id):
+    """An API Gateway HTTP API v2.0 request, which is what lambda_http parses.
+
+    log_classifier is built on lambda_http with only the `apigw_http` feature, so
+    it expects this envelope even on a direct invoke. Verified against the
+    deployed function: a payload with no `job_id` returns its 400 "no job id
+    provided" branch, and a non-numeric one fails in its `parse::<usize>()`,
+    which together show both the envelope and the query string are read.
+    """
+    return {
+        "version": "2.0",
+        "routeKey": "$default",
+        "rawPath": "/",
+        "rawQueryString": f"job_id={job_id}&repo={full_name}",
+        "headers": {},
+        "queryStringParameters": {"job_id": str(job_id), "repo": full_name},
+        "requestContext": {
+            "accountId": "308535385114",
+            "apiId": "gha-log-uploader",
+            "domainName": "lambda-invoke",
+            "domainPrefix": "lambda-invoke",
+            "http": {
+                "method": "GET",
+                "path": "/",
+                "protocol": "HTTP/1.1",
+                "sourceIp": "127.0.0.1",
+                "userAgent": "gha-log-uploader",
+            },
+            "requestId": f"gha-log-uploader-{job_id}",
+            "routeKey": "$default",
+            "stage": "$default",
+            "time": "01/Jan/1970:00:00:00 +0000",
+            "timeEpoch": 0,
+        },
+        "isBase64Encoded": False,
+    }
+
+
+def classify_log(full_name, job_id):
+    """Kick off classification for a log we just stored. Returns True on handoff.
+
+    Asynchronous, and reached through `lambda:InvokeFunction` rather than
+    log_classifier's public function URL, so the path from here to classification
+    never crosses a public endpoint.
+    """
+    try:
+        lambda_client.invoke(
+            FunctionName=LOG_CLASSIFIER_FUNCTION,
+            InvocationType="Event",
+            Payload=json.dumps(classifier_payload(full_name, job_id)).encode(),
+        )
+        return True
+    except Exception as err:
+        # Best effort, deliberately. Raising would make Lambda retry the whole
+        # function, re-downloading a multi-megabyte log from GitHub to retry a
+        # handoff that takes milliseconds. The log itself is already safe in S3.
+        print(f"ERROR invoking the classifier for {full_name} job {job_id}: {err}")
+        return False
+
+
 def download_log(full_name, conclusion, job_id):
     """Fetch a job log from GitHub and archive it. Returns True when stored."""
     response = None
@@ -204,4 +268,11 @@ def lambda_handler(event, context):
         print(f"ERROR downloading log for {full_name} job {job_id}: {err}")
         raise
 
-    return {"repo": full_name, "job_id": job_id, "stored": stored}
+    classified = classify_log(full_name, job_id) if stored else False
+
+    return {
+        "repo": full_name,
+        "job_id": job_id,
+        "stored": stored,
+        "classified": classified,
+    }

aws/lambda/gha-log-uploader/test_lambda_function.py

@@ -1,8 +1,15 @@
+import json
 import unittest
 from unittest.mock import MagicMock, patch
 
 import lambda_function
-from lambda_function import download_log, installation_token, parse_event
+from lambda_function import (
+    classifier_payload,
+    classify_log,
+    download_log,
+    installation_token,
+    parse_event,
+)
 
 
 def make_response(status_code, content=b"log data", headers=None):
@@ -49,6 +56,46 @@ def test_rejects_a_non_object_payload(self):
             parse_event(["pytorch/pytorch", 123])
 
 
+class TestClassifierPayload(unittest.TestCase):
+    def test_is_a_v2_request_the_classifier_can_parse(self):
+        payload = classifier_payload("pytorch/executorch", 999)
+        # log_classifier builds on lambda_http with only the apigw_http feature,
+        # so version 2.0 and requestContext.http are what make it deserialize.
+        self.assertEqual(payload["version"], "2.0")
+        self.assertIn("http", payload["requestContext"])
+        self.assertEqual(
+            payload["queryStringParameters"],
+            {"job_id": "999", "repo": "pytorch/executorch"},
+        )
+        self.assertEqual(payload["rawQueryString"], "job_id=999&repo=pytorch/executorch")
+
+    def test_is_json_serializable(self):
+        json.dumps(classifier_payload("pytorch/pytorch", 1))
+
+
+class TestClassifyLog(unittest.TestCase):
+    def test_invokes_the_classifier_asynchronously(self):
+        with patch.object(lambda_function, "lambda_client") as client:
+            self.assertTrue(classify_log("pytorch/pytorch", 123))
+
+        kwargs = client.invoke.call_args.kwargs
+        self.assertEqual(kwargs["FunctionName"], "log_classifier")
+        # Event, not RequestResponse: waiting on classification is exactly the
+        # mistake that gave github-status-test its multi-hundred-second tails.
+        self.assertEqual(kwargs["InvocationType"], "Event")
+        self.assertEqual(
+            json.loads(kwargs["Payload"])["queryStringParameters"],
+            {"job_id": "123", "repo": "pytorch/pytorch"},
+        )
+
+    def test_a_failed_invoke_is_reported_not_raised(self):
+        # Raising would make Lambda retry the whole function, re-downloading a
+        # multi-megabyte log to retry a handoff that takes milliseconds.
+        with patch.object(lambda_function, "lambda_client") as client:
+            client.invoke.side_effect = RuntimeError("throttled")
+            self.assertFalse(classify_log("pytorch/pytorch", 123))
+
+
 class TestInstallationToken(unittest.TestCase):
     def setUp(self):
         lambda_function._token_cache.clear()
@@ -240,15 +287,35 @@ def test_a_null_conclusion_is_stored_as_empty(self, s3):
 @patch.object(lambda_function, "s3")
 class TestLambdaHandler(unittest.TestCase):
     def test_returns_a_summary(self, s3):
-        with patch.object(lambda_function, "download_log", return_value=True):
+        with patch.object(
+            lambda_function, "download_log", return_value=True
+        ), patch.object(lambda_function, "classify_log", return_value=True):
             self.assertEqual(
                 lambda_function.lambda_handler(
                     {"repo": "pytorch/pytorch", "job_id": 5, "conclusion": "success"},
                     None,
                 ),
-                {"repo": "pytorch/pytorch", "job_id": 5, "stored": True},
+                {
+                    "repo": "pytorch/pytorch",
+                    "job_id": 5,
+                    "stored": True,
+                    "classified": True,
+                },
             )
 
+    def test_does_not_classify_a_log_that_was_never_stored(self, s3):
+        # A 404 from GitHub means there is nothing in S3 for the classifier to
+        # read, so asking it to try would only produce a confusing failure.
+        with patch.object(
+            lambda_function, "download_log", return_value=False
+        ), patch.object(lambda_function, "classify_log") as classify:
+            result = lambda_function.lambda_handler(
+                {"repo": "pytorch/pytorch", "job_id": 5}, None
+            )
+
+        classify.assert_not_called()
+        self.assertFalse(result["classified"])
+
     def test_a_github_blip_propagates_so_lambda_retries(self, s3):
         with patch.object(
             lambda_function,