Refactoring throttled logging

Author: jonathan-buttner

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/InferencePlugin.java

@@ -243,7 +243,9 @@ public List<RestHandler> getRestHandlers(
     @Override
     public Collection<?> createComponents(PluginServices services) {
         var components = new ArrayList<>();
-        var throttlerManager = new ThrottlerManager(settings, services.threadPool(), services.clusterService());
+        var throttlerManager = new ThrottlerManager(settings, services.threadPool());
+        throttlerManager.init(services.clusterService());
+
         var truncator = new Truncator(settings, services.clusterService());
         serviceComponents.set(new ServiceComponents(services.threadPool(), throttlerManager, settings, truncator));
         threadPoolSetOnce.set(services.threadPool());

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/logging/Throttler.java

@@ -9,150 +9,185 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.threadpool.Scheduler;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.Closeable;
 import java.time.Clock;
-import java.time.Duration;
 import java.time.Instant;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
-import static org.elasticsearch.core.Strings.format;
 import static org.elasticsearch.xpack.inference.InferencePlugin.UTILITY_THREAD_POOL_NAME;
 
 /**
- * A class that throttles calls to a logger. If a log call is made during the throttle period a counter is incremented.
- * If a log call occurs after the throttle period, then the call will proceed, and it will include a message like
- * "repeated X times" to indicate how often the message was attempting to be logged.
+ * A class that throttles calls to a logger. The first unique log message is permitted to emit a message. Any subsequent log messages
+ * matching a message that has already been emitted will only increment a counter. A thread runs on an interval
+ * to emit any log messages that have been repeated beyond the initial emitted message. Once the thread emits a repeated
+ * message the counter is reset. If another message is received matching a previously emitted message by the thread, it will be consider
+ * the first time a unique message is received and will be logged.
  */
 public class Throttler implements Closeable {
 
     private static final Logger classLogger = LogManager.getLogger(Throttler.class);
 
-    private final TimeValue resetInterval;
-    private Duration durationToWait;
+    private final TimeValue loggingInterval;
     private final Clock clock;
     private final ConcurrentMap<String, LogExecutor> logExecutors;
     private final AtomicReference<Scheduler.Cancellable> cancellableTask = new AtomicReference<>();
     private final AtomicBoolean isRunning = new AtomicBoolean(true);
+    private final ThreadPool threadPool;
 
     /**
-     * Constructs the throttler and kicks of a scheduled tasks to clear the internal stats.
-     *
-     * @param resetInterval the frequency for clearing the internal stats. This protects against an ever growing
-     *                      cache
-     * @param durationToWait the amount of time to wait before logging a message after the threshold
-     *                       is reached
+     * @param loggingInterval the frequency to run a task to emit repeated log messages
      * @param threadPool a thread pool for running a scheduled task to clear the internal stats
      */
-    public Throttler(TimeValue resetInterval, TimeValue durationToWait, ThreadPool threadPool) {
-        this(resetInterval, durationToWait, Clock.systemUTC(), threadPool, new ConcurrentHashMap<>());
+    public Throttler(TimeValue loggingInterval, ThreadPool threadPool) {
+        this(loggingInterval, Clock.systemUTC(), threadPool, new ConcurrentHashMap<>());
+    }
+
+    /**
+     * @param oldThrottler a previous throttler that is being replaced
+     * @param loggingInterval the frequency to run a task to emit repeated log messages
+     */
+    public Throttler(Throttler oldThrottler, TimeValue loggingInterval) {
+        this(loggingInterval, oldThrottler.clock, oldThrottler.threadPool, new ConcurrentHashMap<>(oldThrottler.logExecutors));
     }
 
     /**
      * This should only be used directly for testing.
      */
-    Throttler(
-        TimeValue resetInterval,
-        TimeValue durationToWait,
-        Clock clock,
-        ThreadPool threadPool,
-        ConcurrentMap<String, LogExecutor> logExecutors
-    ) {
-        Objects.requireNonNull(durationToWait);
-        Objects.requireNonNull(threadPool);
-
-        this.resetInterval = Objects.requireNonNull(resetInterval);
-        this.durationToWait = Duration.ofMillis(durationToWait.millis());
+    Throttler(TimeValue loggingInterval, Clock clock, ThreadPool threadPool, ConcurrentMap<String, LogExecutor> logExecutors) {
+        this.threadPool = Objects.requireNonNull(threadPool);
+        this.loggingInterval = Objects.requireNonNull(loggingInterval);
         this.clock = Objects.requireNonNull(clock);
         this.logExecutors = Objects.requireNonNull(logExecutors);
+    }
 
-        this.cancellableTask.set(startResetTask(threadPool));
+    public void init() {
+        cancellableTask.set(startRepeatingLogEmitter());
     }
 
-    private Scheduler.Cancellable startResetTask(ThreadPool threadPool) {
-        classLogger.debug(() -> format("Reset task scheduled with interval [%s]", resetInterval));
+    private Scheduler.Cancellable startRepeatingLogEmitter() {
+        classLogger.debug(() -> Strings.format("Scheduling repeating log emitter with interval [%s]", loggingInterval));
 
-        return threadPool.scheduleWithFixedDelay(logExecutors::clear, resetInterval, threadPool.executor(UTILITY_THREAD_POOL_NAME));
+        return threadPool.scheduleWithFixedDelay(this::emitRepeatedLogs, loggingInterval, threadPool.executor(UTILITY_THREAD_POOL_NAME));
     }
 
-    public void setDurationToWait(TimeValue durationToWait) {
-        this.durationToWait = Duration.ofMillis(durationToWait.millis());
+    private void emitRepeatedLogs() {
+        if (isRunning.get() == false) {
+            return;
+        }
+
+        for (var iter = logExecutors.values().iterator(); iter.hasNext();) {
+            var executor = iter.next();
+            iter.remove();
+            executor.logRepeatedMessages();
+        }
     }
 
-    public void execute(String message, Consumer<String> consumer) {
+    public void execute(String message, Consumer<String> logCallback) {
         if (isRunning.get() == false) {
             return;
         }
 
-        LogExecutor logExecutor = logExecutors.compute(message, (key, value) -> {
+        var logExecutor = logExecutors.compute(message, (key, value) -> {
             if (value == null) {
-                return new LogExecutor(clock, consumer);
+                return new LogExecutor(clock, logCallback, message);
             }
 
-            return value.compute(consumer, durationToWait);
+            return value;
         });
 
-        // This executes an internal consumer that wraps the passed in one, it will either log the message passed here
-        // unchanged, do nothing if it is in the throttled period, or log this message + some text saying how many times it was repeated
-        logExecutor.log(message);
+        logExecutor.logFirstMessage();
     }
 
     @Override
     public void close() {
         isRunning.set(false);
-        cancellableTask.get().cancel();
+        if (cancellableTask.get() != null) {
+            cancellableTask.get().cancel();
+        }
         logExecutors.clear();
     }
 
-    private static class LogExecutor {
-        private final long skippedLogCalls;
-        private final Instant timeOfLastLogCall;
+    // default for testing
+    static class LogExecutor {
+        // -1 here because we need to determine if we haven't logged the first time
+        // After the first time we'll set it to 0, then the thread that runs on an interval
+        // needs to know if there are any repeated message, if it sees 0, it knows there are none
+        // and skips emitting the message again.
+        private static final long INITIAL_LOG_COUNTER_VALUE = -1;
+
+        private final AtomicLong skippedLogCalls;
+        private Instant timeOfLastLogCall;
         private final Clock clock;
-        private final Consumer<String> consumer;
+        private final Consumer<String> throttledConsumer;
+        private final String originalMessage;
 
-        LogExecutor(Clock clock, Consumer<String> throttledConsumer) {
-            this(clock, 0, throttledConsumer);
+        LogExecutor(Clock clock, Consumer<String> throttledConsumer, String originalMessage) {
+            this(clock, INITIAL_LOG_COUNTER_VALUE, throttledConsumer, originalMessage);
         }
 
-        LogExecutor(Clock clock, long skippedLogCalls, Consumer<String> consumer) {
-            this.skippedLogCalls = skippedLogCalls;
+        LogExecutor(Clock clock, long skippedLogCalls, Consumer<String> throttledConsumer, String originalMessage) {
+            this.skippedLogCalls = new AtomicLong(skippedLogCalls);
             this.clock = Objects.requireNonNull(clock);
             timeOfLastLogCall = Instant.now(this.clock);
-            this.consumer = Objects.requireNonNull(consumer);
+            this.throttledConsumer = Objects.requireNonNull(throttledConsumer);
+            this.originalMessage = Objects.requireNonNull(originalMessage);
         }
 
-        void log(String message) {
-            this.consumer.accept(message);
+        void logRepeatedMessages() {
+            long numSkippedLogCalls;
+            synchronized (skippedLogCalls) {
+                numSkippedLogCalls = skippedLogCalls.get();
+                if (hasRepeatedLogsToEmit(numSkippedLogCalls) == false) {
+                    return;
+                }
+            }
+
+            String enrichedMessage;
+            if (numSkippedLogCalls == 1) {
+                enrichedMessage = Strings.format("%s, repeated 1 time, last message at [%s]", originalMessage, timeOfLastLogCall);
+            } else {
+                enrichedMessage = Strings.format(
+                    "%s, repeated %s times, last message at [%s]",
+                    originalMessage,
+                    skippedLogCalls,
+                    timeOfLastLogCall
+                );
+            }
+
+            throttledConsumer.accept(enrichedMessage);
         }
 
-        LogExecutor compute(Consumer<String> executor, Duration durationToWait) {
-            if (hasDurationExpired(durationToWait)) {
-                String messageToAppend = "";
-                if (this.skippedLogCalls == 1) {
-                    messageToAppend = ", repeated 1 time";
-                } else if (this.skippedLogCalls > 1) {
-                    messageToAppend = format(", repeated %s times", this.skippedLogCalls);
-                }
+        private static boolean hasRepeatedLogsToEmit(long numSkippedLogCalls) {
+            return numSkippedLogCalls > 0;
+        }
 
-                final String stringToAppend = messageToAppend;
-                return new LogExecutor(this.clock, 0, (message) -> executor.accept(message.concat(stringToAppend)));
+        void logFirstMessage() {
+            long numSkippedLogCalls;
+            synchronized (skippedLogCalls) {
+                numSkippedLogCalls = skippedLogCalls.getAndIncrement();
             }
 
-            // This creates a consumer that won't do anything because the original consumer is being throttled
-            return new LogExecutor(this.clock, this.skippedLogCalls + 1, (message) -> {});
+            timeOfLastLogCall = Instant.now(this.clock);
+
+            if (hasLoggedOriginalMessage(numSkippedLogCalls) == false) {
+                this.throttledConsumer.accept(originalMessage);
+            }
         }
 
-        private boolean hasDurationExpired(Duration durationToWait) {
-            Instant now = Instant.now(clock);
-            return now.isAfter(timeOfLastLogCall.plus(durationToWait));
+        private static boolean hasLoggedOriginalMessage(long numSkippedLogCalls) {
+            // a negative value indicates that we haven't yet logged the original message
+            return numSkippedLogCalls >= 0;
         }
     }
 }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/logging/ThrottlerManager.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.core.UpdateForV10;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.Closeable;
@@ -24,8 +25,11 @@
 public class ThrottlerManager implements Closeable {
     private static final TimeValue DEFAULT_STATS_RESET_INTERVAL_TIME = TimeValue.timeValueDays(1);
     /**
-     * A setting specifying the interval for clearing the cached log message stats
+     * Legacy log throttling setting, kept for BWC compatibility. This setting has no effect in 9.1.0 and later. Do not use.
+     * TODO remove in 10.0
      */
+    @UpdateForV10(owner = UpdateForV10.Owner.MACHINE_LEARNING)
+    @Deprecated
     public static final Setting<TimeValue> STATS_RESET_INTERVAL_SETTING = Setting.timeSetting(
         "xpack.inference.logging.reset_interval",
         DEFAULT_STATS_RESET_INTERVAL_TIME,
@@ -35,48 +39,54 @@ public class ThrottlerManager implements Closeable {
 
     private static final TimeValue DEFAULT_WAIT_DURATION_TIME = TimeValue.timeValueHours(1);
     /**
-     * A setting specifying the amount of time to wait after a log call occurs before allowing another log call.
+     * Legacy log throttling setting, kept for BWC compatibility. This setting has no effect in 9.1.0 and later. Do not use.
+     * TODO remove in 10.0
      */
+    @UpdateForV10(owner = UpdateForV10.Owner.MACHINE_LEARNING)
+    @Deprecated
     public static final Setting<TimeValue> LOGGER_WAIT_DURATION_SETTING = Setting.timeSetting(
         "xpack.inference.logging.wait_duration",
         DEFAULT_WAIT_DURATION_TIME,
         Setting.Property.NodeScope,
         Setting.Property.Dynamic
     );
 
-    private final ThreadPool threadPool;
-    private Throttler throttler;
-    private LoggerSettings loggerSettings;
+    private static final TimeValue DEFAULT_LOG_EMIT_INTERVAL = TimeValue.timeValueHours(1);
+    /**
+     * This setting specifies how often a thread will run to emit repeated log messages.
+     */
+    public static final Setting<TimeValue> LOG_EMIT_INTERVAL = Setting.timeSetting(
+        "xpack.inference.logging.log_interval",
+        DEFAULT_LOG_EMIT_INTERVAL,
+        Setting.Property.NodeScope,
+        Setting.Property.Dynamic
+    );
 
-    public ThrottlerManager(Settings settings, ThreadPool threadPool, ClusterService clusterService) {
-        Objects.requireNonNull(settings);
-        Objects.requireNonNull(clusterService);
+    private volatile TimeValue logInterval;
 
-        this.threadPool = Objects.requireNonNull(threadPool);
-        this.loggerSettings = LoggerSettings.fromSettings(settings);
+    private Throttler throttler;
 
-        throttler = new Throttler(loggerSettings.resetInterval(), loggerSettings.waitDuration(), threadPool);
-        this.addSettingsUpdateConsumers(clusterService);
-    }
+    public ThrottlerManager(Settings settings, ThreadPool threadPool) {
+        Objects.requireNonNull(settings);
 
-    private void addSettingsUpdateConsumers(ClusterService clusterService) {
-        clusterService.getClusterSettings().addSettingsUpdateConsumer(STATS_RESET_INTERVAL_SETTING, this::setResetInterval);
-        clusterService.getClusterSettings().addSettingsUpdateConsumer(LOGGER_WAIT_DURATION_SETTING, this::setWaitDuration);
+        throttler = new Throttler(LOG_EMIT_INTERVAL.get(settings), threadPool);
+        throttler.init();
     }
 
-    // default for testing
-    void setWaitDuration(TimeValue waitDuration) {
-        loggerSettings = loggerSettings.createWithWaitDuration(waitDuration);
+    public void init(ClusterService clusterService) {
+        Objects.requireNonNull(clusterService);
 
-        throttler.setDurationToWait(waitDuration);
+        clusterService.getClusterSettings().addSettingsUpdateConsumer(LOG_EMIT_INTERVAL, this::setLogInterval);
     }
 
     // default for testing
-    void setResetInterval(TimeValue resetInterval) {
-        loggerSettings = loggerSettings.createWithResetInterval(resetInterval);
+    void setLogInterval(TimeValue logInterval) {
+        this.logInterval = logInterval;
 
-        throttler.close();
-        throttler = new Throttler(loggerSettings.resetInterval(), loggerSettings.waitDuration(), threadPool);
+        var oldThrottler = throttler;
+        throttler = new Throttler(oldThrottler, this.logInterval);
+        throttler.init();
+        oldThrottler.close();
     }
 
     // default for testing
@@ -103,25 +113,6 @@ public void close() {
     }
 
     public static List<Setting<?>> getSettingsDefinitions() {
-        return List.of(STATS_RESET_INTERVAL_SETTING, LOGGER_WAIT_DURATION_SETTING);
-    }
-
-    private record LoggerSettings(TimeValue resetInterval, TimeValue waitDuration) {
-        LoggerSettings {
-            Objects.requireNonNull(resetInterval);
-            Objects.requireNonNull(waitDuration);
-        }
-
-        static LoggerSettings fromSettings(Settings settings) {
-            return new LoggerSettings(STATS_RESET_INTERVAL_SETTING.get(settings), LOGGER_WAIT_DURATION_SETTING.get(settings));
-        }
-
-        LoggerSettings createWithResetInterval(TimeValue resetInterval) {
-            return new LoggerSettings(resetInterval, waitDuration);
-        }
-
-        LoggerSettings createWithWaitDuration(TimeValue waitDuration) {
-            return new LoggerSettings(resetInterval, waitDuration);
-        }
+        return List.of(LOG_EMIT_INTERVAL, STATS_RESET_INTERVAL_SETTING, LOGGER_WAIT_DURATION_SETTING);
     }
 }