Function wait

misc/python/materialize/cloudtest/util/wait.py:21–76

(
    condition: str,
    resource: str,
    timeout_secs: int = 300,
    context: str = DEFAULT_K8S_CONTEXT_NAME,
    *,
    label: str | None = None,
    namespace: str | None = None,
    server: str | None = None,
)

def wait(
    condition: str,
    resource: str,
    timeout_secs: int = 300,
    context: str = DEFAULT_K8S_CONTEXT_NAME,
    *,
    label: str | None = None,
    namespace: str | None = None,
    server: str | None = None,
) -> None:
    # Use a shorter per-invocation timeout so the outer retry loop can
    # handle transient kubectl errors (e.g. slow image pulls, apiserver blips).
    per_attempt_secs = min(timeout_secs, 60)
    cmd = [
        "kubectl",
        "wait",
        "--for",
        condition,
        resource,
        "--timeout",
        f"{per_attempt_secs}s",
        "--context",
        context,
    ]

    if label is not None:
        cmd.extend(["--selector", label])

    if namespace is not None:
        cmd.extend(["--namespace", namespace])

    if server is not None:
        cmd.extend(["--server", server])

    ui.progress(f'waiting for {" ".join(cmd)} ... ')

    error = None
    for remaining in ui.timeout_loop(timeout_secs, tick=0.1):
        try:
            output = subprocess.check_output(cmd, stderr=subprocess.STDOUT).decode(
                "ascii"
            )
            # output is:
            # - an empty string when a 'delete' condition is satisfied
            # - 'condition met' for all other conditions
            if len(output) == 0 or "condition met" in output:
                ui.progress("success!", finish=True)
                return
        except subprocess.CalledProcessError as e:
            # use a less verbose output than log_subprocess_error here
            LOGGER.info(f"{e} {e.output.decode('ascii')}")
            error = e

    ui.progress(finish=True)
    print_pods()
    raise UIError(f"kubectl wait never returned 'condition met': {error}")