From 915486984a2890f93235a8c5a1ec6963321fdbb0 Mon Sep 17 00:00:00 2001 From: Stefan Pantic Date: Mon, 8 Apr 2019 01:51:32 +0200 Subject: [PATCH] [autoscaler] Add support for separate docker containers on head and worker nodes (#4537) * Added support for running different docker containers on clusters * Remove node specific container names * Keep old options and expand with node specific configuration * Optimized imports * Changed docker fields for autoscaler * Auto reformat * Updated comments * Updated condition * Run linter * Updated example * Changed condition for docker images, updated examples * Removed duplicate line * Fixed setup_commands * Update autoscaler.py * fix_better_image --- python/ray/autoscaler/autoscaler.py | 61 ++++++++++++------- python/ray/autoscaler/aws/example-full.yaml | 8 +++ .../autoscaler/aws/example-gpu-docker.yaml | 8 +++ python/ray/autoscaler/commands.py | 8 +-- python/ray/autoscaler/docker.py | 32 +++++++--- 5 files changed, 79 insertions(+), 38 deletions(-) diff --git a/python/ray/autoscaler/autoscaler.py b/python/ray/autoscaler/autoscaler.py index 5c5b1d19f..047907d55 100644 --- a/python/ray/autoscaler/autoscaler.py +++ b/python/ray/autoscaler/autoscaler.py @@ -3,34 +3,31 @@ from __future__ import division from __future__ import print_function import copy -import json import hashlib +import json +import logging import math import os -from six import string_types -from six.moves import queue import subprocess import threading -import logging import time - from collections import defaultdict import numpy as np +import ray.services as services import yaml - -from ray.ray_constants import AUTOSCALER_MAX_NUM_FAILURES, \ - AUTOSCALER_MAX_LAUNCH_BATCH, AUTOSCALER_MAX_CONCURRENT_LAUNCHES,\ - AUTOSCALER_UPDATE_INTERVAL_S, AUTOSCALER_HEARTBEAT_TIMEOUT_S +from ray.autoscaler.docker import dockerize_if_needed from ray.autoscaler.node_provider import get_node_provider, \ get_default_config -from ray.autoscaler.updater import NodeUpdaterThread -from ray.autoscaler.docker import dockerize_if_needed from ray.autoscaler.tags import (TAG_RAY_LAUNCH_CONFIG, TAG_RAY_RUNTIME_CONFIG, TAG_RAY_NODE_STATUS, TAG_RAY_NODE_TYPE, TAG_RAY_NODE_NAME) - -import ray.services as services +from ray.autoscaler.updater import NodeUpdaterThread +from ray.ray_constants import AUTOSCALER_MAX_NUM_FAILURES, \ + AUTOSCALER_MAX_LAUNCH_BATCH, AUTOSCALER_MAX_CONCURRENT_LAUNCHES, \ + AUTOSCALER_UPDATE_INTERVAL_S, AUTOSCALER_HEARTBEAT_TIMEOUT_S +from six import string_types +from six.moves import queue logger = logging.getLogger(__name__) @@ -38,6 +35,7 @@ REQUIRED, OPTIONAL = True, False # For (a, b), if a is a dictionary object, then # no extra fields can be introduced. + CLUSTER_CONFIG_SCHEMA = { # An unique identifier for the head node and workers of this cluster. "cluster_name": (str, REQUIRED), @@ -91,7 +89,17 @@ CLUSTER_CONFIG_SCHEMA = { { "image": (str, OPTIONAL), # e.g. tensorflow/tensorflow:1.5.0-py3 "container_name": (str, OPTIONAL), # e.g., ray_docker + # shared options for starting head/worker docker "run_options": (list, OPTIONAL), + + # image for head node, takes precedence over "image" if specified + "head_image": (str, OPTIONAL), + # head specific run options, appended to run_options + "head_run_options": (list, OPTIONAL), + # analogous to head_image + "worker_image": (str, OPTIONAL), + # analogous to head_run_options + "worker_run_options": (list, OPTIONAL), }, OPTIONAL), @@ -492,7 +500,6 @@ class StandardAutoscaler(object): new_launch_hash = hash_launch_conf(new_config["worker_nodes"], new_config["auth"]) new_runtime_hash = hash_runtime_conf(new_config["file_mounts"], [ - new_config["setup_commands"], new_config["worker_setup_commands"], new_config["worker_start_ray_commands"] ]) @@ -575,11 +582,9 @@ class StandardAutoscaler(object): if successful_updated and self.config.get("restart_only", False): init_commands = self.config["worker_start_ray_commands"] elif successful_updated and self.config.get("no_restart", False): - init_commands = (self.config["setup_commands"] + - self.config["worker_setup_commands"]) + init_commands = (self.config["worker_setup_commands"]) else: - init_commands = (self.config["setup_commands"] + - self.config["worker_setup_commands"] + + init_commands = (self.config["worker_setup_commands"] + self.config["worker_start_ray_commands"]) return (node_id, init_commands) @@ -618,8 +623,9 @@ class StandardAutoscaler(object): self.launch_queue.put((config, count)) def workers(self): - return self.provider.non_terminated_nodes( - tag_filters={TAG_RAY_NODE_TYPE: "worker"}) + return self.provider.non_terminated_nodes(tag_filters={ + TAG_RAY_NODE_TYPE: "worker" + }) def log_info_string(self, nodes): logger.info("StandardAutoscaler: {}".format(self.info_string(nodes))) @@ -650,7 +656,7 @@ def typename(v): def check_required(config, schema): # Check required schema entries - if type(config) is not dict: + if not isinstance(config, dict): raise ValueError("Config is not a dictionary") for k, (v, kreq) in schema.items(): @@ -668,7 +674,7 @@ def check_required(config, schema): def check_extraneous(config, schema): """Make sure all items of config are in schema""" - if type(config) is not dict: + if not isinstance(config, dict): raise ValueError("Config {} is not a dictionary".format(config)) for k in config: if k not in schema: @@ -691,7 +697,7 @@ def check_extraneous(config, schema): def validate_config(config, schema=CLUSTER_CONFIG_SCHEMA): """Required Dicts indicate that no extra fields can be introduced.""" - if type(config) is not dict: + if not isinstance(config, dict): raise ValueError("Config {} is not a dictionary".format(config)) check_required(config, schema) @@ -701,10 +707,19 @@ def validate_config(config, schema=CLUSTER_CONFIG_SCHEMA): def fillout_defaults(config): defaults = get_default_config(config["provider"]) defaults.update(config) + merge_setup_commands(defaults) dockerize_if_needed(defaults) return defaults +def merge_setup_commands(config): + config["head_setup_commands"] = config["setup_commands"] + \ + config["head_setup_commands"] + config["worker_setup_commands"] = config["setup_commands"] + \ + config["worker_setup_commands"] + return config + + def with_head_node_ip(cmds): head_ip = services.get_node_ip_address() out = [] diff --git a/python/ray/autoscaler/aws/example-full.yaml b/python/ray/autoscaler/aws/example-full.yaml index d24a47c45..82bf92a99 100644 --- a/python/ray/autoscaler/aws/example-full.yaml +++ b/python/ray/autoscaler/aws/example-full.yaml @@ -22,6 +22,14 @@ docker: container_name: "" # e.g. ray_docker run_options: [] # Extra options to pass into "docker run" + # Example of running a GPU head with CPU workers + # head_image: "tensorflow/tensorflow:1.13.1-py3" + # head_run_options: + # - --runtime=nvidia + + # worker_image: "ubuntu:18.04" + # worker_run_options: [] + # The autoscaler will scale up the cluster to this target fraction of resource # usage. For example, if a cluster of 10 nodes is 100% busy and # target_utilization is 0.8, it would resize the cluster to 13. This fraction diff --git a/python/ray/autoscaler/aws/example-gpu-docker.yaml b/python/ray/autoscaler/aws/example-gpu-docker.yaml index 6a10fff5d..962685390 100644 --- a/python/ray/autoscaler/aws/example-gpu-docker.yaml +++ b/python/ray/autoscaler/aws/example-gpu-docker.yaml @@ -23,6 +23,14 @@ docker: run_options: - --runtime=nvidia + # # Example of running a GPU head with CPU workers + # head_image: "tensorflow/tensorflow:1.13.1-gpu-py3" + # head_run_options: + # - --runtime=nvidia + + # worker_image: "tensorflow/tensorflow:1.13.1-py3" + # worker_run_options: [] + # The autoscaler will scale up the cluster to this target fraction of resource # usage. For example, if a cluster of 10 nodes is 100% busy and # target_utilization is 0.8, it would resize the cluster to 13. This fraction diff --git a/python/ray/autoscaler/commands.py b/python/ray/autoscaler/commands.py index 223460bc2..6c819c19d 100644 --- a/python/ray/autoscaler/commands.py +++ b/python/ray/autoscaler/commands.py @@ -226,12 +226,10 @@ def get_or_create_head_node(config, config_file, no_restart, restart_only, yes, if restart_only: init_commands = config["head_start_ray_commands"] elif no_restart: - init_commands = ( - config["setup_commands"] + config["head_setup_commands"]) + init_commands = (config["head_setup_commands"]) else: - init_commands = ( - config["setup_commands"] + config["head_setup_commands"] + - config["head_start_ray_commands"]) + init_commands = (config["head_setup_commands"] + + config["head_start_ray_commands"]) updater = NodeUpdaterThread( node_id=head_node, diff --git a/python/ray/autoscaler/docker.py b/python/ray/autoscaler/docker.py index 5a951c627..4bf233cb6 100644 --- a/python/ray/autoscaler/docker.py +++ b/python/ray/autoscaler/docker.py @@ -15,33 +15,45 @@ logger = logging.getLogger(__name__) def dockerize_if_needed(config): if "docker" not in config: return config + docker_image = config["docker"].get("image") cname = config["docker"].get("container_name") run_options = config["docker"].get("run_options", []) + + head_docker_image = config["docker"].get("head_image", docker_image) + head_run_options = config["docker"].get("head_run_options", []) + + worker_docker_image = config["docker"].get("worker_image", docker_image) + worker_run_options = config["docker"].get("worker_run_options", []) + ssh_user = config["auth"]["ssh_user"] - if not docker_image: + if not docker_image and not (head_docker_image and worker_docker_image): if cname: logger.warning( "dockerize_if_needed: " - "Container name given but no Docker image - continuing...") + "Container name given but no Docker image(s) - continuing...") return config else: assert cname, "Must provide container name!" docker_mounts = {dst: dst for dst in config["file_mounts"]} - config["setup_commands"] = ( - docker_start_cmds(ssh_user, docker_image, docker_mounts, cname, - run_options) + with_docker_exec( - config["setup_commands"], container_name=cname)) + head_docker_start = docker_start_cmds(ssh_user, head_docker_image, + docker_mounts, cname, + run_options + head_run_options) - config["head_setup_commands"] = with_docker_exec( - config["head_setup_commands"], container_name=cname) + worker_docker_start = docker_start_cmds(ssh_user, worker_docker_image, + docker_mounts, cname, + run_options + worker_run_options) + + config["head_setup_commands"] = head_docker_start + \ + with_docker_exec(config["head_setup_commands"], + container_name=cname) config["head_start_ray_commands"] = ( docker_autoscaler_setup(cname) + with_docker_exec( config["head_start_ray_commands"], container_name=cname)) - config["worker_setup_commands"] = with_docker_exec( - config["worker_setup_commands"], container_name=cname) + config["worker_setup_commands"] = worker_docker_start + \ + with_docker_exec(config["worker_setup_commands"], container_name=cname) config["worker_start_ray_commands"] = with_docker_exec( config["worker_start_ray_commands"], container_name=cname,