diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 2be55fbe5c69..ccada98b6fdc 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -406,6 +406,7 @@ 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"), + 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"), @@ -428,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"), diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index b4c834355841..290a484b8e76 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -225,6 +225,7 @@ import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.MBeanWrapper; import org.apache.cassandra.utils.MD5Digest; +import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.OutputHandler; import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.Throwables; @@ -4058,6 +4059,15 @@ protected synchronized void drain(boolean isFinalShutdown) throws IOException, I logger.error("Failed to stop async profiler.", t); } + try + { + NoSpamLogger.shutdown(); + } + catch (Throwable t) + { + logger.warn("Unable to shutdown NoSpamLogger executor within 1 minute.", t); + } + try { // we are not shutting down ScheduledExecutors#scheduledFastTasks to be still able to progress time diff --git a/src/java/org/apache/cassandra/utils/NoSpamLogger.java b/src/java/org/apache/cassandra/utils/NoSpamLogger.java index a2bf815b30f5..1b5f48795f65 100644 --- a/src/java/org/apache/cassandra/utils/NoSpamLogger.java +++ b/src/java/org/apache/cassandra/utils/NoSpamLogger.java @@ -17,15 +17,23 @@ */ package org.apache.cassandra.utils; +import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Supplier; +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 com.google.common.annotations.VisibleForTesting; import org.cliffc.high_scale_lib.NonBlockingHashMap; import org.slf4j.Logger; +import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; +import static org.apache.cassandra.config.CassandraRelevantProperties.NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER; import static org.apache.cassandra.utils.Clock.Global; /** @@ -36,8 +44,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 +75,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,16 +171,51 @@ public boolean error(Object... objects) { return NoSpamLogStatement.this.error(CLOCK.nanoTime(), objects); } + + public long expiry() + { + return minIntervalNanos; + } } private static final NonBlockingHashMap wrappedLoggers = new NonBlockingHashMap<>(); + /** + * Shuts down the shared cache maintenance executor. Should be called during node drain/shutdown. + */ + public static void shutdown() throws InterruptedException, TimeoutException + { + ExecutorUtils.shutdownNowAndWait(1, TimeUnit.MINUTES, CACHE_MAINTENANCE_EXECUTOR); + } + @VisibleForTesting 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 +271,47 @@ public static NoSpamLogStatement getStatement(Logger logger, String message, lon private final Logger wrapped; private final long minIntervalNanos; - private final NonBlockingHashMap lastMessage = new NonBlockingHashMap<>(); + + /** + * Dedicated executor for Caffeine cache maintenance tasks (eviction, expiry) shared across all + * NoSpamLogger instances. Registered with JMX under "internal" path for observability. + */ + private static final ExecutorService CACHE_MAINTENANCE_EXECUTOR = executorFactory().withJmxInternal().sequential("NoSpamLogger"); + + /** + * 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(CACHE_MAINTENANCE_EXECUTOR) + .recordStats() + .build(); private NoSpamLogger(Logger wrapped, long minInterval, TimeUnit timeUnit) { @@ -302,14 +391,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..2ab0933ebc59 100644 --- a/test/unit/org/apache/cassandra/utils/NoSpamLoggerTest.java +++ b/test/unit/org/apache/cassandra/utils/NoSpamLoggerTest.java @@ -32,13 +32,17 @@ import org.slf4j.Logger; import org.slf4j.helpers.SubstituteLogger; +import org.apache.cassandra.distributed.shared.WithProperties; import org.apache.cassandra.utils.NoSpamLogger.Level; import org.apache.cassandra.utils.NoSpamLogger.NoSpamLogStatement; +import static org.apache.cassandra.config.CassandraRelevantProperties.NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER; import static org.junit.Assert.assertArrayEquals; 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 +85,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 +284,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 +317,249 @@ 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; + try (WithProperties properties = new WithProperties().set(NOSPAM_LOGGER_MAX_STATEMENTS_PER_LOGGER, + (long) maxStatementsPerLogger)) + { + 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 + { + 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()); + } + } + } }