Merge "logger: Added logger_queue_limit to the configuration options." into 13

CHANGES

@@ -22,6 +22,15 @@ res_rtp_asterisk
    for a response that can never come until we give up on the response.
    Multiple subnets may be listed.
 
+Logging
+-------------------
+ * Added logger_queue_limit to the configuration options.
+   All log messages go to a queue serviced by a single thread
+   which does all the IO.  This setting controls how big that
+   queue can get (and therefore how much memory is allocated)
+   before new messages are discarded.
+   The default is 1000.
+
 res_pjsip_config_wizard
 ------------------
  * Two new parameters have been added to the pjsip config wizard.

configs/samples/logger.conf.sample

@@ -78,6 +78,14 @@
 ; Directory for log files is configures in asterisk.conf
 ; option astlogdir
 ;
+; All log messages go to a queue serviced by a single thread
+; which does all the IO.  This setting controls how big that
+; queue can get (and therefore how much memory is allocated)
+; before new messages are discarded.
+; The default is 1000
+;logger_queue_limit = 250
+;
+;
 [logfiles]
 ;
 ; Format is "filename" and then "levels" of debugging to be included:

include/asterisk/logger.h

@@ -525,6 +525,22 @@ void ast_verb_console_set(int verb_level);
  */
 int ast_is_logger_initialized(void);
 
+/*!
+ * \brief Set the maximum number of messages allowed in the processing queue
+ *
+ * \param queue_limit
+ *
+ * \return Nothing
+ */
+void ast_logger_set_queue_limit(int queue_limit);
+
+/*!
+ * \brief Get the maximum number of messages allowed in the processing queue
+ *
+ * \return Queue limit
+ */
+int ast_logger_get_queue_limit(void);
+
 #if defined(__cplusplus) || defined(c_plusplus)
 }
 #endif

main/logger.c

@@ -86,6 +86,11 @@ static volatile int next_unique_callid; /* Used to assign unique call_ids to cal
 static int display_callids;
 static void unique_callid_cleanup(void *data);
 
+static int logger_queue_size;
+static int logger_queue_limit = 1000;
+static int logger_messages_discarded;
+static unsigned int high_water_alert;
+
 struct ast_callid {
 	int call_identifier; /* Numerical value of the call displayed in the logs */
 };
@@ -532,6 +537,16 @@ static int init_logger_chain(int locked, const char *altconf)
 			fprintf(stderr, "rotatetimestamp option has been deprecated.  Please use rotatestrategy instead.\n");
 		}
 	}
+	if ((s = ast_variable_retrieve(cfg, "general", "logger_queue_limit"))) {
+		if (sscanf(s, "%30d", &logger_queue_limit) != 1) {
+			fprintf(stderr, "logger_queue_limit has an invalid value.  Leaving at default of %d.\n",
+				logger_queue_limit);
+		}
+		if (logger_queue_limit < 10) {
+			fprintf(stderr, "logger_queue_limit must be >= 10. Setting to 10.\n");
+			logger_queue_limit = 10;
+		}
+	}
 
 	if (!locked) {
 		AST_RWLIST_WRLOCK(&logchannels);
@@ -1117,6 +1132,7 @@ static char *handle_logger_show_channels(struct ast_cli_entry *e, int cmd, struc
 	case CLI_GENERATE:
 		return NULL;
 	}
+	ast_cli(a->fd, "Logger queue limit: %d\n\n", logger_queue_limit);
 	ast_cli(a->fd, FORMATL, "Channel", "Type", "Status");
 	ast_cli(a->fd, "Configuration\n");
 	ast_cli(a->fd, FORMATL, "-------", "----", "------");
@@ -1486,6 +1502,79 @@ static void logger_print_normal(struct logmsg *logmsg)
 	return;
 }
 
+static struct logmsg * __attribute__((format(printf, 6, 0))) format_log_message_ap(int level,
+	const char *file, int line, const char *function, struct ast_callid *callid,
+	const char *fmt, va_list ap)
+{
+	struct logmsg *logmsg = NULL;
+	struct ast_str *buf = NULL;
+	struct ast_tm tm;
+	struct timeval now = ast_tvnow();
+	int res = 0;
+	char datestring[256];
+
+	if (!(buf = ast_str_thread_get(&log_buf, LOG_BUF_INIT_SIZE))) {
+		return NULL;
+	}
+
+	/* Build string */
+	res = ast_str_set_va(&buf, BUFSIZ, fmt, ap);
+
+	/* If the build failed, then abort and free this structure */
+	if (res == AST_DYNSTR_BUILD_FAILED) {
+		return NULL;
+	}
+
+	/* Create a new logging message */
+	if (!(logmsg = ast_calloc_with_stringfields(1, struct logmsg, res + 128))) {
+		return NULL;
+	}
+
+	/* Copy string over */
+	ast_string_field_set(logmsg, message, ast_str_buffer(buf));
+
+	/* Set type */
+	if (level == __LOG_VERBOSE) {
+		logmsg->type = LOGMSG_VERBOSE;
+	} else {
+		logmsg->type = LOGMSG_NORMAL;
+	}
+
+	if (display_callids && callid) {
+		logmsg->callid = ast_callid_ref(callid);
+		/* callid will be unreffed at logmsg destruction */
+	}
+
+	/* Create our date/time */
+	ast_localtime(&now, &tm, NULL);
+	ast_strftime(datestring, sizeof(datestring), dateformat, &tm);
+	ast_string_field_set(logmsg, date, datestring);
+
+	/* Copy over data */
+	logmsg->level = level;
+	logmsg->line = line;
+	ast_string_field_set(logmsg, level_name, levels[level]);
+	ast_string_field_set(logmsg, file, file);
+	ast_string_field_set(logmsg, function, function);
+	logmsg->lwp = ast_get_tid();
+
+	return logmsg;
+}
+
+static struct logmsg * __attribute__((format(printf, 6, 0))) format_log_message(int level,
+	const char *file, int line, const char *function, struct ast_callid *callid,
+	const char *fmt, ...)
+{
+	struct logmsg *logmsg;
+	va_list ap;
+
+	va_start(ap, fmt);
+	logmsg = format_log_message_ap(level, file, line, function, callid, fmt, ap);
+	va_end(ap);
+
+	return logmsg;
+}
+
 /*! \brief Actual logging thread */
 static void *logger_thread(void *data)
 {
@@ -1502,8 +1591,21 @@ static void *logger_thread(void *data)
 				ast_cond_wait(&logcond, &logmsgs.lock);
 			}
 		}
+
+		if (high_water_alert) {
+			msg = format_log_message(__LOG_WARNING, "logger", 0, "***", NULL,
+				"Logging resumed.  %d message%s discarded.\n",
+				logger_messages_discarded, logger_messages_discarded == 1 ? "" : "s");
+			if (msg) {
+				AST_LIST_INSERT_TAIL(&logmsgs, msg, list);
+			}
+			high_water_alert = 0;
+			logger_messages_discarded = 0;
+		}
+
 		next = AST_LIST_FIRST(&logmsgs);
 		AST_LIST_HEAD_INIT_NOLOCK(&logmsgs);
+		logger_queue_size = 0;
 		AST_LIST_UNLOCK(&logmsgs);
 
 		/* Otherwise go through and process each message in the order added */
@@ -1829,74 +1931,35 @@ static void unique_callid_cleanup(void *data)
 /*!
  * \brief send log messages to syslog and/or the console
  */
-static void __attribute__((format(printf, 6, 0))) ast_log_full(int level, const char *file, int line, const char *function, struct ast_callid *callid, const char *fmt, va_list ap)
+static void __attribute__((format(printf, 6, 0))) ast_log_full(int level, const char *file,
+	int line, const char *function, struct ast_callid *callid, const char *fmt, va_list ap)
 {
 	struct logmsg *logmsg = NULL;
-	struct ast_str *buf = NULL;
-	struct ast_tm tm;
-	struct timeval now = ast_tvnow();
-	int res = 0;
-	char datestring[256];
-
-	if (!(buf = ast_str_thread_get(&log_buf, LOG_BUF_INIT_SIZE)))
-		return;
-
-	if (level != __LOG_VERBOSE && AST_RWLIST_EMPTY(&logchannels)) {
-		/*
-		 * we don't have the logger chain configured yet,
-		 * so just log to stdout
-		 */
-		int result;
-		result = ast_str_set_va(&buf, BUFSIZ, fmt, ap); /* XXX BUFSIZ ? */
-		if (result != AST_DYNSTR_BUILD_FAILED) {
-			term_filter_escapes(ast_str_buffer(buf));
-			fputs(ast_str_buffer(buf), stdout);
-		}
-		return;
-	}
 
 	/* Ignore anything that never gets logged anywhere */
-	if (level != __LOG_VERBOSE && !(global_logmask & (1 << level)))
+	if (level != __LOG_VERBOSE && !(global_logmask & (1 << level))) {
 		return;
-
-	/* Build string */
-	res = ast_str_set_va(&buf, BUFSIZ, fmt, ap);
-
-	/* If the build failed, then abort and free this structure */
-	if (res == AST_DYNSTR_BUILD_FAILED)
-		return;
-
-	/* Create a new logging message */
-	if (!(logmsg = ast_calloc_with_stringfields(1, struct logmsg, res + 128)))
-		return;
-
-	/* Copy string over */
-	ast_string_field_set(logmsg, message, ast_str_buffer(buf));
-
-	/* Set type */
-	if (level == __LOG_VERBOSE) {
-		logmsg->type = LOGMSG_VERBOSE;
-	} else {
-		logmsg->type = LOGMSG_NORMAL;
 	}
 
-	if (display_callids && callid) {
+	AST_LIST_LOCK(&logmsgs);
-		logmsg->callid = ast_callid_ref(callid);
+	if (logger_queue_size >= logger_queue_limit && !close_logger_thread) {
-		/* callid will be unreffed at logmsg destruction */
+		logger_messages_discarded++;
+		if (!high_water_alert && !close_logger_thread) {
+			logmsg = format_log_message(__LOG_WARNING, "logger", 0, "***", NULL,
+				"Log queue threshold (%d) exceeded.  Discarding new messages.\n", logger_queue_limit);
+			AST_LIST_INSERT_TAIL(&logmsgs, logmsg, list);
+			high_water_alert = 1;
+			ast_cond_signal(&logcond);
+		}
+		AST_LIST_UNLOCK(&logmsgs);
+		return;
 	}
+	AST_LIST_UNLOCK(&logmsgs);
 
-	/* Create our date/time */
+	logmsg = format_log_message_ap(level, file, line, function, callid, fmt, ap);
-	ast_localtime(&now, &tm, NULL);
+	if (!logmsg) {
-	ast_strftime(datestring, sizeof(datestring), dateformat, &tm);
+		return;
-	ast_string_field_set(logmsg, date, datestring);
+	}
-
-	/* Copy over data */
-	logmsg->level = level;
-	logmsg->line = line;
-	ast_string_field_set(logmsg, level_name, levels[level]);
-	ast_string_field_set(logmsg, file, file);
-	ast_string_field_set(logmsg, function, function);
-	logmsg->lwp = ast_get_tid();
 
 	/* If the logger thread is active, append it to the tail end of the list - otherwise skip that step */
 	if (logthread != AST_PTHREADT_NULL) {
@@ -1906,6 +1969,7 @@ static void __attribute__((format(printf, 6, 0))) ast_log_full(int level, const
 			logmsg_free(logmsg);
 		} else {
 			AST_LIST_INSERT_TAIL(&logmsgs, logmsg, list);
+			logger_queue_size++;
 			ast_cond_signal(&logcond);
 		}
 		AST_LIST_UNLOCK(&logmsgs);
@@ -2375,3 +2439,12 @@ const char *ast_logger_get_dateformat(void)
 	return dateformat;
 }
 
+void ast_logger_set_queue_limit(int queue_limit)
+{
+	logger_queue_limit = queue_limit;
+}
+
+int ast_logger_get_queue_limit(void)
+{
+	return logger_queue_limit;
+}

tests/test_logger.c

@@ -190,9 +190,76 @@ static char *handle_cli_performance_test(struct ast_cli_entry *e, int cmd, struc
 	return CLI_SUCCESS;
 }
 
+static char *handle_cli_queue_test(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
+{
+	unsigned int level;
+	int current_queue_limit;
+	unsigned int x;
+	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";
+		e->usage = ""
+			"Usage: logger test queue\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 %u 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 100 for test.\n", current_queue_limit);
+	ast_logger_set_queue_limit(100);
+
+	ast_cli(a->fd, "Test: You should see SOME 'exceeded' and 'resumed' messages after the test "
+		"is completed.  How many is dependent on system resources.\n");
+
+	start = ast_tvnow();
+	for (x = 0; x < 10000; x++) {
+		ast_log_dynamic_level(level, "Performance test log message %2d\n", x);
+	}
+	end = ast_tvnow();
+	elapsed = ast_tvdiff_ms(end, start);
+	ast_cli(a->fd, "Test: 10,000 messages in %f seconds.\n", (float) elapsed / 1000);
+	ast_cli(a->fd, "Test: Completed.  Resetting queue limit to %d.\n", current_queue_limit);
+	ast_logger_set_queue_limit(current_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"),
 };
 
 static int unload_module(void)