Introduce Compatible Version plugin (#64481)

A RestCompatibilityPlugin and its xpack implementation allow to calculate a version requested on REST request. It uses accept, content-type headers to return a version.
It also performs a validation of allowed combinations of versions and values provided on accept/content-type headers

relates #51816

Author: pgomulka

libs/x-content/src/main/java/org/elasticsearch/common/xcontent/ParsedMediaType.java

@@ -33,13 +33,15 @@
  * @see MediaTypeRegistry
  */
 public class ParsedMediaType {
+    private final String originalHeaderValue;
     private final String type;
     private final String subType;
     private final Map<String, String> parameters;
     // tchar pattern as defined by RFC7230 section 3.2.6
     private static final Pattern TCHAR_PATTERN = Pattern.compile("[a-zA-z0-9!#$%&'*+\\-.\\^_`|~]+");
 
-    private ParsedMediaType(String type, String subType, Map<String, String> parameters) {
+    private ParsedMediaType(String originalHeaderValue, String type, String subType, Map<String, String> parameters) {
+        this.originalHeaderValue = originalHeaderValue;
         this.type = type;
         this.subType = subType;
         this.parameters = Collections.unmodifiableMap(parameters);
@@ -79,7 +81,7 @@ public static ParsedMediaType parseMediaType(String headerValue) {
                 throw new IllegalArgumentException("invalid media-type [" + headerValue + "]");
             }
             if (elements.length == 1) {
-                return new ParsedMediaType(splitMediaType[0].trim(), splitMediaType[1].trim(), Collections.emptyMap());
+                return new ParsedMediaType(headerValue, splitMediaType[0].trim(), splitMediaType[1].trim(), Collections.emptyMap());
             } else {
                 Map<String, String> parameters = new HashMap<>();
                 for (int i = 1; i < elements.length; i++) {
@@ -96,7 +98,7 @@ public static ParsedMediaType parseMediaType(String headerValue) {
                     String parameterValue = keyValueParam[1].toLowerCase(Locale.ROOT).trim();
                     parameters.put(parameterName, parameterValue);
                 }
-                return new ParsedMediaType(splitMediaType[0].trim().toLowerCase(Locale.ROOT),
+                return new ParsedMediaType(headerValue, splitMediaType[0].trim().toLowerCase(Locale.ROOT),
                     splitMediaType[1].trim().toLowerCase(Locale.ROOT), parameters);
             }
         }
@@ -144,4 +146,9 @@ private boolean isValidParameter(String paramName, String value, Map<String, Pat
         //TODO undefined parameters are allowed until https://github.com/elastic/elasticsearch/issues/63080
         return true;
     }
+
+    @Override
+    public String toString() {
+        return originalHeaderValue;
+    }
 }

server/src/main/java/org/elasticsearch/Version.java

@@ -397,7 +397,7 @@ public boolean isCompatible(Version version) {
      * Returns the minimum version that can be used for compatible REST API
      */
     public Version minimumRestCompatibilityVersion() {
-        return Version.CURRENT.previousMajor();
+        return this.previousMajor();
     }
 
     /**

server/src/main/java/org/elasticsearch/node/Node.java

@@ -144,6 +144,7 @@
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.plugins.RepositoryPlugin;
+import org.elasticsearch.plugins.RestCompatibilityPlugin;
 import org.elasticsearch.plugins.ScriptPlugin;
 import org.elasticsearch.plugins.SearchPlugin;
 import org.elasticsearch.plugins.SystemIndexPlugin;
@@ -723,8 +724,16 @@ protected Node(final Environment initialEnvironment,
      * package scope for testing
      */
     CompatibleVersion getRestCompatibleFunction() {
-        // TODO PG Until compatible version plugin is implemented, return current version.
-        return CompatibleVersion.CURRENT_VERSION;
+        List<RestCompatibilityPlugin> restCompatibilityPlugins = pluginsService.filterPlugins(RestCompatibilityPlugin.class);
+        final CompatibleVersion compatibleVersion;
+        if (restCompatibilityPlugins.size() > 1) {
+            throw new IllegalStateException("Only one RestCompatibilityPlugin is allowed");
+        } else if (restCompatibilityPlugins.size() == 1) {
+            compatibleVersion = restCompatibilityPlugins.get(0)::getCompatibleVersion;
+        } else {
+            compatibleVersion = CompatibleVersion.CURRENT_VERSION;
+        }
+        return compatibleVersion;
     }
 
     protected TransportService newTransportService(Settings settings, Transport transport, ThreadPool threadPool,

server/src/main/java/org/elasticsearch/plugins/RestCompatibilityPlugin.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.plugins;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.xcontent.ParsedMediaType;
+
+
+/**
+ * An extension point for Compatible API plugin implementation.
+ */
+public interface RestCompatibilityPlugin {
+    /**
+     * Returns a version which was requested on Accept and Content-Type headers
+     *
+     * @param acceptHeader      - a ParsedMediaType parsed from Accept header
+     * @param contentTypeHeader - a ParsedMediaType parsed from Content-Type header
+     * @param hasContent        - a flag indicating if a request has content
+     * @return a requested Compatible API Version
+     */
+    Version getCompatibleVersion(@Nullable ParsedMediaType acceptHeader, @Nullable ParsedMediaType contentTypeHeader, boolean hasContent);
+}

server/src/main/java/org/elasticsearch/rest/RestController.java

@@ -91,7 +91,7 @@ public class RestController implements HttpServerTransport.Dispatcher {
     /** Rest headers that are copied to internal requests made during a rest request. */
     private final Set<RestHeaderDefinition> headersToCopy;
     private final UsageService usageService;
-    private CompatibleVersion compatibleVersion;
+    private final CompatibleVersion compatibleVersion;
 
     public RestController(Set<RestHeaderDefinition> headersToCopy, UnaryOperator<RestHandler> handlerWrapper,
                           NodeClient client, CircuitBreakerService circuitBreakerService, UsageService usageService,

server/src/test/java/org/elasticsearch/node/NodeTests.java

@@ -20,13 +20,15 @@
 
 import org.apache.lucene.util.LuceneTestCase;
 import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.Version;
 import org.elasticsearch.bootstrap.BootstrapCheck;
 import org.elasticsearch.bootstrap.BootstrapContext;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.network.NetworkModule;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.BoundTransportAddress;
+import org.elasticsearch.common.xcontent.ParsedMediaType;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.index.IndexService;
 import org.elasticsearch.index.engine.Engine.Searcher;
@@ -36,6 +38,8 @@
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.plugins.CircuitBreakerPlugin;
 import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.plugins.RestCompatibilityPlugin;
+import org.elasticsearch.rest.CompatibleVersion;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.InternalTestCluster;
 import org.elasticsearch.test.MockHttpTransport;
@@ -339,4 +343,56 @@ public void setCircuitBreaker(CircuitBreaker circuitBreaker) {
             myCircuitBreaker.set(circuitBreaker);
         }
     }
+
+    public static class TestRestCompatibility1 extends Plugin implements RestCompatibilityPlugin {
+        @Override
+        public Version getCompatibleVersion(ParsedMediaType acceptHeader, ParsedMediaType contentTypeHeader, boolean hasContent) {
+            return Version.CURRENT.previousMajor();
+        }
+    }
+
+    public static class TestRestCompatibility2 extends Plugin implements RestCompatibilityPlugin {
+        @Override
+        public Version getCompatibleVersion(ParsedMediaType acceptHeader, ParsedMediaType contentTypeHeader, boolean hasContent) {
+            return null;
+        }
+    }
+
+    public void testLoadingMultipleRestCompatibilityPlugins() throws IOException {
+        Settings.Builder settings = baseSettings();
+
+        // throw an exception when two plugins are registered
+        List<Class<? extends Plugin>> plugins = basePlugins();
+        plugins.add(TestRestCompatibility1.class);
+        plugins.add(TestRestCompatibility2.class);
+
+        IllegalStateException e = expectThrows(IllegalStateException.class, () -> new MockNode(settings.build(), plugins));
+        assertThat(e.getMessage(), equalTo("Only one RestCompatibilityPlugin is allowed"));
+    }
+
+    public void testCorrectUsageOfRestCompatibilityPlugin() throws IOException {
+        Settings.Builder settings = baseSettings();
+
+        // the correct usage expects one plugin
+        List<Class<? extends Plugin>> plugins = basePlugins();
+        plugins.add(TestRestCompatibility1.class);
+
+        try (Node node = new MockNode(settings.build(), plugins)) {
+            CompatibleVersion restCompatibleFunction = node.getRestCompatibleFunction();
+            assertThat(restCompatibleFunction.get(null, null, false), equalTo(Version.CURRENT.previousMajor()));
+        }
+    }
+
+
+    public void testDefaultingRestCompatibilityPlugin() throws IOException {
+        Settings.Builder settings = baseSettings();
+
+        // default to CompatibleVersion.CURRENT_VERSION when no plugins provided
+        List<Class<? extends Plugin>> plugins = basePlugins();
+
+        try (Node node = new MockNode(settings.build(), plugins)) {
+            CompatibleVersion restCompatibleFunction = node.getRestCompatibleFunction();
+            assertThat(restCompatibleFunction.get(null, null, false), equalTo(Version.CURRENT));
+        }
+    }
 }

x-pack/plugin/rest-compatibility/build.gradle

@@ -0,0 +1,32 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+apply plugin: 'elasticsearch.esplugin'
+
+esplugin {
+  name 'rest-compatibility'
+  description 'A plugin for Compatible Rest API'
+  classname 'org.elasticsearch.compat.CompatibleVersionPlugin'
+  extendedPlugins = ['x-pack-core']
+}
+
+dependencies {
+  compileOnly project(path: xpackModule('core'), configuration: 'default')
+  testImplementation project(path: xpackModule('core'), configuration: 'testArtifacts')
+}

x-pack/plugin/rest-compatibility/src/main/java/org/elasticsearch/compat/CompatibleVersionPlugin.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.compat;
+
+import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.Version;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.xcontent.MediaType;
+import org.elasticsearch.common.xcontent.ParsedMediaType;
+import org.elasticsearch.plugins.Plugin;
+import org.elasticsearch.plugins.RestCompatibilityPlugin;
+import org.elasticsearch.rest.RestStatus;
+
+public class CompatibleVersionPlugin extends Plugin implements RestCompatibilityPlugin {
+
+    @Override
+    public Version getCompatibleVersion(
+        @Nullable ParsedMediaType acceptHeader,
+        @Nullable ParsedMediaType contentTypeHeader,
+        boolean hasContent
+    ) {
+        Byte aVersion = parseVersion(acceptHeader);
+        byte acceptVersion = aVersion == null ? Version.CURRENT.major : Integer.valueOf(aVersion).byteValue();
+        Byte cVersion = parseVersion(contentTypeHeader);
+        byte contentTypeVersion = cVersion == null ? Version.CURRENT.major : Integer.valueOf(cVersion).byteValue();
+
+        // accept version must be current or prior
+        if (acceptVersion > Version.CURRENT.major || acceptVersion < Version.CURRENT.minimumRestCompatibilityVersion().major) {
+            throw new ElasticsearchStatusException(
+                "Accept version must be either version {} or {}, but found {}. Accept={}",
+                RestStatus.BAD_REQUEST,
+                Version.CURRENT.major,
+                Version.CURRENT.minimumRestCompatibilityVersion().major,
+                acceptVersion,
+                acceptHeader
+            );
+        }
+        if (hasContent) {
+
+            // content-type version must be current or prior
+            if (contentTypeVersion > Version.CURRENT.major
+                || contentTypeVersion < Version.CURRENT.minimumRestCompatibilityVersion().major) {
+                throw new ElasticsearchStatusException(
+                    "Content-Type version must be either version {} or {}, but found {}. Content-Type={}",
+                    RestStatus.BAD_REQUEST,
+                    Version.CURRENT.major,
+                    Version.CURRENT.minimumRestCompatibilityVersion().major,
+                    contentTypeVersion,
+                    contentTypeHeader
+                );
+            }
+            // if both accept and content-type are sent, the version must match
+            if (contentTypeVersion != acceptVersion) {
+                throw new ElasticsearchStatusException(
+                    "A compatible version is required on both Content-Type and Accept headers "
+                        + "if either one has requested a compatible version "
+                        + "and the compatible versions must match. Accept={}, Content-Type={}",
+                    RestStatus.BAD_REQUEST,
+                    acceptHeader,
+                    contentTypeHeader
+                );
+            }
+            // both headers should be versioned or none
+            if ((cVersion == null && aVersion != null) || (aVersion == null && cVersion != null)) {
+                throw new ElasticsearchStatusException(
+                    "A compatible version is required on both Content-Type and Accept headers "
+                        + "if either one has requested a compatible version. Accept={}, Content-Type={}",
+                    RestStatus.BAD_REQUEST,
+                    acceptHeader,
+                    contentTypeHeader
+                );
+            }
+            if (contentTypeVersion < Version.CURRENT.major) {
+                return Version.CURRENT.previousMajor();
+            }
+        }
+
+        if (acceptVersion < Version.CURRENT.major) {
+            return Version.CURRENT.previousMajor();
+        }
+
+        return Version.CURRENT;
+    }
+
+    // scope for testing
+    static Byte parseVersion(ParsedMediaType parsedMediaType) {
+        if (parsedMediaType != null) {
+            String version = parsedMediaType.getParameters().get(MediaType.COMPATIBLE_WITH_PARAMETER_NAME);
+            return version != null ? Byte.parseByte(version) : null;
+        }
+        return null;
+    }
+}