Disable validation of cluster config on the cluster to allow for cluster configs with new properties. (#11693)

python/ray/autoscaler/_private/autoscaler.py

@@ -325,7 +325,17 @@ class StandardAutoscaler:
         try:
             with open(self.config_path) as f:
                 new_config = yaml.safe_load(f.read())
-            validate_config(new_config)
+            if new_config != getattr(self, "config", None):
+                try:
+                    validate_config(new_config)
+                except Exception as e:
+                    logger.debug(
+                        "Cluster config validation failed. The version of "
+                        "the ray CLI you launched this cluster with may "
+                        "be higher than the version of ray being run on "
+                        "the cluster. Some new features may not be "
+                        "available until you upgrade ray on your cluster.",
+                        exc_info=e)
             (new_runtime_hash,
              new_file_mounts_contents_hash) = hash_runtime_conf(
                  new_config["file_mounts"],

python/ray/tests/test_autoscaler.py

@@ -427,11 +427,26 @@ class AutoscalingTest(unittest.TestCase):
             f.write(yaml.dump(config))
         return path
 
-    def testInvalidConfig(self):
-        invalid_config = os.devnull
-        with pytest.raises(ValueError):
-            StandardAutoscaler(
-                invalid_config, LoadMetrics(), update_interval_s=0)
+    def testAutoscalerConfigValidationFailNotFatal(self):
+        invalid_config = {**SMALL_CLUSTER, "invalid_property_12345": "test"}
+        # First check that this config is actually invalid
+        with pytest.raises(ValidationError):
+            validate_config(invalid_config)
+
+        config_path = self.write_config(invalid_config)
+        self.provider = MockProvider()
+        runner = MockProcessRunner()
+        autoscaler = StandardAutoscaler(
+            config_path,
+            LoadMetrics(),
+            max_failures=0,
+            process_runner=runner,
+            update_interval_s=0)
+        assert len(self.provider.non_terminated_nodes({})) == 0
+        autoscaler.update()
+        self.waitForNodes(2)
+        autoscaler.update()
+        self.waitForNodes(2)
 
     def testValidation(self):
         """Ensures that schema validation is working."""