Add a new core event triggered for every log message

Author: rvlad-patrascu

cfg.lex

@@ -212,6 +212,7 @@ LOGSTDOUT	log_stdout
 LOGSTDERROR	log_stderror
 STDERROR_ENABLED	stderror_enabled
 SYSLOG_ENABLED	    syslog_enabled
+LOG_EVENT_ENABLED	log_event_enabled
 STDERROR_LEVEL_FILTER	stderror_level_filter
 SYSLOG_LEVEL_FILTER	    syslog_level_filter
 STDERROR_FORMAT stderror_log_format
@@ -431,6 +432,7 @@ SPACE		[ ]
 <INITIAL>{LOGSTDERROR} { yylval.strval=yytext; return LOGSTDERROR; }
 <INITIAL>{STDERROR_ENABLED}	{ yylval.strval=yytext; return STDERROR_ENABLED; }
 <INITIAL>{SYSLOG_ENABLED}	{ yylval.strval=yytext; return SYSLOG_ENABLED; }
+<INITIAL>{LOG_EVENT_ENABLED}	{ yylval.strval=yytext; return LOG_EVENT_ENABLED; }
 <INITIAL>{STDERROR_LEVEL_FILTER}	{ count(); yylval.strval=yytext; return STDERROR_LEVEL_FILTER; }
 <INITIAL>{SYSLOG_LEVEL_FILTER}	{ count(); yylval.strval=yytext; return SYSLOG_LEVEL_FILTER; }
 <INITIAL>{STDERROR_FORMAT} { count(); yylval.strval=yytext; return STDERROR_FORMAT; }

cfg.y

@@ -291,6 +291,7 @@ extern int cfg_parse_only_routes;
 %token LOGSTDERROR
 %token STDERROR_ENABLED
 %token SYSLOG_ENABLED
+%token LOG_EVENT_ENABLED
 %token STDERROR_LEVEL_FILTER
 %token SYSLOG_LEVEL_FILTER
 %token STDERROR_FORMAT
@@ -919,6 +920,16 @@ assign_stm: LOGLEVEL EQUAL snumber { IFOR();
 			}
 			}
 		| SYSLOG_ENABLED EQUAL error { yyerror("boolean value expected"); }
+		| LOG_EVENT_ENABLED EQUAL NUMBER {
+			IFOR();
+			if ($3) {
+				if (init_log_msg_buf(0) < 0) {
+					yyerror("failed to allocate msg log buffer");
+					YYABORT;
+				}
+			}
+			log_event_enabled=$3; }
+		| LOG_EVENT_ENABLED EQUAL error { yyerror("boolean value expected"); }
 		| STDERROR_LEVEL_FILTER EQUAL snumber {
 			IFOR();
 			s_tmp.s=STDERR_CONSUMER_NAME;

dprint.c

@@ -86,6 +86,10 @@ static struct log_consumer_t default_log_consumers[2] ={
 struct log_consumer_t *log_consumers = default_log_consumers;
 int log_consumers_no = 2;
 
+int log_event_enabled = 0;
+static str evi_log_name = str_init("E_CORE_LOG");
+static event_id_t evi_log_id;
+
 static char* str_fac[]={"LOG_AUTH","LOG_CRON","LOG_DAEMON",
 					"LOG_KERN","LOG_LOCAL0","LOG_LOCAL1",
 					"LOG_LOCAL2","LOG_LOCAL3","LOG_LOCAL4","LOG_LOCAL5",
@@ -456,6 +460,110 @@ static void syslog_dprint(int log_level, int facility, char *module, const char
 	}
 }
 
+static str evi_time_str = str_init("time");
+static str evi_pid_str = str_init("pid");
+static str evi_level_str = str_init("level");
+static str evi_module_str = str_init("module");
+static str evi_func_str = str_init("function");
+static str evi_prefix_str = str_init("prefix");
+static str evi_msg_str = str_init("message");
+
+static void event_dprint(int log_level, int facility, char *module, const char *func,
+	char *format, va_list ap)
+{
+	evi_params_p list = NULL;
+	str s;
+	int n;
+	static char in_progress = 0;
+
+	/* prevent reentry from the same process */
+	if (in_progress)
+		return;
+
+	in_progress = 1;
+
+	if (!evi_probe_event(evi_log_id)) {
+		in_progress = 0;
+		return;
+	}
+
+	if (!(list = evi_get_params())) {
+		in_progress = 0;
+		return;
+	}
+
+	init_str(&s, dp_time());
+	if (evi_param_add_str(list, &evi_time_str, &s)) {
+		stderr_dprint_tmp("error: unable to add event parameter\n");
+		goto end_free;
+	}
+	n = dp_my_pid();
+	if (evi_param_add_int(list, &evi_pid_str, &n)) {
+		stderr_dprint_tmp("error: unable to add event parameter\n");
+		goto end_free;
+	}
+	init_str(&s, dp_log_level_str(log_level));
+	if (evi_param_add_str(list, &evi_level_str, &s)) {
+		stderr_dprint_tmp("error: unable to add event parameter\n");
+		goto end_free;
+	}
+
+	if (module && func) {
+		init_str(&s, module);
+		if (evi_param_add_str(list, &evi_module_str, &s)) {
+			stderr_dprint_tmp("error: unable to add event parameter\n");
+			goto end_free;
+		}
+		init_str(&s, func);
+		if (evi_param_add_str(list, &evi_func_str, &s)) {
+			stderr_dprint_tmp("error: unable to add event parameter\n");
+			goto end_free;
+		}
+	}
+
+	init_str(&s, log_prefix);
+	if (s.len) {
+		if (evi_param_add_str(list, &evi_prefix_str, &s)) {
+			stderr_dprint_tmp("error: unable to add event parameter\n");
+			goto end_free;
+		}
+	}
+
+	s.len = vsnprintf(log_msg_buf, log_msg_buf_size, format, ap);
+	if (s.len < 0) {
+		stderr_dprint_tmp("error: vsnprintf() failed!\n");
+		goto end_free;
+	}
+	if (s.len>=log_msg_buf_size) {
+		stderr_dprint_tmp("warning: log message truncated\n");
+		s.len = log_msg_buf_size;
+	}
+
+	/* try to strip \n from the end of the "message" param */
+	if (log_msg_buf[s.len-1] == '\n') {
+		log_msg_buf[s.len-1] = '\0';
+		s.len--;
+	}
+
+	s.s = log_msg_buf;
+	if (evi_param_add_str(list, &evi_msg_str, &s)) {
+		stderr_dprint_tmp("error: unable to add event parameter\n");
+		goto end_free;
+	}
+
+	if (evi_raise_event(evi_log_id, list)) {
+		stderr_dprint_tmp("error: unable to raise '%.*s' event\n",
+			evi_log_name.len, evi_log_name.s);
+	}
+
+	in_progress = 0;
+
+	return;
+end_free:
+	evi_free_params(list);
+	in_progress = 0;
+}
+
 /* generic consumer that registers to the log interface */
 static void gen_consumer_pre_fmt_func(log_print_f gen_print_func, int log_level,
 	int facility, char *module, const char *func,
@@ -587,6 +695,29 @@ int init_log_cons_shm_table(void)
 	return 0;
 }
 
+int init_log_event_cons(void)
+{
+	evi_log_id = evi_publish_event(evi_log_name);
+	if (evi_log_id == EVI_ERROR) {
+		LM_ERR("cannot register '%.*s' event\n",
+			evi_log_name.len, evi_log_name.s);
+		return -1;
+	}
+
+	log_msg_buf = pkg_malloc(log_msg_buf_size+1);
+	if (!log_msg_buf) {
+		LM_ERR("no pkg memory left\n");
+		return -1;
+	}
+
+	if (register_log_consumer(EVENT_CONSUMER_NAME, event_dprint, 0, 1) < 0) {
+		LM_ERR("Failed to register 'event' log consumer\n");
+		return -1;
+	}
+
+	return 0;
+}
+
 static struct log_consumer_t *get_log_consumer_by_name(str *name)
 {
 	int i;
@@ -658,6 +789,22 @@ int get_log_consumer_mute_state(str *name, int *state)
 	return 0;
 }
 
+int set_log_event_cons_cfg_state(void)
+{
+	if (set_log_consumer_mute_state(&str_init(EVENT_CONSUMER_NAME),
+		!log_event_enabled) < 0) {
+		LM_ERR("Failed to set mute state for event consumer\n");
+		return -1;
+	}
+
+	return 0;
+}
+
+void distroy_log_event_cons(void)
+{
+	set_log_consumer_mute_state(&str_init(EVENT_CONSUMER_NAME), 1);
+}
+
 int init_log_level(void)
 {
 	log_level = &pt[process_no].log_level;

dprint.h

@@ -105,10 +105,11 @@
 	#undef NO_LOG
 #endif
 
-#define MAX_LOG_CONS_NO 2
+#define MAX_LOG_CONS_NO 3
 
 #define STDERR_CONSUMER_NAME "stderror"
 #define SYSLOG_CONSUMER_NAME "syslog"
+#define EVENT_CONSUMER_NAME  "event"
 
 #define LOG_PLAIN_NAME    "plain_text"
 #define LOG_JSON_NAME     "json"
@@ -126,6 +127,7 @@ extern int *log_level;
 extern char *log_prefix;
 extern int log_stdout;
 extern int stderr_enabled, syslog_enabled;
+extern int log_event_enabled;
 extern int log_facility;
 extern char* log_name;
 extern char ctime_buf[];
@@ -147,14 +149,17 @@ int init_log_cons_shm_table(void);
 int init_log_json_buf(int realloc);
 int init_log_msg_buf(int realloc);
 
+int init_log_event_cons();
+int set_log_event_cons_cfg_state(void);
+void distroy_log_event_cons();
+
 int set_log_consumer_mute_state(str *name, int state);
 int get_log_consumer_mute_state(str *name, int *state);
 
 int set_log_consumer_level_filter(str *name, int level);
 int get_log_consumer_level_filter(str *name, int *level_filter);
 
 int parse_log_format(str *format);
-
 int dp_my_pid(void);
 
 void stderr_dprint_tmp(char *format, ...);

ipc.c

@@ -289,8 +289,12 @@ void ipc_handle_job(int fd)
 		return;
 	}
 
+	/* we shouldn't print any logs, in case we are handling
+	 * an event_route IPC job for the E_CORE_LOG event */
+	/*
 	LM_DBG("received job type %d[%s] from process %d\n",
 		job.handler_type, ipc_handlers[job.handler_type].name, job.snd_proc);
+	*/
 
 	/* custom handling for RPC type */
 	if (job.handler_type==ipc_rpc_type) {

main.c

@@ -302,6 +302,11 @@ static int main_loop(void)
 		last_check = get_uticks();
 	}
 
+	if (set_log_event_cons_cfg_state() < 0) {
+		LM_ERR("Failed to set the config state for event consumer\n");
+		goto error;
+	}
+
 	if (_ProfilerStart(0, "attendant") != 0) {
 		LM_ERR("failed to start profiling\n");
 		goto error;
@@ -936,6 +941,11 @@ int main(int argc, char** argv)
 		goto error;
 	}
 
+	if (init_log_event_cons() < 0) {
+		LM_ERR("Failed to initialize log event consumer\n");
+		goto error;
+	}
+
 	if (trans_init_all_listeners()<0) {
 		LM_ERR("failed to init all SIP listeners, aborting\n");
 		goto error;

modules/event_kafka/kafka_producer.c

@@ -632,14 +632,20 @@ static int handle_io(struct fd_map *fm, int idx, int event_type)
 
 	switch (fm->type) {
 	case F_KAFKA_JOB:
+		/* supppress any logs, in case we are handling the E_CORE_LOG event */
+		set_proc_log_level(L_ALERT-1);
+
 		job = kafka_receive_job();
 		if (!job) {
 			LM_ERR("Cannot receive job\n");
+			reset_proc_log_level();
 			return 0;
 		}
 
 		if (kafka_handle_job(job) != 0)
 			shm_free(job);
+
+		reset_proc_log_level();
 		break;
 	case F_KAFKA_EVENT:
 		prod = (kafka_producer_t *)fm->data;

modules/event_rabbitmq/rabbitmq_send.c

@@ -541,6 +541,9 @@ void rmq_process(int rank)
 	rmq_init_reader();
 	rmq_send_t * rmqs;
 
+	/* supppress any logs, in case we are handling the E_CORE_LOG event */
+	set_proc_log_level(L_ALERT-1);
+
 	/* waiting for commands */
 	for (;;) {
 		rmqs = rmq_receive();
@@ -576,4 +579,6 @@ void rmq_process(int rank)
 		if (rmqs)
 			shm_free(rmqs);
 	}
+
+	reset_proc_log_level();
 }

modules/event_route/route_send.c

@@ -125,6 +125,9 @@ static void route_received(int sender, void *param)
 	struct sip_msg* req;
 	route_send_t *route_s = (route_send_t *)param;
 
+	/* supppress any logs, in case we are handling the E_CORE_LOG event */
+	set_proc_log_level(L_ALERT-1);
+
 	if (!ref_script_route_check_and_update(route_s->ev_route)){
 		LM_ERR("event route [%.s] no longer available in script\n",
 			route_s->ev_route->name.s);
@@ -149,6 +152,8 @@ static void route_received(int sender, void *param)
 	if (route_s->ev_route)
 		shm_free(route_s->ev_route);
 	shm_free(route_s);
+
+	reset_proc_log_level();
 }
 
 

modules/event_stream/stream_send.c

@@ -661,13 +661,19 @@ static int handle_io(struct fd_map *fm, int idx, int event_type)
 
 	switch (fm->type) {
 		case F_EV_JSONRPC_CMD:
+			/* supppress any logs, in case we are handling the E_CORE_LOG event */
+			set_proc_log_level(L_ALERT-1);
+
 			jsonrpcs = stream_receive();
 			if (!jsonrpcs) {
 				LM_ERR("invalid receive jsonrpc command\n");
+				reset_proc_log_level();
 				return -1;
 			}
 
 			handle_new_stream(jsonrpcs);
+
+			reset_proc_log_level();
 			break;
 		case F_EV_JSONRPC_RPL:
 			con = (struct stream_con *)fm->data;