From 0e586260783dfa590f332c5546a97976ca02fab9 Mon Sep 17 00:00:00 2001 From: Kathiresan Selvaraj <96088452+kathirsvn@users.noreply.github.com> Date: Mon, 15 Jun 2026 15:29:23 +0200 Subject: [PATCH 1/2] CASSANDRA-21474: Added eviction policy to cache in NoSpamLogger (Backport from CNDB-17505) ### What is the issue NoSpamLogger uses unbounded cache that could lead to memory exhaustion ### What does this PR fix and why was it fixed This PR replaces the previous `NonBlockingHashMap` based caching implementation in `NoSpamLogger` with Caffeine cache to prevent unbounded memory growth and improve cache management --- .../config/CassandraRelevantProperties.java | 8 + .../apache/cassandra/utils/NoSpamLogger.java | 90 +++++- .../cassandra/utils/NoSpamLoggerTest.java | 256 +++++++++++++++++- 3 files changed, 339 insertions(+), 15 deletions(-) diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 2be55fbe5c69..bebdfc0839a9 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -406,6 +406,14 @@ public enum CassandraRelevantProperties MEMTABLE_OVERHEAD_SIZE("cassandra.memtable.row_overhead_size", "-1"), MEMTABLE_SHARD_COUNT("cassandra.memtable.shard.count"), MEMTABLE_TRIE_SIZE_LIMIT("cassandra.trie_size_limit_mb"), + + /** + * Maximum number of log statements cached per NoSpamLogger instance. + * This prevents unbounded memory growth when log messages contain dynamic content. + * Defaults to MAX_VALUE as a default behavior since we rely on the cache time-based expiration. + */ + NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER("cassandra.nospam_logger.max_statements_per_logger", String.valueOf(Long.MAX_VALUE)), + METRICS_REPORTER_CONFIG_FILE("cassandra.metricsReporterConfigFile"), /** Defines the maximum number of unique timed out queries that will be reported in the logs. Use a negative number to remove any limit. */ MONITORING_MAX_OPERATIONS("cassandra.monitoring_max_operations", "50"), diff --git a/src/java/org/apache/cassandra/utils/NoSpamLogger.java b/src/java/org/apache/cassandra/utils/NoSpamLogger.java index a2bf815b30f5..09f04d6d8729 100644 --- a/src/java/org/apache/cassandra/utils/NoSpamLogger.java +++ b/src/java/org/apache/cassandra/utils/NoSpamLogger.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.utils; +import java.util.concurrent.ForkJoinPool; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Supplier; @@ -27,6 +28,12 @@ import org.slf4j.Logger; import static org.apache.cassandra.utils.Clock.Global; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Expiry; +import com.github.benmanes.caffeine.cache.Ticker; + +import static org.apache.cassandra.config.CassandraRelevantProperties.NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER; /** * Logging that limits each log statement to firing based on time since the statement last fired. @@ -36,8 +43,11 @@ * result in the original time being used. No warning is provided if there is a mismatch. * * If the statement is cached and used to log directly then only a volatile read will be required in the common case. - * If the Logger is cached then there is a single concurrent hash map lookup + the volatile read. - * If neither the logger nor the statement is cached then it is two concurrent hash map lookups + the volatile read. + * If the Logger is cached then there is a single Caffeine cache lookup + the volatile read. + * If neither the logger nor the statement is cached then it is a NonBlockingHashMap lookup + a Caffeine cache lookup + the volatile read. + * + * The implementation uses Caffeine cache with time-based expiration to automatically evict log statements + * after their minimum interval has passed, preventing unbounded memory growth from dynamic log messages. * */ public class NoSpamLogger @@ -64,6 +74,9 @@ public static void unsafeSetClock(Clock clock) CLOCK = clock; } + @VisibleForTesting + static Ticker TICKER = Ticker.systemTicker(); + public class NoSpamLogStatement extends AtomicLong { private static final long serialVersionUID = 1L; @@ -157,6 +170,11 @@ public boolean error(Object... objects) { return NoSpamLogStatement.this.error(CLOCK.nanoTime(), objects); } + + public long expiry() + { + return minIntervalNanos; + } } private static final NonBlockingHashMap wrappedLoggers = new NonBlockingHashMap<>(); @@ -167,6 +185,28 @@ static void clearWrappedLoggersForTest() wrappedLoggers.clear(); } + /** + * Forces eviction of entries from the {@link NoSpamLogStatement} cache for this logger instance. + * This is useful for testing to ensure cache size limits are enforced immediately. + */ + @VisibleForTesting + void cleanUpStatementsForTest() + { + lastMessage.cleanUp(); + } + + /** + * Returns the current size of the lastMessage cache for this logger instance. + * This is useful for testing cache eviction behavior. + * + * @return the number of log statements currently cached for this logger + */ + @VisibleForTesting + long getStatementsCount() + { + return lastMessage.estimatedSize(); + } + public static NoSpamLogger getLogger(Logger logger, long minInterval, TimeUnit unit) { NoSpamLogger wrapped = wrappedLoggers.get(logger); @@ -222,7 +262,41 @@ public static NoSpamLogStatement getStatement(Logger logger, String message, lon private final Logger wrapped; private final long minIntervalNanos; - private final NonBlockingHashMap lastMessage = new NonBlockingHashMap<>(); + + /** + * Cache of NoSpamLogStatement instances per NoSpamLogger instance. + * Bounded by size and time to prevent memory exhaustion from dynamic log messages. + * Uses Caffeine with W-TinyLFU eviction policy. + * Uses custom per-entry expiry based on each statement's minIntervalNanos. + */ + private final Cache lastMessage = Caffeine.newBuilder() + .maximumSize(NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER.getLong()) + .expireAfter(new Expiry() + { + @Override + public long expireAfterCreate(String key, NoSpamLogStatement value, long currentTime) + { + return value.expiry(); + } + + @Override + public long expireAfterUpdate(String key, NoSpamLogStatement value, + long currentTime, long currentDuration) + { + return value.expiry(); + } + + @Override + public long expireAfterRead(String key, NoSpamLogStatement value, + long currentTime, long currentDuration) + { + return currentDuration; + } + }) + .ticker(TICKER) + .executor(ForkJoinPool.commonPool()) + .recordStats() + .build(); private NoSpamLogger(Logger wrapped, long minInterval, TimeUnit timeUnit) { @@ -302,14 +376,6 @@ public NoSpamLogStatement getStatement(String s, long minIntervalNanos) public NoSpamLogStatement getStatement(String key, String s, long minIntervalNanos) { - NoSpamLogStatement statement = lastMessage.get(key); - if (statement == null) - { - statement = new NoSpamLogStatement(s, minIntervalNanos); - NoSpamLogStatement temp = lastMessage.putIfAbsent(key, statement); - if (temp != null) - statement = temp; - } - return statement; + return lastMessage.get(key, k -> new NoSpamLogStatement(s, minIntervalNanos)); } } diff --git a/test/unit/org/apache/cassandra/utils/NoSpamLoggerTest.java b/test/unit/org/apache/cassandra/utils/NoSpamLoggerTest.java index 93b036028381..16d4e9a7d1ff 100644 --- a/test/unit/org/apache/cassandra/utils/NoSpamLoggerTest.java +++ b/test/unit/org/apache/cassandra/utils/NoSpamLoggerTest.java @@ -39,6 +39,8 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; public class NoSpamLoggerTest @@ -81,11 +83,13 @@ public boolean equals(Object o) static final String statement = "swizzle{}"; static final String param = ""; static long now; + static long tickerTime; @BeforeClass public static void setUpClass() throws Exception { NoSpamLogger.unsafeSetClock(() -> now); + NoSpamLogger.TICKER = () -> tickerTime; } @Before @@ -278,9 +282,9 @@ public void testLoggedResult() now = 45; - assertTrue(nospamStatement.error(param)); - checkMock(Level.ERROR); - } + assertTrue(nospamStatement.error(param)); + checkMock(Level.ERROR); + } @Test public void testSupplierLogging() @@ -311,4 +315,250 @@ public void testSupplierLogging() assertEquals("TESTING {}", loggedMsg.left); assertArrayEquals(params, loggedMsg.right); } + + /** + * Test that the {@link NoSpamLogStatement} cache is bounded and doesn't grow beyond max_statements_per_logger. + * This prevents memory exhaustion from dynamic log messages (e.g., queries with unique strings). + */ + @Test + public void testNoSpamLogStatementCacheBounded() + { + int maxStatementsPerLogger = 10; + System.setProperty("cassandra.nospam_logger.max_statements_per_logger", String.valueOf(maxStatementsPerLogger)); + try + { + NoSpamLogger.clearWrappedLoggersForTest(); + now = 5; + NoSpamLogger logger = NoSpamLogger.getLogger(mock, 5, TimeUnit.NANOSECONDS); + + // Create more unique log statements than the cache can hold + int numberOfLogStatements = (int) (maxStatementsPerLogger * 1.5); + for (int i = 0; i < numberOfLogStatements; i++) + { + String uniqueStatement = "statement" + i + "{}"; + assertTrue("First occurrence of statement " + i + " should succeed", + logger.info(uniqueStatement, param)); + now += 10; // Advance time so each statement can log + } + + assertEquals(numberOfLogStatements, logged.get(Level.INFO).size()); + + // Force cache cleanup to ensure eviction has completed + logger.cleanUpStatementsForTest(); + + // Verify the cache size is bounded to the configured maximum + assertTrue("Cache size should be at most " + maxStatementsPerLogger, logger.getStatementsCount() <= maxStatementsPerLogger); + } + finally + { + System.clearProperty("cassandra.nospam_logger.max_statements_per_logger"); + NoSpamLogger.clearWrappedLoggersForTest(); + } + } + + /** + * Test that log statements expire after the configured inactivity period. + */ + @Test + public void testNoSpamLogStatementsCacheTimeBasedEviction() + { + try + { + int minIntervalInseconds = 10; + NoSpamLogger.clearWrappedLoggersForTest(); + now = 0; + tickerTime = 0; + NoSpamLogger logger = NoSpamLogger.getLogger(mock, minIntervalInseconds, TimeUnit.SECONDS); + + assertTrue(logger.info("test{}", param)); + assertEquals(1, logged.get(Level.INFO).size()); + assertEquals("Cache should contain 1 statement", 1, logger.getStatementsCount()); + + // Try to log again immediately - should be rate-limited + assertFalse(logger.info("test{}", param)); + assertEquals(1, logged.get(Level.INFO).size()); + assertEquals("Cache should still contain 1 statement", 1, logger.getStatementsCount()); + + // Advance BOTH clocks by more than `minIntervalInseconds` seconds + // `now` is used for rate limiting (NoSpamLogger.CLOCK) + // `tickerTime` is used for cache expiration (Caffeine's Ticker) + long advanceTime = TimeUnit.SECONDS.toNanos(minIntervalInseconds + 1); + now += advanceTime; + tickerTime += advanceTime; + + // Trigger cache cleanup to process expired entries + logger.cleanUpStatementsForTest(); + + // Verify the statement was evicted from cache + assertEquals("Cache should be empty after expiration", 0, logger.getStatementsCount()); + + // The statement should have expired from cache, so it should log again + assertTrue("Statement should have expired and can log again", + logger.info("test{}", param)); + assertEquals(2, logged.get(Level.INFO).size()); + assertEquals("Cache should contain 1 statement again", 1, logger.getStatementsCount()); + } + finally + { + NoSpamLogger.clearWrappedLoggersForTest(); + } + } + + /** + * Test that NoSpamLogger instances are cached and reused. + * This test verifies that getting the same logger returns the cached instance, + * and that clearing the cache creates new instances. + */ + @Test + public void testNoSpamLoggerCaching() + { + NoSpamLogger.clearWrappedLoggersForTest(); + now = 0; + + // Create multiple unique logger instances + Logger logger1 = new SubstituteLogger("testLogger1", null, true) + { + @Override + public void info(String statement, Object... args) + { + logged.get(Level.INFO).offer(Pair.create(statement, args)); + } + + @Override + public int hashCode() + { + return System.identityHashCode(this); + } + + @Override + public boolean equals(Object o) + { + return this == o; + } + }; + + Logger logger2 = new SubstituteLogger("testLogger2", null, true) + { + @Override + public void info(String statement, Object... args) + { + logged.get(Level.INFO).offer(Pair.create(statement, args)); + } + + @Override + public int hashCode() + { + return System.identityHashCode(this); + } + + @Override + public boolean equals(Object o) + { + return this == o; + } + }; + + // Get NoSpamLogger instances - these should be cached + NoSpamLogger nsl1 = NoSpamLogger.getLogger(logger1, 5, TimeUnit.NANOSECONDS); + NoSpamLogger nsl2 = NoSpamLogger.getLogger(logger2, 5, TimeUnit.NANOSECONDS); + + assertTrue(nsl1.info("test{}", param)); + assertTrue(nsl2.info("test{}", param)); + assertEquals(2, logged.get(Level.INFO).size()); + + // Verify that getting the same logger returns the cached instance + NoSpamLogger nsl1Again = NoSpamLogger.getLogger(logger1, 5, TimeUnit.NANOSECONDS); + assertSame("Should return cached instance", nsl1, nsl1Again); + + // Forcefully clear all cached loggers + NoSpamLogger.clearWrappedLoggersForTest(); + + // Getting the logger again should create a new instance + NoSpamLogger nsl1New = NoSpamLogger.getLogger(logger1, 5, TimeUnit.NANOSECONDS); + assertNotSame("Should create new instance after cache clear", nsl1, nsl1New); + + // Verify the new instance works correctly + assertTrue("New logger instance should log immediately", nsl1New.info("test{}", param)); + assertEquals(3, logged.get(Level.INFO).size()); + } + + /** + * Test that the NoSpamLogStatement cache uses custom per-entry expiry based on each logger's minIntervalNanos. + * This test verifies that different NoSpamLogger instances with different intervals result in + * different expiry times for their cached statements. + */ + @Test + public void testNoSpamLogStatementCacheCustomExpiry() + { + NoSpamLogger.clearWrappedLoggersForTest(); + now = 0; + tickerTime = 0; + + // Create three NoSpamLogger instances with different intervals + int[] intervals = { 2, 5, 10 }; + NoSpamLogger[] loggers = new NoSpamLogger[intervals.length]; + int logMessagesPerLogger = 3; + for (int i = 0; i < intervals.length; i++) + { + // Create a unique Logger instance for each interval to get separate NoSpamLogger instances + Logger testLogger = new SubstituteLogger("testLogger" + i, null, true) + { + @Override + public void info(String statement, Object... args) + { + logged.get(Level.INFO).offer(Pair.create(statement, args)); + } + + @Override + public int hashCode() + { + return System.identityHashCode(this); + } + + @Override + public boolean equals(Object o) + { + return this == o; + } + }; + + loggers[i] = NoSpamLogger.getLogger(testLogger, intervals[i], TimeUnit.SECONDS); + + // Log 3 messages from each logger + for (int j = 1; j <= logMessagesPerLogger; j++) + { + assertTrue(loggers[i].info("message" + j)); + now += intervals[i] * 1_000_000_000L + 1; // Advance past the interval to allow next log + } + assertEquals(logMessagesPerLogger, loggers[i].getStatementsCount()); + } + + assertEquals(logMessagesPerLogger * intervals.length, logged.get(Level.INFO).size()); + + // Test expiry at different time points + // Entries were created at tickerTime=0, so they expire at their interval time + int[] checkTimes = new int[intervals.length]; + for (int i = 0; i < intervals.length; i++) + { + // Set check time to 1 second after expiry (entries expire at interval seconds) + checkTimes[i] = intervals[i] + 1; + } + + for (int timeIdx = 0; timeIdx < checkTimes.length; timeIdx++) + { + tickerTime = TimeUnit.SECONDS.toNanos(checkTimes[timeIdx]); + + for (int i = 0; i < loggers.length; i++) + { + loggers[i].cleanUpStatementsForTest(); + + // Entries expire at (creation_time + interval), created at time 0 + // So they expire when tickerTime > interval + int expected = (intervals[i] < checkTimes[timeIdx]) ? 0 : logMessagesPerLogger; + assertEquals(String.format("After %ds, %d-second logger should have %d statements", + checkTimes[timeIdx], intervals[i], expected), + expected, loggers[i].getStatementsCount()); + } + } + } } From d3f896c56f5611e690f70dbd10dd3a46f2bd6063 Mon Sep 17 00:00:00 2001 From: viktoriiakotovets Date: Tue, 4 Aug 2026 18:43:52 +0200 Subject: [PATCH 2/2] CASSANDRA-21474: Move NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER in alphabetically order --- .../config/CassandraRelevantProperties.java | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index bebdfc0839a9..ccada98b6fdc 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -407,13 +407,6 @@ public enum CassandraRelevantProperties MEMTABLE_SHARD_COUNT("cassandra.memtable.shard.count"), MEMTABLE_TRIE_SIZE_LIMIT("cassandra.trie_size_limit_mb"), - /** - * Maximum number of log statements cached per NoSpamLogger instance. - * This prevents unbounded memory growth when log messages contain dynamic content. - * Defaults to MAX_VALUE as a default behavior since we rely on the cache time-based expiration. - */ - NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER("cassandra.nospam_logger.max_statements_per_logger", String.valueOf(Long.MAX_VALUE)), - METRICS_REPORTER_CONFIG_FILE("cassandra.metricsReporterConfigFile"), /** Defines the maximum number of unique timed out queries that will be reported in the logs. Use a negative number to remove any limit. */ MONITORING_MAX_OPERATIONS("cassandra.monitoring_max_operations", "50"), @@ -436,6 +429,12 @@ public enum CassandraRelevantProperties * This is an optimization used in unit tests becuase we never restart a node there. The only node is stopoped * when the JVM terminates. Therefore, we can use such optimization and not wait unnecessarily. */ NON_GRACEFUL_SHUTDOWN("cassandra.test.messagingService.nonGracefulShutdown"), + /** + * Maximum number of log statements cached per NoSpamLogger instance. + * This prevents unbounded memory growth when log messages contain dynamic content. + * Defaults to MAX_VALUE as a default behavior since we rely on the cache time-based expiration. + */ + NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER("cassandra.nospam_logger.max_statements_per_logger", String.valueOf(Long.MAX_VALUE)), /** for specific tests */ /** This property indicates whether disable_mbean_registration is true */ ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION("org.apache.cassandra.disable_mbean_registration"),