diff --git a/hack/testing-olm/test-040-eventrouter.sh b/hack/testing-olm/test-040-eventrouter.sh
index 876d376031..6bc7f69294 100755
--- a/hack/testing-olm/test-040-eventrouter.sh
+++ b/hack/testing-olm/test-040-eventrouter.sh
@@ -152,7 +152,7 @@ espod=$(oc -n $LOGGING_NS get pods -l component=elasticsearch -o jsonpath={.item
 os::log::info "Testing Elasticsearch pod ${espod}..."
 os::cmd::try_until_text "oc -n $LOGGING_NS exec -c elasticsearch ${espod} -- es_util --query=/ --request HEAD --head --output /dev/null --write-out %{response_code}" "200" "$(( 1*$minute ))"
 
-warn_nonformatted 'infra'
+warn_nonformatted 'infra-*'
 
 evpod=$(oc -n $LOGGING_NS get pods -l component=eventrouter -o jsonpath={.items[0].metadata.name})
 
diff --git a/internal/generator/fluentd/conf.go b/internal/generator/fluentd/conf.go
index 5982ba68ca..ea35ae2d1a 100644
--- a/internal/generator/fluentd/conf.go
+++ b/internal/generator/fluentd/conf.go
@@ -31,10 +31,6 @@ func Conf(clspec *logging.ClusterLoggingSpec, secrets map[string]*corev1.Secret,
 			Sources(clspec, clfspec, op),
 			"Set of all input sources",
 		},
-		{
-			PrometheusMetrics(clfspec, op),
-			"Section to add measurement, and dispatch to Concat or Ingress pipelines",
-		},
 		{
 			Concat(clfspec, op),
 			`Concat pipeline section`,
diff --git a/internal/generator/fluentd/conf_test.go b/internal/generator/fluentd/conf_test.go
index 12a4c0a18b..bf54f5bc46 100644
--- a/internal/generator/fluentd/conf_test.go
+++ b/internal/generator/fluentd/conf_test.go
@@ -116,7 +116,7 @@ var _ = Describe("Testing Complete Config Generation", func() {
 <source>
   @type systemd
   @id systemd-input
-  @label @MEASURE
+  @label @INGRESS
   path '/var/log/journal'
   <storage>
     @type local
@@ -134,28 +134,20 @@ var _ = Describe("Testing Complete Config Generation", func() {
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
   tag kubernetes.*
   read_from_head "true"
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 
@@ -163,7 +155,7 @@ var _ = Describe("Testing Complete Config Generation", func() {
 <source>
   @type tail
   @id audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/audit/audit.log"
   pos_file "/var/lib/fluentd/pos/audit.log.pos"
   tag linux-audit.log
@@ -176,7 +168,7 @@ var _ = Describe("Testing Complete Config Generation", func() {
 <source>
   @type tail
   @id k8s-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/kube-apiserver/audit.log"
   pos_file "/var/lib/fluentd/pos/kube-apiserver.audit.log.pos"
   tag k8s-audit.log
@@ -193,7 +185,7 @@ var _ = Describe("Testing Complete Config Generation", func() {
 <source>
   @type tail
   @id openshift-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path /var/log/oauth-apiserver/audit.log,/var/log/openshift-apiserver/audit.log
   pos_file /var/lib/fluentd/pos/oauth-apiserver.audit.log
   tag openshift-audit.log
@@ -210,7 +202,7 @@ var _ = Describe("Testing Complete Config Generation", func() {
 <source>
   @type tail
   @id ovn-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/ovn/acl-audit-log.log"
   pos_file "/var/lib/fluentd/pos/acl-audit-log.pos"
   tag ovn-audit.log
@@ -222,72 +214,11 @@ var _ = Describe("Testing Complete Config Generation", func() {
   </parse>
 </source>
 
-# Increment Prometheus metrics
-<label @MEASURE>
-  <filter **>
-    @type record_transformer
-    enable_ruby
-    <record>
-      msg_size ${record.to_s.length}
-    </record>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_total
-      type counter
-      desc The total number of incoming records
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_bytes
-      type counter
-      desc The total bytes of incoming records
-      key msg_size
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type record_transformer
-    remove_keys msg_size
-  </filter>
-  
-  # Journal Logs go to INGRESS pipeline
-  <match journal>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Audit Logs go to INGRESS pipeline
-  <match *audit.log>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Kubernetes Logs go to CONCAT pipeline
-  <match kubernetes.**>
-    @type relabel
-    @label @CONCAT
-  </match>
-</label>
-
 # Concat log lines of container logs, and send to INGRESS pipeline
 <label @CONCAT>
   <filter kubernetes.**>
     @type concat
-    key log
+    key message
     partial_key logtag
     partial_value P
     separator ''
@@ -391,47 +322,22 @@ var _ = Describe("Testing Complete Config Generation", func() {
     </rule>
   </match>
   
-  # Invoke kubernetes apiserver to get kunbernetes metadata
+  # Invoke kubernetes apiserver to get kubernetes metadata
   <filter kubernetes.**>
     @id kubernetes-metadata
     @type kubernetes_metadata
     kubernetes_url 'https://kubernetes.default.svc'
     allow_orphans false
     cache_size '1000'
-    use_journal 'nil'
     ssl_partial_chain 'true'
   </filter>
   
   # Parse Json fields for container, journal and eventrouter logs
-  <filter kubernetes.journal.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  <filter kubernetes.var.log.pods.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
   <filter kubernetes.var.log.pods.**_eventrouter-**>
     @type parse_json_field
     merge_json_log true
     preserve_json_log true
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  # Clean kibana log fields
-  <filter **kibana**>
-    @type record_transformer
-    enable_ruby
-    <record>
-      log ${record['err'] || record['msg'] || record['MESSAGE'] || record['log']}
-    </record>
-    remove_keys req,res,msg,name,level,v,pid,err
+    json_fields 'message'
   </filter>
   
   # Fix level field in audit logs
@@ -452,21 +358,12 @@ var _ = Describe("Testing Complete Config Generation", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
-    elasticsearch_index_prefix_field 'viaq_index_name'
+    enable_prune_empty_fields false
     default_keep_fields CEE,time,@timestamp,aushape,ci_job,collectd,docker,fedora-ci,file,foreman,geoip,hostname,ipaddr4,ipaddr6,kubernetes,level,message,namespace_name,namespace_uuid,offset,openstack,ovirt,pid,pipeline_metadata,rsyslog,service,systemd,tags,testcase,tlog,viaq_msg_id
-    extra_keep_fields ''
     keep_empty_fields 'message'
-    use_undefined false
-    undefined_name 'undefined'
     rename_time true
-    rename_time_if_missing false
-    src_time_name 'time'
-    dest_time_name '@timestamp'
     pipeline_type 'collector'
-    undefined_to_string 'false'
-    undefined_dot_replace_char 'UNUSED'
-    undefined_max_num_fields '-1'
-    process_kubernetes_events 'false'
+    process_kubernetes_events false
     <level>
       name warn
       match 'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"'
@@ -487,50 +384,22 @@ var _ = Describe("Testing Complete Config Generation", func() {
       name debug
       match 'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"'
     </level>
-    <formatter>
-      tag "system.var.log**"
-      type sys_var_log
-      remove_keys host,pid,ident
-    </formatter>
     <formatter>
       tag "journal.system**"
       type sys_journal
       remove_keys log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID
     </formatter>
-    <formatter>
-      tag "kubernetes.journal.container**"
-      type k8s_journal
-      remove_keys 'log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID'
-    </formatter>
     <formatter>
       tag "kubernetes.var.log.pods.**_eventrouter-** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
       process_kubernetes_events 'true'
     </formatter>
     <formatter>
       tag "kubernetes.var.log.pods**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
     </formatter>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** journal.system** system.var.log**"
-      name_type static
-      static_index_name infra-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
-      name_type static
-      static_index_name audit-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "**"
-      name_type static
-      static_index_name app-write
-    </elasticsearch_index_name>
   </filter>
   
   # Generate elasticsearch id
@@ -630,6 +499,10 @@ var _ = Describe("Testing Complete Config Generation", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
diff --git a/internal/generator/fluentd/fluent_conf_test.go b/internal/generator/fluentd/fluent_conf_test.go
index 4fd46303b9..3bdaf2a942 100644
--- a/internal/generator/fluentd/fluent_conf_test.go
+++ b/internal/generator/fluentd/fluent_conf_test.go
@@ -240,97 +240,28 @@ var _ = Describe("Generating fluentd config", func() {
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
   tag kubernetes.*
   read_from_head "true"
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 
-# Increment Prometheus metrics
-<label @MEASURE>
-  <filter **>
-    @type record_transformer
-    enable_ruby
-    <record>
-      msg_size ${record.to_s.length}
-    </record>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_total
-      type counter
-      desc The total number of incoming records
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_bytes
-      type counter
-      desc The total bytes of incoming records
-      key msg_size
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type record_transformer
-    remove_keys msg_size
-  </filter>
-  
-  # Journal Logs go to INGRESS pipeline
-  <match journal>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Audit Logs go to INGRESS pipeline
-  <match *audit.log>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Kubernetes Logs go to CONCAT pipeline
-  <match kubernetes.**>
-    @type relabel
-    @label @CONCAT
-  </match>
-</label>
-
 # Concat log lines of container logs, and send to INGRESS pipeline
 <label @CONCAT>
   <filter kubernetes.**>
     @type concat
-    key log
+    key message
     partial_key logtag
     partial_value P
     separator ''
@@ -434,47 +365,22 @@ var _ = Describe("Generating fluentd config", func() {
     </rule>
   </match>
   
-  # Invoke kubernetes apiserver to get kunbernetes metadata
+  # Invoke kubernetes apiserver to get kubernetes metadata
   <filter kubernetes.**>
 	@id kubernetes-metadata
     @type kubernetes_metadata
     kubernetes_url 'https://kubernetes.default.svc'
     allow_orphans false
     cache_size '1000'
-    use_journal 'nil'
     ssl_partial_chain 'true'
   </filter>
   
   # Parse Json fields for container, journal and eventrouter logs
-  <filter kubernetes.journal.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  <filter kubernetes.var.log.pods.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
   <filter kubernetes.var.log.pods.**_eventrouter-**>
     @type parse_json_field
     merge_json_log true
     preserve_json_log true
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  # Clean kibana log fields
-  <filter **kibana**>
-    @type record_transformer
-    enable_ruby
-    <record>
-      log ${record['err'] || record['msg'] || record['MESSAGE'] || record['log']}
-    </record>
-    remove_keys req,res,msg,name,level,v,pid,err
+    json_fields 'message'
   </filter>
   
   # Fix level field in audit logs
@@ -495,21 +401,12 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
-    elasticsearch_index_prefix_field 'viaq_index_name'
+    enable_prune_empty_fields false
     default_keep_fields CEE,time,@timestamp,aushape,ci_job,collectd,docker,fedora-ci,file,foreman,geoip,hostname,ipaddr4,ipaddr6,kubernetes,level,message,namespace_name,namespace_uuid,offset,openstack,ovirt,pid,pipeline_metadata,rsyslog,service,systemd,tags,testcase,tlog,viaq_msg_id
-    extra_keep_fields ''
     keep_empty_fields 'message'
-    use_undefined false
-    undefined_name 'undefined'
     rename_time true
-    rename_time_if_missing false
-    src_time_name 'time'
-    dest_time_name '@timestamp'
     pipeline_type 'collector'
-    undefined_to_string 'false'
-    undefined_dot_replace_char 'UNUSED'
-    undefined_max_num_fields '-1'
-    process_kubernetes_events 'false'
+    process_kubernetes_events false
     <level>
       name warn
       match 'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"'
@@ -530,50 +427,22 @@ var _ = Describe("Generating fluentd config", func() {
       name debug
       match 'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"'
     </level>
-    <formatter>
-      tag "system.var.log**"
-      type sys_var_log
-      remove_keys host,pid,ident
-    </formatter>
     <formatter>
       tag "journal.system**"
       type sys_journal
       remove_keys log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID
     </formatter>
-    <formatter>
-      tag "kubernetes.journal.container**"
-      type k8s_journal
-      remove_keys 'log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID'
-    </formatter>
     <formatter>
       tag "kubernetes.var.log.pods.**_eventrouter-** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
       process_kubernetes_events 'true'
     </formatter>
     <formatter>
       tag "kubernetes.var.log.pods**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
     </formatter>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** journal.system** system.var.log**"
-      name_type static
-      static_index_name infra-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
-      name_type static
-      static_index_name audit-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "**"
-      name_type static
-      static_index_name app-write
-    </elasticsearch_index_name>
   </filter>
   
   # Generate elasticsearch id
@@ -694,6 +563,10 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -828,6 +701,10 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -1086,97 +963,28 @@ var _ = Describe("Generating fluentd config", func() {
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
   tag kubernetes.*
   read_from_head "true"
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 
-# Increment Prometheus metrics
-<label @MEASURE>
-  <filter **>
-    @type record_transformer
-    enable_ruby
-    <record>
-      msg_size ${record.to_s.length}
-    </record>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_total
-      type counter
-      desc The total number of incoming records
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_bytes
-      type counter
-      desc The total bytes of incoming records
-      key msg_size
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type record_transformer
-    remove_keys msg_size
-  </filter>
-  
-  # Journal Logs go to INGRESS pipeline
-  <match journal>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Audit Logs go to INGRESS pipeline
-  <match *audit.log>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Kubernetes Logs go to CONCAT pipeline
-  <match kubernetes.**>
-    @type relabel
-    @label @CONCAT
-  </match>
-</label>
-
 # Concat log lines of container logs, and send to INGRESS pipeline
 <label @CONCAT>
   <filter kubernetes.**>
     @type concat
-    key log
+    key message
     partial_key logtag
     partial_value P
     separator ''
@@ -1280,47 +1088,22 @@ var _ = Describe("Generating fluentd config", func() {
     </rule>
   </match>
   
-  # Invoke kubernetes apiserver to get kunbernetes metadata
+  # Invoke kubernetes apiserver to get kubernetes metadata
   <filter kubernetes.**>
 	@id kubernetes-metadata
     @type kubernetes_metadata
     kubernetes_url 'https://kubernetes.default.svc'
     allow_orphans false
     cache_size '1000'
-    use_journal 'nil'
     ssl_partial_chain 'true'
   </filter>
   
   # Parse Json fields for container, journal and eventrouter logs
-  <filter kubernetes.journal.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  <filter kubernetes.var.log.pods.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
   <filter kubernetes.var.log.pods.**_eventrouter-**>
     @type parse_json_field
     merge_json_log true
     preserve_json_log true
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  # Clean kibana log fields
-  <filter **kibana**>
-    @type record_transformer
-    enable_ruby
-    <record>
-      log ${record['err'] || record['msg'] || record['MESSAGE'] || record['log']}
-    </record>
-    remove_keys req,res,msg,name,level,v,pid,err
+    json_fields 'message'
   </filter>
   
   # Fix level field in audit logs
@@ -1341,21 +1124,12 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
-    elasticsearch_index_prefix_field 'viaq_index_name'
+    enable_prune_empty_fields false
     default_keep_fields CEE,time,@timestamp,aushape,ci_job,collectd,docker,fedora-ci,file,foreman,geoip,hostname,ipaddr4,ipaddr6,kubernetes,level,message,namespace_name,namespace_uuid,offset,openstack,ovirt,pid,pipeline_metadata,rsyslog,service,systemd,tags,testcase,tlog,viaq_msg_id
-    extra_keep_fields ''
     keep_empty_fields 'message'
-    use_undefined false
-    undefined_name 'undefined'
     rename_time true
-    rename_time_if_missing false
-    src_time_name 'time'
-    dest_time_name '@timestamp'
     pipeline_type 'collector'
-    undefined_to_string 'false'
-    undefined_dot_replace_char 'UNUSED'
-    undefined_max_num_fields '-1'
-    process_kubernetes_events 'false'
+    process_kubernetes_events false
     <level>
       name warn
       match 'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"'
@@ -1376,50 +1150,22 @@ var _ = Describe("Generating fluentd config", func() {
       name debug
       match 'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"'
     </level>
-    <formatter>
-      tag "system.var.log**"
-      type sys_var_log
-      remove_keys host,pid,ident
-    </formatter>
     <formatter>
       tag "journal.system**"
       type sys_journal
       remove_keys log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID
     </formatter>
-    <formatter>
-      tag "kubernetes.journal.container**"
-      type k8s_journal
-      remove_keys 'log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID'
-    </formatter>
     <formatter>
       tag "kubernetes.var.log.pods.**_eventrouter-** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
       process_kubernetes_events 'true'
     </formatter>
     <formatter>
       tag "kubernetes.var.log.pods**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
     </formatter>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** journal.system** system.var.log**"
-      name_type static
-      static_index_name infra-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
-      name_type static
-      static_index_name audit-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "**"
-      name_type static
-      static_index_name app-write
-    </elasticsearch_index_name>
   </filter>
   
   # Generate elasticsearch id
@@ -1523,6 +1269,10 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -1657,6 +1407,10 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -1917,97 +1671,28 @@ var _ = Describe("Generating fluentd config", func() {
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
   tag kubernetes.*
   read_from_head "true"
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 
-# Increment Prometheus metrics
-<label @MEASURE>
-  <filter **>
-    @type record_transformer
-    enable_ruby
-    <record>
-      msg_size ${record.to_s.length}
-    </record>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_total
-      type counter
-      desc The total number of incoming records
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_bytes
-      type counter
-      desc The total bytes of incoming records
-      key msg_size
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type record_transformer
-    remove_keys msg_size
-  </filter>
-  
-  # Journal Logs go to INGRESS pipeline
-  <match journal>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Audit Logs go to INGRESS pipeline
-  <match *audit.log>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Kubernetes Logs go to CONCAT pipeline
-  <match kubernetes.**>
-    @type relabel
-    @label @CONCAT
-  </match>
-</label>
-
 # Concat log lines of container logs, and send to INGRESS pipeline
 <label @CONCAT>
   <filter kubernetes.**>
     @type concat
-    key log
+    key message
     partial_key logtag
     partial_value P
     separator ''
@@ -2111,47 +1796,22 @@ var _ = Describe("Generating fluentd config", func() {
     </rule>
   </match>
   
-  # Invoke kubernetes apiserver to get kunbernetes metadata
+  # Invoke kubernetes apiserver to get kubernetes metadata
   <filter kubernetes.**>
 	@id kubernetes-metadata
     @type kubernetes_metadata
     kubernetes_url 'https://kubernetes.default.svc'
     allow_orphans false
     cache_size '1000'
-    use_journal 'nil'
     ssl_partial_chain 'true'
   </filter>
   
   # Parse Json fields for container, journal and eventrouter logs
-  <filter kubernetes.journal.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  <filter kubernetes.var.log.pods.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
   <filter kubernetes.var.log.pods.**_eventrouter-**>
     @type parse_json_field
     merge_json_log true
     preserve_json_log true
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  # Clean kibana log fields
-  <filter **kibana**>
-    @type record_transformer
-    enable_ruby
-    <record>
-      log ${record['err'] || record['msg'] || record['MESSAGE'] || record['log']}
-    </record>
-    remove_keys req,res,msg,name,level,v,pid,err
+    json_fields 'message'
   </filter>
   
   # Fix level field in audit logs
@@ -2172,21 +1832,12 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
-    elasticsearch_index_prefix_field 'viaq_index_name'
+    enable_prune_empty_fields false
     default_keep_fields CEE,time,@timestamp,aushape,ci_job,collectd,docker,fedora-ci,file,foreman,geoip,hostname,ipaddr4,ipaddr6,kubernetes,level,message,namespace_name,namespace_uuid,offset,openstack,ovirt,pid,pipeline_metadata,rsyslog,service,systemd,tags,testcase,tlog,viaq_msg_id
-    extra_keep_fields ''
     keep_empty_fields 'message'
-    use_undefined false
-    undefined_name 'undefined'
     rename_time true
-    rename_time_if_missing false
-    src_time_name 'time'
-    dest_time_name '@timestamp'
     pipeline_type 'collector'
-    undefined_to_string 'false'
-    undefined_dot_replace_char 'UNUSED'
-    undefined_max_num_fields '-1'
-    process_kubernetes_events 'false'
+    process_kubernetes_events false
     <level>
       name warn
       match 'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"'
@@ -2207,50 +1858,22 @@ var _ = Describe("Generating fluentd config", func() {
       name debug
       match 'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"'
     </level>
-    <formatter>
-      tag "system.var.log**"
-      type sys_var_log
-      remove_keys host,pid,ident
-    </formatter>
     <formatter>
       tag "journal.system**"
       type sys_journal
       remove_keys log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID
     </formatter>
-    <formatter>
-      tag "kubernetes.journal.container**"
-      type k8s_journal
-      remove_keys 'log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID'
-    </formatter>
     <formatter>
       tag "kubernetes.var.log.pods.**_eventrouter-** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
       process_kubernetes_events 'true'
     </formatter>
     <formatter>
       tag "kubernetes.var.log.pods**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
     </formatter>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** journal.system** system.var.log**"
-      name_type static
-      static_index_name infra-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
-      name_type static
-      static_index_name audit-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "**"
-      name_type static
-      static_index_name app-write
-    </elasticsearch_index_name>
   </filter>
   
   # Generate elasticsearch id
@@ -2356,6 +1979,10 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -2490,6 +2117,10 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -2693,97 +2324,28 @@ var _ = Describe("Generating fluentd config", func() {
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
   tag kubernetes.*
   read_from_head "true"
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 
-# Increment Prometheus metrics
-<label @MEASURE>
-  <filter **>
-    @type record_transformer
-    enable_ruby
-    <record>
-      msg_size ${record.to_s.length}
-    </record>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_total
-      type counter
-      desc The total number of incoming records
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_bytes
-      type counter
-      desc The total bytes of incoming records
-      key msg_size
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type record_transformer
-    remove_keys msg_size
-  </filter>
-  
-  # Journal Logs go to INGRESS pipeline
-  <match journal>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Audit Logs go to INGRESS pipeline
-  <match *audit.log>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Kubernetes Logs go to CONCAT pipeline
-  <match kubernetes.**>
-    @type relabel
-    @label @CONCAT
-  </match>
-</label>
-
 # Concat log lines of container logs, and send to INGRESS pipeline
 <label @CONCAT>
   <filter kubernetes.**>
     @type concat
-    key log
+    key message
     partial_key logtag
     partial_value P
     separator ''
@@ -2887,47 +2449,22 @@ var _ = Describe("Generating fluentd config", func() {
     </rule>
   </match>
   
-  # Invoke kubernetes apiserver to get kunbernetes metadata
+  # Invoke kubernetes apiserver to get kubernetes metadata
   <filter kubernetes.**>
 	@id kubernetes-metadata
     @type kubernetes_metadata
     kubernetes_url 'https://kubernetes.default.svc'
     allow_orphans false
     cache_size '1000'
-    use_journal 'nil'
     ssl_partial_chain 'true'
   </filter>
   
   # Parse Json fields for container, journal and eventrouter logs
-  <filter kubernetes.journal.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  <filter kubernetes.var.log.pods.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
   <filter kubernetes.var.log.pods.**_eventrouter-**>
     @type parse_json_field
     merge_json_log true
     preserve_json_log true
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  # Clean kibana log fields
-  <filter **kibana**>
-    @type record_transformer
-    enable_ruby
-    <record>
-      log ${record['err'] || record['msg'] || record['MESSAGE'] || record['log']}
-    </record>
-    remove_keys req,res,msg,name,level,v,pid,err
+    json_fields 'message'
   </filter>
   
   # Fix level field in audit logs
@@ -2948,21 +2485,12 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
-    elasticsearch_index_prefix_field 'viaq_index_name'
+    enable_prune_empty_fields false
     default_keep_fields CEE,time,@timestamp,aushape,ci_job,collectd,docker,fedora-ci,file,foreman,geoip,hostname,ipaddr4,ipaddr6,kubernetes,level,message,namespace_name,namespace_uuid,offset,openstack,ovirt,pid,pipeline_metadata,rsyslog,service,systemd,tags,testcase,tlog,viaq_msg_id
-    extra_keep_fields ''
     keep_empty_fields 'message'
-    use_undefined false
-    undefined_name 'undefined'
     rename_time true
-    rename_time_if_missing false
-    src_time_name 'time'
-    dest_time_name '@timestamp'
     pipeline_type 'collector'
-    undefined_to_string 'false'
-    undefined_dot_replace_char 'UNUSED'
-    undefined_max_num_fields '-1'
-    process_kubernetes_events 'false'
+    process_kubernetes_events false
     <level>
       name warn
       match 'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"'
@@ -2983,50 +2511,22 @@ var _ = Describe("Generating fluentd config", func() {
       name debug
       match 'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"'
     </level>
-    <formatter>
-      tag "system.var.log**"
-      type sys_var_log
-      remove_keys host,pid,ident
-    </formatter>
     <formatter>
       tag "journal.system**"
       type sys_journal
       remove_keys log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID
     </formatter>
-    <formatter>
-      tag "kubernetes.journal.container**"
-      type k8s_journal
-      remove_keys 'log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID'
-    </formatter>
     <formatter>
       tag "kubernetes.var.log.pods.**_eventrouter-** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
       process_kubernetes_events 'true'
     </formatter>
     <formatter>
       tag "kubernetes.var.log.pods**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
     </formatter>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** journal.system** system.var.log**"
-      name_type static
-      static_index_name infra-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
-      name_type static
-      static_index_name audit-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "**"
-      name_type static
-      static_index_name app-write
-    </elasticsearch_index_name>
   </filter>
   
   # Generate elasticsearch id
@@ -3171,7 +2671,7 @@ var _ = Describe("Generating fluentd config", func() {
 <source>
   @type systemd
   @id systemd-input
-  @label @MEASURE
+  @label @INGRESS
   path '/var/log/journal'
   <storage>
     @type local
@@ -3189,28 +2689,20 @@ var _ = Describe("Generating fluentd config", func() {
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
   tag kubernetes.*
   read_from_head "true"
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 
@@ -3218,7 +2710,7 @@ var _ = Describe("Generating fluentd config", func() {
 <source>
   @type tail
   @id audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/audit/audit.log"
   pos_file "/var/lib/fluentd/pos/audit.log.pos"
   tag linux-audit.log
@@ -3231,7 +2723,7 @@ var _ = Describe("Generating fluentd config", func() {
 <source>
   @type tail
   @id k8s-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/kube-apiserver/audit.log"
   pos_file "/var/lib/fluentd/pos/kube-apiserver.audit.log.pos"
   tag k8s-audit.log
@@ -3248,7 +2740,7 @@ var _ = Describe("Generating fluentd config", func() {
 <source>
   @type tail
   @id openshift-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path /var/log/oauth-apiserver/audit.log,/var/log/openshift-apiserver/audit.log
   pos_file /var/lib/fluentd/pos/oauth-apiserver.audit.log
   tag openshift-audit.log
@@ -3265,7 +2757,7 @@ var _ = Describe("Generating fluentd config", func() {
 <source>
   @type tail
   @id ovn-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/ovn/acl-audit-log.log"
   pos_file "/var/lib/fluentd/pos/acl-audit-log.pos"
   tag ovn-audit.log
@@ -3277,72 +2769,11 @@ var _ = Describe("Generating fluentd config", func() {
   </parse>
 </source>
 
-# Increment Prometheus metrics
-<label @MEASURE>
-  <filter **>
-    @type record_transformer
-    enable_ruby
-    <record>
-      msg_size ${record.to_s.length}
-    </record>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_total
-      type counter
-      desc The total number of incoming records
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_bytes
-      type counter
-      desc The total bytes of incoming records
-      key msg_size
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type record_transformer
-    remove_keys msg_size
-  </filter>
-  
-  # Journal Logs go to INGRESS pipeline
-  <match journal>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Audit Logs go to INGRESS pipeline
-  <match *audit.log>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Kubernetes Logs go to CONCAT pipeline
-  <match kubernetes.**>
-    @type relabel
-    @label @CONCAT
-  </match>
-</label>
-
 # Concat log lines of container logs, and send to INGRESS pipeline
 <label @CONCAT>
   <filter kubernetes.**>
     @type concat
-    key log
+    key message
     partial_key logtag
     partial_value P
     separator ''
@@ -3446,47 +2877,22 @@ var _ = Describe("Generating fluentd config", func() {
     </rule>
   </match>
   
-  # Invoke kubernetes apiserver to get kunbernetes metadata
+  # Invoke kubernetes apiserver to get kubernetes metadata
   <filter kubernetes.**>
 	@id kubernetes-metadata
     @type kubernetes_metadata
     kubernetes_url 'https://kubernetes.default.svc'
     allow_orphans false
     cache_size '1000'
-    use_journal 'nil'
     ssl_partial_chain 'true'
   </filter>
   
   # Parse Json fields for container, journal and eventrouter logs
-  <filter kubernetes.journal.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  <filter kubernetes.var.log.pods.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
   <filter kubernetes.var.log.pods.**_eventrouter-**>
     @type parse_json_field
     merge_json_log true
     preserve_json_log true
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  # Clean kibana log fields
-  <filter **kibana**>
-    @type record_transformer
-    enable_ruby
-    <record>
-      log ${record['err'] || record['msg'] || record['MESSAGE'] || record['log']}
-    </record>
-    remove_keys req,res,msg,name,level,v,pid,err
+    json_fields 'message'
   </filter>
   
   # Fix level field in audit logs
@@ -3507,21 +2913,12 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
-    elasticsearch_index_prefix_field 'viaq_index_name'
+    enable_prune_empty_fields false
     default_keep_fields CEE,time,@timestamp,aushape,ci_job,collectd,docker,fedora-ci,file,foreman,geoip,hostname,ipaddr4,ipaddr6,kubernetes,level,message,namespace_name,namespace_uuid,offset,openstack,ovirt,pid,pipeline_metadata,rsyslog,service,systemd,tags,testcase,tlog,viaq_msg_id
-    extra_keep_fields ''
     keep_empty_fields 'message'
-    use_undefined false
-    undefined_name 'undefined'
     rename_time true
-    rename_time_if_missing false
-    src_time_name 'time'
-    dest_time_name '@timestamp'
     pipeline_type 'collector'
-    undefined_to_string 'false'
-    undefined_dot_replace_char 'UNUSED'
-    undefined_max_num_fields '-1'
-    process_kubernetes_events 'false'
+    process_kubernetes_events false
     <level>
       name warn
       match 'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"'
@@ -3542,50 +2939,22 @@ var _ = Describe("Generating fluentd config", func() {
       name debug
       match 'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"'
     </level>
-    <formatter>
-      tag "system.var.log**"
-      type sys_var_log
-      remove_keys host,pid,ident
-    </formatter>
     <formatter>
       tag "journal.system**"
       type sys_journal
       remove_keys log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID
     </formatter>
-    <formatter>
-      tag "kubernetes.journal.container**"
-      type k8s_journal
-      remove_keys 'log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID'
-    </formatter>
     <formatter>
       tag "kubernetes.var.log.pods.**_eventrouter-** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
       process_kubernetes_events 'true'
     </formatter>
     <formatter>
       tag "kubernetes.var.log.pods**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
     </formatter>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** journal.system** system.var.log**"
-      name_type static
-      static_index_name infra-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
-      name_type static
-      static_index_name audit-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "**"
-      name_type static
-      static_index_name app-write
-    </elasticsearch_index_name>
   </filter>
   
   # Generate elasticsearch id
@@ -3703,6 +3072,10 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -3837,6 +3210,10 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -3971,6 +3348,10 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -4105,6 +3486,10 @@ var _ = Describe("Generating fluentd config", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -4381,97 +3766,28 @@ inputs:
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
   tag kubernetes.*
   read_from_head "true"
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 
-# Increment Prometheus metrics
-<label @MEASURE>
-  <filter **>
-    @type record_transformer
-    enable_ruby
-    <record>
-      msg_size ${record.to_s.length}
-    </record>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_total
-      type counter
-      desc The total number of incoming records
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type prometheus
-    <metric>
-      name cluster_logging_collector_input_record_bytes
-      type counter
-      desc The total bytes of incoming records
-      key msg_size
-      <labels>
-        tag ${tag}
-        hostname ${hostname}
-      </labels>
-    </metric>
-  </filter>
-  
-  <filter **>
-    @type record_transformer
-    remove_keys msg_size
-  </filter>
-  
-  # Journal Logs go to INGRESS pipeline
-  <match journal>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Audit Logs go to INGRESS pipeline
-  <match *audit.log>
-    @type relabel
-    @label @INGRESS
-  </match>
-  
-  # Kubernetes Logs go to CONCAT pipeline
-  <match kubernetes.**>
-    @type relabel
-    @label @CONCAT
-  </match>
-</label>
-
 # Concat log lines of container logs, and send to INGRESS pipeline
 <label @CONCAT>
   <filter kubernetes.**>
     @type concat
-    key log
+    key message
     partial_key logtag
     partial_value P
     separator ''
@@ -4575,47 +3891,22 @@ inputs:
     </rule>
   </match>
   
-  # Invoke kubernetes apiserver to get kunbernetes metadata
+  # Invoke kubernetes apiserver to get kubernetes metadata
   <filter kubernetes.**>
 	@id kubernetes-metadata
     @type kubernetes_metadata
     kubernetes_url 'https://kubernetes.default.svc'
 	allow_orphans false
     cache_size '1000'
-    use_journal 'nil'
     ssl_partial_chain 'true'
   </filter>
   
   # Parse Json fields for container, journal and eventrouter logs
-  <filter kubernetes.journal.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  <filter kubernetes.var.log.pods.**>
-    @type parse_json_field
-    merge_json_log 'false'
-    preserve_json_log 'true'
-    json_fields 'log,MESSAGE'
-  </filter>
-  
   <filter kubernetes.var.log.pods.**_eventrouter-**>
     @type parse_json_field
     merge_json_log true
     preserve_json_log true
-    json_fields 'log,MESSAGE'
-  </filter>
-  
-  # Clean kibana log fields
-  <filter **kibana**>
-    @type record_transformer
-    enable_ruby
-    <record>
-      log ${record['err'] || record['msg'] || record['MESSAGE'] || record['log']}
-    </record>
-    remove_keys req,res,msg,name,level,v,pid,err
+    json_fields 'message'
   </filter>
   
   # Fix level field in audit logs
@@ -4636,21 +3927,12 @@ inputs:
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
-    elasticsearch_index_prefix_field 'viaq_index_name'
+    enable_prune_empty_fields false
     default_keep_fields CEE,time,@timestamp,aushape,ci_job,collectd,docker,fedora-ci,file,foreman,geoip,hostname,ipaddr4,ipaddr6,kubernetes,level,message,namespace_name,namespace_uuid,offset,openstack,ovirt,pid,pipeline_metadata,rsyslog,service,systemd,tags,testcase,tlog,viaq_msg_id
-    extra_keep_fields ''
     keep_empty_fields 'message'
-    use_undefined false
-    undefined_name 'undefined'
     rename_time true
-    rename_time_if_missing false
-    src_time_name 'time'
-    dest_time_name '@timestamp'
     pipeline_type 'collector'
-    undefined_to_string 'false'
-    undefined_dot_replace_char 'UNUSED'
-    undefined_max_num_fields '-1'
-    process_kubernetes_events 'false'
+    process_kubernetes_events false
     <level>
       name warn
       match 'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"'
@@ -4671,50 +3953,22 @@ inputs:
       name debug
       match 'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"'
     </level>
-    <formatter>
-      tag "system.var.log**"
-      type sys_var_log
-      remove_keys host,pid,ident
-    </formatter>
     <formatter>
       tag "journal.system**"
       type sys_journal
       remove_keys log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID
     </formatter>
-    <formatter>
-      tag "kubernetes.journal.container**"
-      type k8s_journal
-      remove_keys 'log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID'
-    </formatter>
     <formatter>
       tag "kubernetes.var.log.pods.**_eventrouter-** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
       process_kubernetes_events 'true'
     </formatter>
     <formatter>
       tag "kubernetes.var.log.pods**"
       type k8s_json_file
-      remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+      remove_keys stream
     </formatter>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** journal.system** system.var.log**"
-      name_type static
-      static_index_name infra-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
-      name_type static
-      static_index_name audit-write
-    </elasticsearch_index_name>
-    <elasticsearch_index_name>
-      enabled 'true'
-      tag "**"
-      name_type static
-      static_index_name app-write
-    </elasticsearch_index_name>
   </filter>
   
   # Generate elasticsearch id
diff --git a/internal/generator/fluentd/ingress.go b/internal/generator/fluentd/ingress.go
index db84d27fa9..ff3cfc4b01 100644
--- a/internal/generator/fluentd/ingress.go
+++ b/internal/generator/fluentd/ingress.go
@@ -52,7 +52,7 @@ func Ingress(spec *logging.ClusterLogForwarderSpec, o Options) []Element {
 					TemplateStr:  RetagJournalLogs,
 				},
 				ConfLiteral{
-					Desc:         "Invoke kubernetes apiserver to get kunbernetes metadata",
+					Desc:         "Invoke kubernetes apiserver to get kubernetes metadata",
 					TemplateName: "kubernetesMetadata",
 					TemplateStr:  KubernetesMetadataPlugin,
 				},
@@ -61,11 +61,6 @@ func Ingress(spec *logging.ClusterLogForwarderSpec, o Options) []Element {
 					TemplateName: "parseJsonFields",
 					TemplateStr:  ParseJsonFields,
 				},
-				ConfLiteral{
-					Desc:         "Clean kibana log fields",
-					TemplateName: "cleanKibanaLogs",
-					TemplateStr:  CleanKibanaLogs,
-				},
 				ConfLiteral{
 					Desc:         "Fix level field in audit logs",
 					TemplateName: "fixAuditLevel",
@@ -198,7 +193,6 @@ const KubernetesMetadataPlugin string = `
   kubernetes_url 'https://kubernetes.default.svc'
   allow_orphans false
   cache_size '1000'
-  use_journal 'nil'
   ssl_partial_chain 'true'
 </filter>
 {{end}}
@@ -207,43 +201,14 @@ const KubernetesMetadataPlugin string = `
 const ParseJsonFields string = `
 {{define "parseJsonFields" -}}
 # {{.Desc}}
-<filter kubernetes.journal.**>
-  @type parse_json_field
-  merge_json_log 'false'
-  preserve_json_log 'true'
-  json_fields 'log,MESSAGE'
-</filter>
-
-<filter kubernetes.var.log.pods.**>
-  @type parse_json_field
-  merge_json_log 'false'
-  preserve_json_log 'true'
-  json_fields 'log,MESSAGE'
-</filter>
-
 <filter kubernetes.var.log.pods.**_eventrouter-**>
   @type parse_json_field
   merge_json_log true
   preserve_json_log true
-  json_fields 'log,MESSAGE'
-</filter>
-{{end}}
-`
-
-const CleanKibanaLogs string = `
-{{define "cleanKibanaLogs" -}}
-# {{.Desc}}
-<filter **kibana**>
-  @type record_transformer
-  enable_ruby
-  <record>
-    log ${record['err'] || record['msg'] || record['MESSAGE'] || record['log']}
-  </record>
-  remove_keys req,res,msg,name,level,v,pid,err
+  json_fields 'message'
 </filter>
 {{end}}
 `
-
 const FixAuditLevel string = `
 {{define "fixAuditLevel" -}}
 # {{.Desc}}
@@ -268,21 +233,12 @@ const ViaQDataModel string = `
 # {{.Desc}}
 <filter **>
   @type viaq_data_model
-  elasticsearch_index_prefix_field 'viaq_index_name'
+  enable_prune_empty_fields false
   default_keep_fields CEE,time,@timestamp,aushape,ci_job,collectd,docker,fedora-ci,file,foreman,geoip,hostname,ipaddr4,ipaddr6,kubernetes,level,message,namespace_name,namespace_uuid,offset,openstack,ovirt,pid,pipeline_metadata,rsyslog,service,systemd,tags,testcase,tlog,viaq_msg_id
-  extra_keep_fields ''
   keep_empty_fields 'message'
-  use_undefined false
-  undefined_name 'undefined'
   rename_time true
-  rename_time_if_missing false
-  src_time_name 'time'
-  dest_time_name '@timestamp'
   pipeline_type 'collector'
-  undefined_to_string 'false'
-  undefined_dot_replace_char 'UNUSED'
-  undefined_max_num_fields '-1'
-  process_kubernetes_events 'false'
+  process_kubernetes_events false
   <level>
     name warn
     match 'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"'
@@ -303,50 +259,22 @@ const ViaQDataModel string = `
     name debug
     match 'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"'
   </level>
-  <formatter>
-    tag "system.var.log**"
-    type sys_var_log
-    remove_keys host,pid,ident
-  </formatter>
   <formatter>
     tag "journal.system**"
     type sys_journal
     remove_keys log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID
   </formatter>
-  <formatter>
-    tag "kubernetes.journal.container**"
-    type k8s_journal
-    remove_keys 'log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID'
-  </formatter>
   <formatter>
     tag "kubernetes.var.log.pods.**_eventrouter-** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
     type k8s_json_file
-    remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+    remove_keys stream
     process_kubernetes_events 'true'
   </formatter>
   <formatter>
     tag "kubernetes.var.log.pods**"
     type k8s_json_file
-    remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
+    remove_keys stream
   </formatter>
-  <elasticsearch_index_name>
-    enabled 'true'
-    tag "kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** journal.system** system.var.log**"
-    name_type static
-    static_index_name infra-write
-  </elasticsearch_index_name>
-  <elasticsearch_index_name>
-    enabled 'true'
-    tag "linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
-    name_type static
-    static_index_name audit-write
-  </elasticsearch_index_name>
-  <elasticsearch_index_name>
-    enabled 'true'
-    tag "**"
-    name_type static
-    static_index_name app-write
-  </elasticsearch_index_name>
 </filter>
 {{end}}
 `
@@ -367,7 +295,7 @@ const ConcatLines string = `
 {{define "concatLines" -}}
 <filter kubernetes.**>
   @type concat
-  key log
+  key message
   partial_key logtag
   partial_value P
   separator ''
diff --git a/internal/generator/fluentd/output/elasticsearch/elasticsearch_test.go b/internal/generator/fluentd/output/elasticsearch/elasticsearch_test.go
index fe92846ede..68304df996 100644
--- a/internal/generator/fluentd/output/elasticsearch/elasticsearch_test.go
+++ b/internal/generator/fluentd/output/elasticsearch/elasticsearch_test.go
@@ -52,6 +52,10 @@ var _ = Describe("Generate fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
     elasticsearch_index_prefix_field 'viaq_index_name'
     <elasticsearch_index_name>
       enabled 'true'
@@ -207,6 +211,10 @@ var _ = Describe("Generate fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
     elasticsearch_index_prefix_field 'viaq_index_name'
     <elasticsearch_index_name>
       enabled 'true'
@@ -356,6 +364,10 @@ var _ = Describe("Generate fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
     elasticsearch_index_prefix_field 'viaq_index_name'
     <elasticsearch_index_name>
       enabled 'true'
@@ -494,6 +506,10 @@ var _ = Describe("Generate fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
     elasticsearch_index_prefix_field 'viaq_index_name'
     <elasticsearch_index_name>
       enabled 'true'
@@ -645,6 +661,10 @@ var _ = Describe("Generate fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
     elasticsearch_index_prefix_field 'viaq_index_name'
     <elasticsearch_index_name>
       enabled 'true'
@@ -794,6 +814,10 @@ var _ = Describe("Generate fluentd config", func() {
   # Viaq Data Model
   <filter **>
     @type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
     elasticsearch_index_prefix_field 'viaq_index_name'
     <elasticsearch_index_name>
       enabled 'true'
diff --git a/internal/generator/fluentd/output/elasticsearch/output_conf_es_test.go b/internal/generator/fluentd/output/elasticsearch/output_conf_es_test.go
index 8b16dabe1b..0818fb3c9b 100644
--- a/internal/generator/fluentd/output/elasticsearch/output_conf_es_test.go
+++ b/internal/generator/fluentd/output/elasticsearch/output_conf_es_test.go
@@ -102,6 +102,10 @@ var _ = Describe("Generating fluentd config blocks", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
@@ -266,6 +270,10 @@ var _ = Describe("Generating fluentd config blocks", func() {
   # Viaq Data Model
   <filter **>
 	@type viaq_data_model
+    enable_openshift_model false
+    enable_prune_empty_fields false
+    rename_time false
+    undefined_dot_replace_char UNUSED
 	elasticsearch_index_prefix_field 'viaq_index_name'
 	<elasticsearch_index_name>
 	  enabled 'true'
diff --git a/internal/generator/fluentd/output/elasticsearch/viaq_data_model.go b/internal/generator/fluentd/output/elasticsearch/viaq_data_model.go
index 27677db9d6..e36a2d98fc 100644
--- a/internal/generator/fluentd/output/elasticsearch/viaq_data_model.go
+++ b/internal/generator/fluentd/output/elasticsearch/viaq_data_model.go
@@ -57,6 +57,10 @@ func (im IndexModel) Template() string {
 # Viaq Data Model
 <filter **>
   @type viaq_data_model
+  enable_openshift_model false
+  enable_prune_empty_fields false
+  rename_time false
+  undefined_dot_replace_char UNUSED
   elasticsearch_index_prefix_field 'viaq_index_name'
   <elasticsearch_index_name>
     enabled 'true'
diff --git a/internal/generator/fluentd/output/output_fluentd_buffer_test.go b/internal/generator/fluentd/output/output_fluentd_buffer_test.go
index 0187050f9f..d14bb72af5 100644
--- a/internal/generator/fluentd/output/output_fluentd_buffer_test.go
+++ b/internal/generator/fluentd/output/output_fluentd_buffer_test.go
@@ -88,6 +88,10 @@ var _ = Describe("Generating fluentd config", func() {
 		  # Viaq Data Model
 		  <filter **>
 			@type viaq_data_model
+            enable_openshift_model false
+            enable_prune_empty_fields false
+            rename_time false
+            undefined_dot_replace_char UNUSED
 			elasticsearch_index_prefix_field 'viaq_index_name'
 			<elasticsearch_index_name>
 			  enabled 'true'
@@ -234,6 +238,10 @@ var _ = Describe("Generating fluentd config", func() {
 		  # Viaq Data Model
 		  <filter **>
 			@type viaq_data_model
+            enable_openshift_model false
+            enable_prune_empty_fields false
+            rename_time false
+            undefined_dot_replace_char UNUSED
 			elasticsearch_index_prefix_field 'viaq_index_name'
 			<elasticsearch_index_name>
 			  enabled 'true'
@@ -367,6 +375,10 @@ var _ = Describe("Generating fluentd config", func() {
 		  # Viaq Data Model
 		  <filter **>
 			@type viaq_data_model
+            enable_openshift_model false
+            enable_prune_empty_fields false
+            rename_time false
+            undefined_dot_replace_char UNUSED
 			elasticsearch_index_prefix_field 'viaq_index_name'
 			<elasticsearch_index_name>
 			  enabled 'true'
diff --git a/internal/generator/fluentd/prometheus.go b/internal/generator/fluentd/prometheus.go
index 1ac79c9d70..27cf884c43 100644
--- a/internal/generator/fluentd/prometheus.go
+++ b/internal/generator/fluentd/prometheus.go
@@ -1,10 +1,7 @@
 package fluentd
 
 import (
-	logging "github.com/openshift/cluster-logging-operator/apis/logging/v1"
 	"github.com/openshift/cluster-logging-operator/internal/generator"
-	"github.com/openshift/cluster-logging-operator/internal/generator/fluentd/elements"
-	"github.com/openshift/cluster-logging-operator/internal/generator/fluentd/helpers"
 )
 
 const PrometheusMonitorTemplate = `
@@ -50,83 +47,3 @@ const PrometheusMonitorTemplate = `
 `
 
 type PrometheusMonitor = generator.ConfLiteral
-
-func PrometheusMetrics(spec *logging.ClusterLogForwarderSpec, o generator.Options) []generator.Element {
-	return []generator.Element{
-		elements.Pipeline{
-			InLabel: helpers.LabelName("MEASURE"),
-			Desc:    "Increment Prometheus metrics",
-			SubElements: []generator.Element{
-				generator.ConfLiteral{
-					TemplateName: "EmitMetrics",
-					TemplateStr:  EmitMetrics,
-				},
-				elements.Match{
-					Desc:      "Journal Logs go to INGRESS pipeline",
-					MatchTags: "journal",
-					MatchElement: elements.Relabel{
-						OutLabel: helpers.LabelName("INGRESS"),
-					},
-				},
-				elements.Match{
-					Desc:      "Audit Logs go to INGRESS pipeline",
-					MatchTags: "*audit.log",
-					MatchElement: elements.Relabel{
-						OutLabel: helpers.LabelName("INGRESS"),
-					},
-				},
-				elements.Match{
-					Desc:      "Kubernetes Logs go to CONCAT pipeline",
-					MatchTags: "kubernetes.**",
-					MatchElement: elements.Relabel{
-						OutLabel: helpers.LabelName("CONCAT"),
-					},
-				},
-			},
-		},
-	}
-}
-
-const EmitMetrics string = `
-{{define "EmitMetrics" -}}
-<filter **>
-  @type record_transformer
-  enable_ruby
-  <record>
-    msg_size ${record.to_s.length}
-  </record>
-</filter>
-
-<filter **>
-  @type prometheus
-  <metric>
-    name cluster_logging_collector_input_record_total
-    type counter
-    desc The total number of incoming records
-    <labels>
-      tag ${tag}
-      hostname ${hostname}
-    </labels>
-  </metric>
-</filter>
-
-<filter **>
-  @type prometheus
-  <metric>
-    name cluster_logging_collector_input_record_bytes
-    type counter
-    desc The total bytes of incoming records
-    key msg_size
-    <labels>
-      tag ${tag}
-      hostname ${hostname}
-    </labels>
-  </metric>
-</filter>
-
-<filter **>
-  @type record_transformer
-  remove_keys msg_size
-</filter>
-{{end}}
-`
diff --git a/internal/generator/fluentd/source/audit_logs.go b/internal/generator/fluentd/source/audit_logs.go
index f3103b0f46..8cd5182a47 100644
--- a/internal/generator/fluentd/source/audit_logs.go
+++ b/internal/generator/fluentd/source/audit_logs.go
@@ -86,7 +86,7 @@ const OVNAuditLogTemplate = `
 <source>
   @type tail
   @id ovn-audit-input
-  @label @MEASURE
+  @label @{{.OutLabel}}
   path "/var/log/ovn/acl-audit-log.log"
   pos_file "/var/lib/fluentd/pos/acl-audit-log.pos"
   {{- .ReadLinesLimit }}
diff --git a/internal/generator/fluentd/source/container_logs.go b/internal/generator/fluentd/source/container_logs.go
index e4a4b857eb..dbdd0033aa 100644
--- a/internal/generator/fluentd/source/container_logs.go
+++ b/internal/generator/fluentd/source/container_logs.go
@@ -43,18 +43,10 @@ func (cl ContainerLogs) Template() string {
   skip_refresh_on_startup true
   @label @{{.OutLabel}}
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 {{end}}`
diff --git a/internal/generator/fluentd/sources.go b/internal/generator/fluentd/sources.go
index 86a356601a..13ea9f0552 100644
--- a/internal/generator/fluentd/sources.go
+++ b/internal/generator/fluentd/sources.go
@@ -37,7 +37,7 @@ func LogSources(spec *logging.ClusterLogForwarderSpec, tunings *logging.FluentdI
 		el = append(el,
 			source2.JournalLog{
 				Desc:         "Logs from linux journal",
-				OutLabel:     "MEASURE",
+				OutLabel:     "INGRESS",
 				TemplateName: "inputSourceJournalTemplate",
 				TemplateStr:  source2.JournalLogTemplate,
 			})
@@ -49,7 +49,7 @@ func LogSources(spec *logging.ClusterLogForwarderSpec, tunings *logging.FluentdI
 				Paths:        ContainerLogPaths(),
 				ExcludePaths: ExcludeContainerPaths(),
 				PosFile:      "/var/lib/fluentd/pos/es-containers.log.pos",
-				OutLabel:     "MEASURE",
+				OutLabel:     "CONCAT",
 				Tunings:      tunings,
 			})
 	}
@@ -58,7 +58,7 @@ func LogSources(spec *logging.ClusterLogForwarderSpec, tunings *logging.FluentdI
 			source2.AuditLog{
 				AuditLogLiteral: source2.AuditLogLiteral{
 					Desc:         "linux audit logs",
-					OutLabel:     "MEASURE",
+					OutLabel:     "INGRESS",
 					TemplateName: "inputSourceHostAuditTemplate",
 					TemplateStr:  source2.HostAuditLogTemplate,
 				},
@@ -67,7 +67,7 @@ func LogSources(spec *logging.ClusterLogForwarderSpec, tunings *logging.FluentdI
 			source2.AuditLog{
 				AuditLogLiteral: source2.AuditLogLiteral{
 					Desc:         "k8s audit logs",
-					OutLabel:     "MEASURE",
+					OutLabel:     "INGRESS",
 					TemplateName: "inputSourceK8sAuditTemplate",
 					TemplateStr:  source2.K8sAuditLogTemplate,
 				},
@@ -76,7 +76,7 @@ func LogSources(spec *logging.ClusterLogForwarderSpec, tunings *logging.FluentdI
 			source2.AuditLog{
 				AuditLogLiteral: source2.AuditLogLiteral{
 					Desc:         "Openshift audit logs",
-					OutLabel:     "MEASURE",
+					OutLabel:     "INGRESS",
 					TemplateName: "inputSourceOpenShiftAuditTemplate",
 					TemplateStr:  source2.OpenshiftAuditLogTemplate,
 				},
@@ -85,7 +85,7 @@ func LogSources(spec *logging.ClusterLogForwarderSpec, tunings *logging.FluentdI
 			source2.AuditLog{
 				AuditLogLiteral: source2.AuditLogLiteral{
 					Desc:         "Openshift Virtual Network (OVN) audit logs",
-					OutLabel:     "MEASURE",
+					OutLabel:     "INGRESS",
 					TemplateName: "inputSourceOVNAuditTemplate",
 					TemplateStr:  source2.OVNAuditLogTemplate,
 				},
@@ -97,7 +97,7 @@ func LogSources(spec *logging.ClusterLogForwarderSpec, tunings *logging.FluentdI
 }
 
 func ContainerLogPaths() string {
-	return fmt.Sprintf("%q", "/var/log/pods/**/*.log")
+	return fmt.Sprintf("%q", "/var/log/pods/*/*/*.log")
 }
 
 func ExcludeContainerPaths() string {
@@ -105,5 +105,7 @@ func ExcludeContainerPaths() string {
 	for _, comp := range []string{constants.CollectorName, constants.ElasticsearchName, constants.KibanaName} {
 		paths = append(paths, fmt.Sprintf("\"/var/log/pods/%s_%s-*/*/*.log\"", constants.OpenshiftNS, comp))
 	}
+	paths = append(paths, "\"/var/log/pods/*/*/*.gz\"", "\"/var/log/pods/*/*/*.tmp\"")
+
 	return fmt.Sprintf("[%s]", strings.Join(paths, ", "))
 }
diff --git a/internal/generator/fluentd/sources_test.go b/internal/generator/fluentd/sources_test.go
index 2a75d01010..5ed2edfc6c 100644
--- a/internal/generator/fluentd/sources_test.go
+++ b/internal/generator/fluentd/sources_test.go
@@ -35,28 +35,20 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
   tag kubernetes.*
   read_from_head "true"
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 `,
@@ -87,8 +79,8 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
@@ -96,20 +88,12 @@ var _ = Describe("Testing Config Generation", func() {
   read_from_head "true"
   read_lines_limit 1500
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 `,
@@ -131,7 +115,7 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type systemd
   @id systemd-input
-  @label @MEASURE
+  @label @INGRESS
   path '/var/log/journal'
   <storage>
     @type local
@@ -149,28 +133,20 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
   tag kubernetes.*
   read_from_head "true"
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 `,
@@ -192,7 +168,7 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/audit/audit.log"
   pos_file "/var/lib/fluentd/pos/audit.log.pos"
   tag linux-audit.log
@@ -205,7 +181,7 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id k8s-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/kube-apiserver/audit.log"
   pos_file "/var/lib/fluentd/pos/kube-apiserver.audit.log.pos"
   tag k8s-audit.log
@@ -222,7 +198,7 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id openshift-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path /var/log/oauth-apiserver/audit.log,/var/log/openshift-apiserver/audit.log
   pos_file /var/lib/fluentd/pos/oauth-apiserver.audit.log
   tag openshift-audit.log
@@ -239,7 +215,7 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id ovn-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/ovn/acl-audit-log.log"
   pos_file "/var/lib/fluentd/pos/acl-audit-log.pos"
   tag ovn-audit.log
@@ -278,7 +254,7 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/audit/audit.log"
   pos_file "/var/lib/fluentd/pos/audit.log.pos"
   read_lines_limit 1500
@@ -292,7 +268,7 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id k8s-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/kube-apiserver/audit.log"
   pos_file "/var/lib/fluentd/pos/kube-apiserver.audit.log.pos"
   read_lines_limit 1500
@@ -310,7 +286,7 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id openshift-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path /var/log/oauth-apiserver/audit.log,/var/log/openshift-apiserver/audit.log
   pos_file /var/lib/fluentd/pos/oauth-apiserver.audit.log
   read_lines_limit 1500
@@ -328,7 +304,7 @@ var _ = Describe("Testing Config Generation", func() {
 <source>
   @type tail
   @id ovn-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/ovn/acl-audit-log.log"
   pos_file "/var/lib/fluentd/pos/acl-audit-log.pos"
   read_lines_limit 1500
@@ -366,7 +342,7 @@ const AllSources = `
 <source>
   @type systemd
   @id systemd-input
-  @label @MEASURE
+  @label @INGRESS
   path '/var/log/journal'
   <storage>
     @type local
@@ -384,28 +360,20 @@ const AllSources = `
 <source>
   @type tail
   @id container-input
-  path "/var/log/pods/**/*.log"
-  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log"]
+  path "/var/log/pods/*/*/*.log"
+  exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
   pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
   refresh_interval 5
   rotate_wait 5
   tag kubernetes.*
   read_from_head "true"
   skip_refresh_on_startup true
-  @label @MEASURE
+  @label @CONCAT
   <parse>
-    @type multi_format
-    <pattern>
-      format json
-      time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
-      keep_time_key true
-    </pattern>
-    <pattern>
-      format regexp
-      expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
-      time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
-      keep_time_key true
-    </pattern>
+    @type regexp
+    expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
+    time_key '@timestamp'
+    keep_time_key true
   </parse>
 </source>
 
@@ -413,7 +381,7 @@ const AllSources = `
 <source>
   @type tail
   @id audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/audit/audit.log"
   pos_file "/var/lib/fluentd/pos/audit.log.pos"
   tag linux-audit.log
@@ -426,7 +394,7 @@ const AllSources = `
 <source>
   @type tail
   @id k8s-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/kube-apiserver/audit.log"
   pos_file "/var/lib/fluentd/pos/kube-apiserver.audit.log.pos"
   tag k8s-audit.log
@@ -443,7 +411,7 @@ const AllSources = `
 <source>
   @type tail
   @id openshift-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path /var/log/oauth-apiserver/audit.log,/var/log/openshift-apiserver/audit.log
   pos_file /var/lib/fluentd/pos/oauth-apiserver.audit.log
   tag openshift-audit.log
@@ -460,7 +428,7 @@ const AllSources = `
 <source>
   @type tail
   @id ovn-audit-input
-  @label @MEASURE
+  @label @INGRESS
   path "/var/log/ovn/acl-audit-log.log"
   pos_file "/var/lib/fluentd/pos/acl-audit-log.pos"
   tag ovn-audit.log
diff --git a/internal/generator/vector/sources.go b/internal/generator/vector/sources.go
index fff10617d6..7c251a573b 100644
--- a/internal/generator/vector/sources.go
+++ b/internal/generator/vector/sources.go
@@ -63,7 +63,7 @@ func LogSources(spec *logging.ClusterLogForwarderSpec, op generator.Options) []g
 }
 
 func ContainerLogPaths() string {
-	return fmt.Sprintf("%q", "/var/log/pods/**/*.log")
+	return fmt.Sprintf("%q", "/var/log/pods/*/*/*.log")
 }
 
 func CollectorLogsPath() string {
diff --git a/internal/k8shandler/fluentd.go b/internal/k8shandler/fluentd.go
index 5ae986954d..e19578a0f8 100644
--- a/internal/k8shandler/fluentd.go
+++ b/internal/k8shandler/fluentd.go
@@ -119,6 +119,7 @@ func newFluentdPodSpec(cluster *logging.ClusterLogging, trustedCABundleCM *v1.Co
 		{Name: "K8S_NODE_NAME", ValueFrom: &v1.EnvVarSource{FieldRef: &v1.ObjectFieldSelector{APIVersion: "v1", FieldPath: "spec.nodeName"}}},
 		{Name: "NODE_IPV4", ValueFrom: &v1.EnvVarSource{FieldRef: &v1.ObjectFieldSelector{APIVersion: "v1", FieldPath: "status.hostIP"}}},
 		{Name: "POD_IP", ValueFrom: &v1.EnvVarSource{FieldRef: &v1.ObjectFieldSelector{APIVersion: "v1", FieldPath: "status.podIP"}}},
+		{Name: "RUBY_GC_HEAP_OLDOBJECT_LIMIT_FACTOR", Value: "0.9"},
 	}
 
 	if cluster.Spec.Forwarder != nil {
diff --git a/test/e2e/logforwarding/fluent/fluent_secure_test.go b/test/e2e/logforwarding/fluent/fluent_secure_test.go
index 792717e60f..98dd48a411 100644
--- a/test/e2e/logforwarding/fluent/fluent_secure_test.go
+++ b/test/e2e/logforwarding/fluent/fluent_secure_test.go
@@ -112,7 +112,7 @@ var _ = Describe("[ClusterLogForwarder]", func() {
 				for i := 0; i < 10; {
 					l, err := r.ReadLine()
 					ExpectOK(err)
-					Expect(l).To(ContainSubstring(`"viaq_index_name":"app`)) // Only app logs
+					Expect(l).To(ContainSubstring(`"log_type":"application`)) // Only app logs
 					if strings.Contains(l, secureMessage) {
 						i++ // Count our own app messages, ignore others.
 					}
@@ -177,7 +177,7 @@ var _ = Describe("[ClusterLogForwarder]", func() {
 		for i := 0; i < 10; {
 			l, err := r.ReadLine()
 			ExpectOK(err)
-			Expect(l).To(ContainSubstring(`"viaq_index_name":"app`)) // Only app logs
+			Expect(l).To(ContainSubstring(`"log_type":"application`)) // Only app logs
 			if strings.Contains(l, secureMessage) {
 				i++ // Count our own app messages, ignore others.
 			}
diff --git a/test/e2e/logforwarding/fluent/fluent_test.go b/test/e2e/logforwarding/fluent/fluent_test.go
index 610804b677..a0a729b7e8 100644
--- a/test/e2e/logforwarding/fluent/fluent_test.go
+++ b/test/e2e/logforwarding/fluent/fluent_test.go
@@ -42,7 +42,7 @@ var _ = Describe("[ClusterLogForwarder]", func() {
 			for i := 0; i < 10; {
 				l, err := r.ReadLine()
 				ExpectOK(err)
-				Expect(l).To(ContainSubstring(`"viaq_index_name":"app`)) // Only app logs
+				Expect(l).To(ContainSubstring(`"log_type":"application`)) // Only app logs
 				if strings.Contains(l, message) {
 					i++ // Count our own app messages, ignore others.
 				}
@@ -59,7 +59,7 @@ var _ = Describe("[ClusterLogForwarder]", func() {
 			r := f.Receiver.Sources["infrastructure"].TailReader()
 			l, err := r.ReadLine()
 			ExpectOK(err)
-			Expect(l).To(ContainSubstring(`"viaq_index_name":"inf`)) // Only infra logs
+			Expect(l).To(ContainSubstring(`"log_type":"infrastructure`)) // Only infra logs
 		})
 
 		It("forwards different types to different outputs with labels", func() {
diff --git a/test/framework/functional/message_templates.go b/test/framework/functional/message_templates.go
index 0810512bcd..1fa3072020 100644
--- a/test/framework/functional/message_templates.go
+++ b/test/framework/functional/message_templates.go
@@ -49,13 +49,12 @@ var (
 
 func NewApplicationLogTemplate() types.ApplicationLog {
 	return types.ApplicationLog{
-		Timestamp:     time.Time{},
-		Message:       "*",
-		LogType:       "application",
-		ViaqIndexName: "app-write",
-		Level:         "*",
-		Hostname:      "*",
-		ViaqMsgID:     "*",
+		Timestamp: time.Time{},
+		Message:   "*",
+		LogType:   "application",
+		Level:     "*",
+		Hostname:  "*",
+		ViaqMsgID: "*",
 		Openshift: types.OpenshiftMeta{
 			Labels:   map[string]string{"*": "*"},
 			Sequence: types.NewOptionalInt(""),
diff --git a/test/functional/normalization/eventrouter_test.go b/test/functional/normalization/eventrouter_test.go
index 042a6b13ff..1d3ee0b637 100644
--- a/test/functional/normalization/eventrouter_test.go
+++ b/test/functional/normalization/eventrouter_test.go
@@ -70,7 +70,6 @@ var _ = Describe("[Functional][Normalization] Fluentd normalization for EventRou
 				PipelineMetadata: templateForAnyCollector,
 				Timestamp:        timestamp,
 				LogType:          "application",
-				ViaqIndexName:    "app-write",
 				ViaqMsgID:        "*",
 				OpenshiftLabels:  types.OpenshiftMeta{},
 			}
diff --git a/test/functional/normalization/message_format_test.go b/test/functional/normalization/message_format_test.go
index d58974ad5a..e391ec4c4c 100644
--- a/test/functional/normalization/message_format_test.go
+++ b/test/functional/normalization/message_format_test.go
@@ -44,7 +44,6 @@ var _ = Describe("[Functional][LogForwarding][Normalization] tests for message f
 				Kind:             "Event",
 				Hostname:         functional.FunctionalNodeName,
 				LogType:          "audit",
-				ViaqIndexName:    "audit-write",
 				Level:            "debug",
 				Timestamp:        time.Time{},
 				ViaqMsgID:        "*",
@@ -73,10 +72,9 @@ var _ = Describe("[Functional][LogForwarding][Normalization] tests for message f
 		auditLogLine := functional.NewAuditHostLog(testTime)
 		// Template expected as output Log
 		var outputLogTemplate = types.LinuxAuditLog{
-			Message:       auditLogLine,
-			LogType:       "audit",
-			Hostname:      functional.FunctionalNodeName,
-			ViaqIndexName: "audit-write",
+			Message:  auditLogLine,
+			LogType:  "audit",
+			Hostname: functional.FunctionalNodeName,
 			AuditLinux: types.AuditLinux{
 				Type:     "DAEMON_START",
 				RecordID: "*",
diff --git a/test/functional/outputs/forward_to_elasticsearch_test.go b/test/functional/outputs/forward_to_elasticsearch_test.go
index 9218cc9d1a..0c3ef45c82 100644
--- a/test/functional/outputs/forward_to_elasticsearch_test.go
+++ b/test/functional/outputs/forward_to_elasticsearch_test.go
@@ -19,7 +19,7 @@ var _ = Describe("[Functional][Outputs][ElasticSearch] FluentdForward Output to
 	)
 
 	BeforeEach(func() {
-
+		outputLogTemplate.ViaqIndexName = "app-write"
 		framework = functional.NewCollectorFunctionalFramework()
 		functional.NewClusterLogForwarderBuilder(framework.Forwarder).
 			FromInput(logging.InputNameApplication).