Add DeprecationRestHandler to automatically log deprecated REST calls

This adds a new proxy for RestHandlers and RestControllers so that requests made
to deprecated REST APIs can be automatically logged in the ES logs via the
DeprecationLogger as well as via a "Warning" header (RFC-7234) for all responses.

core/src/main/java/org/elasticsearch/ElasticsearchException.java

@@ -33,7 +33,6 @@
 import org.elasticsearch.transport.TcpTransport;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -102,16 +101,7 @@ public ElasticsearchException(String msg, Throwable cause, Object... args) {
     public ElasticsearchException(StreamInput in) throws IOException {
         super(in.readOptionalString(), in.readException());
         readStackTrace(this, in);
-        int numKeys = in.readVInt();
-        for (int i = 0; i < numKeys; i++) {
-            final String key = in.readString();
-            final int numValues = in.readVInt();
-            final ArrayList<String> values = new ArrayList<>(numValues);
-            for (int j = 0; j < numValues; j++) {
-                values.add(in.readString());
-            }
-            headers.put(key, values);
-        }
+        headers.putAll(in.readMapOfLists());
     }
 
     /**
@@ -206,14 +196,7 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeOptionalString(this.getMessage());
         out.writeException(this.getCause());
         writeStackTraces(this, out);
-        out.writeVInt(headers.size());
-        for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
-            out.writeString(entry.getKey());
-            out.writeVInt(entry.getValue().size());
-            for (String v : entry.getValue()) {
-                out.writeString(v);
-            }
-        }
+        out.writeMapOfLists(headers);
     }
 
     public static ElasticsearchException readException(StreamInput input, int id) throws IOException {

core/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -53,6 +53,7 @@
 import java.nio.file.NoSuchFileException;
 import java.nio.file.NotDirectoryException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
@@ -414,6 +415,21 @@ public Map<String, Object> readMap() throws IOException {
         return (Map<String, Object>) readGenericValue();
     }
 
+    /**
+     * Read a map of strings to string lists.
+     */
+    public Map<String, List<String>> readMapOfLists() throws IOException {
+        int size = readVInt();
+        if (size == 0) {
+            return Collections.emptyMap();
+        }
+        Map<String, List<String>> map = new HashMap<>(size);
+        for (int i = 0; i < size; ++i) {
+            map.put(readString(), readList(StreamInput::readString));
+        }
+        return map;
+    }
+
     @SuppressWarnings({"unchecked"})
     @Nullable
     public Object readGenericValue() throws IOException {

core/src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java

@@ -403,6 +403,21 @@ public void writeMap(@Nullable Map<String, Object> map) throws IOException {
         writeGenericValue(map);
     }
 
+    /**
+     * Writes a map of strings to string lists.
+     */
+    public void writeMapOfLists(Map<String, List<String>> map) throws IOException {
+        writeVInt(map.size());
+
+        for (Map.Entry<String, List<String>> entry : map.entrySet()) {
+            writeString(entry.getKey());
+            writeVInt(entry.getValue().size());
+            for (String v : entry.getValue()) {
+                writeString(v);
+            }
+        }
+    }
+
     @FunctionalInterface
     interface Writer {
         void write(StreamOutput o, Object value) throws IOException;

core/src/main/java/org/elasticsearch/common/logging/DeprecationLogger.java

@@ -20,12 +20,70 @@
 package org.elasticsearch.common.logging;
 
 import org.elasticsearch.common.SuppressLoggerChecks;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
 
 /**
  * A logger that logs deprecation notices.
  */
 public class DeprecationLogger {
 
+    /**
+     * The "Warning" Header comes from RFC-7234. As the RFC describes, it's generally used for caching purposes, but it can be
+     * used for <em>any</em> warning.
+     *
+     * https://tools.ietf.org/html/rfc7234#section-5.5
+     */
+    public static final String DEPRECATION_HEADER = "Warning";
+
+    /**
+     * This is set once by the {@code Node} constructor, but it uses {@link CopyOnWriteArraySet} to ensure that tests can run in parallel.
+     * <p>
+     * Integration tests will create separate nodes within the same classloader, thus leading to a shared, {@code static} state.
+     * In order for all tests to appropriately be handled, this must be able to remember <em>all</em> {@link ThreadContext}s that it is
+     * given in a thread safe manner.
+     * <p>
+     * For actual usage, multiple nodes do not share the same JVM and therefore this will only be set once in practice.
+     */
+    private static final CopyOnWriteArraySet<ThreadContext> THREAD_CONTEXT = new CopyOnWriteArraySet<>();
+
+    /**
+     * Set the {@link ThreadContext} used to add deprecation headers to network responses.
+     * <p>
+     * This is expected to <em>only</em> be invoked by the {@code Node}'s constructor (therefore once outside of tests).
+     *
+     * @param threadContext The thread context owned by the {@code ThreadPool} (and implicitly a {@code Node})
+     * @throws IllegalStateException if this {@code threadContext} has already been set
+     */
+    public static void setThreadContext(ThreadContext threadContext) {
+        assert threadContext != null;
+
+        // add returning false means it _did_ have it already
+        if (THREAD_CONTEXT.add(threadContext) == false) {
+            throw new IllegalStateException("Double-setting ThreadContext not allowed!");
+        }
+    }
+
+    /**
+     * Remove the {@link ThreadContext} used to add deprecation headers to network responses.
+     * <p>
+     * This is expected to <em>only</em> be invoked by the {@code Node}'s {@code close} method (therefore once outside of tests).
+     *
+     * @param threadContext The thread context owned by the {@code ThreadPool} (and implicitly a {@code Node})
+     * @throws IllegalStateException if this {@code threadContext} is unknown (and presumably already unset before)
+     */
+    public static void removeThreadContext(ThreadContext threadContext) {
+        assert threadContext != null;
+
+        // remove returning false means it did not have it already
+        if (THREAD_CONTEXT.remove(threadContext) == false) {
+            throw new IllegalStateException("Removing unknown ThreadContext not allowed!");
+        }
+    }
+
     private final ESLogger logger;
 
     /**
@@ -47,8 +105,37 @@ public DeprecationLogger(ESLogger parentLogger) {
     /**
      * Logs a deprecated message.
      */
-    @SuppressLoggerChecks(reason = "safely delegates to logger")
     public void deprecated(String msg, Object... params) {
-        logger.debug(msg, params);
+        deprecated(THREAD_CONTEXT, msg, params);
+    }
+
+    /**
+     * Logs a deprecated message to the deprecation log, as well as to the local {@link ThreadContext}.
+     *
+     * @param threadContexts The node's {@link ThreadContext} (outside of concurrent tests, this should only ever have one context).
+     * @param msg The deprecation message.
+     * @param params The parameters used to fill in the message, if any exist.
+     */
+    @SuppressLoggerChecks(reason = "safely delegates to logger")
+    void deprecated(Set<ThreadContext> threadContexts, String msg, Object... params) {
+        Iterator<ThreadContext> iterator = threadContexts.iterator();
+
+        if (iterator.hasNext()) {
+            final String formattedMsg = LoggerMessageFormat.format(msg, params);
+
+            while (iterator.hasNext()) {
+                try {
+                    iterator.next().addResponseHeader(DEPRECATION_HEADER, formattedMsg);
+                } catch (IllegalStateException e) {
+                    // ignored; it should be removed shortly
+                }
+            }
+
+            logger.debug(formattedMsg);
+        } else {
+            logger.debug(msg, params);
+        }
+
     }
+
 }