[Autoscaler] Command Line Interface improvements (#9322)

Co-authored-by: Richard Liaw <rliaw@berkeley.edu>
2026-08-17 11:25:34 +08:00 · 2020-07-22 12:21:44 -07:00
parent 456e012029
commit 908c0c630a
11 changed files with 1475 additions and 332 deletions
@@ -13,7 +13,11 @@ import botocore

 from ray.ray_constants import BOTO_MAX_RETRIES
 from ray.autoscaler.tags import NODE_TYPE_WORKER, NODE_TYPE_HEAD
-from ray.autoscaler.aws.utils import LazyDefaultDict
+from ray.autoscaler.aws.utils import LazyDefaultDict, handle_boto_error
+from ray.autoscaler.node_provider import PROVIDER_PRETTY_NAMES
+
+from ray.autoscaler.cli_logger import cli_logger
+import colorful as cf

 logger = logging.getLogger(__name__)

@@ -45,7 +49,8 @@ DEFAULT_AMI = {
    "sa-east-1": "ami-0da2c49fe75e7e5ed",  # SA (Sao Paulo)
 }

-
+# todo: cli_logger should handle this assert properly
+# this should probably also happens somewhere else
 assert StrictVersion(boto3.__version__) >= StrictVersion("1.4.8"), \
    "Boto3 version >= 1.4.8 required, try `pip install -U boto3`"

@@ -70,6 +75,112 @@ def key_pair(i, region, key_name):
 # Suppress excessive connection dropped logs from boto
 logging.getLogger("botocore").setLevel(logging.WARNING)

+_log_info = {}
+
+
+def reload_log_state(override_log_info):
+    _log_info.update(override_log_info)
+
+
+def get_log_state():
+    return _log_info.copy()
+
+
+def _set_config_info(**kwargs):
+    """Record configuration artifacts useful for logging."""
+
+    # todo: this is technically fragile iff we ever use multiple configs
+
+    for k, v in kwargs.items():
+        _log_info[k] = v
+
+
+def _arn_to_name(arn):
+    return arn.split(":")[-1].split("/")[-1]
+
+
+def log_to_cli(config):
+    provider_name = PROVIDER_PRETTY_NAMES.get("aws", None)
+
+    cli_logger.doassert(provider_name is not None,
+                        "Could not find a pretty name for the AWS provider.")
+
+    with cli_logger.group("{} config", provider_name):
+
+        def same_everywhere(key):
+            return config["head_node"][key] == config["worker_nodes"][key]
+
+        def print_info(resource_string,
+                       key,
+                       head_src_key,
+                       workers_src_key,
+                       allowed_tags=["default"],
+                       list_value=False):
+
+            head_tags = {}
+            workers_tags = {}
+
+            if _log_info[head_src_key] in allowed_tags:
+                head_tags[_log_info[head_src_key]] = True
+            if _log_info[workers_src_key] in allowed_tags:
+                workers_tags[_log_info[workers_src_key]] = True
+
+            head_value_str = config["head_node"][key]
+            if list_value:
+                head_value_str = cli_logger.render_list(head_value_str)
+
+            if same_everywhere(key):
+                cli_logger.labeled_value(  # todo: handle plural vs singular?
+                    resource_string + " (head & workers)",
+                    "{}",
+                    head_value_str,
+                    _tags=head_tags)
+            else:
+                workers_value_str = config["worker_nodes"][key]
+                if list_value:
+                    workers_value_str = cli_logger.render_list(
+                        workers_value_str)
+
+                cli_logger.labeled_value(
+                    resource_string + " (head)",
+                    "{}",
+                    head_value_str,
+                    _tags=head_tags)
+                cli_logger.labeled_value(
+                    resource_string + " (workers)",
+                    "{}",
+                    workers_value_str,
+                    _tags=workers_tags)
+
+        tags = {"default": _log_info["head_instance_profile_src"] == "default"}
+        cli_logger.labeled_value(
+            "IAM Profile",
+            "{}",
+            _arn_to_name(config["head_node"]["IamInstanceProfile"]["Arn"]),
+            _tags=tags)
+
+        print_info("EC2 Key pair", "KeyName", "keypair_src", "keypair_src")
+        print_info(
+            "VPC Subnets",
+            "SubnetIds",
+            "head_subnet_src",
+            "workers_subnet_src",
+            list_value=True)
+        print_info(
+            "EC2 Security groups",
+            "SecurityGroupIds",
+            "head_security_group_src",
+            "workers_security_group_src",
+            list_value=True)
+        print_info(
+            "EC2 AMI",
+            "ImageId",
+            "head_ami_src",
+            "workers_ami_src",
+            allowed_tags=["dlami"])
+
+    cli_logger.newline()
+

 def bootstrap_aws(config):
    # The head node needs to have an IAM role that allows it to create further
@@ -94,27 +205,38 @@ def bootstrap_aws(config):

 def _configure_iam_role(config):
    if "IamInstanceProfile" in config["head_node"]:
+        _set_config_info(head_instance_profile_src="config")
        return config
+    _set_config_info(head_instance_profile_src="default")

    profile = _get_instance_profile(DEFAULT_RAY_INSTANCE_PROFILE, config)

    if profile is None:
-        logger.info("_configure_iam_role: "
-                    "Creating new instance profile {}".format(
-                        DEFAULT_RAY_INSTANCE_PROFILE))
+        cli_logger.verbose(
+            "Creating new IAM instance profile {} for use as the default.",
+            cf.bold(DEFAULT_RAY_INSTANCE_PROFILE))
+        cli_logger.old_info(
+            logger, "_configure_iam_role: "
+            "Creating new instance profile {}", DEFAULT_RAY_INSTANCE_PROFILE)
        client = _client("iam", config)
        client.create_instance_profile(
            InstanceProfileName=DEFAULT_RAY_INSTANCE_PROFILE)
        profile = _get_instance_profile(DEFAULT_RAY_INSTANCE_PROFILE, config)
        time.sleep(15)  # wait for propagation

+    cli_logger.doassert(profile is not None,
+                        "Failed to create instance profile.")  # todo: err msg
    assert profile is not None, "Failed to create instance profile"

    if not profile.roles:
        role = _get_role(DEFAULT_RAY_IAM_ROLE, config)
        if role is None:
-            logger.info("_configure_iam_role: "
-                        "Creating new role {}".format(DEFAULT_RAY_IAM_ROLE))
+            cli_logger.verbose(
+                "Creating new IAM role {} for "
+                "use as the default instance role.",
+                cf.bold(DEFAULT_RAY_IAM_ROLE))
+            cli_logger.old_info(logger, "_configure_iam_role: "
+                                "Creating new role {}", DEFAULT_RAY_IAM_ROLE)
            iam = _resource("iam", config)
            iam.create_role(
                RoleName=DEFAULT_RAY_IAM_ROLE,
@@ -130,6 +252,9 @@ def _configure_iam_role(config):
                    ],
                }))
            role = _get_role(DEFAULT_RAY_IAM_ROLE, config)
+
+            cli_logger.doassert(role is not None,
+                                "Failed to create role.")  # todo: err msg
            assert role is not None, "Failed to create role"
        role.attach_policy(
            PolicyArn="arn:aws:iam::aws:policy/AmazonEC2FullAccess")
@@ -138,9 +263,9 @@ def _configure_iam_role(config):
        profile.add_role(RoleName=role.name)
        time.sleep(15)  # wait for propagation

-    logger.info("_configure_iam_role: "
-                "Role not specified for head node, using {}".format(
-                    profile.arn))
+    cli_logger.old_info(
+        logger, "_configure_iam_role: "
+        "Role not specified for head node, using {}", profile.arn)
    config["head_node"]["IamInstanceProfile"] = {"Arn": profile.arn}

    return config
@@ -148,9 +273,19 @@ def _configure_iam_role(config):

 def _configure_key_pair(config):
    if "ssh_private_key" in config["auth"]:
+        _set_config_info(keypair_src="config")
+
+        cli_logger.doassert(  # todo: verify schema beforehand?
+            "KeyName" in config["head_node"],
+            "`KeyName` missing for head node.")  # todo: err msg
+        cli_logger.doassert(
+            "KeyName" in config["worker_nodes"],
+            "`KeyName` missing for worker nodes.")  # todo: err msg
+
        assert "KeyName" in config["head_node"]
        assert "KeyName" in config["worker_nodes"]
        return config
+    _set_config_info(keypair_src="default")

    ec2 = _resource("ec2", config)

@@ -170,8 +305,12 @@ def _configure_key_pair(config):

        # We can safely create a new key.
        if not key and not os.path.exists(key_path):
-            logger.info("_configure_key_pair: "
-                        "Creating new key pair {}".format(key_name))
+            cli_logger.verbose(
+                "Creating new key pair {} for use as the default.",
+                cf.bold(key_name))
+            cli_logger.old_info(
+                logger, "_configure_key_pair: "
+                "Creating new key pair {}", key_name)
            key = ec2.create_key_pair(KeyName=key_name)

            # We need to make sure to _create_ the file with the right
@@ -182,16 +321,25 @@ def _configure_key_pair(config):
            break

    if not key:
+        cli_logger.abort(
+            "No matching local key file for any of the key pairs in this "
+            "account with ids from 0..{}. "
+            "Consider deleting some unused keys pairs from your account.",
+            key_name)  # todo: err msg
        raise ValueError(
            "No matching local key file for any of the key pairs in this "
            "account with ids from 0..{}. ".format(key_name) +
            "Consider deleting some unused keys pairs from your account.")

+    cli_logger.doassert(
+        os.path.exists(key_path), "Private key file " + cf.bold("{}") +
+        " not found for " + cf.bold("{}"), key_path, key_name)  # todo: err msg
    assert os.path.exists(key_path), \
        "Private key file {} not found for {}".format(key_path, key_name)

-    logger.info("_configure_key_pair: "
-                "KeyName not specified for nodes, using {}".format(key_name))
+    cli_logger.old_info(
+        logger, "_configure_key_pair: "
+        "KeyName not specified for nodes, using {}", key_name)

    config["auth"]["ssh_private_key"] = key_path
    config["head_node"]["KeyName"] = key_name
@@ -203,12 +351,25 @@ def _configure_key_pair(config):
 def _configure_subnet(config):
    ec2 = _resource("ec2", config)
    use_internal_ips = config["provider"].get("use_internal_ips", False)
-    subnets = sorted(
-        (s for s in ec2.subnets.all() if s.state == "available" and (
-            use_internal_ips or s.map_public_ip_on_launch)),
-        reverse=True,  # sort from Z-A
-        key=lambda subnet: subnet.availability_zone)
+
+    try:
+        subnets = sorted(
+            (s for s in ec2.subnets.all() if s.state == "available" and (
+                use_internal_ips or s.map_public_ip_on_launch)),
+            reverse=True,  # sort from Z-A
+            key=lambda subnet: subnet.availability_zone)
+    except botocore.exceptions.ClientError as exc:
+        handle_boto_error(exc, "Failed to fetch available subnets from AWS.")
+        raise exc
+
    if not subnets:
+        cli_logger.abort(
+            "No usable subnets found, try manually creating an instance in "
+            "your specified region to populate the list of subnets "
+            "and trying this again.\n"
+            "Note that the subnet must map public IPs "
+            "on instance launch unless you set `use_internal_ips: true` in "
+            "the `provider` config.")  # todo: err msg
        raise Exception(
            "No usable subnets found, try manually creating an instance in "
            "your specified region to populate the list of subnets "
@@ -219,6 +380,12 @@ def _configure_subnet(config):
        azs = config["provider"]["availability_zone"].split(",")
        subnets = [s for s in subnets if s.availability_zone in azs]
        if not subnets:
+            cli_logger.abort(
+                "No usable subnets matching availability zone {} found.\n"
+                "Choose a different availability zone or try "
+                "manually creating an instance in your specified region "
+                "to populate the list of subnets and trying this again.",
+                config["provider"]["availability_zone"])  # todo: err msg
            raise Exception(
                "No usable subnets matching availability zone {} "
                "found. Choose a different availability zone or try "
@@ -229,21 +396,31 @@ def _configure_subnet(config):
    subnet_ids = [s.subnet_id for s in subnets]
    subnet_descr = [(s.subnet_id, s.availability_zone) for s in subnets]
    if "SubnetIds" not in config["head_node"]:
+        _set_config_info(head_subnet_src="default")
        config["head_node"]["SubnetIds"] = subnet_ids
-        logger.info("_configure_subnet: "
-                    "SubnetIds not specified for head node, using {}".format(
-                        subnet_descr))
+        cli_logger.old_info(
+            logger, "_configure_subnet: "
+            "SubnetIds not specified for head node, using {}", subnet_descr)
+    else:
+        _set_config_info(head_subnet_src="config")

    if "SubnetIds" not in config["worker_nodes"]:
+        _set_config_info(workers_subnet_src="default")
        config["worker_nodes"]["SubnetIds"] = subnet_ids
-        logger.info("_configure_subnet: "
-                    "SubnetId not specified for workers,"
-                    " using {}".format(subnet_descr))
+        cli_logger.old_info(
+            logger, "_configure_subnet: "
+            "SubnetId not specified for workers,"
+            " using {}", subnet_descr)
+    else:
+        _set_config_info(workers_subnet_src="config")

    return config


 def _configure_security_group(config):
+    _set_config_info(
+        head_security_group_src="config", workers_security_group_src="config")
+
    node_types_to_configure = [
        node_type for node_type, config_key in NODE_TYPE_CONFIG_KEYS.items()
        if "SecurityGroupIds" not in config[NODE_TYPE_CONFIG_KEYS[node_type]]
@@ -255,17 +432,22 @@ def _configure_security_group(config):

    if NODE_TYPE_HEAD in node_types_to_configure:
        head_sg = security_groups[NODE_TYPE_HEAD]
-        logger.info(
-            "_configure_security_group: "
-            "SecurityGroupIds not specified for head node, using {} ({})"
-            .format(head_sg.group_name, head_sg.id))
+
+        _set_config_info(head_security_group_src="default")
+        cli_logger.old_info(
+            logger, "_configure_security_group: "
+            "SecurityGroupIds not specified for head node, using {} ({})",
+            head_sg.group_name, head_sg.id)
        config["head_node"]["SecurityGroupIds"] = [head_sg.id]

    if NODE_TYPE_WORKER in node_types_to_configure:
        workers_sg = security_groups[NODE_TYPE_WORKER]
-        logger.info("_configure_security_group: "
-                    "SecurityGroupIds not specified for workers, using {} ({})"
-                    .format(workers_sg.group_name, workers_sg.id))
+
+        _set_config_info(workers_security_group_src="default")
+        cli_logger.old_info(
+            logger, "_configure_security_group: "
+            "SecurityGroupIds not specified for workers, using {} ({})",
+            workers_sg.group_name, workers_sg.id)
        config["worker_nodes"]["SecurityGroupIds"] = [workers_sg.id]

    return config
@@ -274,6 +456,8 @@ def _configure_security_group(config):
 def _check_ami(config):
    """Provide helpful message for missing ImageId for node configuration."""

+    _set_config_info(head_ami_src="config", workers_ami_src="config")
+
    region = config["provider"]["region"]
    default_ami = DEFAULT_AMI.get(region)
    if not default_ami:
@@ -282,21 +466,27 @@ def _check_ami(config):

    if config["head_node"].get("ImageId", "").lower() == "latest_dlami":
        config["head_node"]["ImageId"] = default_ami
-        logger.info("_check_ami: head node ImageId is 'latest_dlami'. "
-                    "Using '{ami_id}', which is the default {ami_name} "
-                    "for your region ({region}).".format(
-                        ami_id=default_ami,
-                        ami_name=DEFAULT_AMI_NAME,
-                        region=region))
+        _set_config_info(head_ami_src="dlami")
+        cli_logger.old_info(
+            logger,
+            "_check_ami: head node ImageId is 'latest_dlami'. "
+            "Using '{ami_id}', which is the default {ami_name} "
+            "for your region ({region}).",
+            ami_id=default_ami,
+            ami_name=DEFAULT_AMI_NAME,
+            region=region)

    if config["worker_nodes"].get("ImageId", "").lower() == "latest_dlami":
        config["worker_nodes"]["ImageId"] = default_ami
-        logger.info("_check_ami: worker nodes ImageId is 'latest_dlami'. "
-                    "Using '{ami_id}', which is the default {ami_name} "
-                    "for your region ({region}).".format(
-                        ami_id=default_ami,
-                        ami_name=DEFAULT_AMI_NAME,
-                        region=region))
+        _set_config_info(workers_ami_src="dlami")
+        cli_logger.old_info(
+            logger,
+            "_check_ami: worker nodes ImageId is 'latest_dlami'. "
+            "Using '{ami_id}', which is the default {ami_name} "
+            "for your region ({region}).",
+            ami_id=default_ami,
+            ami_name=DEFAULT_AMI_NAME,
+            region=region)


 def _upsert_security_groups(config, node_types):
@@ -350,6 +540,9 @@ def _get_vpc_id_or_die(ec2, subnet_id):
            "Name": "subnet-id",
            "Values": [subnet_id]
        }]))
+
+    # TODO: better error message
+    cli_logger.doassert(len(subnet) == 1, "Subnet ID not found: {}", subnet_id)
    assert len(subnet) == 1, "Subnet ID not found: {}".format(subnet_id)
    subnet = subnet[0]
    return subnet.vpc_id
@@ -383,8 +576,17 @@ def _create_security_group(config, vpc_id, group_name):
        GroupName=group_name,
        VpcId=vpc_id)
    security_group = _get_security_group(config, vpc_id, group_name)
-    logger.info("_create_security_group: Created new security group {} ({})"
-                .format(security_group.group_name, security_group.id))
+
+    cli_logger.verbose(
+        "Created new security group {}",
+        cf.bold(security_group.group_name),
+        _tags=dict(id=security_group.id))
+    cli_logger.old_info(
+        logger, "_create_security_group: Created new security group {} ({})",
+        security_group.group_name, security_group.id)
+
+    cli_logger.doassert(security_group,
+                        "Failed to create security group")  # err msg
    assert security_group, "Failed to create security group"
    return security_group

@@ -454,6 +656,9 @@ def _get_role(role_name, config):
        if exc.response.get("Error", {}).get("Code") == "NoSuchEntity":
            return None
        else:
+            handle_boto_error(
+                exc, "Failed to fetch IAM role data for {} from AWS.",
+                cf.bold(role_name))
            raise exc


@@ -467,17 +672,26 @@ def _get_instance_profile(profile_name, config):
        if exc.response.get("Error", {}).get("Code") == "NoSuchEntity":
            return None
        else:
+            handle_boto_error(
+                exc,
+                "Failed to fetch IAM instance profile data for {} from AWS.",
+                cf.bold(profile_name))
            raise exc


 def _get_key(key_name, config):
    ec2 = _resource("ec2", config)
-    for key in ec2.key_pairs.filter(Filters=[{
-            "Name": "key-name",
-            "Values": [key_name]
-    }]):
-        if key.name == key_name:
-            return key
+    try:
+        for key in ec2.key_pairs.filter(Filters=[{
+                "Name": "key-name",
+                "Values": [key_name]
+        }]):
+            if key.name == key_name:
+                return key
+    except botocore.exceptions.ClientError as exc:
+        handle_boto_error(exc, "Failed to fetch EC2 key pair {} from AWS.",
+                          cf.bold(key_name))
+        raise exc


 def _client(name, config):
@@ -15,6 +15,10 @@ from ray.autoscaler.tags import TAG_RAY_CLUSTER_NAME, TAG_RAY_NODE_NAME, \
 from ray.ray_constants import BOTO_MAX_RETRIES, BOTO_CREATE_MAX_RETRIES
 from ray.autoscaler.log_timer import LogTimer

+from ray.autoscaler.aws.utils import boto_exception_handler
+from ray.autoscaler.cli_logger import cli_logger
+import colorful as cf
+
 logger = logging.getLogger(__name__)


@@ -133,7 +137,10 @@ class AWSNodeProvider(NodeProvider):
                "Values": [v],
            })

-        nodes = list(self.ec2.instances.filter(Filters=filters))
+        with boto_exception_handler(
+                "Failed to fetch running instances from AWS."):
+            nodes = list(self.ec2.instances.filter(Filters=filters))
+
        # Populate the tag cache with initial information if necessary
        for node in nodes:
            if node.id in self.tag_cache:
@@ -228,19 +235,32 @@ class AWSNodeProvider(NodeProvider):
                self.ec2.instances.filter(Filters=filters))[:count]
            reuse_node_ids = [n.id for n in reuse_nodes]
            if reuse_nodes:
-                logger.info("AWSNodeProvider: reusing instances {}. "
-                            "To disable reuse, set "
-                            "'cache_stopped_nodes: False' in the provider "
-                            "config.".format(reuse_node_ids))
+                cli_logger.print(
+                    # todo: handle plural vs singular?
+                    "Reusing nodes {}. "
+                    "To disable reuse, set `cache_stopped_nodes: False` "
+                    "under `provider` in the cluster configuration.",
+                    cli_logger.render_list(reuse_node_ids))
+                cli_logger.old_info(
+                    logger, "AWSNodeProvider: reusing instances {}. "
+                    "To disable reuse, set "
+                    "'cache_stopped_nodes: False' in the provider "
+                    "config.", reuse_node_ids)

-                for node in reuse_nodes:
-                    self.tag_cache[node.id] = from_aws_format(
-                        {x["Key"]: x["Value"]
-                         for x in node.tags})
-                    if node.state["Name"] == "stopping":
-                        logger.info("AWSNodeProvider: waiting for instance "
-                                    "{} to fully stop...".format(node.id))
-                        node.wait_until_stopped()
+                # todo: timed?
+                with cli_logger.group("Stopping instances to reuse"):
+                    for node in reuse_nodes:
+                        self.tag_cache[node.id] = from_aws_format(
+                            {x["Key"]: x["Value"]
+                             for x in node.tags})
+                        if node.state["Name"] == "stopping":
+                            cli_logger.print("Waiting for instance {} to stop",
+                                             node.id)
+                            cli_logger.old_info(
+                                logger,
+                                "AWSNodeProvider: waiting for instance "
+                                "{} to fully stop...", node.id)
+                            node.wait_until_stopped()

                self.ec2.meta.client.start_instances(
                    InstanceIds=reuse_node_ids)
@@ -300,8 +320,11 @@ class AWSNodeProvider(NodeProvider):
        for attempt in range(1, BOTO_CREATE_MAX_RETRIES + 1):
            try:
                subnet_id = subnet_ids[self.subnet_idx % len(subnet_ids)]
-                logger.info("NodeProvider: calling create_instances "
-                            "with {} (count={}).".format(subnet_id, count))
+
+                cli_logger.old_info(
+                    logger, "NodeProvider: calling create_instances "
+                    "with {} (count={}).", subnet_id, count)
+
                self.subnet_idx += 1
                conf.update({
                    "MinCount": 1,
@@ -310,32 +333,64 @@ class AWSNodeProvider(NodeProvider):
                    "TagSpecifications": tag_specs
                })
                created = self.ec2_fail_fast.create_instances(**conf)
-                for instance in created:
-                    logger.info("NodeProvider: Created instance "
-                                "[id={}, name={}, info={}]".format(
-                                    instance.instance_id,
-                                    instance.state["Name"],
-                                    instance.state_reason["Message"]))
+
+                # todo: timed?
+                # todo: handle plurality?
+                with cli_logger.group(
+                        "Launching {} nodes",
+                        count,
+                        _tags=dict(subnet_id=subnet_id)):
+                    for instance in created:
+                        cli_logger.print(
+                            "Launched instance {}",
+                            instance.instance_id,
+                            _tags=dict(
+                                state=instance.state["Name"],
+                                info=instance.state_reason["Message"]))
+                        cli_logger.old_info(
+                            logger, "NodeProvider: Created instance "
+                            "[id={}, name={}, info={}]", instance.instance_id,
+                            instance.state["Name"],
+                            instance.state_reason["Message"])
                break
            except botocore.exceptions.ClientError as exc:
                if attempt == BOTO_CREATE_MAX_RETRIES:
-                    logger.error(
-                        "create_instances: Max attempts ({}) exceeded.".format(
-                            BOTO_CREATE_MAX_RETRIES))
+                    # todo: err msg
+                    cli_logger.abort(
+                        "Failed to launch instances. Max attempts exceeded.")
+                    cli_logger.old_error(
+                        logger,
+                        "create_instances: Max attempts ({}) exceeded.",
+                        BOTO_CREATE_MAX_RETRIES)
                    raise exc
                else:
-                    logger.error(exc)
+                    # todo: err msg
+                    cli_logger.abort(exc)
+                    cli_logger.old_error(logger, exc)

    def terminate_node(self, node_id):
        node = self._get_cached_node(node_id)
        if self.cache_stopped_nodes:
            if node.spot_instance_request_id:
-                logger.info(
+                cli_logger.print(
+                    "Terminating instance {} " +
+                    cf.gray("(cannot stop spot instances, only terminate)"),
+                    node_id)  # todo: show node name?
+
+                cli_logger.old_info(
+                    logger,
                    "AWSNodeProvider: terminating node {} (spot nodes cannot "
-                    "be stopped, only terminated)".format(node_id))
+                    "be stopped, only terminated)", node_id)
                node.terminate()
            else:
-                logger.info(
+                cli_logger.print("Stopping instance {} " + cf.gray(
+                    "(to terminate instead, "
+                    "set `cache_stopped_nodes: False` "
+                    "under `provider` in the cluster configuration)"),
+                                 node_id)  # todo: show node name?
+
+                cli_logger.old_info(
+                    logger,
                    "AWSNodeProvider: stopping node {}. To terminate nodes "
                    "on stop, set 'cache_stopped_nodes: False' in the "
                    "provider config.".format(node_id))
@@ -360,15 +415,30 @@ class AWSNodeProvider(NodeProvider):
                    on_demand_ids += [node_id]

            if on_demand_ids:
-                logger.info(
+                # todo: show node names?
+                cli_logger.print(
+                    "Stopping instances {} " + cf.gray(
+                        "(to terminate instead, "
+                        "set `cache_stopped_nodes: False` "
+                        "under `provider` in the cluster configuration)"),
+                    cli_logger.render_list(on_demand_ids))
+                cli_logger.old_info(
+                    logger,
                    "AWSNodeProvider: stopping nodes {}. To terminate nodes "
                    "on stop, set 'cache_stopped_nodes: False' in the "
-                    "provider config.".format(on_demand_ids))
+                    "provider config.", on_demand_ids)
+
                self.ec2.meta.client.stop_instances(InstanceIds=on_demand_ids)
            if spot_ids:
-                logger.info(
+                cli_logger.print(
+                    "Terminating instances {} " +
+                    cf.gray("(cannot stop spot instances, only terminate)"),
+                    cli_logger.render_list(spot_ids))
+                cli_logger.old_info(
+                    logger,
                    "AWSNodeProvider: terminating nodes {} (spot nodes cannot "
-                    "be stopped, only terminated)".format(spot_ids))
+                    "be stopped, only terminated)", spot_ids)
+
                self.ec2.meta.client.terminate_instances(InstanceIds=spot_ids)
        else:
            self.ec2.meta.client.terminate_instances(InstanceIds=node_ids)
@@ -1,5 +1,8 @@
 from collections import defaultdict

+from ray.autoscaler.cli_logger import cli_logger
+import colorful as cf
+

 class LazyDefaultDict(defaultdict):
    """
@@ -21,3 +24,99 @@ class LazyDefaultDict(defaultdict):
        """
        self[key] = self.default_factory(key)
        return self[key]
+
+
+def handle_boto_error(exc, msg, *args, **kwargs):
+    if cli_logger.old_style:
+        # old-style logging doesn't do anything here
+        # so we exit early
+        return
+
+    error_code = None
+    error_info = None
+    # todo: not sure if these exceptions always have response
+    if hasattr(exc, "response"):
+        error_info = exc.response.get("Error", None)
+    if error_info is not None:
+        error_code = error_info.get("Code", None)
+
+    generic_message_args = [
+        "{}\n"
+        "Error code: {}",
+        msg.format(*args, **kwargs),
+        cf.bold(error_code)
+    ]
+
+    # apparently
+    # ExpiredTokenException
+    # ExpiredToken
+    # RequestExpired
+    # are all the same pretty much
+    credentials_expiration_codes = [
+        "ExpiredTokenException", "ExpiredToken", "RequestExpired"
+    ]
+
+    if error_code in credentials_expiration_codes:
+        # "An error occurred (ExpiredToken) when calling the
+        # GetInstanceProfile operation: The security token
+        # included in the request is expired"
+
+        # "An error occurred (RequestExpired) when calling the
+        # DescribeKeyPairs operation: Request has expired."
+
+        token_command = (
+            "aws sts get-session-token "
+            "--serial-number arn:aws:iam::" + cf.underlined("ROOT_ACCOUNT_ID")
+            + ":mfa/" + cf.underlined("AWS_USERNAME") + " --token-code " +
+            cf.underlined("TWO_FACTOR_AUTH_CODE"))
+
+        secret_key_var = (
+            "export AWS_SECRET_ACCESS_KEY = " + cf.underlined("REPLACE_ME") +
+            " # found at Credentials.SecretAccessKey")
+        session_token_var = (
+            "export AWS_SESSION_TOKEN = " + cf.underlined("REPLACE_ME") +
+            " # found at Credentials.SessionToken")
+        access_key_id_var = (
+            "export AWS_ACCESS_KEY_ID = " + cf.underlined("REPLACE_ME") +
+            " # found at Credentials.AccessKeyId")
+
+        # fixme: replace with a Github URL that points
+        # to our repo
+        aws_session_script_url = ("https://gist.github.com/maximsmol/"
+                                  "a0284e1d97b25d417bd9ae02e5f450cf")
+
+        cli_logger.verbose_error(*generic_message_args)
+        cli_logger.verbose(vars(exc))
+
+        cli_logger.abort(
+            "Your AWS session has expired.\n\n"
+            "You can request a new one using\n{}\n"
+            "then expose it to Ray by setting\n{}\n{}\n{}\n\n"
+            "You can find a script that automates this at:\n{}",
+            cf.bold(token_command), cf.bold(secret_key_var),
+            cf.bold(session_token_var), cf.bold(access_key_id_var),
+            cf.underlined(aws_session_script_url))
+
+    # todo: any other errors that we should catch separately?
+
+    cli_logger.error(*generic_message_args)
+    cli_logger.newline()
+    with cli_logger.verbatim_error_ctx("Boto3 error:"):
+        cli_logger.verbose(vars(exc))
+        cli_logger.error(exc)
+    cli_logger.abort()
+
+
+def boto_exception_handler(msg, *args, **kwargs):
+    # todo: implement timer
+    class ExceptionHandlerContextManager():
+        def __enter__(self):
+            pass
+
+        def __exit__(self, type, value, tb):
+            import botocore
+
+            if type is botocore.exceptions.ClientError:
+                handle_boto_error(value, msg, *args, **kwargs)
+
+    return ExceptionHandlerContextManager()