feat: structured DEBUG/INFO logging via loguru

- common/logging.py: improve format to timestamp|LEVEL|module:func:line - message
- core/ layer: DEBUG log every subprocess invocation (cmd, rc, byte counts),
  JSON load/dump events, parsed application_id. ERROR log on failures.
- tools/ layer: DEBUG log every public tool entry with key parameters,
  INFO log on business outcomes (saved/submitted/killed/...).
- New tests/unit/test_logging.py: capture loguru output via in-memory sink
  and assert DEBUG + INFO messages are emitted for representative flows.

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")
+logger.add(
+    sys.stderr,
+    level="DEBUG",
+    format=(
+        "<green>{time:HH:mm:ss.SSS}</green> | "
+        "<level>{level: <7}</level> | "
+        "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
+        "<level>{message}</level>"
+    ),
+)

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.

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

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.

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}"
         )

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")

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"}

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,

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

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)

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"}

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