rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-09-06 05:28:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(ci): run query regression on ephemeral Aliyun ECS runners (#8937)

Browse Source 
* feat(ci): add aliyun ecs ephemeral runner path for query regression

Signed-off-by: paomian <xpaomian@gmail.com>

* fix: improve condition for query-regression job execution in workflow

* feat: update Docker installation to use official repository and add GPG key handling

* Refactor query regression runner setup and configuration

- Removed deprecated PersistentVolumeClaim for build cache.
- Introduced a new bootstrap script for setting up the ECS runner host.
- Deleted obsolete Helm values files for runner configuration.
- Updated the Aliyun ECS runner provisioning script to reflect new cache paths.
- Modified GitHub workflows to use the new Aliyun ECS runner setup.
- Adjusted documentation to clarify the new runner lifecycle and provisioning process.

* fix: enhance runner service management during bootstrap process

* fix: update alibabacloud_tea_openapi dependency version in metadata

* feat: enhance ECS runner scripts with region_id and resource_group_id support

* fix: move containerd content store to data root for improved storage management

* feat: rename query-regression runner to ephemeral-github runner and update related scripts

* fix: update sentinel polling method to use serial console output for improved reliability

* fix: add environment variable checks for Alibaba Cloud access keys in ECS client

* fix: improve error handling in GitHub API requests for better diagnostics

* fix: improve cache disk detection logic for Aliyun ECS instances

* fix: enhance cache disk waiting logic with detailed output and error handling

* fix: update dependency version for alibabacloud_tea_openapi in teardown script

* fix: enhance cache disk waiting logic for better compatibility and clarity

* fix: enhance console output handling and add incremental logging during instance provisioning

* fix: add PATH environment variable for runner jobs in service and provision script

* fix: add machine telemetry sampling and logging during query regression jobs

* fix: update query regression documentation and provision script for cache disk handling

* fix: update SCCACHE_CACHE_SIZE validation to 10G for improved caching efficiency

* fix: remove outdated cache size checks and cleanup logic for fresh system disk runs

* fix: enhance instance deletion logic with region handling and console output export

* fix: add swap file setup and OOM handling for ECS runner to improve stability

* fix: update OOM handling and service restart logic for ECS runner to enhance stability

* fix: increase system disk size to 100 GiB for cold double nightly builds to prevent ENOSPC errors

* fix: increase system disk size to 150 GiB for ECS runner to prevent ENOSPC errors

* fix: add keep_instance option to preserve ECS instance for post-mortem debugging

* fix: disable unattended upgrades to prevent job cancellations during library updates

* fix: reduce system disk size to 40 GiB for ECS runner to prevent ENOSPC errors

* feat: Refactor Aliyun ECS runner provisioning and introduce nightly regression comparison

- Update `aliyun-ecs-runner-provision.py` to remove cache disk handling, simplifying the provisioning process.
- Introduce `query-regression-nightly-refs.py` to resolve and compare SHAs from successful nightly builds.
- Create `query-regression-nightly.yml` workflow to trigger nightly comparisons based on successful builds.
- Enhance `query-regression.yml` to include a `test-tooling` job for validating Python scripts before provisioning.
- Update tests for the new nightly reference selection logic and refactor existing tests to align with the new caching strategy.
- Modify documentation to reflect changes in caching and nightly comparison workflows.

* fix: enhance runner image tool verification with detailed checks

* fix: improve error handling in runner image tool verification

* fix: update tool versions in ECS image and workflow for consistency

* fix: correct typo in error message for unparseable ECS creation time

* fix: update README and workflow files for query regression tests and image hygiene

---------

Signed-off-by: paomian <xpaomian@gmail.com>
This commit is contained in:
localhost
2026-08-26 12:11:14 +00:00
committed by GitHub
parent 144f83528d
commit 35ea88a4ef
20 changed files with 2538 additions and 591 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/scripts/aliyun-ecs-runner-provision.py
+485
View File
@@ -0,0 +1,485 @@
#!/usr/bin/env python3
# Copyright 2023 Greptime Team
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# This is the first `.github/scripts/` user of PEP 723 inline script metadata:
# `uv run` installs the pinned Aliyun SDK before executing. The SDK import stays
# lazy (inside `make_ecs_client`) so unit tests can import this module with a
# plain stdlib interpreter.
# /// script
# requires-python = ">=3.10"
# dependencies = [
# # Pin exactly after the first live run against the Aliyun API.
# "alibabacloud_ecs20140526>=4.1.0,<6",
# "alibabacloud_tea_openapi>=0.3.12,<1",
# ]
# ///
"""Provision an ephemeral Aliyun ECS query-regression runner.
Creates one pay-as-you-go ECS instance from the query-regression custom image
and waits until the instance registers itself as an ephemeral GitHub Actions
runner with a per-run label. Build caches live on the instance system disk
and disappear with the VM. The instance id is written to job outputs as soon
as the VM exists so the teardown job is the single DeleteInstance caller (a
timeout here used to delete and then teardown deleted again). On runner
online timeout the console is dumped and the job fails; teardown releases
the instance.
The Aliyun credentials are used only in this control-plane job (ubuntu-latest)
and are never passed to the ECS instance; the instance receives only a
short-lived GitHub runner registration token via user data.
"""
from __future__ import annotations
import argparse
import base64
import json
import os
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
RUNNER_NAME_PREFIX = "qreg-ecs"
RUNNER_LABEL_PREFIX = "query-regression-ecs"
MANAGED_BY_TAG_KEY = "managed-by"
MANAGED_BY_TAG_VALUE = "query-regression-ci"
RUN_TAG_KEY = "query-regression-run-id"
# Runner cache paths on the instance system disk. They are created empty
# every provision and deleted with the VM.
CACHE_PATHS = (
"/home/runner/.cargo/registry",
"/home/runner/.cargo/git",
"/home/runner/query-regression-target",
"/home/runner/query-regression-cache-meta",
"/home/runner/.cache/sccache",
)
RUNNER_ONLINE_TIMEOUT_SECONDS = 10 * 60
POLL_INTERVAL_SECONDS = 5
# Nightly thin-LTO of greptime peaks above ecs.c9i.2xlarge's 16 GiB. A swap
# file plus masking systemd-oomd lets the kernel reclaim rustc pages instead
# of SIGTERM-ing the runner service cgroup (GitHub then reports the step as
# "The operation was canceled"). Sized to match that 16 GiB instance; a
# 32 GiB type still benefits as a safety net. Lives on the system disk next
# to the cold build caches.
SWAP_FILE = "/swapfile"
SWAP_SIZE_GIB = 16
@dataclass(frozen=True)
class ProvisionConfig:
region_id: str
vswitch_id: str
security_group_id: str
image_id: str
instance_type: str
repo: str
run_id: str
github_token: str
# Optional resource group; required when the RAM grant is scoped to one.
resource_group_id: str | None = None
# Runner identity inside the image; the workflow asserts the same values.
runner_uid: str = "1001"
runner_gid: str = "1001"
def runner_name_for_run(run_id: str) -> str:
return f"{RUNNER_NAME_PREFIX}-{run_id}"
def runner_label_for_run(run_id: str) -> str:
return f"{RUNNER_LABEL_PREFIX}-{run_id}"
def render_user_data(
runner_name: str,
runner_label: str,
runner_token: str,
repo: str,
runner_uid: str = "1001",
runner_gid: str = "1001",
) -> str:
"""Render the cloud-init shell script for the runner instance."""
destinations = " ".join(f'"{dst}"' for dst in CACHE_PATHS)
cache_setup = f"""# Caches live on the system disk, are deleted with the instance, and every
# run compiles cold. Within-run reuse (base warming candidate via the shared
# target dir and sccache) still applies.
mkdir -p {destinations}
chown -R {runner_uid}:{runner_gid} /home/runner"""
swap_setup = f"""# Mask systemd-oomd before enabling swap: Ubuntu 24.04 kills the whole
# service cgroup on PSI pressure, and swap thrashing looks like pressure.
systemctl disable --now systemd-oomd.socket systemd-oomd.service || true
systemctl mask systemd-oomd.socket systemd-oomd.service || true
# apt-daily-upgrade / unattended-upgrades can `systemctl restart` services
# whose libraries were updated. That SIGTERM-s the runner mid-job; GitHub
# records UserCancelled (job still valid, child exit 143). These VMs live
# ~1h; security updates belong in the image, not at job time.
systemctl disable --now unattended-upgrades.service apt-daily.timer apt-daily-upgrade.timer apt-daily.service apt-daily-upgrade.service || true
systemctl mask unattended-upgrades.service apt-daily.timer apt-daily-upgrade.timer apt-daily.service apt-daily-upgrade.service || true
systemctl stop unattended-upgrades.service || true
killall -9 unattended-upgr || true
cat > /etc/apt/apt.conf.d/99disable-auto-updates <<'APTEOF'
APT::Periodic::Update-Package-Lists "0";
APT::Periodic::Unattended-Upgrade "0";
APT::Periodic::Download-Upgradeable-Packages "0";
APTEOF
if [[ ! -f "{SWAP_FILE}" ]]; then
fallocate --length {SWAP_SIZE_GIB}G "{SWAP_FILE}"
chmod 600 "{SWAP_FILE}"
mkswap "{SWAP_FILE}"
fi
swapon "{SWAP_FILE}"
sysctl --write vm.swappiness=10
swapon --show
free --human"""
return f"""#!/bin/bash
set -euo pipefail
{cache_setup}
{swap_setup}
cat > /etc/ephemeral-github-runner.env <<'ENVEOF'
RUNNER_NAME={runner_name}
RUNNER_LABELS={runner_label}
RUNNER_TOKEN={runner_token}
REPO_URL=https://github.com/{repo}
# The container image baked /opt/cargo/bin into PATH via Dockerfile ENV; on
# the host nothing inherits that, so publish it through the unit's
# EnvironmentFile: runner job processes inherit the runner's environment.
PATH=/opt/cargo/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
ENVEOF
# Stream the runner service's own logs (config output, job lifecycle, crash
# messages) to the serial console, next to the kernel's OOM-killer records.
# Teardown dumps the console tail before deleting the instance, so the
# witness of what killed the runner survives the machine.
mkdir -p /etc/systemd/system/ephemeral-github-runner.service.d
cat > /etc/systemd/system/ephemeral-github-runner.service.d/console.conf <<'CONFEOF'
[Service]
StandardOutput=journal+console
StandardError=journal+console
CONFEOF
# systemd's DefaultOOMPolicy=stop SIGTERM-s the whole unit when *any*
# process in the cgroup is OOM-killed (typically rustc during thin LTO).
# That is the "Session terminated, killing shell..." / job Canceled path:
# runuser dies, GitHub deletes the ephemeral registration, always() steps
# never run. continue keeps the runner up so cargo can report signal 9.
cat > /etc/systemd/system/ephemeral-github-runner.service.d/oom.conf <<'CONFEOF'
[Service]
OOMPolicy=continue
CONFEOF
systemctl daemon-reload
# restart (not start): the image enables this unit, so it may already
# have come up without the drop-ins if it raced cloud-init. restart
# applies OOMPolicy=continue to the running cgroup. --no-block matters:
# newer image units carry After=cloud-final.service, and this script IS
# cloud-final — a blocking restart would deadlock until TimeoutStartSec.
systemctl restart --no-block ephemeral-github-runner.service
"""
def encode_user_data(script: str) -> str:
return base64.b64encode(script.encode("utf-8")).decode("ascii")
def github_api(token: str, method: str, path: str, body: dict | None = None) -> dict:
data = json.dumps(body).encode("utf-8") if body is not None else None
request = urllib.request.Request(
f"https://api.github.com{path}",
data=data,
method=method,
headers={
"Accept": "application/vnd.github+json",
"Authorization": f"Bearer {token}",
"X-GitHub-Api-Version": "2022-11-28",
},
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as error:
# GitHub's error body says exactly why (e.g. "Must have admin rights to
# Repository" for a PAT without the required scope); surface it instead
# of a bare "HTTP Error 403".
body = error.read().decode("utf-8", "replace")
raise SystemExit(
f"GitHub API {method} {path} failed: HTTP {error.code}: {body}\n"
"The token comes from the GH_PERSONAL_ACCESS_TOKEN secret; it needs "
"'repo' scope (classic PAT) or 'Administration: write' on the "
"repository (fine-grained PAT)."
) from error
def create_registration_token(github_token: str, repo: str) -> str:
response = github_api(
github_token,
"POST",
f"/repos/{repo}/actions/runners/registration-token",
body={},
)
return response["token"]
def find_runner_by_name(github_token: str, repo: str, name: str) -> dict | None:
page = 1
while True:
response = github_api(
github_token,
"GET",
f"/repos/{repo}/actions/runners?per_page=100&page={page}",
)
runners = response.get("runners", [])
for runner in runners:
if runner.get("name") == name:
return runner
if len(runners) < 100:
return None
page += 1
def make_ecs_client(config: ProvisionConfig):
from alibabacloud_ecs20140526.client import Client as EcsClient
from alibabacloud_tea_openapi.models import Config as OpenApiConfig
access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID", "")
access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET", "")
if not access_key_id or not access_key_secret:
raise SystemExit(
"ALIBABA_CLOUD_ACCESS_KEY_ID/SECRET are empty; in CI they come from the "
"ALICLOUD_ECS_ACCESS_KEY_ID/SECRET repository secrets (a missing or "
"misnamed secret expands to an empty string)."
)
return EcsClient(
OpenApiConfig(
access_key_id=access_key_id,
access_key_secret=access_key_secret,
region_id=config.region_id,
endpoint=f"ecs.{config.region_id}.aliyuncs.com",
)
)
def wait_for_instance_status(client, region_id: str, instance_id: str, wanted: str, deadline: float) -> None:
from alibabacloud_ecs20140526 import models as ecs_models
while time.monotonic() < deadline:
response = client.describe_instances(
ecs_models.DescribeInstancesRequest(
region_id=region_id, instance_ids=json.dumps([instance_id])
)
)
instances = response.body.instances.instance
if instances and instances[0].status == wanted:
return
time.sleep(POLL_INTERVAL_SECONDS)
raise TimeoutError(f"Instance {instance_id} did not reach status {wanted} in time")
def fetch_console_output(client, region_id: str, instance_id: str) -> str:
from alibabacloud_ecs20140526 import models as ecs_models
response = client.get_instance_console_output(
ecs_models.GetInstanceConsoleOutputRequest(region_id=region_id, instance_id=instance_id)
)
return base64.b64decode(response.body.console_output or "").decode("utf-8", "replace")
class ConsoleTailer:
"""Incrementally prints new serial-console lines so cloud-init progress is
visible in the CI log while waiting, not only after a failure dump."""
def __init__(self, client, region_id: str, instance_id: str) -> None:
self.client = client
self.region_id = region_id
self.instance_id = instance_id
self.printed_lines = 0
def poll(self) -> None:
try:
lines = fetch_console_output(self.client, self.region_id, self.instance_id).splitlines()
except Exception as error: # noqa: BLE001
print(f"[console] unable to fetch console output: {error}", flush=True)
return
# The API returns a bounded tail; if it ever shrinks, restart from the
# beginning of the new buffer rather than skipping lines.
if len(lines) < self.printed_lines:
self.printed_lines = 0
for line in lines[self.printed_lines :]:
print(f"[console] {line}", flush=True)
self.printed_lines = len(lines)
def dump_console_output(client, region_id: str, instance_id: str) -> None:
try:
output = fetch_console_output(client, region_id, instance_id)
except Exception as error: # noqa: BLE001
print(f"Unable to fetch console output for {instance_id}: {error}", flush=True)
return
tail = "\n".join(output.splitlines()[-80:])
print(f"::group::Console output tail for {instance_id}\n{tail}\n::endgroup::", flush=True)
summary_path = os.environ.get("GITHUB_STEP_SUMMARY")
if summary_path:
with open(summary_path, "a", encoding="utf-8") as summary:
summary.write(f"<details><summary>ECS console output ({instance_id})</summary>\n\n```\n")
summary.write(tail)
summary.write("\n```\n</details>\n")
def append_github_output(name: str, value: str) -> None:
output_path = os.environ.get("GITHUB_OUTPUT")
if output_path:
with open(output_path, "a", encoding="utf-8") as output:
output.write(f"{name}={value}\n")
def run_instance(client, config: ProvisionConfig, user_data: str) -> str:
from alibabacloud_ecs20140526 import models as ecs_models
request = ecs_models.RunInstancesRequest(
region_id=config.region_id,
image_id=config.image_id,
resource_group_id=config.resource_group_id,
instance_type=config.instance_type,
v_switch_id=config.vswitch_id,
security_group_id=config.security_group_id,
instance_name=runner_name_for_run(config.run_id),
host_name=runner_name_for_run(config.run_id),
description=f"Ephemeral query-regression runner for run {config.run_id}",
amount=1,
instance_charge_type="PostPaid",
spot_strategy="NoSpot",
internet_charge_type="PayByTraffic",
internet_max_bandwidth_out=100,
# Holds the image (~12G), the 16G swapfile, the source checkout, the
# base+candidate cluster data homes, and cold build caches. Deleted
# with the instance.
system_disk=ecs_models.RunInstancesRequestSystemDisk(category="cloud_essd", size="40"),
user_data=user_data,
tag=[
ecs_models.RunInstancesRequestTag(key=MANAGED_BY_TAG_KEY, value=MANAGED_BY_TAG_VALUE),
ecs_models.RunInstancesRequestTag(key=RUN_TAG_KEY, value=config.run_id),
],
)
response = client.run_instances(request)
instance_id = response.body.instance_id_sets.instance_id_set[0]
return instance_id
def provision(config: ProvisionConfig) -> int:
client = make_ecs_client(config)
runner_name = runner_name_for_run(config.run_id)
runner_label = runner_label_for_run(config.run_id)
registration_token = create_registration_token(config.github_token, config.repo)
user_data = encode_user_data(
render_user_data(
runner_name,
runner_label,
registration_token,
config.repo,
config.runner_uid,
config.runner_gid,
)
)
print(f"::group::Run instance {runner_name}", flush=True)
instance_id = run_instance(client, config, user_data)
print(f"Instance id: {instance_id}", flush=True)
# Teardown is the only DeleteInstance caller. Write outputs immediately
# so a later status/online failure still releases this VM.
append_github_output("label", runner_label)
append_github_output("instance_id", instance_id)
append_github_output("runner_name", runner_name)
wait_for_instance_status(client, config.region_id, instance_id, "Running", time.monotonic() + 5 * 60)
print("::endgroup::", flush=True)
print("::group::Wait for runner online", flush=True)
deadline = time.monotonic() + RUNNER_ONLINE_TIMEOUT_SECONDS
tailer = ConsoleTailer(client, config.region_id, instance_id)
last_console_poll = 0.0
while time.monotonic() < deadline:
runner = find_runner_by_name(config.github_token, config.repo, runner_name)
if runner is not None:
status = runner.get("status")
print(f"Runner {runner_name} status: {status}", flush=True)
if status == "online":
print("::endgroup::", flush=True)
print(f"Runner {runner_name} is online with label {runner_label}", flush=True)
return 0
# The serial console lags the guest by a minute or so; 30s polling is
# enough to follow cloud-init without tripping API throttling.
if time.monotonic() - last_console_poll >= 30:
tailer.poll()
last_console_poll = time.monotonic()
time.sleep(POLL_INTERVAL_SECONDS)
print("::endgroup::", flush=True)
print(
f"Runner {runner_name} did not come online in time; "
"leaving instance for the teardown job to delete",
flush=True,
)
dump_console_output(client, config.region_id, instance_id)
return 1
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--region-id", default=os.environ.get("ALIYUN_ECS_REGION_ID"))
parser.add_argument("--vswitch-id", default=os.environ.get("ALIYUN_ECS_VSWITCH_ID"))
parser.add_argument("--security-group-id", default=os.environ.get("ALIYUN_ECS_SECURITY_GROUP_ID"))
parser.add_argument("--image-id", default=os.environ.get("QUERY_REGRESSION_ECS_IMAGE_ID"))
parser.add_argument("--instance-type", default=os.environ.get("ALIYUN_ECS_INSTANCE_TYPE"))
parser.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY"))
parser.add_argument("--run-id", default=os.environ.get("GITHUB_RUN_ID"))
parser.add_argument("--github-token", default=os.environ.get("GH_PERSONAL_ACCESS_TOKEN"))
parser.add_argument("--runner-uid", default=os.environ.get("QUERY_REGRESSION_RUNNER_UID", "1001"))
parser.add_argument("--runner-gid", default=os.environ.get("QUERY_REGRESSION_RUNNER_GID", "1001"))
parser.add_argument("--resource-group-id", default=os.environ.get("ALIYUN_ECS_RESOURCE_GROUP_ID"))
args = parser.parse_args()
missing = [
name
for name, value in vars(args).items()
if name not in ("resource_group_id",)
and (value is None or (isinstance(value, str) and not value))
]
if missing:
flags = ", ".join(f"--{name.replace('_', '-')}" for name in missing)
raise SystemExit(f"Missing required configuration: {flags} (or their env defaults)")
config = ProvisionConfig(
region_id=args.region_id,
vswitch_id=args.vswitch_id,
security_group_id=args.security_group_id,
image_id=args.image_id,
instance_type=args.instance_type,
repo=args.repo,
run_id=args.run_id,
github_token=args.github_token,
resource_group_id=args.resource_group_id or None,
runner_uid=args.runner_uid,
runner_gid=args.runner_gid,
)
return provision(config)
if __name__ == "__main__":
raise SystemExit(main())
.github/scripts/aliyun-ecs-runner-teardown.py
+237
View File
@@ -0,0 +1,237 @@
#!/usr/bin/env python3
# Copyright 2023 Greptime Team
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# PEP 723 inline metadata (see aliyun-ecs-runner-provision.py for the
# convention). The SDK import stays lazy so unit tests run on a plain
# stdlib interpreter.
# /// script
# requires-python = ">=3.10"
# dependencies = [
# "alibabacloud_ecs20140526>=4.1.0,<6",
# "alibabacloud_tea_openapi>=0.3.12,<1",
# ]
# ///
"""Tear down ephemeral Aliyun ECS query-regression runners.
Two modes:
- Targeted (per workflow run): delete one instance by id and deregister its
runner by name. Both steps are idempotent and best-effort; a missing
instance or runner is not an error.
- Sweep (scheduled janitor): delete every instance tagged as managed by
query-regression CI whose creation time is older than the given TTL, and
deregister the matching runners. This is the safety net for runs whose
teardown job never executed.
"""
from __future__ import annotations
import argparse
import importlib.util
import os
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
# Reuse the GitHub API client, runner lookup, and tag constants from the
# provision script.
_PROVISION_SPEC = importlib.util.spec_from_file_location(
"aliyun_ecs_runner_provision",
Path(__file__).resolve().parent / "aliyun-ecs-runner-provision.py",
)
assert _PROVISION_SPEC is not None and _PROVISION_SPEC.loader is not None
provision = importlib.util.module_from_spec(_PROVISION_SPEC)
sys.modules[_PROVISION_SPEC.name] = provision
_PROVISION_SPEC.loader.exec_module(provision)
SWEEP_TTL = timedelta(hours=4)
# ECS DescribeInstances creation_time is ISO8601 UTC, e.g. "2026-08-17T02:13Z".
CREATION_TIME_FORMATS = ("%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%MZ")
def parse_creation_time(value: str) -> datetime:
for fmt in CREATION_TIME_FORMATS:
try:
return datetime.strptime(value, fmt).replace(tzinfo=timezone.utc)
except ValueError:
continue
raise ValueError(f"Unparsable ECS creation time: {value}")
def expired_instance_names(
instances: list[tuple[str, str, str]], now: datetime, ttl: timedelta
) -> list[tuple[str, str]]:
"""Pick (instance_id, instance_name) pairs whose creation is older than ttl.
`instances` items are (instance_id, instance_name, creation_time).
"""
expired = []
for instance_id, instance_name, creation_time in instances:
age = now - parse_creation_time(creation_time)
if age >= ttl:
expired.append((instance_id, instance_name))
return expired
def make_ecs_client(region_id: str):
from alibabacloud_ecs20140526.client import Client as EcsClient
from alibabacloud_tea_openapi.models import Config as OpenApiConfig
access_key_id = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID", "")
access_key_secret = os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET", "")
if not access_key_id or not access_key_secret:
raise SystemExit(
"ALIBABA_CLOUD_ACCESS_KEY_ID/SECRET are empty; in CI they come from the "
"ALICLOUD_ECS_ACCESS_KEY_ID/SECRET repository secrets (a missing or "
"misnamed secret expands to an empty string)."
)
return EcsClient(
OpenApiConfig(
access_key_id=access_key_id,
access_key_secret=access_key_secret,
region_id=region_id,
endpoint=f"ecs.{region_id}.aliyuncs.com",
)
)
def delete_instance(client, instance_id: str, region_id: str | None = None) -> bool:
import time
from alibabacloud_ecs20140526 import models as ecs_models
# Export the serial console before destroying the evidence: a cancelled
# run still runs this teardown job, and the console tail (cloud-init,
# runner service logs, kernel OOM records) is the only witness of what
# happened on the machine.
if region_id is not None:
provision.dump_console_output(client, region_id, instance_id)
# DeleteInstance rejects instances that are still Initializing (or in
# another transitional status). Teardown runs right after a cancelled
# run, when the instance may be only seconds old, so retry the transient
# status errors for a few minutes instead of leaking the instance.
deadline = time.monotonic() + 5 * 60
while True:
try:
client.delete_instance(ecs_models.DeleteInstanceRequest(instance_id=instance_id, force=True))
print(f"Deleted instance {instance_id}", flush=True)
return True
except Exception as error: # noqa: BLE001
message = str(error)
if "InvalidInstanceId.NotFound" in message:
print(f"Instance {instance_id} already gone", flush=True)
return True
if "IncorrectInstanceStatus" in message and time.monotonic() < deadline:
print(f"Instance {instance_id} is in a transitional status; retrying delete", flush=True)
time.sleep(15)
continue
print(f"Failed to delete instance {instance_id}: {error}", flush=True)
return False
def deregister_runner(token: str, repo: str, runner_name: str) -> bool:
runner = provision.find_runner_by_name(token, repo, runner_name)
if runner is None:
print(f"Runner {runner_name} is not registered", flush=True)
return True
try:
provision.github_api(token, "DELETE", f"/repos/{repo}/actions/runners/{runner['id']}")
print(f"Deregistered runner {runner_name} (id {runner['id']})", flush=True)
return True
except Exception as error: # noqa: BLE001
print(f"Failed to deregister runner {runner_name}: {error}", flush=True)
return False
def list_managed_instances(client, region_id: str) -> list[tuple[str, str, str]]:
from alibabacloud_ecs20140526 import models as ecs_models
result: list[tuple[str, str, str]] = []
next_token = None
while True:
request = ecs_models.DescribeInstancesRequest(
region_id=region_id,
tag=[
ecs_models.DescribeInstancesRequestTag(
key=provision.MANAGED_BY_TAG_KEY, value=provision.MANAGED_BY_TAG_VALUE
)
],
max_results=100,
next_token=next_token,
)
response = client.describe_instances(request)
for instance in response.body.instances.instance:
result.append((instance.instance_id, instance.instance_name, instance.creation_time))
next_token = response.body.next_token
if not next_token:
return result
def sweep(client, region_id: str, repo: str, github_token: str, ttl: timedelta) -> int:
instances = list_managed_instances(client, region_id)
print(f"Found {len(instances)} managed instance(s) in {region_id}", flush=True)
expired = expired_instance_names(instances, datetime.now(timezone.utc), ttl)
ok = True
for instance_id, instance_name in expired:
print(f"Instance {instance_id} ({instance_name}) exceeds TTL {ttl}; deleting", flush=True)
# Runner names mirror instance names by construction in the provision
# script (both are qreg-ecs-<run_id>).
ok &= delete_instance(client, instance_id, region_id)
ok &= deregister_runner(github_token, repo, instance_name)
return 0 if ok else 1
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--region-id", default=os.environ.get("ALIYUN_ECS_REGION_ID"))
parser.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY"))
parser.add_argument("--github-token", default=os.environ.get("GH_PERSONAL_ACCESS_TOKEN"))
parser.add_argument("--instance-id", default=os.environ.get("QUERY_REGRESSION_ECS_INSTANCE_ID"))
parser.add_argument("--runner-name", default=os.environ.get("QUERY_REGRESSION_ECS_RUNNER_NAME"))
parser.add_argument(
"--sweep",
action="store_true",
help="Janitor mode: delete all managed instances older than --sweep-ttl-hours.",
)
parser.add_argument("--sweep-ttl-hours", type=float, default=SWEEP_TTL.total_seconds() / 3600)
args = parser.parse_args()
for name in ("region_id", "repo", "github_token"):
if not getattr(args, name):
raise SystemExit(f"Missing required configuration: --{name.replace('_', '-')}")
client = make_ecs_client(args.region_id)
if args.sweep:
return sweep(
client, args.region_id, args.repo, args.github_token, timedelta(hours=args.sweep_ttl_hours)
)
ok = True
if args.instance_id:
ok &= delete_instance(client, args.instance_id, args.region_id)
else:
print("No instance id given; skipping instance deletion", flush=True)
if args.runner_name:
ok &= deregister_runner(args.github_token, args.repo, args.runner_name)
else:
print("No runner name given; skipping runner deregistration", flush=True)
return 0 if ok else 1
if __name__ == "__main__":
raise SystemExit(main())
.github/scripts/query-regression-nightly-refs.py
+270
View File
@@ -0,0 +1,270 @@
#!/usr/bin/env python3
# Copyright 2023 Greptime Team
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Resolve consecutive successful Nightly Build SHAs for query-regression.
Query regression builds both binaries from git; this script only picks the
head SHAs of two Nightly Build workflow runs (today vs the previous success).
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from typing import Any
NIGHTLY_WORKFLOW = "nightly-build.yml"
NIGHTLY_EVENTS = frozenset({"schedule", "workflow_dispatch"})
@dataclass(frozen=True)
class WorkflowRun:
id: int
head_sha: str
head_branch: str
html_url: str
created_at: str
conclusion: str
event: str
@dataclass(frozen=True)
class RefPair:
base: WorkflowRun | None
candidate: WorkflowRun | None
skip: bool
reason: str
def parse_run(payload: dict[str, Any]) -> WorkflowRun:
return WorkflowRun(
id=int(payload["id"]),
head_sha=str(payload["head_sha"]),
head_branch=str(payload.get("head_branch") or ""),
html_url=str(payload.get("html_url") or ""),
created_at=str(payload.get("created_at") or ""),
conclusion=str(payload.get("conclusion") or ""),
event=str(payload.get("event") or ""),
)
def github_get(token: str, path: str, query: dict[str, str] | None = None) -> dict[str, Any]:
encoded = f"?{urllib.parse.urlencode(query)}" if query else ""
request = urllib.request.Request(
f"https://api.github.com{path}{encoded}",
method="GET",
headers={
"Accept": "application/vnd.github+json",
"Authorization": f"Bearer {token}",
"X-GitHub-Api-Version": "2022-11-28",
},
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as error:
body = error.read().decode("utf-8", "replace")
raise SystemExit(
f"GitHub API GET {path} failed: HTTP {error.code}: {body}"
) from error
def list_successful_nightly_runs(
token: str,
repo: str,
*,
branch: str | None = None,
per_page: int = 30,
) -> list[WorkflowRun]:
query = {
"status": "success",
"per_page": str(per_page),
}
if branch:
query["branch"] = branch
payload = github_get(
token,
f"/repos/{repo}/actions/workflows/{NIGHTLY_WORKFLOW}/runs",
query,
)
runs = [parse_run(item) for item in payload.get("workflow_runs") or []]
return [
run
for run in runs
if run.conclusion == "success" and run.event in NIGHTLY_EVENTS and run.head_sha
]
def fetch_run(token: str, repo: str, run_id: int) -> WorkflowRun:
payload = github_get(token, f"/repos/{repo}/actions/runs/{run_id}")
return parse_run(payload)
def select_base_and_candidate(
runs: list[WorkflowRun],
*,
candidate_run_id: int | None = None,
candidate: WorkflowRun | None = None,
) -> RefPair:
"""Pick candidate (newest / requested) and the previous successful run.
`runs` must be newest-first. Same-SHA consecutive nightlies are skipped
(no commits that day).
"""
if candidate is None:
if candidate_run_id is not None:
candidate = next((run for run in runs if run.id == candidate_run_id), None)
if candidate is None:
return RefPair(
None,
None,
True,
f"candidate nightly run {candidate_run_id} was not a successful "
f"{NIGHTLY_WORKFLOW} run",
)
elif runs:
candidate = runs[0]
else:
return RefPair(None, None, True, "no successful Nightly Build runs")
branch = candidate.head_branch
older = [
run
for run in runs
if run.id != candidate.id
and (not branch or not run.head_branch or run.head_branch == branch)
and (not candidate.created_at or run.created_at <= candidate.created_at)
]
if not older:
return RefPair(
None,
candidate,
True,
"no previous successful Nightly Build to compare against",
)
base = older[0]
if base.head_sha.lower() == candidate.head_sha.lower():
return RefPair(
base,
candidate,
True,
f"previous nightly SHA {base.head_sha} matches candidate; nothing to compare",
)
return RefPair(base, candidate, False, "")
def _override_run(sha: str) -> WorkflowRun:
return WorkflowRun(
id=0,
head_sha=sha,
head_branch="",
html_url="",
created_at="",
conclusion="success",
event="workflow_dispatch",
)
def override_pair(base_ref: str, candidate_ref: str) -> RefPair:
return RefPair(_override_run(base_ref), _override_run(candidate_ref), False, "explicit refs")
def write_outputs(pair: RefPair) -> None:
values = {
"skip": "true" if pair.skip else "false",
"reason": pair.reason,
"base_sha": pair.base.head_sha if pair.base else "",
"candidate_sha": pair.candidate.head_sha if pair.candidate else "",
"base_run_url": pair.base.html_url if pair.base else "",
"candidate_run_url": pair.candidate.html_url if pair.candidate else "",
"base_run_id": str(pair.base.id) if pair.base and pair.base.id else "",
"candidate_run_id": str(pair.candidate.id) if pair.candidate and pair.candidate.id else "",
}
output_path = os.environ.get("GITHUB_OUTPUT")
if output_path:
with open(output_path, "a", encoding="utf-8") as handle:
for key, value in values.items():
handle.write(f"{key}={value}\n")
for key, value in values.items():
print(f"{key}={value}")
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--repo", default=os.environ.get("GITHUB_REPOSITORY"))
parser.add_argument(
"--token",
default=os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN"),
)
parser.add_argument("--branch", default=os.environ.get("NIGHTLY_BRANCH") or "")
parser.add_argument(
"--candidate-run-id",
default=os.environ.get("CANDIDATE_RUN_ID") or "",
)
parser.add_argument("--base-ref", default=os.environ.get("BASE_REF") or "")
parser.add_argument("--candidate-ref", default=os.environ.get("CANDIDATE_REF") or "")
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
base_ref = args.base_ref.strip()
candidate_ref = args.candidate_ref.strip()
if base_ref and candidate_ref:
write_outputs(override_pair(base_ref, candidate_ref))
return 0
if bool(base_ref) != bool(candidate_ref):
print("Provide both --base-ref and --candidate-ref, or neither.", file=sys.stderr)
return 2
if not args.repo or not args.token:
print("--repo and --token (or GITHUB_REPOSITORY / GITHUB_TOKEN) are required.", file=sys.stderr)
return 2
candidate_run_id = int(args.candidate_run_id) if str(args.candidate_run_id).strip() else None
candidate: WorkflowRun | None = None
branch = args.branch.strip() or None
if candidate_run_id is not None:
candidate = fetch_run(args.token, args.repo, candidate_run_id)
if candidate.conclusion != "success":
write_outputs(
RefPair(
None,
candidate,
True,
f"nightly run {candidate_run_id} conclusion is {candidate.conclusion}",
)
)
return 0
branch = branch or candidate.head_branch or None
runs = list_successful_nightly_runs(args.token, args.repo, branch=branch)
pair = select_base_and_candidate(
runs,
candidate_run_id=candidate_run_id if candidate is None else None,
candidate=candidate,
)
write_outputs(pair)
return 0
if __name__ == "__main__":
raise SystemExit(main())