From 8ededc50a997e010a3cd2ed76e7689cc3865e18a Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 9 Jul 2026 16:24:05 +0800 Subject: [PATCH] feat(connection): add history_server_url config field MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a new optional field to Connection for the Spark History Server base URL. The field is stored as part of the connection but is not yet consumed by any tool — for now it's a labeled place to record where SHS lives on the cluster, and a hook for future SHS-specific tools. To actually fetch SHS endpoints today, use fetch_url with the SHS host added to url_allowlist. The field is Optional[str], default None. Same nullability as yarn_rm_url. The SHS host and the YARN RM host are usually different, so this is independent of yarn_rm_url. Schema: - models.py: Connection.history_server_url (str | None, default None) - requests.py: - SaveConnectionRequest.history_server_url (str | None, default None) - UpdateConnectionRequest.history_server_url (str | None, default None) - tools/connections.py: save_connection signature gains history_server_url with the _UNSET sentinel pattern (same as yarn_rm_url, url_allowlist, etc.) so the upsert path correctly distinguishes "not provided" from "explicitly None". - tools/connections.py: update_connection docstring lists the new field in the mutable fields set. Tests (4 new in tests/unit/test_connection_tools.py): - test_save_connection_with_history_server_url - test_save_connection_history_server_url_defaults_to_none - test_update_connection_changes_history_server_url - test_update_connection_keeps_history_server_url_when_omitted Tests: 398 passed (was 394, +4 net). Co-Authored-By: Claude Fable 5 --- spark_executor/models.py | 12 ++++++++++ spark_executor/tools/connections.py | 6 ++++- spark_executor/tools/requests.py | 17 ++++++++++++++ tests/unit/test_connection_tools.py | 36 +++++++++++++++++++++++++++++ 4 files changed, 70 insertions(+), 1 deletion(-) diff --git a/spark_executor/models.py b/spark_executor/models.py index 96f7bbb..b315ed9 100644 --- a/spark_executor/models.py +++ b/spark_executor/models.py @@ -106,6 +106,18 @@ class Connection(BaseModel): "'ccam50' but not 'ccam50.evil.com'." ), ) + history_server_url: str | None = Field( + default=None, + description=( + "Optional Spark History Server (SHS) base URL, e.g. " + "'http://history.prod.internal:18080'. Stored as part of the " + "connection for reference and to be consumed by future SHS-" + "specific tools. Currently **not consumed by any tool** — to " + "fetch SHS endpoints today, use fetch_url with the SHS host " + "added to url_allowlist. The SHS host and the YARN RM host " + "are usually different, so this is independent of yarn_rm_url." + ), + ) @field_validator("master") @classmethod diff --git a/spark_executor/tools/connections.py b/spark_executor/tools/connections.py index 4c631a7..421b979 100644 --- a/spark_executor/tools/connections.py +++ b/spark_executor/tools/connections.py @@ -26,6 +26,7 @@ def save_connection( auth_principal: str | None = _UNSET, # type: ignore[assignment] auth_keytab: str | None = _UNSET, # type: ignore[assignment] url_allowlist: list[str] | None = _UNSET, # type: ignore[assignment] + history_server_url: str | None = _UNSET, # type: ignore[assignment] ) -> dict[str, object]: logger.debug( f"save_connection enter name={name} master={master} " @@ -56,6 +57,8 @@ def save_connection( fields["auth_keytab"] = auth_keytab if url_allowlist is not _UNSET: fields["url_allowlist"] = url_allowlist or [] + if history_server_url is not _UNSET: + fields["history_server_url"] = history_server_url return update_connection(name, **fields) new_fields: dict[str, object] = { "name": name, @@ -71,6 +74,7 @@ def save_connection( "auth_principal": auth_principal if auth_principal is not _UNSET else None, "auth_keytab": auth_keytab if auth_keytab is not _UNSET else None, "url_allowlist": (url_allowlist if url_allowlist is not _UNSET else None) or [], + "history_server_url": history_server_url if history_server_url is not _UNSET else None, } conn = Connection(**new_fields) store.save(conn) @@ -86,7 +90,7 @@ def update_connection(name: str, **fields) -> dict[str, object]: Mutable fields: master, deploy_mode, yarn_rm_url, spark_conf, ssl_verify, ssl_ca_bundle, auth_type, auth_user, auth_password, - auth_principal, auth_keytab, url_allowlist. + auth_principal, auth_keytab, url_allowlist, history_server_url. """ logger.debug(f"update_connection enter name={name} fields={sorted(fields.keys())}") return store.update(name, **fields).model_dump() diff --git a/spark_executor/tools/requests.py b/spark_executor/tools/requests.py index 39d3c46..b55c0af 100644 --- a/spark_executor/tools/requests.py +++ b/spark_executor/tools/requests.py @@ -134,6 +134,14 @@ class SaveConnectionRequest(BaseModel): "populated via update_connection." ), ) + history_server_url: str | None = Field( + default=None, + description=( + "Optional Spark History Server (SHS) base URL, e.g. " + "'http://history.prod.internal:18080'. Stored for reference. " + "Currently not consumed by any tool — see Connection.history_server_url." + ), + ) class PrepareSubmitJobRequest(BaseModel): connection: str = Field( @@ -477,6 +485,15 @@ class UpdateConnectionRequest(BaseModel): "Example: ['ccam*'] allows ccam1-ccam99." ), ) + history_server_url: str | None = Field( + default=None, + description=( + "New Spark History Server base URL. Omit to keep current. " + "Pass None to clear (use delete_connection + save_connection " + "if you need explicit clear semantics; same caveat as other " + "Optional fields in this tool)." + ), + ) class ListApplicationsRequest(BaseModel): diff --git a/tests/unit/test_connection_tools.py b/tests/unit/test_connection_tools.py index dad74c7..570ab96 100644 --- a/tests/unit/test_connection_tools.py +++ b/tests/unit/test_connection_tools.py @@ -195,3 +195,39 @@ def test_update_connection_persists_to_disk(_fresh_store, tmp_path): connections.update_connection(name="prod", master="spark://new:7077") raw = json.loads((tmp_path / "connections.json").read_text()) assert raw["prod"]["master"] == "spark://new:7077" + + +# --- history_server_url --- + +def test_save_connection_with_history_server_url(_fresh_store): + connections.save_connection( + name="prod", + master="yarn", + history_server_url="http://history.prod.internal:18080", + ) + assert _fresh_store.get("prod").history_server_url == "http://history.prod.internal:18080" + + +def test_save_connection_history_server_url_defaults_to_none(_fresh_store): + connections.save_connection(name="prod", master="yarn") + assert _fresh_store.get("prod").history_server_url is None + + +def test_update_connection_changes_history_server_url(_fresh_store): + connections.save_connection(name="prod", master="yarn") + out = connections.update_connection( + name="prod", + history_server_url="http://history2.prod.internal:18080", + ) + assert out["history_server_url"] == "http://history2.prod.internal:18080" + assert _fresh_store.get("prod").history_server_url == "http://history2.prod.internal:18080" + + +def test_update_connection_keeps_history_server_url_when_omitted(_fresh_store): + connections.save_connection( + name="prod", + master="yarn", + history_server_url="http://history.prod.internal:18080", + ) + out = connections.update_connection(name="prod", master="spark://new:7077") + assert out["history_server_url"] == "http://history.prod.internal:18080"