diff --git a/.build/ci/ci_parser.py b/.build/ci/ci_parser.py index c91b5f15b9c6..b0e69b46acdc 100755 --- a/.build/ci/ci_parser.py +++ b/.build/ci/ci_parser.py @@ -227,9 +227,7 @@ def create_summary_file(test_suites: Dict[str, JUnitTestSuite], xml_files, outpu JUnitResultBuilder.add_style_tags(soup) - # We cut off at 200 failures; if you have > than that chances are you have a bad run and there's no point in - # just continuing to pollute the summary file with it and blow past file size. Since the inlined failures are - # a tool to be used in the attaching / review process and not primarily workflow and fixing. + # Limit failure details by suite to keep the report small, but count every result. total_passed_count = 0 total_skipped_count = 0 total_failure_count = 0 @@ -252,15 +250,15 @@ def create_summary_file(test_suites: Dict[str, JUnitTestSuite], xml_files, outpu failures_builder.add_row(test.row_data()) failures_list_tag.append(BeautifulSoup(failures_builder.build_list(), 'html.parser')) failures_tag.append(BeautifulSoup(failures_builder.build_table(), 'html.parser')) - total_failure_count += failure_count - if total_failure_count > 200: - logger.critical(f'Saw {total_failure_count} failures; greater than 200 threshold. Not appending further failure details to {output}.') + if total_failure_count + failure_count > 200: + logger.critical(f'Saw {total_failure_count + failure_count} failures; greater than 200 threshold. Not appending further failure details to {output}.') + total_failure_count += failure_count total_passed_count += passed_count total_skipped_count += skipped_count # totals, manual html totals_tag = soup.new_tag("div") - totals_tag.string = f"""[Totals]

+ totals_tag.string = f"""[Totals]

diff --git a/.build/ci/ci_parser_test.py b/.build/ci/ci_parser_test.py new file mode 100644 index 000000000000..28450401a827 --- /dev/null +++ b/.build/ci/ci_parser_test.py @@ -0,0 +1,65 @@ +#!/usr/bin/env python3 +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you 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. + +import subprocess +import sys +import tempfile +import unittest +import xml.etree.ElementTree as ET +from pathlib import Path + +from bs4 import BeautifulSoup + + +class TestSummaryTotals(unittest.TestCase): + def report(self, failure_counts): + with tempfile.TemporaryDirectory() as directory: + directory = Path(directory) + for index, failures in enumerate(failure_counts): + suite = ET.Element('testsuite', name=f'suite-{index}') + for number in range(failures): + test = ET.SubElement(suite, 'testcase', classname='ExampleTest', name=f'failure-{number}', time='1') + ET.SubElement(test, 'failure', message='failure detail') + ET.SubElement(suite, 'testcase', classname='ExampleTest', name='passed', time='1') + skipped = ET.SubElement(suite, 'testcase', classname='ExampleTest', name='skipped', time='0') + ET.SubElement(skipped, 'skipped') + ET.ElementTree(suite).write(directory / f'suite-{index}.xml') + output = directory / 'ci_summary.html' + result = subprocess.run([sys.executable, str(Path(__file__).with_name('ci_parser.py').resolve()), + '--input', str(directory), '--output', str(output), '--mute'], + cwd=directory, capture_output=True, text=True, check=False) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + return BeautifulSoup(output.read_text(), 'html.parser') + + def test_detail_limit_does_not_cap_totals(self): + for first_suite_failures in (200, 216): + with self.subTest(first_suite_failures=first_suite_failures): + report = self.report((first_suite_failures, 528 - first_suite_failures)) + totals = report.find('table') + counts = {cells[0].get_text(strip=True): int(cells[-1].get_text(strip=True)) + for row in totals.find_all('tr') if (cells := row.find_all('td'))} + self.assertEqual(counts['Failed'], 528) + self.assertEqual(counts['Total'], 532) + self.assertEqual(counts['Passed'], 2) + self.assertEqual(counts['Skipped'], 2) + self.assertEqual(totals.get('data-failure-count-capped'), 'false') + tables = report.find_all('table')[1:] + self.assertEqual([table.find('th').get_text(strip=True) for table in tables], ['suite-0', 'Suites']) + + +if __name__ == '__main__': + unittest.main() diff --git a/.build/run-ci b/.build/run-ci index 9125f1818007..6ff7c636ed77 100755 --- a/.build/run-ci +++ b/.build/run-ci @@ -41,16 +41,19 @@ import os import re import shutil import socket +import stat import subprocess import sys import tarfile +import tempfile import threading import time +import warnings from contextlib import contextmanager from enum import Enum from pathlib import Path from urllib.error import HTTPError -from urllib.request import urlretrieve +from urllib.parse import quote, unquote, urlsplit, urlunsplit from typing import Optional, Tuple # External Libraries (`pip install -r .build/run-ci.d/requirements.txt`) @@ -75,6 +78,8 @@ def base_job_name(args) -> str: Separate jobs are required because Jenkinsfiles are baked into the job configuration. ref: .jenkins/k8s/jenkins-deployment.yaml JCasC.configScripts.test-job """ + if getattr(args, "job", None): + return args.job if not hasattr(base_job_name, "_cached_result"): raw_url = args.repository.replace("https://github.com/", "https://raw.githubusercontent.com/").removesuffix(".git") + f"/{args.branch}/build.xml" if 200 != requests.head(raw_url, timeout=30).status_code: @@ -155,6 +160,7 @@ DEFAULT_CONTAINER_NAME = "jenkins" JENKINS_API_ERRORS = (jenkins.JenkinsException, requests.exceptions.ConnectionError) # number of consecutive failed polls before the script warns the user POLL_FAILURES_BEFORE_WARNING = 10 +SPLIT_PROGRESS_POLL_SECONDS = 15 # errors urlretrieve raises, including HTTPError (a subclass of OSError) on a missing artifact DOWNLOAD_URL_ERRORS = (requests.exceptions.RequestException, IOError) LOCAL_RESULTS_BASEDIR = CASSANDRA_DIR / "build/ci/" @@ -181,28 +187,134 @@ def load_environment_file(): print("Warning: .build/run-ci.env file not found, or dotenv module not installed.") def setup_environment(kubeconfig, kubecontext) -> client.CoreV1Api: - """Ensures necessary tools are installed and sets up Kubernetes configuration.""" + """Sets up the Kubernetes API client; CLI tools are checked only when invoked.""" # Check Python version required_version = (3, 7) if sys.version_info < required_version: raise EnvironmentError(f"Python {required_version[0]}.{required_version[1]} or higher is required. " f"Current version is {sys.version_info.major}.{sys.version_info.minor}.") - # check command line dependencies - dependencies = ["helm", "kubectl"] - for cmd in dependencies: - if not shutil.which(cmd): - raise EnvironmentError(f"{cmd} must be installed and available in the PATH.") - # Initialize Kubernetes client and API instance config.load_kube_config(config_file=kubeconfig if kubeconfig else None, context=kubecontext or None) return client.CoreV1Api() +def require_command(command): + if not shutil.which(command): + raise EnvironmentError(f"{command} must be installed and available in the PATH for this operation.") + + +def normalize_jenkins_url(value): + """Keep scheme and context path, and exclude credentials from URLs and log messages.""" + parts = urlsplit(value if "://" in value else "http://" + value) + if parts.scheme not in ("http", "https") or not parts.hostname: + raise ValueError("Use a Jenkins HTTP(S) URL with a hostname.") + if parts.username is not None or parts.password is not None or parts.query or parts.fragment: + raise ValueError("Use a Jenkins URL without embedded credentials, query or fragment.") + # Accessing port also validates malformed port numbers. + port = parts.port + if port == 0: + raise ValueError("Jenkins URL port must be greater than zero.") + host = f"[{parts.hostname}]" if ":" in parts.hostname else parts.hostname + if port and (parts.scheme, port) not in (("http", 80), ("https", 443)): + host += f":{port}" + return urlunsplit((parts.scheme, host, parts.path.rstrip("/"), "", "")) + + +def default_credentials_file(): + directory = Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config") + if not directory.is_absolute(): + directory = Path.home() / ".config" + return directory / "cassandra" / "run-ci-credentials.json" + + +def read_credentials_file(path): + """Read an owner-only regular file without following a final symlink.""" + path = Path(path).expanduser() + try: + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + except FileNotFoundError: + return {} + with os.fdopen(descriptor, encoding="utf-8") as handle: + info = os.fstat(handle.fileno()) + if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() + or info.st_nlink != 1 or stat.S_IMODE(info.st_mode) & 0o077): + raise ValueError("Credentials must be an owned regular file with permissions 0600 and no links.") + try: + entries = json.load(handle) + except (ValueError, UnicodeError) as error: + raise ValueError("Invalid credentials file JSON.") from error + if not isinstance(entries, dict) or any( + not isinstance(url, str) or not isinstance(entry, dict) + or not isinstance(entry.get("user"), str) or not entry["user"] + or not isinstance(entry.get("token"), str) or not entry["token"] + for url, entry in entries.items()): + raise ValueError("Invalid credentials file entries; expected URL, user and token.") + return entries + + +def load_credentials(path, url): + return read_credentials_file(path).get(normalize_jenkins_url(url)) + + +def save_credentials(path, url, user, token): + """Replace the file atomically; temporary and final files are accessible only to their owner.""" + path = Path(path).expanduser() + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + parent = path.parent.lstat() + if not stat.S_ISDIR(parent.st_mode) or parent.st_uid != os.getuid() or parent.st_mode & 0o022: + raise ValueError("Credentials directory must be owned by you and not writable by other users.") + entries = read_credentials_file(path) + entries[normalize_jenkins_url(url)] = {"user": user, "token": token} + descriptor, temporary = tempfile.mkstemp(prefix=".run-ci-credentials-", dir=path.parent) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + os.fchmod(handle.fileno(), 0o600) + json.dump(entries, handle, indent=2) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + + +def parse_attach(args, parser): + if not args.attach: + return + if any((args.setup, args.only_setup, args.tear_down, args.only_tear_down, + args.only_node_cleaner, args.download_results)): + parser.error("--attach cannot be combined with setup, teardown, node cleaner or --download-results.") + if args.attach.isdecimal(): + if int(args.attach) < 1: + parser.error("--attach requires a positive build number.") + args.attach = int(args.attach) + return + url = urlsplit(normalize_jenkins_url(args.attach)) + match = re.fullmatch(r"(.*?)((?:/job/[^/]+)+)/([1-9][0-9]*)", url.path) + if not match: + parser.error("--attach needs a build URL ending in /job/JOB/NUMBER/ or a positive build number.") + base = urlunsplit((url.scheme, url.netloc, match[1], "", "")) + job = "/".join(unquote(name) for name in match[2].split("/job/")[1:]) + if args.url and normalize_jenkins_url(args.url) != base: + parser.error("--url and --attach name different Jenkins servers.") + if args.job and args.job != job: + parser.error("--job and --attach name different jobs.") + args.url, args.job, args.attach = base, job, int(match[3]) + + def argument_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Run CI pipeline for Cassandra on K8s using Jenkins.") parser.add_argument("-c", "--kubeconfig", help="Path to a different kubeconfig.") parser.add_argument("-x", "--kubecontext", help="Use a different Kubernetes context.") parser.add_argument("-i", "--url", help="Jenkins url. Suitable when kubectl access in not available. Can also be specified via the JENKINS_URL environment variable (and in .build/.run-ci.env)") parser.add_argument("-u", "--user", help="Jenkins user. Can also be specified via the JENKINS_USER environment variable (and in .build/.run-ci.env)") + parser.add_argument("--credentials-file", type=Path, default=default_credentials_file(), + help="Credentials file (default: $XDG_CONFIG_HOME/cassandra/run-ci-credentials.json or ~/.config/cassandra/run-ci-credentials.json).") + parser.add_argument("--save-credentials", action="store_true", + help="Save or replace credentials after authentication succeeds. Prompts for an API token or password without echoing it.") + parser.add_argument("--attach", metavar="BUILD", + help="Reconnect to a build number or URL. Infer the job from the repository and branch, as for a new build. Wait for completion and download results.") + parser.add_argument("--job", help="Override the inferred Jenkins job name, including folders.") parser.add_argument("-r", "--repository", default=DEFAULT_REPO_URL, help="Repository URL. Defaults to current tracking remote.") parser.add_argument("-b", "--branch", default=DEFAULT_REPO_BRANCH, help="Repository branch. Defaults to current branch.") parser.add_argument("-p", "--profile", choices=['packaging','skinny','pre-commit','pre-commit w/ upgrades','post-commit','custom'], default=DEFAULT_PROFILE, help="CI pipeline profile. Defaults to skinny.") @@ -224,12 +336,24 @@ def parse_arguments() -> argparse.Namespace: Parses command-line arguments and sets environment variables based on inputs. If you update this please also update `.build/run-ci.d/README.md` """ - args = argument_parser().parse_args() + parser = argument_parser() + args = parser.parse_args() + try: + parse_attach(args, parser) + if not args.url: + args.url = os.environ.get("JENKINS_URL") + if args.url: + args.url = normalize_jenkins_url(args.url) + except ValueError as error: + parser.error(str(error)) + if args.job and any(part in ("", ".", "..") for part in args.job.split("/")): + parser.error("--job must contain nonempty job names, separated by / for folders.") # -r defaults to the branch's tracking remote, which is absent when the branch tracks nothing. Only the # flows that build, or that name artefacts after a build, need it: installing and tearing down do not, # so an untracked branch can still deploy the cluster. - if not args.repository and not (args.only_setup or args.only_tear_down or args.only_node_cleaner): + if not args.repository and not (args.only_setup or args.only_tear_down or args.only_node_cleaner + or (args.attach and args.job)): require_tracking_remote(args.branch) assert not args.repository or (args.repository.startswith("https://github.com/") @@ -243,14 +367,11 @@ def parse_arguments() -> argparse.Namespace: assert not (args.values_override and not (args.setup or args.only_setup)), "--values-override requires --setup or --only-setup." assert not (args.values_override and not Path(args.values_override).is_file()), f"No such values override file: {args.values_override}" - if not args.url and os.environ.get("JENKINS_URL"): - args.url = os.environ.get("JENKINS_URL") if not args.user and os.environ.get("JENKINS_USER"): args.user = os.environ.get("JENKINS_USER") assert not (args.url and (args.kubeconfig or args.kubecontext or args.setup or args.only_setup or args.tear_down or args.only_tear_down or args.only_node_cleaner)),\ "Cannot specify both --url and any of --kubeconfig/--kubecontext/--setup/--only-setup/--tear-down/--only-tear-down/--only-node-cleaner. Setting the jenkins url implies no kubectl actions." - assert not (args.url and not args.user), "When specifying --url, --user is required." return args @@ -274,6 +395,7 @@ def run_kubectl_command(kubeconfig: Optional[str], kubecontext: Optional[str], k Runs a kubectl command with the specified kubeconfig and context. Used when functionality is not available in k8s_client. """ + require_command("kubectl") cmd = ["kubectl"] if kubeconfig: cmd += ["--kubeconfig", kubeconfig] @@ -286,6 +408,7 @@ def run_kubectl_command(kubeconfig: Optional[str], kubecontext: Optional[str], k def run_helm_command(kubeconfig: Optional[str], kubecontext: Optional[str], kube_ns: str, command: list, capture_output: bool = True, check: bool = True) -> subprocess.CompletedProcess: """Runs a helm command with the specified kubeconfig, context and namespace.""" + require_command("helm") cmd = ["helm"] if kubeconfig: cmd += ["--kubeconfig", kubeconfig] @@ -563,7 +686,7 @@ def install_jenkins(kubeconfig: Optional[str], kubecontext: Optional[str], kube_ print("Adding Helm repository for Jenkins Operator...") subprocess.run(["helm", "repo", "add", "jenkins", "https://charts.jenkins.io"], check=True) - subprocess.run(["helm", "repo", "update"], check=True) + subprocess.run(["helm", "repo", "update", "jenkins"], check=True) # site customisations are applied last, helm merges each -f over the previous values_files = ["-f", DEPLOY_YAML] + (["-f", values_override] if values_override else []) @@ -572,20 +695,24 @@ def install_jenkins(kubeconfig: Optional[str], kubecontext: Optional[str], kube_ # succeeding. Keep it clear of that budget plus the readinessProbe's, see jenkins-deployment.yaml result = run_helm_command(kubeconfig, kubecontext, kube_ns, ["upgrade", "--install"] + values_files - + ["cassius", "jenkins/jenkins", "--wait", "--timeout", "10m"]) + + ["cassius", "jenkins/jenkins", "--wait", "--timeout", "10m"], check=False) + if result.stdout: + print(result.stdout, end="") + if result.stderr: + print(result.stderr, end="", file=sys.stderr) + if result.returncode != 0: + print("Failed to install Jenkins using Helm.", file=sys.stderr) + sys.exit(result.returncode) run_kubectl_command(kubeconfig, kubecontext, kube_ns, ["exec", DEFAULT_POD_NAME, "--", "curl", "-sS", "https://www.apache.org/logos/originals/cassandra-4.svg", "-o", "/var/jenkins_cache/war/images/svgs/logo.svg"]) - if result.returncode != 0: - print("Failed to install Jenkins Operator using Helm. Check the configuration and/or `kubectl logs cassius-jenkins-0`.") - sys.exit(1) - def wait_for_jenkins_http(ip: str): - host, port = (ip.rsplit(":", 1)[0], int(ip.rsplit(":", 1)[1])) if ":" in ip else (ip, 80) + endpoint = urlsplit(normalize_jenkins_url(ip)) + host, port = endpoint.hostname, endpoint.port or (443 if endpoint.scheme == "https" else 80) spin_while(f"Waiting for Jenkins HTTP at {host}:{port}… ", lambda: _tcp_connect_ok(host, port)) @@ -597,34 +724,87 @@ def _tcp_connect_ok(host: str, port: int) -> bool: return False +def jenkins_service_url(service) -> str: + """Prefer the Service's public DNS name, retaining its load-balancer address as a fallback.""" + annotations = (service.metadata.annotations or {}) if service.metadata else {} + hostname = None + label = r"[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?" + for key in ("external-dns.kubernetes.io/hostname", "external-dns.alpha.kubernetes.io/hostname"): + for candidate in annotations.get(key, "").split(","): + candidate = candidate.strip().removesuffix(".") + if len(candidate) <= 253 and re.fullmatch(rf"{label}(?:\.{label})*", candidate): + hostname = candidate + break + if hostname: + break + + address = hostname + if not address: + balancer = service.status.load_balancer if service.status else None + for ingress in (balancer.ingress or []) if balancer else []: + address = ingress.ip or ingress.hostname + if address: + break + if not address: + raise ValueError("Unable to retrieve Jenkins Service hostname or load-balancer address") + + ports = [port for port in service.spec.ports or [] if port.protocol in (None, "TCP")] + if not ports: + raise ValueError("Unable to retrieve Jenkins Service TCP port") + ssl_cert = annotations.get("service.beta.kubernetes.io/aws-load-balancer-ssl-cert") + ssl_ports_value = annotations.get("service.beta.kubernetes.io/aws-load-balancer-ssl-ports") + ssl_ports = {port.strip() for port in ssl_ports_value.split(",")} if ssl_ports_value is not None else None + + def is_tls(port): + return (port.name == "https" or port.port == 443 + or bool(ssl_cert and (ssl_ports is None or port.name in ssl_ports or str(port.port) in ssl_ports))) + + secure = sorted((port for port in ports if is_tls(port)), key=lambda port: port.port != 443) + plain = sorted((port for port in ports if not is_tls(port)), key=lambda port: port.name != "http" and port.port != 80) + # The raw ELB name generally does not match the site's certificate; keep HTTP as its fallback. + port = (secure + plain if hostname else plain + secure)[0] + scheme = "https" if is_tls(port) else "http" + if ":" in address: + address = f"[{address}]" + return normalize_jenkins_url(f"{scheme}://{address}:{port.port}") + + def get_jenkins(k8s_client: client.CoreV1Api, args, kube_ns: str) -> Tuple[str, jenkins.Jenkins]: - """Authenticates to Jenkins and returns the Jenkins ip and server objects.""" - - def get_jenkins_ip(k8s_client, kube_ns: str) -> str: - svc = k8s_client.read_namespaced_service("cassius-jenkins", kube_ns) - if svc.status.load_balancer.ingress: - # the best we can do is the public IP or hostname of the controller, which may not be the common public url - ingress = svc.status.load_balancer.ingress[0] - ip = ingress.ip if ingress.ip else ingress.hostname - if svc.spec.ports[0].port != 80: - ip += ":" + str(svc.spec.ports[0].port) - print(f"Jenkins: {ip}\n---") - return ip - raise ValueError("Unable to retrieve Jenkins IP address") - - def prompt_for_password(): - return getpass.getpass("Enter Jenkins password: ") - - kubeconfig = args.kubeconfig - kubecontext = args.kubecontext - user = args.user if args.user else "admin" - ip = args.url if args.url else get_jenkins_ip(k8s_client, kube_ns) - - password = prompt_for_password() if args.user \ - else run_kubectl_command(kubeconfig, kubecontext, kube_ns, ["exec", DEFAULT_POD_NAME, "--", "cat", "/run/secrets/additional/chart-admin-password"]) - # Initialize Jenkins API clien - server = jenkins.Jenkins(f"http://{ip}", username=user, password=password) - return ip, server + """Authenticates to Jenkins and returns its base URL and API client.""" + if args.url: + url = normalize_jenkins_url(args.url) + else: + url = jenkins_service_url(k8s_client.read_namespaced_service("cassius-jenkins", kube_ns)) + print(f"Jenkins: {url}\n---") + credentials_file = getattr(args, "credentials_file", default_credentials_file()) + saved = load_credentials(credentials_file, url) + user = args.user or (saved["user"] if saved else None) + saving = getattr(args, "save_credentials", False) + if saved and saved["user"] == user and not saving: + password = saved["token"] + elif args.url or args.user or saved: + if not user: + raise ValueError("Specify --user for this Jenkins URL; --save-credentials remembers it for later runs.") + with warnings.catch_warnings(): + warnings.simplefilter("error", getpass.GetPassWarning) + try: + password = getpass.getpass("Enter Jenkins API token or password: ") + except getpass.GetPassWarning as error: + raise ValueError("Cannot read credentials without echo; use a terminal or saved credentials.") from error + else: + user = "admin" + password = run_kubectl_command(args.kubeconfig, args.kubecontext, kube_ns, + ["exec", DEFAULT_POD_NAME, "--", "cat", "/run/secrets/additional/chart-admin-password"]) + if not password: + raise ValueError("Jenkins credentials cannot be empty.") + server = jenkins.Jenkins(url, username=user, password=password) + if saving: + if server.get_whoami().get("authenticated") is not True: + raise jenkins.JenkinsException("Jenkins did not authenticate these credentials; nothing was saved.") + save_credentials(credentials_file, url, user, password) + print(f"Saved Jenkins credentials to {credentials_file} (owner access only).") + args.user = user + return url, server def ensure_job_parameters_visible(server: jenkins.Jenkins, job_name: str): @@ -711,6 +891,93 @@ def wait_for_build_number(server: jenkins.Jenkins, queue_item: int) -> int: return build_number +def count_test_splits(tree): + """Count the outer parallel branches under Tests, including tasks with just one split.""" + if not isinstance(tree, dict) or tree.get("status") != "ok": + raise ValueError("Invalid pipeline graph response") + stages = tree["data"]["stages"] + if not isinstance(stages, list): + raise ValueError("Invalid pipeline stages") + counts = dict.fromkeys(("total", "finished", "failed", "unstable", "aborted", "skipped", + "running", "queued", "paused", "unknown"), 0) + terminal = {"success", "failure", "unstable", "aborted", "skipped", "not_built", "finished"} + pending = [(stage, False) for stage in stages] + seen = set() + while pending: + node, in_tests = pending.pop() + if not isinstance(node, dict) or not isinstance(node.get("id"), str): + raise ValueError("Invalid pipeline node") + if node["id"] in seen: + continue + seen.add(node["id"]) + if node.get("nextSibling") is not None: + pending.append((node["nextSibling"], in_tests)) + if node.get("placeholder"): + continue + in_tests = in_tests or (node.get("type") == "STAGE" and node.get("name") == "Tests") + if in_tests and node.get("type") == "PARALLEL": + counts["total"] += 1 + status = node.get("state", "unknown") + if status in terminal: + counts["finished"] += 1 + if status not in ("success", "finished"): + counts[{"failure": "failed", "not_built": "skipped"}.get(status, status)] += 1 + else: + counts[status if status in ("running", "queued", "paused") else "unknown"] += 1 + # The branch encloses retries and setup. Their completion does not finish the split. + continue + children = node.get("children", []) + if not isinstance(children, list): + raise ValueError("Invalid pipeline children") + pending.extend((child, in_tests) for child in children) + return counts if counts["total"] else None + + +def format_split_progress(counts): + if counts is None: + return "Splits: waiting for test branches" + text = f"Splits: {counts['finished']}/{counts['total']} finished" + outcomes = [f"{counts[key]} {key}" for key in ("failed", "unstable", "aborted", "skipped") if counts[key]] + if outcomes: + text += " (" + ", ".join(outcomes) + ")" + text += f" | {counts['running']} running | {counts['queued']} queued" + for key in ("paused", "unknown"): + if counts[key]: + text += f" | {counts[key]} {key}" + return text + + +# One polling interface owns the refresh state. +class SplitProgress: # pylint: disable=too-few-public-methods + """Read optional graph data less often than the inexpensive build status poll.""" + def __init__(self, server, job, number): + self.server = server + self.url = jenkins_build_url(server.server, job, number) if isinstance(server.server, str) else None + self.path = "stages/tree" + self.next_poll = 0 + self.text = "Splits: unavailable" + + def update(self, complete=False): + now = time.monotonic() + if self.url is None or (not complete and now < self.next_poll): + return + self.next_poll = now + SPLIT_PROGRESS_POLL_SECONDS + try: + try: + response = self.server.jenkins_open(requests.Request("GET", f"{self.url}/{self.path}"), add_crumb=False) + except jenkins.NotFoundException: + path = "pipeline-overview/tree" if self.path == "stages/tree" else "stages/tree" + response = self.server.jenkins_open(requests.Request("GET", f"{self.url}/{path}"), add_crumb=False) + self.path = path + counts = count_test_splits(json.loads(response)) + self.text = "Splits: no test branches" if counts is None and complete else format_split_progress(counts) + except (jenkins.JenkinsException, requests.exceptions.RequestException, ValueError, TypeError, KeyError) as error: + self.text = "Splits: unavailable" + if isinstance(error, jenkins.NotFoundException): + self.next_poll = now + 60 + debug(f"Split progress unavailable: {error}") + + def wait_for_build_complete(server: jenkins.Jenkins, job_name: str, build_number: int): """ Waits for Jenkins build completion by monitoring the build status. @@ -721,15 +988,19 @@ def wait_for_build_complete(server: jenkins.Jenkins, job_name: str, build_number """ start_time = time.time() state = {"info": {}, "failures": 0, "latched": False} + progress = SplitProgress(server, job_name, build_number) def is_complete() -> bool: info = get_build_info(server, job_name, build_number) if not info: + progress.text = "Splits: unavailable (Jenkins connection lost)" warn_on_repeated_failures(state, f"{job_name} build {build_number}") return False state["failures"] = 0 state["info"] = info result = info.get('result') + complete = info.get('building', True) is False and result is not None + progress.update(complete=complete) # default `building` to True: never read a missing field as finished if info.get('building', True): if result is not None and not state["latched"]: @@ -740,26 +1011,29 @@ def wait_for_build_complete(server: jenkins.Jenkins, job_name: str, build_number return False return result is not None - elapsed_time = spin_while("Waiting for build to complete… ", is_complete) + elapsed_time = spin_while(lambda: progress.text, is_complete) minutes, seconds = divmod(elapsed_time, 60) - print(f"\r---\nBuild completed after {minutes:02}:{seconds:02} with status: {state['info'].get('result')}") + print(f"{progress.text}\n---\nBuild completed after {minutes:02}:{seconds:02} with status: {state['info'].get('result')}") def spin_while(message="", is_complete=lambda: False) -> int: spinner = itertools.cycle(['|', '/', '-', '\\']) start_time = time.time() elapsed_time = 0 - while not is_complete(): - elapsed_time = int(time.time() - start_time) - minutes, seconds = divmod(elapsed_time, 60) - for _ in range(10): - sys.stdout.write(f"\r{message} {minutes:02}:{seconds:02} {next(spinner)}\033[?25l") - sys.stdout.flush() - time.sleep(0.3) - sys.stdout.write("\r" + " " * len(message + " \033[?25h")) - sys.stdout.flush() - return elapsed_time + try: + while not is_complete(): + for _ in range(10): + elapsed_time = int(time.time() - start_time) + minutes, seconds = divmod(elapsed_time, 60) + text = message() if callable(message) else message + sys.stdout.write(f"\r\033[K{text} | {minutes:02}:{seconds:02} {next(spinner)}\033[?25l") + sys.stdout.flush() + time.sleep(0.3) + return int(time.time() - start_time) + finally: + sys.stdout.write("\r\033[K\033[?25h") + sys.stdout.flush() def node_cleaner(k8s_client: client.CoreV1Api, kubeconfig: Optional[str], kubecontext: Optional[str], kube_ns: str): """ @@ -768,8 +1042,22 @@ def node_cleaner(k8s_client: client.CoreV1Api, kubeconfig: Optional[str], kubeco Env variables (per cloud provider): AWS_REGION, GCP_PROJECT_ID, GCP_ZONE """ + stopped = threading.Event() + stop_lock = threading.Lock() + def keep_running() -> bool: - return bool(globals().get("IS_RUNNING", True)) + return not stopped.is_set() and bool(globals().get("IS_RUNNING", True)) + + def stop_on_error(error): + with stop_lock: + if stopped.is_set(): + return + stopped.set() + detail = getattr(error, "stderr", None) or str(error) + print(f"\nNode cleaner stopped after a cluster access failure: {detail}\n" + "Jenkins build monitoring is unaffected. Restore cluster access before restarting " + "--only-node-cleaner; for AWS login credentials, run `aws login` with the same profile.", + file=sys.stderr) def node_cleaner_debug(msg: str): if os.environ.get("NODE_CLEANER_DEBUG"): @@ -792,29 +1080,32 @@ def node_cleaner(k8s_client: client.CoreV1Api, kubeconfig: Optional[str], kubeco k8s_client.patch_node(name=node_name, body={"spec": {"unschedulable": True}}) node_cleaner_debug(f"Node {node_name} cordoned.") except client.exceptions.ApiException as e: - node_cleaner_debug(f"Failed to cordon node {node_name}: {e}") + stop_on_error(e) def drain_node(node_name: str): try: pods = k8s_client.list_pod_for_all_namespaces(field_selector=f"spec.nodeName={node_name}") for pod in pods.items: + if not keep_running(): + return owner_refs = pod.metadata.owner_references or [] # Delete only non-DaemonSet pods if not any(ref.kind == "DaemonSet" for ref in owner_refs): try: k8s_client.delete_namespaced_pod(name=pod.metadata.name, namespace=pod.metadata.namespace) except client.exceptions.ApiException as e: - node_cleaner_debug(f"Failed to delete pod {pod.metadata.name} on {node_name}: {e}") + stop_on_error(e) + return node_cleaner_debug(f"Node {node_name} drained (and all non-DaemonSet pods deleted).") except client.exceptions.ApiException as e: - node_cleaner_debug(f"Failed to drain node {node_name}: {e}") + stop_on_error(e) def delete_k8s_node(node_name: str): try: k8s_client.delete_node(node_name) node_cleaner_debug(f"Node {node_name} deleted from Kubernetes API.") except client.exceptions.ApiException as e: - node_cleaner_debug(f"Failed to delete node {node_name} from K8s API: {e}") + stop_on_error(e) def get_first_node_provider_id() -> Optional[str]: try: @@ -825,7 +1116,8 @@ def node_cleaner(k8s_client: client.CoreV1Api, kubeconfig: Optional[str], kubeco if n.spec and n.spec.provider_id: return n.spec.provider_id return None - except client.exceptions.ApiException: + except client.exceptions.ApiException as e: + stop_on_error(e) return None def detect_cloud_provider(kubeconfig: Optional[str], kubecontext: Optional[str], kube_ns: str, node_name: str) -> Tuple[CloudProvider, str]: @@ -833,13 +1125,15 @@ def node_cleaner(k8s_client: client.CoreV1Api, kubeconfig: Optional[str], kubeco try: node_obj = k8s_client.read_node(node_name) except client.exceptions.ApiException as e: - node_cleaner_debug(f"Failed to read node {node_name}: {e}") + stop_on_error(e) return CloudProvider.UNKNOWN, None - provider_id = getattr(node_obj.spec, "provider_id", None).lower() + provider_id = (getattr(node_obj.spec, "provider_id", None) or "").lower() if not provider_id: - provider_id = get_first_node_provider_id().lower() + provider_id = (get_first_node_provider_id() or "").lower() + if not keep_running(): + return CloudProvider.UNKNOWN, None if provider_id: if provider_id.startswith("aws:"): @@ -856,8 +1150,8 @@ def node_cleaner(k8s_client: client.CoreV1Api, kubeconfig: Optional[str], kubeco return CloudProvider.AWS, provider_id if "gke_" in ctx or "gke" in ctx: return CloudProvider.GCP, provider_id - except subprocess.CalledProcessError: - debug(f"failed to determine provider_id: {e}") + except (subprocess.CalledProcessError, OSError) as e: + stop_on_error(e) return CloudProvider.UNKNOWN, None def parse_aws_provider_id(provider_id: str) -> Tuple[Optional[str], Optional[str]]: @@ -949,21 +1243,25 @@ def node_cleaner(k8s_client: client.CoreV1Api, kubeconfig: Optional[str], kubeco if is_node_in_use(kubeconfig, kubecontext, kube_ns, node_name): node_cleaner_debug(f"Node {node_name} in use [check {attempt}].") return - except (subprocess.CalledProcessError, client.exceptions.ApiException) as e: - node_cleaner_debug(f"Failed to inspect node {node_name} [check {attempt}]: {e}") + except (subprocess.CalledProcessError, OSError, client.exceptions.ApiException) as e: + stop_on_error(e) return # Don't delete nodes we can't inspect safely time.sleep(10) # 2) Determine provider + IDs from providerID of this node + if not keep_running(): + return cloud, provider_id = detect_cloud_provider(kubeconfig, kubecontext, kube_ns, node_name) # 3) Cordon & drain & delete K8s node (shared) - node_cleaner_debug(f"Deleting dangling node {node_name}…") - cordon_node(node_name) - drain_node(node_name) - delete_k8s_node(node_name) + for action in (cordon_node, drain_node, delete_k8s_node): + if not keep_running(): + return + action(node_name) # 4) Cloud-specific instance delete/terminate + if not keep_running(): + return if CloudProvider.AWS == cloud: instance_id, region = parse_aws_provider_id(provider_id) if not instance_id and node_name.startswith("ip-") and "." in node_name: @@ -979,26 +1277,30 @@ def node_cleaner(k8s_client: client.CoreV1Api, kubeconfig: Optional[str], kubeco # Main node_cleaner loop while keep_running(): try: - nodes = k8s_client.list_node().items + # kubectl propagates exec credential failures; the Python loader can log and swallow them. + nodes = json.loads(run_kubectl_command( + kubeconfig, kubecontext, kube_ns, ["get", "nodes", "-o", "json"]))["items"] + node_names = [n["metadata"]["name"] for n in nodes] node_cleaner_debug(f" {len(nodes)} nodes") - except client.exceptions.ApiException as e: - node_cleaner_debug(f"Failed to list nodes: {e}") - time.sleep(10) - continue + except (subprocess.CalledProcessError, OSError, ValueError, KeyError, TypeError) as e: + stop_on_error(e) + break active_threads = {t.name for t in threading.enumerate()} - for n in nodes: - node_name = n.metadata.name + for node_name in node_names: + if not keep_running(): + break # only act on nodes with "agent" in the name node_cleaner_debug(f"Checking node {node_name}…") if node_name not in active_threads: t = threading.Thread(target=check_and_cleanup_node, args=(node_name,), name=node_name, daemon=True) t.start() - time.sleep(10) + if keep_running(): + time.sleep(10) def delete_remote_junit_files(k8s_client, pod_name: str, kube_ns: str, base_job_name: str, build_number: int): debug("Cleaning remote individual JUnit XML files...") - exec_command = ['rm', '-rf', f'/var/jenkins_home/jobs/{base_job_name}/builds/{build_number}/archive/test/output'] + exec_command = ['rm', '-rf', f"/var/jenkins_home/jobs/{base_job_name.replace('/', '/jobs/')}/builds/{build_number}/archive/test/output"] stream.stream(k8s_client.connect_get_namespaced_pod_exec, pod_name, kube_ns, container=DEFAULT_CONTAINER_NAME, command=exec_command, stderr=True, stdin=False, stdout=True, tty=False, _preload_content=False) debug("Remote JUnit XML files cleaned.") @@ -1006,7 +1308,7 @@ def delete_remote_junit_files(k8s_client, pod_name: str, kube_ns: str, base_job_ def is_transient(error) -> bool: """A 404 means the artifact was never archived, so retrying it only wastes time.""" - return not (isinstance(error, HTTPError) and 404 == error.code) + return not (isinstance(error, jenkins.NotFoundException) or (isinstance(error, HTTPError) and 404 == error.code)) def retry_call(action, description: str, errors, max_retries: int = 5, delay: int = 5): @@ -1030,7 +1332,7 @@ def retry_call(action, description: str, errors, max_retries: int = 5, delay: in raise last_error -def download_results_and_print_summary(k8s_client, pod_name: str, kube_ns: str, build_number: int, ip: str, args): +def download_results_and_print_summary(k8s_client, pod_name: str, kube_ns: str, build_number: int, ip: str, args, server=None): def download_console_log(pod_name: str, container_name: str, kubeconfig: Optional[str], kubecontext: Optional[str], kube_ns: str, console_log_path: str, local_console_log: Path): # this runs in a thread, so it must report its own failure. an escaping @@ -1041,8 +1343,6 @@ def download_results_and_print_summary(k8s_client, pod_name: str, kube_ns: str, f"download {pod_name}:{console_log_path}", subprocess.CalledProcessError) except subprocess.CalledProcessError as e: print(f"WARNING: could not download the console log {pod_name}:{console_log_path}: {e}") - return - print(f"Console log saved to {local_console_log}.gz\n") def download_archive_tarball(kubeconfig: Optional[str], kubecontext: Optional[str], kube_ns: str, pod_name: str, container_name: str, remote_path: str, local_path, max_retries=5): retry_call(lambda: run_kubectl_command(kubeconfig, kubecontext, kube_ns, @@ -1066,34 +1366,41 @@ def download_results_and_print_summary(k8s_client, pod_name: str, kube_ns: str, def print_results_summary_console(local_console_log): if local_console_log.exists(): - with open(local_console_log, 'r', encoding="utf-8") as log_file: - log_content = log_file.read() - if "BUILD FAILED" in log_content: - print("---") - failed_index = log_content.index("BUILD FAILED") - # Print the 200 characters after "BUILD FAILED" - print(log_content[failed_index:failed_index + 200]) + result = None with open(local_console_log, 'r', encoding="utf-8") as log_file: for line in log_file: - if "Finished: " in line: - print(line.strip()) - break + match = re.fullmatch(r"Finished: (SUCCESS|UNSTABLE|FAILURE|ABORTED|NOT_BUILT)", line.strip()) + if match: + result = match[1] + print(f"Jenkins result: {result}" if result else "Jenkins result unavailable in console log.") else: print("Missing console log.") def print_results_summary_ci_summary(ci_summary_file): if ci_summary_file.exists(): with open(ci_summary_file, 'r', encoding="utf-8") as log_file: - summary_parts = [] - for line in log_file: - if any(l in line for l in [">Passed<", ">Failed<", ">Skipped<", ">Total<"]): - summary_parts.append(BeautifulSoup(line, 'html.parser').get_text().strip()) - if ">Total<" in line: - break - if summary_parts: - print(" – ".join(summary_parts)) + report = BeautifulSoup(log_file, 'html.parser') + labels = ('Passed', 'Failed', 'Skipped', 'Total') + for table in report.find_all('table'): + totals = {} + for row in table.find_all('tr'): + cells = row.find_all('td', recursive=False) + if len(cells) == 3: + label, count = cells[0].get_text(strip=True), cells[-1].get_text(strip=True) + if label in labels and count.isdecimal(): + totals[label] = int(count) + if all(label in totals for label in labels): + # Older reports stopped counting failures once the detail limit was reached. + capped = totals['Failed'] >= 200 and table.get('data-failure-count-capped') != 'false' + summary_parts = [f"{label} {totals[label]}{'+' if capped and label in ('Failed', 'Total') else ''}" + for label in labels] + print("Tests: " + " | ".join(summary_parts)) + if capped: + print("Counts marked + are lower bounds; this older CI summary may have capped failure totals.") + return + print("Test totals unavailable in CI summary.") else: - print("No tests were run (or missing summary file).") + print("Test totals unavailable: CI summary missing.") def print_results_summary(local_console_log, ci_summary_file): print("--- Build Summary ---") @@ -1104,23 +1411,44 @@ def download_results_and_print_summary(k8s_client, pod_name: str, kube_ns: str, with open(local_console_log, 'rb') as f_in, gzip.open(f"{local_console_log}.gz", 'wb') as f_out: f_out.writelines(f_in) os.remove(local_console_log) + print(f"Console log saved as {local_console_log}.gz") def download_url(url, dest, max_retries=5): - retry_call(lambda: urlretrieve(url, dest), f"download {url}", DOWNLOAD_URL_ERRORS, max_retries) + def transfer(): + temporary = None + try: + with server.jenkins_request(requests.Request("GET", url), add_crumb=False, stream=True) as response: + with tempfile.NamedTemporaryFile(dir=dest.parent, delete=False) as output: + temporary = Path(output.name) + for chunk in response.iter_content(chunk_size=1024 * 1024): + output.write(chunk) + temporary.replace(dest) + finally: + if temporary is not None and temporary.exists(): + temporary.unlink() + retry_call(transfer, f"download {url}", DOWNLOAD_URL_ERRORS + JENKINS_API_ERRORS, max_retries) debug(f" saved {dest}") - local_results_dir = LOCAL_RESULTS_BASEDIR / ip.replace(".", "-") / str(build_number) + endpoint = normalize_jenkins_url(ip) + job_name = base_job_name(args) + server_url = urlsplit(endpoint) + server_directory = quote(server_url.netloc + server_url.path, safe=":[]") + if server_directory in (".", ".."): + server_directory = server_directory.replace(".", "%2E") + local_results_dir = LOCAL_RESULTS_BASEDIR / server_directory / quote(job_name, safe="") / str(build_number) local_results_dir.mkdir(parents=True, exist_ok=True) - repo_owner = args.repository.split('/')[3] if 'https' in args.repository else args.repository.split(':')[1].split('/')[0] - ci_summary_file = local_results_dir / f"ci_summary_{repo_owner}_{args.branch.replace('/', '-')}_{build_number}.html" - ci_details_file = local_results_dir / f"results_details_{repo_owner}_{args.branch.replace('/', '-')}_{build_number}.tar.xz" - if args.url: - artifact_url = f"http://{ip}/job/{base_job_name(args)}/{build_number}/artifact" + repo_owner = urlsplit(args.repository).path.strip("/").split("/")[0] if args.repository else "unknown" + suffix = f"{quote(repo_owner, safe='')}_{quote((args.branch or 'unknown').replace('/', '-'), safe='')}_{build_number}" + ci_summary_file = local_results_dir / f"ci_summary_{suffix}.html" + ci_details_file = local_results_dir / f"results_details_{suffix}.tar.xz" + if args.url or args.attach: + build_url = jenkins_build_url(endpoint, job_name, build_number) + artifact_url = f"{build_url}/artifact" for url, dest in [(f"{artifact_url}/ci_summary.html", ci_summary_file), (f"{artifact_url}/results_details.tar.xz", ci_details_file)]: try: download_url(url, dest) - except DOWNLOAD_URL_ERRORS as e: + except DOWNLOAD_URL_ERRORS + JENKINS_API_ERRORS as e: print(f" Could not download {url}: {e}") print(" A build that was aborted or that failed skips the Summary stage," " and archives no artifacts.") @@ -1130,14 +1458,18 @@ def download_results_and_print_summary(k8s_client, pod_name: str, kube_ns: str, print(f"Details file saved as {ci_details_file}") if (ci_summary_file).exists() or (ci_details_file).exists(): print(" (attach ci_summary….html and results_details….tar.xz to the JIRA ticket)") - print("--- Build Summary ---") - print_results_summary_ci_summary(ci_summary_file) + local_console_log = local_results_dir / "console_log.txt" + try: + download_url(f"{build_url}/consoleText", local_console_log) + except DOWNLOAD_URL_ERRORS + JENKINS_API_ERRORS as error: + print(f"Could not download the console log: {error}") + print_results_summary(local_console_log, ci_summary_file) else: kubeconfig = args.kubeconfig kubecontext = args.kubecontext local_console_log = local_results_dir / "console_log.txt" local_archive_tar = local_results_dir / "archive.tar.gz" - remote_build_dir = f"/var/jenkins_home/jobs/{base_job_name(args)}/builds/{build_number}" + remote_build_dir = f"/var/jenkins_home/jobs/{job_name.replace('/', '/jobs/')}/builds/{build_number}" remote_console_log_path = f"{remote_build_dir}/log" remote_archive_dir = f"{remote_build_dir}/archive" @@ -1196,15 +1528,85 @@ def helm_installation_lock(lock_file: Path, timeout: int = 120): time.sleep(1) -def main_download_results(k8s_client, ip, args): +def jenkins_job_url(url, job): + path = "/".join("job/" + quote(part, safe="") for part in job.split("/")) + return f"{normalize_jenkins_url(url)}/{path}" + + +def jenkins_build_url(url, job, number): + return f"{jenkins_job_url(url, job)}/{number}" + + +def build_parameters(info): + return {parameter["name"]: parameter.get("value") + for action in info.get("actions", []) for parameter in action.get("parameters", [])} + + +def matching_running_build(server, url, job, parameters): + # Executor queries miss pipelines waiting for agents; `builds` alone omits runs older than the latest 100. + tree = "allBuilds[number,building],property[parameterDefinitions[name,defaultParameterValue[value]]]" + response = server.jenkins_open(requests.Request("GET", f"{jenkins_job_url(url, job)}/api/json", + params={"tree": tree}), add_crumb=False) + try: + info = json.loads(response) + except (TypeError, ValueError) as error: + raise jenkins.JenkinsException(f"Could not read running builds for {job}; no build started.") from error + if not isinstance(info, dict) or not isinstance(info.get("allBuilds"), list): + raise jenkins.JenkinsException(f"Missing build list for {job}; no build started.") + expected = {} + for prop in info.get("property", []): + for definition in prop.get("parameterDefinitions", []): + default = definition.get("defaultParameterValue") + if default is not None: + expected[definition["name"]] = default.get("value") + expected.update(parameters) + for build in sorted(info["allBuilds"], key=lambda build: build["number"], reverse=True): + if build.get("building") is not True: + continue + try: + current = server.get_build_info(job, build["number"]) + except jenkins.NotFoundException: + continue # The build was deleted after listing it. + if current.get("building") is True and build_parameters(current) == expected: + return build["number"] + return None + + +def confirm_attachment(job, number): + while True: + try: + answer = input(f"{job} #{number} is running with the same parameters. Attach? [Y/n]: ").strip().lower() + except (EOFError, OSError) as error: + raise SystemExit(f"No build started. Repeat this command with --attach {number} to reconnect.") from error + if answer in ("", "y", "yes"): + return True + if answer in ("n", "no"): + return False + print("Enter y to attach or n to start a new build.") + + +def main_download_results(k8s_client, ip, args, server): build_number = int(args.download_results) - download_results_and_print_summary(k8s_client, DEFAULT_POD_NAME, DEFAULT_KUBE_NS, build_number, ip, args) + download_results_and_print_summary(k8s_client, DEFAULT_POD_NAME, DEFAULT_KUBE_NS, build_number, ip, args, server=server) + + +def attach_build(k8s_client, url, server, args): + args.job = base_job_name(args) + # Read once before polling so an unknown build fails promptly. Its metadata supplies artifact names. + info = server.get_build_info(args.job, args.attach) + parameters = build_parameters(info) + args.repository = parameters.get("repository") + args.branch = parameters.get("branch") + print(f"Reconnecting to {jenkins_build_url(url, args.job, args.attach)}/") + wait_for_build_complete(server, args.job, args.attach) + download_results_and_print_summary(k8s_client, DEFAULT_POD_NAME, DEFAULT_KUBE_NS, + args.attach, url, args, server=server) def main(): load_environment_file() args = parse_arguments() - k8s_client = None if args.url else setup_environment(args.kubeconfig, args.kubecontext) + k8s_client = None if args.url or args.only_tear_down else setup_environment(args.kubeconfig, args.kubecontext) if args.only_tear_down: cleanup_and_maybe_teardown(args.kubeconfig, args.kubecontext, DEFAULT_KUBE_NS, True) @@ -1224,16 +1626,13 @@ def main(): ensure_cassandra_job_parameters_visible(server) if args.only_setup: return + if args.attach: + attach_build(k8s_client, ip, server, args) + return if args.download_results: - main_download_results(k8s_client, ip, args) + main_download_results(k8s_client, ip, args, server) return - # Background node cleaner: checks for dangling nodes and deletes them, can dramatically reduce k8s costs - # set env var NODE_CLEANER_DISABLE to disable - if not os.environ.get("NODE_CLEANER_DISABLE") and not args.url: - threading.Thread(target=node_cleaner, - args=(k8s_client, args.kubeconfig, args.kubecontext, DEFAULT_KUBE_NS), daemon=True).start() - # Trigger Jenkins build with parameters build_params = { "repository": args.repository, @@ -1245,6 +1644,14 @@ def main(): "dtest_branch": args.dtest_branch or "" } + args.job = base_job_name(args) + running = matching_running_build(server, ip, args.job, build_params) + if running is not None and confirm_attachment(args.job, running): + args.attach = running + attach_build(k8s_client, ip, server, args) + cleanup_and_maybe_teardown(args.kubeconfig, args.kubecontext, DEFAULT_KUBE_NS, args.tear_down) + return + if DEFAULT_REPO_URL == args.repository and DEFAULT_REPO_BRANCH == args.branch and is_local_git_dirty(args): print("Local uncommitted/unpushed changes.") print(f"CI only runs on what is pushed in {args.repository} @ {args.branch}") @@ -1254,15 +1661,21 @@ def main(): if "y" != input().strip().lower(): return + # Background node cleaner: checks for dangling nodes and deletes them, can dramatically reduce k8s costs + # set env var NODE_CLEANER_DISABLE to disable + if not os.environ.get("NODE_CLEANER_DISABLE") and not args.url: + threading.Thread(target=node_cleaner, + args=(k8s_client, args.kubeconfig, args.kubecontext, DEFAULT_KUBE_NS), daemon=True).start() + queue_item = trigger_jenkins_build(server, base_job_name(args), **build_params) build_number = wait_for_build_number(server, queue_item) - print(f"Jenkins UI at http://{ip}/job/{base_job_name(args)}/{build_number}/pipeline-overview/") + print(f"Jenkins UI at {jenkins_build_url(ip, base_job_name(args), build_number)}/pipeline-overview/") wait_for_build_complete(server, base_job_name(args), build_number) # Post-build processing and cleanup if not args.url: delete_remote_junit_files(k8s_client, DEFAULT_POD_NAME, DEFAULT_KUBE_NS, base_job_name(args), build_number) - download_results_and_print_summary(k8s_client, DEFAULT_POD_NAME, DEFAULT_KUBE_NS, build_number, ip, args) + download_results_and_print_summary(k8s_client, DEFAULT_POD_NAME, DEFAULT_KUBE_NS, build_number, ip, args, server=server) cleanup_and_maybe_teardown(args.kubeconfig, args.kubecontext, DEFAULT_KUBE_NS, args.tear_down) if __name__ == "__main__": diff --git a/.build/run-ci.d/README.md b/.build/run-ci.d/README.md index d87e5b137880..37be2c0f280e 100644 --- a/.build/run-ci.d/README.md +++ b/.build/run-ci.d/README.md @@ -1,44 +1,141 @@ # Help for `.build/run-ci` -``` -➤ .build/run-ci --help -usage: run-ci [-h] [-c KUBECONFIG] [-x KUBECONTEXT] [-i URL] [-u USER] [-r REPOSITORY] [-b BRANCH] [-p {packaging,skinny,pre-commit,pre-commit w/ upgrades,post-commit,custom}] [-e PROFILE_CUSTOM_REGEXP] [-j JDK] [-d DTEST_REPOSITORY] [-k DTEST_BRANCH] - [-s] [--only-setup] [-v VALUES_OVERRIDE] [--tear-down] [--only-tear-down] [--only-node-cleaner] [-o DOWNLOAD_RESULTS] +- [Tools](#tools) +- [Credentials](#credentials) +- [Attach to a build](#attach-to-a-build) +- [Build progress](#build-progress) +- [Examples](#examples) + +```text +usage: run-ci [-h] [-c KUBECONFIG] [-x KUBECONTEXT] [-i URL] [-u USER] + [--credentials-file CREDENTIALS_FILE] [--save-credentials] + [--attach BUILD] [--job JOB] [-r REPOSITORY] [-b BRANCH] + [-p {packaging,skinny,pre-commit,pre-commit w/ upgrades,post-commit,custom}] + [-e PROFILE_CUSTOM_REGEXP] [-j JDK] [-d DTEST_REPOSITORY] + [-k DTEST_BRANCH] [-s] [--only-setup] [-f VALUES_OVERRIDE] + [--tear-down] [--only-tear-down] [--only-node-cleaner] + [-o DOWNLOAD_RESULTS] Run CI pipeline for Cassandra on K8s using Jenkins. options: -h, --help show this help message and exit - -c KUBECONFIG, --kubeconfig KUBECONFIG + -c, --kubeconfig KUBECONFIG Path to a different kubeconfig. - -x KUBECONTEXT, --kubecontext KUBECONTEXT + -x, --kubecontext KUBECONTEXT Use a different Kubernetes context. - -i URL, --url URL Jenkins url. Suitable when kubectl access in not available. Can also be specified via the JENKINS_URL environment variable (and in .build/.run-ci.env) - -u USER, --user USER Jenkins user. Can also be specified via the JENKINS_USER environment variable (and in .build/.run-ci.env) - -r REPOSITORY, --repository REPOSITORY + -i, --url URL Jenkins url. Suitable when kubectl access in not + available. Can also be specified via the JENKINS_URL + environment variable (and in .build/.run-ci.env) + -u, --user USER Jenkins user. Can also be specified via the + JENKINS_USER environment variable (and in .build/.run- + ci.env) + --credentials-file CREDENTIALS_FILE + Credentials file (default: + $XDG_CONFIG_HOME/cassandra/run-ci-credentials.json or + ~/.config/cassandra/run-ci-credentials.json). + --save-credentials Save or replace credentials after authentication + succeeds. Prompts for an API token or password without + echoing it. + --attach BUILD Reconnect to a build number or URL. Infer the job from + the repository and branch, as for a new build. Wait + for completion and download results. + --job JOB Override the inferred Jenkins job name, including + folders. + -r, --repository REPOSITORY Repository URL. Defaults to current tracking remote. - -b BRANCH, --branch BRANCH - Repository branch. Defaults to current branch. - -p {packaging,skinny,pre-commit,pre-commit w/ upgrades,post-commit,custom}, --profile {packaging,skinny,pre-commit,pre-commit w/ upgrades,post-commit,custom} + -b, --branch BRANCH Repository branch. Defaults to current branch. + -p, --profile {packaging,skinny,pre-commit,pre-commit w/ upgrades,post-commit,custom} CI pipeline profile. Defaults to skinny. - -e PROFILE_CUSTOM_REGEXP, --profile-custom-regexp PROFILE_CUSTOM_REGEXP - Regexp for stages when using custom profile. See `testSteps` in Jenkinsfile for list of stages. Example: 'stress.*|jvm-dtest.' - -j JDK, --jdk JDK Specify JDK version. Defaults to all JDKs the current branch supports. - -d DTEST_REPOSITORY, --dtest-repository DTEST_REPOSITORY + -e, --profile-custom-regexp PROFILE_CUSTOM_REGEXP + Regexp for stages when using custom profile. See + `testSteps` in Jenkinsfile for list of stages. + Example: 'stress.*|jvm-dtest.' + -j, --jdk JDK Specify JDK version. Defaults to all JDKs the current + branch supports. + -d, --dtest-repository DTEST_REPOSITORY DTest repository URL. - -k DTEST_BRANCH, --dtest-branch DTEST_BRANCH + -k, --dtest-branch DTEST_BRANCH DTest repository branch. -s, --setup Set up Jenkins before the build. --only-setup Only install Jenkins into the k8s cluster. - -v VALUES_OVERRIDE, --values-override VALUES_OVERRIDE - Path to an additional helm values file, applied over .jenkins/k8s/jenkins-deployment.yaml. Required when the target cluster carries site customisations, see .jenkins/k8s/README.md + -f, --values-override VALUES_OVERRIDE + Path to an additional helm values file, applied over + .jenkins/k8s/jenkins-deployment.yaml. Required when + the target cluster carries site customisations, see + .jenkins/k8s/README.md --tear-down Tear down Jenkins after the build. --only-tear-down Only tear down Jenkins. - --only-node-cleaner Only run the node cleaner. The node cleaner scans the k8s nodes, eagerly terminating those unused. - -o DOWNLOAD_RESULTS, --download-results DOWNLOAD_RESULTS - Just download the results for the specificed build number. Naming of local artefacts assumes current tracking remote and branch, use -r and -b otherwise. + --only-node-cleaner Only run the node cleaner. The node cleaner scans the + k8s nodes, eagerly terminating those unused. + -o, --download-results DOWNLOAD_RESULTS + Just download the results for the specificed build + number. Naming of local artefacts assumes current + tracking remote and branch, use -r and -b otherwise. +``` + +## Tools + +URL connections need neither Helm nor kubectl, and do not load a kubeconfig. Helm is checked when installing or uninstalling Jenkins. Kubectl is checked when an operation invokes it, such as reading the cluster's admin password or copying results. A cluster connection through the Python Kubernetes client alone does not require either CLI. Python dependencies remain listed in `requirements.txt`. + +Without an explicit URL, `run-ci` reads the `cassius-jenkins` Service in the selected Kubernetes context. It prefers the first concrete hostname in the Service's `external-dns.kubernetes.io/hostname` annotation, or its older `external-dns.alpha.kubernetes.io/hostname` equivalent. For a named host, an HTTPS listener takes precedence over HTTP; custom ports are retained. Without a usable DNS annotation, the Service's load-balancer address is used, preferring HTTP when available because the site's TLS certificate generally does not cover the load balancer's own hostname. `--url` and `JENKINS_URL` override this discovery. Saved credentials are selected by the resulting URL. + +New builds started through a Kubernetes context also run a background node cleaner, unless `NODE_CLEANER_DISABLE=1` is set. The cleaner uses kubectl to list nodes so failed credential refreshes are reported. Any error listing, inspecting or deleting Kubernetes resources stops the cleaner and its pending deletion steps for this invocation, with one diagnostic; Jenkins monitoring continues. This includes transient errors: cleanup does not retry automatically after a failure. Restore cluster access before restarting `--only-node-cleaner`. For AWS login credentials, run `aws login` with the same profile. An already running client does not pick up code changes: use Ctrl-C, then `--attach` with the build URL to continue monitoring and download results without cluster access. Ordinary cluster-based runs still need cluster access for post-build file cleanup and downloads. + +## Credentials + +Add `--save-credentials` to a build, download or attach command to remember the supplied username and API token or password. Input is hidden, and credentials are saved only after Jenkins confirms authentication. Prefer a [Jenkins API token](https://www.jenkins.io/doc/book/system-administration/authenticating-scripted-clients/) and use HTTPS for remote connections. + +```shell +.build/run-ci --url https://ci.example.org --user alice --save-credentials +``` + +Later commands for that Jenkins URL reuse the saved credentials without `--user` or a prompt. Use `--save-credentials` again to replace the token. An explicit different `--user` prompts for that user's credentials. + +The default file is `~/.config/cassandra/run-ci-credentials.json`, or `$XDG_CONFIG_HOME/cassandra/run-ci-credentials.json` when that variable names an absolute directory. Use `--credentials-file PATH` to select another file. The file has mode `0600`; a newly created credentials directory has mode `0700`. Writes are atomic, and symbolic links, hard links, foreign ownership and group/world-readable files are rejected. This is a permissions-protected JSON file, not encrypted storage. Keep it outside the checkout. + +Credentials are keyed by the complete Jenkins base URL, including scheme, port and context path. A token saved for `https://ci.example.org/jenkins` is not reused for HTTP or a different Jenkins path. Tokens and passwords are never accepted as command-line arguments or embedded in URLs. + +## Attach to a build + +Reconnect using the build number: + +```shell +.build/run-ci --attach 42 ``` +The job is inferred from the repository and branch's Cassandra version, as for a new build. Repeat any `-r` and `-b` overrides from the initial invocation. `--job` is an optional override for custom job names or folders. + +Use the same connection options as before, for example `--url https://ci.example.org`, or set `JENKINS_URL`. Without a URL, the selected Kubernetes context supplies the Jenkins address. Saved credentials are reused; add `--user USER` when credentials have not been saved. + +A build URL also supplies the job and server, without a remote branch lookup: + +```shell +.build/run-ci --attach https://ci.example.org/job/cassandra/42/ +``` + +Before starting a new build, `run-ci` checks the inferred job for running builds with identical parameters. Matching includes job defaults for parameters the command does not supply, such as architecture. The newest match prompts `Attach? [Y/n]`: Enter or `y` attaches; `n` starts a new build. Completed builds are not offered. If the prompt cannot read an answer, the command stops and prints `--attach BUILD_NUMBER` instructions. This check does not reserve the job against submissions by other clients. + +Attaching waits for the existing build to finish, then downloads its summary, results archive and console log through the authenticated Jenkins connection. Completed builds download immediately. Artifact names use the build's recorded repository and branch; a clean working tree is not required. Results are stored under `build/ci////`, so different jobs cannot overwrite one another's results. + +Attaching starts no build and runs no node cleaner or remote file deletion. Accepting the matching-build prompt preserves `--tear-down`: Jenkins is uninstalled after monitoring and result downloads return. Explicit `--attach` runs no teardown; setup, teardown and `--download-results` cannot be combined with it. Ctrl-C disconnects the client and leaves the Jenkins build running without teardown. + +## Build progress + +New builds and `--attach` show split progress beside the elapsed waiting time: + +```text +Splits: 128/320 finished (3 failed) | 24 running | 168 queued | 12:34 / +``` + +The counter reads the named parallel branches under `Tests` through the [Pipeline Graph View API](https://github.com/jenkinsci/pipeline-graph-view-plugin/blob/main/openapi.yaml), which the Kubernetes deployment configures. Each selected task/JDK combination counts separately, including tasks with one split; retries remain part of the same branch. A split finishes after its whole branch, including result processing and cleanup, ends. Failed, unstable, aborted and skipped splits count as finished and appear separately in parentheses. + +The total follows the branches Jenkins has published for this build and can grow while the parallel test branches are being created. Before that, the display says `Splits: waiting for test branches`. The counter refreshes every 15 seconds and once at completion; overall build status is still checked approximately every three seconds. If the plugin or its data is unavailable, the display says `Splits: unavailable` and continues waiting for the build. Attaching reconstructs the counter from Jenkins, while the elapsed timer starts at attachment. Split completion is not an estimate of remaining time. + +The final summary labels the Jenkins result and test totals separately, then prints the saved `console_log.txt.gz` path. Ant's `BUILD FAILED` messages belong to individual commands and can appear in an `UNSTABLE` Jenkins build. They remain in the full console log; the summary omits excerpts because parallel task output is interleaved. Missing reports are shown as unavailable totals rather than evidence that no tests ran. + +Results use the Jenkins hostname as their server directory, for example `build/ci/astro-cass.ci/cassandra-6.0/58/`. Non-default ports and encoded context paths, when present, distinguish servers on the same host. The URL scheme is omitted from this directory name; credential lookup still uses the complete URL. + ## Examples Run the current directory's fork and branch through the default "skinny" pipeline, connecting via your default kubeconfig ``` @@ -76,4 +173,4 @@ Uninstall Jenkins from your current kubeconfig. ``` .build/run-ci --only-tear-down ``` -The jenkins-home volume is kept; delete it separately with `kubectl delete pvc cassius-jenkins` \ No newline at end of file +The jenkins-home volume is kept; delete it separately with `kubectl delete pvc cassius-jenkins` diff --git a/.build/run-ci.d/run-ci-test.py b/.build/run-ci.d/run-ci-test.py index 203a3ea5a79a..485f00045627 100644 --- a/.build/run-ci.d/run-ci-test.py +++ b/.build/run-ci.d/run-ci-test.py @@ -28,15 +28,20 @@ import argparse import contextlib import io +import json +import os +import stat from pathlib import Path import subprocess import tempfile import unittest +import warnings from unittest.mock import patch, MagicMock from urllib.error import HTTPError import yaml import jenkins +import run_ci # Import the functions from the script from run_ci import ( @@ -65,10 +70,149 @@ def quietly(action): return action() +class TestNodeCleanerAccess(unittest.TestCase): + EXPIRED_LOGIN = ('aws: [ERROR]: CreateOAuth2Token: The provided authorization grant is ' + 'invalid, expired, revoked, or malformed\nerror: INVALID_REQUEST') + + @contextlib.contextmanager + def cleaner(self, commands): + api = MagicMock() + node = run_ci.client.V1Node( + metadata=run_ci.client.V1ObjectMeta(name='agent-node'), + spec=run_ci.client.V1NodeSpec(provider_id='aws:///us-west-2a/i-123')) + api.list_node.return_value.items = [node] + api.read_node.return_value = node + api.list_pod_for_all_namespaces.return_value.items = [ + run_ci.client.V1Pod(metadata=run_ci.client.V1ObjectMeta(name='pod', namespace='default'))] + output = io.StringIO() + sleeps = 0 + + def bounded_sleep(_seconds): + nonlocal sleeps + sleeps += 1 + if sleeps >= 9: + run_ci.IS_RUNNING = False # Bound retries in the unfixed implementation. + + def start_worker(*, target, args, **_kwargs): + return MagicMock(start=lambda: target(*args)) + + with patch('run_ci.IS_RUNNING', True), patch('run_ci.time.sleep', side_effect=bounded_sleep), \ + patch('run_ci.run_kubectl_command', side_effect=commands) as kubectl, \ + patch('run_ci.threading.Thread', side_effect=start_worker), \ + patch.dict('sys.modules', {'boto3': MagicMock()}) as modules, \ + contextlib.redirect_stderr(output): + yield api, kubectl, modules['boto3'], output + + def command_failure(self, message=EXPIRED_LOGIN): + return subprocess.CalledProcessError(1, ['kubectl', 'get', 'nodes'], stderr=message) + + def test_failed_token_refresh_stops_polling_and_leaves_build_monitor_running(self): + with self.cleaner(self.command_failure()) as (api, kubectl, cloud, output): + run_ci.node_cleaner(api, 'config', 'eks-context', 'default') + self.assertTrue(run_ci.IS_RUNNING) + kubectl.assert_called_once_with('config', 'eks-context', 'default', ['get', 'nodes', '-o', 'json']) + api.list_node.assert_not_called() + api.patch_node.assert_not_called() + cloud.session.Session.assert_not_called() + self.assertEqual(output.getvalue().count('Node cleaner stopped'), 1) + self.assertIn('CreateOAuth2Token', output.getvalue()) + + def test_failed_inspection_stops_repeated_worker_creation(self): + def commands(_config, _context, _namespace, command): + if command[0] == 'get': + return json.dumps({'items': [{'metadata': {'name': name}} + for name in ['agent-node', 'another-node']]}) + raise self.command_failure() + + with self.cleaner(commands) as (api, kubectl, cloud, output): + run_ci.node_cleaner(api, None, None, 'default') + self.assertTrue(run_ci.IS_RUNNING) + self.assertEqual(kubectl.call_count, 2) + api.read_node.assert_not_called() + api.patch_node.assert_not_called() + cloud.session.Session.assert_not_called() + self.assertEqual(output.getvalue().count('Node cleaner stopped'), 1) + + def test_unreadable_node_list_stops_without_deletion(self): + failures = [self.command_failure('Forbidden'), self.command_failure(None), + OSError('kubectl unavailable'), 'not JSON', '{}', '{"items": null}'] + for failure in failures: + with self.subTest(failure=failure), self.cleaner([failure]) as (api, kubectl, cloud, output): + run_ci.node_cleaner(api, None, None, 'default') + self.assertTrue(run_ci.IS_RUNNING) + kubectl.assert_called_once() + api.patch_node.assert_not_called() + cloud.session.Session.assert_not_called() + self.assertEqual(output.getvalue().count('Node cleaner stopped'), 1) + + def test_api_access_failure_aborts_remaining_deletion_steps(self): + steps = ['read_node', 'patch_node', 'list_pod_for_all_namespaces', + 'delete_namespaced_pod', 'delete_node'] + + def commands(_config, _context, _namespace, command): + if command[0] == 'get': + return json.dumps({'items': [{'metadata': {'name': 'agent-node'}}]}) + return 'No Jenkins pods' + + for index, step in enumerate(steps): + with self.subTest(step=step), self.cleaner(commands) as (api, _, cloud, output): + getattr(api, step).side_effect = run_ci.client.exceptions.ApiException(status=401) + run_ci.node_cleaner(api, None, None, 'default') + for remaining in steps[index + 1:]: + getattr(api, remaining).assert_not_called() + cloud.session.Session.assert_not_called() + self.assertTrue(run_ci.IS_RUNNING) + self.assertEqual(output.getvalue().count('Node cleaner stopped'), 1) + + def test_provider_lookup_failure_stops_before_deletion(self): + for lookup in ['list_node', 'current-context']: + with self.subTest(lookup=lookup), self.cleaner( + [json.dumps({'items': [{'metadata': {'name': 'agent-node'}}]})] + + ['No Jenkins pods'] * 6 + [self.command_failure()]) as (api, _, cloud, output): + api.read_node.return_value.spec.provider_id = None + if lookup == 'list_node': + api.list_node.side_effect = run_ci.client.exceptions.ApiException(status=403) + run_ci.node_cleaner(api, None, None, 'default') + api.patch_node.assert_not_called() + cloud.session.Session.assert_not_called() + self.assertTrue(run_ci.IS_RUNNING) + self.assertEqual(output.getvalue().count('Node cleaner stopped'), 1) + + def test_stopping_during_idle_wait_prevents_deletion(self): + sleeps = 0 + + def stop_after_idle_wait(_seconds): + nonlocal sleeps + sleeps += 1 + if sleeps == 6: + run_ci.IS_RUNNING = False + + with self.cleaner([json.dumps({'items': [{'metadata': {'name': 'agent-node'}}]})] + + ['No Jenkins pods'] * 6) as (api, _, cloud, _), \ + patch('run_ci.time.sleep', side_effect=stop_after_idle_wait): + run_ci.node_cleaner(api, None, None, 'default') + api.read_node.assert_not_called() + api.patch_node.assert_not_called() + cloud.session.Session.assert_not_called() + + def test_successful_inspection_still_removes_idle_node(self): + with self.cleaner([json.dumps({'items': [{'metadata': {'name': 'agent-node'}}]})] + + ['No Jenkins pods'] * 6 + [self.command_failure()]) as (api, _, cloud, _): + run_ci.node_cleaner(api, None, None, 'default') + api.patch_node.assert_called_once() + api.delete_namespaced_pod.assert_called_once_with(name='pod', namespace='default') + api.delete_node.assert_called_once_with('agent-node') + cloud.session.Session.return_value.client.return_value.terminate_instance_in_auto_scaling_group.assert_called_once_with( + InstanceId='i-123', ShouldDecrementDesiredCapacity=True) + + class TestCIPipeline(unittest.TestCase): def setUp(self): print("\ntesting ", self._testMethodName) + command_lookup = patch('run_ci.shutil.which', return_value='/mock/tool') + command_lookup.start() + self.addCleanup(command_lookup.stop) @patch('run_ci.os.environ.get') @patch('run_ci.print') @@ -84,14 +228,39 @@ def test_debug(self, mock_print, mock_get): @patch('run_ci.subprocess.run') def test_install_jenkins(self, mock_run): # empty stdout, i.e. nothing deployed yet, so the pre-flight check has nothing to warn about - mock_run.return_value = MagicMock(returncode=0, stdout="") + mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="") install_jenkins("test-namespace", Path("/fake/cassandra/dir"), "default") mock_run.assert_any_call(["helm", "repo", "add", "jenkins", "https://charts.jenkins.io"], check=True) - mock_run.assert_any_call(["helm", "repo", "update"], check=True) + mock_run.assert_any_call(["helm", "repo", "update", "jenkins"], check=True) + + @patch('run_ci.subprocess.run') + def test_install_jenkins_ignores_unrelated_repository_failure(self, mock_run): + def run_command(cmd, **kwargs): + if cmd[:3] == ["helm", "repo", "update"] and (len(cmd) == 3 or "autoscaler" in cmd[3:]): + raise subprocess.CalledProcessError(1, cmd, stderr="lookup kubernetes.github.io: no such host") + return subprocess.CompletedProcess(cmd, 0, "", "") + + mock_run.side_effect = run_command + install_jenkins(None, "test-context", "default") + self.assertTrue(any(call.args[0][0] == "helm" and "upgrade" in call.args[0] + for call in mock_run.call_args_list)) + + @patch('run_ci.subprocess.run') + def test_install_jenkins_stops_when_jenkins_repository_fails(self, mock_run): + def run_command(cmd, **kwargs): + if cmd[:3] == ["helm", "repo", "update"] and (len(cmd) == 3 or "jenkins" in cmd[3:]): + raise subprocess.CalledProcessError(1, cmd, stderr="Jenkins repository unavailable") + return subprocess.CompletedProcess(cmd, 0, "", "") + + mock_run.side_effect = run_command + with self.assertRaises(subprocess.CalledProcessError): + install_jenkins(None, "test-context", "default") + self.assertFalse(any(call.args[0][0] == "helm" and "upgrade" in call.args[0] + for call in mock_run.call_args_list)) @patch('run_ci.subprocess.run') def test_install_jenkins_values_override(self, mock_run): - mock_run.return_value = MagicMock(returncode=0, stdout="") + mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="") with tempfile.NamedTemporaryFile("w", suffix=".yaml") as override: override.write(self.LIVE_STORAGE_CLASS) override.flush() @@ -100,6 +269,28 @@ def test_install_jenkins_values_override(self, mock_run): # the site's overrides must come after, and never replace, the repo's deployment yaml self.assertEqual(["-f", DEPLOY_YAML, "-f", override.name], upgrade_cmd[5:9]) + @patch('run_ci.subprocess.run') + def test_install_jenkins_reports_helm_failure_before_pod_setup(self, mock_run): + def run_command(cmd, **kwargs): + if "upgrade" in cmd: + result = subprocess.CompletedProcess(cmd, 1, "Release cassius failed.\n", + "Error: context deadline exceeded\n") + if kwargs.get("check"): + result.check_returncode() + return result + return subprocess.CompletedProcess(cmd, 0, "", "") + + mock_run.side_effect = run_command + stdout, stderr = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): + with self.assertRaises(SystemExit) as raised: + install_jenkins(None, "test-context", "default") + self.assertEqual(raised.exception.code, 1) + self.assertIn("Release cassius failed.", stdout.getvalue()) + self.assertIn("Error: context deadline exceeded", stderr.getvalue()) + self.assertEqual([], [call for call in mock_run.call_args_list + if call.args[0][0] == "kubectl" and "exec" in call.args[0]]) + @patch('run_ci.sys.stdin.isatty') @patch('run_ci.print') @patch('run_ci.subprocess.run') @@ -162,9 +353,14 @@ def test_get_jenkins(self, mock_jenkins, mock_run): mock_run.return_value = MagicMock(stdout="fake-password") mock_jenkins_instance = MagicMock() mock_jenkins.return_value = mock_jenkins_instance - # hack – use False values instead of None - args = argparse.Namespace(kubeconfig="/fake/kubeconfig", kubecontext="test-context", user=False, url=False) - _, server = get_jenkins(mock_k8s_client, args, "default") + service = mock_k8s_client.read_namespaced_service.return_value + service.metadata.annotations = {} + service.status.load_balancer.ingress = [run_ci.client.V1LoadBalancerIngress(ip='192.0.2.1')] + service.spec.ports = [run_ci.client.V1ServicePort(name='http', port=80)] + with tempfile.TemporaryDirectory() as directory: + args = argparse.Namespace(kubeconfig="/fake/kubeconfig", kubecontext="test-context", user=None, url=None, + credentials_file=Path(directory) / 'credentials.json', save_credentials=False) + _, server = get_jenkins(mock_k8s_client, args, "default") self.assertEqual(server, mock_jenkins_instance) @patch('run_ci.jenkins.Jenkins.build_job') @@ -369,5 +565,913 @@ def test_check_agent_capacity_blocks_contradictory_config(self): # unconfirmable is not the same as contradicted: with that pool at zero there is nothing to check against self.assertEqual(0, self.capacity_check(typo)) +class TestJenkinsServiceDiscovery(unittest.TestCase): + DNS = 'external-dns.alpha.kubernetes.io/hostname' + CERT = 'service.beta.kubernetes.io/aws-load-balancer-ssl-cert' + TLS_PORTS = 'service.beta.kubernetes.io/aws-load-balancer-ssl-ports' + + def service(self, annotations=None, ports=(('http', 80), ('https', 443)), address='example.elb.amazonaws.com'): + return run_ci.client.V1Service( + metadata=run_ci.client.V1ObjectMeta(annotations=annotations), + spec=run_ci.client.V1ServiceSpec(ports=[run_ci.client.V1ServicePort(name=name, port=port, protocol='TCP') + for name, port in ports]), + status=run_ci.client.V1ServiceStatus(load_balancer=run_ci.client.V1LoadBalancerStatus( + ingress=[run_ci.client.V1LoadBalancerIngress(hostname=address)] if address else []))) + + def connect(self, service, explicit_url=None): + api = MagicMock() + api.read_namespaced_service.return_value = service + args = argparse.Namespace(url=explicit_url, kubeconfig=None, kubecontext='selected-context', user=None, + credentials_file=Path('/unused/test-credentials.json'), save_credentials=False) + output = io.StringIO() + with patch('run_ci.load_credentials', return_value={'user': 'alice', 'token': 'test-token'}) as credentials, \ + patch('run_ci.jenkins.Jenkins') as connection, \ + patch('run_ci.run_kubectl_command') as kubectl, contextlib.redirect_stdout(output): + url, _server = get_jenkins(api, args, 'default') + credentials.assert_called_once_with(args.credentials_file, url) + connection.assert_called_once_with(url, username='alice', password='test-token') + kubectl.assert_not_called() + if explicit_url: + api.read_namespaced_service.assert_not_called() + else: + api.read_namespaced_service.assert_called_once_with('cassius-jenkins', 'default') + self.last_output = output.getvalue() + return url + + def test_prefers_service_hostname_and_https(self): + service = self.service({self.DNS: 'astro-cass.ci', self.CERT: 'test-cert', self.TLS_PORTS: 'https'}) + self.assertEqual(self.connect(service), 'https://astro-cass.ci') + self.assertIn('Jenkins: https://astro-cass.ci', self.last_output) + + def test_dns_hostname_works_without_tls_or_load_balancer_status(self): + service = self.service({self.DNS: 'ci.example'}, ports=(('http', 80),), address=None) + self.assertEqual(self.connect(service), 'http://ci.example') + + def test_missing_or_unusable_dns_annotation_preserves_load_balancer_fallback(self): + for annotations in (None, {}, {self.DNS: ''}, {self.DNS: '*.example'}, + {self.DNS: 'https://ci.example'}, {self.DNS: 'user:token@ci.example'}): + with self.subTest(annotations=annotations): + service = self.service(annotations, ports=(('https', 443), ('http', 80))) + self.assertEqual(self.connect(service), 'http://example.elb.amazonaws.com') + + def test_selects_first_concrete_dns_name_and_supports_both_annotation_prefixes(self): + for key in (self.DNS, 'external-dns.kubernetes.io/hostname'): + with self.subTest(key=key): + service = self.service({key: ' *.example, Astro-Cass.ci., alias.example '}) + self.assertEqual(self.connect(service), 'https://astro-cass.ci') + + def test_preserves_custom_http_and_annotated_tls_ports(self): + service = self.service({self.DNS: 'ci.example'}, ports=(('http', 8080),)) + self.assertEqual(self.connect(service), 'http://ci.example:8080') + for tls_ports in ('secure', '8443', None): + with self.subTest(tls_ports=tls_ports): + annotations = {self.DNS: 'ci.example', self.CERT: 'test-cert'} + if tls_ports is not None: + annotations[self.TLS_PORTS] = tls_ports + service = self.service(annotations, ports=(('secure', 8443),)) + self.assertEqual(self.connect(service), 'https://ci.example:8443') + + def test_explicit_url_takes_precedence(self): + self.assertEqual(self.connect(self.service({self.DNS: 'astro-cass.ci'}), 'https://override.example/jenkins'), + 'https://override.example/jenkins') + + def test_ip_fallback_and_missing_address(self): + for address, expected in [('192.0.2.1', 'http://192.0.2.1:8080'), + ('2001:db8::1', 'http://[2001:db8::1]:8080')]: + with self.subTest(address=address): + service = self.service(ports=(('http', 8080),)) + service.status.load_balancer.ingress = [run_ci.client.V1LoadBalancerIngress(ip=address)] + self.assertEqual(self.connect(service), expected) + with self.assertRaisesRegex(ValueError, 'address'): + self.connect(self.service(address=None)) + + +class TestConnectionModes(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.credentials = Path(self.directory.name) / 'auth' / 'credentials.json' + + def args(self, **values): + args = run_ci.argument_parser().parse_args([]) + args.credentials_file = str(self.credentials) + args.save_credentials = False + args.attach = None + args.job = None + vars(args).update(values) + return args + + def parse(self, *arguments): + with patch.dict(os.environ, {}, clear=True), patch('run_ci.sys.argv', ['run-ci', *arguments]), \ + patch('run_ci.DEFAULT_REPO_URL', None), patch('run_ci.require_tracking_remote') as tracking: + result = run_ci.parse_arguments() + tracking.assert_not_called() + return result + + @patch('run_ci.client.CoreV1Api') + @patch('run_ci.config.load_kube_config') + @patch('run_ci.shutil.which', return_value=None) + def test_connecting_to_cluster_does_not_require_cli_tools(self, which, load, core): + self.assertIs(run_ci.setup_environment(None, 'cluster'), core.return_value) + which.assert_not_called() + load.assert_called_once_with(config_file=None, context='cluster') + + @patch('run_ci.shutil.which', return_value=None) + @patch('run_ci.subprocess.run') + def test_cli_tools_checked_only_when_used(self, run, _which): + with self.assertRaisesRegex(EnvironmentError, 'kubectl'): + run_ci.run_kubectl_command(None, None, 'default', ['get', 'pods']) + with self.assertRaisesRegex(EnvironmentError, 'helm'): + run_ci.run_helm_command(None, None, 'default', ['list']) + run.assert_not_called() + + def test_credentials_are_private_and_scoped_to_url(self): + run_ci.save_credentials(self.credentials, 'https://ci.example/jenkins', 'alice', 'test-token') + self.assertEqual(stat.S_IMODE(self.credentials.stat().st_mode), 0o600) + self.assertEqual(stat.S_IMODE(self.credentials.parent.stat().st_mode), 0o700) + self.assertEqual(run_ci.load_credentials(self.credentials, 'https://ci.example/jenkins'), + {'user': 'alice', 'token': 'test-token'}) + for other in ('http://ci.example/jenkins', 'https://other.example/jenkins', 'https://ci.example/other'): + self.assertIsNone(run_ci.load_credentials(self.credentials, other)) + run_ci.save_credentials(self.credentials, 'https://other.example', 'bob', 'other-token') + self.assertEqual(run_ci.load_credentials(self.credentials, 'https://ci.example/jenkins')['token'], 'test-token') + + def test_credentials_reject_public_file_and_symlink(self): + self.credentials.parent.mkdir(mode=0o700) + self.credentials.write_text('{}') + self.credentials.chmod(0o644) + with self.assertRaisesRegex(ValueError, '0600'): + run_ci.load_credentials(self.credentials, 'https://ci.example') + self.credentials.chmod(0o600) + link = self.credentials.parent / 'link.json' + link.symlink_to(self.credentials) + with self.assertRaises((ValueError, OSError)): + run_ci.save_credentials(link, 'https://ci.example', 'alice', 'test-token') + self.assertEqual(self.credentials.read_text(), '{}') + + def test_credentials_reject_shared_parent(self): + self.credentials.parent.mkdir(mode=0o777) + self.credentials.parent.chmod(0o777) + with self.assertRaisesRegex(ValueError, 'directory'): + run_ci.save_credentials(self.credentials, 'https://ci.example', 'alice', 'test-token') + + @patch('run_ci.getpass.getpass', return_value='test-token') + @patch('run_ci.jenkins.Jenkins') + def test_saved_credentials_remove_future_prompts_and_keep_https(self, connection, prompt): + connection.return_value.get_whoami.return_value = {'authenticated': True, 'name': 'alice'} + args = self.args(url='https://ci.example/jenkins/', user='alice', save_credentials=True) + run_ci.get_jenkins(None, args, 'default') + connection.assert_called_with('https://ci.example/jenkins', username='alice', password='test-token') + args.user, args.save_credentials = None, False + run_ci.get_jenkins(None, args, 'default') + self.assertEqual(prompt.call_count, 1) + connection.assert_called_with('https://ci.example/jenkins', username='alice', password='test-token') + args.user = 'bob' + run_ci.get_jenkins(None, args, 'default') + self.assertEqual(prompt.call_count, 2) + + @patch('run_ci.getpass.getpass', return_value='bad-token') + @patch('run_ci.jenkins.Jenkins') + def test_failed_authentication_never_saves_credentials(self, connection, _prompt): + connection.return_value.get_whoami.return_value = {'authenticated': False, 'name': 'anonymous'} + with self.assertRaises(jenkins.JenkinsException): + run_ci.get_jenkins(None, self.args(url='https://ci.example', user='alice', save_credentials=True), 'default') + self.assertFalse(self.credentials.exists()) + + def test_prompt_refuses_an_echoing_terminal(self): + def echoing_prompt(_message): + warnings.warn('input could be echoed', run_ci.getpass.GetPassWarning) + return 'test-token' + with patch('run_ci.getpass.getpass', side_effect=echoing_prompt), \ + patch('run_ci.jenkins.Jenkins') as connection, self.assertRaisesRegex(ValueError, 'echo'): + run_ci.get_jenkins(None, self.args(url='https://ci.example', user='alice'), 'default') + connection.assert_not_called() + + def test_rejected_login_preserves_previous_credentials(self): + run_ci.save_credentials(self.credentials, 'https://ci.example', 'alice', 'valid-token') + original = self.credentials.read_bytes() + with patch('run_ci.jenkins.Jenkins') as connection, patch('run_ci.getpass.getpass', return_value='bad-token'): + connection.return_value.get_whoami.side_effect = jenkins.JenkinsException('denied') + with self.assertRaises(jenkins.JenkinsException): + run_ci.get_jenkins(None, self.args(url='https://ci.example', user='alice', save_credentials=True), 'default') + self.assertEqual(self.credentials.read_bytes(), original) + + def test_credentials_reject_hard_links_and_malformed_data(self): + run_ci.save_credentials(self.credentials, 'https://ci.example', 'alice', 'test-token') + link = self.credentials.parent / 'link.json' + os.link(self.credentials, link) + with self.assertRaises(ValueError): + run_ci.load_credentials(self.credentials, 'https://ci.example') + link.unlink() + for contents in ('not json', json.dumps({'https://ci.example': {'user': 'alice'}})): + self.credentials.write_text(contents) + with self.assertRaisesRegex(ValueError, 'Invalid credentials file'): + run_ci.load_credentials(self.credentials, 'https://ci.example') + + def test_failed_credential_write_preserves_original(self): + run_ci.save_credentials(self.credentials, 'https://ci.example', 'alice', 'valid-token') + original = self.credentials.read_bytes() + with patch('run_ci.os.replace', side_effect=OSError('write failed')), self.assertRaises(OSError): + run_ci.save_credentials(self.credentials, 'https://ci.example', 'alice', 'new-token') + self.assertEqual(self.credentials.read_bytes(), original) + self.assertEqual(list(self.credentials.parent.iterdir()), [self.credentials]) + + def test_url_preserves_https_context_and_rejects_embedded_secrets(self): + self.assertEqual(run_ci.normalize_jenkins_url('HTTPS://CI.EXAMPLE:443/jenkins/'), 'https://ci.example/jenkins') + self.assertEqual(run_ci.normalize_jenkins_url('ci.example:8080/jenkins'), 'http://ci.example:8080/jenkins') + for value in ('https://alice:test-token@ci.example', 'ftp://ci.example', 'https://ci.example?token=secret'): + with self.assertRaises(ValueError) as error: + run_ci.normalize_jenkins_url(value) + self.assertNotIn('test-token', str(error.exception)) + self.assertNotIn('secret', str(error.exception)) + + def test_attach_url_needs_no_git_tracking_or_explicit_user(self): + args = self.parse('--attach', 'https://ci.example/jenkins/job/team/job/cassandra-5.0/42/') + self.assertEqual((args.url, args.job, args.attach), ('https://ci.example/jenkins', 'team/cassandra-5.0', 42)) + self.assertEqual(run_ci.base_job_name(args), 'team/cassandra-5.0') + + def test_attach_number_with_explicit_job(self): + args = self.parse('--attach', '42', '--job', 'cassandra-5.0', '--url', 'ci.example', '--user', 'alice') + self.assertEqual((args.url, args.job, args.attach), ('http://ci.example', 'cassandra-5.0', 42)) + + def test_attach_number_requires_repository_for_job_inference(self): + with patch.dict(os.environ, {}, clear=True), \ + patch('run_ci.sys.argv', ['run-ci', '--attach', '42']), \ + patch('run_ci.DEFAULT_REPO_URL', None), \ + patch('run_ci.require_tracking_remote', side_effect=SystemExit('no tracking remote')) as tracking: + with self.assertRaises(SystemExit): + run_ci.parse_arguments() + tracking.assert_called_once() + + def test_attach_url_overrides_environment_server(self): + with patch.dict(os.environ, {'JENKINS_URL': 'https://other.example'}, clear=True), \ + patch('run_ci.sys.argv', ['run-ci', '--attach', 'https://ci.example/job/cassandra/42/']): + args = run_ci.parse_arguments() + self.assertEqual(args.url, 'https://ci.example') + + def test_credential_location_never_defaults_to_working_directory(self): + for directory in ('', 'relative/directory'): + with patch.dict(os.environ, {'XDG_CONFIG_HOME': directory}): + self.assertEqual(run_ci.default_credentials_file(), + Path.home() / '.config/cassandra/run-ci-credentials.json') + + def test_attach_rejects_ambiguous_or_mutating_requests(self): + for arguments in [('--attach', '0', '--job', 'cassandra'), + ('--attach', 'https://ci.example/job/cassandra/42/', '--setup'), + ('--attach', 'https://ci.example/job/cassandra/42/', '--url', 'https://other.example'), + ('--attach', 'https://ci.example/job/cassandra/42/', '--download-results', '43')]: + with self.subTest(arguments=arguments), self.assertRaises(SystemExit), contextlib.redirect_stderr(io.StringIO()): + self.parse(*arguments) + + @patch('run_ci.cleanup_and_maybe_teardown') + @patch('run_ci.setup_environment') + @patch('run_ci.load_environment_file') + def test_teardown_does_not_initialize_kubernetes(self, _env, setup, cleanup): + with patch('run_ci.parse_arguments', return_value=self.args(only_tear_down=True)): + run_ci.main() + setup.assert_not_called() + cleanup.assert_called_once_with(None, None, 'default', True) + + def test_attach_waits_for_existing_build_without_launch_or_cleanup(self): + args = self.args(url='https://ci.example', job='cassandra-5.0', attach=42) + server = MagicMock() + server.get_build_info.side_effect = [{'building': False, 'result': 'SUCCESS', 'actions': [{'parameters': [ + {'name': 'repository', 'value': 'https://github.com/example/cassandra'}, + {'name': 'branch', 'value': 'feature/123'}]}]}, {'building': False, 'result': 'SUCCESS'}] + with patch('run_ci.load_environment_file'), patch('run_ci.parse_arguments', return_value=args), \ + patch('run_ci.get_jenkins', return_value=('https://ci.example', server)), \ + patch('run_ci.setup_environment') as setup, patch('run_ci.is_local_git_dirty') as dirty, \ + patch('run_ci.wait_for_build_complete') as wait, \ + patch('run_ci.download_results_and_print_summary') as download, \ + patch('run_ci.delete_remote_junit_files') as delete, patch('run_ci.node_cleaner') as cleaner: + quietly(run_ci.main) + setup.assert_not_called() + dirty.assert_not_called() + cleaner.assert_not_called() + delete.assert_not_called() + server.build_job.assert_not_called() + server.stop_build.assert_not_called() + wait.assert_called_once_with(server, 'cassandra-5.0', 42) + download.assert_called_once() + self.assertEqual((args.repository, args.branch), ('https://github.com/example/cassandra', 'feature/123')) + + def test_attach_unknown_build_fails_without_polling_or_launch(self): + args = self.args(url='https://ci.example', job='cassandra', attach=999) + server = MagicMock() + server.get_build_info.side_effect = jenkins.NotFoundException('missing') + with patch('run_ci.load_environment_file'), patch('run_ci.parse_arguments', return_value=args), \ + patch('run_ci.get_jenkins', return_value=('https://ci.example', server)), \ + patch('run_ci.wait_for_build_complete') as wait, self.assertRaises(jenkins.NotFoundException): + run_ci.main() + wait.assert_not_called() + server.build_job.assert_not_called() + + def test_attach_wait_handles_disconnects_and_completed_builds(self): + args = self.args(url='https://ci.example', job='cassandra', attach=42) + for states in ([{'building': False, 'result': 'SUCCESS'}], + [jenkins.JenkinsException('disconnected'), {'building': True, 'result': None}, + {'building': False, 'result': 'SUCCESS'}]): + server = MagicMock() + server.get_build_info.side_effect = [{'actions': []}] + states + with patch('run_ci.download_results_and_print_summary') as download: + quietly(lambda: run_ci.attach_build(None, args.url, server, args)) + self.assertEqual(server.get_build_info.call_count, len(states) + 1) + server.build_job.assert_not_called() + server.stop_build.assert_not_called() + download.assert_called_once() + + def test_interrupting_attach_leaves_build_running(self): + args = self.args(url='https://ci.example', job='cassandra', attach=42) + server = MagicMock() + server.get_build_info.return_value = {'actions': []} + with patch('run_ci.wait_for_build_complete', side_effect=KeyboardInterrupt), \ + patch('run_ci.download_results_and_print_summary') as download, self.assertRaises(KeyboardInterrupt): + quietly(lambda: run_ci.attach_build(None, args.url, server, args)) + download.assert_not_called() + server.stop_build.assert_not_called() + server.build_job.assert_not_called() + + def test_download_uses_authenticated_client_and_exact_job(self): + args = self.args(url='https://ci.example/jenkins', job='team/cassandra-5.0', + repository='https://github.com/example/cassandra', branch='feature/123') + server = MagicMock() + response = server.jenkins_request.return_value.__enter__.return_value + response.iter_content.return_value = [b'test artifact'] + with patch('run_ci.LOCAL_RESULTS_BASEDIR', Path(self.directory.name) / 'results'): + quietly(lambda: run_ci.download_results_and_print_summary(None, 'unused', 'default', 42, + args.url, args, server=server)) + urls = [call.args[0].url for call in server.jenkins_request.call_args_list] + self.assertIn('https://ci.example/jenkins/job/team/job/cassandra-5.0/42/artifact/results_details.tar.xz', urls) + self.assertIn('https://ci.example/jenkins/job/team/job/cassandra-5.0/42/consoleText', urls) + artifacts = list((Path(self.directory.name) / 'results').rglob('*.tar.xz')) + self.assertEqual(len(artifacts), 1) + self.assertEqual(artifacts[0].read_bytes(), b'test artifact') + self.assertEqual(artifacts[0].name, 'results_details_example_feature-123_42.tar.xz') + self.assertEqual(artifacts[0].parent.parent.name, 'team%2Fcassandra-5.0') + + def test_download_directory_uses_host_without_url_scheme(self): + for url, directory in [('https://astro-cass.ci', 'astro-cass.ci'), + ('http://astro-cass.ci', 'astro-cass.ci'), + ('https://astro-cass.ci:443/', 'astro-cass.ci'), + ('http://ci.example:8080', 'ci.example:8080'), + ('http://[::1]:8080', '[::1]:8080'), + ('http://..', '%2E%2E'), + ('https://ci.example/jenkins', 'ci.example%2Fjenkins')]: + with self.subTest(url=url): + args = self.args(url=url, job='cassandra-6.0', repository=None, branch=None) + server = MagicMock() + response = server.jenkins_request.return_value.__enter__.return_value + response.iter_content.return_value = [b'Finished: SUCCESS\n'] + with patch('run_ci.LOCAL_RESULTS_BASEDIR', Path(self.directory.name)), \ + contextlib.redirect_stdout(io.StringIO()) as output: + run_ci.download_results_and_print_summary(None, 'unused', 'default', 58, + url, args, server=server) + destination = Path(self.directory.name) / directory / 'cassandra-6.0' / '58' + self.assertTrue((destination / 'console_log.txt.gz').is_file()) + self.assertTrue((destination / 'ci_summary_unknown_unknown_58.html').is_file()) + self.assertTrue((destination / 'results_details_unknown_unknown_58.tar.xz').is_file()) + self.assertIn(f'CI summary saved as {destination}/', output.getvalue()) + self.assertNotIn('https%3A', output.getvalue()) + self.assertNotIn('http%3A', output.getvalue()) + + def test_interrupted_download_does_not_leave_partial_artifacts(self): + args = self.args(url='https://ci.example', job='cassandra', repository=None, branch=None) + server = MagicMock() + response = server.jenkins_request.return_value.__enter__.return_value + def interrupted(**_kwargs): + yield b'partial data' + raise run_ci.requests.exceptions.ConnectionError('connection lost') + response.iter_content.side_effect = interrupted + destination = Path(self.directory.name) / 'results' + with patch('run_ci.LOCAL_RESULTS_BASEDIR', destination): + quietly(lambda: run_ci.download_results_and_print_summary(None, 'unused', 'default', 42, + args.url, args, server=server)) + self.assertFalse(any(path.is_file() for path in destination.rglob('*'))) + + @patch('run_ci.shutil.which', return_value=None) + def test_url_build_never_checks_cluster_tools(self, which): + args = self.args(url='https://ci.example', job='cassandra', user='alice') + server = MagicMock() + server.jenkins_open.return_value = json.dumps({'allBuilds': [], 'property': []}) + with patch('run_ci.load_environment_file'), patch('run_ci.parse_arguments', return_value=args), \ + patch('run_ci.get_jenkins', return_value=(args.url, server)), \ + patch('run_ci.is_local_git_dirty', return_value=False), \ + patch('run_ci.trigger_jenkins_build', return_value=100) as trigger, \ + patch('run_ci.wait_for_build_number', return_value=42), \ + patch('run_ci.wait_for_build_complete'), patch('run_ci.download_results_and_print_summary'), \ + patch('run_ci.setup_environment') as setup: + quietly(run_ci.main) + setup.assert_not_called() + which.assert_not_called() + trigger.assert_called_once() + + def test_missing_authenticated_artifact_is_not_retried(self): + calls = MagicMock(side_effect=jenkins.NotFoundException('not archived')) + with self.assertRaises(jenkins.NotFoundException): + run_ci.retry_call(calls, 'get artifact', run_ci.JENKINS_API_ERRORS, 5, 0) + calls.assert_called_once() + + +class TestBuildResultSummary(unittest.TestCase): + TOTALS = '''
Passed {total_passed_count}
Failed {total_failure_count}
Skipped {total_skipped_count}
+ + + + +
Passed 112234
Failed 216
Skipped 1950
Total    114400
''' + + def summary(self, console, totals=TOTALS): + args = run_ci.argument_parser().parse_args([]) + args.url, args.job = 'https://ci.example', 'cassandra-6.0' + args.repository, args.branch = None, None + server = MagicMock() + + def response(request, **_kwargs): + if request.url.endswith('/consoleText'): + content = console + elif request.url.endswith('/ci_summary.html'): + content = totals + else: + content = 'details archive' + if content is None: + raise jenkins.NotFoundException('not archived') + result = MagicMock() + result.__enter__.return_value.iter_content.return_value = [content.encode()] + return result + + server.jenkins_request.side_effect = response + with tempfile.TemporaryDirectory() as directory, \ + patch('run_ci.LOCAL_RESULTS_BASEDIR', Path(directory)), \ + contextlib.redirect_stdout(io.StringIO()) as output: + run_ci.download_results_and_print_summary(None, 'unused', 'default', 58, + args.url, args, server=server) + compressed = list(Path(directory).rglob('console_log.txt.gz')) + if console is not None: + self.assertEqual(len(compressed), 1) + with run_ci.gzip.open(compressed[0], 'rt') as log: + self.assertEqual(log.read(), console) + else: + self.assertEqual(compressed, []) + return output.getvalue().split('--- Build Summary ---\n', 1)[1] + + def test_unstable_result_does_not_print_interleaved_ant_failure_excerpt(self): + console = ('BUILD FAILED\n' + '/home/cassandra/cassandra/build.xml:1985: The following error occurred while executing this line:\n' + '[Checks API] No suitable checks publisher found.\n[Pipeline] sh\n' + ' [java] DEBUG [org.apache.cassandra.test.microbench.SnapshotTakingBench]\n' + '/home/cassandra/cassandra/build.xml:1377: Some test(s) failed.\nFinished: UNSTABLE\n') + output = self.summary(console) + self.assertIn('Jenkins result: UNSTABLE\n', output) + self.assertIn('Tests: Passed 112234 | Failed 216 | Skipped 1950 | Total 114400\n', output) + self.assertRegex(output, r'Console log saved as .*/console_log.txt.gz\n') + for excerpt in ('BUILD FAILED', '[Checks API]', '[Pipeline]', '[java]', 'build.xml:'): + self.assertNotIn(excerpt, output) + + def test_final_result_wins_over_earlier_log_messages(self): + output = self.summary('Example: Finished: FAILURE\nFinished: FAILURE\n' + 'Retry succeeded\nFinished: SUCCESS\n') + self.assertIn('Jenkins result: SUCCESS\n', output) + self.assertNotIn('Jenkins result: FAILURE', output) + + def test_failed_build_without_reports_keeps_result_and_full_log(self): + output = self.summary('BUILD FAILED\ncompiler error\nFinished: FAILURE\n', totals=None) + self.assertIn('Jenkins result: FAILURE\n', output) + self.assertIn('Test totals unavailable: CI summary missing.', output) + self.assertNotIn('No tests were run', output) + + def test_log_without_final_result_does_not_infer_failure(self): + output = self.summary('BUILD FAILED\n[Pipeline] More stages still running\n') + self.assertIn('Jenkins result unavailable in console log.', output) + self.assertNotIn('Jenkins result: FAILURE', output) + + def test_missing_console_preserves_test_totals(self): + output = self.summary(None) + self.assertIn('Missing console log.', output) + self.assertIn('Tests: Passed 112234 | Failed 216 | Skipped 1950 | Total 114400', output) + + def test_report_without_totals_does_not_claim_no_tests_ran(self): + output = self.summary('Finished: ABORTED\n', totals='Incomplete report') + self.assertIn('Jenkins result: ABORTED\n', output) + self.assertIn('Test totals unavailable in CI summary.', output) + self.assertNotIn('No tests were run', output) + + def test_legacy_failure_limit_marks_failed_and_total_as_lower_bounds(self): + for failures in (200, 216): + with self.subTest(failures=failures): + total = 112234 + failures + 1950 + report = self.TOTALS.replace(' data-failure-count-capped="false"', '') + report = report.replace('216', str(failures)).replace('114400', str(total)) + output = self.summary('Finished: UNSTABLE\n', totals=report) + self.assertIn(f'Tests: Passed 112234 | Failed {failures}+ | Skipped 1950 | Total {total}+\n', output) + self.assertIn('lower bounds', output) + + def test_legacy_failure_count_below_limit_is_exact(self): + report = self.TOTALS.replace(' data-failure-count-capped="false"', '') + report = report.replace('216', '199').replace('114400', '114383') + output = self.summary('Finished: UNSTABLE\n', totals=report) + self.assertIn('Failed 199 | Skipped 1950 | Total 114383\n', output) + self.assertNotIn('lower bounds', output) + + def test_uncapped_reports_keep_exact_counts_at_and_above_old_limit(self): + for failures in (200, 528): + with self.subTest(failures=failures): + total = 112234 + failures + 1950 + report = self.TOTALS.replace('216', str(failures)).replace('114400', str(total)) + # HTML formatting must not decide which totals we can read. + report = run_ci.BeautifulSoup(report, 'html.parser').prettify() + output = self.summary('Finished: UNSTABLE\n', totals=report) + self.assertIn(f'Failed {failures} | Skipped 1950 | Total {total}\n', output) + self.assertNotIn('lower bounds', output) + + def test_partial_totals_are_unavailable(self): + output = self.summary('Finished: ABORTED\n', totals='
Passed2
') + self.assertIn('Test totals unavailable in CI summary.', output) + + +class TestBuildAttachment(unittest.TestCase): + PARAMETERS = { + 'repository': 'https://github.com/example/cassandra', 'branch': 'feature/123', + 'profile': 'skinny', 'profile_custom_regexp': '', 'jdk': '', + 'dtest_repository': 'https://github.com/apache/cassandra-dtest.git', 'dtest_branch': 'trunk', + } + + def setUp(self): + self.args = run_ci.argument_parser().parse_args([]) + vars(self.args).update(self.PARAMETERS, url='https://ci.example/jenkins', user='alice') + self.server = MagicMock() + self.builds = {} + self.server.get_build_info.side_effect = lambda _job, number: self.builds[number] + self.job_info = {'allBuilds': [], 'property': [{'parameterDefinitions': [ + {'name': 'architecture', 'defaultParameterValue': {'value': 'amd64'}}]}]} + self.server.jenkins_open.side_effect = lambda *_args, **_kwargs: json.dumps(self.job_info) + # Exercise normal version-based job inference, without using GitHub or the cluster. + if hasattr(run_ci.base_job_name, '_cached_result'): + del run_ci.base_job_name._cached_result + self.addCleanup(lambda: vars(run_ci.base_job_name).pop('_cached_result', None)) + + def build(self, number, building=True): + parameters = dict(self.PARAMETERS, architecture='amd64') + self.builds[number] = {'building': building, 'result': None if building else 'SUCCESS', + 'actions': [{'parameters': [{'name': name, 'value': value} + for name, value in parameters.items()]}]} + self.job_info['allBuilds'].append({'number': number, 'building': building}) + + @contextlib.contextmanager + def invocation(self, answer=''): + with contextlib.ExitStack() as stack: + mocks = {} + for name, values in { + 'load_environment_file': {}, 'parse_arguments': {'return_value': self.args}, + 'get_jenkins': {'return_value': ('https://ci.example/jenkins', self.server)}, + 'setup_environment': {}, 'is_local_git_dirty': {'return_value': False}, + 'trigger_jenkins_build': {'return_value': 100}, 'wait_for_build_number': {'return_value': 999}, + 'wait_for_build_complete': {}, 'download_results_and_print_summary': {}, + 'delete_remote_junit_files': {}, 'cleanup_and_maybe_teardown': {}, 'threading.Thread': {}, + 'requests.head': {'return_value': MagicMock(status_code=200)}, + 'requests.get': {'return_value': MagicMock(text='')}, + }.items(): + mocks[name] = stack.enter_context(patch('run_ci.' + name, **values)) + mocks['input'] = stack.enter_context(patch('builtins.input', return_value=answer)) + stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + yield mocks + + def test_attach_number_infers_job_from_repository_and_branch(self): + for version, job in [('5.0.9', 'cassandra-5.0'), ('6.0-alpha1', 'cassandra-6.0'), ('7.0', 'cassandra')]: + with self.subTest(version=version): + vars(run_ci.base_job_name).pop('_cached_result', None) + with patch.dict(os.environ, {}, clear=True), \ + patch('run_ci.DEFAULT_REPO_URL', self.PARAMETERS['repository']), \ + patch('run_ci.DEFAULT_REPO_BRANCH', self.PARAMETERS['branch']), \ + patch('run_ci.sys.argv', ['run-ci', '--attach', '42', '--url', 'https://ci.example/jenkins']): + self.args = run_ci.parse_arguments() + self.assertIsNone(self.args.job) + self.build(42) + with self.invocation() as mocks: + mocks['requests.get'].return_value.text = f'' + run_ci.main() + mocks['wait_for_build_complete'].assert_called_once_with(self.server, job, 42) + mocks['download_results_and_print_summary'].assert_called_once() + mocks['trigger_jenkins_build'].assert_not_called() + mocks['input'].assert_not_called() + self.assertEqual(self.args.job, job) + + def test_matching_build_offers_attach_before_launch_or_cleanup(self): + self.args.url = None # Cluster connection must not start its background cleaner when attaching. + self.build(40) + self.build(42) + with self.invocation() as mocks: + run_ci.main() + mocks['input'].assert_called_once() + self.assertIn('42', mocks['input'].call_args.args[0]) + mocks['wait_for_build_complete'].assert_called_once_with(self.server, 'cassandra-5.0', 42) + mocks['download_results_and_print_summary'].assert_called_once() + for name in ('trigger_jenkins_build', 'is_local_git_dirty', 'delete_remote_junit_files', + 'threading.Thread'): + mocks[name].assert_not_called() + mocks['cleanup_and_maybe_teardown'].assert_called_once_with(None, None, 'default', False) + self.assertEqual(self.args.attach, 42) + + def test_prompt_attachment_preserves_teardown_after_download(self): + self.args.url = None + self.args.setup = True + self.args.tear_down = True + self.args.kubeconfig = 'config' + self.args.kubecontext = 'context' + self.build(42) + with self.invocation() as mocks, patch('run_ci.init_k8s_namespace'), \ + patch('run_ci.helm_installation_lock'), patch('run_ci.install_jenkins'), \ + patch('run_ci.wait_for_jenkins_http'), patch('run_ci.ensure_cassandra_job_parameters_visible'): + calls = MagicMock() + for name in ('wait_for_build_complete', 'download_results_and_print_summary', 'cleanup_and_maybe_teardown'): + calls.attach_mock(mocks[name], name) + run_ci.main() + self.assertEqual([call[0] for call in calls.mock_calls], + ['wait_for_build_complete', 'download_results_and_print_summary', 'cleanup_and_maybe_teardown']) + mocks['cleanup_and_maybe_teardown'].assert_called_once_with('config', 'context', 'default', True) + mocks['trigger_jenkins_build'].assert_not_called() + + def test_interrupted_prompt_attachment_does_not_teardown(self): + self.args.url = None + self.args.tear_down = True + self.build(42) + with self.invocation() as mocks: + mocks['wait_for_build_complete'].side_effect = KeyboardInterrupt + with self.assertRaises(KeyboardInterrupt): + run_ci.main() + mocks['download_results_and_print_summary'].assert_not_called() + mocks['cleanup_and_maybe_teardown'].assert_not_called() + + def test_declining_attachment_starts_new_build(self): + self.build(42) + with self.invocation(answer='n') as mocks: + run_ci.main() + mocks['input'].assert_called_once() + mocks['trigger_jenkins_build'].assert_called_once_with(self.server, 'cassandra-5.0', **self.PARAMETERS) + mocks['wait_for_build_complete'].assert_called_once_with(self.server, 'cassandra-5.0', 999) + + def test_every_submitted_parameter_and_job_default_must_match(self): + for name in (*self.PARAMETERS, 'architecture'): + with self.subTest(parameter=name): + self.build(42) + parameters = self.builds[42]['actions'][0]['parameters'] + next(parameter for parameter in parameters if parameter['name'] == name)['value'] = 'different' + with self.invocation() as mocks: + run_ci.main() + mocks['input'].assert_not_called() + mocks['trigger_jenkins_build'].assert_called_once() + + def test_completed_builds_and_missing_parameter_metadata_do_not_match(self): + self.build(40, building=False) + self.build(41) + self.builds[41]['actions'] = [] + self.build(42) + self.builds[42]['building'] = False # Finished between the job query and build query. + with self.invocation() as mocks: + run_ci.main() + mocks['input'].assert_not_called() + mocks['trigger_jenkins_build'].assert_called_once() + + def test_finds_old_running_pipeline_without_querying_executors_or_completed_builds(self): + for number in range(150, 42, -1): + self.build(number, building=False) + self.build(42) + self.builds[42]['result'] = 'UNSTABLE' # Pipelines can set a result before finishing. + self.args.job = 'team/cassandra-5.0' + with self.invocation() as mocks: + run_ci.main() + mocks['trigger_jenkins_build'].assert_not_called() + mocks['wait_for_build_complete'].assert_called_once_with(self.server, 'team/cassandra-5.0', 42) + self.server.get_running_builds.assert_not_called() + self.assertTrue(all(call.args == ('team/cassandra-5.0', 42) + for call in self.server.get_build_info.call_args_list)) + request = self.server.jenkins_open.call_args.args[0] + self.assertEqual(request.url, 'https://ci.example/jenkins/job/team/job/cassandra-5.0/api/json') + self.assertIn('allBuilds[number,building]', request.params['tree']) + + def test_inspection_failure_does_not_launch_a_duplicate(self): + self.server.jenkins_open.side_effect = jenkins.JenkinsException('unavailable') + with self.invocation() as mocks, self.assertRaises(jenkins.JenkinsException): + run_ci.main() + mocks['trigger_jenkins_build'].assert_not_called() + + def test_incomplete_job_response_does_not_launch_a_duplicate(self): + for response in ('not JSON', 'null', '{}'): + with self.subTest(response=response): + self.server.jenkins_open.side_effect = None + self.server.jenkins_open.return_value = response + with self.invocation() as mocks, self.assertRaises(jenkins.JenkinsException): + run_ci.main() + mocks['trigger_jenkins_build'].assert_not_called() + + def test_deleted_build_does_not_hide_an_older_match(self): + self.build(42) + self.build(43) + def lookup(_job, number): + if number == 43: + raise jenkins.NotFoundException('deleted') + return self.builds[number] + self.server.get_build_info.side_effect = lookup + with self.invocation() as mocks: + run_ci.main() + mocks['wait_for_build_complete'].assert_called_once_with(self.server, 'cassandra-5.0', 42) + mocks['trigger_jenkins_build'].assert_not_called() + + def test_requested_values_override_job_defaults_when_matching(self): + self.job_info['property'][0]['parameterDefinitions'].extend([ + {'name': 'branch', 'defaultParameterValue': {'value': 'trunk'}}, + {'name': 'profile', 'defaultParameterValue': {'value': 'post-commit'}}, + {'name': 'optional', 'defaultParameterValue': None}, + ]) + self.build(42) + with self.invocation() as mocks: + run_ci.main() + mocks['wait_for_build_complete'].assert_called_once_with(self.server, 'cassandra-5.0', 42) + mocks['trigger_jenkins_build'].assert_not_called() + + def test_invalid_prompt_answer_does_not_start_a_duplicate(self): + self.build(42) + with self.invocation() as mocks: + mocks['input'].side_effect = ['maybe', 'Y'] + run_ci.main() + self.assertEqual(mocks['input'].call_count, 2) + mocks['trigger_jenkins_build'].assert_not_called() + + def test_unavailable_prompt_leaves_matching_build_running(self): + self.build(42) + with self.invocation() as mocks: + mocks['input'].side_effect = EOFError + with self.assertRaises(SystemExit) as error: + run_ci.main() + self.assertIn('--attach 42', str(error.exception)) + mocks['trigger_jenkins_build'].assert_not_called() + self.server.stop_build.assert_not_called() + + +class TestSplitProgress(unittest.TestCase): + @staticmethod + def node(number, name, state='running', kind='PARALLEL', children=None, **extra): + return dict(id=str(number), name=name, state=state, type=kind, children=children or [], **extra) + + def tree(self, *branches): + return {'status': 'ok', 'data': {'stages': [ + self.node(1, 'jar', kind='STAGE', children=[self.node(2, 'jar jdk11', 'success')]), + self.node(3, 'Tests', kind='STAGE', children=list(branches)), + self.node(10000, 'Summary', kind='STAGE'), + ]}} + + def server(self, *graphs): + server = MagicMock(server='https://ci.example/jenkins/') + server.jenkins_open.side_effect = [json.dumps(graph) if isinstance(graph, dict) else graph for graph in graphs] + return server + + def test_counts_selected_branches_and_terminal_outcomes(self): + graph = self.tree(*(self.node(index + 10, f'test jdk11 {index + 1}/11', state) + for index, state in enumerate(['success', 'failure', 'unstable', 'aborted', + 'skipped', 'not_built', 'finished', 'running', + 'queued', 'paused', 'unknown']))) + counts = run_ci.count_test_splits(graph) + self.assertEqual(counts, dict(total=11, finished=7, failed=1, unstable=1, aborted=1, skipped=2, + running=1, queued=1, paused=1, unknown=1)) + + def test_retry_children_and_repeated_node_references_are_counted_once(self): + retry_attempt = self.node(21, 'test jdk11 1/2', 'failure', kind='STAGE') + branch = self.node(20, 'test jdk11 1/2', children=[retry_attempt]) + other = self.node(30, 'test jdk11 2/2', 'queued') + branch['nextSibling'] = other + graph = self.tree(self.node(10, 'parallel', kind='PARALLEL_BLOCK', children=[branch, other])) + counts = run_ci.count_test_splits(graph) + self.assertEqual((counts['total'], counts['finished'], counts['failed'], counts['running']), (2, 0, 0, 1)) + + def test_single_split_tasks_and_different_jdks_are_separate(self): + graph = self.tree(self.node(10, 'fqltool-test jdk11', 'success'), + self.node(11, 'fqltool-test jdk17', 'queued'), + self.node(12, 'test jdk11 1/20', 'running')) + counts = run_ci.count_test_splits(graph) + self.assertEqual((counts['total'], counts['finished']), (3, 1)) + + def test_all_branches_are_counted_without_a_display_limit(self): + graph = self.tree(*(self.node(i + 10, f'dtest jdk11 {i + 1}/400', 'success') for i in range(400))) + self.assertEqual(run_ci.count_test_splits(graph)['finished'], 400) + + def test_before_fanout_does_not_claim_zero_of_zero_complete(self): + for graph in (self.tree(), {'status': 'ok', 'data': {'stages': []}}): + self.assertIsNone(run_ci.count_test_splits(graph)) + + def test_unknown_state_is_not_counted_as_finished(self): + for state in ('new-plugin-state', 'failed', 'total'): + with self.subTest(state=state): + counts = run_ci.count_test_splits(self.tree(self.node(10, 'test jdk11 1/1', state))) + self.assertEqual((counts['total'], counts['finished'], counts['unknown']), (1, 0, 1)) + + def test_placeholder_branches_are_not_counted_as_selected_splits(self): + graph = self.tree(self.node(10, 'test jdk11 1/2', 'success'), + self.node(11, 'test jdk17 1/2', 'not_built', placeholder=True, + children=[self.node(12, 'nested unused branch')])) + self.assertEqual(run_ci.count_test_splits(graph)['total'], 1) + + def test_graph_poll_is_throttled_but_completion_refreshes_immediately(self): + graph = self.tree(self.node(10, 'test jdk11 1/1', 'success')) + server = self.server(graph, graph, graph) + progress = run_ci.SplitProgress(server, 'cassandra', 42) + with patch('run_ci.time.monotonic', side_effect=[0, 3, 14, 15, 16]): + for _ in range(3): + progress.update() + self.assertEqual(server.jenkins_open.call_count, 1) + progress.update() + self.assertEqual(server.jenkins_open.call_count, 2) + progress.update(complete=True) + self.assertEqual(server.jenkins_open.call_count, 3) + + def test_overview_endpoint_is_reused_when_stages_endpoint_is_missing(self): + graph = self.tree(self.node(10, 'test jdk11 1/1', 'success')) + server = self.server(jenkins.NotFoundException('no stages route'), graph, graph) + progress = run_ci.SplitProgress(server, 'cassandra', 42) + with patch('run_ci.time.monotonic', side_effect=[0, 15]): + progress.update() + progress.update() + self.assertEqual([call.args[0].url.rsplit('/42/', 1)[1] for call in server.jenkins_open.call_args_list], + ['stages/tree', 'pipeline-overview/tree', 'pipeline-overview/tree']) + self.assertIn('1/1 finished', progress.text) + + def test_absent_graph_is_retried_after_startup_with_backoff(self): + graph = self.tree(self.node(10, 'test jdk11 1/1', 'running')) + server = self.server(jenkins.NotFoundException('not ready'), jenkins.NotFoundException('not ready'), graph) + progress = run_ci.SplitProgress(server, 'cassandra', 42) + with patch('run_ci.time.monotonic', side_effect=[0, 15, 59, 60]): + for _ in range(3): + progress.update() + self.assertEqual(server.jenkins_open.call_count, 2) + self.assertEqual(progress.text, 'Splits: unavailable') + progress.update() + self.assertIn('0/1 finished', progress.text) + + def test_failed_and_aborted_splits_are_visible_in_summary(self): + counts = run_ci.count_test_splits(self.tree(self.node(10, 'test jdk11 1/3', 'failure'), + self.node(11, 'test jdk11 2/3', 'aborted'), + self.node(12, 'test jdk11 3/3', 'queued'))) + self.assertEqual(run_ci.format_split_progress(counts), + 'Splits: 2/3 finished (1 failed, 1 aborted) | 0 running | 1 queued') + + def test_polling_displays_progress_then_recovers_after_graph_failure(self): + graph = self.tree(self.node(10, 'test jdk11 1/2', 'success'), self.node(11, 'test jdk11 2/2', 'queued')) + server = self.server(graph, jenkins.JenkinsException('temporary outage'), graph, graph) + server.get_build_info.side_effect = [{'building': True, 'result': None}] * 3 + [ + {'building': False, 'result': 'SUCCESS'}] + output = io.StringIO() + with patch('run_ci.time.sleep'), patch('run_ci.time.monotonic', side_effect=range(0, 1000, 20)), \ + contextlib.redirect_stdout(output): + run_ci.wait_for_build_complete(server, 'team/cassandra-5.0', 42) + text = output.getvalue() + self.assertIn('Splits: 1/2 finished', text) + self.assertIn('1 queued', text) + self.assertIn('Splits: unavailable', text) + self.assertIn('status: SUCCESS', text) + self.assertGreater(text.rfind('Splits: 1/2 finished'), text.find('Splits: unavailable')) + request = server.jenkins_open.call_args.args[0] + self.assertEqual(request.url, 'https://ci.example/jenkins/job/team/job/cassandra-5.0/42/stages/tree') + self.assertFalse(server.jenkins_open.call_args.kwargs['add_crumb']) + + def test_missing_plugin_or_bad_response_does_not_stop_waiting(self): + for result in (jenkins.NotFoundException('no plugin'), jenkins.JenkinsException('forbidden'), + run_ci.requests.exceptions.Timeout('timeout'), 'not JSON', + {'status': 'error'}, {'status': 'ok', 'data': {'stages': 'invalid'}}): + with self.subTest(result=result): + server = self.server() + server.jenkins_open.side_effect = result if isinstance(result, Exception) else None + if not isinstance(result, Exception): + server.jenkins_open.return_value = json.dumps(result) if isinstance(result, dict) else result + server.get_build_info.side_effect = [{'building': True, 'result': None}, + {'building': False, 'result': 'SUCCESS'}] + output = io.StringIO() + with patch('run_ci.time.sleep'), contextlib.redirect_stdout(output): + run_ci.wait_for_build_complete(server, 'cassandra', 42) + self.assertIn('Splits: unavailable', output.getvalue()) + self.assertIn('status: SUCCESS', output.getvalue()) + + def test_attach_reconstructs_existing_progress(self): + server = self.server(self.tree(self.node(10, 'test jdk11 1/2', 'success'), + self.node(11, 'test jdk11 2/2', 'running')), + self.tree(self.node(10, 'test jdk11 1/2', 'success'), + self.node(11, 'test jdk11 2/2', 'success'))) + server.get_build_info.side_effect = [{'actions': []}, {'building': True, 'result': None}, + {'building': False, 'result': 'SUCCESS'}] + args = argparse.Namespace(job='cassandra', attach=42) + output = io.StringIO() + with patch('run_ci.time.sleep'), contextlib.redirect_stdout(output), \ + patch('run_ci.download_results_and_print_summary') as download: + run_ci.attach_build(None, server.server, server, args) + self.assertIn('Splits: 1/2 finished', output.getvalue()) + download.assert_called_once() + server.build_job.assert_not_called() + + def test_spinner_clears_old_counts_and_restores_cursor_on_interrupt(self): + output = io.StringIO() + with patch('run_ci.time.sleep'), contextlib.redirect_stdout(output): + with self.assertRaises(KeyboardInterrupt): + run_ci.spin_while(lambda: 'Splits: 1/2 finished', MagicMock(side_effect=[False, KeyboardInterrupt])) + self.assertIn('Splits: 1/2 finished', output.getvalue()) + self.assertTrue(output.getvalue().endswith('\r\033[K\033[?25h')) + + if __name__ == '__main__': unittest.main() diff --git a/.github/workflows/jenkins-check.yaml b/.github/workflows/jenkins-check.yaml index 6b39f9dc10c8..00048770120c 100644 --- a/.github/workflows/jenkins-check.yaml +++ b/.github/workflows/jenkins-check.yaml @@ -10,12 +10,14 @@ on: paths: - '.build/run-ci' - '.build/run-ci.d/**' + - '.build/ci/**' - '.jenkins/**' - '.github/workflows/jenkins-check.yaml' pull_request: paths: - '.build/run-ci' - '.build/run-ci.d/**' + - '.build/ci/**' - '.jenkins/**' - '.github/workflows/jenkins-check.yaml' @@ -31,11 +33,13 @@ jobs: with: python-version: '3.x' - name: Install dependencies - run: pip install pylint -r .build/run-ci.d/requirements.txt + run: pip install pylint -r .build/run-ci.d/requirements.txt -r .build/ci/requirements.txt - name: Lint run-ci run: pylint --disable=C0301,C0302,W0511,C0103,W0702,C0415,C0116,C0115,R0914,W0603,R0915,R0913,R0917,R0911,W0212,W0621 .build/run-ci - name: Run run-ci unit tests run: python .build/run-ci.d/run-ci-test.py + - name: Test CI report totals + run: python .build/ci/ci_parser_test.py - name: Check run-ci argument parsing run: python .build/run-ci --help diff --git a/.jenkins/Jenkinsfile b/.jenkins/Jenkinsfile index 6b900b77d3ab..a87180a0b3b4 100644 --- a/.jenkins/Jenkinsfile +++ b/.jenkins/Jenkinsfile @@ -96,9 +96,9 @@ pipeline { } parameters { string(name: 'repository', defaultValue: params.repository ?: scm.userRemoteConfigs[0].url, description: 'Cassandra Repository') - string(name: 'branch', defaultValue: params.branch ?: scm.branch, description: 'Branch') + string(name: 'branch', defaultValue: params.branch?.trim() ?: scm.branches[0].name.replaceFirst(/^\*\//, ''), description: 'Branch') - choice(name: 'profile', choices: pipelineProfileNames(params.profile ?: ''), description: 'Pick a pipeline profile.') + choice(name: 'profile', choices: pipelineProfileNames(params.profile), description: 'Pick a pipeline profile.') string(name: 'profile_custom_regexp', defaultValue: params.profile_custom_regexp ?: '', description: 'Regexp for stages when using custom profile. See `testSteps` in Jenkinsfile for list of stages. Example: stress.*|jvm-dtest.*') choice(name: 'architecture', choices: archsSupported + "all", description: 'Pick architecture. The ARM64 is disabled by default at the moment.') @@ -175,10 +175,9 @@ def pipelineProfiles() { @NonCPS def pipelineProfileNames(putFirst) { - set = pipelineProfiles().keySet() as List - set = set - putFirst - set.add(0, putFirst) - return set + def names = pipelineProfiles().keySet() as List + def selected = names.contains(putFirst) ? putFirst : 'skinny' + return [selected] + (names - selected) } @NonCPS diff --git a/.jenkins/k8s/README.md b/.jenkins/k8s/README.md index b8d631d7595a..5651011565f2 100644 --- a/.jenkins/k8s/README.md +++ b/.jenkins/k8s/README.md @@ -70,6 +70,8 @@ kubectl exec -it svc/cassius-jenkins -c jenkins -- /bin/cat /run/secrets/additio This leaves the controller running, a single e2-standard-8 instance. All other node-pools downscale to zero. +The shared job setup preserves saved build parameters across redeploys on both GKE and EKS. New jobs, and existing jobs with blank defaults, get their configured SCM branch (`trunk`, `cassandra-6.0`, or `cassandra-5.0`) and the `skinny` profile. Parameter forms are available before the first build. The branch parameter selects the code to test; the job's SCM configuration still selects which Jenkinsfile to load. + ## Upgrading an existing instance (e.g. pre-ci.cassandra.apache.org) A long-lived site like pre-ci.cassandra.apache.org may carry customisations: hostname, cloud load-balancer, storage class; that are deliberately absent from `jenkins-deployment.yaml`. diff --git a/.jenkins/k8s/jenkins-deployment.yaml b/.jenkins/k8s/jenkins-deployment.yaml index 37e257da19b5..63e4b607c651 100644 --- a/.jenkins/k8s/jenkins-deployment.yaml +++ b/.jenkins/k8s/jenkins-deployment.yaml @@ -102,54 +102,55 @@ controller: # TODO: add new version each release branching test-job: | jobs: - - script: > - pipelineJob('cassandra') { - definition { - cpsScm { - scm { - git { - remote { - url('https://github.com/apache/cassandra') - } - branch('trunk') - scriptPath('.jenkins/Jenkinsfile') - } - } - lightweight() - } + - script: | + import javaposse.jobdsl.dsl.JobConfigurationNotFoundException + + ['cassandra': 'trunk', 'cassandra-6.0': 'cassandra-6.0', 'cassandra-5.0': 'cassandra-5.0'].each { jobName, scmBranch -> + // Job DSL replaces config.xml on reload. Preserve the defaults saved by the Jenkinsfile. + def savedParameters = [] + try { + savedParameters = new XmlParser().parseText(jm.getConfig(jobName)).properties. + 'hudson.model.ParametersDefinitionProperty'.parameterDefinitions[0]?.children() ?: [] + } catch (JobConfigurationNotFoundException ignored) { + // A new job gets the defaults below before its first build. } - } - - script: > - pipelineJob('cassandra-6.0') { - definition { - cpsScm { - scm { - git { - remote { - url('https://github.com/apache/cassandra') - } - branch('cassandra-6.0') - scriptPath('.jenkins/Jenkinsfile') + pipelineJob(jobName) { + parameters { + stringParam('repository', 'https://github.com/apache/cassandra', 'Cassandra Repository') + stringParam('branch', scmBranch, 'Branch') + choiceParam('profile', ['skinny', 'packaging', 'pre-commit', 'pre-commit w/ upgrades', 'post-commit', 'performance', 'custom'], 'Pick a pipeline profile.') + stringParam('profile_custom_regexp', '', 'Regexp for stages when using custom profile. See testSteps in Jenkinsfile.') + choiceParam('architecture', ['amd64', 'arm64', 'all'], 'Pick architecture.') + stringParam('jdk', '', 'Restrict JDK versions.') + stringParam('dtest_repository', 'https://github.com/apache/cassandra-dtest', 'Cassandra DTest Repository') + stringParam('dtest_branch', 'trunk', 'DTest Branch') + } + configure { project -> + def definitions = project.properties.'hudson.model.ParametersDefinitionProperty'.parameterDefinitions[0] + savedParameters.each { saved -> + def name = saved.name.text() + if (name == 'branch' && !saved.defaultValue.text().trim()) return + if (name == 'profile') { + def firstChoice = saved.choices[0]?.depthFirst()?.find { it instanceof Node && it.name() == 'string' } + if (!firstChoice?.text()?.trim()) return } + definitions.children().removeAll { it.name.text() == name } + definitions.append(saved.clone()) } - lightweight() } - } - } - - script: > - pipelineJob('cassandra-5.0') { - definition { - cpsScm { - scm { - git { - remote { - url('https://github.com/apache/cassandra') + definition { + cpsScm { + scm { + git { + remote { + url('https://github.com/apache/cassandra') + } + branch(scmBranch) + scriptPath('.jenkins/Jenkinsfile') } - branch('cassandra-5.0') - scriptPath('.jenkins/Jenkinsfile') } + lightweight() } - lightweight() } } } diff --git a/.jenkins/k8s/jenkins-parameters-test.groovy b/.jenkins/k8s/jenkins-parameters-test.groovy new file mode 100644 index 000000000000..22252b0ea74d --- /dev/null +++ b/.jenkins/k8s/jenkins-parameters-test.groovy @@ -0,0 +1,235 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +import groovy.json.JsonSlurper +import groovy.xml.XmlUtil +import org.codehaus.groovy.control.CompilerConfiguration +import org.codehaus.groovy.control.customizers.ImportCustomizer + +// Execute the checked-in scripts with in-memory job storage and a small Job DSL adapter. +// This checks parameter XML across reloads without a Jenkins controller or build agents. +class SeedJob { + Node xml = new Node(null, 'flow-definition') + List configureBlocks = [] + String scmBranch + + SeedJob() { xml.appendNode('properties') } + + void runContext(Closure body) { + def action = body.rehydrate(this, body.owner, body.thisObject) + action.resolveStrategy = Closure.DELEGATE_FIRST + action() + } + + void parameters(Closure body) { + xml.properties[0].appendNode('hudson.model.ParametersDefinitionProperty').appendNode('parameterDefinitions') + runContext(body) + } + + void stringParam(String name, String value, String description) { + def node = definitions().appendNode('hudson.model.StringParameterDefinition') + node.appendNode('name', name) + node.appendNode('description', description) + node.appendNode('defaultValue', value) + } + + void choiceParam(String name, List values, String description) { + def node = definitions().appendNode('hudson.model.ChoiceParameterDefinition') + node.appendNode('name', name) + node.appendNode('description', description) + def choices = node.appendNode('choices', [class: 'java.util.Arrays$ArrayList']) + .appendNode('a', [class: 'string-array']) + values.each { choices.appendNode('string', it) } + } + + Node definitions() { xml.properties[0].'hudson.model.ParametersDefinitionProperty'[0].parameterDefinitions[0] } + void configure(Closure body) { configureBlocks.add(body) } + + def methodMissing(String name, args) { + if (name in ['definition', 'cpsScm', 'scm', 'git', 'remote']) { + runContext(args[0]) + } else if (name == 'branch') { + scmBranch = args[0] + } else if (!(name in ['url', 'scriptPath', 'lightweight'])) { + throw new MissingMethodException(name, getClass(), args) + } + } + + Node render() { + configureBlocks.each { it.call(xml) } + xml.appendNode('definition').appendNode('scm').appendNode('branch', scmBranch) + xml + } +} + +class ParameterDsl { + void agent(Closure ignored) {} + void options(Closure ignored) {} + void stages(Closure ignored) {} + void post(Closure ignored) {} + void parameters(Closure body) { body.call() } +} + +// Jenkins uses Groovy 2; the existing validation container uses Groovy 4. +def xmlParserName = GroovySystem.version.startsWith('2.') ? 'groovy.util.XmlParser' : 'groovy.xml.XmlParser' +def xmlParserClass = getClass().classLoader.loadClass(xmlParserName) +def parseXml = { text -> xmlParserClass.newInstance().parseText(text) } +def compiler = new CompilerConfiguration() +compiler.addCompilationCustomizers(new ImportCustomizer().addImports(xmlParserName)) +def loader = new GroovyClassLoader(getClass().classLoader) +def missingConfig = loader.parseClass(''' + package javaposse.jobdsl.dsl + class JobConfigurationNotFoundException extends RuntimeException {} +''') +loader.parseClass('@interface NonCPS {}') +loader.parseClass('package hudson; class AbortException extends IOException {}') +def scripts = new JsonSlurper().parse(new File(args[0])) +def store = [:] +def readFailure = null +def reload = { + def generated = [:] + def management = new Expando(getConfig: { String name -> + if (readFailure != null) throw readFailure + if (!store.containsKey(name)) throw missingConfig.newInstance() + store[name] + }) + scripts.each { source -> + def binding = new Binding(jm: management, pipelineJob: { String name, Closure body -> + def job = new SeedJob() + job.runContext(body) + generated[name] = job + }) + new GroovyShell(loader, binding, compiler).evaluate(source) + } + generated.each { name, job -> store[name] = XmlUtil.serialize(job.render()) } +} +def definitions = { String name -> + parseXml(store[name]).properties.'hudson.model.ParametersDefinitionProperty'.parameterDefinitions[0] +} +def parameter = { String job, String name -> definitions(job)?.children()?.find { it.name.text() == name } } +def choices = { Node definition -> definition == null ? [] : definition.choices[0].depthFirst().findAll { it instanceof Node && it.name() == 'string' }*.text() } +def failures = [] +def check = { String name, Closure test -> + try { + test() + println "PASS ${name}" + } catch (AssertionError error) { + failures.add(name) + println "FAIL ${name}: ${error.message}" + } +} + +reload() +check('fresh jobs have branch and profile defaults before their first build') { + ['cassandra': 'trunk', 'cassandra-6.0': 'cassandra-6.0', 'cassandra-5.0': 'cassandra-5.0'].each { job, branch -> + assert parameter(job, 'branch')?.defaultValue?.text() == branch + assert choices(parameter(job, 'profile'))[0] == 'skinny' + assert definitions(job).children().size() == 8 + assert choices(parameter(job, 'architecture')) == ['amd64', 'arm64', 'all'] + } +} + +// A build can update defaults and add parameter definitions unknown to the seed script. +store['cassandra'] = ''' + + branchfeature/CASSANDRA-123 + profilepre-commitskinnycustom + repositoryhttps://github.com/example/cassandra + extratrue + +old-definition +''' +2.times { reload() } +check('repeated reloads preserve saved defaults and extra parameters while updating SCM') { + assert parameter('cassandra', 'branch')?.defaultValue?.text() == 'feature/CASSANDRA-123' + assert choices(parameter('cassandra', 'profile')) == ['pre-commit', 'skinny', 'custom'] + assert parameter('cassandra', 'repository')?.defaultValue?.text() == 'https://github.com/example/cassandra' + assert parameter('cassandra', 'extra')?.defaultValue?.text() == 'true' + assert definitions('cassandra').children().size() == 9 + assert parseXml(store.cassandra).definition.scm.branch.text() == 'trunk' + assert parameter('cassandra-5.0', 'branch')?.defaultValue?.text() == 'cassandra-5.0' +} + +store['cassandra'] = store['cassandra'].replace('feature/CASSANDRA-123', ' ').replace('pre-commit', '') +store['cassandra-6.0'] = '' +reload() +check('reload repairs blank defaults and jobs with no parameter property') { + assert parameter('cassandra', 'branch')?.defaultValue?.text() == 'trunk' + assert choices(parameter('cassandra', 'profile'))[0] == 'skinny' + assert parameter('cassandra-6.0', 'branch')?.defaultValue?.text() == 'cassandra-6.0' + assert choices(parameter('cassandra-6.0', 'profile'))[0] == 'skinny' + assert parameter('cassandra', 'extra')?.defaultValue?.text() == 'true' +} + +check('a config read failure cannot replace saved parameters with defaults') { + def before = store.clone() + readFailure = new SecurityException('read denied') + try { + reload() + assert false: 'reload must fail when existing configuration cannot be read' + } catch (SecurityException expected) { + assert store == before + } finally { + readFailure = null + } +} + +// Capture just the real Jenkinsfile's parameter declarations; never execute build stages. +def pipelineParameters = { Map params, String branch -> + def declarations = [:] + def script = new GroovyShell(loader, new Binding(params: params, + scm: [branch: null, branches: [[name: branch]], userRemoteConfigs: [[url: 'https://github.com/apache/cassandra']]]), compiler) + .parse(new File(args[1]).text, 'CassandraPipeline.groovy') + script.metaClass.pipeline = { Closure body -> + def dsl = new ParameterDsl() + def action = body.rehydrate(dsl, script, script) + action.resolveStrategy = Closure.DELEGATE_FIRST + action() + } + script.metaClass.string = { Map definition -> declarations[definition.name] = definition } + script.metaClass.choice = { Map definition -> declarations[definition.name] = definition } + script.run() + declarations +} +check('Jenkinsfile has nonblank defaults with missing or blank parameters') { + [[:], [branch: '', profile: ''], [branch: ' ', profile: ' ']].each { params -> + def result = pipelineParameters(params, '*/cassandra-6.0') + assert result.branch.defaultValue == 'cassandra-6.0' + assert result.profile.choices[0] == 'skinny' + assert result.profile.choices.every { it.trim() } + assert result.profile.choices.unique(false) == result.profile.choices + } +} +check('Jenkinsfile retains a selected branch and valid profile') { + def result = pipelineParameters([branch: 'feature/CASSANDRA-456', profile: 'pre-commit'], 'trunk') + assert result.branch.defaultValue == 'feature/CASSANDRA-456' + assert result.profile.choices[0] == 'pre-commit' +} +check('seeded defaults remain selected when the Jenkinsfile defines parameters') { + ['cassandra', 'cassandra-6.0', 'cassandra-5.0'].each { job -> + def result = pipelineParameters([ + branch: parameter(job, 'branch')?.defaultValue?.text(), + profile: choices(parameter(job, 'profile'))[0] + ], parseXml(store[job]).definition.scm.branch.text()) + assert result.branch.defaultValue == (job == 'cassandra' ? 'trunk' : job) + assert result.profile.choices[0] == 'skinny' + assert result.profile.choices.toSet() == choices(parameter(job, 'profile')).toSet() + } +} +check('Jenkinsfile replaces a removed profile with skinny') { + assert pipelineParameters([profile: 'removed-profile'], 'trunk').profile.choices[0] == 'skinny' +} +assert failures.empty: "${failures.size()} parameter regression(s) failed: ${failures.join(', ')}" diff --git a/.jenkins/k8s/jenkins-test.sh b/.jenkins/k8s/jenkins-test.sh index eef4f47f7e3d..4050396f0cd4 100755 --- a/.jenkins/k8s/jenkins-test.sh +++ b/.jenkins/k8s/jenkins-test.sh @@ -18,6 +18,7 @@ # # Validates the CI declarations under .jenkins/ without deploying anything: # - the Jenkinsfile parses as groovy +# - job parameters survive redeploys and have nonblank branch/profile defaults # - jenkins-deployment.yaml renders through the jenkins helm chart # - the yaml embedded in it (agent pod templates, JCasC config scripts) parses # @@ -50,12 +51,25 @@ args.each { cu.addSource(new File(it)) } cu.compile(Phases.CONVERSION) println " ${args.join(', ')} parses" EOF +python3 - "${JENKINS_DIR}/k8s/jenkins-deployment.yaml" "${syntax_check_dir}/job-seeds.json" << 'EOF' +import json, sys, yaml +from pathlib import Path + +values = yaml.safe_load(Path(sys.argv[1]).read_text(encoding="utf-8")) +jobs = yaml.safe_load(values["controller"]["JCasC"]["configScripts"]["test-job"])["jobs"] +Path(sys.argv[2]).write_text(json.dumps([job["script"] for job in jobs]), encoding="utf-8") +EOF if command -v groovy > /dev/null ; then groovy "${syntax_check}" "${JENKINS_DIR}/Jenkinsfile" || status=1 + groovy "${JENKINS_DIR}/k8s/jenkins-parameters-test.groovy" "${syntax_check_dir}/job-seeds.json" \ + "${JENKINS_DIR}/Jenkinsfile" || status=1 elif command -v docker > /dev/null ; then # absolute paths, the image's working directory is not where the script was mounted docker run --rm -v "${syntax_check_dir}:/check:ro" -v "${JENKINS_DIR}:/jenkins:ro" \ groovy:4.0-jdk17 groovy /check/syntax-check.groovy /jenkins/Jenkinsfile || status=1 + docker run --rm -v "${syntax_check_dir}:/check:ro" -v "${JENKINS_DIR}:/jenkins:ro" \ + groovy:4.0-jdk17 groovy /jenkins/k8s/jenkins-parameters-test.groovy /check/job-seeds.json \ + /jenkins/Jenkinsfile || status=1 else echo " SKIPPED: neither groovy nor docker found" fi