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-12629-0b9e75a6ceb846cd08b06184127560cbd96c01b1 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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
