Add log throttling capability.

include/tscore/Diags.h

@@ -41,6 +41,7 @@
 #include "ink_inet.h"
 #include "BaseLogFile.h"
 #include "SourceLocation.h"
+#include "ThrottledMessage.h"
 
 #define DIAGS_MAGIC 0x12345678
 #define BYTES_IN_MB 1000000
@@ -285,6 +286,9 @@ class Diags
 
 extern inkcoreapi Diags *diags;
 
+// Note that the log functions being implemented as a macro has the advantage
+// that the pre-compiler expands this in place such that the call to
+// MakeSourceLocation happens at the call site for the function.
 #define DiagsError(level, fmt, ...)                \
   do {                                             \
     SourceLocation loc = MakeSourceLocation();     \
@@ -313,6 +317,18 @@ extern inkcoreapi Diags *diags;
 #define AlertV(fmt, ap) DiagsErrorV(DL_Alert, fmt, ap)
 #define EmergencyV(fmt, ap) DiagsErrorV(DL_Emergency, fmt, ap)
 
+#define THROTTLED_MESSAGE(throttle_func, ...)          \
+  do {                                                 \
+    const SourceLocation loc = MakeSourceLocation();   \
+    static ThrottledMessage throttled_message;         \
+    throttled_message.throttle_func(loc, __VA_ARGS__); \
+  } while (0)
+
+#define ThrottledStatus(...) THROTTLED_MESSAGE(status, __VA_ARGS__)
+#define ThrottledNote(...) THROTTLED_MESSAGE(note, __VA_ARGS__)
+#define ThrottledWarning(...) THROTTLED_MESSAGE(warning, __VA_ARGS__)
+#define ThrottledError(...) THROTTLED_MESSAGE(error, __VA_ARGS__)
+
 #if TS_USE_DIAGS
 
 #define Diag(tag, ...)                                 \

include/tscore/ThrottledMessage.h

@@ -0,0 +1,54 @@
+/** @file
+
+  ThrottledMessage declaration.
+
+  @section license License
+
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF 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.
+ */
+
+#pragma once
+
+#include "SourceLocation.h"
+#include "Throttler.h"
+
+#include <chrono>
+
+/** Represents a set of log messages for which throttling is desired. */
+class ThrottledMessage
+{
+public:
+  /**
+   * @param[in] seconds_between The minimum number of desired seconds between
+   * log events.
+   *
+   * TODO: Add proxy.config.log.throttle_delay? We'll have to be clear that
+   * this would only throttle the subset of programmatically registered log
+   * messages.
+   */
+  ThrottledMessage(std::chrono::seconds seconds_between = std::chrono::seconds{60});
+
+  /* TODO: Add BufferWriter overloads for these. */
+  void status(SourceLocation const &loc, const char *fmt, ...);
+  void note(SourceLocation const &loc, const char *fmt, ...);
+  void warning(SourceLocation const &loc, const char *fmt, ...);
+  void error(SourceLocation const &loc, const char *fmt, ...);
+
+private:
+  /** Used to throttle the log messages to the specified periodicity. */
+  Throttler _throttler;
+};

include/tscore/Throttler.h

@@ -0,0 +1,102 @@
+/** @file
+
+  A class for generic throttling.
+
+  @section license License
+
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF 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.
+ */
+
+#pragma once
+
+#include <atomic>
+#include <chrono>
+#include <cstdint>
+
+/** A class that exposes an interface for generic throttling of some action
+ * against a certain periodicity.
+ *
+ * To use:
+ *
+ * 1. Create an instance of this class specifying the periodicity for which
+ * something should be throttled.
+ *
+ * 2. Prepend each decision for a given throttled action with a boolean call
+ * against the instance created in step one.
+ *
+ *   2a. If the boolean operation returns true, then at least the configured
+ *   number of milliseconds has elapsed since the last time the operation
+ *   returned true. The number of times the check has been called is provided
+ *   in the skipped_count output parameter.
+ *
+ *   2b. If the boolean operation returns false, then not enough time has
+ *   elapsed since the last time the operation returned true.
+ *
+ * For instance:
+ *
+ *    void foo()
+ *    {
+ *      using namespace std::chrono_literals;
+ *      static Throttler t(300ms);
+ *      uint64_t skipped_count;
+ *      if (t(skipped_count)) {
+ *        std::printf("Alan bought another monitor\n");
+ *        std::printf("We ignored Alan buying a monitor %llu times\n", skipped_count);
+ *      }
+ *    }
+ */
+class Throttler
+{
+public:
+  /**
+   * @param[in] periodicity The minimum number of milliseconds between
+   * calls to Throttler which should return true.
+   */
+  Throttler(std::chrono::milliseconds periodicity);
+
+  /** Whether enough time has passed since the last allowed action.
+   *
+   * @param[out] skipped_count If the return of this call is true,
+   * this is populated with the approximate number of times the operator has
+   * been queried since before this function was called. Otherwise the value is
+   * not set.
+   *
+   * @return True if the action is emitted per the configured periodicity,
+   * false otherwise.
+   */
+  bool operator()(uint64_t &skipped_count);
+
+private:
+  /// Base clock.
+  using Clock = std::chrono::system_clock;
+  /// Time point type, based on the clock to be used.
+  using TimePoint = Clock::time_point;
+
+  /// Time that the last item was emitted.
+  // It is strange that we need to explicitly default construct this with a
+  // default constructed TimePoint. Without it, however, I get a compiler error
+  // in gcc 8.x and 9.x.  Playing around in godbolt I notice that neither clang
+  // nor gcc versions starting from 10.x require this, so I suspect it is a
+  // compiler bug.
+  std::atomic<TimePoint> _last_allowed_time{TimePoint{}};
+
+  /// The minimum number of milliseconds desired between actions.
+  std::chrono::milliseconds const _periodicity{0};
+
+  /// The number of calls to Throttler since the last
+  uint64_t _skipped_count = 0;
+};

proxy/logging/Log.cc

@@ -1376,7 +1376,7 @@ Log::flush_thread_main(void * /* args ATS_UNUSED */)
       // make sure we're open & ready to write
       logfile->check_fd();
       if (!logfile->is_open()) {
-        Warning("File:%s was closed, have dropped (%d) bytes.", logfile->get_name(), total_bytes);
+        ThrottledWarning("File:%s was closed, have dropped (%d) bytes.", logfile->get_name(), total_bytes);
 
         RecIncrRawStat(log_rsb, mutex->thread_holding, log_stat_bytes_lost_before_written_to_disk_stat, total_bytes);
         delete fdata;
@@ -1402,8 +1402,8 @@ Log::flush_thread_main(void * /* args ATS_UNUSED */)
         len = ::write(logfilefd, &buf[bytes_written], total_bytes - bytes_written);
 
         if (len < 0) {
-          Error("Failed to write log to %s: [tried %d, wrote %d, %s]", logfile->get_name(), total_bytes - bytes_written,
-                bytes_written, strerror(errno));
+          ThrottledError("Failed to write log to %s: [tried %d, wrote %d, %s]", logfile->get_name(), total_bytes - bytes_written,
+                         bytes_written, strerror(errno));
 
           RecIncrRawStat(log_rsb, mutex->thread_holding, log_stat_bytes_lost_before_written_to_disk_stat,
                          total_bytes - bytes_written);

proxy/logging/LogFile.cc

@@ -718,7 +718,7 @@ LogFile::writeln(char *data, int len, int fd, const char *path)
     }
 
     if ((bytes_this_write = static_cast<int>(::writev(fd, (const struct iovec *)wvec, vcnt))) < 0) {
-      Warning("An error was encountered in writing to %s: %s.", ((path) ? path : "logfile"), strerror(errno));
+      ThrottledWarning("An error was encountered in writing to %s: %s.", ((path) ? path : "logfile"), strerror(errno));
     } else {
       total_bytes = bytes_this_write;
     }

src/tscore/Makefile.am

@@ -117,6 +117,8 @@ libtscore_la_SOURCES = \
 	signals.cc \
 	SourceLocation.cc \
 	TextBuffer.cc \
+	ThrottledMessage.cc \
+	Throttler.cc \
 	Tokenizer.cc \
 	ts_file.cc \
 	Version.cc \
@@ -177,12 +179,13 @@ test_tscore_SOURCES = \
 	unit_tests/test_List.cc \
 	unit_tests/test_MemArena.cc \
 	unit_tests/test_MT_hashtable.cc \
-  unit_tests/test_ParseRules.cc \
+	unit_tests/test_ParseRules.cc \
 	unit_tests/test_PriorityQueue.cc \
 	unit_tests/test_Ptr.cc \
 	unit_tests/test_Regex.cc \
 	unit_tests/test_Scalar.cc \
 	unit_tests/test_scoped_resource.cc \
+	unit_tests/test_Throttler.cc \
 	unit_tests/test_Tokenizer.cc \
 	unit_tests/test_ts_file.cc \
 	unit_tests/test_Version.cc \

src/tscore/ThrottledMessage.cc

@@ -0,0 +1,81 @@
+/** @file
+
+  ThrottledMessage declaration.
+
+  @section license License
+
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF 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.
+ */
+
+#include "tscore/ThrottledMessage.h"
+
+#include "tscore/Diags.h"
+
+namespace
+{
+void
+message_helper(DiagsLevel level, SourceLocation const &loc, Throttler &throttler, const char *fmt, va_list ap)
+{
+  uint64_t count_since_last_log = 0;
+  if (!throttler(count_since_last_log)) {
+    return;
+  }
+  if (count_since_last_log > 0) {
+    diags->error(level, &loc, "Skipped the following message %lu times.", count_since_last_log);
+  }
+  diags->error_va(level, &loc, fmt, ap);
+}
+
+} // namespace
+
+ThrottledMessage::ThrottledMessage(std::chrono::seconds seconds_between) : _throttler{seconds_between} {}
+
+void
+ThrottledMessage::status(SourceLocation const &loc, const char *fmt, ...)
+{
+  va_list args;
+  va_start(args, fmt);
+  message_helper(DL_Status, loc, _throttler, fmt, args);
+  va_end(args);
+}
+
+void
+ThrottledMessage::note(SourceLocation const &loc, const char *fmt, ...)
+{
+  va_list args;
+  va_start(args, fmt);
+  message_helper(DL_Note, loc, _throttler, fmt, args);
+  va_end(args);
+}
+
+void
+ThrottledMessage::warning(SourceLocation const &loc, const char *fmt, ...)
+{
+  va_list args;
+  va_start(args, fmt);
+  message_helper(DL_Warning, loc, _throttler, fmt, args);
+  va_end(args);
+}
+
+void
+ThrottledMessage::error(SourceLocation const &loc, const char *fmt, ...)
+{
+  va_list args;
+  va_start(args, fmt);
+  message_helper(DL_Error, loc, _throttler, fmt, args);
+  va_end(args);
+}

src/tscore/Throttler.cc

@@ -0,0 +1,42 @@
+/** @file
+
+  Implement Throttler.
+
+  @section license License
+
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF 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.
+ */
+
+#include "tscore/Throttler.h"
+
+Throttler::Throttler(std::chrono::milliseconds periodicity) : _periodicity{periodicity} {}
+
+bool
+Throttler::operator()(uint64_t &skipped_count)
+{
+  TimePoint const now = Clock::now();
+  TimePoint last_allowed_time{_last_allowed_time};
+  if ((last_allowed_time + _periodicity) <= now) {
+    if (_last_allowed_time.compare_exchange_strong(last_allowed_time, now)) {
+      skipped_count  = _skipped_count;
+      _skipped_count = 0;
+      return true;
+    }
+  }
+  ++_skipped_count;
+  return false;
+}