otel: support system/metrics input with experimental otel runtime (#8148) (#8327)

* otel: support system/metrics input with experimental otel runtime

* use EventuallyWithTf

* add tests comparing final documents

* trim test

* remove TestSystemMetricsInput

* remove unused vars

* ignore host.network.ingress fields

* fix typo in expectation message

* don't compare system metrics directly, only compare the keys are present

* ignore host.network.* fields

* remove dot from prefix

* ignore network activity fields

* ignore flaky fields from network

* stop ignoring elastic_agent.{snapshot,version}

* reestructure test to use a single config

* Revert "reestructure test to use a single config"

Windows throws errors about named pipes. I suspect system/metrics input
tries to register monitoring on the same socket.

This reverts commit aa6b4c6.

* remove agent.version from ignore list

* reorder stripNondeterminism call

* print documents in case of failure

* compare host-wide and normal network events separately

* simplify code

* Add Sudo to define.Require

* match for host metrics

* filter network interface events

* use tt.metricset, add note about network events

* add metricsets to template

* Update testing/integration/beat_receivers_test.go



* use AssertMapstrEqual

---------


(cherry picked from commit 5648918)

Co-authored-by: Mauri de Souza Meneguzzo <[email protected]>
Co-authored-by: Mikołaj Świątek <[email protected]>

Author: 3 people

internal/pkg/otel/configtranslate/otelconfig.go

@@ -39,7 +39,7 @@ type exporterConfigTranslationFunc func(*config.C) (map[string]any, error)
 
 var (
 	OtelSupportedOutputTypes         = []string{"elasticsearch"}
-	OtelSupportedInputTypes          = []string{"filestream", "http/metrics", "beat/metrics"}
+	OtelSupportedInputTypes          = []string{"filestream", "http/metrics", "beat/metrics", "system/metrics"}
 	configTranslationFuncForExporter = map[otelcomponent.Type]exporterConfigTranslationFunc{
 		otelcomponent.MustNewType("elasticsearch"): translateEsOutputToExporter,
 	}

internal/pkg/otel/configtranslate/otelconfig_test.go

@@ -179,6 +179,33 @@ func TestGetOtelConfig(t *testing.T) {
 			},
 		},
 	}
+	systemMetricsConfig := map[string]any{
+		"id":         "test",
+		"use_output": "default",
+		"type":       "system/metrics",
+		"streams": []any{
+			map[string]any{
+				"id": "test-1",
+				"data_stream": map[string]any{
+					"dataset": "generic-1",
+				},
+				"metricsets": map[string]any{
+					"cpu": map[string]any{
+						"data_stream.dataset": "system.cpu",
+					},
+					"memory": map[string]any{
+						"data_stream.dataset": "system.memory",
+					},
+					"network": map[string]any{
+						"data_stream.dataset": "system.network",
+					},
+					"filesystem": map[string]any{
+						"data_stream.dataset": "system.filesystem",
+					},
+				},
+			},
+		},
+	}
 	esOutputConfig := map[string]any{
 		"type":             "elasticsearch",
 		"hosts":            []any{"localhost:9200"},
@@ -505,6 +532,110 @@ func TestGetOtelConfig(t *testing.T) {
 				},
 			}),
 		},
+		{
+			name: "system/metrics",
+			model: &component.Model{
+				Components: []component.Component{
+					{
+						ID:         "system-metrics",
+						InputType:  "system/metrics",
+						OutputType: "elasticsearch",
+						InputSpec: &component.InputRuntimeSpec{
+							BinaryName: "agentbeat",
+							Spec: component.InputSpec{
+								Command: &component.CommandSpec{
+									Args: []string{"metricbeat"},
+								},
+							},
+						},
+						Units: []component.Unit{
+							{
+								ID:     "system/metrics",
+								Type:   client.UnitTypeInput,
+								Config: component.MustExpectedConfig(systemMetricsConfig),
+							},
+							{
+								ID:     "system/metrics-default",
+								Type:   client.UnitTypeOutput,
+								Config: component.MustExpectedConfig(esOutputConfig),
+							},
+						},
+					},
+				},
+			},
+			expectedConfig: confmap.NewFromStringMap(map[string]any{
+				"exporters": expectedESConfig,
+				"receivers": map[string]any{
+					"metricbeatreceiver/_agent-component/system-metrics": map[string]any{
+						"metricbeat": map[string]any{
+							"modules": []map[string]any{
+								{
+									"module":      "system",
+									"data_stream": map[string]any{"dataset": "generic-1"},
+									"id":          "test-1",
+									"index":       "metrics-generic-1-default",
+									"metricsets": map[string]any{
+										"cpu": map[string]any{
+											"data_stream.dataset": "system.cpu",
+										},
+										"memory": map[string]any{
+											"data_stream.dataset": "system.memory",
+										},
+										"network": map[string]any{
+											"data_stream.dataset": "system.network",
+										},
+										"filesystem": map[string]any{
+											"data_stream.dataset": "system.filesystem",
+										},
+									},
+									"processors": defaultProcessors("test-1", "generic-1", "metrics"),
+								},
+							},
+						},
+						"output": map[string]any{
+							"otelconsumer": map[string]any{},
+						},
+						"path": map[string]any{
+							"data": filepath.Join(paths.Run(), "system-metrics"),
+						},
+						"queue": map[string]any{
+							"mem": map[string]any{
+								"events": uint64(3200),
+								"flush": map[string]any{
+									"min_events": uint64(1600),
+									"timeout":    "10s",
+								},
+							},
+						},
+						"logging": map[string]any{
+							"with_fields": map[string]any{
+								"component": map[string]any{
+									"binary":  "metricbeat",
+									"dataset": "elastic_agent.metricbeat",
+									"type":    "system/metrics",
+									"id":      "system-metrics",
+								},
+								"log": map[string]any{
+									"source": "system-metrics",
+								},
+							},
+						},
+						"http": map[string]any{
+							"enabled": true,
+							"host":    "localhost",
+						},
+					},
+				},
+				"service": map[string]any{
+					"pipelines": map[string]any{
+						"logs/_agent-component/system-metrics": map[string][]string{
+							"exporters": []string{"elasticsearch/_agent-component/default"},
+							"receivers": []string{"metricbeatreceiver/_agent-component/system-metrics"},
+						},
+					},
+				},
+			}),
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {