This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12688-a9db0e65428bd95dfd00ee21484f10dc6908db27 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 714aeed0c486818d91e83d73b4d8e5dfbfbacc12 Author: SEZ <[email protected]> AuthorDate: Wed Oct 14 12:52:26 2026 +0000 [Feature][seatunnel-cli] Diagnose a gradual slowdown from checkpoint history (#12688) --- seatunnel-cli/README.md | 85 +++ seatunnel-cli/README.zh-CN.md | 74 +++ seatunnel-cli/seatunnel_cli/cli.py | 217 +++++++ .../seatunnel_cli/diagnostics/checkpoints.py | 468 ++++++++++++++ seatunnel-cli/seatunnel_cli/diagnostics/jobs.py | 525 ++++++++++++++++ seatunnel-cli/tests/test_diagnose_checkpoints.py | 576 +++++++++++++++++ seatunnel-cli/tests/test_diagnose_cli.py | 575 +++++++++++++++++ seatunnel-cli/tests/test_diagnose_job.py | 683 +++++++++++++++++++++ 8 files changed, 3203 insertions(+) diff --git a/seatunnel-cli/README.md b/seatunnel-cli/README.md index 58f6888e0c..71d1d07e29 100644 --- a/seatunnel-cli/README.md +++ b/seatunnel-cli/README.md @@ -276,6 +276,90 @@ seatunnel "从 Kafka 读取订单数据写入 ClickHouse" -o my_job.conf seatunnel "Read CSV files and write to Elasticsearch" --provider openai --model gpt-4o ``` +### Diagnosing a Job + +`--diagnose` answers "why is this job not making progress?" from what the +engine already publishes. It is read-only -- a `GET /job-info/<id>` and a +`GET /jobs/checkpoints/<id>`, never a submit or a stop -- and uses no LLM at +all, so it works with no API key and no provider configured: + +```bash +seatunnel --diagnose 852362670771666945 + +# Another host, or a cluster whose enable-dynamic-port moved the listener +SEATUNNEL_API_BASE=http://zeta-master:8080 seatunnel --diagnose 852362670771666945 +``` + +`/job-info` is part of the v2 REST API, which Zeta serves from its Jetty HTTP +port -- `8080` in the packaged `config/seatunnel.yaml`. Point +`SEATUNNEL_API_BASE` there if your cluster is elsewhere or uses another port. + +``` +Job 852362670771666945 mysql-to-doris — RUNNING + [warning] Pipeline 1 has restarted 4 times and its current attempt is only 20s old. + It is failing and being restored repeatedly, so the job looks alive while + making no progress. (restoreCount=4/10 pipelineStatus=RUNNING running for 20s + since the last restore) +``` + +What it reports: the parsed error code and root cause of a failed job, a +pipeline restarting in a loop (and whether it has used up its restore budget), +a job that has sat in a pre-running state too long to be normal, and a running +job whose source reads nothing or whose sink writes nothing. Exit status is +non-zero only when the job itself is `FAILED`, so a stuck-but-live job does not +look like a command failure. + +It also reports **why a job that used to be fast has got slower**, which the +job state alone cannot explain: a job that is lagging is still `RUNNING` with +both row counters moving. Checkpoint history supplies the missing comparison: + +``` +Job 852362670771666945 mysql-cdc-to-doris — RUNNING + [warning] Pipeline 1 checkpoints are getting slower: recently 48.0s against 4.0s + earlier in the retained history. (median durationMillis 4000 -> 48000 over 16 checkpoints) + [warning] Pipeline 1 now spends almost all its time checkpointing: a checkpoint takes + 48.0s and one is triggered every 1m00s. (recent median duration 48000ms vs interval 60000ms) + [info] Pipeline 1 checkpoint state is growing: recently 120.0MiB against 12.0MiB earlier. +``` + +A stalled checkpoint is also named down to the subtasks holding up the barrier +(`3 of 8 subtasks have acknowledged it`), which points at one stage rather than +at the job as a whole. + +The engine keeps the last 32 checkpoints per pipeline, so the trend covers that +window rather than the whole run. The "spends its time checkpointing" rule +needs no baseline and so still fires on a job that degraded earlier and has +been slow since. If the cluster predates the checkpoint endpoint, or its +monitor service is off, the rest of the diagnosis is printed unchanged, with +one line noting that this part of the analysis did not run. + +Every rule is deliberately limited to what one response can prove, because a +wrong hint costs more than a missing one: + +- `restoreCount` counts restores since submission, so what it means depends on + the pipeline's current state. A pipeline that is `FINISHED` or `CANCELED` is + over and says nothing; a `FAILED` one is reported only if it spent its whole + restore budget, in the past tense; one that is shutting down reports its + restores as history. A loop is only claimed when the pipeline is between + attempts, or has been running for less than 10 minutes since its last + restore — so a loop whose attempts each survive longer than that is reported + as history rather than as a live fault. +- the row-count rules wait until the job has been RUNNING for two minutes, and + for a `job.mode = STREAMING` job an idle source is reported as a note rather + than a warning -- a quiet topic and a misconfigured one read the same here. +- `DOING_SAVEPOINT` is reported as a duration, not as a hung task: a large + state legitimately takes minutes to write. +- a checkpoint failure is only a warning while the **latest** failure is newer + than the latest completion. `counts.failed` is cumulative, so one expired + checkpoint would otherwise keep it above zero forever. +- the stalled-checkpoint rule fires after 2 minutes in progress, which is above + the packaged `checkpoint.timeout` of 60s. On default settings a hung + checkpoint expires first and is reported as a failure instead, so this rule + matters on clusters that have raised that timeout. +- durations are measured against the engine's own clock -- the job rules use + `diagnostics.generatedAt`, the checkpoint rules use `updatedAt` -- so a + client clock that disagrees with the cluster cannot invent a stuck job. + ### CLI Arguments ``` @@ -289,6 +373,7 @@ Options: --provider PROVIDER LLM provider: bedrock | bedrock-mantle | anthropic | openai | orcarouter --model MODEL Override primary model ID --fast-model MODEL Override fast model ID + --diagnose JOB_ID Explain a job's current state and exit (read-only, no LLM) --sync-catalog PATH Regenerate connector catalog from SeaTunnel source -V, --version Show version -h, --help Show help message diff --git a/seatunnel-cli/README.zh-CN.md b/seatunnel-cli/README.zh-CN.md index f57c82fc23..a5481d4767 100644 --- a/seatunnel-cli/README.zh-CN.md +++ b/seatunnel-cli/README.zh-CN.md @@ -244,6 +244,79 @@ seatunnel "从 Kafka 读取订单数据写入 ClickHouse" -o my_job.conf seatunnel "Read CSV files and write to Elasticsearch" --provider openai --model gpt-4o ``` +### 诊断作业 + +`--diagnose` 用引擎已经暴露的信息回答"这个作业为什么不往前走"。它是只读的—— +只发 `GET /job-info/<id>` 和 `GET /jobs/checkpoints/<id>`，不会提交也不会停止 +作业——并且完全不调用 LLM，因此在没有 API key、没有配置任何提供商的情况下也能用： + +```bash +seatunnel --diagnose 852362670771666945 + +# 集群在其他主机上，或 enable-dynamic-port 让监听端口发生了变化 +SEATUNNEL_API_BASE=http://zeta-master:8080 seatunnel --diagnose 852362670771666945 +``` + +`/job-info` 属于 v2 REST API，由 Zeta 的 Jetty HTTP 端口提供服务——打包配置 +`config/seatunnel.yaml` 中是 `8080`。集群在别处或端口不同时，请把 +`SEATUNNEL_API_BASE` 指向该地址。 + +``` +Job 852362670771666945 mysql-to-doris — RUNNING + [warning] Pipeline 1 has restarted 4 times and its current attempt is only 20s old. + It is failing and being restored repeatedly, so the job looks alive while + making no progress. (restoreCount=4/10 pipelineStatus=RUNNING running for 20s + since the last restore) +``` + +它会报告：失败作业解析出的错误码与根因、反复重启的 pipeline（以及是否已用尽 +重启次数）、在运行前状态停留过久的作业、以及源端读不到数据或目标端写不出数据的 +运行中作业。只有作业本身处于 `FAILED` 时退出码才非零——卡住但仍存活的作业不会被 +当成命令执行失败。 + +它还会回答**原本很快的作业为什么变慢了**——这是作业状态本身无法解释的：出现延迟的 +作业仍然是 `RUNNING`，两侧的行数也都在涨。checkpoint 历史补上了缺失的那个对比： + +``` +Job 852362670771666945 mysql-cdc-to-doris — RUNNING + [warning] Pipeline 1 checkpoints are getting slower: recently 48.0s against 4.0s + earlier in the retained history. (median durationMillis 4000 -> 48000 over 16 checkpoints) + [warning] Pipeline 1 now spends almost all its time checkpointing: a checkpoint takes + 48.0s and one is triggered every 1m00s. (recent median duration 48000ms vs interval 60000ms) + [info] Pipeline 1 checkpoint state is growing: recently 120.0MiB against 12.0MiB earlier. +``` + +卡住的 checkpoint 还会定位到是哪些 subtask 在挡住 barrier（`3 of 8 subtasks have +acknowledged it`），指向具体的某一环而不是整个作业。 + +引擎每个 pipeline 只保留最近 32 个 checkpoint，因此趋势只覆盖这个窗口，而非整个 +运行周期。「几乎全部时间都在做 checkpoint」这条规则不需要基线，所以对于更早就已经 +退化、之后一直慢的作业同样有效。如果集群版本没有该 checkpoint 端点，或其监控服务 +未开启，诊断的其余部分照常输出，并会额外打印一行说明这部分分析没有执行。 + +每条规则都刻意只说单次响应能证明的事情，因为错误的提示比没有提示代价更大： + +- `restoreCount` 统计的是自提交以来的重启次数，所以它的含义完全取决于 pipeline + 当前的状态。`FINISHED` / `CANCELED` 的 pipeline 已经结束，不会输出任何内容； + `FAILED` 的只在重启预算用尽时以过去时报告；正在关停的则把重启次数作为历史 + 记录输出。只有当 pipeline 处于两次尝试之间、或距上次重启运行不足 10 分钟时， + 才会判定为重启循环——因此每次尝试都能存活超过 10 分钟的循环会被当作历史 + 记录，而不是当前故障。 +- 行数规则要等作业进入 RUNNING 满两分钟后才生效；对 `job.mode = STREAMING` + 的作业，源端空闲只作为提示而非告警——安静的 topic 和配错的 topic 在这里 + 读数完全相同。 +- `DOING_SAVEPOINT` 只报告持续时长，不会说成任务挂死：状态大时写 savepoint + 本来就要几分钟。 +- checkpoint 失败只有在**最近一次失败比最近一次成功更新**时才告警。 + `counts.failed` 是自提交以来的累计值，一次早已过期的 checkpoint 会让它永远 + 大于零。 +- checkpoint 卡住的判定阈值是进行中超过 2 分钟，高于打包配置里的 + `checkpoint.timeout`（60 秒）。默认配置下挂死的 checkpoint 会先超时、以失败 + 的形式报出，因此这条规则主要服务于调高了该超时的集群。 +- 时长以引擎自己的时钟为基准计算（作业规则用 `diagnostics.generatedAt`， + checkpoint 规则用 `updatedAt`），客户端时钟与集群不一致时不会凭空造出一个 + "卡住"的作业。 + ### CLI 参数 ``` @@ -257,6 +330,7 @@ seatunnel [request] [options] --provider PROVIDER LLM 提供商：bedrock | bedrock-mantle | anthropic | openai | orcarouter --model MODEL 覆盖主模型 ID --fast-model MODEL 覆盖快速模型 ID + --diagnose JOB_ID 诊断作业当前状态后退出（只读，不调用 LLM） --sync-catalog PATH 从 SeaTunnel 源码重新生成连接器目录 -V, --version 显示版本 -h, --help 显示帮助信息 diff --git a/seatunnel-cli/seatunnel_cli/cli.py b/seatunnel-cli/seatunnel_cli/cli.py index b7e4512b0e..af1ec00cb9 100644 --- a/seatunnel-cli/seatunnel_cli/cli.py +++ b/seatunnel-cli/seatunnel_cli/cli.py @@ -28,6 +28,7 @@ from pathlib import Path from rich.console import Console from rich.live import Live from rich.markdown import Markdown +from rich.markup import escape from rich.panel import Panel from rich.spinner import Spinner from rich.syntax import Syntax @@ -43,6 +44,8 @@ from .diagnostics import ParsedError, parse_error from .llm_provider import create_provider, format_llm_error from .agents import Orchestrator +logger = logging.getLogger(__name__) + # ─── Theme ─── @@ -1573,6 +1576,206 @@ class SeaTunnelCLI: self.console.print(f" [error]Failed to execute: {e}[/error]") +_SEVERITY_STYLE = {"error": "error", "warning": "warning", "info": "info"} + +# The engine parses a job id as a decimal long, so anything else can only be a +# typo or a mangled copy-paste. Catching it here turns "1/2" -- which would +# otherwise be sent as the URL path `/job-info/1/2` and come back as a 404 -- +# into a message that names the actual problem. +_JOB_ID_RE = re.compile(r"-?\d+$") + +# An engine error body is bounded only by rest.MAX_RESPONSE_BYTES (8 MiB). +# That is the right limit for a response being parsed; it is the wrong amount +# of text to print into someone's terminal, so the printed form is clipped. +_ERROR_BODY_MAX_CHARS = 2000 + + +def _clip(text: str, limit: int = _ERROR_BODY_MAX_CHARS) -> str: + """Shorten engine-supplied text for display, saying so when it is cut.""" + return text if len(text) <= limit else f"{text[:limit]}... [{len(text)} chars total]" + + +def run_diagnose(job_id: str, console: Console) -> int: + """Explain one job's current state from ``/job-info``. Returns an exit code. + + Deliberately model-free. Every line printed here is derived from a field + the engine publishes, so this works with no API key, no provider configured + and no network beyond the cluster -- which is exactly the situation someone + is in when a job is stuck in production and they want an answer now. + + It is also read-only: two GETs, no submit, no stop, no savepoint. That is + what makes it safe to point at a live production job. + """ + from .connectors import _ENGINE_API_BASE + from .diagnostics.checkpoints import diagnose_checkpoints + from .diagnostics.jobs import as_int, diagnose_job, is_non_terminal, sort_findings + from .memory import redact_credentials + + if not _JOB_ID_RE.match(job_id): + console.print( + f"[error]{escape(job_id)} is not a job id.[/error] A SeaTunnel job id is a " + f"decimal number, as printed on submit and listed by /running-jobs." + ) + return 1 + + url = f"{_ENGINE_API_BASE}/job-info/{job_id}" + try: + job_info = rest.request_json(url, timeout=10) + except rest.RestError as e: + # A 404 here is the single most likely mistake (wrong id, a job old + # enough to have been evicted from the finished-job history, or the + # base URL pointing at a port that does not serve the v2 REST API), so + # it is worth separating from a cluster that is not answering at all. + if e.status == 404: + console.print( + f"[error]No job {escape(job_id)} at {escape(_ENGINE_API_BASE)}.[/error] It may " + f"have been evicted from the finished-job history, or the id may be wrong. " + f"If the cluster is elsewhere, set SEATUNNEL_API_BASE to its REST address " + f"(the Jetty HTTP port, 8080 in the packaged config -- not the member port 5801)." + ) + else: + # Redacted and capped for the same reasons the finding lines are: + # the engine body can quote a JDBC URL with a password, and + # rest.py only bounds it at MAX_RESPONSE_BYTES (8 MiB), which is + # far more than anyone wants pasted into a terminal. + console.print( + f"[error]Could not read job {escape(job_id)} ({e.status}):[/error] " + f"{escape(_clip(redact_credentials(str(e.body))))}" + ) + return 1 + except Exception as e: + console.print( + f"[error]Could not reach the engine at {escape(_ENGINE_API_BASE)}:[/error] " + f"{escape(str(e))}\n" + f"Set SEATUNNEL_API_BASE if the REST API is on another host or port." + ) + return 1 + + # The engine does not 404 for an id it does not know: JobInfoService falls + # through to `{"jobId": "<id>"}` with status 200 when the job is in + # neither the running-job map nor the finished-job state. Without this + # check that response reaches the rules as a job with no status, nothing + # matches, and the output reads "Nothing to report" -- i.e. "your job is + # fine" about a job that does not exist. A non-object body (a bare `null` + # or a list from a proxy) is the same situation and is caught here too, + # before anything calls .get() on it. + status = job_info.get("jobStatus") if isinstance(job_info, dict) else None + if not isinstance(status, str) or not status.strip(): + console.print( + f"[error]No job {escape(job_id)} on the cluster at " + f"{escape(_ENGINE_API_BASE)}.[/error] The engine returned no status for it, " + f"which means the id is unknown there: either it is wrong, or the job has " + f"been evicted from the finished-job history." + ) + return 1 + status = status.strip() + + name = job_info.get("jobName") or "(unnamed)" + # escape(): the job name is chosen by whoever submitted the job, so a name + # like "[prod] sync" would otherwise be read as a style tag and vanish. + console.print( + f"Job [bold]{escape(job_id)}[/bold] {escape(str(name))} " + f"— [bold]{escape(status)}[/bold]" + ) + + # The `diagnostics` block carries the restore counters and state + # timestamps, so the crash-loop and stuck-state rules are built entirely + # on it. When it is missing those rules have no input, and saying nothing + # would present "no data" as "no problem". + # + # Only while the job can still move, though: the engine builds the block + # for a job the master coordinates, so for a finished or cancelled job its + # absence is expected and the notice would be noise. + diagnostics_present = isinstance(job_info.get("diagnostics"), dict) + live_job = is_non_terminal(status) + # `generatedAt` is stamped by the master when it builds the response, so + # it is a usable "now" for both rule sets. None means fall back to the + # local clock inside each. + engine_now_ms = None + if diagnostics_present: + generated_at = as_int(job_info["diagnostics"].get("generatedAt")) + engine_now_ms = generated_at if generated_at and generated_at > 0 else None + if live_job and not diagnostics_present: + console.print( + " Restart counts and state ages were not available in this response, " + "so the crash-loop and stuck-state checks did not run (the master may " + "not be serving diagnostics for this job).", + style="info", + ) + + findings = diagnose_job(job_info) + + # Checkpoint history answers "why did it get slower", which job-info alone + # cannot. It is fetched best-effort on purpose: an older engine has no + # such endpoint and a cluster with the monitor service off returns nothing + # useful, and in neither case should a working diagnosis turn into a + # failure. Whatever job-info gave us is still worth printing. + try: + checkpoints = rest.request_json( + f"{_ENGINE_API_BASE}/jobs/checkpoints/{job_id}", timeout=10 + ) + except Exception as e: + checkpoints = None + logger.debug("Checkpoint overview for job %s unavailable: %s", job_id, e) + + # A 200 is not the same as data. When the monitor service is off, or has + # nothing recorded for the job yet, CheckpointMonitorRestService answers + # 200 with only {"jobId": ...} and no `pipelines` -- no exception to catch. + # Both that and a failed request leave the slowdown rules with no input, + # and staying silent would let the output end on "Nothing to report", + # which reads as "the job is fine". + checkpoints_usable = isinstance(checkpoints, dict) and isinstance( + checkpoints.get("pipelines"), list + ) + if live_job and not checkpoints_usable: + console.print( + " No checkpoint history for this job, so the slowdown checks did not run " + "(the cluster may predate the endpoint, its monitor service may be off, or " + "the job may not have checkpointed yet).", + style="info", + ) + if checkpoints_usable: + # The engine's own clock, from the /job-info response fetched a moment + # ago: diagnostics.generatedAt is stamped at response time, so the + # checkpoint ages are computed in the same clock as the timestamps + # they are subtracted from. + findings = sort_findings( + findings + diagnose_checkpoints(checkpoints, now_ms=engine_now_ms) + ) + + if not findings: + # Only claim a clean bill of health when the rules actually had their + # input; otherwise the notices above are the whole story. + inputs_complete = not live_job or (diagnostics_present and checkpoints_usable) + if inputs_complete: + console.print(" Nothing to report: no rule matched this job's state.", style="info") + return 0 + + for finding in findings: + # The error message can carry a JDBC URL with a password, and a + # finding quotes it, so redact on the way out even though nothing is + # being sent anywhere -- terminal output gets pasted into issues. + # + # markup=False because the finding text is data, not markup. Three + # reasons it matters: `render()` prefixes "[error]", and "error" is a + # style name in THEME, so rich would consume the prefix as a style tag + # and the severity would reach the user by colour alone -- invisible in + # a pipe, in CI, or under NO_COLOR. An engine message containing + # "ErrorDescription:[master is down]" would lose the bracketed part. + # And a fragment that looks like a closing tag, such as a path in + # "[/data/x.csv]", raises MarkupError -- crashing the one command + # someone runs when a job is already broken. + console.print( + f" {redact_credentials(finding.render())}", + style=_SEVERITY_STYLE.get(finding.severity, "info"), + markup=False, + ) + # Only a terminal failure is an error for the shell. A stuck or + # crash-looping job is a live job, and reporting it as a command failure + # would break `seatunnel --diagnose x && ...` for a job that is merely slow. + return 1 if status == "FAILED" else 0 + + def main(): """Entry point for seatunnel CLI.""" parser = argparse.ArgumentParser( @@ -1611,6 +1814,13 @@ def main(): action="store_true", help="Interactive first-time setup: choose LLM provider and save config", ) + parser.add_argument( + "--diagnose", + metavar="JOB_ID", + help="Explain the current state of a job on the cluster and exit. " + "Read-only and model-free: needs no API key, only a reachable " + "engine REST API (see SEATUNNEL_API_BASE).", + ) parser.add_argument( "--export-metadata", nargs="?", @@ -1629,6 +1839,13 @@ def main(): SeaTunnelCLI.run_init(console) sys.exit(0) + # --diagnose: read-only cluster query. Handled before any provider setup + # below, because it must work for a user who has no LLM configured at all. + if args.diagnose: + console = Console(theme=THEME) + _install_secret_log_filter() + sys.exit(run_diagnose(args.diagnose, console)) + # --export-metadata: run Java exporter, no LLM needed if args.export_metadata: from .connectors import export_runtime_metadata diff --git a/seatunnel-cli/seatunnel_cli/diagnostics/checkpoints.py b/seatunnel-cli/seatunnel_cli/diagnostics/checkpoints.py new file mode 100644 index 0000000000..45297adb3b --- /dev/null +++ b/seatunnel-cli/seatunnel_cli/diagnostics/checkpoints.py @@ -0,0 +1,468 @@ +# +# 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. +# + +"""Turn one ``/jobs/checkpoints/{jobId}`` response into findings. + +The rules in ``jobs.py`` all compare against zero or against a fixed +threshold, which answers "did it break" and "is it stuck" but not the question +people actually ask about a long-running CDC job: it was fine at first and now +it lags. The counters keep moving there, just more slowly, so nothing those +rules look at ever trips. + +This endpoint closes that gap because it is the one place the engine publishes +a *series* rather than a current value: up to 32 history entries per pipeline, +each with how long that checkpoint took and how big its state was. Comparing a +job against its own recent past then becomes arithmetic. + +Two independent signals are worth separating, because they fail to answer +different questions: + +* a **trend** -- recent checkpoints are slower than older ones. Needs a + baseline, so it only sees degradation still visible inside the retained + window (32 checkpoints; at a 10s interval that is roughly five minutes, not + three days). +* a **ratio** -- checkpoints take about as long as the gap between them, so the + job is checkpoint-bound right now. Needs no baseline at all, which is what + makes it useful for the job that got slower yesterday and then stayed slow. + +Pure functions over dicts: no I/O, no network, no model calls. Unknown or +missing input yields fewer findings, never a wrong one. +""" + +from __future__ import annotations + +import time + +from seatunnel_cli.diagnostics.jobs import Finding, as_int, sort_findings + +# A checkpoint in progress for longer than this is worth mentioning. The +# endpoint does not publish the configured interval, so there is no way to +# scale this to the job; the value is deliberately well above any healthy +# checkpoint so that a slow-but-working job is not reported as stalled. +# Note how this sits against the engine's own timeout: the packaged +# config/seatunnel.yaml sets `checkpoint.timeout: 60000`, so on default +# settings a hung checkpoint expires after 60s and surfaces through +# `checkpoint_failing` instead. This rule therefore earns its keep on +# clusters that have raised `checkpoint.timeout` above 120s, where a barrier +# can hang for minutes without the engine calling it failed. +STALLED_CHECKPOINT_SECONDS = 120 + +# "Triggered but never completed" is the normal reading for the first interval +# of every healthy job, so it takes a second round before it means anything. +MIN_TRIGGERS_NEVER_COMPLETED = 2 + +# Halving the history to compare medians needs enough samples that one slow +# outlier cannot carry the result. Six is the smallest split where each side +# has a real median rather than a single value or a pair. +MIN_TREND_SAMPLES = 6 + +# Doubling is a large enough step that it is not measurement noise, and it is +# the point where a user would describe the job as having got slower. +GROWTH_FACTOR = 2.0 + +# Below this, a doubling is arithmetically true and operationally irrelevant: +# 8ms to 20ms is not why a CDC pipeline is lagging, and reporting it would +# train people to ignore the rule. +DURATION_FLOOR_MS = 5_000 + +# Same reasoning for state: a few KiB doubling says nothing. +STATE_FLOOR_BYTES = 1 << 20 + +# At this fraction of the trigger interval the job is spending most of its +# time checkpointing, so the next checkpoint starts as the previous one ends +# and there is little room left for actual throughput. +INTERVAL_SATURATION = 0.8 + +_COMPLETED = "COMPLETED" + + +def _ms(millis: float) -> str: + """Format a millisecond duration the way an operator would say it.""" + millis = int(millis) + if millis < 1000: + return f"{millis}ms" + seconds = millis / 1000.0 + if seconds < 60: + return f"{seconds:.1f}s" + seconds = int(seconds) + if seconds < 3600: + return f"{seconds // 60}m{seconds % 60:02d}s" + return f"{seconds // 3600}h{(seconds % 3600) // 60:02d}m" + + +def _bytes(count: float) -> str: + count = float(count) + if count < 1024: + return f"{count:.0f}B" + for unit in ("KiB", "MiB"): + count /= 1024 + if count < 1024: + return f"{count:.1f}{unit}" + return f"{count / 1024:.1f}GiB" + + +def _median(values: list[float]) -> float: + ordered = sorted(values) + middle = len(ordered) // 2 + if len(ordered) % 2: + return float(ordered[middle]) + return (ordered[middle - 1] + ordered[middle]) / 2.0 + + +def _checkpoints(pipeline: dict) -> list[dict]: + """The ``checkpoint`` objects of one pipeline's history, newest first. + + Order matters and is not obvious: the engine prepends each new entry + (``PipelineCheckpointOverview.addHistory`` uses ``addFirst``) and the + history endpoint sorts by ``triggerTimestamp`` descending. A trend read in + array order without accounting for that reports the opposite conclusion, + so it is stated once here and relied on by every caller below. The test + suite pins it from the other direction too, with a history that is getting + faster rather than slower. + """ + history = pipeline.get("history") + if not isinstance(history, list): + return [] + out = [] + for entry in history: + if not isinstance(entry, dict): + continue + checkpoint = entry.get("checkpoint") + if isinstance(checkpoint, dict): + out.append(checkpoint) + return out + + +def _completed_only(checkpoints: list[dict]) -> list[dict]: + """Drop everything that is not a completed checkpoint. + + Failed and canceled checkpoints are in the same history array, and both + carry misleading values rather than absent ones: ``durationMillis`` and + ``completedTimestamp`` are omitted from the JSON when null, while + ``stateSize`` is a primitive ``long`` and so arrives as a truthful-looking + ``0``. Left in, they would drag a duration trend toward nothing and make a + growing state look like a shrinking one. + """ + return [c for c in checkpoints if c.get("status") == _COMPLETED] + + +def _trend(values: list[float]) -> tuple[float, float] | None: + """Medians of (recent half, older half), or None without enough samples. + + ``values`` is newest-first, so the recent half is the front of the list. + """ + if len(values) < MIN_TREND_SAMPLES: + return None + half = len(values) // 2 + return _median(values[:half]), _median(values[half:]) + + +def _recent_median(values: list[float]) -> float: + """Median of the recent half, or of everything when samples are few. + + Taking the median across the whole window would answer "what was this job + like on average recently", which is not the question. A job that was + healthy for most of the retained history and is checkpoint-bound now gets + its current state diluted by its own healthy past, and the rule stays + silent exactly when it is needed. + """ + if len(values) >= MIN_TREND_SAMPLES: + return _median(values[: len(values) // 2]) + return _median(values) + + +def _duration_findings(pipeline_id, completed: list[dict]) -> list[Finding]: + durations = [d for d in (as_int(c.get("durationMillis")) for c in completed) if d is not None] + trend = _trend([float(d) for d in durations]) + if trend is None: + return [] + recent, older = trend + if recent < DURATION_FLOOR_MS or older <= 0 or recent < older * GROWTH_FACTOR: + return [] + return [ + Finding( + "warning", + "checkpoint_duration_growing", + f"Pipeline {pipeline_id} checkpoints are getting slower: recently " + f"{_ms(recent)} against {_ms(older)} earlier in the retained history. " + f"A checkpoint that takes longer every round holds up the source, so " + f"throughput falls even though the job stays RUNNING.", + f"median durationMillis {int(older)} -> {int(recent)} " + f"over {len(durations)} checkpoints", + ) + ] + + +def _state_findings(pipeline_id, completed: list[dict]) -> list[Finding]: + sizes = [s for s in (as_int(c.get("stateSize")) for c in completed) if s is not None] + trend = _trend([float(s) for s in sizes]) + if trend is None: + return [] + recent, older = trend + if recent < STATE_FLOOR_BYTES or older <= 0 or recent < older * GROWTH_FACTOR: + return [] + return [ + Finding( + "info", + "checkpoint_state_growing", + f"Pipeline {pipeline_id} checkpoint state is growing: recently " + f"{_bytes(recent)} against {_bytes(older)} earlier. This is usually what " + f"is behind checkpoints taking longer.", + f"median stateSize {int(older)} -> {int(recent)} bytes", + ) + ] + + +def _interval_findings( + pipeline_id, checkpoints: list[dict], completed: list[dict] +) -> list[Finding]: + """Compare how long checkpoints take against how often they are triggered. + + Unlike the trend rules this needs no baseline, so it still fires on a job + that degraded before the retained window began and has been slow ever + since -- which is the common case by the time someone asks for help. + + Intervals come from every history entry, not just the completed ones: a + checkpoint that failed was still triggered on schedule, and dropping it + would overstate the gap between triggers. + """ + triggers = [t for t in (as_int(c.get("triggerTimestamp")) for c in checkpoints) if t] + durations = [d for d in (as_int(c.get("durationMillis")) for c in completed) if d is not None] + if len(triggers) < 3 or len(durations) < 3: + return [] + # Newest first, so the earlier element is the later timestamp. + gaps = [ + float(triggers[i] - triggers[i + 1]) + for i in range(len(triggers) - 1) + if triggers[i] > triggers[i + 1] + ] + if len(gaps) < 2: + return [] + interval = _recent_median(gaps) + duration = _recent_median([float(d) for d in durations]) + if interval <= 0 or duration < interval * INTERVAL_SATURATION: + return [] + return [ + Finding( + "warning", + "checkpoint_bound", + f"Pipeline {pipeline_id} now spends almost all its time checkpointing: " + f"a checkpoint takes {_ms(duration)} and one is triggered every " + f"{_ms(interval)}. Raise checkpoint.interval, or find why the snapshot " + f"is slow -- until then the job cannot go faster than its checkpoints.", + f"recent median duration {int(duration)}ms vs interval {int(interval)}ms", + ) + ] + + +def _triggered_at(pipeline: dict, key: str) -> int | None: + """``triggerTimestamp`` of ``latestCompleted`` / ``latestFailed``, if usable.""" + latest = pipeline.get(key) + if not isinstance(latest, dict): + return None + stamp = as_int(latest.get("triggerTimestamp")) + return stamp if stamp and stamp > 0 else None + + +def _oldest_in_progress_age_ms(pipeline: dict, now_ms: int) -> int | None: + in_progress = pipeline.get("inProgress") + if not isinstance(in_progress, list): + return None + ages = [ + now_ms - stamp + for stamp in ( + as_int(c.get("triggerTimestamp")) for c in in_progress if isinstance(c, dict) + ) + if stamp and stamp > 0 + ] + return max(ages) if ages else None + + +def _counts_findings(pipeline_id, pipeline: dict, now_ms: int) -> list[Finding]: + """Rules over ``counts``, which are lifetime totals rather than a state. + + Every value here counts since the job was submitted, so neither "failed" + nor "completed none" says anything about the present on its own. Both + rules therefore need a second piece of evidence -- the relative age of the + latest failure and completion, or how many rounds have been triggered -- + before they are allowed to describe the job as it is now. + """ + counts = pipeline.get("counts") + if not isinstance(counts, dict): + return [] + triggered = as_int(counts.get("triggered")) + completed = as_int(counts.get("completed")) + failed = as_int(counts.get("failed")) + + findings = [] + # `completed == 0`, not `not completed`: an absent or unparsable count is + # unknown, and reporting "completed none" from it would state as measured + # something the engine never said. + if triggered and completed == 0: + # A job that has just started legitimately reads triggered=1, + # completed=0 with that first checkpoint still in flight. Saying + # "nothing has been committed" there describes every healthy job in + # its first interval, so it needs either a second round or a first + # round that has already been running too long. + stalled_ms = _oldest_in_progress_age_ms(pipeline, now_ms) + rounds_enough = triggered >= MIN_TRIGGERS_NEVER_COMPLETED + first_round_overdue = stalled_ms is not None and ( + stalled_ms >= STALLED_CHECKPOINT_SECONDS * 1000 + ) + if rounds_enough or first_round_overdue: + evidence = f"triggered={triggered} completed=0" + if stalled_ms is not None: + evidence += f" oldest in progress for {_ms(stalled_ms)}" + findings.append( + Finding( + "warning", + "checkpoint_never_completed", + f"Pipeline {pipeline_id} has triggered {triggered} checkpoints and " + f"completed none. Nothing has been committed, so a restart replays " + f"from the beginning rather than resuming.", + evidence, + ) + ) + if failed: + reason = pipeline.get("latestFailed") + reason = reason.get("failureReason") if isinstance(reason, dict) else None + failed_at = _triggered_at(pipeline, "latestFailed") + completed_at = _triggered_at(pipeline, "latestCompleted") + evidence = f"failed={failed}" + if completed is not None: + evidence += f" completed={completed}" + + # The deciding question is not "has anything ever failed" but "did the + # last attempt fail". A pipeline with one expired checkpoint weeks ago + # and thousands of completions since keeps `failed` above zero for the + # rest of its life, and a warning claiming it makes "no durable + # progress" would be wrong for as long as it runs. + failing_now = failed_at is not None and (completed_at is None or failed_at > completed_at) + if failing_now: + detail = ( + f"Pipeline {pipeline_id} has {failed} failed checkpoints and its latest " + f"checkpoint failed rather than completed. Offsets stop advancing on " + f"every failure, so the job can appear to be reading while making no " + f"durable progress." + ) + if isinstance(reason, str) and reason.strip(): + detail += f" Last failure: {reason.strip()}" + findings.append(Finding("warning", "checkpoint_failing", detail, evidence)) + elif completed_at is not None: + detail = ( + f"Pipeline {pipeline_id} has {failed} failed checkpoints since it was " + f"submitted, but the most recent checkpoint completed, so it is " + f"committing progress now." + ) + if isinstance(reason, str) and reason.strip(): + detail += f" Last failure was: {reason.strip()}" + findings.append(Finding("info", "checkpoint_failed_before", detail, evidence)) + else: + # Counter present, no usable timestamps: report the count and the + # reason, but not a conclusion about now. + detail = ( + f"Pipeline {pipeline_id} has {failed} failed checkpoints since it was " + f"submitted. Whether the latest one failed is not visible in this " + f"response." + ) + if isinstance(reason, str) and reason.strip(): + detail += f" Last failure was: {reason.strip()}" + findings.append(Finding("info", "checkpoint_failed_before", detail, evidence)) + return findings + + +def _in_progress_findings(pipeline_id, pipeline: dict, now_ms: int) -> list[Finding]: + in_progress = pipeline.get("inProgress") + if not isinstance(in_progress, list): + return [] + findings = [] + for checkpoint in in_progress: + if not isinstance(checkpoint, dict): + continue + triggered_at = as_int(checkpoint.get("triggerTimestamp")) + if not triggered_at or triggered_at <= 0: + continue + age_ms = now_ms - triggered_at + if age_ms < STALLED_CHECKPOINT_SECONDS * 1000: + continue + acknowledged = as_int(checkpoint.get("acknowledged")) + total = as_int(checkpoint.get("total")) + checkpoint_id = checkpoint.get("checkpointId", "?") + detail = ( + f"Pipeline {pipeline_id} checkpoint {checkpoint_id} has been in progress " + f"for {_ms(age_ms)} without completing." + ) + evidence = f"triggerTimestamp={triggered_at}" + # Naming the subtasks that have not acknowledged turns "the job is slow" + # into "this stage is slow", which is the difference between a usable + # diagnosis and a restatement of the symptom. + if acknowledged is not None and total: + detail += ( + f" {acknowledged} of {total} subtasks have acknowledged it, so the " + f"remaining {total - acknowledged} are what the barrier is waiting on." + ) + evidence += f" acknowledged={acknowledged}/{total}" + findings.append(Finding("error", "checkpoint_stalled", detail, evidence)) + return findings + + +def diagnose_checkpoints(overview: dict, *, now_ms: int | None = None) -> list[Finding]: + """Apply every checkpoint rule to one overview response, worst finding first. + + The response carries only ``jobId`` when the monitor service is not + running or has nothing recorded for the job, which is a legitimate state + rather than an error: it yields no findings. + + ``now_ms`` should be the engine's own "now" where the caller has one -- + ``diagnostics.generatedAt`` from the ``/job-info`` response fetched moments + earlier, which the master stamps at response time. The local clock is the + fallback. + + Deliberately *not* the overview's own ``updatedAt``, which looks like the + right field and is not: the engine writes it inside + ``HazelcastCheckpointOverviewStateStore.updateOverview``, i.e. only when a + checkpoint event is recorded. For a checkpoint stuck waiting on its last + subtask the final write is that last acknowledgement, and nothing is + written afterwards, so ``updatedAt - triggerTimestamp`` freezes a few + seconds after the trigger and never reaches the stalled threshold. Using + it would make ``checkpoint_stalled`` unable to fire in the one case it + exists for. + """ + if not isinstance(overview, dict): + return [] + + if now_ms is None: + now_ms = int(time.time() * 1000) + + pipelines = overview.get("pipelines") + if not isinstance(pipelines, list): + return [] + + findings: list[Finding] = [] + for pipeline in pipelines: + if not isinstance(pipeline, dict): + continue + pipeline_id = pipeline.get("pipelineId", "?") + checkpoints = _checkpoints(pipeline) + completed = _completed_only(checkpoints) + + findings += _in_progress_findings(pipeline_id, pipeline, now_ms) + findings += _counts_findings(pipeline_id, pipeline, now_ms) + findings += _duration_findings(pipeline_id, completed) + findings += _interval_findings(pipeline_id, checkpoints, completed) + findings += _state_findings(pipeline_id, completed) + return sort_findings(findings) diff --git a/seatunnel-cli/seatunnel_cli/diagnostics/jobs.py b/seatunnel-cli/seatunnel_cli/diagnostics/jobs.py new file mode 100644 index 0000000000..1f47e6d7ed --- /dev/null +++ b/seatunnel-cli/seatunnel_cli/diagnostics/jobs.py @@ -0,0 +1,525 @@ +# +# 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. +# + +"""Turn one ``/job-info`` response into findings, without calling a model. + +Every rule here reads a field the engine already publishes. The ``diagnostics`` +block in particular is a dedicated structure -- per-state entry timestamps at +both job and pipeline level, plus the pipeline restore counter -- so the two +questions users actually ask about a non-finishing job can be answered by +arithmetic: + +* "it says RUNNING but nothing is moving" -> a pipeline is restarting in a + loop, visible as ``restoreCount`` climbing toward ``maxRestoreCount`` +* "it has been stuck for ages" -> how long ago the job entered its current + state, from ``stateTimestamps`` + +Both were previously answerable only by reading server logs. The point of +doing this with rules rather than a model is that these answers are derived, +not guessed: a restore count of 7 is a crash loop whatever a model thinks. + +Pure functions over dicts: no I/O, no network, no model calls. Unknown or +missing input yields fewer findings, never a wrong one -- the caller prints +what is here and says nothing about what is not. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass + +from seatunnel_cli.diagnostics.errors import parse_error + +# How long a job may sit in a non-terminal state before it is worth +# mentioning. Submitting, scheduling and deploying a pipeline legitimately +# takes a few seconds on a cold cluster, and reporting that as "stuck" would +# make the normal case look broken, so the threshold is deliberately well +# above any healthy startup. +STUCK_SECONDS = 120 + +# Restarting once is normal operation -- a transient connection drop costs one +# restore and the job carries on. The signal is repetition, so the crash-loop +# rule only fires from the second restore. +CRASH_LOOP_RESTORES = 2 + +# ``restoreCount`` counts restores since submission, so on its own it says +# nothing about *now*: a streaming job that recovered twice from transient +# errors weeks ago carries the same count as one restarting every minute. The +# distinguishing evidence is how long the pipeline has been in its current +# state -- the engine overwrites the pipeline state timestamps on each +# restore, so a recent entry into RUNNING means the latest attempt is young. +# Ten minutes is comfortably longer than the restart interval of a real loop +# (which is bounded by deploy time, seconds) and comfortably shorter than an +# uptime anyone would call recovered. +CRASH_LOOP_RECENT_SECONDS = 600 + +# The PipelineStatus values, grouped by what a restore count means in each. +# +# This matters because a restorable pipeline is only in an end state for an +# instant: `SubPlan.prepareRestorePipeline` increments the restore counter and +# then immediately `reset()`s the pipeline back toward CREATED, before the +# restore interval elapses. So a pipeline that a response shows *in* an end +# state is not mid-restart -- it is done, and saying "it is restarting again" +# would be false. `PipelineStatus.isEndState()` covers FINISHED, CANCELED and +# FAILED; FAILED is separated here because it is worth a past-tense note when +# the restore budget was spent, while the other two are worth nothing at all. +_PIPELINE_ENDED = {"FINISHED", "CANCELED"} +_PIPELINE_FAILED = {"FAILED"} +_PIPELINE_TEARING_DOWN = {"CANCELING", "FAILING"} +_PIPELINE_RESTARTING = {"CREATED", "SCHEDULED", "DEPLOYING", "INITIALIZING"} + +# The JobStatus values declared EndState.NOT_END, split by what a long stay +# in them actually means. Terminal states are absent on purpose: there, a +# duration is just how long ago the job ended and says nothing about health. +# +# Before running, the usual cause is the cluster not granting the slots the +# job asked for. While tearing down, it is a task that will not stop. A +# savepoint is a third case: it legitimately takes as long as the state is +# large. Those need different wording, because sending someone to check +# cluster resources over a slow savepoint wastes their time. +_PRE_RUNNING = {"INITIALIZING", "CREATED", "PENDING", "SCHEDULED"} +_TEARING_DOWN = {"FAILING", "CANCELING"} +_SAVEPOINT = {"DOING_SAVEPOINT"} +_NON_TERMINAL = _PRE_RUNNING | _TEARING_DOWN | _SAVEPOINT | {"RUNNING"} + +_SEVERITY_ORDER = {"error": 0, "warning": 1, "info": 2} + + +@dataclass(frozen=True) +class Finding: + """One observation about a job. + + ``evidence`` is the field values the rule fired on, kept separate from + ``detail`` so a reader can check the conclusion against the numbers rather + than having to trust it. + """ + + severity: str + rule: str + detail: str + evidence: str | None = None + + def render(self) -> str: + line = f"[{self.severity}] {self.detail}" + return f"{line} ({self.evidence})" if self.evidence else line + + +def is_non_terminal(status) -> bool: + """Whether a job status is one the engine can still move out of. + + Exposed because the caller needs the same notion to decide whether a + missing ``diagnostics`` block is a gap worth reporting: the engine builds + that block only for a job the master still coordinates, so for a finished + or cancelled job its absence is by design rather than lost coverage. + """ + return isinstance(status, str) and status.strip().upper() in _NON_TERMINAL + + +def sort_findings(findings: list[Finding]) -> list[Finding]: + """Worst severity first, stable so equal severities keep insertion order. + + Shared rather than kept per module because the caller prints findings from + several rule sets as a single list: sorting each set on its own would put + an informational note from one above an error from another. + """ + return sorted(findings, key=lambda f: _SEVERITY_ORDER.get(f.severity, 9)) + + +def as_int(value) -> int | None: + """Engine JSON sends counters as ints, but metrics arrive as strings.""" + if isinstance(value, bool) or value is None: + return None + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _state_entered_at(timestamps: dict, status: str) -> int | None: + """Epoch millis of when the job entered ``status``, or None if unusable. + + The engine records the *last* entry into each state, keyed by state name, + so the entry for the current status is the right answer even for a job + that was restored and re-entered an earlier state. The map is unordered, + so the fallback when the current status has no entry of its own is the + largest timestamp present -- never the last key, which would be an + arbitrary and possibly very stale state. + """ + if not isinstance(timestamps, dict): + return None + own = as_int(timestamps.get(status)) + if own and own > 0: + return own + values = [v for v in (as_int(v) for v in timestamps.values()) if v and v > 0] + return max(values) if values else None + + +def _humanize(seconds: float) -> str: + seconds = int(seconds) + if seconds < 60: + return f"{seconds}s" + if seconds < 3600: + return f"{seconds // 60}m{seconds % 60:02d}s" + return f"{seconds // 3600}h{(seconds % 3600) // 60:02d}m" + + +def _failure_findings(job_info: dict) -> list[Finding]: + error_msg = job_info.get("errorMsg") + if not isinstance(error_msg, str) or not error_msg.strip(): + return [ + Finding( + "error", + "failed_without_error", + "The job failed but reported no error message; the cause is only in the server log.", + ) + ] + parsed = parse_error(error_msg) + findings = [Finding("error", "job_failed", parsed.headline())] + if parsed.hint: + findings.append(Finding("info", "failure_hint", parsed.hint)) + return findings + + +def _crash_loop_findings(pipelines: list, now_ms: int) -> list[Finding]: + """Restore counters, reported as a live problem only when they are one. + + ``restoreCount`` is cumulative, so what it means depends entirely on the + pipeline's current state. The grouping above encodes the engine's own + lifecycle: a restorable pipeline occupies an end state for an instant + only, because ``SubPlan.prepareRestorePipeline`` increments the counter + and immediately ``reset()``s the pipeline back toward ``CREATED`` before + the restore interval elapses. A pipeline observed in an end state is + therefore finished, not restarting, and its restore count is history. + """ + findings = [] + for pipeline in pipelines: + if not isinstance(pipeline, dict): + continue + restores = as_int(pipeline.get("restoreCount")) + if restores is None or restores < CRASH_LOOP_RESTORES: + continue + pipeline_id = pipeline.get("pipelineId", "?") + limit = as_int(pipeline.get("maxRestoreCount")) + exhausted = limit is not None and limit > 0 and restores >= limit + + status = pipeline.get("pipelineStatus") + status = status.strip().upper() if isinstance(status, str) and status.strip() else None + + evidence = f"restoreCount={restores}" + if limit is not None and limit > 0: + evidence += f"/{limit}" + if status is not None: + evidence += f" pipelineStatus={status}" + + # Finished or cancelled: the pipeline is done. In a multi-pipeline job + # one pipeline finishing after two earlier restores is normal, and + # calling it a crash loop would be plainly wrong. + if status in _PIPELINE_ENDED: + continue + + # Failed: past tense. A failure that was not restorable, or one after + # the budget ran out, is covered by the job-level failure finding; the + # only thing worth adding is that the restores were all spent. + if status in _PIPELINE_FAILED: + if exhausted: + findings.append( + Finding( + "error", + "pipeline_restore_exhausted", + f"Pipeline {pipeline_id} used all {restores} of its restores and " + f"then failed.", + evidence, + ) + ) + continue + + # Budget gone while the pipeline is still live: the engine will not + # restore again, so the next failure is terminal. + if exhausted: + findings.append( + Finding( + "error", + "pipeline_restore_exhausted", + f"Pipeline {pipeline_id} has restarted {restores} times, which is " + f"its limit: the next failure ends the job.", + evidence, + ) + ) + continue + + # Shutting down. Whatever the restores were, they are not what is + # happening now, and the stop may well have been requested. + if status in _PIPELINE_TEARING_DOWN: + findings.append( + Finding( + "info", + "pipeline_restored_before", + f"Pipeline {pipeline_id} restarted {restores} times since it was " + f"submitted and is now {status}.", + evidence, + ) + ) + continue + + # Between attempts: not yet running, with restores behind it, which is + # what a restart in progress looks like. + if status in _PIPELINE_RESTARTING: + findings.append( + Finding( + "warning", + "pipeline_crash_loop", + f"Pipeline {pipeline_id} has restarted {restores} times and is " + f"{status} rather than RUNNING, so it is between attempts rather " + f"than processing data.", + evidence, + ) + ) + continue + + if status == "RUNNING": + attempt_started = _state_entered_at(pipeline.get("stateTimestamps", {}), "RUNNING") + if attempt_started is not None: + uptime_seconds = (now_ms - attempt_started) / 1000.0 + evidence += f" running for {_humanize(uptime_seconds)} since the last restore" + if uptime_seconds >= CRASH_LOOP_RECENT_SECONDS: + # Running steadily since the last restore: the count is a + # record of past incidents, not a description of now. + findings.append( + Finding( + "info", + "pipeline_restored_before", + f"Pipeline {pipeline_id} has restarted {restores} times since " + f"it was submitted, most recently " + f"{_humanize(uptime_seconds)} ago, and has been running since.", + evidence, + ) + ) + continue + # Restarted recently and running again. That the restores are + # still ongoing is not something one snapshot can show, so + # this reports the two facts and points at the logs that can. + findings.append( + Finding( + "warning", + "pipeline_crash_loop", + f"Pipeline {pipeline_id} has restarted {restores} times, and the " + f"current attempt is only {_humanize(uptime_seconds)} old. Check " + f"the task logs for what triggered the restores.", + evidence, + ) + ) + continue + + # No usable status, or RUNNING without a timestamp: the count is all + # there is, so report exactly that and let the reader judge. An + # unproven "crash loop" in the output is worse than a plain number. + findings.append( + Finding( + "info", + "pipeline_restored_before", + f"Pipeline {pipeline_id} has restarted {restores} times since it was " + f"submitted. How recently is not visible in this response.", + evidence, + ) + ) + return findings + + +def _stuck_findings(job_info: dict, diagnostics: dict, now_ms: int) -> list[Finding]: + status = job_info.get("jobStatus") + if not isinstance(status, str): + return [] + status = status.upper() + if status not in _NON_TERMINAL: + return [] + + entered = _state_entered_at(diagnostics.get("stateTimestamps", {}), status) + if entered is None: + return [] + held_seconds = (now_ms - entered) / 1000.0 + if held_seconds < STUCK_SECONDS: + return [] + + held = _humanize(held_seconds) + if status == "RUNNING": + # RUNNING for a long time is what a streaming job is supposed to do, + # so this is only worth a line of context, never a warning. + return [Finding("info", "running_duration", f"Running for {held}.", f"state={status}")] + if status in _PRE_RUNNING: + return [ + Finding( + "warning", + "stuck_before_running", + f"The job has been {status} for {held} without starting. That usually " + f"means the cluster could not give it the slots it asked for -- check " + f"free slots and worker count on the cluster.", + f"state={status}", + ) + ] + if status in _SAVEPOINT: + # A savepoint writes out the whole job state, so a large state takes + # minutes by design. Calling that "not responding" sends someone + # looking for a hung task when the job is doing exactly what it was + # asked to do, so this only reports the duration. + return [ + Finding( + "warning", + "savepoint_in_progress", + f"The job has been {status} for {held}. A savepoint takes as long as " + f"the job state is large, so this can be normal; if it does not finish, " + f"a task is not completing its snapshot.", + f"state={status}", + ) + ] + return [ + Finding( + "warning", + "stuck_tearing_down", + f"The job has been {status} for {held}. It is shutting down and not " + f"finishing, which means a task is not responding to the stop request.", + f"state={status}", + ) + ] + + +def _is_streaming(job_info: dict) -> bool: + """Whether the job declares ``job.mode = STREAMING`` in its env options. + + Only used to pick wording. A streaming source that has not produced a row + yet may simply have nothing to produce -- a quiet topic, a CDC job waiting + for its first change -- while a batch source that reads nothing has a + problem by definition. + """ + env = job_info.get("envOptions") + mode = env.get("job.mode") if isinstance(env, dict) else None + return isinstance(mode, str) and mode.strip().upper() == "STREAMING" + + +def _progress_findings(job_info: dict, diagnostics: dict, now_ms: int) -> list[Finding]: + """Row counts, read from the job-wide scalar metrics. + + ``SourceReceivedCount``/``SinkWriteCount`` are the cluster-wide totals. + The ``Table``-prefixed names are deliberately *not* used here: those are + objects keyed by table name, not scalars, and every metric leaf is + rendered as a string by the engine, so both the nesting and the quoting + have to be respected rather than assumed. + + Both rules are gated on how long the job has been RUNNING. A zero counter + a second after the job started running is the expected reading, not a + symptom, and reporting it would make every freshly submitted job look + broken. + """ + if job_info.get("jobStatus") != "RUNNING": + return [] + metrics = job_info.get("metrics") + if not isinstance(metrics, dict): + return [] + read = as_int(metrics.get("SourceReceivedCount")) + written = as_int(metrics.get("SinkWriteCount")) + if read is None or written is None: + return [] + + # The RUNNING entry specifically, not the newest timestamp of any state: + # without it there is no way to tell a job that has read nothing for an + # hour from one that started two seconds ago, and guessing is what these + # rules are supposed to avoid. + timestamps = diagnostics.get("stateTimestamps") + entered = as_int(timestamps.get("RUNNING")) if isinstance(timestamps, dict) else None + if entered is None or entered <= 0: + return [] + running_seconds = (now_ms - entered) / 1000.0 + if running_seconds < STUCK_SECONDS: + return [] + running_for = _humanize(running_seconds) + + if read == 0: + if _is_streaming(job_info): + return [ + Finding( + "info", + "no_rows_read", + f"The source has produced no rows in the {running_for} since the job " + f"started running. For a streaming source with nothing to read yet -- " + f"a quiet topic, a CDC job waiting for its first change -- that is " + f"normal; if you expected rows by now, check the table or path it " + f"points at and any filter on the source.", + f"SourceReceivedCount=0 running for {running_for}", + ) + ] + return [ + Finding( + "warning", + "no_rows_read", + f"The source has produced no rows in the {running_for} since the job " + f"started running. Check the table or path it points at, and any filter " + f"or where clause on the source.", + f"SourceReceivedCount=0 running for {running_for}", + ) + ] + if written == 0: + return [ + Finding( + "warning", + "rows_read_none_written", + f"The source read {read} rows but the sink wrote none, so rows are " + "being dropped between them -- usually a transform filter, or a " + "sink that has not committed its first batch yet.", + f"SourceReceivedCount={read} SinkWriteCount=0 running for {running_for}", + ) + ] + return [] + + +def diagnose_job(job_info: dict, *, now_ms: int | None = None) -> list[Finding]: + """Apply every rule to one ``/job-info`` response, worst finding first. + + ``now_ms`` is injectable because the stuck-state rule is a clock + comparison, and a test that cannot pin the clock can only assert on + thresholds it has to reach by sleeping. + + When it is not injected, ``diagnostics.generatedAt`` is preferred over the + local clock: every timestamp it is compared against was written by the + master, so using the master's own "now" keeps the arithmetic inside one + clock. A laptop a few minutes off a cluster would otherwise invent a + stuck job, or hide one. + """ + if not isinstance(job_info, dict): + return [] + + diagnostics = job_info.get("diagnostics") + diagnostics = diagnostics if isinstance(diagnostics, dict) else {} + pipelines = diagnostics.get("pipelines") + pipelines = pipelines if isinstance(pipelines, list) else [] + + if now_ms is None: + generated_at = as_int(diagnostics.get("generatedAt")) + now_ms = ( + generated_at + if generated_at is not None and generated_at > 0 + else int(time.time() * 1000) + ) + + findings: list[Finding] = [] + if job_info.get("jobStatus") == "FAILED": + findings += _failure_findings(job_info) + findings += _crash_loop_findings(pipelines, now_ms) + findings += _stuck_findings(job_info, diagnostics, now_ms) + findings += _progress_findings(job_info, diagnostics, now_ms) + + # Stable sort, so rules of equal severity keep the order above: failure + # first, then why it is not progressing, then throughput. + return sort_findings(findings) diff --git a/seatunnel-cli/tests/test_diagnose_checkpoints.py b/seatunnel-cli/tests/test_diagnose_checkpoints.py new file mode 100644 index 0000000000..2ecfc513d6 --- /dev/null +++ b/seatunnel-cli/tests/test_diagnose_checkpoints.py @@ -0,0 +1,576 @@ +# +# 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. +# + +"""Tests for the checkpoint-history rules. + +The payloads are shaped the way `/jobs/checkpoints/{jobId}` actually sends +them, and three of those details are what a diagnosis built on assumptions +would get wrong: + +* `history` is **newest first**, so a trend read in array order is inverted; +* failed and canceled checkpoints share the array but have **no** + `durationMillis` (the field is omitted when null); +* their `stateSize` is a primitive `long`, so it arrives as a plausible `0` + rather than being absent, which would make growing state look like + shrinking state. + +Each of those has a test of its own below, because all three fail silently -- +they produce a confident finding that is simply backwards. +""" + +import time + +from seatunnel_cli.diagnostics.checkpoints import ( + DURATION_FLOOR_MS, + MIN_TREND_SAMPLES, + MIN_TRIGGERS_NEVER_COMPLETED, + STALLED_CHECKPOINT_SECONDS, + diagnose_checkpoints, +) + +NOW = 1_760_000_000_000 + + +def _rules(findings): + return [f.rule for f in findings] + + +def _by_rule(findings, rule): + return next(f for f in findings if f.rule == rule) + + +def _history(durations, *, state_sizes=None, interval_ms=600_000, status="COMPLETED"): + """Build a newest-first history array. + + ``durations`` is given newest first, matching the wire format, so a test + that wants "recently slow" puts the large values at the front. + """ + entries = [] + for index, duration in enumerate(durations): + checkpoint = { + "checkpointId": 1000 - index, + "checkpointType": "CHECKPOINT_TYPE", + "status": status, + # Newest first, so each successive entry is older. + "triggerTimestamp": NOW - index * interval_ms, + "stateSize": (state_sizes[index] if state_sizes else 1024), + } + if duration is not None: + checkpoint["durationMillis"] = duration + checkpoint["completedTimestamp"] = checkpoint["triggerTimestamp"] + duration + entries.append({"pipelineId": 1, "checkpoint": checkpoint}) + return entries + + +def _overview(pipeline): + return {"jobId": "852", "updatedAt": NOW, "pipelines": [pipeline]} + + +def test_checkpoints_getting_slower_are_reported(): + findings = diagnose_checkpoints( + _overview({"pipelineId": 1, "history": _history([30_000] * 4 + [6_000] * 4)}), + now_ms=NOW, + ) + + growing = _by_rule(findings, "checkpoint_duration_growing") + assert growing.severity == "warning" + assert "30.0s" in growing.detail and "6.0s" in growing.detail + + +def test_checkpoints_getting_faster_are_not_reported_as_slower(): + # This is the ordering trap. `history` is newest first, so these durations + # describe a job that *recovered*: it used to take 30s and now takes 6s. + # Reading the array as oldest-first would report a 5x slowdown. + findings = diagnose_checkpoints( + _overview({"pipelineId": 1, "history": _history([6_000] * 4 + [30_000] * 4)}), + now_ms=NOW, + ) + + assert "checkpoint_duration_growing" not in _rules(findings) + + +def test_failed_checkpoints_do_not_poison_the_duration_trend(): + # Failed entries carry no durationMillis at all. Treating the missing + # field as a zero would halve the recent median and hide a real slowdown. + history = _history([30_000, None, 30_000, None, 30_000, 30_000, 6_000, 6_000, 6_000, 6_000]) + for entry in history: + if "durationMillis" not in entry["checkpoint"]: + entry["checkpoint"]["status"] = "FAILED" + entry["checkpoint"]["failureReason"] = "checkpoint expired" + + findings = diagnose_checkpoints(_overview({"pipelineId": 1, "history": history}), now_ms=NOW) + + assert "checkpoint_duration_growing" in _rules(findings) + + +def test_failed_checkpoints_zero_state_size_does_not_invert_the_state_trend(): + # stateSize is a primitive long on the engine side, so a failed checkpoint + # reports a convincing 0 rather than omitting the field. Counted in, those + # zeroes would drag the recent median down and turn growing state into + # apparently shrinking state. + sizes = [50 << 20] * 4 + [10 << 20] * 4 + history = _history([30_000] * 8, state_sizes=sizes) + # Two recent failures, which is exactly when this is most likely to bite. + for entry in history[:2]: + entry["checkpoint"]["status"] = "FAILED" + entry["checkpoint"]["stateSize"] = 0 + del entry["checkpoint"]["durationMillis"] + + findings = diagnose_checkpoints(_overview({"pipelineId": 1, "history": history}), now_ms=NOW) + + assert "checkpoint_state_growing" in _rules(findings) + + +def test_a_doubling_of_a_trivial_duration_is_not_reported(): + # 8ms to 20ms is a doubling and is not why anything is lagging. + findings = diagnose_checkpoints( + _overview( + {"pipelineId": 1, "history": _history([20] * 4 + [8] * 4, interval_ms=10_000)} + ), + now_ms=NOW, + ) + + assert "checkpoint_duration_growing" not in _rules(findings) + assert DURATION_FLOOR_MS > 20 + + +def test_too_few_checkpoints_yields_no_trend(): + findings = diagnose_checkpoints( + _overview( + {"pipelineId": 1, "history": _history([30_000, 30_000, 6_000, 6_000])}, + ), + now_ms=NOW, + ) + + assert MIN_TREND_SAMPLES > 4 + assert "checkpoint_duration_growing" not in _rules(findings) + + +def test_a_job_that_spends_its_time_checkpointing_is_reported_without_a_baseline(): + # Durations are flat, so the trend rule stays silent -- this is the job + # that degraded before the retained window began and has been slow since. + findings = diagnose_checkpoints( + _overview( + {"pipelineId": 1, "history": _history([9_000] * 8, interval_ms=10_000)}, + ), + now_ms=NOW, + ) + + assert "checkpoint_duration_growing" not in _rules(findings) + bound = _by_rule(findings, "checkpoint_bound") + assert "9.0s" in bound.detail and "10.0s" in bound.detail + + +def test_a_job_that_became_checkpoint_bound_recently_is_still_reported(): + # Healthy for the older half (4s of a 60s interval), checkpoint-bound in + # the recent half (48s of 60s). A median over the whole window is 26s, + # which is under the threshold, so taking it would stay silent exactly + # when the job has just gone bad. + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "history": _history([48_000] * 8 + [4_000] * 8, interval_ms=60_000), + } + ), + now_ms=NOW, + ) + + bound = _by_rule(findings, "checkpoint_bound") + assert "48.0s" in bound.detail and "1m00s" in bound.detail + + +def test_checkpoints_that_finish_well_within_the_interval_are_not_flagged(): + findings = diagnose_checkpoints( + _overview( + {"pipelineId": 1, "history": _history([1_000] * 8, interval_ms=60_000)}, + ), + now_ms=NOW, + ) + + assert findings == [] + + +def test_a_stalled_checkpoint_names_the_subtasks_holding_the_barrier(): + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 2, + "inProgress": [ + { + "checkpointId": 77, + "triggerTimestamp": NOW - (STALLED_CHECKPOINT_SECONDS + 300) * 1000, + "acknowledged": 3, + "total": 8, + } + ], + } + ), + now_ms=NOW, + ) + + stalled = _by_rule(findings, "checkpoint_stalled") + assert stalled.severity == "error" + assert "3 of 8" in stalled.detail + assert "remaining 5" in stalled.detail + assert "acknowledged=3/8" in stalled.evidence + + +def test_a_checkpoint_that_just_started_is_not_called_stalled(): + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "inProgress": [ + {"checkpointId": 78, "triggerTimestamp": NOW - 3000, "acknowledged": 1, + "total": 8}, + ], + } + ), + now_ms=NOW, + ) + + assert findings == [] + + +def test_never_having_completed_a_checkpoint_is_reported(): + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "counts": {"triggered": 12, "completed": 0, "failed": 0, "inProgress": 1}, + } + ), + now_ms=NOW, + ) + + never = _by_rule(findings, "checkpoint_never_completed") + assert never.evidence == "triggered=12 completed=0" + + +def test_a_job_in_its_first_checkpoint_interval_is_not_reported_as_never_completing(): + # triggered=1, completed=0, one in flight is what every healthy job looks + # like before its first checkpoint lands. Saying "nothing has been + # committed, a restart replays from the beginning" there describes normal + # startup as a fault. + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "counts": {"triggered": 1, "completed": 0, "failed": 0, "inProgress": 1}, + "inProgress": [{"checkpointId": 1, "triggerTimestamp": NOW - 3_000}], + } + ), + now_ms=NOW, + ) + + assert findings == [] + + +def test_a_first_checkpoint_that_is_already_overdue_is_reported(): + # One trigger is still enough once that checkpoint has been in flight + # longer than any healthy one would be. + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "counts": {"triggered": 1, "completed": 0, "failed": 0, "inProgress": 1}, + "inProgress": [ + { + "checkpointId": 1, + "triggerTimestamp": NOW - (STALLED_CHECKPOINT_SECONDS + 10) * 1000, + } + ], + } + ), + now_ms=NOW, + ) + + never = _by_rule(findings, "checkpoint_never_completed") + assert "oldest in progress for" in never.evidence + + +def test_a_second_trigger_with_nothing_completed_is_reported_without_timestamps(): + # An engine that sends counts but no inProgress detail still supports the + # "two rounds, nothing committed" reading. + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "counts": { + "triggered": MIN_TRIGGERS_NEVER_COMPLETED, + "completed": 0, + "failed": 0, + }, + } + ), + now_ms=NOW, + ) + + assert "checkpoint_never_completed" in _rules(findings) + + +def test_a_stuck_checkpoint_is_detected_even_though_updated_at_is_stale(): + # `updatedAt` looks like a response timestamp and is not one: the engine + # writes it inside HazelcastCheckpointOverviewStateStore.updateOverview, + # so only when a checkpoint event is recorded. For a checkpoint hung on + # its last subtask the newest write is that last acknowledgement, a few + # seconds after the trigger, and nothing follows. Taking it as "now" + # freezes the age there, and the stalled rule -- the whole point of which + # is a hung barrier -- could never fire. + trigger = NOW - (STALLED_CHECKPOINT_SECONDS + 30) * 1000 + overview = { + "jobId": "852", + "updatedAt": trigger + 3_000, + "pipelines": [ + { + "pipelineId": 1, + "inProgress": [ + { + "checkpointId": 7, + "triggerTimestamp": trigger, + "acknowledged": 3, + "total": 8, + } + ], + } + ], + } + + stalled = _by_rule(diagnose_checkpoints(overview, now_ms=NOW), "checkpoint_stalled") + assert "2m30s" in stalled.detail + assert "3 of 8 subtasks" in stalled.detail + + +def test_without_an_injected_clock_the_local_one_is_used_not_updated_at(): + # The fallback has to be the local clock. If `updatedAt` were consulted + # here the age would come out as 3s and the rule would stay silent. + now = int(time.time() * 1000) + trigger = now - (STALLED_CHECKPOINT_SECONDS + 60) * 1000 + overview = { + "jobId": "852", + "updatedAt": trigger + 3_000, + "pipelines": [ + {"pipelineId": 1, "inProgress": [{"checkpointId": 7, "triggerTimestamp": trigger}]} + ], + } + + assert "checkpoint_stalled" in _rules(diagnose_checkpoints(overview)) + + +def test_failed_checkpoints_are_reported_with_the_engine_reason(): + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "counts": {"triggered": 40, "completed": 36, "failed": 4}, + "latestCompleted": {"checkpointId": 39, "triggerTimestamp": NOW - 120_000}, + "latestFailed": { + "checkpointId": 40, + "status": "FAILED", + "triggerTimestamp": NOW - 60_000, + "failureReason": "checkpoint expired before completing", + }, + } + ), + now_ms=NOW, + ) + + failing = _by_rule(findings, "checkpoint_failing") + assert "checkpoint expired before completing" in failing.detail + assert failing.evidence == "failed=4 completed=36" + + +def test_old_failures_under_a_newer_completion_are_not_a_warning(): + # `counts.failed` never goes down, so one expired checkpoint keeps it above + # zero for the rest of the job's life. A long-running pipeline that has + # completed thousands of checkpoints since is committing progress, and + # telling its owner it makes "no durable progress" is simply false. + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "counts": {"triggered": 5000, "completed": 4999, "failed": 1}, + "latestFailed": { + "checkpointId": 12, + "triggerTimestamp": NOW - 20 * 86400 * 1000, + "failureReason": "CHECKPOINT_EXPIRED", + }, + "latestCompleted": {"checkpointId": 5000, "triggerTimestamp": NOW - 10_000}, + } + ), + now_ms=NOW, + ) + + assert "checkpoint_failing" not in _rules(findings) + note = _by_rule(findings, "checkpoint_failed_before") + assert note.severity == "info" + assert "committing progress now" in note.detail + assert "no durable progress" not in note.detail + + +def test_a_failure_newer_than_the_last_completion_is_a_warning(): + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "counts": {"triggered": 5000, "completed": 4990, "failed": 10}, + "latestCompleted": {"checkpointId": 4990, "triggerTimestamp": NOW - 600_000}, + "latestFailed": {"checkpointId": 5000, "triggerTimestamp": NOW - 5_000}, + } + ), + now_ms=NOW, + ) + + failing = _by_rule(findings, "checkpoint_failing") + assert failing.severity == "warning" + assert "latest checkpoint failed" in failing.detail + + +def test_failures_with_no_completion_on_record_are_a_warning(): + # Nothing has ever completed, so the latest failure is the only outcome + # there is. + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "counts": {"triggered": 3, "completed": 0, "failed": 3}, + "latestFailed": {"checkpointId": 3, "triggerTimestamp": NOW - 5_000}, + } + ), + now_ms=NOW, + ) + + assert _by_rule(findings, "checkpoint_failing").severity == "warning" + + +def test_an_unknown_completed_count_is_not_reported_as_zero(): + # A count the engine did not send, or sent unusably, is unknown. Saying + # "completed none" from it would present a guess as a measurement. + for completed in (None, "x", {}): + findings = diagnose_checkpoints( + _overview( + {"pipelineId": 1, "counts": {"triggered": 12, "completed": completed}} + ), + now_ms=NOW, + ) + + assert "checkpoint_never_completed" not in _rules(findings) + + +def test_byte_sizes_are_rendered_at_a_readable_scale(): + findings = diagnose_checkpoints( + _overview( + { + "pipelineId": 1, + "history": _history( + [30_000] * 8, state_sizes=[6 << 30] * 4 + [1 << 30] * 4 + ), + } + ), + now_ms=NOW, + ) + + growing = _by_rule(findings, "checkpoint_state_growing") + assert "6.0GiB" in growing.detail and "1.0GiB" in growing.detail + + +def test_a_cluster_with_no_checkpoint_monitor_data_yields_nothing(): + # The endpoint answers with just the jobId when the monitor service is not + # running or has nothing for this job. That is a normal state, not an error. + assert diagnose_checkpoints({"jobId": "852"}, now_ms=NOW) == [] + + +def test_findings_from_several_pipelines_are_ordered_worst_first(): + findings = diagnose_checkpoints( + { + "jobId": "852", + "pipelines": [ + {"pipelineId": 1, "history": _history([30_000] * 4 + [6_000] * 4)}, + { + "pipelineId": 2, + "inProgress": [ + { + "checkpointId": 9, + "triggerTimestamp": NOW - (STALLED_CHECKPOINT_SECONDS + 10) * 1000, + "acknowledged": 0, + "total": 4, + } + ], + }, + ], + }, + now_ms=NOW, + ) + + # The stalled checkpoint on pipeline 2 is an error and must outrank the + # slowdown warning on pipeline 1, even though pipeline 1 is listed first. + assert findings[0].rule == "checkpoint_stalled" + assert "checkpoint_duration_growing" in _rules(findings) + + +def test_malformed_payloads_yield_no_findings_rather_than_raising(): + for payload in ( + None, + [], + "nope", + {}, + {"pipelines": None}, + {"pipelines": "none"}, + {"pipelines": [None, 7, "x"]}, + {"pipelines": [{"pipelineId": 1, "history": "none"}]}, + {"pipelines": [{"pipelineId": 1, "history": [None, 3]}]}, + {"pipelines": [{"pipelineId": 1, "history": [{"checkpoint": "no"}]}]}, + {"pipelines": [{"pipelineId": 1, "counts": "none"}]}, + {"pipelines": [{"pipelineId": 1, "counts": {"triggered": "x", "completed": None}}]}, + {"pipelines": [{"pipelineId": 1, "inProgress": "none"}]}, + {"pipelines": [{"pipelineId": 1, "inProgress": [None, 2]}]}, + {"pipelines": [{"pipelineId": 1, "inProgress": [{"triggerTimestamp": "soon"}]}]}, + ): + assert diagnose_checkpoints(payload, now_ms=NOW) == [] + + +def test_a_failed_count_still_reports_when_the_reason_is_unusable(): + # latestFailed can be an empty object, since the engine omits null fields, + # and a count of failures is worth reporting with or without a reason -- + # dropping the finding because the explanation is missing would discard the + # part that was actually measured. + for latest_failed in ({}, "none", None, 7): + findings = diagnose_checkpoints( + _overview( + {"pipelineId": 1, "counts": {"failed": 2}, "latestFailed": latest_failed} + ), + now_ms=NOW, + ) + + # Without a timestamp there is no evidence about the latest attempt, + # so the count is reported as a count rather than as a live fault. + assert _rules(findings) == ["checkpoint_failed_before"] + assert findings[0].severity == "info" + assert "not visible in this response" in findings[0].detail + assert "Last failure" not in findings[0].detail + + +def test_identical_timestamps_do_not_divide_by_zero(): + # A clock with coarse resolution, or a burst of savepoints, can give + # several entries the same trigger timestamp. + findings = diagnose_checkpoints( + _overview({"pipelineId": 1, "history": _history([9_000] * 8, interval_ms=0)}), + now_ms=NOW, + ) + + assert "checkpoint_bound" not in _rules(findings) diff --git a/seatunnel-cli/tests/test_diagnose_cli.py b/seatunnel-cli/tests/test_diagnose_cli.py new file mode 100644 index 0000000000..8962a6de0a --- /dev/null +++ b/seatunnel-cli/tests/test_diagnose_cli.py @@ -0,0 +1,575 @@ +# +# 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. +# + +"""End-to-end tests for ``seatunnel --diagnose <jobId>``. + +These drive `run_diagnose` against a real loopback HTTP server so the whole +path is covered: URL construction, the REST call, the rules, redaction and the +exit code. No LLM provider is constructed anywhere here, which is itself part +of what is being asserted -- the flag has to work for a user with no API key. +""" + +import json +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer + +import pytest +from rich.console import Console + +from seatunnel_cli import cli as cli_module + + +class _Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_GET(self): + self.server.paths.append(self.path) + status, body = self.server.script(self) + payload = body.encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + [email protected] +def engine(monkeypatch): + httpd = HTTPServer(("127.0.0.1", 0), _Handler) + httpd.paths = [] + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + base = f"http://127.0.0.1:{httpd.server_port}"; + # `connectors` resolves the base at import time, so patch the module + # attribute rather than the environment. + monkeypatch.setattr("seatunnel_cli.connectors._ENGINE_API_BASE", base) + try: + yield httpd + finally: + httpd.shutdown() + httpd.server_close() + + +def _run(job_id="852"): + console = Console(theme=cli_module.THEME, width=200, record=True) + code = cli_module.run_diagnose(job_id, console) + return code, " ".join(console.export_text().split()) + + +def test_a_failed_job_prints_the_error_code_and_exits_nonzero(engine): + engine.script = lambda h: ( + 200, + json.dumps( + { + "jobId": "852", + "jobName": "mysql-to-doris", + "jobStatus": "FAILED", + "errorMsg": ( + "Caused by: org.apache.seatunnel.common.exception.SeaTunnelRuntimeException: " + "ErrorCode:[JDBC-05], ErrorDescription:[Connection failed]" + ), + } + ), + ) + + code, out = _run() + + assert code == 1 + assert "mysql-to-doris" in out + assert "JDBC-05" in out + # Both reads are issued, and both are GETs, so the command stays read-only. + assert engine.paths == ["/job-info/852", "/jobs/checkpoints/852"] + + +_NOW = 1_760_000_000_000 + + +def _crash_looping_job(name="mysql-to-doris"): + # `generatedAt` is what the rules use as "now", so pinning it here makes + # the restore recency deterministic without touching the system clock. + return json.dumps( + { + "jobId": "852", + "jobName": name, + "jobStatus": "RUNNING", + "diagnostics": { + "generatedAt": _NOW, + "pipelines": [ + { + "pipelineId": 1, + "restoreCount": 5, + "maxRestoreCount": 10, + "pipelineStatus": "RUNNING", + "stateTimestamps": {"RUNNING": _NOW - 20_000}, + } + ], + }, + } + ) + + +def test_a_crash_looping_job_is_reported_but_exits_zero(engine): + # The job is alive, so this must not look like a command failure. + engine.script = lambda h: (200, _crash_looping_job()) + + code, out = _run() + + assert code == 0 + assert "restarted 5 times" in out + + +def test_the_severity_label_reaches_the_terminal(engine): + # `Finding.render()` prefixes "[warning]", and "warning" is a style name in + # THEME, so with rich markup enabled the label is consumed as a tag and the + # severity survives only as colour -- invisible in a pipe, in CI logs and + # under NO_COLOR, which is where this output usually ends up. + engine.script = lambda h: (200, _crash_looping_job()) + + _, out = _run() + + assert "[warning]" in out + + +def test_bracketed_engine_text_is_not_swallowed(engine): + # A SeaTunnel error message is full of lowercase bracketed fields, and + # those look exactly like style tags. + engine.script = lambda h: ( + 200, + json.dumps( + { + "jobId": "852", + "jobStatus": "FAILED", + "errorMsg": "ErrorCode:[JDBC-05], ErrorDescription:[connection refused]", + } + ), + ) + + code, out = _run() + + assert code == 1 + assert "[error]" in out + assert "connection refused" in out + + +def test_a_path_that_looks_like_a_closing_tag_does_not_crash(engine): + # "[/data/x.csv]" parses as a closing tag and raises MarkupError, which + # would crash the command in the one situation it exists for. + engine.script = lambda h: ( + 200, + json.dumps( + { + "jobId": "852", + "jobStatus": "FAILED", + "errorMsg": "java.io.IOException: cannot write [/data/x.csv]", + } + ), + ) + + code, out = _run() + + assert code == 1 + assert "/data/x.csv" in out + + +def test_a_bracketed_job_name_is_not_read_as_markup(engine): + # The job name comes from whoever submitted the job. + engine.script = lambda h: (200, _crash_looping_job(name="[prod] nightly")) + + _, out = _run() + + assert "[prod] nightly" in out + + +def test_a_non_numeric_job_id_is_rejected_before_any_request(engine): + # The engine parses a job id as a decimal long, so "1/2" would be sent as + # the path /job-info/1/2 and come back as a 404 about a missing job. + engine.script = lambda h: (200, "{}") + + code, out = _run("1/2") + + assert code == 1 + assert "is not a job id" in out + assert engine.paths == [] + + +def test_a_healthy_job_says_nothing_to_report(engine): + engine.script = lambda h: (200, json.dumps({"jobId": "852", "jobStatus": "FINISHED"})) + + code, out = _run() + + assert code == 0 + assert "Nothing to report" in out + + +def test_an_unknown_job_id_is_reported_from_the_response_the_engine_really_sends(engine): + # The decisive case, and not the obvious one: `JobInfoService` does not + # 404 for an id it does not know. When the job is in neither the + # running-job map nor the finished-job state it falls through to + # `{"jobId": "<id>"}` with HTTP 200. Treated as a job, that response has + # no status, no rule matches, and the output used to read "Nothing to + # report" -- telling someone their typo'd job id is healthy. + engine.script = lambda h: (200, json.dumps({"jobId": "999999999999999999"})) + + code, out = _run("999999999999999999") + + assert code == 1 + assert "No job 999999999999999999" in out + assert "Nothing to report" not in out + + +def test_a_non_object_json_body_does_not_crash(engine): + # A bare `null` or a list (a proxy, or a future response shape) reaches + # the same path and must not raise AttributeError. + for body in ("null", "[]", '"a string"'): + engine.script = lambda h, b=body: (200, b) + + code, out = _run() + + assert code == 1 + assert "No job 852" in out + + +def test_a_real_404_is_still_handled(engine): + # Jetty does not send it for a missing job, but a proxy or a different + # engine build can, and that branch still has to say something useful. + engine.script = lambda h: (404, "job not found") + + code, out = _run("999999999999999999") + + assert code == 1 + assert "No job 999999999999999999" in out + # A 404 is also what a base URL pointing at the member port rather than + # the Jetty HTTP port looks like, so the message has to offer that way out. + assert "SEATUNNEL_API_BASE" in out + + +def test_a_running_job_without_diagnostics_says_the_checks_did_not_run(engine): + # The engine omits the `diagnostics` block when the master cannot serve + # it, which leaves the crash-loop and stuck-state rules with no input. + # Silence there is the same false reassurance as the unknown-id case. + engine.script = lambda h: ( + 200, + json.dumps({"jobId": "852", "jobStatus": "RUNNING", "metrics": {}}), + ) + + code, out = _run() + + assert code == 0 + assert "Restart counts and state ages were not available" in out + assert "Nothing to report" not in out + + +def test_a_finished_job_without_diagnostics_is_not_warned_about(engine): + # A terminal job legitimately has no diagnostics block: the engine builds + # it only for a job the master still coordinates. Reporting that as a gap + # would fire on every finished job. + engine.script = lambda h: (200, json.dumps({"jobId": "852", "jobStatus": "FINISHED"})) + + code, out = _run() + + assert code == 0 + assert "were not available" not in out + assert "Nothing to report" in out + + +def test_the_http_error_body_is_redacted_and_capped(engine): + # The body is printed, so it needs the same treatment as the findings: + # it can quote a JDBC URL with a password, and rest.py only bounds it at + # 8 MiB, which is not a terminal-sized amount of text. + secret = "jdbc:mysql://db:3306/app?user=root&password=sup3rs3cret" + engine.script = lambda h: (500, f"submit failed for {secret} " + ("x" * 5000)) + + code, out = _run() + + assert code == 1 + assert "sup3rs3cret" not in out + assert "chars total" in out + assert len(out) < 5000 + + +def test_other_http_errors_show_the_engine_body(engine): + engine.script = lambda h: (500, "ErrorCode:[API-02], ErrorDescription:[master is down]") + + code, out = _run() + + assert code == 1 + # The bracketed description is the useful half of the body; asserting only + # on "API-02" is what let the markup bug through in the first place. + assert "ErrorDescription:[master is down]" in out + + +def test_an_unreachable_engine_names_the_override_env_var(monkeypatch): + # Nothing is listening on port 1, so this is the real "engine is down" + # path, not a simulated one. + monkeypatch.setattr("seatunnel_cli.connectors._ENGINE_API_BASE", "http://127.0.0.1:1";) + + code, out = _run() + + assert code == 1 + assert "SEATUNNEL_API_BASE" in out + + +def test_credentials_in_the_error_message_are_redacted(engine): + # A JDBC failure quotes the URL it tried, password included, and this + # output routinely gets pasted into a bug report. + engine.script = lambda h: ( + 200, + json.dumps( + { + "jobId": "852", + "jobStatus": "FAILED", + "errorMsg": ( + "Caused by: java.sql.SQLException: Could not connect to " + "jdbc:mysql://db:3306/app?user=root&password=sup3rs3cret" + ), + } + ), + ) + + code, out = _run() + + assert code == 1 + assert "sup3rs3cret" not in out + + +def _routed(job_info, checkpoints=None, checkpoint_status=200): + """Serve job-info and the checkpoint overview from one fake engine.""" + + def script(handler): + if handler.path.startswith("/jobs/checkpoints/"): + if checkpoints is None: + return checkpoint_status, "no checkpoint data" + return checkpoint_status, json.dumps(checkpoints) + return 200, json.dumps(job_info) + + return script + + +def _slow_checkpoint_history(count=8): + # Newest first, matching the engine. Recent checkpoints take 40s, older + # ones took 4s, so the job has visibly got slower. + now = 1_760_000_000_000 + entries = [] + for index in range(count): + duration = 40_000 if index < count // 2 else 4_000 + entries.append( + { + "pipelineId": 1, + "checkpoint": { + "checkpointId": 500 - index, + "status": "COMPLETED", + "triggerTimestamp": now - index * 600_000, + "completedTimestamp": now - index * 600_000 + duration, + "durationMillis": duration, + "stateSize": 4096, + }, + } + ) + return entries + + +def test_a_slowing_cdc_job_is_explained_even_though_rows_still_flow(engine): + # This is the case job-info alone cannot answer: RUNNING, both counters + # moving, no restores. Without the checkpoint read the only output would + # be "Running for ...", which restates the symptom. + engine.script = _routed( + { + "jobId": "852", + "jobName": "mysql-cdc-to-doris", + "jobStatus": "RUNNING", + "metrics": {"SourceReceivedCount": "184203311", "SinkWriteCount": "184201902"}, + }, + {"jobId": "852", "pipelines": [{"pipelineId": 1, "history": _slow_checkpoint_history()}]}, + ) + + code, out = _run() + + assert code == 0 + assert "getting slower" in out + assert "40.0s" in out and "4.0s" in out + + +def test_a_missing_checkpoint_endpoint_does_not_break_the_diagnosis(engine): + # An older engine has no such endpoint. The job diagnosis must survive it, + # because losing the whole answer to gain one optional rule is a bad trade. + engine.script = _routed( + { + "jobId": "852", + "jobStatus": "FAILED", + "errorMsg": "ErrorCode:[JDBC-05], ErrorDescription:[Connection failed]", + }, + checkpoint_status=404, + ) + + code, out = _run() + + assert code == 1 + assert "JDBC-05" in out + # No notice for a terminal job: a finished or failed job legitimately has + # no checkpoint overview retained, so reporting its absence would fire on + # every one of them. The live-job case is covered below. + assert "No checkpoint history" not in out + + +def test_a_healthy_job_with_no_checkpoint_data_does_not_look_fully_checked(engine): + # The combination that matters: the job-level rules ran and found nothing, + # while the checkpoint read failed. Without the notice this prints only + # "Nothing to report", which reads as "the slowdown checks ran and found + # nothing". The job carries a diagnostics block so that this test isolates + # the checkpoint gap rather than also tripping the job-level one. + engine.script = _routed( + { + "jobId": "852", + "jobStatus": "RUNNING", + "diagnostics": {"generatedAt": _NOW, "stateTimestamps": {"RUNNING": _NOW - 5_000}}, + }, + checkpoint_status=404, + ) + + code, out = _run() + + assert code == 0 + assert "No checkpoint history for this job" in out + # The notice replaces "Nothing to report" rather than sitting next to it: + # the slowdown half never ran, so a clean bill of health is not something + # this output can give. + assert "Nothing to report" not in out + + +def test_a_cluster_with_the_monitor_off_is_reported_not_passed_over(engine): + # The shape that actually occurs: with the monitor service off, or with + # nothing recorded for the job yet, CheckpointMonitorRestService answers + # 200 with only {"jobId": ...} and no `pipelines`. That is not an + # exception, so an except-only notice would never print and the output + # would end on "Nothing to report" with the slowdown half never run. + engine.script = _routed( + { + "jobId": "852", + "jobStatus": "RUNNING", + "diagnostics": {"generatedAt": _NOW, "stateTimestamps": {"RUNNING": _NOW - 5_000}}, + }, + checkpoints={"jobId": "852"}, + ) + + code, out = _run() + + assert code == 0 + assert "No checkpoint history for this job" in out + assert "Nothing to report" not in out + + +def test_an_unparsable_checkpoint_response_does_not_break_the_diagnosis(engine): + engine.script = _routed( + { + "jobId": "852", + "jobStatus": "FAILED", + "errorMsg": "ErrorCode:[JDBC-05], ErrorDescription:[Connection failed]", + }, + checkpoint_status=200, + ) + + code, out = _run() + + assert code == 1 + assert "JDBC-05" in out + + +def test_a_checkpoint_error_outranks_a_job_level_info_line(engine): + # The two rule sets are merged and sorted together. Sorting them + # separately would print "Running for 72h" above a stalled checkpoint. + now = 1_760_000_000_000 + engine.script = _routed( + { + "jobId": "852", + "jobStatus": "RUNNING", + "diagnostics": {"stateTimestamps": {"RUNNING": now - 72 * 3600 * 1000}}, + }, + { + "jobId": "852", + "pipelines": [ + { + "pipelineId": 1, + "inProgress": [ + { + "checkpointId": 91, + "triggerTimestamp": now - 1800 * 1000, + "acknowledged": 2, + "total": 6, + } + ], + } + ], + }, + ) + + code, out = _run() + + assert code == 0 + lines = [line for line in out.splitlines() if "[error]" in line or "[info]" in line] + assert "[error]" in lines[0] + assert "2 of 6" in out + + +def test_the_severity_label_is_printed_and_not_eaten_as_markup(engine): + # "error" is a style name in THEME, so rich would treat the "[error]" + # prefix as a style tag and drop it, leaving severity carried by colour + # alone -- lost in a pipe, in CI, or under NO_COLOR. + engine.script = _routed( + { + "jobId": "852", + "jobStatus": "FAILED", + "errorMsg": "ErrorCode:[JDBC-05], ErrorDescription:[Connection failed]", + }, + checkpoint_status=404, + ) + console = Console(theme=cli_module.THEME, width=200, record=True, no_color=True) + + code = cli_module.run_diagnose("852", console) + out = console.export_text() + + assert code == 1 + assert "[error]" in out + + +def test_brackets_in_an_engine_message_cannot_corrupt_the_output(engine): + # The checkpoint failure reason is engine text passed straight through to + # the terminal, and a Java message routinely contains bracketed text. The + # error headline would not do here: it keeps only the code and + # description, so brackets never reach the output through that path. + engine.script = _routed( + {"jobId": "852", "jobStatus": "RUNNING"}, + { + "jobId": "852", + "pipelines": [ + { + "pipelineId": 1, + "counts": {"triggered": 10, "completed": 6, "failed": 4}, + "latestFailed": { + "failureReason": "expired at Table[/bold] row[nope] idx[0]" + }, + } + ], + }, + ) + + code, out = _run() + + # Rich wraps the line, so compare with whitespace normalised. + flat = " ".join(out.split()) + assert code == 0 + assert "Table[/bold] row[nope] idx[0]" in flat diff --git a/seatunnel-cli/tests/test_diagnose_job.py b/seatunnel-cli/tests/test_diagnose_job.py new file mode 100644 index 0000000000..53dcee7338 --- /dev/null +++ b/seatunnel-cli/tests/test_diagnose_job.py @@ -0,0 +1,683 @@ +# +# 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. +# + +"""Tests for the rule-based job diagnosis. + +The job-info payloads here are shaped the way the engine actually sends them, +which is not the way one would guess: every metric leaf is a *string* because +`BaseService.metricsToJsonObject` calls `toString()` on it, the job-wide row +counts are `SourceReceivedCount`/`SinkWriteCount` while the `Table`-prefixed +names are objects keyed by table, and `stateTimestamps` is a state-name map +rather than an ordered list. Those three details are the ones a diagnosis +built on assumptions would get wrong, so they are pinned here. +""" + +from seatunnel_cli.diagnostics.jobs import ( + CRASH_LOOP_RECENT_SECONDS, + CRASH_LOOP_RESTORES, + STUCK_SECONDS, + diagnose_job, +) + +NOW = 1_760_000_000_000 # fixed clock, so the stuck rule is not time-dependent + +# Long enough for the progress rules to apply at all: a zero row count on a +# job that started seconds ago is the expected reading, not a symptom. +RUNNING_LONG_ENOUGH = {"RUNNING": NOW - (STUCK_SECONDS + 60) * 1000} + + +def _rules(findings): + return [f.rule for f in findings] + + +def _by_rule(findings, rule): + return next(f for f in findings if f.rule == rule) + + +def test_failed_job_leads_with_the_parsed_error_code(): + findings = diagnose_job( + { + "jobStatus": "FAILED", + "errorMsg": ( + "java.util.concurrent.CompletionException: wrapper\n" + "Caused by: org.apache.seatunnel.common.exception.SeaTunnelRuntimeException: " + "ErrorCode:[JDBC-05], ErrorDescription:[Connection failed]\n" + "Caused by: java.sql.SQLException: Access denied for user 'bench'\n" + ), + }, + now_ms=NOW, + ) + + assert "job_failed" in _rules(findings) + assert "JDBC-05" in _by_rule(findings, "job_failed").detail + + +def test_failed_with_no_error_message_says_so_instead_of_staying_silent(): + # Silence here would be the worst outcome: the user would think the tool + # had nothing to say about a job that definitely broke. + findings = diagnose_job({"jobStatus": "FAILED", "errorMsg": " "}, now_ms=NOW) + + assert _rules(findings) == ["failed_without_error"] + + +def test_repeated_restores_with_a_young_attempt_are_reported_as_a_crash_loop(): + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": { + "pipelines": [ + { + "pipelineId": 1, + "restoreCount": 4, + "maxRestoreCount": 10, + "pipelineStatus": "RUNNING", + # Restored 30s ago: the loop is happening now. + "stateTimestamps": {"RUNNING": NOW - 30_000}, + }, + ], + }, + }, + now_ms=NOW, + ) + + loop = _by_rule(findings, "pipeline_crash_loop") + assert loop.severity == "warning" + assert "restoreCount=4/10" in loop.evidence + assert "30s old" in loop.detail + + +def test_old_restores_on_a_steady_pipeline_are_history_not_a_crash_loop(): + # `restoreCount` counts restores since submission, so a streaming job that + # recovered from two transient faults weeks ago carries the same count as + # one restarting every minute. Reporting the first as "making no progress" + # is simply false, and it is the reading a long-lived job most often has. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": { + "pipelines": [ + { + "pipelineId": 1, + "restoreCount": 2, + "maxRestoreCount": 10, + "pipelineStatus": "RUNNING", + "stateTimestamps": {"RUNNING": NOW - 14 * 86400 * 1000}, + }, + ], + }, + }, + now_ms=NOW, + ) + + assert "pipeline_crash_loop" not in _rules(findings) + note = _by_rule(findings, "pipeline_restored_before") + assert note.severity == "info" + assert "has been running since" in note.detail + assert "no progress" not in note.detail + + +def test_a_pipeline_not_running_with_restores_behind_it_is_a_crash_loop(): + # Mid-restart is the one moment the loop is directly observable. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": { + "pipelines": [ + { + "pipelineId": 1, + "restoreCount": 5, + "maxRestoreCount": 10, + "pipelineStatus": "SCHEDULED", + "stateTimestamps": {"SCHEDULED": NOW - 1000}, + }, + ], + }, + }, + now_ms=NOW, + ) + + loop = _by_rule(findings, "pipeline_crash_loop") + assert loop.severity == "warning" + assert "SCHEDULED rather than RUNNING" in loop.detail + + +def test_restores_with_no_timing_evidence_are_reported_as_a_bare_count(): + # An older engine, or a master that has just taken over, can send the + # counter without pipeline state timestamps. Without them there is no + # evidence of a loop, so the count is reported as a count. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": { + "pipelines": [{"pipelineId": 1, "restoreCount": 4, "maxRestoreCount": 10}], + }, + }, + now_ms=NOW, + ) + + note = _by_rule(findings, "pipeline_restored_before") + assert note.severity == "info" + assert "not visible in this response" in note.detail + + +def _pipeline(status, restores=3, limit=10, **extra): + p = { + "pipelineId": 1, + "restoreCount": restores, + "maxRestoreCount": limit, + "pipelineStatus": status, + "stateTimestamps": {status: NOW - 1000}, + } + p.update(extra) + return p + + +def test_finished_and_cancelled_pipelines_are_not_called_restarting(): + # A restorable pipeline is in an end state for an instant only: + # prepareRestorePipeline increments the counter and immediately resets + # back toward CREATED. So a pipeline *observed* in an end state is over, + # and a restore count on it is history. In a multi-pipeline job this is + # the normal reading for a pipeline that finished after two restores. + for status in ("FINISHED", "CANCELED"): + findings = diagnose_job( + {"jobStatus": "RUNNING", "diagnostics": {"pipelines": [_pipeline(status)]}}, + now_ms=NOW, + ) + + assert findings == [], status + + +def test_an_end_state_pipeline_at_its_restore_limit_is_not_an_error_about_the_future(): + # "The next failure ends the job" is meaningless for a pipeline that has + # already finished, and it was previously reported as an error. + for status in ("FINISHED", "CANCELED"): + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": {"pipelines": [_pipeline(status, restores=10, limit=10)]}, + }, + now_ms=NOW, + ) + + assert findings == [], status + + +def test_a_failed_pipeline_is_reported_in_the_past_tense_only_when_exhausted(): + exhausted = diagnose_job( + { + "jobStatus": "FAILING", + "diagnostics": {"pipelines": [_pipeline("FAILED", restores=10, limit=10)]}, + }, + now_ms=NOW, + ) + spent = _by_rule(exhausted, "pipeline_restore_exhausted") + assert "used all 10 of its restores and then failed" in spent.detail + assert "next failure" not in spent.detail + + # Below the limit a failure was not restorable; the job-level failure + # finding covers it, so there is nothing to add here. + not_exhausted = diagnose_job( + { + "jobStatus": "FAILING", + "diagnostics": {"pipelines": [_pipeline("FAILED", restores=3, limit=10)]}, + }, + now_ms=NOW, + ) + assert not_exhausted == [] + + +def test_a_tearing_down_pipeline_reports_restores_as_history(): + for status in ("CANCELING", "FAILING"): + findings = diagnose_job( + {"jobStatus": "CANCELING", "diagnostics": {"pipelines": [_pipeline(status)]}}, + now_ms=NOW, + ) + + note = _by_rule(findings, "pipeline_restored_before") + assert note.severity == "info", status + assert "no progress" not in note.detail + + +def test_a_finished_pipeline_next_to_a_running_one_is_not_reported(): + # The multi-pipeline case the end-state bug showed up in: one pipeline + # finishing must not make the whole job look like it is crash-looping. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": { + "pipelines": [ + _pipeline("FINISHED", restores=4), + { + "pipelineId": 2, + "restoreCount": 4, + "maxRestoreCount": 10, + "pipelineStatus": "RUNNING", + "stateTimestamps": {"RUNNING": NOW - 30_000}, + }, + ] + }, + }, + now_ms=NOW, + ) + + rules = _rules(findings) + assert rules.count("pipeline_crash_loop") == 1 + assert "Pipeline 2" in _by_rule(findings, "pipeline_crash_loop").detail + + +def test_a_pipeline_between_attempts_is_reported_as_such(): + findings = diagnose_job( + {"jobStatus": "RUNNING", "diagnostics": {"pipelines": [_pipeline("DEPLOYING")]}}, + now_ms=NOW, + ) + + loop = _by_rule(findings, "pipeline_crash_loop") + assert loop.severity == "warning" + assert "between attempts" in loop.detail + + +def test_a_pipeline_restored_just_under_the_recency_window_still_warns(): + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": { + "pipelines": [ + { + "pipelineId": 1, + "restoreCount": 3, + "maxRestoreCount": 10, + "pipelineStatus": "RUNNING", + "stateTimestamps": { + "RUNNING": NOW - (CRASH_LOOP_RECENT_SECONDS - 1) * 1000 + }, + }, + ], + }, + }, + now_ms=NOW, + ) + + assert _by_rule(findings, "pipeline_crash_loop").severity == "warning" + + +def test_a_pipeline_at_its_restore_limit_is_an_error_not_a_warning(): + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": { + "pipelines": [ + {"pipelineId": 2, "restoreCount": 3, "maxRestoreCount": 3}, + ], + }, + }, + now_ms=NOW, + ) + + exhausted = _by_rule(findings, "pipeline_restore_exhausted") + assert exhausted.severity == "error" + + +def test_a_single_restore_is_not_flagged(): + # One restore is ordinary recovery from a transient fault; flagging it + # would make healthy clusters look sick. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": { + "pipelines": [{"pipelineId": 1, "restoreCount": CRASH_LOOP_RESTORES - 1}], + }, + }, + now_ms=NOW, + ) + + assert findings == [] + + +def test_long_scheduled_state_is_reported_as_stuck(): + entered = NOW - (STUCK_SECONDS + 60) * 1000 + findings = diagnose_job( + { + "jobStatus": "SCHEDULED", + "diagnostics": {"stateTimestamps": {"CREATED": entered - 5000, "SCHEDULED": entered}}, + }, + now_ms=NOW, + ) + + stuck = _by_rule(findings, "stuck_before_running") + assert "SCHEDULED for 3m00s" in stuck.detail + assert "slots" in stuck.detail + + +def test_the_duration_comes_from_the_current_state_not_an_older_one(): + # A restored job re-enters an earlier state, so the map holds entries for + # several states. Reading the wrong one -- the last key, or the oldest -- + # reports a duration off by days. + findings = diagnose_job( + { + "jobStatus": "SCHEDULED", + "diagnostics": { + "stateTimestamps": { + "SCHEDULED": NOW - (STUCK_SECONDS + 1) * 1000, + "CREATED": NOW - 10 * 24 * 3600 * 1000, + } + }, + }, + now_ms=NOW, + ) + + assert "2m01s" in _by_rule(findings, "stuck_before_running").detail + + +def test_a_state_with_no_entry_of_its_own_falls_back_to_the_newest_entry(): + # PENDING has no timestamp here, so the best available answer is the most + # recent transition the engine did record. + findings = diagnose_job( + { + "jobStatus": "PENDING", + "diagnostics": { + "stateTimestamps": { + "CREATED": NOW - 10 * 24 * 3600 * 1000, + "INITIALIZING": NOW - (STUCK_SECONDS + 1) * 1000, + } + }, + }, + now_ms=NOW, + ) + + assert "2m01s" in _by_rule(findings, "stuck_before_running").detail + + +def test_a_job_stuck_tearing_down_is_not_blamed_on_cluster_slots(): + # CANCELING forever is a task ignoring the stop request, so pointing the + # user at free slots would send them to the wrong place entirely. + findings = diagnose_job( + { + "jobStatus": "CANCELING", + "diagnostics": { + "stateTimestamps": {"CANCELING": NOW - (STUCK_SECONDS + 1) * 1000} + }, + }, + now_ms=NOW, + ) + + stuck = _by_rule(findings, "stuck_tearing_down") + assert "slots" not in stuck.detail + assert "stop request" in stuck.detail + + +def test_a_long_savepoint_is_not_described_as_an_unresponsive_task(): + # A savepoint writes the whole job state, so minutes can be exactly what + # the operator asked for. "Not responding to the stop request" would send + # someone hunting a hung task over a job doing its job. + findings = diagnose_job( + { + "jobStatus": "DOING_SAVEPOINT", + "diagnostics": { + "stateTimestamps": {"DOING_SAVEPOINT": NOW - (STUCK_SECONDS + 1) * 1000} + }, + }, + now_ms=NOW, + ) + + savepoint = _by_rule(findings, "savepoint_in_progress") + assert "stop request" not in savepoint.detail + assert "can be normal" in savepoint.detail + + +def test_the_masters_generated_at_is_preferred_over_the_local_clock(): + # Every timestamp compared here was written by the master, so mixing in a + # local clock that is minutes off invents stuck jobs (or hides them). With + # no `now_ms` injected, `generatedAt` is the right "now". + job_info = { + "jobStatus": "SCHEDULED", + "diagnostics": { + "generatedAt": NOW, + "stateTimestamps": {"SCHEDULED": NOW - (STUCK_SECONDS + 60) * 1000}, + }, + } + + assert "3m00s" in _by_rule(diagnose_job(job_info), "stuck_before_running").detail + + +def test_a_missing_generated_at_falls_back_to_the_local_clock(): + # An older engine sends no diagnostics block at all; the rules that do not + # need it must still work, so the fallback has to stay. + job_info = { + "jobStatus": "SCHEDULED", + "diagnostics": {"stateTimestamps": {"SCHEDULED": 1_000}}, + } + + assert "stuck_before_running" in _rules(diagnose_job(job_info)) + + +def test_a_freshly_scheduled_job_is_not_called_stuck(): + findings = diagnose_job( + { + "jobStatus": "SCHEDULED", + "diagnostics": {"stateTimestamps": {"SCHEDULED": NOW - 3000}}, + }, + now_ms=NOW, + ) + + assert findings == [] + + +def test_a_long_running_job_is_information_not_a_warning(): + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": {"stateTimestamps": {"RUNNING": NOW - 7200 * 1000}}, + }, + now_ms=NOW, + ) + + running = _by_rule(findings, "running_duration") + assert running.severity == "info" + assert "2h00m" in running.detail + + +def test_a_finished_job_is_not_measured_for_stuckness(): + # FINISHED is terminal, so the age of the state is just how long ago the + # job ended and says nothing about health. + findings = diagnose_job( + { + "jobStatus": "FINISHED", + "diagnostics": {"stateTimestamps": {"FINISHED": NOW - 30 * 86400 * 1000}}, + }, + now_ms=NOW, + ) + + assert findings == [] + + +def test_zero_rows_read_is_reported_from_string_valued_metrics(): + # The engine stringifies every metric leaf, so an int comparison on the + # raw value would never match. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "metrics": {"SourceReceivedCount": "0", "SinkWriteCount": "0"}, + "diagnostics": {"stateTimestamps": RUNNING_LONG_ENOUGH}, + }, + now_ms=NOW, + ) + + assert _by_rule(findings, "no_rows_read").severity == "warning" + + +def test_a_freshly_running_job_that_has_read_nothing_yet_is_not_flagged(): + # A zero counter one second into RUNNING is what every job looks like + # before its first batch arrives. Warning here would mean every job is + # reported as broken for its first two minutes. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "metrics": {"SourceReceivedCount": "0", "SinkWriteCount": "0"}, + "diagnostics": {"stateTimestamps": {"RUNNING": NOW - 1000}}, + }, + now_ms=NOW, + ) + + assert findings == [] + + +def test_progress_rules_stay_silent_without_a_running_timestamp(): + # No RUNNING entry means the job's uptime is unknown, and the difference + # between "read nothing for an hour" and "started a second ago" is the + # whole finding. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "metrics": {"SourceReceivedCount": "0", "SinkWriteCount": "0"}, + "diagnostics": {"stateTimestamps": {"CREATED": NOW - 86400 * 1000}}, + }, + now_ms=NOW, + ) + + assert "no_rows_read" not in _rules(findings) + + +def test_an_idle_streaming_source_is_a_note_not_a_warning(): + # A quiet topic or a CDC job waiting for its first change event reads + # nothing and is perfectly healthy, so for a streaming job this cannot be + # asserted as a fault. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "envOptions": {"job.mode": "STREAMING"}, + "metrics": {"SourceReceivedCount": "0", "SinkWriteCount": "0"}, + "diagnostics": {"stateTimestamps": RUNNING_LONG_ENOUGH}, + }, + now_ms=NOW, + ) + + idle = _by_rule(findings, "no_rows_read") + assert idle.severity == "info" + assert "normal" in idle.detail + + +def test_a_batch_source_reading_nothing_is_still_a_warning(): + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "envOptions": {"job.mode": "BATCH"}, + "metrics": {"SourceReceivedCount": "0", "SinkWriteCount": "0"}, + "diagnostics": {"stateTimestamps": RUNNING_LONG_ENOUGH}, + }, + now_ms=NOW, + ) + + assert _by_rule(findings, "no_rows_read").severity == "warning" + + +def test_rows_read_but_none_written_points_between_source_and_sink(): + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "metrics": {"SourceReceivedCount": "1500", "SinkWriteCount": "0"}, + "diagnostics": {"stateTimestamps": RUNNING_LONG_ENOUGH}, + }, + now_ms=NOW, + ) + + dropped = _by_rule(findings, "rows_read_none_written") + assert "1500" in dropped.detail + + +def test_healthy_throughput_produces_no_progress_finding(): + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "metrics": {"SourceReceivedCount": "1500", "SinkWriteCount": "1500"}, + "diagnostics": {"stateTimestamps": RUNNING_LONG_ENOUGH}, + }, + now_ms=NOW, + ) + + assert _rules(findings) == ["running_duration"] + + +def test_table_scoped_metrics_are_not_mistaken_for_scalars(): + # "TableSourceReceivedCount" is an object keyed by table name. Reading it + # as a count would make `int(...)` fail or, worse, coerce to something + # meaningless -- either way a false "no rows read" on a healthy job. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "metrics": { + "TableSourceReceivedCount": {"db.public.users": "1500"}, + "TableSinkWriteCount": {"db.public.users": "1500"}, + }, + "diagnostics": {"stateTimestamps": RUNNING_LONG_ENOUGH}, + }, + now_ms=NOW, + ) + + assert _rules(findings) == ["running_duration"] + + +def test_findings_are_ordered_worst_first(): + entered = NOW - (STUCK_SECONDS + 10) * 1000 + findings = diagnose_job( + { + "jobStatus": "FAILED", + "errorMsg": "ErrorCode:[JDBC-05], ErrorDescription:[Connection failed]", + "diagnostics": { + "pipelines": [{"pipelineId": 1, "restoreCount": 2, "maxRestoreCount": 3}], + "stateTimestamps": {"FAILED": entered}, + }, + }, + now_ms=NOW, + ) + + severities = [f.severity for f in findings] + assert severities == sorted(severities, key=["error", "warning", "info"].index) + assert findings[0].rule == "job_failed" + + +def test_malformed_payloads_yield_no_findings_rather_than_raising(): + # Everything here is a shape the engine should never send, but a crash + # while diagnosing is strictly worse than saying nothing. + for payload in ( + None, + [], + "FAILED", + {}, + {"jobStatus": None}, + {"jobStatus": "RUNNING", "diagnostics": "unavailable"}, + {"jobStatus": "RUNNING", "diagnostics": {"pipelines": "none"}}, + {"jobStatus": "RUNNING", "diagnostics": {"pipelines": [None, 7]}}, + {"jobStatus": "RUNNING", "diagnostics": {"stateTimestamps": []}}, + {"jobStatus": "RUNNING", "diagnostics": {"stateTimestamps": {"RUNNING": "soon"}}}, + {"jobStatus": "RUNNING", "metrics": None}, + {"jobStatus": "RUNNING", "metrics": {"SourceReceivedCount": "n/a"}}, + ): + assert diagnose_job(payload, now_ms=NOW) == [] + + +def test_restore_count_of_true_is_not_counted_as_one(): + # bool is a subclass of int in Python, so an unguarded int() would turn a + # stray `true` into a restore count. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": {"pipelines": [{"pipelineId": 1, "restoreCount": True}]}, + }, + now_ms=NOW, + ) + + assert findings == []
