Write deprecation logs to a data stream

server/src/main/java/org/elasticsearch/common/logging/DeprecationIndexingService.java

@@ -0,0 +1,154 @@
+package org.elasticsearch.common.logging;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.DocWriteRequest;
+import org.elasticsearch.action.admin.indices.template.put.PutComposableIndexTemplateAction;
+import org.elasticsearch.action.index.IndexAction;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.client.OriginSettingClient;
+import org.elasticsearch.cluster.ClusterChangedEvent;
+import org.elasticsearch.cluster.ClusterStateListener;
+import org.elasticsearch.cluster.metadata.ComposableIndexTemplate;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.xcontent.NamedXContentRegistry;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.elasticsearch.common.Strings.isNullOrEmpty;
+
+/**
+ * This service is responsible for writing deprecation messages to a data stream. It also creates
+ * the data stream if necessary. The writing of messages can be toggled using the
+ * {@link #WRITE_DEPRECATION_LOGS_TO_INDEX} setting.
+ */
+public class DeprecationIndexingService implements ClusterStateListener {
+    private static final Logger LOGGER = LogManager.getLogger(DeprecationIndexingService.class);
+
+    private static final String DATA_STREAM_NAME = "logs-deprecation-elasticsearch";
+    private static final String TEMPLATE_NAME = DATA_STREAM_NAME + "-template";
+    private static final String TEMPLATE_MAPPING = TEMPLATE_NAME + ".json";
+
+    private static final String DEPRECATION_ORIGIN = "deprecation";
+
+    public static final Setting<Boolean> WRITE_DEPRECATION_LOGS_TO_INDEX = Setting.boolSetting(
+        "cluster.deprecation_indexing.enabled",
+        true,
+        Setting.Property.NodeScope,
+        Setting.Property.Dynamic
+    );
+
+    private final Client client;
+    private boolean hasTriedToLoadTemplate = false;
+    private volatile boolean isEnabled = false;
+
+    public DeprecationIndexingService(ClusterService clusterService, Client client) {
+        this.client = new OriginSettingClient(client, DEPRECATION_ORIGIN);
+
+        clusterService.addListener(this);
+    }
+
+    /**
+     * Indexes a deprecation message.
+     * @param key       the key that was used to determine if this deprecation should have been be logged.
+     *                  Useful when aggregating the recorded messages.
+     * @param message   the message to log
+     * @param xOpaqueId the associated "X-Opaque-ID" header value if any, or <code>null</code>
+     * @param params    parameters to the message, if any
+     */
+    public void writeMessage(String key, String message, String xOpaqueId, Object[] params) {
+        if (this.isEnabled == false) {
+            return;
+        }
+
+        Map<String, Object> payload = new HashMap<>();
+        payload.put("@timestamp", Instant.now().toString());
+        payload.put("key", key);
+        payload.put("message", message);
+
+        if (isNullOrEmpty(xOpaqueId) == false) {
+            payload.put("x-opaque-id", xOpaqueId);
+        }
+
+        if (params != null && params.length > 0) {
+            payload.put("params", params);
+        }
+
+        new IndexRequestBuilder(client, IndexAction.INSTANCE).setIndex(DATA_STREAM_NAME)
+            .setOpType(DocWriteRequest.OpType.CREATE)
+            .setSource(payload)
+            .execute(new ActionListener<>() {
+                @Override
+                public void onResponse(IndexResponse indexResponse) {
+                    // Nothing to do
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    LOGGER.error("Failed to index deprecation message", e);
+                }
+            });
+    }
+
+    /**
+     * Listens for changes to the cluster state, in order to know whether to toggle indexing.
+     */
+    @Override
+    public void clusterChanged(ClusterChangedEvent event) {
+        this.isEnabled = WRITE_DEPRECATION_LOGS_TO_INDEX.get(event.state().getMetadata().settings());
+
+        if (this.isEnabled == false || this.hasTriedToLoadTemplate == true) {
+            return;
+        }
+
+        // We only ever try to load the template once, because if there's a problem, we'll spam
+        // the log with the failure on every cluster state update
+        this.hasTriedToLoadTemplate = true;
+
+        if (event.state().getMetadata().templatesV2().containsKey(TEMPLATE_NAME)) {
+            return;
+        }
+
+        loadTemplate();
+    }
+
+    /*
+     * Attempts to load a template for the deprecation logs data stream
+     */
+    private void loadTemplate() {
+        try (InputStream is = getClass().getResourceAsStream(TEMPLATE_MAPPING)) {
+            final XContentParser parser = JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY, null, is);
+
+            PutComposableIndexTemplateAction.Request request = new PutComposableIndexTemplateAction.Request(TEMPLATE_NAME);
+            request.cause("auto (deprecation indexing service)");
+            request.indexTemplate(ComposableIndexTemplate.parse(parser));
+
+            this.client.execute(PutComposableIndexTemplateAction.INSTANCE, request, new ActionListener<>() {
+                @Override
+                public void onResponse(AcknowledgedResponse acknowledgedResponse) {
+                    if (acknowledgedResponse.isAcknowledged() == false) {
+                        LOGGER.error("The attempt to create a deprecations index template was not acknowledged.");
+                    }
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    LOGGER.error("Failed to create the deprecations index template: " + e.getMessage(), e);
+                }
+            });
+        } catch (IOException e) {
+            LOGGER.error("Failed to load " + TEMPLATE_MAPPING + ": " + e.getMessage(), e);
+        }
+    }
+}

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

@@ -37,6 +37,7 @@
  * @see ThrottlingAndHeaderWarningLogger for throttling and header warnings implementation details
  */
 public class DeprecationLogger {
+    private static DeprecationIndexingService indexingService;
     private final ThrottlingAndHeaderWarningLogger deprecationLogger;
 
     /**
@@ -45,8 +46,8 @@ public class DeprecationLogger {
      * it replaces "org.elasticsearch" with "org.elasticsearch.deprecation" to maintain
      * the "org.elasticsearch" namespace.
      */
-    private DeprecationLogger(Logger parentLogger) {
-        deprecationLogger = new ThrottlingAndHeaderWarningLogger(parentLogger);
+    public DeprecationLogger(Logger parentLogger) {
+        deprecationLogger = new ThrottlingAndHeaderWarningLogger(deprecatedLoggerName(parentLogger.getName()), indexingService);
     }
 
     public static DeprecationLogger getLogger(Class<?> aClass) {
@@ -79,6 +80,14 @@ public static void removeThreadContext(ThreadContext threadContext) {
         HeaderWarning.removeThreadContext(threadContext);
     }
 
+    public static void setIndexingService(DeprecationIndexingService indexingService) {
+        DeprecationLogger.indexingService = indexingService;
+    }
+
+    public static void removeIndexingService() {
+        DeprecationLogger.indexingService = null;
+    }
+
     /**
      * Logs a deprecation message, adding a formatted warning message as a response header on the thread context.
      * The deprecation message will be throttled to deprecation log.

server/src/main/java/org/elasticsearch/common/logging/RateLimiter.java

@@ -0,0 +1,51 @@
+/*
+ * 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.common.logging;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * <p>
+ * This class limits the number of times that actions are carried out.
+ * <p>
+ * The throttling algorithm relies on a LRU set of keys, which evicts entries when its size exceeds 128.
+ * When a {@code key} is seen for the first time, the {@code runnable} will be executed, but then will not be
+ * executed again for that key until the key is removed from the set.
+ */
+class RateLimiter {
+
+    // LRU set of keys used to determine if a message should be emitted to the logs
+    private final Set<String> keys = Collections.newSetFromMap(Collections.synchronizedMap(new LinkedHashMap<>() {
+        @Override
+        protected boolean removeEldestEntry(final Map.Entry<String, Boolean> eldest) {
+            return size() > 128;
+        }
+    }));
+
+    void limit(String key, Runnable runnable) {
+        boolean shouldRun = keys.add(key);
+        if (shouldRun) {
+            runnable.run();
+        }
+    }
+}

server/src/main/java/org/elasticsearch/common/logging/ThrottlingAndHeaderWarningLogger.java

@@ -20,16 +20,32 @@
 package org.elasticsearch.common.logging;
 
 import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.message.Message;
+import org.elasticsearch.common.SuppressLoggerChecks;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+
+import java.security.AccessController;
+import java.security.PrivilegedAction;
 
 /**
- * This class wraps both <code>HeaderWarningLogger</code> and <code>ThrottlingLogger</code>
- * which is a common use case across Elasticsearch
+ * This class composes {@link HeaderWarning}, {@link DeprecationIndexingService} and {@link Logger},
+ * in order to apply a single message to multiple destination.
+ * <p>
+ * Logging and indexing are throttled in order to avoid filling the destination with duplicates.
+ * Throttling is implemented using a mandatory per-message key combined with any <code>X-Opaque-Id</code>
+ * HTTP header value. This header allows throttling per user. This value is set in {@link ThreadContext}.
+ * <p>
+ * TODO wrapping logging this way limits the usage of %location. It will think this is used from that class.
  */
 class ThrottlingAndHeaderWarningLogger {
-    private final ThrottlingLogger throttlingLogger;
+    private final Logger logger;
+    private final RateLimiter rateLimiter;
+    private final DeprecationIndexingService indexingService;
 
-    ThrottlingAndHeaderWarningLogger(Logger logger) {
-        this.throttlingLogger = new ThrottlingLogger(logger);
+    ThrottlingAndHeaderWarningLogger(Logger logger, DeprecationIndexingService indexingService) {
+        this.logger = logger;
+        this.rateLimiter = new RateLimiter();
+        this.indexingService = indexingService;
     }
 
     /**
@@ -43,7 +59,24 @@ void throttleLogAndAddWarning(final String key, ESLogMessage message) {
         String messagePattern = message.getMessagePattern();
         Object[] arguments = message.getArguments();
         HeaderWarning.addWarning(messagePattern, arguments);
-        throttlingLogger.throttleLog(key, message);
+
+        String xOpaqueId = HeaderWarning.getXOpaqueId();
+        this.rateLimiter.limit(xOpaqueId + key, () -> {
+            log(message);
+            if (indexingService != null) {
+                indexingService.writeMessage(key, messagePattern, xOpaqueId, arguments);
+            }
+        });
     }
 
+    private void log(Message message) {
+        AccessController.doPrivileged(new PrivilegedAction<Void>() {
+            @SuppressLoggerChecks(reason = "safely delegates to logger")
+            @Override
+            public Void run() {
+                logger.warn(message);
+                return null;
+            }
+        });
+    }
 }