logmngr: use separate fluent-bit instances for main and user conf

So the philosophy here is that there will be one main fluent-bit
instance, which will read from /dev/kmsg and /dev/log and write to
/var/log/messages.
This is done to reduce the chance of introducing errors in config
and making sure that var/log/messages is always populated because
logread depends on it.
For example, an unavailable URL for a udp based syslog output plugin
can cause fluent-bit to exit entirely.
The other fluent-bit instance uses a different "user" config file.
Which will handle log_file, log_remote and included configs. It has
a single hard coded input tail plugin which reads from
/var/log/messages.
Any filter that is applied will act on the main config and thus,
only filtered logs will be available to both fluent-bit instances.

logmngr/Makefile

@@ -74,6 +74,7 @@ ifeq ($(CONFIG_LOGMNGR_BACKEND_FLUENTBIT),y)
 	$(INSTALL_BIN) ./files/logread $(1)/sbin/
 	$(INSTALL_DATA) ./files/lib/logmngr/fluent-bit.sh $(1)/lib/logmngr/
 	$(INSTALL_BIN) ./files/etc/hotplug.d/ntp/20-reload_fluent_bit $(1)/etc/hotplug.d/ntp/
+	$(INSTALL_DATA) ./files/etc/uci-defaults/20-add-parser $(1)/etc/uci-defaults/
 else ifeq ($(CONFIG_LOGMNGR_BACKEND_SYSLOG_NG),y)
 	$(INSTALL_DATA) ./files/lib/logmngr/syslog-ng.sh $(1)/lib/logmngr/
 endif

logmngr/files/etc/uci-defaults/20-add-parser

@@ -0,0 +1,27 @@
+#!/bin/sh
+
+# add a parser to extract message from /var/log/messages file
+# this is needed because tail plugin treats the entire line as the one string without this
+PARSER_FILE="/etc/fluent-bit/parsers.conf"
+PARSER_NAME="syslog_message"
+
+# Check if parser already exists
+if grep -q "Name\s\+$PARSER_NAME" "$PARSER_FILE"; then
+    echo "Fluent Bit parser '$PARSER_NAME' already exists. Skipping."
+    exit 0
+fi
+
+# Append the parser to the file
+cat << EOF >> "$PARSER_FILE"
+
+[PARSER]
+    Name        $PARSER_NAME
+    Format      regex
+    Regex       ^(?<timestamp>\w+\s+\d+\s+\d+:\d+:\d+)\s+(?<hostname>\S+)\s+(?<process>[^:]+):\s+(?<message>.+)
+    Time_Key    timestamp
+    Time_Format %b %d %H:%M:%S
+EOF
+
+echo "Added Fluent Bit parser '$PARSER_NAME' to $PARSER_FILE"
+
+exit 0

logmngr/files/lib/logmngr/fluent-bit.sh

@@ -3,8 +3,14 @@
 . /lib/functions.sh
 . /lib/logmngr/logrotate.sh
 
+PROG=/usr/sbin/fluent-bit
 CONF_FILE=/etc/fluent-bit/fluent-bit.conf
-TMP_CONF_FILE=/tmp/fluent-bit/fluent-bit.conf
+FLUENT_BIT_TMP_DIR=/tmp/fluent-bit
+MAIN_CONF="${FLUENT_BIT_TMP_DIR}/main.fluent-bit.conf"
+USER_CONF="${FLUENT_BIT_TMP_DIR}/user.fluent-bit.conf"
+# in future if USER_TAG has to be changed
+# we will need to check that no file in FLUENT_BIT_CONF_DIR uses it
+USER_TAG="user_logs"
 FLUENT_BIT_CONF_DIR=/etc/fluent-bit/conf.d
 PROCESSED_SYSLOG_TAGS=""
 PROCESSED_KMSG_TAGS=""
@@ -38,35 +44,53 @@ kmsg_tag_already_processed() {
 	return 1
 }
 
-append_conf() {
+append_both_conf() {
-	echo "$*" >> ${TMP_CONF_FILE}
+	echo "$*" >> ${MAIN_CONF}
+	echo "$*" >> ${USER_CONF}
 }
 
-create_config_file() {
+append_user_conf() {
-	mkdir -p /tmp/fluent-bit
+	echo "$*" >> ${USER_CONF}
-	rm -f ${TMP_CONF_FILE}
+}
-	touch ${TMP_CONF_FILE}
+
-	# include all files placed in FLUENT_BIT_CONF_DIR directory
+append_conf() {
-	# fluent-bit does not support using directory in include directive
+	echo "$*" >> ${MAIN_CONF}
-	# also, if no file is found then fluent-bit aborts
+}
-	# so only add include if any file is present in the FLUENT_BIT_CONF_DIR
+
-	if [ -d "$FLUENT_BIT_CONF_DIR" ] && [ "$(ls -A "$FLUENT_BIT_CONF_DIR")" ]; then
+create_main_config_file() {
-		append_conf "@INCLUDE ${FLUENT_BIT_CONF_DIR}/*"
+	mkdir -p ${FLUENT_BIT_TMP_DIR}
-	fi
+
-	append_conf ""
+	rm -f ${MAIN_CONF}
+	touch ${MAIN_CONF}
+}
+
+create_user_config_file() {
+	mkdir -p ${FLUENT_BIT_TMP_DIR}
+	rm -f ${USER_CONF}
+	touch ${USER_CONF}
+
+	append_user_conf "[INPUT]"
+	append_user_conf "    name tail"
+	append_user_conf "    tag $USER_TAG"
+	append_user_conf "    path /var/log/messages"
+
+	# We have to use this custom parser because /var/log/messages lines do not have PRI infront of them.
+	# So, they would be rejected by default parsers.
+	append_user_conf "    parser syslog_message"
+	append_user_conf ""
 }
 
 create_service_section() {
 	# the service section of the fluent-bit.conf file has hardcoded values,
 	# no need to lookup any uci section to configure this section
-	append_conf "[SERVICE]"
+	append_both_conf "[SERVICE]"
-	append_conf "    flush        1"
+	append_both_conf "    flush        1"
-	append_conf "    daemon       off"
+	append_both_conf "    daemon       off"
-	append_conf "    log_level    info"
+	append_both_conf "    log_level    info"
-	append_conf "    coro_stack_size    1048576"
+	append_both_conf "    coro_stack_size    1048576"
-	append_conf "    parsers_file /etc/fluent-bit/parsers.conf"
+	append_both_conf "    parsers_file /etc/fluent-bit/parsers.conf"
-	append_conf "    hot_reload        on"
+	append_both_conf "    hot_reload        on"
-	append_conf ""
+	append_both_conf ""
 }
 
 create_lua_filter_for_severity_facility() {
@@ -123,6 +147,8 @@ populate_allowed_logs() {
 
 	[ -z "$section" ] && return
 
+	# Shared state populated by populate_allowed_logs()
+	# Used by generate_syslog_filter(), create_kmsg_input_section(), etc.
 	# reset
 	match_pattern=""
 	facilities=""
@@ -180,7 +206,7 @@ populate_allowed_logs() {
 				# equal
 				severities="$sev_level"
 			else
-				# equl or higher
+				# equal or higher
 				severities="$(seq 0 $sev_level | xargs)"
 			fi
 			;;
@@ -309,25 +335,41 @@ handle_log_file() {
 		return
 	fi
 
-	append_conf "[OUTPUT]"
+	# since /var/log/messages is critically needed, it is populated by main fluent-bit instance
-	append_conf "    name         file"
+	if [ "$file" = "/var/log/messages" ]; then
-	append_conf "    workers      2"
+		append_conf "[OUTPUT]"
-	append_conf "    match_regex  $match_regex"
+		append_conf "    name         file"
-	append_conf "    file         $file"
+		append_conf "    workers      2"
+		append_conf "    match_regex  $match_regex"
+		append_conf "    file         $file"
 
 
-	if [ -n "$template" ]; then
+		if [ -n "$template" ]; then
-		append_conf "    format       template"
+			append_conf "    format       template"
-		append_conf "    template     ${template}"
+			append_conf "    template     ${template}"
+		fi
+
+		append_conf ""
+	else
+		append_user_conf "[OUTPUT]"
+		append_user_conf "    name         file"
+		append_user_conf "    workers      2"
+		append_user_conf "    match_regex  $USER_TAG"
+		append_user_conf "    file         $file"
+
+
+		if [ -n "$template" ]; then
+			append_user_conf "    format       template"
+			append_user_conf "    template     ${template}"
+		fi
+
+		append_user_conf ""
 	fi
-
-	append_conf ""
 }
 
 handle_log_remote() {
 	local section="$1"
 	local linker="$2"
-	local match_regex="$3"
 	local action_ref
 
 	config_get action_ref $section action
@@ -347,44 +389,45 @@ handle_log_remote() {
 		return
 	fi
 
-	append_conf "[OUTPUT]"
+	# log remote will always be sent via secondary (or user) fluent-bit instance
-	append_conf "    name         syslog"
+	append_user_conf "[OUTPUT]"
-	append_conf "    match_regex  $match_regex"
+	append_user_conf "    name         syslog"
-	append_conf "    host         $address"
+	append_user_conf "    match_regex  $USER_TAG"
-	append_conf "    syslog_appname_key   ident"
+	append_user_conf "    host         $address"
-	append_conf "    syslog_procid_key    pid"
+	append_user_conf "    syslog_appname_key   ident"
-	append_conf "    syslog_message_key   message"
+	append_user_conf "    syslog_procid_key    pid"
-	append_conf "    syslog_hostname_key  hostname"
+	append_user_conf "    syslog_message_key   message"
+	append_user_conf "    syslog_hostname_key  hostname"
 
 	local proto # holds value tcp or udp
 	config_get proto ${section} proto
 	if [ -n "$proto" ]; then
 		if [ "$proto" == "tls" ]; then
-			append_conf "    mode         tcp"
+			append_user_conf "    mode         tcp"
-			append_conf "    tls          on"
+			append_user_conf "    tls          on"
 		else
-			append_conf "    mode         $proto"
+			append_user_conf "    mode         $proto"
 		fi
 	fi
 
 	local port
 	config_get port $section port
 	if [ -n "$port" ]; then
-		append_conf "    port         $port"
+		append_user_conf "    port         $port"
 	fi
 
 	local cert
 	local peer_verify
 	config_get cert $section cert
 	if [ -n "$cert" ]; then
-		append_conf "    tls.crt_file $cert"
+		append_user_conf "    tls.crt_file $cert"
 
 		config_get_bool peer_verify $section peer_verify
 		if [ "$peer_verify" = "1" ]; then
-			append_conf "    tls.verify     on"
+			append_user_conf "    tls.verify     on"
 		fi
 	fi
-	append_conf ""
+	append_user_conf ""
 }
 
 resolve_source_section() {
@@ -462,6 +505,8 @@ handle_action() {
 	local source_tag_syslog source_tag_kmsg
 
 	# shared variables set by populate_allowed_logs
+	# Shared state populated by populate_allowed_logs()
+	# Used by generate_syslog_filter(), create_kmsg_input_section(), etc.
 	match_pattern=""
 	facilities=""
 	all_facilities=0
@@ -509,21 +554,34 @@ handle_action() {
 	# section so figure out if any out_log or out_syslog section is associated
 	# with this and action and setup output accordingly.
 	config_foreach handle_log_file log_file "$action_name" "$tag_regex" "$log_template"
-	config_foreach handle_log_remote log_remote "$action_name" "$tag_regex"
+
+	# tag_regex not needed because log_remote will go to user conf, which has one fixed input for /var/log/messages
+	config_foreach handle_log_remote log_remote "$action_name"
 }
 
 handle_action_section() {
 	config_foreach handle_action action
 }
 
-PROG=/usr/sbin/fluent-bit
+# So the philosophy here is that there will be one main fluent-bit instance,
+# which will read from /dev/kmsg and /dev/log and write to /var/log/messages.
+# This is done to reduce the chance of introducing errors in config and
+# making sure that /var/log/messages is always populated because logread depends on it.
+# For example, an unavailable URL can make a udp based syslog output plugin can cause
+# fluent-bit to exit entirely.
+# The other fluent-bit instance uses a different "user" config file.
+# log_file, log_remote and included configs go into this user config.
+# It has a single hard coded input tail plugin which reads from /var/log/messages.
+# Any filter that is applied will act on the main config and thus only filtered logs
+# will be available to both fluent-bit instances.
 logmngr_init() {
 	local enabled
 
 	config_load logmngr
 	config_get_bool enabled globals enable "1"
 
-	create_config_file
+	create_main_config_file
+	create_user_config_file
 	create_service_section
 	create_default_filters
 	handle_action_section
@@ -536,13 +594,14 @@ logmngr_init() {
 		return
 	fi
 
-	procd_open_instance logmngr
+	procd_open_instance logmngr_main
-	if [ -s "${TMP_CONF_FILE}" ]; then
+	if [ -s "${MAIN_CONF}" ]; then
-		procd_set_param command $PROG -c ${TMP_CONF_FILE}
+		procd_set_param command $PROG -c ${MAIN_CONF}
-		procd_set_param file ${TMP_CONF_FILE}
+		procd_set_param file ${MAIN_CONF}
 	elif [ -s "${CONF_FILE}" ]; then
 		procd_set_param command $PROG -c ${CONF_FILE}
 		procd_set_param file ${CONF_FILE}
+
 	fi
 
 	# if process finishes later than respawn_threshold, it is restarted unconditionally, regardless of error code
@@ -551,4 +610,22 @@ logmngr_init() {
 	# this is done to make sure that eventually when the url can be resolved then logging resumes
 	procd_set_param respawn ${respawn_threshold:-1} ${respawn_timeout:-5}
 	procd_close_instance
+
+	if [ -s "${USER_CONF}" ]; then
+		# include all files placed in FLUENT_BIT_CONF_DIR directory
+		# fluent-bit does not support using directory in include directive
+		# also, if no file is found then fluent-bit aborts
+		# so only add include if any file is present in the FLUENT_BIT_CONF_DIR
+		if [ -d "$FLUENT_BIT_CONF_DIR" ] && [ "$(ls -A "$FLUENT_BIT_CONF_DIR")" ]; then
+			append_user_conf "@INCLUDE ${FLUENT_BIT_CONF_DIR}/*"
+		fi
+
+		procd_open_instance logmngr_user
+		procd_set_param command $PROG -c ${USER_CONF}
+		procd_set_param file ${USER_CONF}
+
+		# same logic as above
+		procd_set_param respawn ${respawn_threshold:-1} ${respawn_timeout:-5}
+		procd_close_instance
+	fi
 }