Enable GCS server when running python unit tests (#7101)

* Enable GCS server when running python unit tests

* restart ci

* restart ci

* fix code style

* restart ci

* restart ci

* restart ci

* restart ci

* restart ci

* Define RAY_GCS_SERVICE_ENABLED as a constant

* fix review comments

* fix code style

* fix code style

* fix code style

* fix code style

* fix review comments

* add gcs service python testcase

* fix TESTSUITE name bug

python/ray/node.py

@@ -609,7 +609,7 @@ class Node:
         # If this is the head node, start the relevant head node processes.
         self.start_redis()
 
-        if os.environ.get("RAY_GCS_SERVICE_ENABLED", None):
+        if os.environ.get(ray_constants.RAY_GCS_SERVICE_ENABLED, None):
             self.start_gcs_server()
 
         self.start_monitor()

python/ray/ray_constants.py

@@ -203,3 +203,8 @@ NODE_DEFAULT_IP = "127.0.0.1"
 
 # The Mach kernel page size in bytes.
 MACH_PAGE_SIZE_BYTES = 4096
+
+# RAY_GCS_SERVICE_ENABLED only set in ci job.
+# TODO(ffbin): Once we entirely migrate to service-based GCS, we should
+# remove it.
+RAY_GCS_SERVICE_ENABLED = "RAY_GCS_SERVICE_ENABLED"

python/ray/tests/test_component_failures_3.py

@@ -84,6 +84,9 @@ def test_driver_lives_sequential(ray_start_regular):
     ray.worker._global_node.kill_monitor()
     ray.worker._global_node.kill_raylet_monitor()
 
+    if os.environ.get(ray_constants.RAY_GCS_SERVICE_ENABLED, None):
+        ray.worker._global_node.kill_gcs_server()
+
     # If the driver can reach the tearDown method, then it is still alive.
 
 
@@ -92,12 +95,24 @@ def test_driver_lives_sequential(ray_start_regular):
     reason="Hanging with new GCS API.")
 def test_driver_lives_parallel(ray_start_regular):
     all_processes = ray.worker._global_node.all_processes
-    process_infos = (all_processes[ray_constants.PROCESS_TYPE_PLASMA_STORE] +
-                     all_processes[ray_constants.PROCESS_TYPE_RAYLET] +
-                     all_processes[ray_constants.PROCESS_TYPE_LOG_MONITOR] +
-                     all_processes[ray_constants.PROCESS_TYPE_MONITOR] +
-                     all_processes[ray_constants.PROCESS_TYPE_RAYLET_MONITOR])
-    assert len(process_infos) == 5
+
+    if os.environ.get(ray_constants.RAY_GCS_SERVICE_ENABLED, None):
+        process_infos = (
+            all_processes[ray_constants.PROCESS_TYPE_PLASMA_STORE] +
+            all_processes[ray_constants.PROCESS_TYPE_GCS_SERVER] +
+            all_processes[ray_constants.PROCESS_TYPE_RAYLET] +
+            all_processes[ray_constants.PROCESS_TYPE_LOG_MONITOR] +
+            all_processes[ray_constants.PROCESS_TYPE_MONITOR] +
+            all_processes[ray_constants.PROCESS_TYPE_RAYLET_MONITOR])
+        assert len(process_infos) == 6
+    else:
+        process_infos = (
+            all_processes[ray_constants.PROCESS_TYPE_PLASMA_STORE] +
+            all_processes[ray_constants.PROCESS_TYPE_RAYLET] +
+            all_processes[ray_constants.PROCESS_TYPE_LOG_MONITOR] +
+            all_processes[ray_constants.PROCESS_TYPE_MONITOR] +
+            all_processes[ray_constants.PROCESS_TYPE_RAYLET_MONITOR])
+        assert len(process_infos) == 5
 
     # Kill all the components in parallel.
     for process_info in process_infos:

python/ray/tests/test_tempfile.py

@@ -3,6 +3,7 @@ import shutil
 import time
 import pytest
 import ray
+import ray.ray_constants as ray_constants
 from ray.cluster_utils import Cluster
 
 
@@ -98,12 +99,18 @@ def test_raylet_tempfiles(shutdown_only):
     top_levels = set(os.listdir(node.get_session_dir_path()))
     assert top_levels.issuperset({"sockets", "logs"})
     log_files = set(os.listdir(node.get_logs_dir_path()))
-    assert log_files.issuperset({
+    log_files_expected = {
         "log_monitor.out", "log_monitor.err", "plasma_store.out",
         "plasma_store.err", "monitor.out", "monitor.err", "raylet_monitor.out",
         "raylet_monitor.err", "redis-shard_0.out", "redis-shard_0.err",
         "redis.out", "redis.err", "raylet.out", "raylet.err"
-    })  # with raylet logs
+    }
+
+    if os.environ.get(ray_constants.RAY_GCS_SERVICE_ENABLED, None):
+        log_files_expected.update({"gcs_server.out", "gcs_server.err"})
+
+    assert log_files.issuperset(log_files_expected)
+
     socket_files = set(os.listdir(node.get_sockets_dir_path()))
     assert socket_files == {"plasma_store", "raylet"}
     ray.shutdown()
@@ -114,12 +121,8 @@ def test_raylet_tempfiles(shutdown_only):
     assert top_levels.issuperset({"sockets", "logs"})
     time.sleep(3)  # wait workers to start
     log_files = set(os.listdir(node.get_logs_dir_path()))
-    assert log_files.issuperset({
-        "log_monitor.out", "log_monitor.err", "plasma_store.out",
-        "plasma_store.err", "monitor.out", "monitor.err", "raylet_monitor.out",
-        "raylet_monitor.err", "redis-shard_0.out", "redis-shard_0.err",
-        "redis.out", "redis.err", "raylet.out", "raylet.err"
-    })  # with raylet logs
+
+    assert log_files.issuperset(log_files_expected)
 
     # Check numbers of worker log file.
     assert sum(