Centralized internal logging

lib/filter/filter-call.c

@@ -51,8 +51,7 @@ filter_call_eval(FilterExprNode *s, LogMessage **msgs, gint num_msg, LogTemplate
     stats_counter_inc(self->super.not_matched);
 
   msg_trace("filter() evaluation started",
-            evt_tag_str("called-rule", self->rule),
-            evt_tag_msg_reference(msgs[num_msg - 1]));
+            evt_tag_str("called-rule", self->rule));
 
   return res ^ s->comp;
 }

lib/filter/filter-cmp.c

@@ -302,8 +302,7 @@ fop_cmp_eval(FilterExprNode *s, LogMessage **msgs, gint num_msg, LogTemplateEval
             evt_tag_str("compare_mode", _compare_mode_to_string(self->compare_mode)),
             evt_tag_str("left_type", log_msg_value_type_to_str(left_type)),
             evt_tag_str("right_type", log_msg_value_type_to_str(right_type)),
-            evt_tag_int("result", result),
-            evt_tag_msg_reference(msgs[num_msg - 1]));
+            evt_tag_int("result", result));
 
   scratch_buffers_reclaim_marked(marker);
   return result ^ s->comp;

lib/filter/filter-in-list.c

@@ -51,8 +51,7 @@ filter_in_list_eval(FilterExprNode *s, LogMessage **msgs, gint num_msg, LogTempl
 
   gboolean result = (g_tree_lookup(self->tree, value) != NULL);
   msg_trace("in-list() evaluation started",
-            evt_tag_str("value", value),
-            evt_tag_msg_reference(msg));
+            evt_tag_str("value", value));
 
   return result ^ s->comp;
 }

lib/filter/filter-netmask.c

@@ -66,8 +66,7 @@ filter_netmask_eval(FilterExprNode *s, LogMessage **msgs, gint num_msg, LogTempl
   msg_trace("netmask() evaluation started",
             evt_tag_inaddr("msg_address", addr),
             evt_tag_inaddr("address", &self->address),
-            evt_tag_inaddr("netmask", &self->netmask),
-            evt_tag_msg_reference(msg));
+            evt_tag_inaddr("netmask", &self->netmask));
   return res ^ s->comp;
 }
 

lib/filter/filter-netmask6.c

@@ -131,8 +131,7 @@ _eval(FilterExprNode *s, LogMessage **msgs, gint num_msg, LogTemplateEvalOptions
   msg_trace("netmask6() evaluation started",
             evt_tag_inaddr6("msg_address", address),
             evt_tag_inaddr6("address", &self->address),
-            evt_tag_int("prefix", self->prefix),
-            evt_tag_msg_reference(msg));
+            evt_tag_int("prefix", self->prefix));
   return result ^ s->comp;
 }
 

lib/filter/filter-pipe.c

@@ -61,16 +61,14 @@ log_filter_pipe_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_op
 
   msg_trace(">>>>>> filter rule evaluation begin",
             evt_tag_str("rule", self->name),
-            log_pipe_location_tag(s),
-            evt_tag_msg_reference(msg));
+            log_pipe_location_tag(s));
 
   res = filter_expr_eval_root(self->expr, &msg, path_options);
 
   msg_trace("<<<<<< filter rule evaluation result",
             evt_tag_str("result", res ? "matched" : "unmatched"),
             evt_tag_str("rule", self->name),
-            log_pipe_location_tag(s),
-            evt_tag_msg_reference(msg));
+            log_pipe_location_tag(s));
 
   if (res)
     {

lib/filter/filter-pri.c

@@ -51,8 +51,7 @@ filter_facility_eval(FilterExprNode *s, LogMessage **msgs, gint num_msg, LogTemp
     }
   msg_trace("facility() evaluation started",
             evt_tag_int("fac", fac_num),
-            evt_tag_printf("valid_fac", "%08x", self->valid),
-            evt_tag_msg_reference(msg));
+            evt_tag_printf("valid_fac", "%08x", self->valid));
 
   return res ^ s->comp;
 }
@@ -81,8 +80,7 @@ filter_severity_eval(FilterExprNode *s, LogMessage **msgs, gint num_msg, LogTemp
 
   msg_trace("severity() evaluation started",
             evt_tag_int("pri", pri),
-            evt_tag_printf("valid_pri", "%08x", self->valid),
-            evt_tag_msg_reference(msg));
+            evt_tag_printf("valid_pri", "%08x", self->valid));
 
   return res ^ s->comp;
 }

lib/filter/filter-re.c

@@ -46,8 +46,7 @@ filter_re_eval(FilterExprNode *s, LogMessage **msgs, gint num_msg, LogTemplateEv
   msg_trace("match() evaluation started against a name-value pair",
             evt_tag_msg_value_name("name", self->value_handle),
             evt_tag_msg_value("value", msg, self->value_handle),
-            evt_tag_str("pattern", self->matcher->pattern),
-            evt_tag_msg_reference(msg));
+            evt_tag_str("pattern", self->matcher->pattern));
   result = log_matcher_match_value(self->matcher, msg, self->value_handle);
   return result ^ s->comp;
 }
@@ -180,8 +179,7 @@ filter_match_eval_against_program_pid_msg(FilterExprNode *s, LogMessage **msgs,
 
   msg_trace("match() evaluation started against constructed $PROGRAM[$PID]: $MESSAGE string for compatibility",
             evt_tag_printf("input", "%s", str),
-            evt_tag_str("pattern", self->super.matcher->pattern),
-            evt_tag_msg_reference(msg));
+            evt_tag_str("pattern", self->super.matcher->pattern));
 
   result = log_matcher_match_buffer(self->super.matcher, msg, str, -1);
 
@@ -198,8 +196,7 @@ filter_match_eval_against_template(FilterExprNode *s, LogMessage **msgs, gint nu
   msg_trace("match() evaluation started against template",
             evt_tag_template("input", self->template, msg, options),
             evt_tag_str("pattern", self->super.matcher->pattern),
-            evt_tag_str("template", self->template->template_str),
-            evt_tag_msg_reference(msg));
+            evt_tag_str("template", self->template->template_str));
   gboolean result = log_matcher_match_template(self->super.matcher, msg, self->template, options);
   return result ^ s->comp;
 }

lib/filter/filter-tags.c

@@ -46,8 +46,7 @@ filter_tags_eval(FilterExprNode *s, LogMessage **msgs, gint num_msg, LogTemplate
       if (log_msg_is_tag_by_id(msg, tag_id))
         {
           msg_trace("tags() evaluation result, matching tag is found",
-                    evt_tag_str("tag", log_tags_get_by_id(tag_id)),
-                    evt_tag_msg_reference(msg));
+                    evt_tag_str("tag", log_tags_get_by_id(tag_id)));
 
           res = TRUE;
           return res ^ s->comp;
@@ -56,13 +55,11 @@ filter_tags_eval(FilterExprNode *s, LogMessage **msgs, gint num_msg, LogTemplate
         {
           msg_trace("tags() evaluation progress, tag is not set",
                     evt_tag_str("tag", log_tags_get_by_id(tag_id)),
-                    evt_tag_int("value", log_msg_is_tag_by_id(msg, tag_id)),
-                    evt_tag_msg_reference(msg));
+                    evt_tag_int("value", log_msg_is_tag_by_id(msg, tag_id)));
         }
     }
 
-  msg_trace("tags() evaluation result, none of the tags is present",
-            evt_tag_msg_reference(msg));
+  msg_trace("tags() evaluation result, none of the tags is present");
   res = FALSE;
   return res ^ s->comp;
 }

lib/filterx/filterx-pipe.c

@@ -61,8 +61,7 @@ log_filterx_pipe_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_o
 
   msg_trace(">>>>>> filterx rule evaluation begin",
             evt_tag_str("rule", self->name),
-            log_pipe_location_tag(s),
-            evt_tag_msg_reference(msg));
+            log_pipe_location_tag(s));
 
   NVTable *payload = nv_table_ref(msg->payload);
   eval_res = filterx_eval_exec(&eval_context, self->block, msg);
@@ -71,8 +70,7 @@ log_filterx_pipe_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_o
             filterx_format_eval_result(eval_res),
             evt_tag_str("rule", self->name),
             log_pipe_location_tag(s),
-            evt_tag_int("dirty", filterx_scope_is_dirty(eval_context.scope)),
-            evt_tag_msg_reference(msg));
+            evt_tag_int("dirty", filterx_scope_is_dirty(eval_context.scope)));
 
   local_path_options.filterx_context = &eval_context;
   switch (eval_res)

lib/logmsg/logmsg.c

@@ -633,8 +633,7 @@ log_msg_set_value_with_type(LogMessage *self, NVHandle handle,
       msg_trace("Setting value",
                 evt_tag_str("name", name),
                 evt_tag_mem("value", value, value_len),
-                evt_tag_str("type", log_msg_value_type_to_str(type)),
-                evt_tag_msg_reference(self));
+                evt_tag_str("type", log_msg_value_type_to_str(type)));
     }
 
   if (!log_msg_chk_flag(self, LF_STATE_OWN_PAYLOAD))
@@ -689,8 +688,7 @@ log_msg_unset_value(LogMessage *self, NVHandle handle)
   if (_log_name_value_updates(self))
     {
       msg_trace("Unsetting value",
-                evt_tag_str("name", log_msg_get_value_name(handle, NULL)),
-                evt_tag_msg_reference(self));
+                evt_tag_str("name", log_msg_get_value_name(handle, NULL)));
     }
 
   if (!log_msg_chk_flag(self, LF_STATE_OWN_PAYLOAD))
@@ -754,8 +752,7 @@ log_msg_set_value_indirect_with_type(LogMessage *self, NVHandle handle,
                 evt_tag_str("type", log_msg_value_type_to_str(type)),
                 evt_tag_int("ref_handle", ref_handle),
                 evt_tag_int("ofs", ofs),
-                evt_tag_int("len", len),
-                evt_tag_msg_reference(self));
+                evt_tag_int("len", len));
     }
 
   if (!log_msg_chk_flag(self, LF_STATE_OWN_PAYLOAD))
@@ -788,6 +785,7 @@ log_msg_set_value_indirect_with_type(LogMessage *self, NVHandle handle,
   if (new_entry)
     log_msg_update_sdata(self, handle, name, name_len);
   log_msg_update_num_matches(self, handle);
+
 }
 
 void
@@ -1510,8 +1508,7 @@ log_msg_clone_cow(LogMessage *msg, const LogPathOptions *path_options)
   msg->allocated_bytes = allocated_bytes;
 
   msg_trace("Message was cloned",
-            evt_tag_printf("original_msg", "%p", msg),
-            evt_tag_msg_reference(self));
+            evt_tag_printf("original_msg", "%p", msg));
 
   /* every field _must_ be initialized explicitly if its direct
    * copying would cause problems (like copying a pointer by value) */
@@ -1537,6 +1534,7 @@ log_msg_clone_cow(LogMessage *msg, const LogPathOptions *path_options)
 
   if (self->num_tags == 0)
     self->flags |= LF_STATE_OWN_TAGS;
+
   return self;
 }
 

lib/logmsg/logmsg.h

@@ -552,17 +552,12 @@ void log_msg_registry_init(void);
 void log_msg_registry_deinit(void);
 void log_msg_global_init(void);
 void log_msg_global_deinit(void);
-void log_msg_stats_global_init(void);
 void log_msg_registry_foreach(GHFunc func, gpointer user_data);
 
 gint log_msg_lookup_time_stamp_name(const gchar *name);
 
 gssize log_msg_get_size(LogMessage *self);
 
-#define evt_tag_msg_reference(msg)             \
-    evt_tag_printf("msg", "%p", (msg)),        \
-    evt_tag_printf("rcptid", "%" G_GUINT64_FORMAT, (msg)->rcptid)
-
 static inline EVTTAG *
 evt_tag_msg_value(const gchar *name, LogMessage *msg, NVHandle value_handle)
 {

lib/logqueue-fifo.c

@@ -489,6 +489,7 @@ log_queue_fifo_pop_head(LogQueue *s, LogPathOptions *path_options)
   if (!node->flow_control_requested)
     self->backlog_queue.non_flow_controlled_len++;
 
+  msg_set_context(msg);
   return msg;
 }
 

lib/logreader.c

@@ -459,9 +459,9 @@ log_reader_handle_line(LogReader *self, const guchar *line, gint length, LogTran
   LogMessage *m;
 
   m = msg_format_construct_message(&self->options->parse_options, line, length);
+  msg_set_context(m);
   msg_debug("Incoming log entry",
-            evt_tag_mem("input", line, length),
-            evt_tag_msg_reference(m));
+            evt_tag_mem("input", line, length));
 
   msg_format_parse_into(&self->options->parse_options, m, line, length);
 
@@ -487,6 +487,7 @@ log_reader_handle_line(LogReader *self, const guchar *line, gint length, LogTran
   log_transport_aux_data_foreach(aux, _add_aux_nvpair, m);
 
   log_source_post(&self->super, m);
+  msg_set_context(NULL);
   log_msg_refcache_stop();
   return log_source_free_to_send(&self->super);
 }

lib/logscheduler.c

@@ -87,6 +87,7 @@ _work(gpointer s, gpointer arg)
           iv_list_del(&node->list);
 
           LogMessage *msg = log_msg_ref(node->msg);
+          msg_set_context(msg);
 
           LogPathOptions path_options = LOG_PATH_OPTIONS_INIT;
           path_options.ack_needed = node->ack_needed;
@@ -96,6 +97,7 @@ _work(gpointer s, gpointer arg)
 
           log_msg_refcache_start_consumer(msg, &path_options);
           _reinject_message(partition->front_pipe, msg, &path_options);
+          msg_set_context(NULL);
           log_msg_unref(msg);
           log_msg_refcache_stop();
         }

lib/logsource.c

@@ -633,11 +633,8 @@ log_source_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_options
   LogSource *self = (LogSource *) s;
   gint i;
 
-  msg_set_context(msg);
-
   msg_diagnostics(">>>>>> Source side message processing begin",
-                  log_pipe_location_tag(s),
-                  evt_tag_msg_reference(msg));
+                  log_pipe_location_tag(s));
 
   /* $HOST setup */
   log_source_mangle_hostname(self, msg);
@@ -686,8 +683,6 @@ log_source_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_options
                   log_pipe_location_tag(s),
                   evt_tag_printf("msg", "%p", msg),
                   evt_tag_printf("rcptid", "%" G_GUINT64_FORMAT, rcptid));
-
-  msg_set_context(NULL);
 }
 
 static void

lib/logthrdest/logthrdestdrv.c

@@ -484,7 +484,6 @@ _perform_inserts(LogThreadedDestWorker *self)
           break;
         }
 
-      msg_set_context(msg);
       log_msg_refcache_start_consumer(msg, &path_options);
 
       self->batch_size++;
@@ -496,7 +495,6 @@ _perform_inserts(LogThreadedDestWorker *self)
         _perform_flush(self);
 
       log_msg_unref(msg);
-      msg_set_context(NULL);
       log_msg_refcache_stop();
 
 flush_error:

lib/logthrsource/logthrsourcedrv.c

@@ -411,8 +411,7 @@ log_threaded_source_worker_post(LogThreadedSourceWorker *self, LogMessage *msg)
   msg_debug("Incoming log message",
             evt_tag_str("input", log_msg_get_value(msg, LM_V_MESSAGE, NULL)),
             evt_tag_str("driver", self->control->super.super.id),
-            evt_tag_int("worker_index", self->worker_index),
-            evt_tag_msg_reference(msg));
+            evt_tag_int("worker_index", self->worker_index));
   _apply_message_attributes(self->control, msg);
   log_source_post(&self->super, msg);
 

lib/logwriter.c

@@ -1218,7 +1218,6 @@ log_writer_write_message(LogWriter *self, LogMessage *msg, LogPathOptions *path_
   *write_error = FALSE;
 
   log_msg_refcache_start_consumer(msg, path_options);
-  msg_set_context(msg);
 
   log_writer_format_log(self, msg, self->line_buffer);
 

lib/messages.c

@@ -50,6 +50,7 @@ typedef struct _MsgContext
   guint16 recurse_state;
   guint recurse_warning:1;
   gchar recurse_trigger[128];
+  guint64 original_msg_rcptid;
 } MsgContext;
 
 static gint active_log_level = -1;
@@ -84,6 +85,8 @@ msg_set_context(LogMessage *msg)
 {
   MsgContext *context = msg_get_context();
 
+  context->original_msg_rcptid = msg ? msg->rcptid : 0;
+
   if (msg && (msg->flags & LF_INTERNAL))
     {
       if (msg->recursed)
@@ -240,6 +243,12 @@ msg_event_create(gint prio, const gchar *desc, EVTTAG *tag1, ...)
       evt_rec_add_tagsv(e, va);
       va_end(va);
     }
+  MsgContext *msg_context = msg_get_context();
+  if (msg_context->original_msg_rcptid != 0)
+    {
+      EVTTAG *rcptid_tag = evt_tag_printf("rcptid", "%" G_GUINT64_FORMAT, msg_context->original_msg_rcptid);
+      evt_rec_add_tag(e, rcptid_tag);
+    }
   g_mutex_unlock(&evtlog_lock);
   return e;
 }

lib/parser/parser-expr.c

@@ -93,16 +93,14 @@ log_parser_queue_method(LogPipe *s, LogMessage *msg, const LogPathOptions *path_
 
   msg_trace(">>>>>> parser rule evaluation begin",
             evt_tag_str("rule", self->name),
-            log_pipe_location_tag(s),
-            evt_tag_msg_reference(msg));
+            log_pipe_location_tag(s));
 
   success = log_parser_process_message(self, &msg, path_options);
 
   msg_trace("<<<<<< parser rule evaluation result",
             evt_tag_str("result", success ? "accepted" : "rejected"),
             evt_tag_str("rule", self->name),
-            log_pipe_location_tag(s),
-            evt_tag_msg_reference(msg));
+            log_pipe_location_tag(s));
 
   if (success)
     {