[ci][micro/4] support async query of test results (ray-project#45095)

- Support async query of test results. This speed up the query
significantly, especially for long test history
- Increase the test history of microcheck coverage to 500. 100 is about
one week of data which is too short for activities in the ray
repository.

Test:
- CI
- microcheck jobs:
https://buildkite.com/ray-project/postmerge/builds/4293#_

---------

Signed-off-by: can <[email protected]>

ci/ray_ci/automation/determine_microcheck_tests.py

@@ -14,7 +14,7 @@
 @click.command()
 @click.argument("team", required=True, type=str)
 @click.argument("coverage", required=True, type=int)
-@click.option("--test-history-length", default=100, type=int)
+@click.option("--test-history-length", default=500, type=int)
 @click.option("--test-prefix", default=LINUX_TEST_PREFIX, type=str)
 @click.option("--production", is_flag=True, default=False)
 def main(

release/BUILD.bazel

@@ -290,6 +290,7 @@ py_library(
         bk_require("anyscale"),
         bk_require("aws-requests-auth"),
         bk_require("bazel-runfiles"),
+        bk_require("aioboto3"),
         bk_require("boto3"),
         bk_require("botocore"),
         bk_require("click"),

release/ray_release/test.py

@@ -1,12 +1,15 @@
+import asyncio
+import concurrent.futures
 import enum
 import os
 import platform
 import json
 import time
 from itertools import chain
-from typing import Optional, List, Dict
+from typing import Awaitable, Optional, List, Dict
 from dataclasses import dataclass
 
+import aioboto3
 import boto3
 from botocore.exceptions import ClientError
 from github import Repository
@@ -40,6 +43,8 @@
 WINDOWS_BISECT_DAILY_RATE_LIMIT = 0  # windows bisect is disabled
 BISECT_DAILY_RATE_LIMIT = 10
 
+_asyncio_thread_pool = concurrent.futures.ThreadPoolExecutor()
+
 
 def _convert_env_list_to_dict(env_list: List[str]) -> Dict[str, str]:
     env_dict = {}
@@ -464,22 +469,36 @@ def get_test_results(
             key=lambda file: int(file["LastModified"].timestamp()),
             reverse=True,
         )[:limit]
-        self.test_results = [
-            TestResult.from_dict(
-                json.loads(
-                    s3_client.get_object(
-                        Bucket=bucket,
-                        Key=file["Key"],
-                    )
-                    .get("Body")
-                    .read()
-                    .decode("utf-8")
-                )
+        self.test_results = _asyncio_thread_pool.submit(
+            lambda: asyncio.run(
+                self._gen_test_results(bucket, [file["Key"] for file in files])
             )
-            for file in files
-        ]
+        ).result()
+
         return self.test_results
 
+    async def _gen_test_results(
+        self,
+        bucket: str,
+        keys: List[str],
+    ) -> Awaitable[List[TestResult]]:
+        session = aioboto3.Session()
+        async with session.client("s3") as s3_client:
+            return await asyncio.gather(
+                *[self._gen_test_result(s3_client, bucket, key) for key in keys]
+            )
+
+    async def _gen_test_result(
+        self,
+        s3_client: aioboto3.Session.client,
+        bucket: str,
+        key: str,
+    ) -> Awaitable[TestResult]:
+        object = await s3_client.get_object(Bucket=bucket, Key=key)
+        object_body = await object["Body"].read()
+
+        return TestResult.from_dict(json.loads(object_body.decode("utf-8")))
+
     def persist_result_to_s3(self, result: Result) -> bool:
         """
         Persist result object to s3

release/ray_release/tests/test_test.py

@@ -1,12 +1,14 @@
+import asyncio
 import json
 import sys
 import os
 import platform
 from unittest import mock
 
+import aioboto3
 import boto3
 import pytest
-from unittest.mock import patch
+from unittest.mock import patch, AsyncMock
 
 from ray_release.bazel import bazel_runfile
 from ray_release.configs.global_config import (
@@ -18,6 +20,7 @@
     TestResult,
     TestState,
     TestType,
+    ResultStatus,
     _convert_env_list_to_dict,
     DATAPLANE_ECR_REPO,
     DATAPLANE_ECR_ML_REPO,
@@ -42,6 +45,17 @@ def _stub_test(val: dict) -> Test:
     return test
 
 
+def _stub_test_result(status: ResultStatus) -> TestResult:
+    return TestResult(
+        status=status.value,
+        commit="1234567890",
+        branch="master",
+        url="url",
+        timestamp=0,
+        pull_request="1",
+    )
+
+
 def test_is_byod_cluster():
     assert not _stub_test({}).is_byod_cluster()
     assert _stub_test({"cluster": {"byod": {}}}).is_byod_cluster()
@@ -264,5 +278,34 @@ def test_is_high_impact() -> None:
     assert not _stub_test({"name": "test"}).is_high_impact()
 
 
+@patch("ray_release.test.Test._gen_test_result")
+def test_gen_test_results(mock_gen_test_result) -> None:
+    def _mock_gen_test_result(
+        client: aioboto3.Session.client,
+        bucket: str,
+        key: str,
+    ) -> TestResult:
+        return (
+            _stub_test_result(ResultStatus.SUCCESS)
+            if key == "good"
+            else _stub_test_result(ResultStatus.ERROR)
+        )
+
+    mock_gen_test_result.side_effect = AsyncMock(side_effect=_mock_gen_test_result)
+
+    results = asyncio.run(
+        _stub_test({})._gen_test_results(
+            bucket="bucket",
+            keys=["good", "bad", "bad", "good"],
+        )
+    )
+    assert [result.status for result in results] == [
+        ResultStatus.SUCCESS.value,
+        ResultStatus.ERROR.value,
+        ResultStatus.ERROR.value,
+        ResultStatus.SUCCESS.value,
+    ]
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-v", __file__]))

release/requirements_buildkite.in

@@ -2,6 +2,7 @@
 # Copy anyscale pin to requirements.txt and util.py
 anyscale
 bazel-runfiles
+aioboto3
 boto3
 click
 freezegun