mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-13 23:42:32 +00:00
ci: add optional AWS runners for observability benchmarks
Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
@@ -49,10 +49,10 @@ import base64
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
|
||||
from runner_utils import append_github_output, create_registration_token, find_runner_by_name, github_api
|
||||
|
||||
RUNNER_NAME_PREFIX = "qreg-ecs"
|
||||
RUNNER_LABEL_PREFIX = "query-regression-ecs"
|
||||
MANAGED_BY_TAG_KEY = "managed-by"
|
||||
@@ -243,63 +243,6 @@ 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:
|
||||
if response.status == 204:
|
||||
return {}
|
||||
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
|
||||
@@ -388,13 +331,6 @@ def dump_console_output(client, region_id: str, instance_id: str) -> None:
|
||||
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
|
||||
|
||||
|
||||
@@ -47,6 +47,8 @@ import sys
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from runner_utils import deregister_runner
|
||||
|
||||
# Reuse the GitHub API client, runner lookup, and tag constants from the
|
||||
# provision script.
|
||||
_PROVISION_SPEC = importlib.util.spec_from_file_location(
|
||||
@@ -159,20 +161,6 @@ def delete_instance(client, instance_id: str, region_id: str | None = None) -> b
|
||||
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, str | None]]:
|
||||
|
||||
@@ -0,0 +1,265 @@
|
||||
#!/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.
|
||||
|
||||
# /// script
|
||||
# requires-python = ">=3.10"
|
||||
# dependencies = ["boto3>=1.35,<2"]
|
||||
# ///
|
||||
|
||||
"""Provision an ephemeral EC2 CI runner from an Ubuntu x86_64 AMI."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
|
||||
from runner_utils import append_github_output, create_registration_token, find_runner_by_name, github_api
|
||||
|
||||
MANAGED_BY = "query-regression-ci"
|
||||
REPO_TAG = "github-repository"
|
||||
RUN_TAG = "query-regression-run-id"
|
||||
RUNNER_TAG = "github-runner-name"
|
||||
TTL_TAG_KEY = "runner-ttl-hours"
|
||||
|
||||
|
||||
def make_client(region: str):
|
||||
import boto3
|
||||
from botocore.config import Config
|
||||
|
||||
return boto3.client("ec2", region_name=region, config=Config(
|
||||
connect_timeout=10, read_timeout=30,
|
||||
retries={"mode": "standard", "max_attempts": 3},
|
||||
))
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProvisionConfig:
|
||||
region: str
|
||||
image_id: str
|
||||
instance_type: str
|
||||
subnet_id: str
|
||||
security_group_id: str
|
||||
repo: str
|
||||
run_id: str
|
||||
github_token: str
|
||||
system_disk_gib: int = 500
|
||||
ttl_hours: int = 8
|
||||
|
||||
def __post_init__(self):
|
||||
if not re.fullmatch(r"[0-9]+-[0-9]+", self.run_id):
|
||||
raise ValueError("run-id must be GitHub run_id-run_attempt")
|
||||
if not 20 <= self.system_disk_gib <= 2048:
|
||||
raise ValueError("system-disk-gib must be between 20 and 2048")
|
||||
if not 1 <= self.ttl_hours <= 168:
|
||||
raise ValueError("ttl-hours must be between 1 and 168")
|
||||
|
||||
@property
|
||||
def runner_name(self) -> str:
|
||||
return f"ci-ec2-{self.run_id}"
|
||||
|
||||
|
||||
def resolve_image(client, image_id: str) -> str:
|
||||
if image_id != "auto":
|
||||
return image_id
|
||||
images = client.describe_images(
|
||||
Owners=["099720109477"],
|
||||
Filters=[
|
||||
{"Name": "name", "Values": ["ubuntu/images/hvm-ssd-gp3/ubuntu-noble-24.04-amd64-server-*"]},
|
||||
{"Name": "state", "Values": ["available"]},
|
||||
{"Name": "architecture", "Values": ["x86_64"]},
|
||||
],
|
||||
)["Images"]
|
||||
if not images:
|
||||
raise ValueError("No Canonical Ubuntu 24.04 amd64 AMI found in this region")
|
||||
image = max(images, key=lambda image: image["CreationDate"])
|
||||
print(f"Using AMI {image['ImageId']}: {image['Name']}")
|
||||
return image["ImageId"]
|
||||
|
||||
|
||||
def root_disk(client, image_id: str, size: int) -> dict:
|
||||
images = client.describe_images(ImageIds=[image_id])["Images"]
|
||||
if len(images) != 1:
|
||||
raise ValueError(f"AMI not found: {image_id}")
|
||||
image = images[0]
|
||||
if image["State"] != "available" or image["Architecture"] != "x86_64":
|
||||
raise ValueError("Runner AMI must be available and x86_64")
|
||||
root = next((d for d in image["BlockDeviceMappings"]
|
||||
if d["DeviceName"] == image["RootDeviceName"] and "Ebs" in d), None)
|
||||
if root is None:
|
||||
raise ValueError("Runner AMI must have an EBS root disk")
|
||||
minimum = root["Ebs"]["VolumeSize"]
|
||||
if size < minimum:
|
||||
raise ValueError(f"Requested root disk {size} GiB is smaller than AMI minimum {minimum} GiB")
|
||||
return {"DeviceName": image["RootDeviceName"], "Ebs": {
|
||||
"VolumeSize": size, "VolumeType": "gp3", "Iops": 3000,
|
||||
"Throughput": 125, "DeleteOnTermination": True,
|
||||
}}
|
||||
|
||||
|
||||
def runner_download(token: str, repo: str) -> str:
|
||||
runners = github_api(token, "GET", f"/repos/{repo}/actions/runners/downloads")
|
||||
return next(r["download_url"] for r in runners if r["os"] == "linux" and r["architecture"] == "x64")
|
||||
|
||||
|
||||
def render_user_data(name: str, token: str, repo: str, download_url: str) -> str:
|
||||
# Quote every external value before embedding it in the root shell script.
|
||||
name, token, repo, download_url = map(shlex.quote, (name, token, repo, download_url))
|
||||
return f"""#!/bin/bash
|
||||
set -euo pipefail
|
||||
for tool in git curl tar sudo; do command -v "$tool"; done
|
||||
# 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 "/swapfile" ]]; then
|
||||
fallocate --length 16G "/swapfile"
|
||||
chmod 600 "/swapfile"
|
||||
mkswap "/swapfile"
|
||||
fi
|
||||
swapon "/swapfile"
|
||||
sysctl --write vm.swappiness=10
|
||||
swapon --show
|
||||
free --human
|
||||
|
||||
id runner >/dev/null 2>&1 || useradd --create-home --shell /bin/bash runner
|
||||
# Establish membership before the runner service starts; setup installs Docker later.
|
||||
getent group docker >/dev/null || groupadd --system docker
|
||||
usermod -aG docker runner
|
||||
printf 'runner ALL=(ALL) NOPASSWD: ALL\\n' > /etc/sudoers.d/ci-runner
|
||||
chmod 0440 /etc/sudoers.d/ci-runner
|
||||
visudo -cf /etc/sudoers.d/ci-runner
|
||||
install -d -o runner -g "$(id -gn runner)" /home/runner/benchmark-data /opt/ci-github-runner
|
||||
cd /opt/ci-github-runner
|
||||
curl --fail --location --retry 3 {download_url} | tar xz
|
||||
./bin/installdependencies.sh
|
||||
chown -R runner:"$(id -gn runner)" /opt/ci-github-runner
|
||||
runuser -u runner -- ./config.sh --unattended --ephemeral --disableupdate --url https://github.com/{repo} --token {token} --name {name} --labels {name}
|
||||
./svc.sh install runner
|
||||
# Match the ECS runner's behavior when a child process is OOM-killed.
|
||||
service_name=$(cat .service)
|
||||
mkdir -p "/etc/systemd/system/$service_name.d"
|
||||
cat > "/etc/systemd/system/$service_name.d/oom.conf" <<'EOF'
|
||||
[Service]
|
||||
OOMPolicy=continue
|
||||
EOF
|
||||
systemctl daemon-reload
|
||||
# Package installation in a job must not restart its runner service.
|
||||
install -d /etc/needrestart/conf.d
|
||||
cat > /etc/needrestart/conf.d/actions_runner_services.conf <<'EOF'
|
||||
$nrconf{{override_rc}}{{qr(^actions\\.runner\\..+\\.service$)}} = 0;
|
||||
EOF
|
||||
./svc.sh start
|
||||
"""
|
||||
|
||||
|
||||
def run_instance(client, config: ProvisionConfig) -> str:
|
||||
image_id = resolve_image(client, config.image_id)
|
||||
disk = root_disk(client, image_id, config.system_disk_gib)
|
||||
instance = client.describe_instance_types(InstanceTypes=[config.instance_type])["InstanceTypes"][0]
|
||||
if "x86_64" not in instance["ProcessorInfo"]["SupportedArchitectures"]:
|
||||
raise ValueError("Runner instance type must support x86_64")
|
||||
token = create_registration_token(config.github_token, config.repo)
|
||||
print(f"::add-mask::{token}")
|
||||
script = render_user_data(config.runner_name, token, config.repo, runner_download(config.github_token, config.repo))
|
||||
tags = [
|
||||
{"Key": "Name", "Value": config.runner_name},
|
||||
{"Key": "managed-by", "Value": MANAGED_BY},
|
||||
{"Key": REPO_TAG, "Value": config.repo},
|
||||
{"Key": RUN_TAG, "Value": config.run_id},
|
||||
{"Key": RUNNER_TAG, "Value": config.runner_name},
|
||||
{"Key": TTL_TAG_KEY, "Value": str(config.ttl_hours)},
|
||||
]
|
||||
response = client.run_instances(
|
||||
ImageId=image_id, InstanceType=config.instance_type,
|
||||
MinCount=1, MaxCount=1, SubnetId=config.subnet_id,
|
||||
SecurityGroupIds=[config.security_group_id],
|
||||
BlockDeviceMappings=[disk], UserData=script,
|
||||
ClientToken=config.runner_name,
|
||||
MetadataOptions={"HttpTokens": "required", "HttpEndpoint": "enabled"},
|
||||
TagSpecifications=[{"ResourceType": kind, "Tags": tags} for kind in ("instance", "volume")],
|
||||
)
|
||||
return response["Instances"][0]["InstanceId"]
|
||||
|
||||
|
||||
def dump_console(client, instance_id: str) -> None:
|
||||
try:
|
||||
# Boto3 decodes the API's base64 console output.
|
||||
output = client.get_console_output(InstanceId=instance_id, Latest=True).get("Output", "")
|
||||
print(f"::group::EC2 console {instance_id}")
|
||||
print("\n".join(output.splitlines()[-80:]))
|
||||
print("::endgroup::")
|
||||
except Exception as error:
|
||||
print(f"Unable to read console for {instance_id}: {error}")
|
||||
|
||||
|
||||
def provision(client, config: ProvisionConfig) -> int:
|
||||
instance_id = run_instance(client, config)
|
||||
# Publish before waiting: failed registration must still reach teardown.
|
||||
for key, value in (("instance_id", instance_id),
|
||||
("label", config.runner_name), ("runner_name", config.runner_name)):
|
||||
append_github_output(key, value)
|
||||
try:
|
||||
client.get_waiter("instance_running").wait(
|
||||
InstanceIds=[instance_id], WaiterConfig={"Delay": 5, "MaxAttempts": 60})
|
||||
deadline = time.monotonic() + 10 * 60
|
||||
while time.monotonic() < deadline:
|
||||
runner = find_runner_by_name(config.github_token, config.repo, config.runner_name)
|
||||
if runner and runner.get("status") == "online":
|
||||
print(f"Runner {config.runner_name} online: {instance_id}")
|
||||
return 0
|
||||
time.sleep(5)
|
||||
raise TimeoutError(f"Runner {config.runner_name} did not come online")
|
||||
finally:
|
||||
dump_console(client, instance_id)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
for name, env in (("region", "AWS_REGION"), ("repo", "GITHUB_REPOSITORY"),
|
||||
("github-token", "GH_PERSONAL_ACCESS_TOKEN"),
|
||||
("run-id", "AWS_EC2_RUN_ID"), ("image-id", "AWS_EC2_IMAGE_ID"),
|
||||
("instance-type", "AWS_EC2_INSTANCE_TYPE"),
|
||||
("subnet-id", "AWS_EC2_SUBNET_ID"),
|
||||
("security-group-id", "AWS_EC2_SECURITY_GROUP_ID")):
|
||||
parser.add_argument(f"--{name}", default=os.environ.get(env) or ("auto" if name == "image-id" else None))
|
||||
parser.add_argument("--system-disk-gib", type=int, default=os.environ.get("AWS_EC2_SYSTEM_DISK_GIB", "500"))
|
||||
parser.add_argument("--ttl-hours", type=int, default=os.environ.get("AWS_EC2_TTL_HOURS", "8"))
|
||||
args = parser.parse_args()
|
||||
for name, value in vars(args).items():
|
||||
if value is None or value == "":
|
||||
parser.error(f"Missing {name.replace('_', '-')}")
|
||||
config = ProvisionConfig(**vars(args))
|
||||
return provision(make_client(args.region), config)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,160 @@
|
||||
#!/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.
|
||||
|
||||
# /// script
|
||||
# requires-python = ">=3.10"
|
||||
# dependencies = ["boto3>=1.35,<2"]
|
||||
# ///
|
||||
|
||||
"""Delete managed EC2 CI runners, or sweep expired instances by ownership tags."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from runner_utils import deregister_runner
|
||||
|
||||
_SPEC = importlib.util.spec_from_file_location("aws_ec2_runner_provision", Path(__file__).with_name("aws-ec2-runner-provision.py"))
|
||||
assert _SPEC is not None and _SPEC.loader is not None
|
||||
provision = importlib.util.module_from_spec(_SPEC)
|
||||
sys.modules[_SPEC.name] = provision
|
||||
_SPEC.loader.exec_module(provision)
|
||||
make_client = provision.make_client
|
||||
dump_console = provision.dump_console
|
||||
MANAGED_BY = provision.MANAGED_BY
|
||||
REPO_TAG = provision.REPO_TAG
|
||||
RUN_TAG = provision.RUN_TAG
|
||||
RUNNER_TAG = provision.RUNNER_TAG
|
||||
TTL_TAG_KEY = provision.TTL_TAG_KEY
|
||||
SWEEP_TTL_HOURS = 4
|
||||
|
||||
|
||||
def managed_instances(client, repo: str, run_id: str | None = None,
|
||||
instance_id: str | None = None) -> list[dict]:
|
||||
filters = [{"Name": "tag:managed-by", "Values": [MANAGED_BY]},
|
||||
{"Name": f"tag:{REPO_TAG}", "Values": [repo]}]
|
||||
if instance_id:
|
||||
# A retried teardown may have a newer workflow attempt than creation.
|
||||
filters.append({"Name": "instance-id", "Values": [instance_id]})
|
||||
elif run_id:
|
||||
filters.append({"Name": f"tag:{RUN_TAG}", "Values": [run_id]})
|
||||
return [instance
|
||||
for page in client.get_paginator("describe_instances").paginate(Filters=filters)
|
||||
for reservation in page["Reservations"] for instance in reservation["Instances"]]
|
||||
|
||||
|
||||
def delete_instance(client, instance_id: str) -> bool:
|
||||
dump_console(client, instance_id)
|
||||
for attempt in range(12):
|
||||
try:
|
||||
client.terminate_instances(InstanceIds=[instance_id])
|
||||
client.get_waiter("instance_terminated").wait(
|
||||
InstanceIds=[instance_id], WaiterConfig={"Delay": 5, "MaxAttempts": 60})
|
||||
return True
|
||||
except Exception as error:
|
||||
code = getattr(error, "response", {}).get("Error", {}).get("Code")
|
||||
if code == "InvalidInstanceID.NotFound":
|
||||
if attempt < 11:
|
||||
time.sleep(5)
|
||||
continue
|
||||
# The ownership lookup observed this instance. Treat sustained
|
||||
# NotFound after bounded retries as an already-deleted VM.
|
||||
return True
|
||||
print(f"Failed to terminate {instance_id}: {error}")
|
||||
return False
|
||||
return False
|
||||
|
||||
|
||||
def teardown(client, repo: str, token: str, run_id: str | None = None,
|
||||
instance_id: str | None = None, now: datetime | None = None,
|
||||
runner_name: str | None = None) -> int:
|
||||
# Lookup by ownership/run tags also recovers a create whose response or job
|
||||
# outputs were lost. Sweep never touches legacy build/JSONBench instances.
|
||||
instances = managed_instances(client, repo, run_id, instance_id)
|
||||
# Tags can lag RunInstances. Retry before concluding a targeted VM is gone.
|
||||
if run_id and not instances:
|
||||
for _ in range(11):
|
||||
time.sleep(5)
|
||||
instances = managed_instances(client, repo, run_id, instance_id)
|
||||
if instances:
|
||||
break
|
||||
ok = True
|
||||
found = False
|
||||
for instance in instances:
|
||||
identity = instance["InstanceId"]
|
||||
if instance_id and identity != instance_id:
|
||||
continue
|
||||
found = True
|
||||
tags = {t["Key"]: t["Value"] for t in instance.get("Tags", [])}
|
||||
name = tags.get(RUNNER_TAG, "")
|
||||
if not name.startswith("ci-ec2-"):
|
||||
print(f"Skipping {identity}: missing managed runner name")
|
||||
ok = False
|
||||
continue
|
||||
if not run_id:
|
||||
ttl = tags.get(TTL_TAG_KEY)
|
||||
try:
|
||||
hours = SWEEP_TTL_HOURS if ttl is None else int(ttl)
|
||||
if not 1 <= hours <= 168:
|
||||
raise ValueError("TTL outside 1..168 hours")
|
||||
except (TypeError, ValueError):
|
||||
print(f"Skipping {identity}: invalid TTL {ttl!r}")
|
||||
continue
|
||||
if (now or datetime.now(timezone.utc)) - instance["LaunchTime"] < timedelta(hours=hours):
|
||||
continue
|
||||
# A GitHub outage must not prevent EC2 deletion, or vice versa.
|
||||
ok &= delete_instance(client, identity)
|
||||
try:
|
||||
ok &= deregister_runner(token, repo, name)
|
||||
except (Exception, SystemExit) as error:
|
||||
print(f"Failed to unregister {name}: {error}")
|
||||
ok = False
|
||||
if run_id and not found:
|
||||
# EC2 may already be gone while its GitHub registration remains.
|
||||
name = runner_name or f"ci-ec2-{run_id}"
|
||||
if not name.startswith("ci-ec2-"):
|
||||
raise ValueError("Invalid CI runner name")
|
||||
ok &= deregister_runner(token, repo, name)
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
for name, env in (("region", "AWS_REGION"), ("repo", "GITHUB_REPOSITORY"),
|
||||
("github-token", "GH_PERSONAL_ACCESS_TOKEN"),
|
||||
("run-id", "AWS_EC2_RUN_ID"), ("instance-id", "AWS_EC2_INSTANCE_ID"),
|
||||
("runner-name", "AWS_EC2_RUNNER_NAME")):
|
||||
parser.add_argument(f"--{name}", default=os.environ.get(env))
|
||||
parser.add_argument("--sweep", action="store_true")
|
||||
args = parser.parse_args()
|
||||
for name in ("region", "repo", "github_token"):
|
||||
if not getattr(args, name):
|
||||
parser.error(f"Missing {name.replace('_', '-')}")
|
||||
if not args.sweep and not args.run_id:
|
||||
parser.error("Missing run-id for targeted teardown")
|
||||
return teardown(make_client(args.region), args.repo, args.github_token,
|
||||
None if args.sweep else args.run_id,
|
||||
None if args.sweep else args.instance_id,
|
||||
runner_name=None if args.sweep else args.runner_name)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,47 @@
|
||||
#!/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 dispatch provider and instance type before allocating a runner."""
|
||||
|
||||
import argparse
|
||||
import re
|
||||
|
||||
from runner_utils import append_github_output
|
||||
|
||||
DEFAULT_INSTANCE_TYPES = {"Aliyun": "ecs.c9i.2xlarge", "AWS": "c7i.2xlarge"}
|
||||
|
||||
|
||||
def resolve_instance_type(provider: str, requested: str) -> str:
|
||||
if provider not in DEFAULT_INSTANCE_TYPES:
|
||||
raise ValueError("provider must be Aliyun or AWS")
|
||||
instance_type = DEFAULT_INSTANCE_TYPES[provider] if requested == "auto" else requested
|
||||
pattern = r"ecs\.[a-z0-9-]+\.[a-z0-9-]+" if provider == "Aliyun" else r"[a-z0-9-]+\.[a-z0-9-]+"
|
||||
if not re.fullmatch(pattern, instance_type):
|
||||
raise ValueError(f"Invalid {provider} instance type: {instance_type!r}")
|
||||
return instance_type
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--provider", required=True)
|
||||
parser.add_argument("--instance-type", default="auto")
|
||||
args = parser.parse_args()
|
||||
instance_type = resolve_instance_type(args.provider, args.instance_type)
|
||||
append_github_output("instance_type", instance_type)
|
||||
print(f"CI runner: {args.provider} / {instance_type}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,71 @@
|
||||
# CI runners
|
||||
|
||||
The agent observability and long-range benchmark workflows select `Aliyun` or
|
||||
`AWS` through the `provider` dispatch input. `instance_type: auto` selects
|
||||
`ecs.c9i.2xlarge` or `c7i.2xlarge`, respectively (8 vCPU, 16 GiB).
|
||||
Existing build/release and JSONBench workflows keep their current runner actions.
|
||||
|
||||
Aliyun uses the existing ECS actions and image. Its provision and teardown
|
||||
scripts share only GitHub API and output helpers from `runner_utils.py` with
|
||||
the new AWS scripts.
|
||||
|
||||
AWS uses `aws-ec2-runner-provision.py` and `aws-ec2-runner-teardown.py` directly.
|
||||
Repository variables supply `EC2_RUNNER_REGION`, `EC2_RUNNER_SUBNET_ID`, and
|
||||
`EC2_RUNNER_SECURITY_GROUP_ID`; credentials use `AWS_ACCESS_KEY_ID`,
|
||||
`AWS_SECRET_ACCESS_KEY`, and `GH_PERSONAL_ACCESS_TOKEN` secrets.
|
||||
The current region is Singapore (`ap-southeast-1`). By default the script selects
|
||||
the newest available Canonical Ubuntu 24.04 amd64 server AMI in that region
|
||||
(owner `099720109477`) and logs its ID. Set `BENCHMARK_EC2_IMAGE_ID` to pin a
|
||||
prepared Ubuntu AMI. Existing build CI's `EC2_RUNNER_LINUX_AMD64_IMAGE_ID` is
|
||||
unaffected. Both benchmark workflows default to 500 GiB for both providers;
|
||||
explicit disk overrides must be at least the selected AMI's root snapshot size.
|
||||
The official 24.04 image does not have the old custom AMI's 500 GiB minimum.
|
||||
|
||||
AWS user data registers a fresh ephemeral runner. Git, curl, tar, and sudo are
|
||||
expected in the Ubuntu server image. Docker group membership is set before the
|
||||
runner service starts, and this dedicated VM grants the runner passwordless sudo
|
||||
so its job can install dependencies. Like ECS, the host creates 16 GiB swap with
|
||||
swappiness 10, disables systemd-oomd and background automatic updates, and sets
|
||||
`OOMPolicy=continue` on the runner service. Explicit setup package installation
|
||||
remains enabled.
|
||||
|
||||
Both benchmark jobs call `.github/actions/setup-benchmark` after checkout. It
|
||||
reuses existing tools, installs missing curl/jq/python3/rsync, installs Docker CE
|
||||
if absent, and checks Docker access. Missing packages require passwordless sudo;
|
||||
Aliyun's prepared image normally provides the tools and Docker already. Existing
|
||||
Docker installations are preserved, so exact versions need not match between
|
||||
providers. Provisioning must establish Docker permissions before starting the
|
||||
runner; adding a group inside the running job would not update its groups.
|
||||
|
||||
The subnet must provide outbound access to Ubuntu and Docker packages, GitHub,
|
||||
and the configured container registries. The IAM identity needs EC2
|
||||
image/instance-type discovery, instance and volume tagging, launch, describe,
|
||||
console output, and termination permissions. GitHub credentials need repository
|
||||
runner administration access.
|
||||
|
||||
Creation publishes instance identity before waiting for registration. Teardown
|
||||
can recover missing outputs using repository and workflow-attempt tags. The
|
||||
daily AWS janitor (03:23 UTC) removes expired instances tagged `managed-by=query-regression-ci`
|
||||
and unregisters their runners. Both providers use the existing
|
||||
`managed-by=query-regression-ci`, `query-regression-run-id`, and
|
||||
`runner-ttl-hours` tags. AWS retains a run-attempt suffix in its run identity to
|
||||
isolate retries, plus repository and runner-name tags to restrict cleanup.
|
||||
Legacy build instances do not match these ownership filters.
|
||||
Both janitors use a 4-hour fallback when the TTL tag is absent and skip invalid
|
||||
TTL tags. Benchmark jobs explicitly tag their configured TTL (default 8 hours).
|
||||
Cleanup does not check whether the GitHub job is still running. Expired instances
|
||||
are removed on the next daily sweep; normal workflow teardown runs immediately.
|
||||
TTL is measured from instance launch, so it must cover provisioning and the
|
||||
benchmark. The scheduled janitor becomes active after landing on the default
|
||||
branch.
|
||||
|
||||
Local validation:
|
||||
|
||||
```sh
|
||||
python3 tests/perf/test_ci_runner_scripts.py
|
||||
python3 tests/perf/test_aliyun_ecs_runner_scripts.py
|
||||
actionlint .github/workflows/agent-observability.yml .github/workflows/vmbench-long-range.yml .github/workflows/aws-ec2-runner-janitor.yml
|
||||
```
|
||||
|
||||
A real AWS dispatch is still needed to verify AMI boot, package
|
||||
installation, registry connectivity, and end-to-end benchmark execution.
|
||||
@@ -0,0 +1,101 @@
|
||||
#!/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.
|
||||
|
||||
"""Small shared GitHub runner API and workflow-output helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
|
||||
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:
|
||||
if response.status == 204:
|
||||
return {}
|
||||
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 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 deregister_runner(token: str, repo: str, runner_name: str) -> bool:
|
||||
runner = 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:
|
||||
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
|
||||
Reference in New Issue
Block a user