Capture stdout/stderr of spawned components

.gitignore

@@ -45,7 +45,6 @@ fleet.enc.lock
 # Files generated with the bump version automations
 *.bck
 
-
 # agent
 build/
 elastic-agent

NOTICE.txt

@@ -1273,11 +1273,11 @@ SOFTWARE
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/elastic-agent-libs
-Version: v0.2.6
+Version: v0.2.15
 Licence type (autodetected): Apache-2.0
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/[email protected].6/LICENSE:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/[email protected].15/LICENSE:
 
                                  Apache License
                            Version 2.0, January 2004

changelog/fragments/1669236059-Capture-stdout-stderr-of-all-spawned-components-to-simplify-logging.yaml

@@ -0,0 +1,31 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Capture stdout/stderr of all spawned components to simplify logging
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+#description:
+
+# Affected component; a word indicating the component this changeset affects.
+component:
+
+# PR number; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: 1702
+
+# Issue number; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: 221

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/elastic/e2e-testing v1.99.2-0.20220117192005-d3365c99b9c4
 	github.com/elastic/elastic-agent-autodiscover v0.2.1
 	github.com/elastic/elastic-agent-client/v7 v7.0.0-20220804181728-b0328d2fe484
-	github.com/elastic/elastic-agent-libs v0.2.6
+	github.com/elastic/elastic-agent-libs v0.2.15
 	github.com/elastic/elastic-agent-system-metrics v0.4.4
 	github.com/elastic/go-licenser v0.4.0
 	github.com/elastic/go-sysinfo v1.8.1

go.sum

@@ -387,8 +387,8 @@ github.com/elastic/elastic-agent-autodiscover v0.2.1/go.mod h1:gPnzzfdYNdgznAb+i
 github.com/elastic/elastic-agent-client/v7 v7.0.0-20220804181728-b0328d2fe484 h1:uJIMfLgCenJvxsVmEjBjYGxt0JddCgw2IxgoNfcIXOk=
 github.com/elastic/elastic-agent-client/v7 v7.0.0-20220804181728-b0328d2fe484/go.mod h1:fkvyUfFwyAG5OnMF0h+FV9sC0Xn9YLITwQpSuwungQs=
 github.com/elastic/elastic-agent-libs v0.2.5/go.mod h1:chO3rtcLyGlKi9S0iGVZhYCzDfdDsAQYBc+ui588AFE=
-github.com/elastic/elastic-agent-libs v0.2.6 h1:DpcUcCVYZ7lNtHLUlyT1u/GtGAh49wpL15DTH7+8O5o=
-github.com/elastic/elastic-agent-libs v0.2.6/go.mod h1:chO3rtcLyGlKi9S0iGVZhYCzDfdDsAQYBc+ui588AFE=
+github.com/elastic/elastic-agent-libs v0.2.15 h1:hdAbrZZ2mCPcQLRCE3E8xw3mHKl8HFMt36w7jan/XGo=
+github.com/elastic/elastic-agent-libs v0.2.15/go.mod h1:0J9lzJh+BjttIiVjYDLncKYCEWUUHiiqnuI64y6C6ss=
 github.com/elastic/elastic-agent-system-metrics v0.4.4 h1:Br3S+TlBhijrLysOvbHscFhgQ00X/trDT5VEnOau0E0=
 github.com/elastic/elastic-agent-system-metrics v0.4.4/go.mod h1:tF/f9Off38nfzTZHIVQ++FkXrDm9keFhFpJ+3pQ00iI=
 github.com/elastic/elastic-package v0.32.1/go.mod h1:l1fEnF52XRBL6a5h6uAemtdViz2bjtjUtgdQcuRhEAY=

internal/pkg/agent/application/monitoring/v1_monitor.go

@@ -58,7 +58,7 @@ var (
 	supportedBeatsComponents = []string{"filebeat", "metricbeat", "apm-server", "fleet-server", "auditbeat", "cloudbeat", "heartbeat", "osquerybeat", "packetbeat"}
 )
 
-// Beats monitor is providing V1 monitoring support.
+// BeatsMonitor is providing V1 monitoring support for metrics and logs for endpoint-security only.
 type BeatsMonitor struct {
 	enabled         bool // feature flag disabling whole v1 monitoring story
 	config          *monitoringConfig
@@ -178,21 +178,10 @@ func (b *BeatsMonitor) EnrichArgs(unit, binary string, args []string) []string {
 		}
 	}
 
-	loggingPath := loggingPath(unit, b.operatingSystem)
-	if loggingPath != "" {
+	if !b.config.C.LogMetrics {
 		appendix = append(appendix,
-			"-E", "logging.files.path="+filepath.Dir(loggingPath),
-			"-E", "logging.files.name="+filepath.Base(loggingPath),
-			"-E", "logging.files.keepfiles=7",
-			"-E", "logging.files.permission=0640",
-			"-E", "logging.files.interval=1h",
+			"-E", "logging.metrics.enabled=false",
 		)
-
-		if !b.config.C.LogMetrics {
-			appendix = append(appendix,
-				"-E", "logging.metrics.enabled=false",
-			)
-		}
 	}
 
 	return append(args, appendix...)
@@ -291,24 +280,21 @@ func (b *BeatsMonitor) injectMonitoringOutput(source, dest map[string]interface{
 
 func (b *BeatsMonitor) injectLogsInput(cfg map[string]interface{}, componentIDToBinary map[string]string, monitoringOutput string) error {
 	monitoringNamespace := b.monitoringNamespace()
-	//fixedAgentName := strings.ReplaceAll(agentName, "-", "_")
 	logsDrop := filepath.Dir(loggingPath("unit", b.operatingSystem))
 
 	streams := []interface{}{
 		map[string]interface{}{
-			idKey: "filestream-monitoring-agent",
-			// "data_stream" is not used when creating an Input on Filebeat
-			"data_stream": map[string]interface{}{
-				"type":      "filestream",
-				"dataset":   "elastic_agent",
-				"namespace": monitoringNamespace,
-			},
+			idKey:  "filestream-monitoring-agent",
 			"type": "filestream",
 			"paths": []interface{}{
 				filepath.Join(logsDrop, agentName+"-*.ndjson"),
 				filepath.Join(logsDrop, agentName+"-watcher-*.ndjson"),
 			},
-			"index": fmt.Sprintf("logs-elastic_agent-%s", monitoringNamespace),
+			"data_stream": map[string]interface{}{
+				"type":      "logs",
+				"dataset":   "elastic_agent",
+				"namespace": monitoringNamespace,
+			},
 			"close": map[string]interface{}{
 				"on_state_change": map[string]interface{}{
 					"inactive": "5m",
@@ -325,133 +311,86 @@ func (b *BeatsMonitor) injectLogsInput(cfg map[string]interface{}, componentIDTo
 				},
 			},
 			"processors": []interface{}{
+				// copy original dataset so we can drop the dataset field
 				map[string]interface{}{
-					"add_fields": map[string]interface{}{
-						"target": "data_stream",
-						"fields": map[string]interface{}{
-							"type":      "logs",
-							"dataset":   "elastic_agent",
-							"namespace": monitoringNamespace,
-						},
-					},
-				},
-				map[string]interface{}{
-					"add_fields": map[string]interface{}{
-						"target": "event",
-						"fields": map[string]interface{}{
-							"dataset": "elastic_agent",
-						},
-					},
-				},
-				map[string]interface{}{
-					"add_fields": map[string]interface{}{
-						"target": "elastic_agent",
-						"fields": map[string]interface{}{
-							"id":       b.agentInfo.AgentID(),
-							"version":  b.agentInfo.Version(),
-							"snapshot": b.agentInfo.Snapshot(),
-						},
-					},
-				},
-				map[string]interface{}{
-					"add_fields": map[string]interface{}{
-						"target": "agent",
-						"fields": map[string]interface{}{
-							"id": b.agentInfo.AgentID(),
+					"copy_fields": map[string]interface{}{
+						"fields": []interface{}{
+							map[string]interface{}{
+								"from": "data_stream.dataset",
+								"to":   "data_stream.dataset_original",
+							},
 						},
 					},
 				},
+				// drop the dataset field so following copy_field can copy to it
 				map[string]interface{}{
 					"drop_fields": map[string]interface{}{
 						"fields": []interface{}{
-							"ecs.version", //coming from logger, already added by libbeat
+							"data_stream.dataset",
 						},
-						"ignore_missing": true,
-					},
-				}},
-		},
-	}
-	for unit, binaryName := range componentIDToBinary {
-		if !isSupportedBinary(binaryName) {
-			continue
-		}
-
-		fixedBinaryName := strings.ReplaceAll(binaryName, "-", "_")
-		name := strings.ReplaceAll(unit, "-", "_") // conform with index naming policy
-		logFile := loggingPath(unit, b.operatingSystem)
-		streams = append(streams, map[string]interface{}{
-			idKey: "filestream-monitoring-" + name,
-			"data_stream": map[string]interface{}{
-				// "data_stream" is not used when creating an Input on Filebeat
-				"type":      "filestream",
-				"dataset":   fmt.Sprintf("elastic_agent.%s", fixedBinaryName),
-				"namespace": monitoringNamespace,
-			},
-			"type":  "filestream",
-			"index": fmt.Sprintf("logs-elastic_agent.%s-%s", fixedBinaryName, monitoringNamespace),
-			"paths": []interface{}{logFile, logFile + "*"},
-			"close": map[string]interface{}{
-				"on_state_change": map[string]interface{}{
-					"inactive": "5m",
-				},
-			},
-			"parsers": []interface{}{
-				map[string]interface{}{
-					"ndjson": map[string]interface{}{
-						"message_key":    "message",
-						"overwrite_keys": true,
-						"add_error_key":  true,
-						"target":         "",
 					},
 				},
-			},
-			"processors": []interface{}{
+				// copy component.dataset as the real dataset
 				map[string]interface{}{
-					"add_fields": map[string]interface{}{
-						"target": "data_stream",
-						"fields": map[string]interface{}{
-							"type":      "logs",
-							"dataset":   fmt.Sprintf("elastic_agent.%s", fixedBinaryName),
-							"namespace": monitoringNamespace,
+					"copy_fields": map[string]interface{}{
+						"fields": []interface{}{
+							map[string]interface{}{
+								"from": "component.dataset",
+								"to":   "data_stream.dataset",
+							},
 						},
+						"fail_on_error":  false,
+						"ignore_missing": true,
 					},
 				},
+				// possible it's a log message from agent itself (doesn't have component.dataset)
 				map[string]interface{}{
-					"add_fields": map[string]interface{}{
-						"target": "event",
-						"fields": map[string]interface{}{
-							"dataset": fmt.Sprintf("elastic_agent.%s", fixedBinaryName),
+					"copy_fields": map[string]interface{}{
+						"fields": []interface{}{
+							map[string]interface{}{
+								"from": "data_stream.dataset_original",
+								"to":   "data_stream.dataset",
+							},
 						},
+						"fail_on_error": false,
 					},
 				},
+				// drop the original dataset copied and the event.dataset (as it will be updated)
 				map[string]interface{}{
-					"add_fields": map[string]interface{}{
-						"target": "elastic_agent",
-						"fields": map[string]interface{}{
-							"id":       b.agentInfo.AgentID(),
-							"version":  b.agentInfo.Version(),
-							"snapshot": b.agentInfo.Snapshot(),
+					"drop_fields": map[string]interface{}{
+						"fields": []interface{}{
+							"data_stream.dataset_original",
+							"event.dataset",
 						},
 					},
 				},
+				// update event.dataset with the now used data_stream.dataset
 				map[string]interface{}{
-					"add_fields": map[string]interface{}{
-						"target": "agent",
-						"fields": map[string]interface{}{
-							"id": b.agentInfo.AgentID(),
+					"copy_fields": map[string]interface{}{
+						"fields": []interface{}{
+							map[string]interface{}{
+								"from": "data_stream.dataset",
+								"to":   "event.dataset",
+							},
 						},
 					},
 				},
+				// coming from logger, added by agent (drop)
 				map[string]interface{}{
 					"drop_fields": map[string]interface{}{
 						"fields": []interface{}{
-							"ecs.version", //coming from logger, already added by libbeat
+							"ecs.version",
 						},
 						"ignore_missing": true,
 					},
 				},
-			},
-		})
+				// adjust destination data_stream based on the data_stream fields
+				map[string]interface{}{
+					"add_formatted_index": map[string]interface{}{
+						"index": "%{[data_stream.type]}-%{[data_stream.dataset]}-%{[data_stream.namespace]}",
+					},
+				}},
+		},
 	}
 
 	inputs := []interface{}{
@@ -460,10 +399,7 @@ func (b *BeatsMonitor) injectLogsInput(cfg map[string]interface{}, componentIDTo
 			"name":       "filestream-monitoring-agent",
 			"type":       "filestream",
 			useOutputKey: monitoringOutput,
-			"data_stream": map[string]interface{}{
-				"namespace": monitoringNamespace,
-			},
-			"streams": streams,
+			"streams":    streams,
 		},
 	}
 	inputsNode, found := cfg[inputsKey]