filterx: delegate per LogPipe filterx <> message sync to filterx

Signed-off-by: Balazs Scheidler <balazs.scheidler@axoflow.com>
bazsi · May 5, 2024 · 9782a80 · 9782a80
1 parent 7db26bf
commit 9782a80
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 14 deletions.
diff --git a/lib/filterx/filterx-eval.h b/lib/filterx/filterx-eval.h
@@ -60,4 +60,26 @@ void filterx_eval_store_weak_ref(FilterXObject *object);
 void filterx_eval_init_context(FilterXEvalContext *context, FilterXEvalContext *previous_context);
 void filterx_eval_deinit_context(FilterXEvalContext *context);
 
+static inline void
+filterx_eval_sync_message(FilterXEvalContext *context, LogMessage **pmsg, const LogPathOptions *path_options)
+{
+  if (!context)
+    return;
+
+  if (!filterx_scope_is_dirty(context->scope))
+    return;
+
+  log_msg_make_writable(pmsg, path_options);
+  filterx_scope_sync(context->scope, *pmsg);
+}
+
+static inline void
+filterx_eval_prepare_for_fork(FilterXEvalContext *context, LogMessage **pmsg, const LogPathOptions *path_options)
+{
+  filterx_eval_sync_message(context, pmsg, path_options);
+  if (context)
+    filterx_scope_write_protect(context->scope);
+  log_msg_write_protect(*pmsg);
+}
+
 #endif
diff --git a/lib/logmpx.c b/lib/logmpx.c
@@ -88,13 +88,12 @@ log_multiplexer_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_op
   log_path_options_push_junction(&local_path_options, &matched, path_options);
   if (_has_multiple_arcs(self))
     {
-      if (path_options->filterx_scope)
-        {
-          log_msg_make_writable(&msg, path_options);
-          filterx_scope_sync(path_options->filterx_scope, msg);
-          filterx_scope_write_protect(path_options->filterx_scope);
-        }
-      log_msg_write_protect(msg);
+      /* if we are delivering to multiple branches, we need to sync the
+       * filterx state with our message and also need to make everything
+       * write protected so that changes in those branches don't overwrite
+       * data we still need */
+
+      filterx_eval_prepare_for_fork(path_options->filterx_context, &msg, path_options);
     }
   for (fallback = 0; (fallback == 0) || (fallback == 1 && self->fallback_exists && !delivered); fallback++)
     {

diff --git a/lib/logpipe.h b/lib/logpipe.h
@@ -458,13 +458,8 @@ log_pipe_queue(LogPipe *s, LogMessage *msg, const LogPathOptions *path_options)
         }
     }
 
-  if ((s->flags & PIF_SYNC_SCOPE) &&
-      path_options->filterx_scope &&
-      filterx_scope_is_dirty(path_options->filterx_scope))
-    {
-      log_msg_make_writable(&msg, path_options);
-      filterx_scope_sync(path_options->filterx_scope, msg);
-    }
+  if ((s->flags & PIF_SYNC_SCOPE))
+    filterx_eval_sync_message(path_options->filterx_context, &msg, path_options);
 
   if (G_UNLIKELY(s->flags & (PIF_HARD_FLOW_CONTROL | PIF_JUNCTION_END | PIF_CONDITIONAL_MIDPOINT)))
     {