Hotfix [nodes -> workers] + [count failed nodes as stopped]

python/ray/autoscaler/_private/autoscaler.py

@@ -303,6 +303,8 @@ def _update(self):
  " Failed to update node."
  " Node has already been terminated.")
  if nodes_to_terminate:
+ self.prom_metrics.stopped_nodes.inc(
+ len(nodes_to_terminate))
  self.provider.terminate_nodes(nodes_to_terminate)
  nodes = self.workers()
 
@@ -750,7 +752,7 @@ def workers(self):
  nodes = self.provider.non_terminated_nodes(
  tag_filters={TAG_RAY_NODE_KIND: NODE_KIND_WORKER})
  # Update running nodes gauge whenever we check workers
- self.prom_metrics.running_nodes.set(len(nodes))
+ self.prom_metrics.running_workers.set(len(nodes))
  return nodes
 
  def unmanaged_workers(self):

python/ray/autoscaler/_private/prom_metrics.py

@@ -29,9 +29,9 @@ def __init__(self, registry: CollectorRegistry = None):
  unit="nodes",
  namespace="autoscaler",
  registry=self.registry)
- self.running_nodes: Gauge = Gauge(
- "running_nodes",
- "Number of nodes running.",
+ self.running_workers: Gauge = Gauge(
+ "running_workers",
+ "Number of worker nodes running.",
  unit="nodes",
  namespace="autoscaler",
  registry=self.registry)

python/ray/tests/test_autoscaler.py

@@ -901,8 +901,8 @@ def testScaleUp(self):
  autoscaler.update()
  self.waitForNodes(2)
 
- # running_nodes metric should be set to 2
- mock_metrics.running_nodes.set.assert_called_with(2)
+ # running_workers metric should be set to 2
+ mock_metrics.running_workers.set.assert_called_with(2)
 
  def testTerminateOutdatedNodesGracefully(self):
  config = SMALL_CLUSTER.copy()
@@ -998,7 +998,7 @@ def testDynamicScaling(self):
  assert ("Removing 1 nodes of type "
  "ray-legacy-worker-node-type (max workers)." in events), events
  assert mock_metrics.stopped_nodes.inc.call_count == 1
- mock_metrics.running_nodes.set.assert_called_with(10)
+ mock_metrics.running_workers.set.assert_called_with(10)
 
  def testInitialWorkers(self):
  """initial_workers is deprecated, this tests that it is ignored."""
@@ -2305,12 +2305,14 @@ def testNodeTerminatedDuringUpdate(self):
  self.provider = MockProvider()
  runner = MockProcessRunner()
  lm = LoadMetrics()
+ mock_metrics = Mock(spec=AutoscalerPrometheusMetrics())
  autoscaler = StandardAutoscaler(
  config_path,
  lm,
  max_failures=0,
  process_runner=runner,
- update_interval_s=0)
+ update_interval_s=0,
+ prom_metrics=mock_metrics)
 
  # Scale up to two up-to-date workers
  autoscaler.update()
@@ -2388,6 +2390,8 @@ def terminate_worker_zero():
  assert set(autoscaler.workers()) == {2, 3},\
  "Unexpected node_ids"
 
+ assert len(mock_metrics.stopped_nodes.mock_calls) == 1
+
  def testProviderException(self):
  config_path = self.write_config(SMALL_CLUSTER)
  self.provider = MockProvider()

python/ray/tests/test_metrics_agent.py

@@ -54,7 +54,7 @@
 _AUTOSCALER_METRICS = [
  "autoscaler_config_validation_exceptions",
  "autoscaler_node_launch_exceptions", "autoscaler_pending_nodes",
- "autoscaler_reset_exceptions", "autoscaler_running_nodes",
+ "autoscaler_reset_exceptions", "autoscaler_running_workers",
  "autoscaler_started_nodes", "autoscaler_stopped_nodes",
  "autoscaler_update_loop_exceptions"
 ]