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-12731-c70a7a49f24418aac42ffb96ee7a4350d050e4e7 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 01b474a0dc4a1de58997875f5fadab7a16cfe6b8 Author: SEZ <[email protected]> AuthorDate: Thu Oct 22 14:20:54 2026 +0000 [Fix][seatunnel-cli] Correct the --diagnose docs and two out-of-order restore-rule branches (#12784) --- seatunnel-cli/README.md | 22 +++++--- seatunnel-cli/README.zh-CN.md | 19 ++++--- seatunnel-cli/seatunnel_cli/cli.py | 4 +- seatunnel-cli/seatunnel_cli/connectors.py | 4 +- seatunnel-cli/seatunnel_cli/diagnostics/jobs.py | 48 +++++++++++------ seatunnel-cli/tests/test_diagnose_cli.py | 72 +++++++++++++++++++++++++ seatunnel-cli/tests/test_diagnose_job.py | 51 ++++++++++++++++++ 7 files changed, 186 insertions(+), 34 deletions(-) diff --git a/seatunnel-cli/README.md b/seatunnel-cli/README.md index 71d1d07e29..1ec5d1f4f1 100644 --- a/seatunnel-cli/README.md +++ b/seatunnel-cli/README.md @@ -296,10 +296,9 @@ port -- `8080` in the packaged `config/seatunnel.yaml`. Point ``` 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) + [warning] Pipeline 1 has restarted 4 times, and the current attempt is only 20s old. + Check the task logs for what triggered the restores. (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 @@ -343,7 +342,9 @@ wrong hint costs more than a missing one: 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. + as history rather than as a live fault. One exception: a pipeline that has + used its whole restore budget is reported even after a single restore, which + you will see on a job with `job.retry.times = 1`. - 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. @@ -356,9 +357,14 @@ wrong hint costs more than a missing one: 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. +- durations are measured against the engine's own clock: `diagnostics.generatedAt` + from the `/job-info` response, which the master stamps when it builds the + reply, is used for both the job rules and the checkpoint rules, so a client + clock that disagrees with the cluster cannot invent a stuck job. The + checkpoint overview's own `updatedAt` is deliberately not used -- the engine + writes it only when a checkpoint event is recorded, so for a checkpoint hung + on its last subtask it stops advancing and would hide the very stall the + rule looks for. ### CLI Arguments diff --git a/seatunnel-cli/README.zh-CN.md b/seatunnel-cli/README.zh-CN.md index a5481d4767..5cf2a2a588 100644 --- a/seatunnel-cli/README.zh-CN.md +++ b/seatunnel-cli/README.zh-CN.md @@ -263,10 +263,9 @@ SEATUNNEL_API_BASE=http://zeta-master:8080 seatunnel --diagnose 8523626707716669 ``` 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) + [warning] Pipeline 1 has restarted 4 times, and the current attempt is only 20s old. + Check the task logs for what triggered the restores. (restoreCount=4/10 + pipelineStatus=RUNNING running for 20s since the last restore) ``` 它会报告：失败作业解析出的错误码与根因、反复重启的 pipeline（以及是否已用尽 @@ -301,7 +300,8 @@ acknowledged it`），指向具体的某一环而不是整个作业。 `FAILED` 的只在重启预算用尽时以过去时报告；正在关停的则把重启次数作为历史 记录输出。只有当 pipeline 处于两次尝试之间、或距上次重启运行不足 10 分钟时， 才会判定为重启循环——因此每次尝试都能存活超过 10 分钟的循环会被当作历史 - 记录，而不是当前故障。 + 记录，而不是当前故障。有一个例外：重启预算已经用尽的 pipeline 即使只重启过 + 一次也会被报告，`job.retry.times = 1` 的作业就会遇到这种情况。 - 行数规则要等作业进入 RUNNING 满两分钟后才生效；对 `job.mode = STREAMING` 的作业，源端空闲只作为提示而非告警——安静的 topic 和配错的 topic 在这里 读数完全相同。 @@ -313,9 +313,12 @@ acknowledged it`），指向具体的某一环而不是整个作业。 - checkpoint 卡住的判定阈值是进行中超过 2 分钟，高于打包配置里的 `checkpoint.timeout`（60 秒）。默认配置下挂死的 checkpoint 会先超时、以失败 的形式报出，因此这条规则主要服务于调高了该超时的集群。 -- 时长以引擎自己的时钟为基准计算（作业规则用 `diagnostics.generatedAt`， - checkpoint 规则用 `updatedAt`），客户端时钟与集群不一致时不会凭空造出一个 - "卡住"的作业。 +- 时长以引擎自己的时钟为基准计算：作业规则和 checkpoint 规则都使用 + `/job-info` 响应里的 `diagnostics.generatedAt`（由 master 在构造响应时打上）， + 因此客户端时钟与集群不一致时不会凭空造出一个"卡住"的作业。checkpoint + overview 自带的 `updatedAt` 刻意不使用——引擎只在记录 checkpoint 事件时才 + 写入它，所以当某个 checkpoint 卡在最后一个 subtask 上时它就不再前进，反而 + 会把这条规则本该发现的卡死隐藏掉。 ### CLI 参数 diff --git a/seatunnel-cli/seatunnel_cli/cli.py b/seatunnel-cli/seatunnel_cli/cli.py index af1ec00cb9..c81db9bf44 100644 --- a/seatunnel-cli/seatunnel_cli/cli.py +++ b/seatunnel-cli/seatunnel_cli/cli.py @@ -1582,7 +1582,9 @@ _SEVERITY_STYLE = {"error": "error", "warning": "warning", "info": "info"} # 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+$") +# `\Z`, not `$`: `$` also matches just before a trailing newline, so a job +# id pasted with the newline still got interpolated into the URL path. +_JOB_ID_RE = re.compile(r"-?\d+\Z") # 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 diff --git a/seatunnel-cli/seatunnel_cli/connectors.py b/seatunnel-cli/seatunnel_cli/connectors.py index ade13972ef..b49fc4b188 100644 --- a/seatunnel-cli/seatunnel_cli/connectors.py +++ b/seatunnel-cli/seatunnel_cli/connectors.py @@ -61,7 +61,9 @@ logger = logging.getLogger(__name__) # # In local mode (-m local): REST API is NOT available. # Users can override with SEATUNNEL_API_BASE env var — needed when -# enable-dynamic-port moves the listener off 8080, or for a remote cluster. +# enable-dynamic-port moves the listener off 8080, when a non-empty +# seatunnel.engine.http.context-path mounts every servlet under a prefix +# (the code default is empty), or for a remote cluster. _ENGINE_API_BASE = os.environ.get("SEATUNNEL_API_BASE", "http://localhost:8080";) _API_CACHE: dict[str, dict] = {} # memory cache: "source:Jdbc" → parsed response _ENGINE_AVAILABLE: bool | None = None # None = not checked yet diff --git a/seatunnel-cli/seatunnel_cli/diagnostics/jobs.py b/seatunnel-cli/seatunnel_cli/diagnostics/jobs.py index 1f47e6d7ed..1c7db694a8 100644 --- a/seatunnel-cli/seatunnel_cli/diagnostics/jobs.py +++ b/seatunnel-cli/seatunnel_cli/diagnostics/jobs.py @@ -53,7 +53,9 @@ 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. +# rule only fires from the second restore. The exhausted-budget finding is the +# exception: it does not need repetition, because a spent budget means the next +# failure is terminal whether that took one restore or ten. CRASH_LOOP_RESTORES = 2 # ``restoreCount`` counts restores since submission, so on its own it says @@ -210,12 +212,23 @@ def _crash_loop_findings(pipelines: list, now_ms: int) -> list[Finding]: if not isinstance(pipeline, dict): continue restores = as_int(pipeline.get("restoreCount")) - if restores is None or restores < CRASH_LOOP_RESTORES: + # Nothing to say about a pipeline that has never restored. Subsumed by + # the gate below, but kept so that case is explicit rather than an + # accident of how `exhausted` evaluates at zero. + if restores is None or restores < 1: continue pipeline_id = pipeline.get("pipelineId", "?") limit = as_int(pipeline.get("maxRestoreCount")) exhausted = limit is not None and limit > 0 and restores >= limit + # A spent restore budget is worth reporting however small the budget + # was: with `job.retry.times = 1` a pipeline reaches its limit on the + # first restore, and gating on CRASH_LOOP_RESTORES first made that + # case produce no finding at all. Below the limit, one restore is + # still ordinary recovery and stays silent. + if not exhausted and restores < CRASH_LOOP_RESTORES: + continue + status = pipeline.get("pipelineStatus") status = status.strip().upper() if isinstance(status, str) and status.strip() else None @@ -247,29 +260,32 @@ def _crash_loop_findings(pipelines: list, now_ms: int) -> list[Finding]: ) continue - # Budget gone while the pipeline is still live: the engine will not - # restore again, so the next failure is terminal. - if exhausted: + # Shutting down. Whatever the restores were, they are not what is + # happening now, and the stop may well have been requested. Checked + # before the exhausted branch: a pipeline that is being torn down has + # no "next failure" to warn about, which is the same ordering mistake + # the end-state check above avoids. + if status in _PIPELINE_TEARING_DOWN: 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.", + "info", + "pipeline_restored_before", + f"Pipeline {pipeline_id} restarted {restores} times since it was " + f"submitted and is now {status}.", 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: + # 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( - "info", - "pipeline_restored_before", - f"Pipeline {pipeline_id} restarted {restores} times since it was " - f"submitted and is now {status}.", + "error", + "pipeline_restore_exhausted", + f"Pipeline {pipeline_id} has restarted {restores} times, which is " + f"its limit: the next failure ends the job.", evidence, ) ) diff --git a/seatunnel-cli/tests/test_diagnose_cli.py b/seatunnel-cli/tests/test_diagnose_cli.py index 8962a6de0a..b878efd084 100644 --- a/seatunnel-cli/tests/test_diagnose_cli.py +++ b/seatunnel-cli/tests/test_diagnose_cli.py @@ -195,6 +195,78 @@ def test_a_bracketed_job_name_is_not_read_as_markup(engine): assert "[prod] nightly" in out +def test_a_job_id_with_a_trailing_newline_is_rejected(engine): + # `$` in the validation pattern also matches before a trailing newline, so + # an id pasted with the newline attached used to pass and go straight into + # the URL path. + engine.script = lambda h: (200, "{}") + + code, out = _run("852\n") + + assert code == 1 + assert "is not a job id" in out + assert engine.paths == [] + + +def test_the_engine_clock_is_passed_to_the_checkpoint_rules(engine): + # The plumbing, not just the rule: `diagnostics.generatedAt` from the + # /job-info response has to reach diagnose_checkpoints, or the checkpoint + # ages are computed against the local clock and a stalled checkpoint is + # missed on any host whose clock differs from the cluster's. + # + # `generatedAt` here is far in the past relative to the real clock, and the + # checkpoint was triggered just before it. If the local clock were used the + # age would be enormous and the stalled rule would fire; with the engine + # clock it is 10s and must not. + engine.script = _routed( + { + "jobId": "852", + "jobStatus": "RUNNING", + "diagnostics": {"generatedAt": _NOW, "stateTimestamps": {"RUNNING": _NOW - 5_000}}, + }, + checkpoints={ + "jobId": "852", + "pipelines": [ + { + "pipelineId": 1, + "inProgress": [{"checkpointId": 9, "triggerTimestamp": _NOW - 10_000}], + } + ], + }, + ) + + code, out = _run() + + assert code == 0 + assert "has been in progress" not in out + + +def test_a_checkpoint_stalled_against_the_engine_clock_is_still_reported(engine): + # The other direction, so the test above cannot pass by the rule simply + # never firing: same plumbing, a trigger well past the threshold. + engine.script = _routed( + { + "jobId": "852", + "jobStatus": "RUNNING", + "diagnostics": {"generatedAt": _NOW, "stateTimestamps": {"RUNNING": _NOW - 5_000}}, + }, + checkpoints={ + "jobId": "852", + "pipelines": [ + { + "pipelineId": 1, + "inProgress": [{"checkpointId": 9, "triggerTimestamp": _NOW - 300_000}], + } + ], + }, + ) + + code, out = _run() + + assert code == 0 + assert "has been in progress for 5m00s" 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. diff --git a/seatunnel-cli/tests/test_diagnose_job.py b/seatunnel-cli/tests/test_diagnose_job.py index 53dcee7338..35b7f80b16 100644 --- a/seatunnel-cli/tests/test_diagnose_job.py +++ b/seatunnel-cli/tests/test_diagnose_job.py @@ -186,6 +186,57 @@ def _pipeline(status, restores=3, limit=10, **extra): return p +def test_a_spent_restore_budget_is_reported_even_when_the_limit_is_one(): + # `job.retry.times = 1` gives maxRestoreCount 1, so the budget is spent on + # the very first restore. The crash-loop threshold used to be applied + # first, which made this case produce no finding at all -- the one reading + # where the job is certain to die on the next failure. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": {"pipelines": [_pipeline("RUNNING", restores=1, limit=1)]}, + }, + now_ms=NOW, + ) + + spent = _by_rule(findings, "pipeline_restore_exhausted") + assert spent.severity == "error" + assert "restoreCount=1/1" in spent.evidence + + +def test_one_restore_below_the_limit_is_still_silent(): + # The threshold still applies when the budget is not spent: a single + # recovery from a transient fault is ordinary operation. + findings = diagnose_job( + { + "jobStatus": "RUNNING", + "diagnostics": {"pipelines": [_pipeline("RUNNING", restores=1, limit=10)]}, + }, + now_ms=NOW, + ) + + assert findings == [] + + +def test_a_tearing_down_pipeline_at_its_limit_is_not_warned_about_a_next_failure(): + # Ordering: the teardown check has to precede the exhausted check. A + # pipeline being cancelled has no "next failure" to warn about, which is + # the same mistake the end-state check already avoids. + for status in ("CANCELING", "FAILING"): + findings = diagnose_job( + { + "jobStatus": "CANCELING", + "diagnostics": {"pipelines": [_pipeline(status, restores=5, limit=5)]}, + }, + now_ms=NOW, + ) + + note = _by_rule(findings, "pipeline_restored_before") + assert note.severity == "info", status + assert "next failure" not in note.detail + assert "pipeline_restore_exhausted" not in _rules(findings) + + 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
