def testDelayedLaunch(self):
        config_path = self.write_config(SMALL_CLUSTER)
        self.provider = MockProvider()
        autoscaler = StandardAutoscaler(
            config_path,
            LoadMetrics(),
            max_launch_batch=5,
            max_concurrent_launches=5,
            max_failures=0,
            update_interval_s=0)
        assert len(self.provider.non_terminated_nodes({})) == 0

        # Update will try to create, but will block until we set the flag
        self.provider.ready_to_create.clear()
        autoscaler.update()
        assert autoscaler.num_launches_pending.value == 2
        assert len(self.provider.non_terminated_nodes({})) == 0

        # Set the flag, check it updates
        self.provider.ready_to_create.set()
        self.waitForNodes(2)
        assert autoscaler.num_launches_pending.value == 0

        # Update the config to reduce the cluster size
        new_config = SMALL_CLUSTER.copy()
        new_config["max_workers"] = 1
        self.write_config(new_config)
        autoscaler.update()
        assert len(self.provider.non_terminated_nodes({})) == 1

 def testMaxFailures(self):
     config_path = self.write_config(SMALL_CLUSTER)
     self.provider = MockProvider()
     self.provider.throw = True
     autoscaler = StandardAutoscaler(
         config_path, LoadMetrics(), max_failures=2, update_interval_s=0)
     autoscaler.update()
     autoscaler.update()
     self.assertRaises(Exception, autoscaler.update)

 def testScaleUp(self):
     config_path = self.write_config(SMALL_CLUSTER)
     self.provider = MockProvider()
     autoscaler = StandardAutoscaler(
         config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
     assert len(self.provider.non_terminated_nodes({})) == 0
     autoscaler.update()
     self.waitForNodes(2)
     autoscaler.update()
     self.waitForNodes(2)

 def testScaleUp(self):
     config_path = self.write_config(SMALL_CLUSTER)
     self.provider = MockProvider()
     autoscaler = StandardAutoscaler(
         config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
     self.assertEqual(len(self.provider.nodes({})), 0)
     autoscaler.update()
     self.assertEqual(len(self.provider.nodes({})), 2)
     autoscaler.update()
     self.assertEqual(len(self.provider.nodes({})), 2)

    def testDontScaleBelowTarget(self):
        config = SMALL_CLUSTER.copy()
        config["min_workers"] = 0
        config["max_workers"] = 2
        config["target_utilization_fraction"] = 0.5
        config_path = self.write_config(config)
        self.provider = MockProvider()
        lm = LoadMetrics()
        autoscaler = StandardAutoscaler(
            config_path, lm, max_failures=0, update_interval_s=0)
        assert len(self.provider.non_terminated_nodes({})) == 0
        autoscaler.update()
        assert autoscaler.num_launches_pending.value == 0
        assert len(self.provider.non_terminated_nodes({})) == 0

        # Scales up as nodes are reported as used
        local_ip = services.get_node_ip_address()
        lm.update(local_ip, {"CPU": 2}, {"CPU": 0})  # head
        # 1.0 nodes used => target nodes = 2 => target workers = 1
        autoscaler.update()
        self.waitForNodes(1)

        # Make new node idle, and never used.
        # Should hold steady as target is still 2.
        lm.update("172.0.0.0", {"CPU": 0}, {"CPU": 0})
        lm.last_used_time_by_ip["172.0.0.0"] = 0
        autoscaler.update()
        assert len(self.provider.non_terminated_nodes({})) == 1

        # Reduce load on head => target nodes = 1 => target workers = 0
        lm.update(local_ip, {"CPU": 2}, {"CPU": 1})
        autoscaler.update()
        assert len(self.provider.non_terminated_nodes({})) == 0

    def testDynamicScaling(self):
        config_path = self.write_config(SMALL_CLUSTER)
        self.provider = MockProvider()
        autoscaler = StandardAutoscaler(
            config_path, LoadMetrics(), max_concurrent_launches=5,
            max_failures=0, update_interval_s=0)
        self.assertEqual(len(self.provider.nodes({})), 0)
        autoscaler.update()
        self.assertEqual(len(self.provider.nodes({})), 2)

        # Update the config to reduce the cluster size
        new_config = SMALL_CLUSTER.copy()
        new_config["max_workers"] = 1
        self.write_config(new_config)
        autoscaler.update()
        self.assertEqual(len(self.provider.nodes({})), 1)

        # Update the config to reduce the cluster size
        new_config["min_workers"] = 10
        new_config["max_workers"] = 10
        self.write_config(new_config)
        autoscaler.update()
        self.assertEqual(len(self.provider.nodes({})), 6)
        autoscaler.update()
        self.assertEqual(len(self.provider.nodes({})), 10)

 def testUpdateThrottling(self):
     config_path = self.write_config(SMALL_CLUSTER)
     self.provider = MockProvider()
     autoscaler = StandardAutoscaler(
         config_path, LoadMetrics(), max_concurrent_launches=5,
         max_failures=0, update_interval_s=10)
     autoscaler.update()
     self.assertEqual(len(self.provider.nodes({})), 2)
     new_config = SMALL_CLUSTER.copy()
     new_config["max_workers"] = 1
     self.write_config(new_config)
     autoscaler.update()
     self.assertEqual(len(self.provider.nodes({})), 2)  # not updated yet

 def testLaunchNewNodeOnOutOfBandTerminate(self):
     config_path = self.write_config(SMALL_CLUSTER)
     self.provider = MockProvider()
     autoscaler = StandardAutoscaler(
         config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
     autoscaler.update()
     autoscaler.update()
     self.assertEqual(len(self.provider.nodes({})), 2)
     for node in self.provider.mock_nodes.values():
         node.state = "terminated"
     self.assertEqual(len(self.provider.nodes({})), 0)
     autoscaler.update()
     self.assertEqual(len(self.provider.nodes({})), 2)

    def testDelayedLaunchWithFailure(self):
        config = SMALL_CLUSTER.copy()
        config["min_workers"] = 10
        config["max_workers"] = 10
        config_path = self.write_config(config)
        self.provider = MockProvider()
        autoscaler = StandardAutoscaler(
            config_path,
            LoadMetrics(),
            max_launch_batch=5,
            max_concurrent_launches=8,
            max_failures=0,
            update_interval_s=0)
        assert len(self.provider.non_terminated_nodes({})) == 0

        # update() should launch a wave of 5 nodes (max_launch_batch)
        # Force this first wave to block.
        rtc1 = self.provider.ready_to_create
        rtc1.clear()
        autoscaler.update()
        # Synchronization: wait for launchy thread to be blocked on rtc1
        if hasattr(rtc1, '_cond'):  # Python 3.5
            waiters = rtc1._cond._waiters
        else:  # Python 2.7
            waiters = rtc1._Event__cond._Condition__waiters
        self.waitFor(lambda: len(waiters) == 1)
        assert autoscaler.num_launches_pending.value == 5
        assert len(self.provider.non_terminated_nodes({})) == 0

        # Call update() to launch a second wave of 3 nodes,
        # as 5 + 3 = 8 = max_concurrent_launches.
        # Make this wave complete immediately.
        rtc2 = threading.Event()
        self.provider.ready_to_create = rtc2
        rtc2.set()
        autoscaler.update()
        self.waitForNodes(3)
        assert autoscaler.num_launches_pending.value == 5

        # The first wave of 5 will now tragically fail
        self.provider.fail_creates = True
        rtc1.set()
        self.waitFor(lambda: autoscaler.num_launches_pending.value == 0)
        assert len(self.provider.non_terminated_nodes({})) == 3

        # Retry the first wave, allowing it to succeed this time
        self.provider.fail_creates = False
        autoscaler.update()
        self.waitForNodes(8)
        assert autoscaler.num_launches_pending.value == 0

        # Final wave of 2 nodes
        autoscaler.update()
        self.waitForNodes(10)
        assert autoscaler.num_launches_pending.value == 0

 def testConfiguresNewNodes(self):
     config_path = self.write_config(SMALL_CLUSTER)
     self.provider = MockProvider()
     runner = MockProcessRunner()
     autoscaler = StandardAutoscaler(
         config_path,
         LoadMetrics(),
         max_failures=0,
         process_runner=runner,
         update_interval_s=0)
     autoscaler.update()
     autoscaler.update()
     self.waitForNodes(2)
     for node in self.provider.mock_nodes.values():
         node.state = "running"
     autoscaler.update()
     self.waitForNodes(2, tag_filters={TAG_RAY_NODE_STATUS: "up-to-date"})

    def testRecoverUnhealthyWorkers(self):
        config_path = self.write_config(SMALL_CLUSTER)
        self.provider = MockProvider()
        runner = MockProcessRunner()
        lm = LoadMetrics()
        autoscaler = StandardAutoscaler(
            config_path,
            lm,
            max_failures=0,
            process_runner=runner,
            verbose_updates=True,
            node_updater_cls=NodeUpdaterThread,
            update_interval_s=0)
        autoscaler.update()
        self.waitForNodes(2)
        for node in self.provider.mock_nodes.values():
            node.state = "running"
        autoscaler.update()
        self.waitForNodes(2, tag_filters={TAG_RAY_NODE_STATUS: "up-to-date"})

        # Mark a node as unhealthy
        lm.last_heartbeat_time_by_ip["172.0.0.0"] = 0
        num_calls = len(runner.calls)
        autoscaler.update()
        self.waitFor(lambda: len(runner.calls) > num_calls)

    def testIgnoresCorruptedConfig(self):
        config_path = self.write_config(SMALL_CLUSTER)
        self.provider = MockProvider()
        autoscaler = StandardAutoscaler(
            config_path,
            LoadMetrics(),
            max_launch_batch=10,
            max_concurrent_launches=10,
            max_failures=0,
            update_interval_s=0)
        autoscaler.update()
        self.waitForNodes(2)

        # Write a corrupted config
        self.write_config("asdf")
        for _ in range(10):
            autoscaler.update()
        time.sleep(0.1)
        assert autoscaler.num_launches_pending.value == 0
        assert len(self.provider.non_terminated_nodes({})) == 2

        # New a good config again
        new_config = SMALL_CLUSTER.copy()
        new_config["min_workers"] = 10
        new_config["max_workers"] = 10
        self.write_config(new_config)
        autoscaler.update()
        self.waitForNodes(10)

 def testConfiguresOutdatedNodes(self):
     config_path = self.write_config(SMALL_CLUSTER)
     self.provider = MockProvider()
     runner = MockProcessRunner()
     autoscaler = StandardAutoscaler(
         config_path,
         LoadMetrics(),
         max_failures=0,
         process_runner=runner,
         verbose_updates=True,
         node_updater_cls=NodeUpdaterThread,
         update_interval_s=0)
     autoscaler.update()
     autoscaler.update()
     self.waitForNodes(2)
     for node in self.provider.mock_nodes.values():
         node.state = "running"
     autoscaler.update()
     self.waitForNodes(2, tag_filters={TAG_RAY_NODE_STATUS: "up-to-date"})
     runner.calls = []
     new_config = SMALL_CLUSTER.copy()
     new_config["worker_setup_commands"] = ["cmdX", "cmdY"]
     self.write_config(new_config)
     autoscaler.update()
     autoscaler.update()
     self.waitFor(lambda: len(runner.calls) > 0)

    def testLaunchConfigChange(self):
        config_path = self.write_config(SMALL_CLUSTER)
        self.provider = MockProvider()
        autoscaler = StandardAutoscaler(
            config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
        autoscaler.update()
        self.assertEqual(len(self.provider.nodes({})), 2)

        # Update the config to change the node type
        new_config = SMALL_CLUSTER.copy()
        new_config["worker_nodes"]["InstanceType"] = "updated"
        self.write_config(new_config)
        existing_nodes = set(self.provider.nodes({}))
        for _ in range(5):
            autoscaler.update()
        new_nodes = set(self.provider.nodes({}))
        self.assertEqual(len(new_nodes), 2)
        self.assertEqual(len(new_nodes.intersection(existing_nodes)), 0)

    def testLaunchConfigChange(self):
        config_path = self.write_config(SMALL_CLUSTER)
        self.provider = MockProvider()
        autoscaler = StandardAutoscaler(
            config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
        autoscaler.update()
        self.waitForNodes(2)

        # Update the config to change the node type
        new_config = SMALL_CLUSTER.copy()
        new_config["worker_nodes"]["InstanceType"] = "updated"
        self.write_config(new_config)
        self.provider.ready_to_create.clear()
        for _ in range(5):
            autoscaler.update()
        self.waitForNodes(0)
        self.provider.ready_to_create.set()
        self.waitForNodes(2)

 def testReportsConfigFailures(self):
     config_path = self.write_config(SMALL_CLUSTER)
     self.provider = MockProvider()
     runner = MockProcessRunner(fail_cmds=["cmd1"])
     autoscaler = StandardAutoscaler(
         config_path, LoadMetrics(), max_failures=0, process_runner=runner,
         verbose_updates=True, node_updater_cls=NodeUpdaterThread,
         update_interval_s=0)
     autoscaler.update()
     autoscaler.update()
     self.assertEqual(len(self.provider.nodes({})), 2)
     for node in self.provider.mock_nodes.values():
         node.state = "running"
     assert len(self.provider.nodes(
         {TAG_RAY_NODE_STATUS: "Uninitialized"})) == 2
     autoscaler.update()
     self.waitFor(
         lambda: len(self.provider.nodes(
             {TAG_RAY_NODE_STATUS: "UpdateFailed"})) == 2)

    def testTerminateOutdatedNodesGracefully(self):
        config = SMALL_CLUSTER.copy()
        config["min_workers"] = 5
        config["max_workers"] = 5
        config_path = self.write_config(config)
        self.provider = MockProvider()
        self.provider.create_node({}, {TAG_RAY_NODE_TYPE: "Worker"}, 10)
        autoscaler = StandardAutoscaler(
            config_path, LoadMetrics(), max_failures=0, update_interval_s=0)
        self.assertEqual(len(self.provider.nodes({})), 10)

        # Gradually scales down to meet target size, never going too low
        for _ in range(10):
            autoscaler.update()
            self.assertLessEqual(len(self.provider.nodes({})), 5)
            self.assertGreaterEqual(len(self.provider.nodes({})), 4)

        # Eventually reaches steady state
        self.assertEqual(len(self.provider.nodes({})), 5)

 def testUpdateThrottling(self):
     config_path = self.write_config(SMALL_CLUSTER)
     self.provider = MockProvider()
     autoscaler = StandardAutoscaler(
         config_path,
         LoadMetrics(),
         max_launch_batch=5,
         max_concurrent_launches=5,
         max_failures=0,
         update_interval_s=10)
     autoscaler.update()
     self.waitForNodes(2)
     assert autoscaler.num_launches_pending.value == 0
     new_config = SMALL_CLUSTER.copy()
     new_config["max_workers"] = 1
     self.write_config(new_config)
     autoscaler.update()
     # not updated yet
     # note that node termination happens in the main thread, so
     # we do not need to add any delay here before checking
     assert len(self.provider.non_terminated_nodes({})) == 2
     assert autoscaler.num_launches_pending.value == 0

 def testConfiguresNewNodes(self):
     config_path = self.write_config(SMALL_CLUSTER)
     self.provider = MockProvider()
     runner = MockProcessRunner()
     autoscaler = StandardAutoscaler(
         config_path,
         LoadMetrics(),
         max_failures=0,
         process_runner=runner,
         verbose_updates=True,
         node_updater_cls=NodeUpdaterThread,
         update_interval_s=0)
     autoscaler.update()
     autoscaler.update()
     self.waitForNodes(2)
     for node in self.provider.mock_nodes.values():
         node.state = "running"
     assert len(
         self.provider.nodes({
             TAG_RAY_NODE_STATUS: "uninitialized"
         })) == 2
     autoscaler.update()
     self.waitForNodes(2, tag_filters={TAG_RAY_NODE_STATUS: "up-to-date"})

    def __init__(self, redis_address, autoscaling_config, redis_password=None):
        # Initialize the Redis clients.
        self.state = ray.experimental.state.GlobalState()
        redis_ip_address = get_ip_address(args.redis_address)
        redis_port = get_port(args.redis_address)
        self.state._initialize_global_state(
            redis_ip_address, redis_port, redis_password=redis_password)
        self.redis = ray.services.create_redis_client(
            redis_address, password=redis_password)
        # Setup subscriptions to the primary Redis server and the Redis shards.
        self.primary_subscribe_client = self.redis.pubsub(
            ignore_subscribe_messages=True)
        # Keep a mapping from local scheduler client ID to IP address to use
        # for updating the load metrics.
        self.local_scheduler_id_to_ip_map = {}
        self.load_metrics = LoadMetrics()
        if autoscaling_config:
            self.autoscaler = StandardAutoscaler(autoscaling_config,
                                                 self.load_metrics)
        else:
            self.autoscaler = None

        # Experimental feature: GCS flushing.
        self.issue_gcs_flushes = "RAY_USE_NEW_GCS" in os.environ
        self.gcs_flush_policy = None
        if self.issue_gcs_flushes:
            # Data is stored under the first data shard, so we issue flushes to
            # that redis server.
            addr_port = self.redis.lrange("RedisShards", 0, -1)
            if len(addr_port) > 1:
                logger.warning(
                    "Monitor: "
                    "TODO: if launching > 1 redis shard, flushing needs to "
                    "touch shards in parallel.")
                self.issue_gcs_flushes = False
            else:
                addr_port = addr_port[0].split(b":")
                self.redis_shard = redis.StrictRedis(
                    host=addr_port[0],
                    port=addr_port[1],
                    password=redis_password)
                try:
                    self.redis_shard.execute_command("HEAD.FLUSH 0")
                except redis.exceptions.ResponseError as e:
                    logger.info(
                        "Monitor: "
                        "Turning off flushing due to exception: {}".format(
                            str(e)))
                    self.issue_gcs_flushes = False