[ML] Create endpoint cache feature

We want to gate the inference endpoint cache behind a NodeFeature so
that it will only enable if every node in the cluster can support it.
This closes the risk that a master node can be an earlier version of the
cache, potentially handling the ClearCache action without having the
code for it.

Resolve #134809

Author: prwhelan

qa/lucene-index-compatibility/src/javaRestTest/java/org/elasticsearch/lucene/AbstractIndexCompatibilityTestCase.java

@@ -26,9 +26,7 @@
 import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.test.XContentTestUtils;
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
-import org.elasticsearch.test.cluster.local.DefaultSettingsProvider;
 import org.elasticsearch.test.cluster.local.LocalClusterConfigProvider;
-import org.elasticsearch.test.cluster.local.LocalClusterSpec;
 import org.elasticsearch.test.cluster.local.distribution.DistributionType;
 import org.elasticsearch.test.cluster.util.Version;
 import org.elasticsearch.test.rest.ESRestTestCase;
@@ -80,16 +78,6 @@ public abstract class AbstractIndexCompatibilityTestCase extends ESRestTestCase
         .setting("xpack.security.enabled", "false")
         .setting("xpack.ml.enabled", "false")
         .setting("path.repo", () -> REPOSITORY_PATH.getRoot().getPath())
-        .settings(new DefaultSettingsProvider() {
-            @Override
-            public Map<String, String> get(LocalClusterSpec.LocalNodeSpec nodeSpec) {
-                var settings = super.get(nodeSpec);
-                if (nodeSpec.getVersion().onOrAfter(Version.fromString("9.2.0"))) {
-                    settings.put("xpack.inference.endpoint.cache.enabled", "false");
-                }
-                return settings;
-            }
-        })
         .apply(() -> clusterConfig)
         .build();
 

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/InferenceFeatures.java

@@ -50,6 +50,13 @@ public class InferenceFeatures implements FeatureSpecification {
     public static final NodeFeature SEMANTIC_TEXT_HIGHLIGHTING_FLAT = new NodeFeature("semantic_text.highlighter.flat_index_options");
     private static final NodeFeature SEMANTIC_TEXT_FIELDS_CHUNKS_FORMAT = new NodeFeature("semantic_text.fields_chunks_format");
 
+    public static final NodeFeature INFERENCE_ENDPOINT_CACHE = new NodeFeature("inference.endpoint.cache");
+
+    @Override
+    public Set<NodeFeature> getFeatures() {
+        return Set.of(INFERENCE_ENDPOINT_CACHE);
+    }
+
     @Override
     public Set<NodeFeature> getTestFeatures() {
         var testFeatures = new HashSet<>(
@@ -90,6 +97,7 @@ public Set<NodeFeature> getTestFeatures() {
                 TEXT_SIMILARITY_RERANKER_SNIPPETS
             )
         );
+        testFeatures.addAll(getFeatures());
         return testFeatures;
     }
 }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/InferencePlugin.java

@@ -391,7 +391,8 @@ public Collection<?> createComponents(PluginServices services) {
                 settings,
                 modelRegistry.get(),
                 serviceRegistry,
-                services.projectResolver()
+                services.projectResolver(),
+                services.featureService()
             )
         );
 

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/registry/InferenceEndpointRegistry.java

@@ -19,11 +19,14 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.features.FeatureService;
 import org.elasticsearch.inference.InferenceServiceRegistry;
 import org.elasticsearch.inference.Model;
+import org.elasticsearch.xpack.inference.InferenceFeatures;
 
 import java.util.Collection;
 import java.util.List;
+import java.util.function.Supplier;
 
 /**
  * A registry that assembles and caches Inference Endpoints, {@link Model}, for reuse.
@@ -64,14 +67,16 @@ public static Collection<? extends Setting<?>> getSettingsDefinitions() {
     private final InferenceServiceRegistry serviceRegistry;
     private final ProjectResolver projectResolver;
     private final Cache<InferenceIdAndProject, Model> cache;
-    private volatile boolean cacheEnabled;
+    private final Supplier<Boolean> cacheEnabledViaFeature;
+    private volatile boolean cacheEnabledViaSetting;
 
     public InferenceEndpointRegistry(
         ClusterService clusterService,
         Settings settings,
         ModelRegistry modelRegistry,
         InferenceServiceRegistry serviceRegistry,
-        ProjectResolver projectResolver
+        ProjectResolver projectResolver,
+        FeatureService featureService
     ) {
         this.modelRegistry = modelRegistry;
         this.serviceRegistry = serviceRegistry;
@@ -80,15 +85,19 @@ public InferenceEndpointRegistry(
             .setMaximumWeight(INFERENCE_ENDPOINT_CACHE_WEIGHT.get(settings))
             .setExpireAfterWrite(INFERENCE_ENDPOINT_CACHE_EXPIRY.get(settings))
             .build();
-        this.cacheEnabled = INFERENCE_ENDPOINT_CACHE_ENABLED.get(settings);
+        this.cacheEnabledViaFeature = () -> {
+            var state = clusterService.state();
+            return state.clusterRecovered() && featureService.clusterHasFeature(state, InferenceFeatures.INFERENCE_ENDPOINT_CACHE);
+        };
+        this.cacheEnabledViaSetting = INFERENCE_ENDPOINT_CACHE_ENABLED.get(settings);
 
         clusterService.getClusterSettings()
-            .addSettingsUpdateConsumer(INFERENCE_ENDPOINT_CACHE_ENABLED, enabled -> this.cacheEnabled = enabled);
+            .addSettingsUpdateConsumer(INFERENCE_ENDPOINT_CACHE_ENABLED, enabled -> this.cacheEnabledViaSetting = enabled);
     }
 
     public void getEndpoint(String inferenceEntityId, ActionListener<Model> listener) {
         var key = new InferenceIdAndProject(inferenceEntityId, projectResolver.getProjectId());
-        var cachedModel = cacheEnabled ? cache.get(key) : null;
+        var cachedModel = cacheEnabled() ? cache.get(key) : null;
         if (cachedModel != null) {
             log.trace("Retrieved [{}] from cache.", inferenceEntityId);
             listener.onResponse(cachedModel);
@@ -98,7 +107,7 @@ public void getEndpoint(String inferenceEntityId, ActionListener<Model> listener
     }
 
     void invalidateAll(ProjectId projectId) {
-        if (cacheEnabled) {
+        if (cacheEnabled()) {
             var cacheKeys = cache.keys().iterator();
             while (cacheKeys.hasNext()) {
                 if (cacheKeys.next().projectId.equals(projectId)) {
@@ -126,23 +135,23 @@ private void loadFromIndex(InferenceIdAndProject idAndProject, ActionListener<Mo
                 unparsedModel.secrets()
             );
 
-            if (cacheEnabled) {
+            if (cacheEnabled()) {
                 cache.put(idAndProject, model);
             }
             l.onResponse(model);
         }));
     }
 
     public Cache.Stats stats() {
-        return cacheEnabled ? cache.stats() : EMPTY;
+        return cacheEnabled() ? cache.stats() : EMPTY;
     }
 
     public int cacheCount() {
-        return cacheEnabled ? cache.count() : 0;
+        return cacheEnabled() ? cache.count() : 0;
     }
 
     public boolean cacheEnabled() {
-        return cacheEnabled;
+        return cacheEnabledViaSetting && cacheEnabledViaFeature.get();
     }
 
     private record InferenceIdAndProject(String inferenceEntityId, ProjectId projectId) {}