From 1cd31285229dfe32664ff09b347e1d4b43279add Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 24 Jun 2026 14:43:56 +0800 Subject: [PATCH] feat: add get_job_logs tool with tail --- spark_executor/tools/logs.py | 23 ++++++++++++++++++ tests/unit/test_logs_tool.py | 47 ++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+) create mode 100644 spark_executor/tools/logs.py create mode 100644 tests/unit/test_logs_tool.py diff --git a/spark_executor/tools/logs.py b/spark_executor/tools/logs.py new file mode 100644 index 0000000..ba0f8e0 --- /dev/null +++ b/spark_executor/tools/logs.py @@ -0,0 +1,23 @@ +# coding=utf-8 +""" +@Time :2026/6/24 +@Author :tao.chen +""" +from common.logging import logger +from spark_executor.core.job_store import JobStore +from spark_executor.core.yarn_client import get_application_logs + +store = JobStore() + + +def get_job_logs(job_id: str, tail_chars: int = 5000) -> str: + job = store.get(job_id) + if job is None: + raise KeyError(f"Unknown job_id: {job_id}") + full = get_application_logs(job.application_id) + tailed = full[-tail_chars:] if len(full) > tail_chars else full + logger.info( + f"get_job_logs job_id={job_id} application_id={job.application_id} " + f"chars={len(tailed)}" + ) + return tailed diff --git a/tests/unit/test_logs_tool.py b/tests/unit/test_logs_tool.py new file mode 100644 index 0000000..3c349de --- /dev/null +++ b/tests/unit/test_logs_tool.py @@ -0,0 +1,47 @@ +# coding=utf-8 +from datetime import datetime +from unittest.mock import patch + +from spark_executor.core.job_store import JobStore +from spark_executor.models import Job +from spark_executor.tools import logs + + +def _seed(job_id="abc", app_id="application_1"): + logs.store = JobStore() + logs.store.put( + Job( + job_id=job_id, + application_id=app_id, + script_path="/tmp/j.py", + queue="default", + submit_time=datetime(2026, 6, 24), + connection="prod", + ) + ) + + +def test_get_job_logs_tails_to_default_5000(): + _seed() + big = "x" * 8000 + "\nEND" + with patch("spark_executor.tools.logs.get_application_logs", return_value=big): + out = logs.get_job_logs("abc") + assert out.endswith("END") + assert len(out) == 5000 + + +def test_get_job_logs_respects_custom_tail_chars(): + _seed() + with patch( + "spark_executor.tools.logs.get_application_logs", + return_value="0123456789", + ): + out = logs.get_job_logs("abc", tail_chars=3) + assert out == "789" + + +def test_get_job_logs_raises_for_unknown_job(): + _seed() + import pytest + with pytest.raises(KeyError): + logs.get_job_logs("missing")