[Jobs] Always run jobs controllers with debug logging

The jobs controller process inherits the launching client's log level via
SKYPILOT_DEBUG (through os.environ in consolidation mode, or the controller
env file in remote mode). When the client does not opt into debug logging,
the controller's own logs are emitted at INFO level and without timestamps.
This makes it hard to reconstruct incidents from controller logs (e.g.
determining when controllers re-claim jobs during failover).

Always run the controller with debug logging enabled, independent of the
client's log 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 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,17 @@ async def main(controller_uuid: str):
 
 
 if __name__ == '__main__':
+    # Always run the jobs controller with debug logging enabled, regardless of
+    # the launching client's log level. The controller process inherits the
+    # client's SKYPILOT_DEBUG (via os.environ in consolidation mode, or the
+    # controller env file in remote mode), which leaves the controller's own
+    # logs at INFO level and without timestamps when the client did not 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