add extended runtime test

.buildkite/pipeline.yml

@@ -223,6 +223,16 @@ steps:
       provider: "gcp"
       machineType: "n1-standard-8"
 
+  - label: "Extended runtime leak tests"
+    key: "extended-integration-tests"
+    command: ".buildkite/scripts/steps/integration_tests.sh stateful integration:TestForResourceLeaks"
+    artifact_paths:
+      - "build/TEST-**"
+      - "build/diagnostics/*"
+    agents:
+      provider: "gcp"
+      machineType: "n1-standard-8"    
+
   - label: "Integration tests"
     key: "integration-tests"
     command: ".buildkite/scripts/steps/integration_tests.sh stateful"

NOTICE.txt

@@ -5823,6 +5823,36 @@ OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 SOFTWARE.
 
 
+--------------------------------------------------------------------------------
+Dependency : github.com/sajari/regression
+Version: v1.0.1
+Licence type (autodetected): MIT
+--------------------------------------------------------------------------------
+
+Contents of probable licence file $GOMODCACHE/github.com/sajari/regression@v1.0.1/LICENSE:
+
+The MIT License (MIT)
+
+Copyright (c) 2014 Sajari Pty Ltd
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
+
 --------------------------------------------------------------------------------
 Dependency : github.com/schollz/progressbar/v3
 Version: v3.13.1

dev-tools/mage/pkg.go

@@ -230,7 +230,7 @@ func TestPackages(options ...TestPackagesOption) error {
 		if mg.Verbose() {
 			fmt.Println(out)
 		}
-		return err
+		return fmt.Errorf("error running package_test.go: %w, stdout: %s", err, out)
 	}
 
 	return nil

docs/test-framework-dev-guide.md

@@ -117,6 +117,11 @@ We pass a `-test.count` flag along with the name match
 We pass a `-test.run` flag along with the names of the tests we want to run in OR
 `GOTEST_FLAGS="-test.run ^(TestStandaloneUpgrade|TestFleetManagedUpgrade)$" mage integration:test`
 
+##### Run Extended Runtime Leak Test
+The test framework includes a "long running" test to check for resource leaks and stability.
+The runtime of the test can be set via the `LONG_TEST_RUNTIME` environment variable.
+The test itself can be run via the `integration:TestLongRunningAgentForLeaks` mage target.
+
 ##### Limitations
 Due to the way the parameters are passed to `devtools.GoTest` the value of the environment variable
 is split on space, so not all combination of flags and their values may be correctly split.

go.mod

@@ -50,6 +50,7 @@ require (
 	github.com/pierrre/gotestcover v0.0.0-20160517101806-924dca7d15f0
 	github.com/pkg/errors v0.9.1
 	github.com/rs/zerolog v1.27.0
+	github.com/sajari/regression v1.0.1
 	github.com/schollz/progressbar/v3 v3.13.1
 	github.com/shirou/gopsutil/v3 v3.24.1
 	github.com/sirupsen/logrus v1.9.3

go.sum

@@ -1700,6 +1700,8 @@ github.com/ryanuber/columnize v2.1.0+incompatible/go.mod h1:sm1tb6uqfes/u+d4ooFo
 github.com/safchain/ethtool v0.0.0-20190326074333-42ed695e3de8/go.mod h1:Z0q5wiBQGYcxhMZ6gUqHn6pYNLypFAvaL3UvgZLR0U4=
 github.com/safchain/ethtool v0.0.0-20210803160452-9aa261dae9b1/go.mod h1:Z0q5wiBQGYcxhMZ6gUqHn6pYNLypFAvaL3UvgZLR0U4=
 github.com/sagikazarmark/crypt v0.9.0/go.mod h1:RnH7sEhxfdnPm1z+XMgSLjWTEIjyK4z2dw6+4vHTMuo=
+github.com/sajari/regression v1.0.1 h1:iTVc6ZACGCkoXC+8NdqH5tIreslDTT/bXxT6OmHR5PE=
+github.com/sajari/regression v1.0.1/go.mod h1:NeG/XTW1lYfGY7YV/Z0nYDV/RGh3wxwd1yW46835flM=
 github.com/santhosh-tekuri/jsonschema v1.2.4 h1:hNhW8e7t+H1vgY+1QeEQpveR6D4+OwKPXCfD2aieJis=
 github.com/santhosh-tekuri/jsonschema v1.2.4/go.mod h1:TEAUOeZSmIxTTuHatJzrvARHiuO9LYd+cIxzgEHCQI4=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=

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

@@ -15,6 +15,7 @@ import (
 	"time"
 	"unicode"
 
+	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/utils"
 
@@ -606,6 +607,10 @@ func (b *BeatsMonitor) injectMetricsInput(cfg map[string]interface{}, componentI
 			},
 		},
 	}
+	dbgLog := logp.L()
+	for _, comp := range componentList {
+		dbgLog.Infof("component: %#v\n\t componentData: %#v", comp, comp.Component.String())
+	}
 	for unit, binaryName := range componentIDToBinary {
 		if !isSupportedMetricsBinary(binaryName) {
 			continue

magefile.go

@@ -1950,6 +1950,14 @@ func (Integration) TestBeatServerless(ctx context.Context, beatname string) erro
 	return integRunner(ctx, false, "TestBeatsServerless")
 }
 
+func (Integration) TestForResourceLeaks(ctx context.Context) error {
+	err := os.Setenv("TEST_LONG_RUNNING", "true")
+	if err != nil {
+		return fmt.Errorf("error setting TEST_LONG_RUNNING: %w", err)
+	}
+	return integRunner(ctx, false, "TestLongRunningAgentForLeaks")
+}
+
 // TestOnRemote shouldn't be called locally (called on remote host to perform testing)
 func (Integration) TestOnRemote(ctx context.Context) error {
 	mg.Deps(Build.TestBinaries)
@@ -2176,6 +2184,9 @@ func createTestRunner(matrix bool, singleTest string, goTestFlags string, batche
 		extraEnv["AGENT_KEEP_INSTALLED"] = os.Getenv("AGENT_KEEP_INSTALLED")
 	}
 
+	extraEnv["TEST_LONG_RUNNING"] = os.Getenv("TEST_LONG_RUNNING")
+	extraEnv["LONG_TEST_RUNTIME"] = os.Getenv("LONG_TEST_RUNTIME")
+
 	// these following two env vars are currently not used by anything, but can be used in the future to test beats or
 	// other binaries, see https://github.com/elastic/elastic-agent/pull/3258
 	binaryName := os.Getenv("TEST_BINARY_NAME")

pkg/testing/fixture.go

@@ -701,7 +701,10 @@ func (f *Fixture) ExecStatus(ctx context.Context, opts ...process.CmdOption) (Ag
 				}, uerr))
 	}
 
-	return status, err
+	if err != nil {
+		return status, fmt.Errorf("error running command (output: %s): %w", string(out), err)
+	}
+	return status, nil
 }
 
 // ExecInspect executes to inspect subcommand on the prepared Elastic Agent binary.
@@ -777,7 +780,7 @@ func (f *Fixture) ExecDiagnostics(ctx context.Context, cmd ...string) (string, e
 func (f *Fixture) IsHealthy(ctx context.Context, opts ...process.CmdOption) error {
 	status, err := f.ExecStatus(ctx, opts...)
 	if err != nil {
-		return fmt.Errorf("agent status returned and error: %w", err)
+		return fmt.Errorf("agent status returned an error: %w", err)
 	}
 
 	if status.State != int(cproto.State_HEALTHY) {

pkg/testing/tools/epr.go

@@ -0,0 +1,57 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package tools
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+)
+
+const eprProd = "https://epr.elastic.co"
+
+// / PackageSearchResult contains basic info on a package returned by a search
+type PackageSearchResult struct {
+	Name    string `json:"name"`
+	Version string `json:"version"`
+	Release string `json:"release"`
+	Path    string `json:"path"`
+}
+
+// GetLatestPackageRelease returns the version string of the latest package release
+func GetLatestPackageRelease(ctx context.Context, packageName string) (string, error) {
+	endpoint := fmt.Sprintf("%s/search?package=%s&all=false", eprProd, packageName)
+	req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil)
+	if err != nil {
+		return "", fmt.Errorf("error creating HTTP request: %w", err)
+	}
+	resp, err := http.DefaultClient.Do(req) //nolint:gosec,nolintlint // it's a test
+	//create body before we check for errors, easier to format error strings that way
+	body, errRead := io.ReadAll(resp.Body)
+	if errRead != nil {
+		return "", fmt.Errorf("error reading body of HTTP resp: %w", err)
+	}
+	resp.Body.Close()
+	if err != nil {
+		return "", fmt.Errorf("failed to create search request for EPR (%s): %w", body, err)
+	}
+	if resp.StatusCode >= 300 {
+		return "", fmt.Errorf("bad status code in response from EPR: %d - %s", resp.StatusCode, resp.Status)
+	}
+
+	parsedResp := []PackageSearchResult{}
+	err = json.Unmarshal(body, &parsedResp)
+	if err != nil {
+		return "", fmt.Errorf("error parsing search response: %w", err)
+	}
+	// if we set &all=false, we'll get at most one result
+	if len(parsedResp) < 1 {
+		return "", fmt.Errorf("no packages matching '%s' found", packageName)
+	}
+
+	return parsedResp[0].Version, nil
+}

pkg/testing/tools/estools/elasticsearch.go

@@ -477,6 +477,44 @@ func GetLogsForAgentID(ctx context.Context, client elastictransport.Interface, i
 	return handleDocsResponse(res)
 }
 
+// GetResultsForAgentAndDatastream returns any documents match both the given agent ID and data stream
+func GetResultsForAgentAndDatastream(ctx context.Context, client elastictransport.Interface, dataset string, agentID string) (Documents, error) {
+	indexQuery := map[string]interface{}{
+		"query": map[string]interface{}{
+			"bool": map[string]interface{}{
+				"must": []map[string]interface{}{
+					{
+						"match": map[string]interface{}{"data_stream.dataset": dataset},
+					},
+					{
+						"match": map[string]interface{}{"agent.id": agentID},
+					},
+				},
+			},
+		},
+	}
+
+	var buf bytes.Buffer
+	err := json.NewEncoder(&buf).Encode(indexQuery)
+	if err != nil {
+		return Documents{}, fmt.Errorf("error creating ES query: %w", err)
+	}
+
+	es := esapi.New(client)
+	res, err := es.Search(
+		es.Search.WithExpandWildcards("all"),
+		es.Search.WithBody(&buf),
+		es.Search.WithTrackTotalHits(true),
+		es.Search.WithContext(ctx),
+		es.Search.WithSize(300),
+	)
+	if err != nil {
+		return Documents{}, fmt.Errorf("error performing ES search: %w", err)
+	}
+
+	return handleDocsResponse(res)
+}
+
 // GetLogsForDatasetWithContext returns any logs associated with the datastream
 func GetLogsForDatasetWithContext(ctx context.Context, client elastictransport.Interface, index string) (Documents, error) {
 	indexQuery := map[string]interface{}{
@@ -546,6 +584,38 @@ func performQueryForRawQuery(ctx context.Context, queryRaw map[string]interface{
 	return handleDocsResponse(res)
 }
 
+// FindMatchingLogLinesForAgentWithContext returns the matching `message` line field for an agent with the matching ID
+func FindMatchingLogLinesForAgentWithContext(ctx context.Context, client elastictransport.Interface, agentID, line string) (Documents, error) {
+	queryRaw := map[string]interface{}{
+		"query": map[string]interface{}{
+			"bool": map[string]interface{}{
+				"must": []map[string]interface{}{
+					{
+						"match_phrase": map[string]interface{}{
+							"message": line,
+						},
+					},
+					{
+						"term": map[string]interface{}{
+							"agent.id": map[string]interface{}{
+								"value": agentID,
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	var buf bytes.Buffer
+	err := json.NewEncoder(&buf).Encode(queryRaw)
+	if err != nil {
+		return Documents{}, fmt.Errorf("error creating ES query: %w", err)
+	}
+
+	return performQueryForRawQuery(ctx, queryRaw, "logs-elastic_agent*", client)
+}
+
 // GetLogsForDatastream returns any logs associated with the datastream
 func GetLogsForDatastream(
 	ctx context.Context,

pkg/testing/tools/slope.go

@@ -0,0 +1,48 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package tools
+
+import (
+	"time"
+
+	"github.com/sajari/regression"
+)
+
+// Slope is a slim wrapper around a regression library for calculating rate of change over time in tests.
+type Slope struct {
+	handler *regression.Regression
+}
+
+func NewSlope(label string) Slope {
+	handler := new(regression.Regression)
+	handler.SetObserved(label)
+	handler.SetVar(0, "time")
+	return Slope{handler: handler}
+}
+
+// add a datapoint and timestamp to the calculaton.
+func (slope Slope) AddDatapoint(count float64, timeSinceStart time.Duration) {
+	slope.handler.Train(regression.DataPoint(count, []float64{timeSinceStart.Seconds()}))
+}
+
+// Run the regression on the supplied data
+func (slope Slope) Run() error {
+	return slope.handler.Run()
+}
+
+// return the slope of the regression
+func (slope Slope) GetSlope() float64 {
+	return slope.handler.GetCoeffs()[1]
+}
+
+// Formula returns a string representation of the regression formula
+func (slope Slope) Formula() string {
+	return slope.handler.Formula
+}
+
+// Debug returns a string representation of the regression, including all datapoints
+func (slope Slope) Debug() string {
+	return slope.handler.String()
+}