wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-15 11:25:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

[autoscaler] Add support for separate docker containers on head and worker nodes (#4537)

Browse Source 
* 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
This commit is contained in:
Stefan Pantic
2019-04-07 16:51:32 -07:00
committed by Kristian Hartikainen
parent da5a471485
commit 915486984a
5 changed files with 79 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/autoscaler/autoscaler.py
+38 -23
View File
@@ -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 = []
python/ray/autoscaler/aws/example-full.yaml
+8
View File
@@ -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
python/ray/autoscaler/aws/example-gpu-docker.yaml
+8
View File
@@ -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
python/ray/autoscaler/commands.py
+3 -5
View File
@@ -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,
python/ray/autoscaler/docker.py
+22 -10
View File
@@ -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,