[GPU] Licensing checks via LicensedFeature (#139142)

This PR changes the gpu plugin license check to use a LicensedFeature, so users can get feedback messages when the license becomes invalid/expires, like for other x-pack features.
To ensure better feedback, it also updates the check logic used to return GPU-enabled vector formats/codecs with errors and warnings in case other conditions are met, but we lack a correct license.

Author: ldematte

x-pack/plugin/core/src/main/java/org/elasticsearch/license/XPackLicenseState.java

@@ -86,6 +86,7 @@ public class XPackLicenseState {
         );
         messages.put(XPackField.REDACT_PROCESSOR, new String[] { "Executing a redact processor in an ingest pipeline will fail." });
         messages.put(XPackField.INFERENCE, new String[] { "The Inference API is disabled" });
+        messages.put(XPackField.GPU_INDEXING, new String[] { "Indexing using a GPU is disabled." });
         EXPIRATION_MESSAGES = Collections.unmodifiableMap(messages);
     }
 
@@ -109,6 +110,7 @@ public class XPackLicenseState {
         messages.put(XPackField.REDACT_PROCESSOR, XPackLicenseState::redactProcessorAcknowledgementMessages);
         messages.put(XPackField.ESQL, XPackLicenseState::esqlAcknowledgementMessages);
         messages.put(XPackField.INFERENCE, XPackLicenseState::inferenceApiAcknowledgementMessages);
+        messages.put(XPackField.GPU_INDEXING, XPackLicenseState::gpuIndexingAcknowledgementMessages);
         ACKNOWLEDGMENT_MESSAGES = Collections.unmodifiableMap(messages);
     }
 
@@ -376,6 +378,22 @@ private static String[] redactProcessorAcknowledgementMessages(OperationMode cur
         return Strings.EMPTY_ARRAY;
     }
 
+    private static String[] gpuIndexingAcknowledgementMessages(OperationMode currentMode, OperationMode newMode) {
+        switch (newMode) {
+            case BASIC:
+            case STANDARD:
+            case GOLD:
+            case PLATINUM:
+                switch (currentMode) {
+                    case TRIAL:
+                    case ENTERPRISE:
+                        return new String[] { "Indexing using a GPU will be disabled" };
+                }
+                break;
+        }
+        return Strings.EMPTY_ARRAY;
+    }
+
     private static boolean isBasic(OperationMode mode) {
         return mode == OperationMode.BASIC;
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackField.java

@@ -94,6 +94,8 @@ public final class XPackField {
 
     /** Name constant for the redact processor feature. */
     public static final String REDACT_PROCESSOR = "redact_processor";
+    /** Name constant for the GPU indexing feature. */
+    public static final String GPU_INDEXING = "gpu_indexing";
     public static final String ENTERPRISE_GEOIP_DOWNLOADER = "enterprise_geoip_downloader";
     /** Name for Universal Profiling. */
     public static final String UNIVERSAL_PROFILING = "universal_profiling";

x-pack/plugin/gpu/src/main/java/org/elasticsearch/xpack/gpu/GPUPlugin.java

@@ -14,23 +14,34 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.FeatureFlag;
+import org.elasticsearch.core.Strings;
 import org.elasticsearch.gpu.GPUSupport;
 import org.elasticsearch.gpu.codec.ES92GpuHnswSQVectorsFormat;
 import org.elasticsearch.gpu.codec.ES92GpuHnswVectorsFormat;
 import org.elasticsearch.index.mapper.vectors.DenseVectorFieldMapper;
 import org.elasticsearch.index.mapper.vectors.VectorsFormatProvider;
 import org.elasticsearch.license.License;
+import org.elasticsearch.license.LicensedFeature;
+import org.elasticsearch.logging.LogManager;
+import org.elasticsearch.logging.Logger;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.internal.InternalVectorFormatProviderPlugin;
+import org.elasticsearch.xpack.core.XPackField;
 import org.elasticsearch.xpack.core.XPackPlugin;
 
 import java.util.List;
 
 public class GPUPlugin extends Plugin implements InternalVectorFormatProviderPlugin {
 
+    private static final Logger log = LogManager.getLogger(GPUPlugin.class);
+
     public static final FeatureFlag GPU_FORMAT = new FeatureFlag("gpu_vectors_indexing");
 
-    private static final License.OperationMode MINIMUM_ALLOWED_LICENSE = License.OperationMode.ENTERPRISE;
+    public static final LicensedFeature.Momentary GPU_INDEXING_FEATURE = LicensedFeature.momentary(
+        null,
+        XPackField.GPU_INDEXING,
+        License.OperationMode.ENTERPRISE
+    );
 
     private final GpuMode gpuMode;
 
@@ -75,7 +86,7 @@ public List<Setting<?>> getSettings() {
     // Allow tests to override the license state
     protected boolean isGpuIndexingFeatureAllowed() {
         var licenseState = XPackPlugin.getSharedLicenseState();
-        return licenseState != null && licenseState.isAllowedByLicense(MINIMUM_ALLOWED_LICENSE);
+        return licenseState != null && GPU_INDEXING_FEATURE.check(licenseState);
     }
 
     @Override
@@ -115,10 +126,25 @@ public ReferenceDocs referenceDocs() {
     @Override
     public VectorsFormatProvider getVectorsFormatProvider() {
         return (indexSettings, indexOptions, similarity, elementType) -> {
-            if (GPU_FORMAT.isEnabled() && isGpuIndexingFeatureAllowed()) {
-                if ((gpuMode == GpuMode.TRUE || (gpuMode == GpuMode.AUTO && GPUSupport.isSupported()))
-                    && vectorIndexAndElementTypeSupported(indexOptions.getType(), elementType)) {
-                    return getVectorsFormat(indexOptions, similarity);
+            if (GPU_FORMAT.isEnabled()) {
+                if (vectorIndexAndElementTypeSupported(indexOptions.getType(), elementType) == false) {
+                    return null;
+                }
+
+                if (gpuMode == GpuMode.TRUE || (gpuMode == GpuMode.AUTO && GPUSupport.isSupported())) {
+                    assert GPUSupport.isSupported();
+                    if (isGpuIndexingFeatureAllowed()) {
+                        return getVectorsFormat(indexOptions, similarity);
+                    } else {
+                        log.warn(
+                            Strings.format(
+                                "The current configuration supports GPU indexing, but it is not allowed by the current license. "
+                                    + "If this is intentional, it is possible to suppress this message by setting [%s] to FALSE",
+                                VECTORS_INDEXING_USE_GPU_NODE_SETTING.getKey()
+                            )
+                        );
+                        return null;
+                    }
                 }
             }
             return null;