Better logging of message keys ignored during indexing (#17601)

changelog/unreleased/pr-17601.toml

@@ -0,0 +1,7 @@
+type = "c"
+message = "Changed logging in Message class. When invalid message key is found and ignored, that fact is logged with INFO level. Rate limited log is used in order to not overwhelm logs with this kind of log messages."
+
+issues = [""]
+pulls = ["17601"]
+
+

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/processors/PipelineInterpreter.java

@@ -48,6 +48,7 @@
 import org.graylog2.plugin.Messages;
 import org.graylog2.plugin.messageprocessors.MessageProcessor;
 import org.graylog2.plugin.streams.Stream;
+import org.graylog2.plugin.utilities.ratelimitedlog.RateLimitedLogFactory;
 import org.graylog2.shared.buffers.processors.ProcessBufferProcessor;
 import org.graylog2.shared.messageq.MessageQueueAcknowledger;
 import org.graylog2.shared.metrics.MetricUtils;
@@ -470,12 +471,8 @@ public String className() {
         }
     }
 
-    public static RateLimitedLog getRateLimitedLog(Class clazz) {
-        final Logger baseLog = LoggerFactory.getLogger(clazz);
-        return RateLimitedLog
-                .withRateLimit(baseLog)
-                .maxRate(5).every(Duration.ofSeconds(10))
-                .build();
+    public static RateLimitedLog getRateLimitedLog(final Class<?> clazz) {
+        return RateLimitedLogFactory.createRateLimitedLog(clazz, 5, Duration.ofSeconds(10));
     }
 
     public static class State {

graylog2-server/src/main/java/org/graylog2/outputs/OutputRegistry.java

@@ -27,7 +27,6 @@
 import com.google.common.eventbus.EventBus;
 import com.google.common.eventbus.Subscribe;
 import com.google.common.util.concurrent.UncheckedExecutionException;
-import com.swrve.ratelimitedlogger.RateLimitedLog;
 import org.graylog2.database.NotFoundException;
 import org.graylog2.notifications.Notification;
 import org.graylog2.notifications.NotificationService;
@@ -37,6 +36,7 @@
 import org.graylog2.plugin.streams.Output;
 import org.graylog2.plugin.streams.Stream;
 import org.graylog2.plugin.system.NodeId;
+import org.graylog2.plugin.utilities.ratelimitedlog.RateLimitedLogFactory;
 import org.graylog2.streams.OutputService;
 import org.graylog2.streams.StreamService;
 import org.graylog2.streams.events.StreamsChangedEvent;
@@ -59,10 +59,7 @@
 @Singleton
 public class OutputRegistry {
     private static final Logger LOG = LoggerFactory.getLogger(OutputRegistry.class);
-    private static final Logger RATE_LIMITED_LOG = RateLimitedLog.withRateLimit(LOG)
-            .maxRate(1)
-            .every(Duration.ofSeconds(5))
-            .build();
+    private static final Logger RATE_LIMITED_LOG = RateLimitedLogFactory.createRateLimitedLog(LOG, 1, Duration.ofSeconds(5));
 
     private final Cache<String, MessageOutput> runningMessageOutputs;
     private final MessageOutput defaultMessageOutput;

graylog2-server/src/main/java/org/graylog2/plugin/Message.java

@@ -38,6 +38,7 @@
 import org.graylog2.indexer.messages.Indexable;
 import org.graylog2.plugin.streams.Stream;
 import org.graylog2.plugin.utilities.date.DateTimeConverter;
+import org.graylog2.plugin.utilities.ratelimitedlog.RateLimitedLogFactory;
 import org.graylog2.shared.utilities.ExceptionUtils;
 import org.joda.time.DateTime;
 import org.slf4j.Logger;
@@ -47,6 +48,7 @@
 import javax.annotation.Nullable;
 import javax.annotation.concurrent.NotThreadSafe;
 import java.net.InetAddress;
+import java.time.Duration;
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -83,6 +85,7 @@
 @NotThreadSafe
 public class Message implements Messages, Indexable {
     private static final Logger LOG = LoggerFactory.getLogger(Message.class);
+    private static final Logger RATE_LIMITED_LOG = RateLimitedLogFactory.createRateLimitedLog(LOG, 3, Duration.ofMinutes(1));
 
     /**
      * The "_id" is used as document ID to address the document in Elasticsearch.
@@ -552,6 +555,8 @@ private void addField(final String key, final Object value, final boolean isRequ
         if ((RESERVED_FIELDS.contains(trimmedKey) && !RESERVED_SETTABLE_FIELDS.contains(trimmedKey)) || !validKey(trimmedKey)) {
             if (LOG.isDebugEnabled()) {
                 LOG.debug("Ignoring invalid or reserved key {} for message {}", trimmedKey, getId());
+            } else {
+                RATE_LIMITED_LOG.info("Ignoring invalid or reserved key {} for message {}", trimmedKey, getId());
             }
             return;
         }

graylog2-server/src/main/java/org/graylog2/plugin/utilities/ratelimitedlog/RateLimitedLogFactory.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.plugin.utilities.ratelimitedlog;
+
+import com.swrve.ratelimitedlogger.RateLimitedLog;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+
+public class RateLimitedLogFactory {
+
+    public static RateLimitedLog createRateLimitedLog(final Logger logger,
+                                                      final int maxRate,
+                                                      final Duration duration) {
+        return RateLimitedLog
+                .withRateLimit(logger)
+                .maxRate(maxRate)
+                .every(duration)
+                .build();
+    }
+
+    public static RateLimitedLog createRateLimitedLog(final Class<?> clazz,
+                                                      final int maxRate,
+                                                      final Duration duration) {
+        return createRateLimitedLog(LoggerFactory.getLogger(clazz), maxRate, duration);
+    }
+
+}