[Jobs] Always run jobs controllers with debug logging

The jobs controller process inherits SKYPILOT_DEBUG from whichever process
starts it, and that inherited level is unreliable and rarely debug:
- Consolidation mode: the controller subprocess inherits the os.environ of
  the process that starts it. In steady state that is the managed-job refresh
  daemon (a thread in the main API server process), so it inherits the API
  server's level; when a `sky jobs launch` request worker starts it directly,
  it inherits the client's level.
- Remote mode: it comes from the controller env file (built from the
  launching client's level).

None of these is guaranteed to be debug, so the controller's own
("controller_system") logs are emitted at INFO level and without timestamps
unless someone happened to opt in. This makes it hard to reconstruct incidents
from controller logs (e.g. determining when controllers re-claim jobs during
a failover/rolling upgrade).

Always run the controller with debug logging enabled, independent of the
inherited level, in both consolidation and remote modes:

- At the controller process entry point (__main__), force SKYPILOT_DEBUG=1
  and reload the logger so the controller-process ("controller_system") logs
  get debug detail + timestamps. Setting the env var also propagates to
  subprocesses spawned by the controller.
- In the per-job loop, override SKYPILOT_DEBUG to '1' after applying the
  job's env (which carries the launching client's value), so per-job
  controller logs are also at debug level.

Both entry points run the same `python -m sky.jobs.controller` module, so
the fix covers consolidation-mode and remote controllers alike.

Tested:
- Repro'd in consolidation mode: controller_system log had no timestamps and
  no debug lines when the client did not set SKYPILOT_DEBUG.
- After fix, verified end-to-end on Kubernetes in both consolidation mode and
  remote-controller mode: both the controller_system log and per-job logs are
  emitted at debug level with timestamps even when the launching client did
  not set SKYPILOT_DEBUG (env file still records SKYPILOT_DEBUG='0').
- Added unit tests for the per-job env override helper.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FmmoQDXj9gwbXvS7PhXuLF

Author: cg505

sky/jobs/controller.py

@@ -49,6 +49,7 @@
 from sky.utils import context_utils
 from sky.utils import controller_utils
 from sky.utils import dag_utils
+from sky.utils import env_options
 from sky.utils import status_lib
 from sky.utils import ux_utils
 from sky.utils.plugin_extensions import ExternalClusterFailure
@@ -103,6 +104,25 @@ def _get_dag(job_id: int) -> 'sky.Dag':
     return dag
 
 
+def _apply_job_env_overrides(
+        ctx: context.SkyPilotContext,
+        env_vars: typing.Mapping[str, Optional[str]]) -> None:
+    """Apply a job's env vars to ``ctx``, always forcing debug logging on.
+
+    The per-job env carries the launching client's SKYPILOT_DEBUG (e.g. '0').
+    We override it to '1' after applying the env so the controller's per-job
+    logs always include debug detail + timestamps, regardless of the client's
+    log level. This mirrors the controller-process-level counterpart in the
+    ``__main__`` block. These logs are essential for reconstructing incidents
+    and the extra verbosity on the controller is an acceptable cost.
+    """
+    for key, value in env_vars.items():
+        if value is not None:
+            ctx.override_envs({key: value})
+            logger.debug('Set environment variable: %s=%s', key, value)
+    ctx.override_envs({env_options.Options.SHOW_DEBUG_INFO.env_key: '1'})
+
+
 def _add_k8s_annotations(task: 'sky.Task', job_id: int) -> None:
     """Adds Kubernetes pod config annotations to the task resources.
 
@@ -2148,11 +2168,7 @@ async def run_job_loop(self,
                 logger.info('Loading %d environment variables for job %s',
                             len(env_vars), job_id)
                 if ctx is not None:
-                    for key, value in env_vars.items():
-                        if value is not None:
-                            ctx.override_envs({key: value})
-                            logger.debug('Set environment variable: %s=%s', key,
-                                         value)
+                    _apply_job_env_overrides(ctx, env_vars)
 
                     # Restore config file if needed
                     file_content_utils.restore_job_config_file(job_id)
@@ -2534,4 +2550,24 @@ async def main(controller_uuid: str):
 
 
 if __name__ == '__main__':
+    # Always run the jobs controller with debug logging enabled, regardless of
+    # whatever SKYPILOT_DEBUG it would otherwise inherit. That inherited level
+    # is unreliable and rarely debug:
+    #   - Consolidation mode: the controller subprocess inherits the os.environ
+    #     of whichever process started it. In steady state that is the
+    #     managed-job refresh daemon (a thread in the main API server process),
+    #     so it inherits the *API server's* level; when a `sky jobs launch`
+    #     request worker starts it directly, it inherits the *client's* level.
+    #   - Remote mode: it comes from the controller env file (built from the
+    #     launching client's level).
+    # None of these is guaranteed to be debug, so the controller's own
+    # ("controller_system") logs land at INFO and without timestamps unless
+    # someone happened to opt in. Controller debug logs (timestamps + extra
+    # detail) are essential for reconstructing incidents -- e.g. determining
+    # when controllers re-claim jobs during failover -- and the extra verbosity
+    # on the controller is an acceptable cost. We set the env var (so
+    # subprocesses spawned by the controller inherit it) and reload the logger
+    # so the level and timestamped formatter take effect immediately.
+    os.environ[env_options.Options.SHOW_DEBUG_INFO.env_key] = '1'
+    sky_logging.reload_logger()
     asyncio.run(main(sys.argv[1]))

tests/unit_tests/test_sky/jobs/test_controller.py

@@ -9,6 +9,7 @@
 and file mount cleanup in task_cleanup().
 """
 import asyncio
+import os
 from typing import Dict, List, Optional, Tuple
 from unittest.mock import AsyncMock
 from unittest.mock import MagicMock
@@ -17,8 +18,11 @@
 import pytest
 
 from sky.jobs import state as managed_job_state
+from sky.jobs.controller import _apply_job_env_overrides
 from sky.jobs.controller import ControllerManager
 from sky.utils import common
+from sky.utils import context
+from sky.utils import env_options
 
 
 class TestNormalJobRecovery:
@@ -1077,3 +1081,53 @@ def download_side_effect(task_id, handle, job_id_on_pool):
                 0, mock_handle_0, None)
             controller.download_log_and_stream.assert_any_call(
                 1, mock_handle_1, None)
+
+
+class TestApplyJobEnvOverrides:
+    """Tests that the controller always forces debug logging for the per-job
+    loop, regardless of the launching client's log level.
+
+    The per-job env carries the client's SKYPILOT_DEBUG value (e.g. '0' when
+    the client did not opt into debug logging). The controller must override
+    this to '1' so the per-job controller logs always include debug detail and
+    timestamps -- essential for reconstructing incidents.
+    """
+
+    @pytest.fixture
+    def ctx(self):
+        context.initialize()
+        yield context.get()
+        if context.get()._log_file_handle is not None:
+            context.get()._log_file_handle.close()
+
+    def test_forces_debug_when_client_disabled_it(self, ctx):
+        """SKYPILOT_DEBUG='0' from the client is overridden to '1'."""
+        debug_key = env_options.Options.SHOW_DEBUG_INFO.env_key
+        _apply_job_env_overrides(ctx, {debug_key: '0', 'FOO': 'bar'})
+        # Other env vars are passed through unchanged.
+        assert ctx.env_overrides['FOO'] == 'bar'
+        # Debug is forced on regardless of the client's value.
+        assert ctx.env_overrides[debug_key] == '1'
+
+    def test_forces_debug_when_client_omitted_it(self, ctx):
+        """Debug is forced on even when the env has no SKYPILOT_DEBUG."""
+        debug_key = env_options.Options.SHOW_DEBUG_INFO.env_key
+        _apply_job_env_overrides(ctx, {'FOO': 'bar'})
+        assert ctx.env_overrides[debug_key] == '1'
+
+    def test_skips_none_values(self, ctx):
+        """None values (unset in the env file) are not applied."""
+        _apply_job_env_overrides(ctx, {'FOO': None})
+        assert 'FOO' not in ctx.env_overrides
+
+    def test_debug_visible_through_contextual_environ(self, ctx):
+        """With the context active, SHOW_DEBUG_INFO.get() reflects the
+        forced value even though the client supplied '0'."""
+        debug_key = env_options.Options.SHOW_DEBUG_INFO.env_key
+        _apply_job_env_overrides(ctx, {debug_key: '0'})
+        original_environ = os.environ
+        try:
+            os.environ = context.ContextualEnviron(original_environ)
+            assert env_options.Options.SHOW_DEBUG_INFO.get() is True
+        finally:
+            os.environ = original_environ