: logging: default disable log forwarding (#1878)

Summary:

this diff sets the following defaults for hyperactor-mesh global configuration values:
- `MESH_ENABLE_LOG_FORWARDING=false`
- `MESH_ENABLE_FILE_CAPTURE=false`
- `MESH_TAIL_LOG_LINES=0`

the effect of this is to disable log forwarding, prevent allocating resources for log forwarding, no file capture at the hyperactor mesh level (including no "exit tail" capture), in fact, under these defaults there is no interception of child process stdio at all.

a [workplace post is planned to announce this change](https://fb.workplace.com/groups/1399849971389924/permalink/1602002844507968/) in default configuration.

this diff is built on: D85783397 provide config for enabling/disabling hyperactor-mesh logging interception features, D85919326 for avoiding spinning up `LogForwardActor` meshes when log forwarding is enabled and D85969320 for some detailed testing.

Reviewed By: zdevito, vidhyav

Differential Revision: D86994420

Author: shayne-fletcher

hyperactor_mesh/src/bootstrap.rs

@@ -99,7 +99,7 @@ declare_attrs! {
         env_name: Some("HYPERACTOR_MESH_ENABLE_LOG_FORWARDING".to_string()),
         py_name: None,
     })
-    pub attr MESH_ENABLE_LOG_FORWARDING: bool = true;
+    pub attr MESH_ENABLE_LOG_FORWARDING: bool = false;
 
     /// When `true`: if stdio is piped, each child's `StreamFwder`
     /// also forwards lines to a host-scoped `FileAppender` managed by
@@ -124,7 +124,7 @@ declare_attrs! {
         env_name: Some("HYPERACTOR_MESH_ENABLE_FILE_CAPTURE".to_string()),
         py_name: None,
     })
-    pub attr MESH_ENABLE_FILE_CAPTURE: bool = true;
+    pub attr MESH_ENABLE_FILE_CAPTURE: bool = false;
 
     /// Maximum number of log lines retained in a proc's stderr/stdout
     /// tail buffer. Used by [`StreamFwder`] when wiring child
@@ -133,7 +133,7 @@ declare_attrs! {
         env_name: Some("HYPERACTOR_MESH_TAIL_LOG_LINES".to_string()),
         py_name: None,
     })
-    pub attr MESH_TAIL_LOG_LINES: usize = 100;
+    pub attr MESH_TAIL_LOG_LINES: usize = 0;
 
     /// If enabled (default), bootstrap child processes install
     /// `PR_SET_PDEATHSIG(SIGKILL)` so the kernel reaps them if the
@@ -3692,6 +3692,10 @@ mod tests {
     #[tokio::test]
     async fn exit_tail_is_attached_and_logged() {
         hyperactor_telemetry::initialize_logging_for_test();
+
+        let lock = hyperactor::config::global::lock();
+        let _guard = lock.override_key(MESH_TAIL_LOG_LINES, 100);
+
         // Spawn a child that writes to stderr then exits 7.
         let mut cmd = Command::new("sh");
         cmd.arg("-c")

hyperactor_mesh/src/v1/host_mesh.rs

@@ -1125,6 +1125,7 @@ mod tests {
 
     use super::*;
     use crate::Bootstrap;
+    use crate::bootstrap::MESH_TAIL_LOG_LINES;
     use crate::resource::Status;
     use crate::v1::ActorMesh;
     use crate::v1::testactor;
@@ -1321,6 +1322,9 @@ mod tests {
     #[tokio::test]
     #[cfg(fbcode_build)]
     async fn test_failing_proc_allocation() {
+        let lock = hyperactor::config::global::lock();
+        let _guard = lock.override_key(MESH_TAIL_LOG_LINES, 100);
+
         let program = crate::testresource::get("monarch/hyperactor_mesh/bootstrap");
 
         let hosts = vec![free_localhost_addr(), free_localhost_addr()];

monarch_hyperactor/src/v1/logging.rs

@@ -310,9 +310,9 @@ impl LoggingMeshClient {
             // re-spawning infra, which we deliberately don't do at
             // runtime.
             (None, true) => {
-                return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
-                    "log forwarding disabled by config at startup; cannot enable streaming_to_client",
-                ));
+                // return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
+                //     "log forwarding disabled by config at startup; cannot enable streaming_to_client",
+                // ));
             }
         }
 
@@ -592,6 +592,9 @@ mod tests {
                     );
                 }
 
+                /*
+                // Update (SF: 2025, 11, 13): We now ignore stream to client requests if
+                // log forwarding is enabled.
                 // (c) stream_to_client = true when forwarding was
                 //     never spawned -> Err
                 let res = client_ref.set_mode(&py_instance, true, None, 10);
@@ -606,6 +609,7 @@ mod tests {
                         "unexpected err when enabling streaming with no forwarders: {msg}"
                     );
                 }
+                */
             });
 
             drop(client_py); // See note "NOTE ON LIFECYCLE / CLEANUP"

python/monarch/_src/actor/v1/proc_mesh.py

@@ -365,8 +365,8 @@ def rank_tensors(self) -> Dict[str, "Tensor"]:
 
     async def logging_option(
         self,
-        stream_to_client: bool = True,
-        aggregate_window_sec: int | None = 3,
+        stream_to_client: bool = False,
+        aggregate_window_sec: int | None = None,
         level: int = logging.INFO,
     ) -> None:
         """

python/tests/test_allocator.py

@@ -254,9 +254,19 @@ async def test_allocate_failure_message(self) -> None:
             r"exited with code 1: Traceback \(most recent call last\).*",
         ):
             with remote_process_allocator(
-                envs={"MONARCH_ERROR_DURING_BOOTSTRAP_FOR_TESTING": "1"}
+                envs={
+                    "MONARCH_ERROR_DURING_BOOTSTRAP_FOR_TESTING": "1",
+                    "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+                    "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+                    "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+                }
             ) as host1, remote_process_allocator(
-                envs={"MONARCH_ERROR_DURING_BOOTSTRAP_FOR_TESTING": "1"}
+                envs={
+                    "MONARCH_ERROR_DURING_BOOTSTRAP_FOR_TESTING": "1",
+                    "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+                    "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+                    "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+                }
             ) as host2:
                 allocator = RemoteAllocator(
                     world_id="test_remote_allocator",

python/tests/test_python_actors.py

@@ -547,6 +547,16 @@ def _handle_undeliverable_message(
 
 @pytest.mark.timeout(60)
 async def test_actor_log_streaming() -> None:
+    old_env = {}
+    env_vars = {
+        "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+        "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+        "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+    }
+    for key, value in env_vars.items():
+        old_env[key] = os.environ.get(key)
+        os.environ[key] = value
+
     # Save original file descriptors
     original_stdout_fd = os.dup(1)  # stdout
     original_stderr_fd = os.dup(2)  # stderr
@@ -684,6 +694,12 @@ async def test_actor_log_streaming() -> None:
         ), stderr_content
 
     finally:
+        for key, value in old_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+
         # Ensure file descriptors are restored even if something goes wrong
         try:
             os.dup2(original_stdout_fd, 1)
@@ -699,6 +715,16 @@ async def test_alloc_based_log_streaming() -> None:
     """Test both AllocHandle.stream_logs = False and True cases."""
 
     async def test_stream_logs_case(stream_logs: bool, test_name: str) -> None:
+        old_env = {}
+        env_vars = {
+            "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+            "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+            "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+        }
+        for key, value in env_vars.items():
+            old_env[key] = os.environ.get(key)
+            os.environ[key] = value
+
         # Save original file descriptors
         original_stdout_fd = os.dup(1)  # stdout
 
@@ -778,6 +804,11 @@ def _stream_logs(self) -> bool:
                 ), f"stream_logs=True case: {stdout_content}"
 
         finally:
+            for key, value in old_env.items():
+                if value is None:
+                    os.environ.pop(key, None)
+                else:
+                    os.environ[key] = value
             # Ensure file descriptors are restored even if something goes wrong
             try:
                 os.dup2(original_stdout_fd, 1)
@@ -792,6 +823,16 @@ def _stream_logs(self) -> bool:
 
 @pytest.mark.timeout(60)
 async def test_logging_option_defaults() -> None:
+    old_env = {}
+    env_vars = {
+        "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+        "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+        "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+    }
+    for key, value in env_vars.items():
+        old_env[key] = os.environ.get(key)
+        os.environ[key] = value
+
     # Save original file descriptors
     original_stdout_fd = os.dup(1)  # stdout
     original_stderr_fd = os.dup(2)  # stderr
@@ -870,6 +911,12 @@ async def test_logging_option_defaults() -> None:
         ), stderr_content
 
     finally:
+        for key, value in old_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+
         # Ensure file descriptors are restored even if something goes wrong
         try:
             os.dup2(original_stdout_fd, 1)
@@ -906,6 +953,15 @@ def __init__(self):
 @pytest.mark.oss_skip
 async def test_flush_called_only_once() -> None:
     """Test that flush is called only once when ending an ipython cell"""
+    old_env = {}
+    env_vars = {
+        "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+        "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+        "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+    }
+    for key, value in env_vars.items():
+        old_env[key] = os.environ.get(key)
+        os.environ[key] = value
     mock_ipython = MockIPython()
     with unittest.mock.patch(
         "monarch._src.actor.logging.get_ipython",
@@ -926,14 +982,29 @@ async def test_flush_called_only_once() -> None:
 
         # now, flush should be called only once
         mock_ipython.events.trigger("post_run_cell", unittest.mock.MagicMock())
+
         assert mock_flush.call_count == 1
+        for key, value in old_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
 
 
 # oss_skip: pytest keeps complaining about mocking get_ipython module
 @pytest.mark.oss_skip
 @pytest.mark.timeout(180)
 async def test_flush_logs_ipython() -> None:
     """Test that logs are flushed when get_ipython is available and post_run_cell event is triggered."""
+    old_env = {}
+    env_vars = {
+        "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+        "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+        "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+    }
+    for key, value in env_vars.items():
+        old_env[key] = os.environ.get(key)
+        os.environ[key] = value
     # Save original file descriptors
     original_stdout_fd = os.dup(1)  # stdout
 
@@ -1025,6 +1096,11 @@ async def test_flush_logs_ipython() -> None:
         ), stdout_content
 
     finally:
+        for key, value in old_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
         # Ensure file descriptors are restored even if something goes wrong
         try:
             os.dup2(original_stdout_fd, 1)
@@ -1036,6 +1112,15 @@ async def test_flush_logs_ipython() -> None:
 # oss_skip: importlib not pulling resource correctly in git CI, needs to be revisited
 @pytest.mark.oss_skip
 async def test_flush_logs_fast_exit() -> None:
+    old_env = {}
+    env_vars = {
+        "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+        "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+        "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+    }
+    for key, value in env_vars.items():
+        old_env[key] = os.environ.get(key)
+        os.environ[key] = value
     # We use a subprocess to run the test so we can handle the flushed logs at the end.
     # Otherwise, it is hard to restore the original stdout/stderr.
 
@@ -1062,13 +1147,29 @@ async def test_flush_logs_fast_exit() -> None:
         == 1
     ), process.stdout
 
+    for key, value in old_env.items():
+        if value is None:
+            os.environ.pop(key, None)
+        else:
+            os.environ[key] = value
+
 
 @pytest.mark.timeout(60)
 async def test_flush_on_disable_aggregation() -> None:
     """Test that logs are flushed when disabling aggregation.
 
     This tests the corner case: "Make sure we flush whatever in the aggregators before disabling aggregation."
     """
+    old_env = {}
+    env_vars = {
+        "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+        "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+        "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+    }
+    for key, value in env_vars.items():
+        old_env[key] = os.environ.get(key)
+        os.environ[key] = value
+
     # Save original file descriptors
     original_stdout_fd = os.dup(1)  # stdout
 
@@ -1148,6 +1249,12 @@ async def test_flush_on_disable_aggregation() -> None:
         ), f"Expected 10 single log lines, got {total_single} from {stdout_content}"
 
     finally:
+        for key, value in old_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+
         # Ensure file descriptors are restored even if something goes wrong
         try:
             os.dup2(original_stdout_fd, 1)
@@ -1163,6 +1270,15 @@ async def test_multiple_ongoing_flushes_no_deadlock() -> None:
     Because now a flush call is purely sync, it is very easy to get into a deadlock.
     So we assert the last flush call will not get into such a state.
     """
+    old_env = {}
+    env_vars = {
+        "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+        "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+        "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+    }
+    for key, value in env_vars.items():
+        old_env[key] = os.environ.get(key)
+        os.environ[key] = value
     pm = this_host().spawn_procs(per_host={"gpus": 4})
     am = pm.spawn("printer", Printer)
 
@@ -1185,13 +1301,29 @@ async def test_multiple_ongoing_flushes_no_deadlock() -> None:
     # The last flush should not block
     futures[-1].get()
 
+    for key, value in old_env.items():
+        if value is None:
+            os.environ.pop(key, None)
+        else:
+            os.environ[key] = value
+
 
 @pytest.mark.timeout(60)
 async def test_adjust_aggregation_window() -> None:
     """Test that the flush deadline is updated when the aggregation window is adjusted.
 
     This tests the corner case: "This can happen if the user has adjusted the aggregation window."
     """
+    old_env = {}
+    env_vars = {
+        "HYPERACTOR_MESH_ENABLE_LOG_FORWARDING": "true",
+        "HYPERACTOR_MESH_ENABLE_FILE_CAPTURE": "true",
+        "HYPERACTOR_MESH_TAIL_LOG_LINES": "100",
+    }
+    for key, value in env_vars.items():
+        old_env[key] = os.environ.get(key)
+        os.environ[key] = value
+
     # Save original file descriptors
     original_stdout_fd = os.dup(1)  # stdout
 
@@ -1258,6 +1390,12 @@ async def test_adjust_aggregation_window() -> None:
         ), stdout_content
 
     finally:
+        for key, value in old_env.items():
+            if value is None:
+                os.environ.pop(key, None)
+            else:
+                os.environ[key] = value
+
         # Ensure file descriptors are restored even if something goes wrong
         try:
             os.dup2(original_stdout_fd, 1)