Compatible API header parsing plugin

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

@@ -137,6 +137,10 @@ public static XContentBuilder builder(XContent xContent, Set<String> includes, S
         DATE_TRANSFORMERS = Collections.unmodifiableMap(dateTransformers);
     }
 
+    public XContentBuilder withCompatibleMajorVersion(byte compatibleVersion) {
+        return this;
+    }
+
     @FunctionalInterface
     public interface Writer {
         void write(XContentBuilder builder, Object value) throws IOException;

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

@@ -257,6 +257,7 @@ private static Version fromStringSlow(String version) {
     public final byte build;
     public final org.apache.lucene.util.Version luceneVersion;
     private final String toString;
+    public final int previousMajorId;
 
     Version(int id, org.apache.lucene.util.Version luceneVersion) {
         this.id = id;
@@ -266,6 +267,7 @@ private static Version fromStringSlow(String version) {
         this.build = (byte) (id % 100);
         this.luceneVersion = Objects.requireNonNull(luceneVersion);
         this.toString = major + "." + minor + "." + revision;
+        this.previousMajorId = major > 0 ? (major - 1) * 1000000 + 99 : major;
     }
 
     public boolean after(Version version) {
@@ -294,6 +296,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         return builder.value(toString());
     }
 
+    public Version previousMajor() {
+        return Version.fromId(previousMajorId);
+    }
+
     /*
      * We need the declared versions when computing the minimum compatibility version. As computing the declared versions uses reflection it
      * is not cheap. Since computing the minimum compatibility version can occur often, we use this holder to compute the declared versions

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -259,6 +259,7 @@
 import org.elasticsearch.persistent.UpdatePersistentTaskStatusAction;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.ActionPlugin.ActionHandler;
+import org.elasticsearch.rest.CompatibleVersion;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
 import org.elasticsearch.rest.RestHeaderDefinition;
@@ -427,7 +428,8 @@ public class ActionModule extends AbstractModule {
     public ActionModule(Settings settings, IndexNameExpressionResolver indexNameExpressionResolver,
                         IndexScopedSettings indexScopedSettings, ClusterSettings clusterSettings, SettingsFilter settingsFilter,
                         ThreadPool threadPool, List<ActionPlugin> actionPlugins, NodeClient nodeClient,
-                        CircuitBreakerService circuitBreakerService, UsageService usageService, SystemIndices systemIndices) {
+                        CircuitBreakerService circuitBreakerService, UsageService usageService, SystemIndices systemIndices,
+                        CompatibleVersion compatibleVersion) {
         this.settings = settings;
         this.indexNameExpressionResolver = indexNameExpressionResolver;
         this.indexScopedSettings = indexScopedSettings;
@@ -459,10 +461,9 @@ public ActionModule(Settings settings, IndexNameExpressionResolver indexNameExpr
         indicesAliasesRequestRequestValidators = new RequestValidators<>(
                 actionPlugins.stream().flatMap(p -> p.indicesAliasesRequestValidators().stream()).collect(Collectors.toList()));
 
-        restController = new RestController(headers, restWrapper, nodeClient, circuitBreakerService, usageService);
+        restController = new RestController(headers, restWrapper, nodeClient, circuitBreakerService, usageService, compatibleVersion);
     }
 
-
     public Map<String, ActionHandler<?, ?>> getActions() {
         return actions;
     }

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

@@ -143,6 +143,8 @@
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.plugins.RepositoryPlugin;
+import org.elasticsearch.plugins.RestCompatibilityPlugin;
+import org.elasticsearch.rest.CompatibleVersion;
 import org.elasticsearch.plugins.ScriptPlugin;
 import org.elasticsearch.plugins.SearchPlugin;
 import org.elasticsearch.plugins.SystemIndexPlugin;
@@ -529,7 +531,8 @@ protected Node(final Environment initialEnvironment,
 
             ActionModule actionModule = new ActionModule(settings, clusterModule.getIndexNameExpressionResolver(),
                 settingsModule.getIndexScopedSettings(), settingsModule.getClusterSettings(), settingsModule.getSettingsFilter(),
-                threadPool, pluginsService.filterPlugins(ActionPlugin.class), client, circuitBreakerService, usageService, systemIndices);
+                threadPool, pluginsService.filterPlugins(ActionPlugin.class), client, circuitBreakerService, usageService,
+                systemIndices, getRestCompatibleFunction());
             modules.add(actionModule);
 
             final RestController restController = actionModule.getRestController();
@@ -704,6 +707,23 @@ protected Node(final Environment initialEnvironment,
         }
     }
 
+    /**
+     * @return A function that can be used to determine the requested REST compatible version
+     * package scope for testing
+     */
+    CompatibleVersion getRestCompatibleFunction() {
+        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,
                                                    TransportInterceptor interceptor,
                                                    Function<BoundTransportAddress, DiscoveryNode> localNodeFactory,

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

@@ -0,0 +1,39 @@
+/*
+ * 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;
+
+
+/**
+ * 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 media-type value from Accept header
+     * @param contentTypeHeader - a media-type value from Content-Type header
+     * @param hasContent        - a flag indicating if a request has content
+     * @return a requested Compatible API Version
+     */
+    Version getCompatibleVersion(@Nullable String acceptHeader, @Nullable String contentTypeHeader, boolean hasContent);
+}

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

@@ -0,0 +1,34 @@
+/*
+ * 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.rest;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.common.Nullable;
+
+/**
+ * An interface used to specify a function that returns a compatible API version
+ * Intended to be used in a code base instead of a plugin.
+ */
+@FunctionalInterface
+public interface CompatibleVersion {
+    Version get(@Nullable String acceptHeader, @Nullable String contentTypeHeader, boolean hasContent);
+
+    CompatibleVersion CURRENT_VERSION = (acceptHeader, contentTypeHeader, hasContent) -> Version.CURRENT;
+}

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

@@ -23,6 +23,7 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.Version;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Strings;
@@ -90,11 +91,15 @@ 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;
+
 
     public RestController(Set<RestHeaderDefinition> headersToCopy, UnaryOperator<RestHandler> handlerWrapper,
-            NodeClient client, CircuitBreakerService circuitBreakerService, UsageService usageService) {
+                          NodeClient client, CircuitBreakerService circuitBreakerService, UsageService usageService,
+                          CompatibleVersion compatibleVersion) {
         this.headersToCopy = headersToCopy;
         this.usageService = usageService;
+        this.compatibleVersion = compatibleVersion;
         if (handlerWrapper == null) {
             handlerWrapper = h -> h; // passthrough if no wrapper set
         }
@@ -220,7 +225,8 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
         }
     }
 
-    private void dispatchRequest(RestRequest request, RestChannel channel, RestHandler handler) throws Exception {
+    private void dispatchRequest(RestRequest request, RestChannel channel, RestHandler handler, Version compatibleApiVersion)
+        throws Exception {
         final int contentLength = request.contentLength();
         if (contentLength > 0) {
             final XContentType xContentType = request.getXContentType();
@@ -242,7 +248,7 @@ private void dispatchRequest(RestRequest request, RestChannel channel, RestHandl
                 inFlightRequestsBreaker(circuitBreakerService).addWithoutBreaking(contentLength);
             }
             // iff we could reserve bytes for the request we need to send the response also over this channel
-            responseChannel = new ResourceHandlingHttpChannel(channel, circuitBreakerService, contentLength);
+            responseChannel = new ResourceHandlingHttpChannel(channel, circuitBreakerService, contentLength, compatibleApiVersion);
             // TODO: Count requests double in the circuit breaker if they need copying?
             if (handler.allowsUnsafeBuffers() == false) {
                 request.ensureSafeBuffers();
@@ -318,6 +324,9 @@ private void tryAllHandlers(final RestRequest request, final RestChannel channel
         final String rawPath = request.rawPath();
         final String uri = request.uri();
         final RestRequest.Method requestMethod;
+        //TODO: USAGE_1 now that we have a version we can implement a REST handler that accepts path, method AND version
+        Version version = compatibleVersion.get(request.header("Accept"), request.header("Content-Type"), request.hasContent());
+
         try {
             // Resolves the HTTP method and fails if the method is invalid
             requestMethod = request.method();
@@ -336,7 +345,7 @@ private void tryAllHandlers(final RestRequest request, final RestChannel channel
                       return;
                   }
                 } else {
-                    dispatchRequest(request, channel, handler);
+                    dispatchRequest(request, channel, handler, version);
                     return;
                 }
             }
@@ -454,12 +463,15 @@ private static final class ResourceHandlingHttpChannel implements RestChannel {
         private final RestChannel delegate;
         private final CircuitBreakerService circuitBreakerService;
         private final int contentLength;
+        private final Version compatibleVersion;
         private final AtomicBoolean closed = new AtomicBoolean();
 
-        ResourceHandlingHttpChannel(RestChannel delegate, CircuitBreakerService circuitBreakerService, int contentLength) {
+        ResourceHandlingHttpChannel(RestChannel delegate, CircuitBreakerService circuitBreakerService, int contentLength,
+                                    Version compatibleVersion) {
             this.delegate = delegate;
             this.circuitBreakerService = circuitBreakerService;
             this.contentLength = contentLength;
+            this.compatibleVersion = compatibleVersion;
         }
 
         @Override
@@ -474,13 +486,15 @@ public XContentBuilder newErrorBuilder() throws IOException {
 
         @Override
         public XContentBuilder newBuilder(@Nullable XContentType xContentType, boolean useFiltering) throws IOException {
-            return delegate.newBuilder(xContentType, useFiltering);
+            return delegate.newBuilder(xContentType, useFiltering)
+                .withCompatibleMajorVersion(compatibleVersion.major);
         }
 
         @Override
         public XContentBuilder newBuilder(XContentType xContentType, XContentType responseContentType, boolean useFiltering)
                 throws IOException {
-            return delegate.newBuilder(xContentType, responseContentType, useFiltering);
+            return delegate.newBuilder(xContentType, responseContentType, useFiltering)
+                        .withCompatibleMajorVersion(compatibleVersion.major);
         }
 
         @Override

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

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.rest;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.xcontent.XContent;
 import org.elasticsearch.rest.RestRequest.Method;
@@ -99,6 +100,16 @@ default boolean allowSystemIndexAccessByDefault() {
         return false;
     }
 
+    /**
+     * Returns the version that a handler is compatible with.
+     * The version is used to find a handler for a request that specified a compatible with version.
+     * If no version is specified, handler is assumed to be compatible with <code>Version.CURRENT</code>
+     * @return a version
+     */
+    default Version compatibleWithVersion() {
+        return Version.CURRENT;
+    }
+
     class Route {
 
         private final String path;

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

@@ -79,11 +79,18 @@ public boolean isContentConsumed() {
         return contentConsumed;
     }
 
+    // for testing
     protected RestRequest(NamedXContentRegistry xContentRegistry, Map<String, String> params, String path,
                           Map<String, List<String>> headers, HttpRequest httpRequest, HttpChannel httpChannel) {
         this(xContentRegistry, params, path, headers, httpRequest, httpChannel, requestIdGenerator.incrementAndGet());
     }
 
+    protected RestRequest(RestRequest restRequest) {
+        this(restRequest.getXContentRegistry(), restRequest.params(), restRequest.path(), restRequest.getHeaders(),
+            restRequest.getHttpRequest(), restRequest.getHttpChannel(), restRequest.getRequestId());
+    }
+
+
     private RestRequest(NamedXContentRegistry xContentRegistry, Map<String, String> params, String path,
                         Map<String, List<String>> headers, HttpRequest httpRequest, HttpChannel httpChannel, long requestId) {
         final XContentType xContentType;
@@ -104,10 +111,7 @@ private RestRequest(NamedXContentRegistry xContentRegistry, Map<String, String>
         this.requestId = requestId;
     }
 
-    protected RestRequest(RestRequest restRequest) {
-        this(restRequest.getXContentRegistry(), restRequest.params(), restRequest.path(), restRequest.getHeaders(),
-            restRequest.getHttpRequest(), restRequest.getHttpChannel(), restRequest.getRequestId());
-    }
+
 
     /**
      * Invoke {@link HttpRequest#releaseAndCopy()} on the http request in this instance and replace a pooled http request

server/src/test/java/org/elasticsearch/VersionTests.java

@@ -369,4 +369,11 @@ public void testUnreleasedVersion() {
         VersionTests.assertUnknownVersion(VERSION_5_1_0_UNRELEASED);
     }
 
+    public void testPreviousVersion() {
+        Version current = Version.CURRENT;
+        assertThat(current.previousMajor(), equalTo(Version.fromString(Version.CURRENT.major - 1 + ".0.0")));
+        assertThat(Version.fromString("7.8.1").previousMajor(), equalTo(Version.fromString("6.0.0")));
+        assertThat(Version.V_EMPTY.previousMajor(), equalTo(Version.V_EMPTY));
+    }
+
 }

server/src/test/java/org/elasticsearch/action/ActionModuleTests.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.plugins.ActionPlugin;
 import org.elasticsearch.plugins.ActionPlugin.ActionHandler;
+import org.elasticsearch.rest.CompatibleVersion;
 import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
@@ -111,7 +112,7 @@ public void testSetupRestHandlerContainsKnownBuiltin() {
         ActionModule actionModule = new ActionModule(settings.getSettings(),
             new IndexNameExpressionResolver(new ThreadContext(Settings.EMPTY)), settings.getIndexScopedSettings(),
             settings.getClusterSettings(), settings.getSettingsFilter(), null, emptyList(), null,
-            null, usageService, null);
+            null, usageService, null, CompatibleVersion.CURRENT_VERSION);
         actionModule.initRestHandlers(null);
         // At this point the easiest way to confirm that a handler is loaded is to try to register another one on top of it and to fail
         Exception e = expectThrows(IllegalArgumentException.class, () ->
@@ -151,7 +152,7 @@ public String getName() {
             ActionModule actionModule = new ActionModule(settings.getSettings(),
                 new IndexNameExpressionResolver(threadPool.getThreadContext()), settings.getIndexScopedSettings(),
                 settings.getClusterSettings(), settings.getSettingsFilter(), threadPool, singletonList(dupsMainAction),
-                null, null, usageService, null);
+                null, null, usageService, null, CompatibleVersion.CURRENT_VERSION);
             Exception e = expectThrows(IllegalArgumentException.class, () -> actionModule.initRestHandlers(null));
             assertThat(e.getMessage(), startsWith("Cannot replace existing handler for [/] for method: GET"));
         } finally {
@@ -186,7 +187,7 @@ public List<RestHandler> getRestHandlers(Settings settings, RestController restC
             ActionModule actionModule = new ActionModule(settings.getSettings(),
                 new IndexNameExpressionResolver(threadPool.getThreadContext()), settings.getIndexScopedSettings(),
                 settings.getClusterSettings(), settings.getSettingsFilter(), threadPool, singletonList(registersFakeHandler),
-                null, null, usageService, null);
+                null, null, usageService, null, CompatibleVersion.CURRENT_VERSION);
             actionModule.initRestHandlers(null);
             // At this point the easiest way to confirm that a handler is loaded is to try to register another one on top of it and to fail
             Exception e = expectThrows(IllegalArgumentException.class, () ->