Bug 1554293 - logging-eventrouter event not formatted correctly in El…

…asticsearch when using MUX

When fluentd is configured as a collector and MUX, event logs from the event
router need to be processed by MUX not by the collector fluentd for the both
MUX_CLIENT_MODE=maximal and minimal cases.  It is because if an event log is
formatted in the collector (note: the event record is put under the kubernetes
key), the log is forwarded to MUX and passed to the k8s-meta plugin and the
existing kubernetes record is overwritten.

To avoid the replacement, if the log is from event router, the tag is rewritten
to ${tag}.raw in input-post-forward-mux.conf, which makes the log treated in
the MUX_CLIENT_MODE=minimal way.

There was another bug in ansible.  That is, the environment variable TRANSFORM_
EVENTS was not set in MUX even if openshift_logging_install_eventrouter is set
to true.  This PR fixes the issue.
openshift/openshift-ansible#10207

In Fluentd run.sh, "process_kubernetes_events false" is set in the filter-viaq-
data-model plugin to suppress processing the event logs.  It is _not_ set, i.e.,
event router log processing is enabled, when TRANSFORM_EVENTS is true and the
fluentd is standalone (no MUX configured) or the fluentd is MUX.

Correct the order of shutdown fluentd, reset MUX_CLIENT_MODE, then restart.

fluentd/configs.d/input-post-forward-mux.conf

@@ -54,8 +54,9 @@
   <match system.var.log.messages** kubernetes.** journal.container** journal.system** audit.log**>
     @type rewrite_tag_filter
     @label @INGRESS
-    rewriterule1 @timestamp .+ ${tag}.mux
-    rewriterule2 @timestamp !.+ ${tag}.raw
+    rewriterule1 CONTAINER_NAME ^k8s_kube-eventrouter ${tag}.raw
+    rewriterule2 @timestamp .+ ${tag}.mux
+    rewriterule3 @timestamp !.+ ${tag}.raw
   </match>
 
   # If we got here, then these are external logs.  The namespace to use

fluentd/run.sh

@@ -305,7 +305,8 @@ if [[ "${USE_REMOTE_SYSLOG:-}" = "true" ]] ; then
     fi
 fi
 
-if [ "${TRANSFORM_EVENTS:-}" != true ] ; then
+# Disable process_kubernetes_events if TRANSFORM_EVENTS is false or MUX client.
+if [ "${TRANSFORM_EVENTS:-}" != true -o -n "${MUX_CLIENT_MODE:-}" ] ; then
     sed -i 's/\(.*@type viaq_data_model.*\)/\1\n  process_kubernetes_events false/' $CFG_DIR/openshift/filter-viaq-data-model.conf
 fi
 

test/eventrouter.sh

@@ -6,6 +6,8 @@ source "$(dirname "${BASH_SOURCE[0]}" )/../hack/lib/init.sh"
 source "${OS_O_A_L_DIR}/hack/testing/util.sh"
 os::test::junit::declare_suite_start "test/eventrouter"
 
+FLUENTD_WAIT_TIME=${FLUENTD_WAIT_TIME:-$(( 2 * minute ))}
+
 function warn_nonformatted() {
     local es_svc=$1
     local index=$2
@@ -19,6 +21,12 @@ function get_eventrouter_pod() {
     oc get pods --namespace=default -l component=eventrouter --no-headers | awk '$3 == "Running" {print $1}'
 }
 
+function logs_count_is_gt() {
+    local expected=$1
+    local actual=$( curl_es $esopssvc /.operations.*/_count?q=kubernetes.event.verb:* | get_count_from_json )
+    test $actual -gt $expected
+}
+
 evpod=$( get_eventrouter_pod )
 if [ -z "$evpod" ]; then
     os::log::warning "Eventrouter not deployed"
@@ -27,8 +35,38 @@ else
     esopssvc=$( get_es_svc es-ops )
     esopssvc=${esopssvc:-$essvc}
 
+    # Make sure there's no MUX
+    # undeploy fluentd
+    oc label node --all logging-infra-fluentd- 2>&1 | artifact_out
+    os::cmd::try_until_text "oc get daemonset logging-fluentd -o jsonpath='{ .status.numberReady }'" "0" $FLUENTD_WAIT_TIME
+    oc set env ds/logging-fluentd MUX_CLIENT_MODE- 2>&1 | artifact_out
+    oc label node --all logging-infra-fluentd=true 2>&1 | artifact_out
+    os::cmd::try_until_text "oc get pods -l component=fluentd" "^logging-fluentd-.* Running "
+
     warn_nonformatted $essvc '/project.*'
     warn_nonformatted $esopssvc '/.operations.*/'
 
     os::cmd::expect_success_and_not_text "curl_es $esopssvc /.operations.*/_count?q=kubernetes.event.verb:* | get_count_from_json" "^0\$"
+    prev_event_count=$( curl_es $esopssvc /.operations.*/_count?q=kubernetes.event.verb:* | get_count_from_json )
+
+    # utilize mux if mux pod exists
+    if oc get dc/logging-mux > /dev/null 2>&1 ; then
+        # MUX_CLIENT_MODE: maximal; oc set env restarts logging-fluentd
+        oc label node --all logging-infra-fluentd- 2>&1 | artifact_out
+        os::cmd::try_until_text "oc get daemonset logging-fluentd -o jsonpath='{ .status.numberReady }'" "0" $FLUENTD_WAIT_TIME
+        oc set env ds/logging-fluentd MUX_CLIENT_MODE=maximal 2>&1 | artifact_out
+        oc label node --all logging-infra-fluentd=true 2>&1 | artifact_out
+        os::cmd::try_until_text "oc get pods -l component=fluentd" "^logging-fluentd-.* Running "
+        os::cmd::try_until_success "logs_count_is_gt $prev_event_count"
+        prev_event_count=$( curl_es $esopssvc /.operations.*/_count?q=kubernetes.event.verb:* | get_count_from_json )
+
+        # MUX_CLIENT_MODE: minimal; oc set env restarts logging-fluentd
+        oc label node --all logging-infra-fluentd- 2>&1 | artifact_out
+        os::cmd::try_until_text "oc get daemonset logging-fluentd -o jsonpath='{ .status.numberReady }'" "0" $FLUENTD_WAIT_TIME
+        oc set env ds/logging-fluentd MUX_CLIENT_MODE=minimal 2>&1 | artifact_out
+        oc label node --all logging-infra-fluentd=true 2>&1 | artifact_out
+        os::cmd::try_until_text "oc get pods -l component=fluentd" "^logging-fluentd-.* Running "
+        os::cmd::try_until_success "logs_count_is_gt $prev_event_count"
+    fi
+
 fi