Merge pull request #209 from Yelp/mpiano/CLUSTERMAN-691

Author: piax93

clusterman/autoscaler/pool_manager.py

@@ -84,12 +84,13 @@ def __init__(
         if fetch_state:
             self.reload_state()
 
-    def reload_state(self) -> None:
+    def reload_state(self, **cluster_connector_kwargs) -> None:
         """Fetch any state that may have changed behind our back, but which we do not want to change during an
         ``Autoscaler.run()``.
         """
         logger.info("Reloading cluster connector state")
-        self.cluster_connector.reload_state()
+        # TODO: update mypy to avoid having to ignore this error (CLUSTERMAN-692)
+        self.cluster_connector.reload_state(**cluster_connector_kwargs)  # type: ignore
 
         logger.info("Reloading resource groups")
         self._reload_resource_groups()

clusterman/kubernetes/kubernetes_cluster_connector.py

@@ -94,20 +94,23 @@ def __init__(self, cluster: str, pool: Optional[str], init_crd: bool = False) ->
             )
             self._label_selectors.append(f"{node_label_selector}={self.pool}")
 
-    def reload_state(self) -> None:
+    def reload_state(self, load_pods_info: bool = True) -> None:
         logger.info("Reloading nodes")
 
         self.reload_client()
 
         # store the previous _nodes_by_ip for use in get_removed_nodes_before_last_reload()
         self._prev_nodes_by_ip = copy.deepcopy(self._nodes_by_ip)
         self._nodes_by_ip = self._get_nodes_by_ip()
-        logger.info("Reloading pods")
-        (self._pods_by_ip, self._unschedulable_pods, self._excluded_pods_by_ip,) = (
-            self._get_pods_info_with_label()
-            if self.pool_config.read_bool("use_labels_for_pods", default=False)
-            else self._get_pods_info()
-        )
+        if load_pods_info:
+            logger.info("Reloading pods")
+            self._pods_by_ip, self._unschedulable_pods, self._excluded_pods_by_ip = (
+                self._get_pods_info_with_label()
+                if self.pool_config.read_bool("use_labels_for_pods", default=False)
+                else self._get_pods_info()
+            )
+        else:
+            self._pods_by_ip, self._unschedulable_pods, self._excluded_pods_by_ip = ({}, [], {})
 
     def reload_client(self) -> None:
         self._core_api = CachedCoreV1Api(self.kubeconfig_path)

clusterman/migration/settings.py

@@ -78,6 +78,7 @@ class WorkerSetup(NamedTuple):
     bootstrap_timeout: float
     disable_autoscaling: bool
     expected_duration: float
+    ignore_pod_health: bool = False
 
     @classmethod
     def from_config(cls, config: dict) -> "WorkerSetup":
@@ -90,4 +91,5 @@ def from_config(cls, config: dict) -> "WorkerSetup":
             bootstrap_timeout=parse_time_interval_seconds(strat_conf.get("bootstrap_timeout", DEFAULT_NODE_BOOT_WAIT)),
             disable_autoscaling=config.get("disable_autoscaling", False),
             expected_duration=parse_time_interval_seconds(config.get("expected_duration", DEFAULT_WORKER_TIMEOUT)),
+            ignore_pod_health=config.get("ignore_pod_health", False),
         )

clusterman/migration/worker.py

@@ -66,27 +66,27 @@ class NodeMigrationError(Exception):
 
 
 def _monitor_pool_health(
-    manager: PoolManager, timeout: float, drained: Collection[ClusterNodeMetadata], check_pods: bool = True
+    manager: PoolManager, timeout: float, drained: Collection[ClusterNodeMetadata], ignore_pod_health: bool = False
 ) -> bool:
     """Monitor pool health after nodes were submitted for draining
 
     :param PoolManager manager: pool manager instance
     :param float timeout: timestamp after which giving up
     :param Collection[ClusterNodeMetadata] drained: nodes which were submitted for draining
-    :param bool check_pods: check that pods can successfully be scheduled
+    :param bool ignore_pod_health: If set, do not check that pods can successfully be scheduled
     :return: true if capacity is fulfilled
     """
     draining_happened = False
     connector = cast(KubernetesClusterConnector, manager.cluster_connector)
     while time.time() < timeout:
-        manager.reload_state()
+        manager.reload_state(load_pods_info=not ignore_pod_health)
         draining_happened = draining_happened or not any(
             node.agent.agent_id == connector.get_agent_metadata(node.instance.ip_address).agent_id for node in drained
         )
         if (
             draining_happened
             and manager.is_capacity_satisfied()
-            and (not check_pods or connector.has_enough_capacity_for_pods())
+            and (ignore_pod_health or connector.has_enough_capacity_for_pods())
         ):
             return True
         time.sleep(HEALTH_CHECK_INTERVAL_SECONDS)
@@ -114,7 +114,9 @@ def _drain_node_selection(
             logger.info(f"Recycling node {node.instance.instance_id}")
             manager.submit_for_draining(node)
         time.sleep(worker_setup.bootstrap_wait)
-        if not _monitor_pool_health(manager, start_time + worker_setup.bootstrap_timeout, selection_chunk):
+        if not _monitor_pool_health(
+            manager, start_time + worker_setup.bootstrap_timeout, selection_chunk, worker_setup.ignore_pod_health
+        ):
             logger.warning(
                 f"Pool {manager.cluster}:{manager.pool} did not come back"
                 " to desired capacity, stopping selection draining"
@@ -145,7 +147,7 @@ def uptime_migration_worker(
         else:
             logger.warning(f"Pool {cluster}:{pool} is currently underprovisioned, skipping uptime migration iteration")
         time.sleep(UPTIME_CHECK_INTERVAL_SECONDS)
-        manager.reload_state()
+        manager.reload_state(load_pods_info=not worker_setup.ignore_pod_health)
 
 
 def event_migration_worker(migration_event: MigrationEvent, worker_setup: WorkerSetup, pool_lock: LockBase) -> None:
@@ -158,7 +160,7 @@ def event_migration_worker(migration_event: MigrationEvent, worker_setup: Worker
     manager = PoolManager(migration_event.cluster, migration_event.pool, SUPPORTED_POOL_SCHEDULER, fetch_state=False)
     connector = cast(KubernetesClusterConnector, manager.cluster_connector)
     connector.set_label_selectors(migration_event.label_selectors, add_to_existing=True)
-    manager.reload_state()
+    manager.reload_state(load_pods_info=not worker_setup.ignore_pod_health)
     try:
         pool_lock.acquire(timeout=worker_setup.expected_duration)
         pool_lock_acquired = True
@@ -177,7 +179,10 @@ def event_migration_worker(migration_event: MigrationEvent, worker_setup: Worker
             prescaled_capacity = round(manager.target_capacity + (offset * avg_weight))
             manager.modify_target_capacity(prescaled_capacity)
         if not _monitor_pool_health(
-            manager, time.time() + INITIAL_POOL_HEALTH_TIMEOUT_SECONDS, drained=[], check_pods=False
+            manager,
+            time.time() + INITIAL_POOL_HEALTH_TIMEOUT_SECONDS,
+            drained=[],
+            ignore_pod_health=True,
         ):
             raise NodeMigrationError(f"Pool {migration_event.cluster}:{migration_event.pool} is not healthy")
         node_selector = lambda node: node.agent.agent_id and not migration_event.condition.matches(node)  # noqa

clusterman/simulator/simulated_pool_manager.py

@@ -56,7 +56,7 @@ def __init__(
             MAX_MIN_NODE_SCALEIN_UPTIME_SECONDS,
         )
 
-    def reload_state(self) -> None:
+    def reload_state(self, **cluster_connector_kwargs) -> None:
         pass
 
     def get_node_metadatas(

docs/source/metrics.rst

@@ -85,17 +85,17 @@ cpus_allocated|cluster=norcal-prod,pool=appA_pool 1502405756    22
 mem_allocated|cluster=norcal-prod,pool=appB_pool  1502405810    20
 ================================================= ========== =====
 
-+---------------------------------------------------------------------------------------------------+-------------------------+-------------------------+
-| Metadata                                                                                          |                         |                         |
-+-----------------------------------------------------+------------+--------------------------------+-------------------------+-------------------------+
-| metric name                                         | timestamp  | value                          | <c3.xlarge, us-west-2a> | <c3.xlarge, us-west-2c> |
-+=====================================================+============+================================+=========================+=========================+
++----------------------------------------------------------------------------+------------+--------------------------------+-------------------------+-------------------------+
+| Metadata                                                                   |            |                                |                         |                         |
++----------------------------------------------------------------------------+------------+--------------------------------+-------------------------+-------------------------+
+| metric name                                                                | timestamp  | value                          | <c3.xlarge, us-west-2a> | <c3.xlarge, us-west-2c> |
++============================================================================+============+================================+=========================+=========================+
 | spot_prices|aws_availability_zone=us-west-2a,aws_instance_type=c3.xlarge   | 1502405756 | 1.30                           |                         |                         |
-+-----------------------------------------------------+------------+--------------------------------+-------------------------+-------------------------+
++----------------------------------------------------------------------------+------------+--------------------------------+-------------------------+-------------------------+
 | spot_prices|aws_availability_zone=us-west-2c,aws_instance_type=c3.xlarge   | 1502405756 | 5.27                           |                         |                         |
-+-----------------------------------------------------+------------+--------------------------------+-------------------------+-------------------------+
-| fulfilled_capacity|cluster=norcal-prod,pool=seagull | 1502409314 |                                |                       4 |                      20 |
-+-----------------------------------------------------+------------+--------------------------------+-------------------------+-------------------------+
++----------------------------------------------------------------------------+------------+--------------------------------+-------------------------+-------------------------+
+| fulfilled_capacity|cluster=norcal-prod,pool=seagull                        | 1502409314 |                                |                       4 |                      20 |
++----------------------------------------------------------------------------+------------+--------------------------------+-------------------------+-------------------------+
 
 .. _metric_name_reference:
 

docs/source/node_migration.rst

@@ -42,6 +42,8 @@ The allowed values for the migration settings are as follows:
 
 * ``disable_autoscaling``: turn off autoscaler while recycling instances (false by default).
 
+* ``ignore_pod_health``: avoid loading and checking pod information to determine pool health (false by default).
+
 * ``expected_duration``: estimated duration for migration of the whole pool; human readable time string (1 day by default).
 
 See :ref:`pool_configuration` for how an example configuration block would look like.

tests/migration/migration_worker_test.py

@@ -108,13 +108,15 @@ def test_drain_node_selection(mock_monitor, mock_time):
                     ClusterNodeMetadata(AgentMetadata(agent_id=5, task_count=20), InstanceMetadata(None, None)),
                     ClusterNodeMetadata(AgentMetadata(agent_id=4, task_count=22), InstanceMetadata(None, None)),
                 ],
+                False,
             ),
             call(
                 mock_manager,
                 3,
                 [
                     ClusterNodeMetadata(AgentMetadata(agent_id=3, task_count=24), InstanceMetadata(None, None)),
                 ],
+                False,
             ),
         ]
     )