mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-11-10 01:25:21 +00:00
285 lines
8.5 KiB
Python
285 lines
8.5 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from typing import Any, Iterator, List, Union, Optional, Sequence
|
|
|
|
import requests
|
|
|
|
|
|
class Envs:
|
|
GITHUB_REPOSITORY = os.getenv("GITHUB_REPOSITORY", "ClickHouse/ClickHouse")
|
|
WORKFLOW_RESULT_FILE = os.getenv(
|
|
"WORKFLOW_RESULT_FILE", "/tmp/workflow_results.json"
|
|
)
|
|
S3_BUILDS_BUCKET = os.getenv("S3_BUILDS_BUCKET", "clickhouse-builds")
|
|
GITHUB_WORKFLOW = os.getenv("GITHUB_WORKFLOW", "")
|
|
|
|
|
|
class WithIter(type):
|
|
def __iter__(cls):
|
|
return (v for k, v in cls.__dict__.items() if not k.startswith("_"))
|
|
|
|
|
|
@contextmanager
|
|
def cd(path: Union[Path, str]) -> Iterator[None]:
|
|
oldpwd = os.getcwd()
|
|
os.chdir(path)
|
|
try:
|
|
yield
|
|
finally:
|
|
os.chdir(oldpwd)
|
|
|
|
|
|
class GH:
|
|
class ActionsNames:
|
|
RunConfig = "RunConfig"
|
|
|
|
class ActionStatuses:
|
|
ERROR = "error"
|
|
FAILURE = "failure"
|
|
PENDING = "pending"
|
|
SUCCESS = "success"
|
|
SKIPPED = "skipped"
|
|
|
|
@classmethod
|
|
def get_workflow_results(cls):
|
|
if not Path(Envs.WORKFLOW_RESULT_FILE).exists():
|
|
print(
|
|
f"ERROR: Failed to get workflow results from file [{Envs.WORKFLOW_RESULT_FILE}]"
|
|
)
|
|
return {}
|
|
with open(Envs.WORKFLOW_RESULT_FILE, "r", encoding="utf-8") as json_file:
|
|
try:
|
|
res = json.load(json_file)
|
|
except json.JSONDecodeError as e:
|
|
print(f"ERROR: json decoder exception {e}")
|
|
json_file.seek(0)
|
|
print(" File content:")
|
|
print(json_file.read())
|
|
return {}
|
|
return res
|
|
|
|
@classmethod
|
|
def print_workflow_results(cls):
|
|
res = cls.get_workflow_results()
|
|
results = [f"{job}: {data['result']}" for job, data in res.items()]
|
|
cls.print_in_group("Workflow results", results)
|
|
|
|
@classmethod
|
|
def is_workflow_ok(cls) -> bool:
|
|
res = cls.get_workflow_results()
|
|
for _job, data in res.items():
|
|
if data["result"] == "failure":
|
|
return False
|
|
return bool(res)
|
|
|
|
@classmethod
|
|
def get_workflow_job_result(cls, wf_job_name: str) -> Optional[str]:
|
|
res = cls.get_workflow_results()
|
|
if wf_job_name in res:
|
|
return res[wf_job_name]["result"] # type: ignore
|
|
else:
|
|
return None
|
|
|
|
@staticmethod
|
|
def print_in_group(group_name: str, lines: Union[Any, List[Any]]) -> None:
|
|
lines = list(lines)
|
|
print(f"::group::{group_name}")
|
|
for line in lines:
|
|
print(line)
|
|
print("::endgroup::")
|
|
|
|
@staticmethod
|
|
def get_commit_status_by_name(
|
|
token: str, commit_sha: str, status_name: Union[str, Sequence]
|
|
) -> str:
|
|
assert len(token) == 40
|
|
assert len(commit_sha) == 40
|
|
assert Utils.is_hex(commit_sha)
|
|
assert not Utils.is_hex(token)
|
|
|
|
url = f"https://api.github.com/repos/{Envs.GITHUB_REPOSITORY}/commits/{commit_sha}/statuses"
|
|
headers = {
|
|
"Authorization": f"token {token}",
|
|
"Accept": "application/vnd.github.v3+json",
|
|
}
|
|
|
|
if isinstance(status_name, str):
|
|
status_name = (status_name,)
|
|
|
|
while url:
|
|
response = requests.get(url, headers=headers, timeout=5)
|
|
if response.status_code == 200:
|
|
statuses = response.json()
|
|
for status in statuses:
|
|
if status["context"] in status_name:
|
|
return status["state"] # type: ignore
|
|
|
|
# Check if there is a next page
|
|
url = response.links.get("next", {}).get("url")
|
|
else:
|
|
break
|
|
|
|
return ""
|
|
|
|
@staticmethod
|
|
def check_wf_completed(token: str, commit_sha: str) -> bool:
|
|
headers = {
|
|
"Authorization": f"token {token}",
|
|
"Accept": "application/vnd.github.v3+json",
|
|
}
|
|
url = f"https://api.github.com/repos/{Envs.GITHUB_REPOSITORY}/commits/{commit_sha}/check-runs?per_page={100}"
|
|
|
|
for i in range(3):
|
|
try:
|
|
response = requests.get(url, headers=headers, timeout=5)
|
|
response.raise_for_status()
|
|
# assert "next" not in response.links, "Response truncated"
|
|
|
|
data = response.json()
|
|
assert data["check_runs"], "?"
|
|
|
|
for check in data["check_runs"]:
|
|
if check["status"] != "completed":
|
|
print(
|
|
f" Check workflow status: Check not completed [{check['name']}]"
|
|
)
|
|
return False
|
|
return True
|
|
except Exception as e:
|
|
print(f"ERROR: exception after attempt [{i}]: {e}")
|
|
time.sleep(1)
|
|
|
|
return False
|
|
|
|
@staticmethod
|
|
def get_pr_url_by_branch(branch, repo=None):
|
|
repo = repo or Envs.GITHUB_REPOSITORY
|
|
get_url_cmd = f"gh pr list --repo {repo} --head {branch} --json url --jq '.[0].url' --state open"
|
|
url = Shell.get_output(get_url_cmd)
|
|
if not url:
|
|
print(f"WARNING: No open PR found, branch [{branch}] - search for merged")
|
|
get_url_cmd = f"gh pr list --repo {repo} --head {branch} --json url --jq '.[0].url' --state merged"
|
|
url = Shell.get_output(get_url_cmd)
|
|
if not url:
|
|
print(f"ERROR: PR nor found, branch [{branch}]")
|
|
return url
|
|
|
|
@staticmethod
|
|
def is_latest_release_branch(branch):
|
|
latest_branch = Shell.get_output(
|
|
'gh pr list --label release --repo ClickHouse/ClickHouse --search "sort:created" -L1 --json headRefName'
|
|
)
|
|
if latest_branch:
|
|
latest_branch = json.loads(latest_branch)[0]["headRefName"]
|
|
print(
|
|
f"Latest branch [{latest_branch}], release branch [{branch}], release latest [{latest_branch == branch}]"
|
|
)
|
|
return latest_branch == branch
|
|
|
|
|
|
class Shell:
|
|
@classmethod
|
|
def get_output_or_raise(cls, command):
|
|
return cls.get_output(command, strict=True)
|
|
|
|
@classmethod
|
|
def get_output(cls, command, strict=False):
|
|
res = subprocess.run(
|
|
command,
|
|
shell=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
check=strict,
|
|
)
|
|
return res.stdout.strip()
|
|
|
|
@classmethod
|
|
def check(
|
|
cls,
|
|
command,
|
|
strict=False,
|
|
verbose=False,
|
|
dry_run=False,
|
|
stdin_str=None,
|
|
**kwargs,
|
|
):
|
|
if dry_run:
|
|
print(f"Dry-ryn. Would run command [{command}]")
|
|
return True
|
|
if verbose:
|
|
print(f"Run command [{command}]")
|
|
proc = subprocess.Popen(
|
|
command,
|
|
shell=True,
|
|
stderr=subprocess.STDOUT,
|
|
stdout=subprocess.PIPE,
|
|
stdin=subprocess.PIPE if stdin_str else None,
|
|
universal_newlines=True,
|
|
start_new_session=True,
|
|
bufsize=1,
|
|
errors="backslashreplace",
|
|
**kwargs,
|
|
)
|
|
if stdin_str:
|
|
proc.communicate(input=stdin_str)
|
|
elif proc.stdout:
|
|
for line in proc.stdout:
|
|
sys.stdout.write(line)
|
|
proc.wait()
|
|
if strict:
|
|
assert proc.returncode == 0
|
|
return proc.returncode == 0
|
|
|
|
|
|
class Utils:
|
|
@staticmethod
|
|
def get_failed_tests_number(description: str) -> Optional[int]:
|
|
description = description.lower()
|
|
|
|
pattern = r"fail:\s*(\d+)\s*(?=,|$)"
|
|
match = re.search(pattern, description)
|
|
if match:
|
|
return int(match.group(1))
|
|
return None
|
|
|
|
@staticmethod
|
|
def is_killed_with_oom():
|
|
if Shell.check(
|
|
"sudo dmesg -T | grep -q -e 'Out of memory: Killed process' -e 'oom_reaper: reaped process' -e 'oom-kill:constraint=CONSTRAINT_NONE'"
|
|
):
|
|
return True
|
|
return False
|
|
|
|
@staticmethod
|
|
def clear_dmesg():
|
|
Shell.check("sudo dmesg --clear", verbose=True)
|
|
|
|
@staticmethod
|
|
def is_hex(s):
|
|
try:
|
|
int(s, 16)
|
|
return True
|
|
except ValueError:
|
|
return False
|
|
|
|
@staticmethod
|
|
def normalize_string(string: str) -> str:
|
|
res = string.lower()
|
|
for r in (
|
|
(" ", "_"),
|
|
("(", "_"),
|
|
(")", "_"),
|
|
(",", "_"),
|
|
("/", "_"),
|
|
("-", "_"),
|
|
):
|
|
res = res.replace(*r)
|
|
return res
|