diff --git a/common/logging.py b/common/logging.py
index 1325ddf..50bc3f2 100644
--- a/common/logging.py
+++ b/common/logging.py
@@ -1,10 +1,29 @@
# coding=utf-8
"""
@Time :2026/6/24
-@Author :tao.chen
+@Author :tao.chen
+
+Process-wide loguru configuration. Import `logger` from here in every
+module instead of instantiating new loggers.
+
+Levels used in this project:
+ DEBUG - entry/exit of public tools, subprocess commands, file I/O paths
+ INFO - business events (job submitted, status changed, connection saved)
+ WARNING - recoverable problems (transient YARN issues, retry-able)
+ ERROR - raised exceptions (caller will see the traceback)
"""
import sys
+
from loguru import logger
logger.remove()
-logger.add(sys.stderr, level="DEBUG")
\ No newline at end of file
+logger.add(
+ sys.stderr,
+ level="DEBUG",
+ format=(
+ "{time:HH:mm:ss.SSS} | "
+ "{level: <7} | "
+ "{name}:{function}:{line} - "
+ "{message}"
+ ),
+)
diff --git a/spark_executor/core/connection_store.py b/spark_executor/core/connection_store.py
index 503f67b..15729a5 100644
--- a/spark_executor/core/connection_store.py
+++ b/spark_executor/core/connection_store.py
@@ -9,6 +9,7 @@ import tempfile
from pathlib import Path
from threading import Lock
+from common.logging import logger
from spark_executor.models import Connection
DEFAULT_DATA_DIR = os.environ.get("SPARK_EXECUTOR_DATA_DIR", "./data")
@@ -37,8 +38,10 @@ class ConnectionStore:
def _load(self) -> dict[str, Connection]:
if not self.path.exists():
+ logger.debug(f"ConnectionStore._load: file {self.path} absent, returning empty")
return {}
raw = json.loads(self.path.read_text(encoding="utf-8"))
+ logger.debug(f"ConnectionStore._load: loaded {len(raw)} records from {self.path}")
return {name: Connection.model_validate(c) for name, c in raw.items()}
def _dump(self, records: dict[str, Connection]) -> None:
@@ -52,6 +55,7 @@ class ConnectionStore:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, ensure_ascii=False)
os.replace(tmp_path, self.path)
+ logger.debug(f"ConnectionStore._dump: wrote {len(records)} records to {self.path}")
except Exception:
if os.path.exists(tmp_path):
os.unlink(tmp_path)
@@ -70,6 +74,7 @@ class ConnectionStore:
records = self._load()
records[conn.name] = conn
self._dump(records)
+ logger.info(f"connection saved name={conn.name} master={conn.master}")
def delete(self, name: str) -> bool:
with self._lock:
@@ -78,7 +83,8 @@ class ConnectionStore:
return False
del records[name]
self._dump(records)
- return True
+ logger.info(f"connection deleted name={name}")
+ return True
# Module-level singleton; replaced in tests.
diff --git a/spark_executor/core/log_parser.py b/spark_executor/core/log_parser.py
index ad1b72e..306966b 100644
--- a/spark_executor/core/log_parser.py
+++ b/spark_executor/core/log_parser.py
@@ -5,6 +5,8 @@
"""
import re
+from common.logging import logger
+
_APP_ID_RE = re.compile(r"Submitted application (\S+)")
_TRACKING_URL_RE = re.compile(r"tracking URL:\s+(\S+)")
@@ -20,14 +22,20 @@ def parse_spark_submit_output(stderr: str) -> tuple[str, str | None]:
else:
url_match = _TRACKING_URL_RE.search(stderr)
if not url_match:
+ logger.error("Could not find application_id in spark-submit output")
raise ValueError("Could not find application_id in spark-submit output")
# tracking URL is of the form http://rm:8088/proxy/application_xxx/
tracking_url = url_match.group(1)
tail = tracking_url.rstrip("/").rsplit("/", 1)[-1]
if not tail.startswith("application_"):
+ logger.error("Could not find application_id in spark-submit output (bad tracking URL)")
raise ValueError("Could not find application_id in spark-submit output")
application_id = tail
url_match = _TRACKING_URL_RE.search(stderr)
tracking_url = url_match.group(1) if url_match else None
+ logger.debug(
+ f"parse_spark_submit_output -> application_id={application_id} "
+ f"tracking_url={tracking_url}"
+ )
return application_id, tracking_url
diff --git a/spark_executor/core/pending_store.py b/spark_executor/core/pending_store.py
index 661229e..8487881 100644
--- a/spark_executor/core/pending_store.py
+++ b/spark_executor/core/pending_store.py
@@ -9,6 +9,7 @@ import tempfile
from pathlib import Path
from threading import Lock
+from common.logging import logger
from spark_executor.models import PendingSubmission
DEFAULT_DATA_DIR = os.environ.get("SPARK_EXECUTOR_DATA_DIR", "./data")
@@ -33,8 +34,10 @@ class PendingStore:
def _load(self) -> dict[str, PendingSubmission]:
if not self.path.exists():
+ logger.debug(f"PendingStore._load: file {self.path} absent, returning empty")
return {}
raw = json.loads(self.path.read_text(encoding="utf-8"))
+ logger.debug(f"PendingStore._load: loaded {len(raw)} records from {self.path}")
return {pid: PendingSubmission.model_validate(p) for pid, p in raw.items()}
def _dump(self, records: dict[str, PendingSubmission]) -> None:
@@ -47,6 +50,7 @@ class PendingStore:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(payload, f, indent=2, ensure_ascii=False, default=str)
os.replace(tmp_path, self.path)
+ logger.debug(f"PendingStore._dump: wrote {len(records)} records to {self.path}")
except Exception:
if os.path.exists(tmp_path):
os.unlink(tmp_path)
@@ -65,6 +69,10 @@ class PendingStore:
records = self._load()
records[pending.pending_id] = pending
self._dump(records)
+ logger.info(
+ f"pending saved pending_id={pending.pending_id} "
+ f"status={pending.status} connection={pending.connection}"
+ )
def delete(self, pending_id: str) -> bool:
with self._lock:
@@ -73,7 +81,8 @@ class PendingStore:
return False
del records[pending_id]
self._dump(records)
- return True
+ logger.info(f"pending deleted pending_id={pending_id}")
+ return True
# Module-level singleton; replaced in tests.
diff --git a/spark_executor/core/spark_submit.py b/spark_executor/core/spark_submit.py
index 559845d..d0f759d 100644
--- a/spark_executor/core/spark_submit.py
+++ b/spark_executor/core/spark_submit.py
@@ -5,6 +5,8 @@
"""
import subprocess
+from common.logging import logger
+
class SparkSubmitError(Exception):
"""Raised when spark-submit exits with a non-zero return code."""
@@ -33,12 +35,19 @@ def build_spark_submit_command(
for key, value in (spark_conf or {}).items():
cmd.extend(["--conf", f"{key}={value}"])
cmd.append(script_path)
+ logger.debug(f"build_spark_submit_command -> {cmd}")
return cmd
def run_spark_submit(cmd: list[str]) -> "subprocess.CompletedProcess[str]":
+ logger.debug(f"run_spark_submit exec: {cmd}")
result = subprocess.run(cmd, capture_output=True, text=True, errors="replace")
+ logger.debug(
+ f"run_spark_submit done rc={result.returncode} "
+ f"stdout_len={len(result.stdout)} stderr_len={len(result.stderr)}"
+ )
if result.returncode != 0:
+ logger.error(f"spark-submit failed (rc={result.returncode}): {result.stderr[:500]}")
raise SparkSubmitError(
f"spark-submit failed (rc={result.returncode}): {result.stderr}"
)
diff --git a/spark_executor/core/yarn_client.py b/spark_executor/core/yarn_client.py
index 15bcec1..89ab071 100644
--- a/spark_executor/core/yarn_client.py
+++ b/spark_executor/core/yarn_client.py
@@ -6,6 +6,8 @@
import re
import subprocess
+from common.logging import logger
+
class YarnError(Exception):
"""Raised when a yarn CLI invocation fails."""
@@ -15,33 +17,54 @@ _STATE_RE = re.compile(r"State\s*:\s*(\S+)")
def _run(cmd: list[str]) -> "subprocess.CompletedProcess[str]":
- return subprocess.run(cmd, capture_output=True, text=True, errors="replace")
+ logger.debug(f"yarn _run exec: {cmd}")
+ result = subprocess.run(cmd, capture_output=True, text=True, errors="replace")
+ logger.debug(
+ f"yarn _run done rc={result.returncode} "
+ f"stdout_len={len(result.stdout)} stderr_len={len(result.stderr)}"
+ )
+ return result
def get_application_status(application_id: str) -> tuple[str, str]:
proc = _run(["yarn", "application", "-status", application_id])
if proc.returncode != 0:
+ logger.error(
+ f"yarn application -status failed (rc={proc.returncode}) for "
+ f"{application_id}: {proc.stderr[:500]}"
+ )
raise YarnError(
f"yarn application -status failed (rc={proc.returncode}): {proc.stderr}"
)
match = _STATE_RE.search(proc.stdout)
if not match:
raise YarnError(f"Could not parse YARN state from output: {proc.stdout!r}")
- return match.group(1), proc.stdout
+ state = match.group(1)
+ logger.info(f"yarn status {application_id} -> {state}")
+ return state, proc.stdout
def get_application_logs(application_id: str) -> str:
proc = _run(["yarn", "logs", "-applicationId", application_id])
if proc.returncode != 0:
+ logger.error(
+ f"yarn logs failed (rc={proc.returncode}) for {application_id}: {proc.stderr[:500]}"
+ )
raise YarnError(
f"yarn logs failed (rc={proc.returncode}): {proc.stderr}"
)
+ logger.info(f"yarn logs {application_id} -> {len(proc.stdout)} chars")
return proc.stdout
def kill_application(application_id: str) -> None:
proc = _run(["yarn", "application", "-kill", application_id])
if proc.returncode != 0:
+ logger.error(
+ f"yarn application -kill failed (rc={proc.returncode}) for "
+ f"{application_id}: {proc.stderr[:500]}"
+ )
raise YarnError(
f"yarn application -kill failed (rc={proc.returncode}): {proc.stderr}"
)
+ logger.info(f"yarn kill {application_id} -> ok")
diff --git a/spark_executor/tools/connections.py b/spark_executor/tools/connections.py
index ccff910..f76c8f9 100644
--- a/spark_executor/tools/connections.py
+++ b/spark_executor/tools/connections.py
@@ -16,6 +16,10 @@ def save_connection(
yarn_rm_url: str | None = None,
spark_conf: dict[str, str] | None = None,
) -> dict[str, str]:
+ logger.debug(
+ f"save_connection enter name={name} master={master} deploy_mode={deploy_mode} "
+ f"yarn_rm_url={yarn_rm_url} spark_conf_keys={list((spark_conf or {}).keys())}"
+ )
conn = Connection(
name=name,
master=master,
@@ -24,15 +28,16 @@ def save_connection(
spark_conf=spark_conf or {},
)
store.save(conn)
- logger.info(f"save_connection name={name} master={master}")
return {"name": name, "status": "SAVED"}
def list_connections() -> list[dict[str, object]]:
+ logger.debug("list_connections enter")
return [c.model_dump() for c in store.list_all()]
def get_connection(name: str) -> dict[str, object]:
+ logger.debug(f"get_connection enter name={name}")
conn = store.get(name)
if conn is None:
raise KeyError(f"Unknown connection: {name}")
@@ -40,8 +45,8 @@ def get_connection(name: str) -> dict[str, object]:
def delete_connection(name: str) -> dict[str, str]:
+ logger.debug(f"delete_connection enter name={name}")
removed = store.delete(name)
if not removed:
raise KeyError(f"Unknown connection: {name}")
- logger.info(f"delete_connection name={name}")
return {"name": name, "status": "DELETED"}
diff --git a/spark_executor/tools/kill.py b/spark_executor/tools/kill.py
index da692c1..8d9409e 100644
--- a/spark_executor/tools/kill.py
+++ b/spark_executor/tools/kill.py
@@ -11,11 +11,12 @@ store = JobStore()
def kill_job(job_id: str) -> dict[str, str]:
+ logger.debug(f"kill_job enter job_id={job_id}")
job = store.get(job_id)
if job is None:
raise KeyError(f"Unknown job_id: {job_id}")
kill_application(job.application_id)
- logger.info(f"kill_job job_id={job_id} application_id={job.application_id}")
+ logger.info(f"kill_job ok job_id={job_id} application_id={job.application_id}")
return {
"job_id": job_id,
"application_id": job.application_id,
diff --git a/spark_executor/tools/logs.py b/spark_executor/tools/logs.py
index ba0f8e0..2834878 100644
--- a/spark_executor/tools/logs.py
+++ b/spark_executor/tools/logs.py
@@ -11,13 +11,14 @@ store = JobStore()
def get_job_logs(job_id: str, tail_chars: int = 5000) -> str:
+ logger.debug(f"get_job_logs enter job_id={job_id} tail_chars={tail_chars}")
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)}"
+ f"get_job_logs ok job_id={job_id} application_id={job.application_id} "
+ f"full_chars={len(full)} returned_chars={len(tailed)}"
)
return tailed
diff --git a/spark_executor/tools/status.py b/spark_executor/tools/status.py
index 048ff1a..17b2f86 100644
--- a/spark_executor/tools/status.py
+++ b/spark_executor/tools/status.py
@@ -12,9 +12,10 @@ store = JobStore()
def get_job_status(job_id: str) -> JobStatus:
+ logger.debug(f"get_job_status enter job_id={job_id}")
job = store.get(job_id)
if job is None:
raise KeyError(f"Unknown job_id: {job_id}")
state, raw = get_application_status(job.application_id)
- logger.info(f"get_job_status job_id={job_id} state={state}")
+ logger.info(f"get_job_status ok job_id={job_id} application_id={job.application_id} state={state}")
return JobStatus(application_id=job.application_id, state=state, raw=raw)
diff --git a/spark_executor/tools/submit.py b/spark_executor/tools/submit.py
index 8e09d12..db72311 100644
--- a/spark_executor/tools/submit.py
+++ b/spark_executor/tools/submit.py
@@ -4,6 +4,7 @@
@Author :tao.chen
"""
import secrets
+import uuid
from datetime import datetime
from common.logging import logger
@@ -17,8 +18,6 @@ from spark_executor.core.spark_submit import (
run_spark_submit,
)
from spark_executor.models import Job, PendingSubmission, SubmitResult
-import uuid
-from datetime import datetime as _datetime
def _new_pending_id() -> str:
@@ -35,6 +34,11 @@ def prepare_submit_job(
num_executors: int = 2,
) -> dict[str, object]:
"""Snapshot connection params and persist a PendingSubmission. Does NOT submit."""
+ logger.debug(
+ f"prepare_submit_job enter connection={connection} script_path={script_path} "
+ f"queue={queue} executor_memory={executor_memory} executor_cores={executor_cores} "
+ f"num_executors={num_executors}"
+ )
conn = conn_store.get(connection)
if conn is None:
raise KeyError(f"Unknown connection: {connection}")
@@ -56,7 +60,8 @@ def prepare_submit_job(
)
pending_store.save(pending)
logger.info(
- f"prepare_submit_job pending_id={pending_id} connection={connection} master={conn.master}"
+ f"prepare_submit_job ok pending_id={pending_id} connection={connection} "
+ f"master={conn.master} script_path={script_path}"
)
return {
"pending_id": pending_id,
@@ -71,6 +76,7 @@ job_store: JobStore = JobStore()
def confirm_submit_job(*, pending_id: str) -> SubmitResult:
"""Actually invoke spark-submit for a previously-prepared PendingSubmission."""
+ logger.debug(f"confirm_submit_job enter pending_id={pending_id}")
pending = pending_store.get(pending_id)
if pending is None:
raise KeyError(f"Unknown pending_id: {pending_id}")
@@ -89,13 +95,17 @@ def confirm_submit_job(*, pending_id: str) -> SubmitResult:
num_executors=pending.num_executors,
spark_conf=pending.spark_conf,
)
- logger.info(f"confirm_submit_job pending_id={pending_id} cmd={cmd}")
+ logger.info(
+ f"confirm_submit_job start pending_id={pending_id} "
+ f"application_target={pending.master} script_path={pending.script_path}"
+ )
try:
result = run_spark_submit(cmd)
except SparkSubmitError as exc:
pending.status = "FAILED"
pending.error = str(exc)
pending_store.save(pending)
+ logger.error(f"confirm_submit_job failed pending_id={pending_id} err={exc}")
raise
application_id, tracking_url = parse_spark_submit_output(result.stderr)
@@ -107,7 +117,7 @@ def confirm_submit_job(*, pending_id: str) -> SubmitResult:
application_id=application_id,
script_path=pending.script_path,
queue=pending.queue,
- submit_time=_datetime.utcnow(),
+ submit_time=datetime.utcnow(),
connection=pending.connection,
)
)
@@ -117,7 +127,8 @@ def confirm_submit_job(*, pending_id: str) -> SubmitResult:
pending.application_id = application_id
pending_store.save(pending)
logger.info(
- f"confirm_submit_job pending_id={pending_id} job_id={job_id} application_id={application_id}"
+ f"confirm_submit_job ok pending_id={pending_id} job_id={job_id} "
+ f"application_id={application_id}"
)
return SubmitResult(
job_id=job_id,
@@ -127,10 +138,12 @@ def confirm_submit_job(*, pending_id: str) -> SubmitResult:
def list_pending_jobs() -> list[dict[str, object]]:
+ logger.debug("list_pending_jobs enter")
return [p.model_dump() for p in pending_store.list_all()]
def get_pending_job(pending_id: str) -> dict[str, object]:
+ logger.debug(f"get_pending_job enter pending_id={pending_id}")
p = pending_store.get(pending_id)
if p is None:
raise KeyError(f"Unknown pending_id: {pending_id}")
@@ -138,6 +151,7 @@ def get_pending_job(pending_id: str) -> dict[str, object]:
def cancel_pending_job(pending_id: str) -> dict[str, str]:
+ logger.debug(f"cancel_pending_job enter pending_id={pending_id}")
p = pending_store.get(pending_id)
if p is None:
raise KeyError(f"Unknown pending_id: {pending_id}")
@@ -147,5 +161,5 @@ def cancel_pending_job(pending_id: str) -> dict[str, str]:
)
p.status = "CANCELLED"
pending_store.save(p)
- logger.info(f"cancel_pending_job pending_id={pending_id}")
+ logger.info(f"cancel_pending_job ok pending_id={pending_id}")
return {"pending_id": pending_id, "status": "CANCELLED"}
diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py
new file mode 100644
index 0000000..44b746b
--- /dev/null
+++ b/tests/unit/test_logging.py
@@ -0,0 +1,63 @@
+# coding=utf-8
+import io
+from pathlib import Path
+
+import pytest
+
+from spark_executor.core import connection_store, pending_store
+from spark_executor.core.connection_store import ConnectionStore
+from spark_executor.core.pending_store import PendingStore
+from spark_executor.models import Connection
+from spark_executor.tools import connections, submit
+
+
+@pytest.fixture(autouse=True)
+def _fresh_stores(tmp_path: Path, monkeypatch):
+ monkeypatch.setattr(connection_store, "DEFAULT_DATA_DIR", str(tmp_path))
+ monkeypatch.setattr(connection_store, "store", ConnectionStore())
+ monkeypatch.setattr(pending_store, "DEFAULT_DATA_DIR", str(tmp_path))
+ monkeypatch.setattr(pending_store, "store", PendingStore())
+ connections.store = connection_store.store
+ submit.conn_store = connection_store.store
+ submit.pending_store = pending_store.store
+ submit.job_store = submit.job_store.__class__() # fresh in-memory job store
+
+
+@pytest.fixture
+def log_capture():
+ """Attach an in-memory sink to loguru so tests can assert on emitted lines."""
+ from common.logging import logger
+ buf = io.StringIO()
+ handler_id = logger.add(buf, level="DEBUG", format="{level}|{message}")
+ yield buf
+ logger.remove(handler_id)
+
+
+def test_save_connection_emits_info_log(log_capture):
+ connections.save_connection(name="prod", master="yarn")
+ text = log_capture.getvalue()
+ assert "INFO" in text
+ assert "save_connection enter" in text
+ assert "DEBUG" in text
+ assert "connection saved" in text
+
+
+def test_prepare_submit_job_emits_debug_and_info(log_capture):
+ connections.save_connection(name="prod", master="yarn", deploy_mode="cluster")
+ log_capture.truncate(0); log_capture.seek(0)
+ submit.prepare_submit_job(connection="prod", script_path="/tmp/j.py", queue="research")
+ text = log_capture.getvalue()
+ assert "DEBUG|prepare_submit_job enter" in text
+ assert "INFO|prepare_submit_job ok" in text
+ assert "script_path=/tmp/j.py" in text
+ assert "queue=research" in text
+
+
+def test_get_unknown_pending_job_emits_debug(log_capture):
+ log_capture.truncate(0); log_capture.seek(0)
+ import pytest as _pytest
+ with _pytest.raises(KeyError):
+ submit.get_pending_job("p_doesnotexist")
+ text = log_capture.getvalue()
+ assert "DEBUG|get_pending_job enter" in text
+ assert "p_doesnotexist" in text