Logger: Don't discard WARNING or ERROR messages.

When the logger reaches it's threshold for messages, don't discard the
WARNING or ERROR messages up to a certain amount. If THAT threshold
reaches its maximum, start discarding the oldest non-WARNING/ERROR
messages until we only have WARNING and ERROR messages left. Everything
will be discarded after that until we have room again for more.

Also added a test that limits the queue sizes and logs many messages to
see the queues hitting their maximums and discarding the appropriate
messages.

Fixes: #1882

include/asterisk/logger.h

@@ -530,6 +530,24 @@ void ast_logger_set_queue_limit(int queue_limit);
  */
 int ast_logger_get_queue_limit(void);
 
+/*!
+ * \brief Set the threshold for the maximum number of WARNING/ERROR messages allowed in the
+ * processing queue that can exceed the logger queue's threshold. This acts as a buffer for
+ * WARNING/ERROR messages so that we can continue to queue them while discarding other message
+ * types. This threshold sits on top of the logger queue's message threshold, so the actual
+ * queue size will be the logger queue threshold PLUS this threshold.
+ *
+ * \param limit
+ */
+void ast_logger_set_over_threshold_queue_limit(int limit);
+
+/*!
+ * \brief Get the over threshold maximum number of messages allowed in the processing queue
+ *
+ * \return Over threshold queue limit
+ */
+int ast_logger_get_over_threshold_queue_limit(void);
+
 
 /*! \defgroup Scope_Trace Scope Trace
  * @{

main/logger.c

@@ -70,6 +70,7 @@
 #include "asterisk/ast_version.h"
 #include "asterisk/backtrace.h"
 #include "asterisk/json.h"
+#include "asterisk/dlinkedlists.h"
 
 static int logger_register_level(const char *name);
 static int logger_unregister_level(const char *name);
@@ -91,8 +92,15 @@ AST_THREADSTORAGE(unique_callid);
 
 static int logger_queue_size;
 static int logger_queue_limit = 1000;
+static int logger_queue_over_threshold_limit = 200; /* How many specific messages (WARNING, ERROR) can go over the limit */
 static int logger_messages_discarded;
 static unsigned int high_water_alert;
+/* This gets set when we are at the absolute maximum number of WARNING/ERROR messages in the
+ * logger queue. This includes the logger queue threshold and the over threshold limit, which
+ * means that the queue is COMPLETELY full of WARNING/ERROR messages, so any new messages
+ * coming in will just need to be discarded, no matter what they are.
+ */
+static unsigned int discard_all_new_msgs;
 
 /* On some platforms, like those with MUSL as the runtime, BUFSIZ is
  * unreasonably small (1024). Use a larger value in those environments.
@@ -181,7 +189,7 @@ struct logmsg {
 		AST_STRING_FIELD(message);
 		AST_STRING_FIELD(level_name);
 	);
-	AST_LIST_ENTRY(logmsg) list;
+	AST_DLLIST_ENTRY(logmsg) list;
 };
 
 static void logmsg_free(struct logmsg *msg)
@@ -190,7 +198,15 @@ static void logmsg_free(struct logmsg *msg)
 	ast_free(msg);
 }
 
-static AST_LIST_HEAD_STATIC(logmsgs, logmsg);
+static AST_DLLIST_HEAD_STATIC(logmsgs, logmsg);
+
+/* This pointer should not be modified. It will be set if and only if the logger queue
+ * gets to a point where we need to start discarding messages to make room for new
+ * WARNING/ERROR messages. At that point, it will always point to the last logmsg that
+ * was looked at for discard, or NULL if we discarded everything we looked at, which
+ * means that we just start at the HEAD of the list for the next search.
+ */
+static struct logmsg *last_logmsg_checked_for_discard = NULL;
 static pthread_t logthread = AST_PTHREADT_NULL;
 static ast_cond_t logcond;
 static int close_logger_thread = 0;
@@ -2070,10 +2086,10 @@ static void *logger_thread(void *data)
 
 	for (;;) {
 		/* We lock the message list, and see if any message exists... if not we wait on the condition to be signalled */
-		AST_LIST_LOCK(&logmsgs);
-		if (AST_LIST_EMPTY(&logmsgs)) {
+		AST_DLLIST_LOCK(&logmsgs);
+		if (AST_DLLIST_EMPTY(&logmsgs)) {
 			if (close_logger_thread) {
-				AST_LIST_UNLOCK(&logmsgs);
+				AST_DLLIST_UNLOCK(&logmsgs);
 				break;
 			} else {
 				ast_cond_wait(&logcond, &logmsgs.lock);
@@ -2085,21 +2101,24 @@ static void *logger_thread(void *data)
 				"Logging resumed.  %d message%s discarded.\n",
 				logger_messages_discarded, logger_messages_discarded == 1 ? "" : "s");
 			if (msg) {
-				AST_LIST_INSERT_TAIL(&logmsgs, msg, list);
+				AST_DLLIST_INSERT_TAIL(&logmsgs, msg, list);
 			}
 			high_water_alert = 0;
 			logger_messages_discarded = 0;
 		}
 
-		next = AST_LIST_FIRST(&logmsgs);
-		AST_LIST_HEAD_INIT_NOLOCK(&logmsgs);
+		discard_all_new_msgs = 0;
+		last_logmsg_checked_for_discard = NULL;
+
+		next = AST_DLLIST_FIRST(&logmsgs);
+		AST_DLLIST_HEAD_INIT_NOLOCK(&logmsgs);
 		logger_queue_size = 0;
-		AST_LIST_UNLOCK(&logmsgs);
+		AST_DLLIST_UNLOCK(&logmsgs);
 
 		/* Otherwise go through and process each message in the order added */
 		while ((msg = next)) {
 			/* Get the next entry now so that we can free our current structure later */
-			next = AST_LIST_NEXT(msg, list);
+			next = AST_DLLIST_NEXT(msg, list);
 
 			/* Depending on the type, send it to the proper function */
 			logger_print_normal(msg);
@@ -2224,10 +2243,10 @@ void close_logger(void)
 	logger_initialized = 0;
 
 	/* Stop logger thread */
-	AST_LIST_LOCK(&logmsgs);
+	AST_DLLIST_LOCK(&logmsgs);
 	close_logger_thread = 1;
 	ast_cond_signal(&logcond);
-	AST_LIST_UNLOCK(&logmsgs);
+	AST_DLLIST_UNLOCK(&logmsgs);
 
 	if (logthread != AST_PTHREADT_NULL) {
 		pthread_join(logthread, NULL);
@@ -2353,6 +2372,49 @@ void ast_callid_threadstorage_auto_clean(ast_callid callid, int callid_created)
 	}
 }
 
+/*!
+ * \internal
+ * \brief Flush the oldest non-warning/error log messages out of the queue
+ *
+ * \pre The log messages list is locked
+ *
+ * \param start A pointer to the logmsg to start at
+ * \param amount The number of messages to flush
+ *
+ * \retval A pointer to the last WARNING / ERROR message checked (NULL otherwise)
+ */
+static struct logmsg *flush_logger_queue_messages(struct logmsg *start, int amount)
+{
+	struct logmsg *current, *previous;
+	int flushed = 0;
+
+	previous = NULL;
+	current = start;
+
+	while (current) {
+		struct logmsg *next;
+
+		if (current->level == __LOG_WARNING || current->level == __LOG_ERROR) {
+			previous = current;
+			current = AST_DLLIST_NEXT(current, list);
+			continue;
+		}
+
+		next = AST_DLLIST_NEXT(current, list);
+		AST_DLLIST_REMOVE(&logmsgs, current, list);
+		logmsg_free(current);
+		current = next;
+		logger_queue_size--;
+		flushed++;
+
+		if (flushed >= amount) {
+			break;
+		}
+	}
+
+	return previous;
+}
+
 /*!
  * \brief send log messages to syslog and/or the console
  */
@@ -2380,20 +2442,55 @@ static void __attribute__((format(printf, 7, 0))) ast_log_full(int level, int su
 		return;
 	}
 
-	AST_LIST_LOCK(&logmsgs);
+	AST_DLLIST_LOCK(&logmsgs);
 	if (logger_queue_size >= logger_queue_limit && !close_logger_thread) {
-		logger_messages_discarded++;
 		if (!high_water_alert && !close_logger_thread) {
 			logmsg = format_log_message(__LOG_WARNING, 0, "logger", 0, "***", 0,
-				"Log queue threshold (%d) exceeded.  Discarding new messages.\n", logger_queue_limit);
-			AST_LIST_INSERT_TAIL(&logmsgs, logmsg, list);
+				"Log queue threshold (%d) exceeded.  Discarding non-warning/error messages.\n", logger_queue_limit);
+			AST_DLLIST_INSERT_TAIL(&logmsgs, logmsg, list);
 			high_water_alert = 1;
 			ast_cond_signal(&logcond);
 		}
-		AST_LIST_UNLOCK(&logmsgs);
-		return;
+
+		if (discard_all_new_msgs || (level != __LOG_WARNING && level != __LOG_ERROR)) {
+			logger_messages_discarded++;
+			AST_DLLIST_UNLOCK(&logmsgs);
+			return;
+		}
+
+		if (logger_queue_size >= logger_queue_limit + logger_queue_over_threshold_limit) {
+			int size_before = logger_queue_size;
+			struct logmsg *msg_ptr;
+
+			msg_ptr = last_logmsg_checked_for_discard ?
+				AST_DLLIST_NEXT(last_logmsg_checked_for_discard, list) : AST_DLLIST_FIRST(&logmsgs);
+
+			msg_ptr = flush_logger_queue_messages(msg_ptr, logger_queue_over_threshold_limit);
+			last_logmsg_checked_for_discard = msg_ptr ? msg_ptr : last_logmsg_checked_for_discard;
+
+			if (size_before == logger_queue_size) {
+				/* We didn't flush anything, so we have to discard the message */
+				discard_all_new_msgs = 1;
+				logger_messages_discarded++;
+				logmsg = format_log_message(__LOG_WARNING, 0, "logger", 0, "***", 0,
+					"Log queue does not have any non-WARNING/ERROR messages left to discard. "
+					"Discarding all new messages.");
+				AST_DLLIST_INSERT_TAIL(&logmsgs, logmsg, list);
+				ast_cond_signal(&logcond);
+				AST_DLLIST_UNLOCK(&logmsgs);
+				return;
+			}
+			else {
+				/* We flushed some non-WARNING/ERROR messages */
+				logmsg = format_log_message(__LOG_WARNING, 0, "logger", 0, "***", 0,
+					"Log queue WARNING / ERROR threshold (%d) exceeded. Discarded %d oldest non-"
+					"WARNING/ERROR messages.\n", logger_queue_over_threshold_limit, size_before - logger_queue_size);
+				AST_DLLIST_INSERT_TAIL(&logmsgs, logmsg, list);
+				ast_cond_signal(&logcond);
+			}
+		}
 	}
-	AST_LIST_UNLOCK(&logmsgs);
+	AST_DLLIST_UNLOCK(&logmsgs);
 
 	logmsg = format_log_message_ap(level, sublevel, file, line, function, callid, fmt, ap);
 	if (!logmsg) {
@@ -2404,16 +2501,16 @@ static void __attribute__((format(printf, 7, 0))) ast_log_full(int level, int su
 
 	/* If the logger thread is active, append it to the tail end of the list - otherwise skip that step */
 	if (logthread != AST_PTHREADT_NULL) {
-		AST_LIST_LOCK(&logmsgs);
+		AST_DLLIST_LOCK(&logmsgs);
 		if (close_logger_thread) {
 			/* Logger is either closing or closed.  We cannot log this message. */
 			logmsg_free(logmsg);
 		} else {
-			AST_LIST_INSERT_TAIL(&logmsgs, logmsg, list);
+			AST_DLLIST_INSERT_TAIL(&logmsgs, logmsg, list);
 			logger_queue_size++;
 			ast_cond_signal(&logcond);
 		}
-		AST_LIST_UNLOCK(&logmsgs);
+		AST_DLLIST_UNLOCK(&logmsgs);
 	} else {
 		logger_print_normal(logmsg);
 		logmsg_free(logmsg);
@@ -2909,6 +3006,16 @@ int ast_logger_get_queue_limit(void)
 	return logger_queue_limit;
 }
 
+void ast_logger_set_over_threshold_queue_limit(int limit)
+{
+	logger_queue_over_threshold_limit = limit;
+}
+
+int ast_logger_get_over_threshold_queue_limit(void)
+{
+	return logger_queue_over_threshold_limit;
+}
+
 static int reload_module(void)
 {
 	return reload_logger(0, NULL);

tests/test_logger.c

@@ -254,10 +254,90 @@ error:
 	return CLI_SUCCESS;
 }
 
+static char *handle_cli_queue_over_threshold_test(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
+{
+	int level;
+	int current_queue_limit, current_over_threshold_queue_limit;
+	unsigned int x, custom = 1000, warning = 200, error = 200;
+	struct timeval start, end;
+	int elapsed;
+	char tmppath[] = "/tmp/asterisk_logger_queue.XXXXXX";
+	int fd;
+
+	switch (cmd) {
+	case CLI_INIT:
+		e->command = "logger test queue over threshold";
+		e->usage = ""
+			"Usage: logger test queue over threshold\n"
+			"";
+		return NULL;
+	case CLI_GENERATE:
+		return NULL;
+	}
+
+	fd = mkstemp(tmppath);
+	if (fd < 0) {
+		ast_cli(a->fd, "Test: Failed, could not create temporary log file '%s'.\n", tmppath);
+		return CLI_SUCCESS;
+	}
+
+	level = ast_logger_register_level("queuetest");
+	if (level < 0) {
+		ast_cli(a->fd, "Test: Failed, could not register level 'queuetest'.\n");
+		return CLI_SUCCESS;
+	}
+	ast_cli(a->fd, "Test: got level %d for 'queuetest'.\n", level);
+
+	if (ast_logger_create_channel(tmppath, "queuetest") != AST_LOGGER_SUCCESS) {
+		ast_cli(a->fd, "Test: Unable to create logger channel '%s'\n", tmppath);
+		goto error;
+	}
+
+	current_queue_limit = ast_logger_get_queue_limit();
+	ast_cli(a->fd, "Test: Current queue limit: %d.  Setting to 10 for test.\n", current_queue_limit);
+	ast_logger_set_queue_limit(10);
+
+	current_over_threshold_queue_limit = ast_logger_get_over_threshold_queue_limit();
+	ast_cli(a->fd, "Test: Current over threshold queue limit: %d. Setting to 10 for test.\n", current_over_threshold_queue_limit);
+	ast_logger_set_over_threshold_queue_limit(10);
+
+	ast_cli(a->fd, "Test: You should see SOME 'exceeded', 'resumed', 'discarded', and WARNING/ERROR "
+		"messages after the test is completed. How many is dependent on system resources.\n");
+
+	start = ast_tvnow();
+	for (x = 0; x < custom; x++) {
+		ast_log_dynamic_level(level, "Performance test log message %2d\n", x);
+	}
+	for (; x < custom + warning; x++) {
+		ast_log(LOG_WARNING, "Performance test log message %2d\n", x);
+	}
+	for (; x < custom + warning + error; x++) {
+		ast_log(LOG_ERROR, "Performance test log message %2d\n", x);
+	}
+	end = ast_tvnow();
+	elapsed = ast_tvdiff_ms(end, start);
+	ast_cli(a->fd, "Test: %d messages in %f seconds.\n", x, (float) elapsed / 1000);
+	ast_cli(a->fd, "Test: Completed. Resetting queue limit to %d and over threshold limit to %d.\n",
+		current_queue_limit, current_over_threshold_queue_limit);
+	ast_logger_set_queue_limit(current_queue_limit);
+	ast_logger_set_over_threshold_queue_limit(current_over_threshold_queue_limit);
+
+error:
+
+	ast_logger_remove_channel(tmppath);
+	ast_logger_unregister_level("queuetest");
+	close(fd);
+	unlink(tmppath);
+
+	return CLI_SUCCESS;
+}
+
+
 static struct ast_cli_entry cli_logger[] = {
 	AST_CLI_DEFINE(handle_cli_dynamic_level_test, "Test the dynamic logger level implementation"),
 	AST_CLI_DEFINE(handle_cli_performance_test, "Test the logger performance"),
 	AST_CLI_DEFINE(handle_cli_queue_test, "Test the logger queue"),
+	AST_CLI_DEFINE(handle_cli_queue_over_threshold_test, "Test the logger queue over threshold for WARNING/ERRORs"),
 };
 
 static int unload_module(void)