Modify e2e test

Author: swiatekm

testing/integration/beat_receivers_test.go

@@ -7,26 +7,21 @@
 package integration
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"io"
 	"net/http"
 	"runtime"
 	"testing"
-	"text/template"
 	"time"
 
 	"gopkg.in/yaml.v2"
 
 	"github.com/elastic/elastic-agent-libs/kibana"
 	"github.com/elastic/elastic-agent-libs/testing/estools"
-	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
 	atesting "github.com/elastic/elastic-agent/pkg/testing"
 	"github.com/elastic/elastic-agent/pkg/testing/define"
 	"github.com/elastic/elastic-agent/pkg/testing/tools/testcontext"
-	"github.com/elastic/elastic-agent/pkg/utils"
-
 	"github.com/gofrs/uuid/v5"
 	"github.com/stretchr/testify/require"
 )
@@ -79,73 +74,77 @@ func TestAgentMonitoring(t *testing.T) {
 		Force:          true,
 	}
 
+	// prepare the policy and marshalled configuration
+	policyCtx, policyCancel := testcontext.WithDeadline(t, context.Background(), time.Now().Add(5*time.Minute))
+	t.Cleanup(policyCancel)
+
+	// 1. Create and install policy with just monitoring
+	createPolicyReq := kibana.AgentPolicy{
+		Name:        fmt.Sprintf("%s-%s", t.Name(), uuid.Must(uuid.NewV4()).String()),
+		Namespace:   info.Namespace,
+		Description: fmt.Sprintf("%s policy", t.Name()),
+		MonitoringEnabled: []kibana.MonitoringEnabledOption{
+			kibana.MonitoringEnabledLogs,
+			kibana.MonitoringEnabledMetrics,
+		},
+	}
+	policyResponse, err := info.KibanaClient.CreatePolicy(policyCtx, createPolicyReq)
+	require.NoError(t, err, "error creating policy")
+
+	// 2. Download the policy, add the API key
+	downloadURL := fmt.Sprintf("/api/fleet/agent_policies/%s/download", policyResponse.ID)
+	resp, err := info.KibanaClient.Connection.SendWithContext(policyCtx, http.MethodGet, downloadURL, nil, nil, nil)
+	require.NoError(t, err, "error downloading policy")
+	policyBytes, err := io.ReadAll(resp.Body)
+	require.NoError(t, err, "error reading policy response")
+	defer resp.Body.Close()
+
+	apiKeyResponse, err := createESApiKey(info.ESClient)
+	require.NoError(t, err, "failed to get api key")
+	require.True(t, len(apiKeyResponse.Encoded) > 1, "api key is invalid %q", apiKeyResponse)
+	apiKey, err := getDecodedApiKey(apiKeyResponse)
+	require.NoError(t, err, "error decoding api key")
+
+	type PolicyOutputs struct {
+		Type   string   `yaml:"type"`
+		Hosts  []string `yaml:"hosts"`
+		Preset string   `yaml:"preset"`
+		ApiKey string   `yaml:"api_key"`
+	}
+	type PolicyStruct struct {
+		ID                string                   `yaml:"id"`
+		Revision          int                      `yaml:"revision"`
+		Outputs           map[string]PolicyOutputs `yaml:"outputs"`
+		Fleet             map[string]any           `yaml:"fleet"`
+		OutputPermissions map[string]any           `yaml:"output_permissions"`
+		Agent             map[string]any           `yaml:"agent"`
+		Inputs            []map[string]any         `yaml:"inputs"`
+		Signed            map[string]any           `yaml:"signed"`
+		SecretReferences  []map[string]any         `yaml:"secret_references"`
+		Namespaces        []map[string]any         `yaml:"namespaces"`
+	}
+
+	policy := PolicyStruct{}
+	err = yaml.Unmarshal(policyBytes, &policy)
+	require.NoError(t, err, "error unmarshalling policy")
+	d, prs := policy.Outputs["default"]
+	require.True(t, prs, "default must be in outputs")
+	d.ApiKey = string(apiKey)
+	policy.Outputs["default"] = d
+
 	// Flow
 	// 1. Create and install policy with just monitoring
 	// 2. Download the policy, add the API key
 	// 3. Install without enrolling in fleet
 	// 4. Make sure logs and metrics for agent monitoring are being received
 	t.Run("verify elastic-agent monitoring functionality", func(t *testing.T) {
 		ctx, cancel := testcontext.WithDeadline(t, context.Background(), time.Now().Add(5*time.Minute))
-		defer cancel()
-
-		// 1. Create and install policy with just monitoring
-		createPolicyReq := kibana.AgentPolicy{
-			Name:        fmt.Sprintf("%s-%s", t.Name(), uuid.Must(uuid.NewV4()).String()),
-			Namespace:   info.Namespace,
-			Description: fmt.Sprintf("%s policy", t.Name()),
-			MonitoringEnabled: []kibana.MonitoringEnabledOption{
-				kibana.MonitoringEnabledLogs,
-				kibana.MonitoringEnabledMetrics,
-			},
-		}
-		policyResponse, err := info.KibanaClient.CreatePolicy(ctx, createPolicyReq)
-		require.NoError(t, err, "error creating policy")
-
-		// 2. Download the policy, add the API key
-		downloadURL := fmt.Sprintf("/api/fleet/agent_policies/%s/download", policyResponse.ID)
-		resp, err := info.KibanaClient.Connection.SendWithContext(ctx, http.MethodGet, downloadURL, nil, nil, nil)
-		require.NoError(t, err, "error downloading policy")
-		policy, err := io.ReadAll(resp.Body)
-		require.NoError(t, err, "error reading policy response")
-		defer resp.Body.Close()
-
-		apiKeyResponse, err := createESApiKey(info.ESClient)
-		require.NoError(t, err, "failed to get api key")
-		require.True(t, len(apiKeyResponse.Encoded) > 1, "api key is invalid %q", apiKeyResponse)
-		apiKey, err := getDecodedApiKey(apiKeyResponse)
-		require.NoError(t, err, "error decoding api key")
-
-		type PolicyOutputs struct {
-			Type   string   `yaml:"type"`
-			Hosts  []string `yaml:"hosts"`
-			Preset string   `yaml:"preset"`
-			ApiKey string   `yaml:"api_key"`
-		}
-		type PolicyStruct struct {
-			ID                string                   `yaml:"id"`
-			Revision          int                      `yaml:"revision"`
-			Outputs           map[string]PolicyOutputs `yaml:"outputs"`
-			Fleet             map[string]any           `yaml:"fleet"`
-			OutputPermissions map[string]any           `yaml:"output_permissions"`
-			Agent             map[string]any           `yaml:"agent"`
-			Inputs            []map[string]any         `yaml:"inputs"`
-			Signed            map[string]any           `yaml:"signed"`
-			SecretReferences  []map[string]any         `yaml:"secret_references"`
-			Namespaces        []map[string]any         `yaml:"namespaces"`
-		}
-
-		y := PolicyStruct{}
-		err = yaml.Unmarshal(policy, &y)
-		require.NoError(t, err, "error unmarshalling policy")
-		d, prs := y.Outputs["default"]
-		require.True(t, prs, "default must be in outputs")
-		d.ApiKey = string(apiKey)
-		y.Outputs["default"] = d
-		policyBytes, err := yaml.Marshal(y)
-		require.NoErrorf(t, err, "error marshalling policy, struct was %v", y)
+		t.Cleanup(cancel)
+		updatedPolicyBytes, err := yaml.Marshal(policy)
+		require.NoErrorf(t, err, "error marshalling policy, struct was %v", policy)
 		t.Cleanup(func() {
 			if t.Failed() {
-				t.Logf("policy was %s", string(policyBytes))
+				t.Logf("policy was %s", string(updatedPolicyBytes))
 			}
 		})
 
@@ -156,7 +155,7 @@ func TestAgentMonitoring(t *testing.T) {
 		err = fixture.Prepare(ctx)
 		require.NoError(t, err, "error preparing fixture")
 
-		err = fixture.Configure(ctx, policyBytes)
+		err = fixture.Configure(ctx, updatedPolicyBytes)
 		require.NoError(t, err, "error configuring fixture")
 
 		output, err := fixture.InstallWithoutEnroll(ctx, &installOpts)
@@ -225,198 +224,50 @@ func TestAgentMonitoring(t *testing.T) {
 			}, "monitoring logs from elastic-agent should exist before proceeding",
 		)
 
-		type configOptions struct {
-			InputPath      string
-			ESEndpoint     string
-			ESApiKey       string
-			SocketEndpoint string
-			Namespace      string
-		}
-		esEndpoint, err := getESHost()
-		require.NoError(t, err, "error getting elasticsearch endpoint")
-		esApiKey, err := createESApiKey(info.ESClient)
-		require.NoError(t, err, "error creating API key")
-		require.NotEmptyf(t, esApiKey.Encoded, "api key is invalid %q", esApiKey)
-
-		// Start monitoring in otel mode
-		fixture, err := define.NewFixtureFromLocalBuild(t, define.Version())
-		require.NoError(t, err)
-
 		ctx, cancel := testcontext.WithDeadline(t, context.Background(), time.Now().Add(5*time.Minute))
-		defer cancel()
+		t.Cleanup(cancel)
 
-		err = fixture.Prepare(ctx)
-		require.NoError(t, err)
+		// switch monitoring to the otel runtime
+		monitoring := policy.Agent["monitoring"]
+		monitoringMap := monitoring.(map[any]any)
+		monitoringMap["_runtime_experimental"] = "otel"
+		policy.Agent["monitoring"] = monitoringMap
 
-		// installs elastic-agent with empty elastic-agent.yml to get its working dir first
-		err = fixture.Configure(ctx, []byte{})
-		require.NoError(t, err)
-
-		output, err := fixture.InstallWithoutEnroll(ctx, &installOpts)
-		require.NoErrorf(t, err, "error install withouth enroll: %s\ncombinedoutput:\n%s", err, string(output))
-
-		// Ensure elastic-agent is healthy, otherwise we cannot perform retstart operation
-		require.Eventually(t, func() bool {
-			err = fixture.IsHealthy(ctx)
-			if err != nil {
-				t.Logf("waiting for agent healthy: %s", err.Error())
-				return false
-			}
-			return true
-		}, 30*time.Second, 1*time.Second)
-
-		configTemplateOTel := `
-receivers:
-  filebeatreceiver/filestream-monitoring:
-    filebeat:
-      inputs:
-        - type: filestream
-          enabled: true
-          id: filestream-monitoring-agent
-          paths:
-            -  {{.InputPath}}/data/elastic-agent-*/logs/elastic-agent-*.ndjson
-            -  {{.InputPath}}/data/elastic-agent-*/logs/elastic-agent-watcher-*.ndjson
-          close:
-            on_state_change:
-              inactive: 5m
-          parsers:
-            - ndjson:
-                add_error_key: true
-                message_key: message
-                overwrite_keys: true
-                target: ""
-          processors:
-            - add_fields:
-                fields:
-                  dataset: elastic_agent
-                  namespace: {{.Namespace}}
-                  type: logs
-                target: data_stream
-            - add_fields:
-                fields:
-                  dataset: elastic_agent
-                target: event
-            - add_fields:
-                fields:
-                  id: 0ddca301-e7c0-4eac-8432-7dd05bc9cb06
-                  snapshot: false
-                  version: 8.19.0
-                target: elastic_agent
-            - add_fields:
-                fields:
-                  id: 0879f47d-df41-464d-8462-bc2b8fef45bf
-                target: agent
-            - drop_event:
-                when:
-                  regexp:
-                    component.id: .*-monitoring$
-            - drop_event:
-                when:
-                  regexp:
-                    message: ^Non-zero metrics in the last
-            - copy_fields:
-                fields:
-                  - from: data_stream.dataset
-                    to: data_stream.dataset_original
-            - drop_fields:
-                fields:
-                  - data_stream.dataset
-            - copy_fields:
-                fail_on_error: false
-                fields:
-                  - from: component.dataset
-                    to: data_stream.dataset
-                ignore_missing: true
-            - copy_fields:
-                fail_on_error: false
-                fields:
-                  - from: data_stream.dataset_original
-                    to: data_stream.dataset
-            - drop_fields:
-                fields:
-                  - data_stream.dataset_original
-                  - event.dataset
-            - copy_fields:
-                fields:
-                  - from: data_stream.dataset
-                    to: event.dataset
-            - drop_fields:
-                fields:
-                  - ecs.version
-                ignore_missing: true
-    output:
-      otelconsumer:
-    queue:
-      mem:
-        flush:
-          timeout: 0s
-    logging:
-      level: info
-      selectors:
-        - '*'
-    http.enabled: true
-    http.host: {{ .SocketEndpoint }}
-exporters:
-  debug:
-    use_internal_logger: false
-    verbosity: detailed
-  elasticsearch/log:
-    endpoints:
-      - {{.ESEndpoint}}
-    compression: none
-    api_key: {{.ESApiKey}}
-    logs_dynamic_index:
-      enabled: true
-    batcher:
-      enabled: true
-      flush_timeout: 0.5s
-    mapping:
-      mode: bodymap
-service:
-  pipelines:
-    logs:
-      receivers:
-        - filebeatreceiver/filestream-monitoring
-      exporters:
-        - elasticsearch/log
-`
-		socketEndpoint := utils.SocketURLWithFallback(uuid.Must(uuid.NewV4()).String(), paths.TempDir())
-
-		// configure elastic-agent.yml with new config
-		var configBuffer bytes.Buffer
-		template.Must(template.New("config").Parse(configTemplateOTel)).Execute(&configBuffer,
-			configOptions{
-				InputPath:      fixture.WorkDir(),
-				ESEndpoint:     esEndpoint,
-				ESApiKey:       esApiKey.Encoded,
-				SocketEndpoint: socketEndpoint,
-				Namespace:      info.Namespace,
-			})
-		configOTelContents := configBuffer.Bytes()
+		updatedPolicyBytes, err := yaml.Marshal(policy)
+		require.NoErrorf(t, err, "error marshalling policy, struct was %v", policy)
 		t.Cleanup(func() {
 			if t.Failed() {
-				t.Logf("Contents of agent config file:\n%s\n", string(configOTelContents))
+				t.Logf("policy was %s", string(updatedPolicyBytes))
 			}
 		})
-		err = fixture.Configure(ctx, configOTelContents)
+
+		// 3. Install without enrolling in fleet
+		fixture, err := define.NewFixtureFromLocalBuild(t, define.Version())
 		require.NoError(t, err)
 
-		// Get the timestamp before restarting. Required to separate logs from agent and otel
+		err = fixture.Prepare(ctx)
+		require.NoError(t, err, "error preparing fixture")
+
+		err = fixture.Configure(ctx, updatedPolicyBytes)
+		require.NoError(t, err, "error configuring fixture")
+
+		// Get the timestamp before starting. Required to separate logs from agent and otel.
 		timestamp := time.Now().UTC().Format("2006-01-02T15:04:05.000Z")
 
 		fmt.Println(time.Now())
-		// Restart elastic-agent
-		output, err = fixture.Exec(ctx, []string{"restart"})
-		require.NoErrorf(t, err, "error restarting agent: %s\ncombinedoutput:\n%s", err, string(output))
 
+		output, err := fixture.InstallWithoutEnroll(ctx, &installOpts)
+		require.NoErrorf(t, err, "error install withouth enroll: %s\ncombinedoutput:\n%s", err, string(output))
+
+		// Ensure elastic-agent is healthy, otherwise we cannot perform restart operation
 		require.Eventually(t, func() bool {
 			err = fixture.IsHealthy(ctx)
 			if err != nil {
 				t.Logf("waiting for agent healthy: %s", err.Error())
 				return false
 			}
 			return true
-		}, 30*time.Second, 1*time.Second)
+		}, 1*time.Minute, 1*time.Second)
 
 		// run this only for logs for now
 		tc := tests[0]