LOG-5931: keep stream field (stdout,stderr) in log event

Signed-off-by: Vitalii Parfonov <[email protected]>
openshift · Aug 8, 2024 · b8e813b · b8e813b
1 parent 30ec01e
commit b8e813b
Show file tree

Hide file tree

Showing 16 changed files with 29 additions and 14 deletions.
diff --git a/internal/generator/vector/conf/complex.toml b/internal/generator/vector/conf/complex.toml
@@ -269,7 +269,7 @@ if .log_source == "container" {
   del(._partial)
   del(.file)
   del(.source_type)
-  del(.stream)
+  .kubernetes.stream = del(.stream)
   del(.kubernetes.pod_ips)
   del(.kubernetes.node_labels)
   del(.timestamp_end)

diff --git a/internal/generator/vector/conf/complex_http_receiver.toml b/internal/generator/vector/conf/complex_http_receiver.toml
@@ -303,7 +303,7 @@ if .log_source == "container" {
   del(._partial)
   del(.file)
   del(.source_type)
-  del(.stream)
+  .kubernetes.stream = del(.stream)
   del(.kubernetes.pod_ips)
   del(.kubernetes.node_labels)
   del(.timestamp_end)

diff --git a/internal/generator/vector/conf/container.toml b/internal/generator/vector/conf/container.toml
@@ -111,7 +111,7 @@ if .log_source == "container" {
   del(._partial)
   del(.file)
   del(.source_type)
-  del(.stream)
+  .kubernetes.stream = del(.stream)
   del(.kubernetes.pod_ips)
   del(.kubernetes.node_labels)
   del(.timestamp_end)

diff --git a/internal/generator/vector/filter/multilineexception/filter.go b/internal/generator/vector/filter/multilineexception/filter.go
@@ -27,7 +27,7 @@ func (d DetectExceptions) Template() string {
 type = "detect_exceptions"
 inputs = {{.Inputs}}
 languages = ["All"]
-group_by = ["kubernetes.namespace_name","kubernetes.pod_name","kubernetes.container_name", "kubernetes.pod_id"]
+group_by = ["kubernetes.namespace_name","kubernetes.pod_name","kubernetes.container_name", "kubernetes.pod_id", "stream"]
 expire_after_ms = 2000
 multiline_flush_interval_ms = 1000
 {{end}}`

diff --git a/internal/generator/vector/filter/openshift/viaq/filter.go b/internal/generator/vector/filter/openshift/viaq/filter.go
@@ -43,7 +43,7 @@ if .log_source == "%s" {
 		RemovePartial,
 		RemoveFile,
 		RemoveSourceType,
-		RemoveStream,
+		HandleStream,
 		RemovePodIPs,
 		RemoveNodeLabels,
 		RemoveTimestampEnd,

diff --git a/internal/generator/vector/filter/openshift/viaq/normalize.go b/internal/generator/vector/filter/openshift/viaq/normalize.go
@@ -50,7 +50,7 @@ if starts_with(pod_name, "eventrouter-") {
   }
 }
 `
-	RemoveStream       = `del(.stream)`
+	HandleStream       = `.kubernetes.stream = del(.stream)`
 	RemovePodIPs       = `del(.kubernetes.pod_ips)`
 	RemoveNodeLabels   = `del(.kubernetes.node_labels)`
 	RemoveTimestampEnd = `del(.timestamp_end)`

diff --git a/internal/generator/vector/output/otlp/group_by.go b/internal/generator/vector/output/otlp/group_by.go
@@ -43,7 +43,7 @@ func GroupByContainer(id string, inputs []string) Element {
 		Inputs:      helpers.MakeInputs(inputs...),
 		MaxEvents:   "250",
 		GroupBy: MakeGroupBys(".openshift.cluster_id",
-			".kubernetes.namespace_name", ".kubernetes.pod_name", ".kubernetes.container_name"),
+			".kubernetes.namespace_name", ".kubernetes.pod_name", ".kubernetes.container_name", ".kubernetes.stream"),
 	}
 }
 

diff --git a/internal/generator/vector/output/otlp/otlp_all.toml b/internal/generator/vector/output/otlp/otlp_all.toml
@@ -24,7 +24,8 @@ source = '''
   resource.attributes = append( resource.attributes,
       [{"key": "k8s.pod.name", "value": {"stringValue": get!(.,["kubernetes","pod_name"])}},
       {"key": "k8s.container.name", "value": {"stringValue": get!(.,["kubernetes","container_name"])}},
-      {"key": "k8s.namespace.name", "value": {"stringValue": get!(.,["kubernetes","namespace_name"])}}]
+      {"key": "k8s.namespace.name", "value": {"stringValue": get!(.,["kubernetes","namespace_name"])}},
+      {"key": "k8s.stream", "value": {"stringValue": get!(.,["kubernetes","stream"])}}]
   )
   # Create logRecord object
   r = {}

diff --git a/internal/generator/vector/output/otlp/transform.go b/internal/generator/vector/output/otlp/transform.go
@@ -28,7 +28,8 @@ resource.attributes = append( resource.attributes,
 resource.attributes = append( resource.attributes,
     [{"key": "k8s.pod.name", "value": {"stringValue": get!(.,["kubernetes","pod_name"])}},
     {"key": "k8s.container.name", "value": {"stringValue": get!(.,["kubernetes","container_name"])}},
-    {"key": "k8s.namespace.name", "value": {"stringValue": get!(.,["kubernetes","namespace_name"])}}]
+    {"key": "k8s.namespace.name", "value": {"stringValue": get!(.,["kubernetes","namespace_name"])}},
+    {"key": "k8s.stream", "value": {"stringValue": get!(.,["kubernetes","stream"])}}]
 )
 `
 	LogRecord = `

diff --git a/internal/generator/vector/pipeline/adapter_test_drop_filter.toml b/internal/generator/vector/pipeline/adapter_test_drop_filter.toml
@@ -57,7 +57,7 @@ if .log_source == "container" {
   del(._partial)
   del(.file)
   del(.source_type)
-  del(.stream)
+  .kubernetes.stream = del(.stream)
   del(.kubernetes.pod_ips)
   del(.kubernetes.node_labels)
   del(.timestamp_end)

diff --git a/internal/generator/vector/pipeline/adapter_test_prune_inNotIn_filter.toml b/internal/generator/vector/pipeline/adapter_test_prune_inNotIn_filter.toml
@@ -49,7 +49,7 @@ if .log_source == "container" {
   del(._partial)
   del(.file)
   del(.source_type)
-  del(.stream)
+  .kubernetes.stream = del(.stream)
   del(.kubernetes.pod_ips)
   del(.kubernetes.node_labels)
   del(.timestamp_end)

diff --git a/internal/generator/vector/pipeline/adapter_test_prune_inOnly_filter.toml b/internal/generator/vector/pipeline/adapter_test_prune_inOnly_filter.toml
@@ -49,7 +49,7 @@ if .log_source == "container" {
   del(._partial)
   del(.file)
   del(.source_type)
-  del(.stream)
+  .kubernetes.stream = del(.stream)
   del(.kubernetes.pod_ips)
   del(.kubernetes.node_labels)
   del(.timestamp_end)

diff --git a/internal/generator/vector/pipeline/adapter_test_prune_notIn_only_filter.toml b/internal/generator/vector/pipeline/adapter_test_prune_notIn_only_filter.toml
@@ -49,7 +49,7 @@ if .log_source == "container" {
   del(._partial)
   del(.file)
   del(.source_type)
-  del(.stream)
+  .kubernetes.stream = del(.stream)
   del(.kubernetes.pod_ips)
   del(.kubernetes.node_labels)
   del(.timestamp_end)

diff --git a/test/framework/functional/message_templates.go b/test/framework/functional/message_templates.go
@@ -31,6 +31,7 @@ var (
 		FlatLabels:       []string{"*"},
 		NamespaceLabels:  map[string]string{"*": "*"},
 		Annotations:      map[string]string{"*": "*"},
+		Stream:           "stdout",
 	}
 	templateForInfraKubernetes = types.Kubernetes{
 		ContainerID:       "**optional**",

diff --git a/test/functional/normalization/message_format_test.go b/test/functional/normalization/message_format_test.go
@@ -77,15 +77,23 @@ var _ = Describe("[Functional][LogForwarding][Normalization] tests for message f
 		outputLogTemplate.Message = fmt.Sprintf("regex:^%s.*$", message)
 		outputLogTemplate.Level = "*"
 
-		// Write log line as input
+		// Write log line as stdout
 		applicationLogLine := fmt.Sprintf("%s stdout F %s $n", timestamp, message)
 		Expect(framework.WriteMessagesToApplicationLog(applicationLogLine, 1)).To(BeNil())
 
+		// Write log line as stderr
+		applicationLogLine = fmt.Sprintf("%s stderr F %s $n", timestamp, message)
+		Expect(framework.WriteMessagesToApplicationLog(applicationLogLine, 1)).To(BeNil())
+
 		logs, err := framework.ReadApplicationLogsFrom(string(obs.OutputTypeElasticsearch))
 		Expect(err).To(BeNil(), "Expected no errors reading the logs")
 		// Compare to expected template
 		outputTestLog := logs[0]
 		Expect(outputTestLog).To(FitLogFormatTemplate(outputLogTemplate))
+
+		outputLogTemplate.Kubernetes.Stream = "stderr"
+		outputTestLog = logs[1]
+		Expect(outputTestLog).To(FitLogFormatTemplate(outputLogTemplate))
 	})
 
 })
diff --git a/test/helpers/types/types.go b/test/helpers/types/types.go
@@ -116,6 +116,10 @@ type Kubernetes struct {
 	// NamespaceLabels are the labels present on the pod namespace
 	// +optional
 	NamespaceLabels map[string]string `json:"namespace_labels,omitempty"`
+
+	// The name of the stream the log line was submitted to (e.g.: stdout, stderr)
+	// +optional
+	Stream string `json:"stream,omitempty"`
 }
 
 type Collector struct {