From 5d408eecfd0e9bbe30d7c6e77cad710de6e7ef4f Mon Sep 17 00:00:00 2001
From: Yang Wang <yang3.wang@intel.com>
Date: Thu, 3 Sep 2020 10:47:57 +0800
Subject: [PATCH] fix duplicated ray worker logs (#2799)

* fix multiple ray worker logs

* refine comments

* Update raycontext.py
---
 python/orca/src/bigdl/orca/ray/raycontext.py | 30 ++++++++++++++++++++
 1 file changed, 30 insertions(+)

diff --git a/python/orca/src/bigdl/orca/ray/raycontext.py b/python/orca/src/bigdl/orca/ray/raycontext.py
index 3958b068779..278a5e1496d 100755
--- a/python/orca/src/bigdl/orca/ray/raycontext.py
+++ b/python/orca/src/bigdl/orca/ray/raycontext.py
@@ -48,6 +48,32 @@ def register_pids(pids):
             raise err
 
 
+def kill_redundant_log_monitors(redis_address):
+
+    """
+    Killing redundant log_monitor.py processes.
+    If multiple ray nodes are started on the same machine,
+    there will be multiple ray log_monitor.py processes
+    monitoring the same log dir. As a result, the logs
+    will be replicated multiple times and forwarded to driver.
+    See issue https://github.com/ray-project/ray/issues/10392
+    """
+
+    import psutil
+    import subprocess
+    log_monitor_processes = []
+    for proc in psutil.process_iter(["name", "cmdline"]):
+        cmdline = subprocess.list2cmdline(proc.cmdline())
+        is_log_monitor = "log_monitor.py" in cmdline
+        is_same_redis = "--redis-address={}".format(redis_address)
+        if is_log_monitor and is_same_redis in cmdline:
+            log_monitor_processes.append(proc)
+
+    if len(log_monitor_processes) > 1:
+        for proc in log_monitor_processes[1:]:
+            proc.kill()
+
+
 class RayServiceFuncGenerator(object):
     """
     This should be a pickable class.
@@ -204,6 +230,7 @@ def _start_ray_services(iter):
                             object_store_memory=self.object_store_memory,
                             extra_params=self.extra_params),
                         tag="raylet")
+                    kill_redundant_log_monitors(redis_address=redis_address)
 
             yield process_info
         return _start_ray_services
@@ -418,6 +445,9 @@ def init(self, driver_cores=0):
             else:
                 self._start_cluster()
                 self._address_info = self._start_driver(num_cores=driver_cores)
+
+            print(self._address_info)
+            kill_redundant_log_monitors(self._address_info["redis_address"])
             self.initialized = True
         return self._address_info