[Autoscaler] make --redis-address not required (ray-project#22083)

`--redis-address` should not be required, since starting autoscaler with `--gcs-address` is supported too.

python/ray/_private/services.py

@@ -2103,61 +2103,6 @@ def determine_plasma_store_config(
     return plasma_directory, object_store_memory
 
 
-def start_worker(
-    node_ip_address,
-    object_store_name,
-    raylet_name,
-    redis_address,
-    worker_path,
-    temp_dir,
-    raylet_ip_address=None,
-    stdout_file=None,
-    stderr_file=None,
-    fate_share=None,
-):
-    """This method starts a worker process.
-
-    Args:
-        node_ip_address (str): The IP address of the node that this worker is
-            running on.
-        object_store_name (str): The socket name of the object store.
-        raylet_name (str): The socket name of the raylet server.
-        redis_address (str): The address that the Redis server is listening on.
-        worker_path (str): The path of the source code which the worker process
-            will run.
-        temp_dir (str): The path of the temp dir.
-        raylet_ip_address (str): The IP address of the worker's raylet. If not
-            provided, it defaults to the node_ip_address.
-        stdout_file: A file handle opened for writing to redirect stdout to. If
-            no redirection should happen, then this should be None.
-        stderr_file: A file handle opened for writing to redirect stderr to. If
-            no redirection should happen, then this should be None.
-
-    Returns:
-        ProcessInfo for the process that was started.
-    """
-    command = [
-        sys.executable,
-        "-u",
-        worker_path,
-        "--node-ip-address=" + node_ip_address,
-        "--object-store-name=" + object_store_name,
-        "--raylet-name=" + raylet_name,
-        "--redis-address=" + str(redis_address),
-        "--temp-dir=" + temp_dir,
-    ]
-    if raylet_ip_address is not None:
-        command.append("--raylet-ip-address=" + raylet_ip_address)
-    process_info = start_ray_process(
-        command,
-        ray_constants.PROCESS_TYPE_WORKER,
-        stdout_file=stdout_file,
-        stderr_file=stderr_file,
-        fate_share=fate_share,
-    )
-    return process_info
-
-
 def start_monitor(
     redis_address,
     gcs_address,
@@ -2199,12 +2144,13 @@ def start_monitor(
         "-u",
         monitor_path,
         f"--logs-dir={logs_dir}",
-        f"--redis-address={redis_address}",
         f"--logging-rotate-bytes={max_bytes}",
         f"--logging-rotate-backup-count={backup_count}",
-        f"--gcs-address={gcs_address}",
     ]
-
+    if redis_address is not None:
+        command.append(f"--redis-address={redis_address}")
+    if gcs_address is not None:
+        command.append(f"--gcs-address={gcs_address}")
     if stdout_file is None and stderr_file is None:
         # If not redirecting logging to files, unset log filename.
         # This will cause log records to go to stderr.

python/ray/autoscaler/_private/monitor.py

@@ -531,7 +531,7 @@ def log_resource_batch_data_if_desired(
         "--gcs-address", required=False, type=str, help="The address (ip:port) of GCS."
     )
     parser.add_argument(
-        "--redis-address", required=True, type=str, help="the address to use for Redis"
+        "--redis-address", required=False, type=str, help="the address to use for Redis"
     )
     parser.add_argument(
         "--autoscaling-config",
@@ -620,8 +620,14 @@ def log_resource_batch_data_if_desired(
     else:
         autoscaling_config = None
 
+    bootstrap_address = (
+        args.gcs_address if use_gcs_for_bootstrap() else args.redis_address
+    )
+    if bootstrap_address is None:
+        raise ValueError("One of --gcs-address or --redis-address must be set!")
+
     monitor = Monitor(
-        args.gcs_address if use_gcs_for_bootstrap() else args.redis_address,
+        bootstrap_address,
         autoscaling_config,
         redis_password=args.redis_password,
         monitor_ip=args.monitor_ip,