mirror of
https://github.com/IfcOpenShell/IfcOpenShell.git
synced 2026-08-06 07:51:47 +00:00
Andrej730
105 lines
2.9 KiB
Python
105 lines
2.9 KiB
Python
import os
|
|
import pathlib
|
|
import shutil
|
|
import zipfile
|
|
|
|
import requests
|
|
|
|
# To test this locally, set these environment variables
|
|
REPO_OWNER = os.environ.get("REPO_OWNER", "IfcOpenShell/IfcOpenShell")
|
|
MY_WORKFLOW = os.environ.get("MY_WORKFLOW", "ci-ifcopenshell-conda-daily")
|
|
WORKFLOW_RUN_ID = os.environ.get("WORKFLOW_RUN_ID", None)
|
|
TOKEN = os.environ.get("GITHUB_TOKEN", None)
|
|
# To test this locally create a fine-grained personal access token for this repo with permissions "actions:read"
|
|
# See https://github.com/settings/tokens?type=beta
|
|
|
|
# This is a list of strings that indicate that a job has stopped abruptly.
|
|
SIGNS_OF_STOPPAGE = [
|
|
"Error: The operation was canceled.",
|
|
"fatal error C1060: compiler is out of heap space",
|
|
]
|
|
|
|
|
|
def start_request_session():
|
|
s = requests.Session()
|
|
headers = {
|
|
"Accept": "application/vnd.github+json",
|
|
"X-GitHub-Api-Version": "2022-11-28",
|
|
}
|
|
if TOKEN:
|
|
headers["Authorization"] = f"Bearer {TOKEN}"
|
|
|
|
s.headers = headers
|
|
return s
|
|
|
|
|
|
def get_ci_run(repo_name, run_id):
|
|
url = f"https://api.github.com/repos/{repo_name}/actions/runs/{run_id}"
|
|
s = start_request_session()
|
|
response = s.get(url)
|
|
return response.json()
|
|
|
|
|
|
def evaluate_log_file_for_abrupt_stop(log_file):
|
|
with open(log_file) as f:
|
|
for i, line in enumerate(f):
|
|
for sign in SIGNS_OF_STOPPAGE:
|
|
if sign in line:
|
|
print(f"Found sign of abrupt stoppage in [line {i}]: '{sign}'")
|
|
return True
|
|
return False
|
|
|
|
|
|
def get_ci_specific_run_specific_failure_details(repo_name, run_id):
|
|
url = f"https://api.github.com/repos/{repo_name}/actions/runs/{run_id}/attempts/1/logs"
|
|
s = start_request_session()
|
|
response = s.get(url)
|
|
|
|
# SAVE zip file
|
|
with open("logs.zip", "wb") as f:
|
|
f.write(response.content)
|
|
|
|
# unzip file
|
|
with zipfile.ZipFile("logs.zip", "r") as zip_ref:
|
|
zip_ref.extractall("logs")
|
|
|
|
failed_logs = []
|
|
for file in pathlib.Path("logs").iterdir():
|
|
if file.is_dir():
|
|
continue
|
|
|
|
if evaluate_log_file_for_abrupt_stop(file):
|
|
failed_logs.append(file)
|
|
|
|
return failed_logs
|
|
|
|
|
|
def restart_job(repo_name, job_id):
|
|
url = f"https://api.github.com/repos/{repo_name}/actions/runs/{job_id}/rerun-failed-jobs"
|
|
s = start_request_session()
|
|
response = s.post(url)
|
|
return response.json()
|
|
|
|
|
|
def eval_jobs():
|
|
run = get_ci_run(REPO_OWNER, WORKFLOW_RUN_ID)
|
|
if run["run_attempt"] > 1:
|
|
print("This is not the first attempt. Exiting")
|
|
return
|
|
failed_logs = get_ci_specific_run_specific_failure_details(REPO_OWNER, WORKFLOW_RUN_ID)
|
|
|
|
if len(failed_logs) == 0:
|
|
print("No runs exhibit signs of abrupt stoppage")
|
|
return
|
|
|
|
print("restarting job", WORKFLOW_RUN_ID)
|
|
r = restart_job(REPO_OWNER, WORKFLOW_RUN_ID)
|
|
print(r)
|
|
|
|
shutil.rmtree("logs")
|
|
os.remove("logs.zip")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
eval_jobs()
|