This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 6071eabae5 [Improve][seatunnel-cli] Route engine REST calls through
one module (#12629)
6071eabae5 is described below
commit 6071eabae585977efbb7f15e066d0787f9c0bf2c
Author: SEZ <[email protected]>
AuthorDate: Fri Oct 9 08:08:08 2026 +0000
[Improve][seatunnel-cli] Route engine REST calls through one module (#12629)
---
seatunnel-cli/seatunnel_cli/agents.py | 27 ++----
seatunnel-cli/seatunnel_cli/cli.py | 32 ++-----
seatunnel-cli/seatunnel_cli/connectors.py | 22 ++---
seatunnel-cli/seatunnel_cli/rest.py | 106 ++++++++++++++++++++++
seatunnel-cli/tests/test_rest.py | 146 ++++++++++++++++++++++++++++++
5 files changed, 274 insertions(+), 59 deletions(-)
diff --git a/seatunnel-cli/seatunnel_cli/agents.py
b/seatunnel-cli/seatunnel_cli/agents.py
index 77f0b707e8..81c42ed7a4 100644
--- a/seatunnel-cli/seatunnel_cli/agents.py
+++ b/seatunnel-cli/seatunnel_cli/agents.py
@@ -664,28 +664,13 @@ def dry_run_config(config_str: str) -> dict:
pass
# ── Phase 3: REST API validation ──
- from .connectors import _check_engine, _ENGINE_API_BASE
+ from .connectors import _check_engine
if _check_engine():
- try:
- import urllib.request
- # Use submit-job endpoint with a dry-run approach:
- # We validate by checking if config parses on server side
- # without actually starting the job
- url = f"{_ENGINE_API_BASE}/submit-job"
- headers = {"Content-Type": "application/json"}
- # Submit with an invalid job name pattern to trigger validation
- # without actual execution — this is a best-effort approach
- # since SeaTunnel doesn't have a dedicated validate endpoint
- data = json.dumps({
- "env": {"job.mode": "BATCH"},
- "params": {"config": config_str, "format": "hocon"},
- }).encode("utf-8")
-
- # For now, just verify the config format is accepted by the API
- # A full submit-and-cancel approach would be too risky
- result["phase3_api"] = "SKIPPED (no dedicated validate endpoint)"
- except Exception as e:
- result["phase3_api"] = f"ERROR: {e}"
+ # The engine has no validate-only endpoint, and submitting a real job
to
+ # validate it is not an option, so there is nothing to call here. This
+ # used to build a request URL, headers and body and then discard them
+ # without ever sending anything; only the status below had any effect.
+ result["phase3_api"] = "SKIPPED (no dedicated validate endpoint)"
# ── Summary ──
phases_passed = ["Local: " + local_result]
diff --git a/seatunnel-cli/seatunnel_cli/cli.py
b/seatunnel-cli/seatunnel_cli/cli.py
index 83db23a9e8..b7e4512b0e 100644
--- a/seatunnel-cli/seatunnel_cli/cli.py
+++ b/seatunnel-cli/seatunnel_cli/cli.py
@@ -38,7 +38,7 @@ from prompt_toolkit import prompt as pt_prompt
from prompt_toolkit.completion import WordCompleter
from prompt_toolkit.history import FileHistory
-from . import __version__, get_data_dir
+from . import __version__, get_data_dir, rest
from .diagnostics import ParsedError, parse_error
from .llm_provider import create_provider, format_llm_error
from .agents import Orchestrator
@@ -1331,29 +1331,21 @@ class SeaTunnelCLI:
def _run_via_rest_api(self, api_base: str):
"""Submit job via REST API and poll status."""
- import json as _json
- import urllib.request
- import urllib.error
-
self.console.print(" Submitting job via REST API...", style="info")
try:
- url = f"{api_base}/submit-job?format=hocon"
- req = urllib.request.Request(
- url,
+ body = rest.request_json(
+ f"{api_base}/submit-job?format=hocon",
+ method="POST",
data=self.last_config.encode("utf-8"),
headers={"Content-Type": "text/plain"},
- method="POST",
+ timeout=30,
)
- with urllib.request.urlopen(req, timeout=30) as resp:
- body = _json.loads(resp.read().decode("utf-8"))
-
job_id = body.get("jobId")
job_name = body.get("jobName", "")
self.console.print(f" Job submitted: [bold]{job_id}[/bold]
({job_name})", style="success")
- except urllib.error.HTTPError as e:
- error_body = e.read().decode("utf-8", errors="replace")
- self.console.print(f" [error]Submit failed ({e.code}):[/error]")
- self._show_error_and_diagnose(error_body)
+ except rest.RestError as e:
+ self.console.print(f" [error]Submit failed ({e.status}):[/error]")
+ self._show_error_and_diagnose(e.body)
return
except Exception as e:
self.console.print(f" [error]Submit failed: {e}[/error]")
@@ -1364,9 +1356,6 @@ class SeaTunnelCLI:
def _poll_job_status(self, api_base: str, job_id: str):
"""Poll job status until terminal state, then show results."""
- import json as _json
- import urllib.request
-
terminal_states = {"FINISHED", "CANCELED", "FAILED"}
self.console.print(" Waiting for job to complete...", style="info")
@@ -1377,10 +1366,7 @@ class SeaTunnelCLI:
for _ in range(max_polls):
time.sleep(poll_interval)
try:
- url = f"{api_base}/job-info/{job_id}"
- req = urllib.request.Request(url, method="GET")
- with urllib.request.urlopen(req, timeout=10) as resp:
- info = _json.loads(resp.read().decode("utf-8"))
+ info = rest.request_json(f"{api_base}/job-info/{job_id}",
timeout=10)
status = info.get("jobStatus", "UNKNOWN")
if status != last_status:
diff --git a/seatunnel-cli/seatunnel_cli/connectors.py
b/seatunnel-cli/seatunnel_cli/connectors.py
index 1fa2e73769..ecce50c197 100644
--- a/seatunnel-cli/seatunnel_cli/connectors.py
+++ b/seatunnel-cli/seatunnel_cli/connectors.py
@@ -40,10 +40,10 @@ import os
import re
import subprocess
import time
-import urllib.request
-import urllib.error
from pathlib import Path
+from . import rest
+
logger = logging.getLogger(__name__)
# ─── Runtime API client ───
@@ -72,12 +72,7 @@ def _check_engine() -> bool:
now = time.monotonic()
if _ENGINE_AVAILABLE is not None and (now - _ENGINE_CHECK_TS) < 60:
return _ENGINE_AVAILABLE
- try:
- req = urllib.request.Request(f"{_ENGINE_API_BASE}/running-jobs",
method="GET")
- with urllib.request.urlopen(req, timeout=2) as resp:
- _ENGINE_AVAILABLE = resp.status == 200
- except Exception:
- _ENGINE_AVAILABLE = False
+ _ENGINE_AVAILABLE = rest.is_reachable(f"{_ENGINE_API_BASE}/running-jobs",
timeout=2)
_ENGINE_CHECK_TS = now
return _ENGINE_AVAILABLE
@@ -127,13 +122,10 @@ def _fetch_option_rules(plugin_type: str, plugin_name:
str) -> dict | None:
if _check_engine():
try:
url =
f"{_ENGINE_API_BASE}/option-rules?type={plugin_type}&plugin={plugin_name}"
- req = urllib.request.Request(url, method="GET")
- with urllib.request.urlopen(req, timeout=5) as resp:
- if resp.status == 200:
- data = json.loads(resp.read().decode("utf-8"))
- _API_CACHE[cache_key] = data
- _write_disk_cache(cache_key, data)
- return data
+ data = rest.request_json(url, timeout=5)
+ _API_CACHE[cache_key] = data
+ _write_disk_cache(cache_key, data)
+ return data
except Exception as e:
logger.debug(f"option-rules API call failed for {cache_key}: {e}")
diff --git a/seatunnel-cli/seatunnel_cli/rest.py
b/seatunnel-cli/seatunnel_cli/rest.py
new file mode 100644
index 0000000000..de844889c0
--- /dev/null
+++ b/seatunnel-cli/seatunnel_cli/rest.py
@@ -0,0 +1,106 @@
+#
+# 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.
+#
+
+"""One place where the CLI talks to the SeaTunnel engine REST API.
+
+The same six lines of `urllib.request.Request` / `urlopen` / `json.loads` were
+repeated at four call sites across three modules, each with its own timeout and
+its own idea of what to do with a non-200. That made two things awkward: there
+was nowhere to put behaviour that every engine call needs (the response size
+bound below), and a caller that wanted an HTTP error *body* had to reach into
+`urllib.error.HTTPError` itself.
+
+Nothing here adds a dependency; it is the same `urllib` calls behind one door.
+"""
+
+import json
+import logging
+import urllib.error
+import urllib.request
+
+logger = logging.getLogger(__name__)
+
+# Engine responses are JSON status documents, not payloads. A job list on a
busy
+# cluster or an errorMsg carrying a full stack trace is the realistic upper
end,
+# so this is generous while still refusing to buffer an unbounded body into
+# memory from a host the CLI does not control.
+MAX_RESPONSE_BYTES = 8 * 1024 * 1024
+
+DEFAULT_TIMEOUT = 10
+
+
+class RestError(Exception):
+ """An HTTP-level failure, carrying the body so callers can diagnose it.
+
+ `_run_via_rest_api` needs the response body of a failed submit -- that is
+ where the engine explains why a config was rejected -- so the body is read
+ here rather than leaving callers to remember that `HTTPError` is also a
+ readable file object.
+ """
+
+ def __init__(self, status: int, body: str, url: str):
+ super().__init__(f"HTTP {status} from {url}")
+ self.status = status
+ self.body = body
+ self.url = url
+
+
+def request_json(
+ url: str,
+ *,
+ method: str = "GET",
+ data: bytes | None = None,
+ headers: dict[str, str] | None = None,
+ timeout: float = DEFAULT_TIMEOUT,
+) -> dict:
+ """Call the engine and return the decoded JSON body.
+
+ Raises `RestError` for an HTTP error status; transport failures and
+ undecodable bodies propagate as-is, because the callers here already treat
+ "could not reach the engine" and "engine said no" differently.
+ """
+ req = urllib.request.Request(
+ url, data=data, headers=headers or {}, method=method
+ )
+ try:
+ with urllib.request.urlopen(req, timeout=timeout) as resp:
+ raw = resp.read(MAX_RESPONSE_BYTES + 1)
+ except urllib.error.HTTPError as e:
+ body = e.read(MAX_RESPONSE_BYTES).decode("utf-8", errors="replace")
+ raise RestError(e.code, body, url) from e
+
+ if len(raw) > MAX_RESPONSE_BYTES:
+ raise RestError(
+ 0,
+ f"response from {url} exceeded {MAX_RESPONSE_BYTES} bytes",
+ url,
+ )
+ return json.loads(raw.decode("utf-8"))
+
+
+def is_reachable(url: str, *, timeout: float = 2) -> bool:
+ """Whether the engine answered 200 at `url`. Never raises.
+
+ Used for the liveness probe, where any failure -- refused, timed out, 500
--
+ means the same thing to the caller: fall back to offline metadata.
+ """
+ req = urllib.request.Request(url, method="GET")
+ try:
+ with urllib.request.urlopen(req, timeout=timeout) as resp:
+ return resp.status == 200
+ except Exception:
+ return False
diff --git a/seatunnel-cli/tests/test_rest.py b/seatunnel-cli/tests/test_rest.py
new file mode 100644
index 0000000000..5562723a49
--- /dev/null
+++ b/seatunnel-cli/tests/test_rest.py
@@ -0,0 +1,146 @@
+#
+# 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 shared engine REST helper.
+
+These run against a real loopback HTTP server rather than a patched `urlopen`,
+so they exercise the actual `urllib` behaviour the call sites depend on --
+in particular that an error *body* survives, which is what the submit path
+feeds into failure diagnosis.
+"""
+
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+import pytest
+
+from seatunnel_cli import rest
+
+
+class _Handler(BaseHTTPRequestHandler):
+ """Serves whatever the active test put in `self.server.script`."""
+
+ def log_message(self, *args): # keep pytest output clean
+ pass
+
+ def _respond(self):
+ status, body = self.server.script(self)
+ payload = body.encode("utf-8") if isinstance(body, str) else body
+ 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)
+
+ do_GET = _respond
+ do_POST = _respond
+
+
[email protected]
+def server():
+ httpd = HTTPServer(("127.0.0.1", 0), _Handler)
+ thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+ thread.start()
+ httpd.base = f"http://127.0.0.1:{httpd.server_port}";
+ try:
+ yield httpd
+ finally:
+ httpd.shutdown()
+ httpd.server_close()
+
+
+def test_get_returns_decoded_json(server):
+ server.script = lambda h: (200, json.dumps({"jobStatus": "RUNNING"}))
+
+ assert rest.request_json(f"{server.base}/job-info/7") == {"jobStatus":
"RUNNING"}
+
+
+def test_post_sends_body_and_headers(server):
+ seen = {}
+
+ def script(handler):
+ seen["method"] = handler.command
+ seen["type"] = handler.headers.get("Content-Type")
+ length = int(handler.headers.get("Content-Length", 0))
+ seen["body"] = handler.rfile.read(length).decode("utf-8")
+ return 200, json.dumps({"jobId": "852"})
+
+ server.script = script
+
+ body = rest.request_json(
+ f"{server.base}/submit-job?format=hocon",
+ method="POST",
+ data=b"env {}\n",
+ headers={"Content-Type": "text/plain"},
+ timeout=5,
+ )
+
+ assert body == {"jobId": "852"}
+ assert seen["method"] == "POST"
+ assert seen["type"] == "text/plain"
+ assert seen["body"] == "env {}\n"
+
+
+def test_http_error_keeps_status_and_body(server):
+ # The submit path shows this body to the user and feeds it to diagnosis,
+ # so losing it would turn a specific rejection into "Submit failed (400)".
+ server.script = lambda h: (400, "ErrorCode:[API-02], ErrorDescription:[bad
config]")
+
+ with pytest.raises(rest.RestError) as excinfo:
+ rest.request_json(f"{server.base}/submit-job")
+
+ assert excinfo.value.status == 400
+ assert "ErrorCode:[API-02]" in excinfo.value.body
+ assert excinfo.value.url.endswith("/submit-job")
+
+
+def test_oversized_response_is_refused_rather_than_buffered(server):
+ big = json.dumps({"errorMsg": "x" * (rest.MAX_RESPONSE_BYTES + 64)})
+ server.script = lambda h: (200, big)
+
+ with pytest.raises(rest.RestError) as excinfo:
+ rest.request_json(f"{server.base}/job-info/7")
+
+ assert "exceeded" in excinfo.value.body
+
+
+def test_a_response_at_the_limit_is_still_accepted(server):
+ filler = "y" * (rest.MAX_RESPONSE_BYTES - len('{"errorMsg": ""}'))
+ payload = json.dumps({"errorMsg": filler})
+ assert len(payload) == rest.MAX_RESPONSE_BYTES
+ server.script = lambda h: (200, payload)
+
+ assert rest.request_json(f"{server.base}/job-info/7")["errorMsg"] == filler
+
+
+def test_is_reachable_true_on_200(server):
+ server.script = lambda h: (200, "[]")
+
+ assert rest.is_reachable(f"{server.base}/running-jobs") is True
+
+
+def test_is_reachable_false_on_error_status(server):
+ server.script = lambda h: (500, "boom")
+
+ assert rest.is_reachable(f"{server.base}/running-jobs") is False
+
+
+def test_is_reachable_false_when_nothing_is_listening():
+ # The liveness probe must not raise: every failure means "engine offline"
+ # and the caller falls back to offline metadata.
+ assert rest.is_reachable("http://127.0.0.1:1/running-jobs";, timeout=1) is
False
