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"