From 21d80f95dedfa3c419881011bce7cdc45806da8f Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 28 May 2026 12:14:38 +0200 Subject: [PATCH 1/6] Implemented a log4j filter that uses Guava's Bloom filter to check if a log message at certain level was already emitted --- .../org/logstash/log/DeduplicationFilter.java | 93 ++++++++++++++++++ .../logstash/log/DeduplicationFilterTest.java | 95 +++++++++++++++++++ 2 files changed, 188 insertions(+) create mode 100644 logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java create mode 100644 logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java diff --git a/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java b/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java new file mode 100644 index 0000000000..0c4d2ff2e4 --- /dev/null +++ b/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java @@ -0,0 +1,93 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. 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.logstash.log; + +import com.google.common.hash.BloomFilter; +import com.google.common.hash.Funnels; +import org.apache.logging.log4j.core.Appender; +import org.apache.logging.log4j.core.Core; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.config.plugins.Plugin; +import org.apache.logging.log4j.core.config.plugins.PluginAttribute; +import org.apache.logging.log4j.core.config.plugins.PluginFactory; +import org.apache.logging.log4j.core.filter.AbstractFilter; + +import java.nio.charset.StandardCharsets; + +/** + * Log4j2 filter that suppresses repeated log lines using a Guava {@link BloomFilter}. + *

+ * Deduplication key is {@code level + formattedMessage}. The first occurrence of a key + * yields {@link Result#NEUTRAL}; subsequent occurrences yield {@link Result#DENY}. + *

+ *

+ * Example {@code log4j2.properties} wiring on an appender: + *

+ * 
+ * appender.rolling.filter.dedup.type = DeduplicationFilter
+ * appender.rolling.filter.dedup.falsePositiveProbability = 0.01
+ *
+ */ +@Plugin(name = "DeduplicationFilter", category = Core.CATEGORY_NAME, elementType = Appender.ELEMENT_TYPE, printObject = true) +public final class DeduplicationFilter extends AbstractFilter { + + static final double DEFAULT_FALSE_POSITIVE_PROBABILITY = 0.01; + private static final int DEFAULT_EXPECTED_INSERTIONS = 1_000_000; + + private final BloomFilter seenKeys; + + @PluginFactory + public static DeduplicationFilter createFilter( + @PluginAttribute(value = "falsePositiveProbability", defaultDouble = DEFAULT_FALSE_POSITIVE_PROBABILITY) + final double falsePositiveProbability) { + return new DeduplicationFilter(resolveFalsePositiveProbability(falsePositiveProbability)); + } + + private DeduplicationFilter(final double falsePositiveProbability) { + seenKeys = BloomFilter.create( + Funnels.stringFunnel(StandardCharsets.UTF_8), + DEFAULT_EXPECTED_INSERTIONS, + falsePositiveProbability + ); + } + + static double resolveFalsePositiveProbability(final double falsePositiveProbability) { + if (falsePositiveProbability > 0.0 && falsePositiveProbability < 1.0) { + return falsePositiveProbability; + } + return DEFAULT_FALSE_POSITIVE_PROBABILITY; + } + + @Override + public Result filter(final LogEvent event) { + final CharSequence key = dedupKey(event); + synchronized (seenKeys) { + if (seenKeys.mightContain(key)) { + return Result.DENY; + } + seenKeys.put(key); + return Result.NEUTRAL; + } + } + + private static CharSequence dedupKey(final LogEvent event) { + return event.getLevel().name() + '\0' + event.getMessage().getFormattedMessage(); + } +} diff --git a/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java b/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java new file mode 100644 index 0000000000..508857a8f9 --- /dev/null +++ b/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java @@ -0,0 +1,95 @@ +/* + * Licensed to Elasticsearch B.V. under one or more contributor + * license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright + * ownership. Elasticsearch B.V. 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.logstash.log; + +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.core.Filter; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.impl.Log4jLogEvent; +import org.apache.logging.log4j.message.SimpleMessage; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class DeduplicationFilterTest { + + @Test + public void givenAStringAtInfoLevelWhenAppearsFirstTimeThenIsForwarded() { + final DeduplicationFilter filter = DeduplicationFilter.createFilter( + DeduplicationFilter.DEFAULT_FALSE_POSITIVE_PROBABILITY); + + final Filter.Result result = filter.filter(logEvent(Level.INFO, "duplicate me")); + + assertEquals(Filter.Result.NEUTRAL, result); + } + + @Test + public void givenAStringAtInfoLevelWhenAppearsMultipleTimesThenIsDenied() { + final DeduplicationFilter filter = DeduplicationFilter.createFilter( + DeduplicationFilter.DEFAULT_FALSE_POSITIVE_PROBABILITY); + + filter.filter(logEvent(Level.WARN, "same line")); + final Filter.Result result = filter.filter(logEvent(Level.WARN, "same line")); + + assertEquals(Filter.Result.DENY, result); + } + + @Test + public void givenAStringWhenAppearsAtDifferentLevelThenIsForwarded() { + final DeduplicationFilter filter = DeduplicationFilter.createFilter( + DeduplicationFilter.DEFAULT_FALSE_POSITIVE_PROBABILITY); + + filter.filter(logEvent(Level.INFO, "shared text")); + final Filter.Result result = filter.filter(logEvent(Level.ERROR, "shared text")); + + assertEquals(Filter.Result.NEUTRAL, result); + } + + @Test + public void invalidFalsePositiveProbabilityFallsBackToDefault() { + assertEquals( + DeduplicationFilter.DEFAULT_FALSE_POSITIVE_PROBABILITY, + DeduplicationFilter.resolveFalsePositiveProbability(0.0), + 0.0 + ); + assertEquals( + DeduplicationFilter.DEFAULT_FALSE_POSITIVE_PROBABILITY, + DeduplicationFilter.resolveFalsePositiveProbability(1.0), + 0.0 + ); + assertEquals( + DeduplicationFilter.DEFAULT_FALSE_POSITIVE_PROBABILITY, + DeduplicationFilter.resolveFalsePositiveProbability(-0.5), + 0.0 + ); + } + + @Test + public void validFalsePositiveProbabilityIsPreserved() { + assertEquals(0.001, DeduplicationFilter.resolveFalsePositiveProbability(0.001), 0.0); + } + + private static LogEvent logEvent(final Level level, final String message) { + return Log4jLogEvent.newBuilder() + .setLevel(level) + .setMessage(new SimpleMessage(message)) + .build(); + } +} From 657241aaf2eeb9c55600c88d32f5ddd0ba22c47f Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 28 May 2026 17:04:42 +0200 Subject: [PATCH 2/6] Moved Bloom filter from CharSequence to String --- .../src/main/java/org/logstash/log/DeduplicationFilter.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java b/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java index 0c4d2ff2e4..f99a04e1b1 100644 --- a/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java +++ b/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java @@ -51,7 +51,7 @@ public final class DeduplicationFilter extends AbstractFilter { static final double DEFAULT_FALSE_POSITIVE_PROBABILITY = 0.01; private static final int DEFAULT_EXPECTED_INSERTIONS = 1_000_000; - private final BloomFilter seenKeys; + private final BloomFilter seenKeys; @PluginFactory public static DeduplicationFilter createFilter( @@ -77,7 +77,7 @@ static double resolveFalsePositiveProbability(final double falsePositiveProbabil @Override public Result filter(final LogEvent event) { - final CharSequence key = dedupKey(event); + final String key = dedupKey(event); synchronized (seenKeys) { if (seenKeys.mightContain(key)) { return Result.DENY; @@ -87,7 +87,7 @@ public Result filter(final LogEvent event) { } } - private static CharSequence dedupKey(final LogEvent event) { + private static String dedupKey(final LogEvent event) { return event.getLevel().name() + '\0' + event.getMessage().getFormattedMessage(); } } From 3da4a29d4cfd3f17af272ef225589ca7c2bccf3a Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 28 May 2026 17:11:31 +0200 Subject: [PATCH 3/6] Inverted the condition on the Bloom filter, eventually permit the log only if not previously seen so that it avoid the false positive, being on the safe side of filter, where it's certain if an element is not part of the filter. --- .../org/logstash/log/DeduplicationFilter.java | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java b/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java index f99a04e1b1..09ccf51be5 100644 --- a/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java +++ b/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java @@ -38,11 +38,12 @@ * yields {@link Result#NEUTRAL}; subsequent occurrences yield {@link Result#DENY}. *

*

- * Example {@code log4j2.properties} wiring on an appender: + * Example {@code log4j2.properties} wiring on a logger: *

* 
- * appender.rolling.filter.dedup.type = DeduplicationFilter
- * appender.rolling.filter.dedup.falsePositiveProbability = 0.01
+ * logger.periodic_flusher.name = org.logstash.execution.PeriodicFlush
+ * logger.periodic_flusher.level = DEBUG
+ * logger.periodic_flusher.filter.dedup.type = DeduplicationFilter
  *
*/ @Plugin(name = "DeduplicationFilter", category = Core.CATEGORY_NAME, elementType = Appender.ELEMENT_TYPE, printObject = true) @@ -79,11 +80,16 @@ static double resolveFalsePositiveProbability(final double falsePositiveProbabil public Result filter(final LogEvent event) { final String key = dedupKey(event); synchronized (seenKeys) { - if (seenKeys.mightContain(key)) { - return Result.DENY; + if (!seenKeys.mightContain(key)) { + seenKeys.put(key); + return Result.NEUTRAL; } - seenKeys.put(key); - return Result.NEUTRAL; + return Result.DENY; +// if (seenKeys.mightContain(key)) { +// return Result.DENY; +// } +// seenKeys.put(key); +// return Result.NEUTRAL; } } From 05a98323c8b23a3e2c5a00416e808e1767278c82 Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 28 May 2026 17:16:02 +0200 Subject: [PATCH 4/6] [Test] renamed test --- .../src/test/java/org/logstash/log/DeduplicationFilterTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java b/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java index 508857a8f9..2023f4be5b 100644 --- a/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java +++ b/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java @@ -41,7 +41,7 @@ public void givenAStringAtInfoLevelWhenAppearsFirstTimeThenIsForwarded() { } @Test - public void givenAStringAtInfoLevelWhenAppearsMultipleTimesThenIsDenied() { + public void givenAStringAtWarnLevelWhenAppearsMultipleTimesThenIsDenied() { final DeduplicationFilter filter = DeduplicationFilter.createFilter( DeduplicationFilter.DEFAULT_FALSE_POSITIVE_PROBABILITY); From 4687dcb1418dd1102fb651cad996810e870e30ba Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 28 May 2026 17:26:57 +0200 Subject: [PATCH 5/6] Minor, removed commented code --- .../src/main/java/org/logstash/log/DeduplicationFilter.java | 5 ----- 1 file changed, 5 deletions(-) diff --git a/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java b/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java index 09ccf51be5..f7f3b61ce8 100644 --- a/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java +++ b/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java @@ -85,11 +85,6 @@ public Result filter(final LogEvent event) { return Result.NEUTRAL; } return Result.DENY; -// if (seenKeys.mightContain(key)) { -// return Result.DENY; -// } -// seenKeys.put(key); -// return Result.NEUTRAL; } } From 16e9f872346cce26068eb17e1e48b95341e2b5bd Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 16 Jun 2026 14:06:58 +0200 Subject: [PATCH 6/6] Lowered maximum accepted falsePositiveProbability to 5% and log on status logger when defaulting becuase of invalid value --- .../org/logstash/log/DeduplicationFilter.java | 7 ++- .../logstash/log/DeduplicationFilterTest.java | 49 +++++++++++++++++++ 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java b/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java index f7f3b61ce8..040618a40d 100644 --- a/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java +++ b/logstash-core/src/main/java/org/logstash/log/DeduplicationFilter.java @@ -21,6 +21,7 @@ import com.google.common.hash.BloomFilter; import com.google.common.hash.Funnels; +import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.core.Appender; import org.apache.logging.log4j.core.Core; import org.apache.logging.log4j.core.LogEvent; @@ -28,6 +29,7 @@ import org.apache.logging.log4j.core.config.plugins.PluginAttribute; import org.apache.logging.log4j.core.config.plugins.PluginFactory; import org.apache.logging.log4j.core.filter.AbstractFilter; +import org.apache.logging.log4j.status.StatusLogger; import java.nio.charset.StandardCharsets; @@ -51,6 +53,7 @@ public final class DeduplicationFilter extends AbstractFilter { static final double DEFAULT_FALSE_POSITIVE_PROBABILITY = 0.01; private static final int DEFAULT_EXPECTED_INSERTIONS = 1_000_000; + private static final Logger STATUS_LOGGER = StatusLogger.getLogger(); private final BloomFilter seenKeys; @@ -70,9 +73,11 @@ private DeduplicationFilter(final double falsePositiveProbability) { } static double resolveFalsePositiveProbability(final double falsePositiveProbability) { - if (falsePositiveProbability > 0.0 && falsePositiveProbability < 1.0) { + if (falsePositiveProbability > 0.0 && falsePositiveProbability <= 0.5) { return falsePositiveProbability; } + STATUS_LOGGER.warn("falsePositiveProbability is expected to be in the range (0, 5%] but was {}, defaulting to {}", + falsePositiveProbability, DEFAULT_FALSE_POSITIVE_PROBABILITY); return DEFAULT_FALSE_POSITIVE_PROBABILITY; } diff --git a/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java b/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java index 2023f4be5b..ff20bfadb1 100644 --- a/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java +++ b/logstash-core/src/test/java/org/logstash/log/DeduplicationFilterTest.java @@ -24,8 +24,18 @@ import org.apache.logging.log4j.core.LogEvent; import org.apache.logging.log4j.core.impl.Log4jLogEvent; import org.apache.logging.log4j.message.SimpleMessage; +import org.apache.logging.log4j.status.StatusData; +import org.apache.logging.log4j.status.StatusListener; +import org.apache.logging.log4j.status.StatusLogger; import org.junit.Test; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +import static org.hamcrest.MatcherAssert.assertThat; + +import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; public class DeduplicationFilterTest { @@ -92,4 +102,43 @@ private static LogEvent logEvent(final Level level, final String message) { .setMessage(new SimpleMessage(message)) .build(); } + + private static class SpyListener implements StatusListener { + + private final List spiedMessages = new ArrayList<>(); + + @Override + public void log(StatusData data) { + spiedMessages.add(data); + } + + @Override + public Level getStatusLevel() { + return Level.WARN; + } + + @Override + public void close() throws IOException { + + } + } + + @Test + public void givenFalsePositiveProbabilitySetToValueOutsideExpectedRangeThenOverrideToDefaultAndLog() throws IOException { + // setup + try (SpyListener loggerSpy = new SpyListener()) { + StatusLogger.getLogger().registerListener(loggerSpy); + + // Exercise + DeduplicationFilter.resolveFalsePositiveProbability(1.0); + + // Verify + final StatusData data = loggerSpy.spiedMessages.get(0); + assertEquals(Level.WARN, data.getLevel()); + assertThat(data.getMessage().getFormattedMessage(), containsString("falsePositiveProbability is expected to be in the range (0, 5%] but was")); + + // teardown + StatusLogger.getLogger().removeListener(loggerSpy); + } + } }