Receive and use package.version from Elastic Agent (#37553)

In managed mode (running under Agent), Beats now receive agent information alongside connection details. This includes the agent's package version, which Beats will report instead of their own. That means the version added to event will be the agent's package version.

The beats test framework (libbeat/tests/integration/framework.go) now utilizes exec.Cmd instead of os.StartProcess to initiate the test beat. Furthermore, the StdinPipe is now exposed instead of binding os.Stdin to the process stdin.

A new utility, `testing/certutil/certutil`, has been created to provide root CA and child certificates for use in tests.

CHANGELOG.next.asciidoc

@@ -15,6 +15,9 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 platform, and when viewed from a metadata API standpoint, it is impossible to differentiate it from OpenStack. If you
 know that your deployments run on Huawei Cloud exclusively, and you wish to have `cloud.provider` value as `huawei`,
 you can achieve this by overwriting the value using an `add_fields` processor. {pull}35184[35184]
+ - In managed mode, Beats running under Elastic Agent will report the package
+version of Elastic Agent as their own version. This includes all additional
+fields added to events containing the Beats version. {pull}37553[37553]
 
 *Auditbeat*
 

NOTICE.txt

@@ -12494,11 +12494,11 @@ Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-a
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/elastic-agent-client/v7
-Version: v7.6.0
+Version: v7.8.0
 Licence type (autodetected): Elastic
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-client/v7@v7.6.0/LICENSE.txt:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-client/v7@v7.8.0/LICENSE.txt:
 
 ELASTIC LICENSE AGREEMENT
 
@@ -25546,11 +25546,11 @@ Contents of probable licence file $GOMODCACHE/google.golang.org/[email protected]/LIC
 
 --------------------------------------------------------------------------------
 Dependency : google.golang.org/protobuf
-Version: v1.31.0
+Version: v1.32.0
 Licence type (autodetected): BSD-3-Clause
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/google.golang.org/protobuf@v1.31.0/LICENSE:
+Contents of probable licence file $GOMODCACHE/google.golang.org/protobuf@v1.32.0/LICENSE:
 
 Copyright (c) 2018 The Go Authors. All rights reserved.
 

docs/devguide/testing.asciidoc

@@ -50,24 +50,28 @@ In Metricbeat, run the command from within a module like this: `go test --tags i
 
 A note about tags: the `--data` flag is a custom flag added by Metricbeat and Packetbeat frameworks. It will not be present in case tags do not match, as the relevant code will not be run and silently skipped (without the tag the test file is ignored by Go compiler so the framework doesn't load). This may happen if there are different tags in the build tags of the metricset under test (i.e. the GCP billing metricset requires the `billing` tag too).
 
-==== Running Python Tests
+==== Running System (integration) Tests (Python and Go)
 
-Python system tests are defined in the `tests/system` directory. They require a testing binary to be available and the python environment to be set up.
+The system tests are defined in the `tests/system` (for legacy Python test) and on `tests/integration` (for Go tests) directory. They require a testing binary to be available and the python environment to be set up.
 
-To create the testing binary run `mage buildSystemTestBinary`. This will create the test binary in the beat directory. To setup the testing environment run `mage pythonVirtualEnv` which will create a virtual environment with all test dependencies and print its location. To activate it, the instructions depend on your operating system. See the https://packaging.python.org/en/latest/guides/installing-using-pip-and-virtual-environments/#activating-a-virtual-environment[virtualenv documentation].
+To create the testing binary run `mage buildSystemTestBinary`. This will create the test binary in the beat directory. To set up the Python testing environment run `mage pythonVirtualEnv` which will create a virtual environment with all test dependencies and print its location. To activate it, the instructions depend on your operating system. See the https://packaging.python.org/en/latest/guides/installing-using-pip-and-virtual-environments/#activating-a-virtual-environment[virtualenv documentation].
 
 To run the system and integration tests use the `mage pythonIntegTest` target, which will start the required services using https://docs.docker.com/compose/[docker-compose] and run all integration tests. Similar to Go integration tests, the individual steps can be done manually to allow selecting which tests should be run:
 
 [source,bash]
 ----
 # Create and activate the system test virtual environment (assumes a Unix system).
 source $(mage pythonVirtualEnv)/bin/activate
+
 # Pull and build the containers. Only needs to be done once unless you change the containers.
 mage docker:composeBuild
+
 # Bring up all containers, wait until they are healthy, and put them in the background.
 mage docker:composeUp
+
 # Run all system and integration tests.
 INTEGRATION_TESTS=1 pytest ./tests/system
+
 # Stop all started containers.
 mage docker:composeDown
 ----

go.mod

@@ -69,7 +69,7 @@ require (
 	github.com/dustin/go-humanize v1.0.1
 	github.com/eapache/go-resiliency v1.2.0
 	github.com/eclipse/paho.mqtt.golang v1.3.5
-	github.com/elastic/elastic-agent-client/v7 v7.6.0
+	github.com/elastic/elastic-agent-client/v7 v7.8.0
 	github.com/elastic/go-concert v0.2.0
 	github.com/elastic/go-libaudit/v2 v2.5.0
 	github.com/elastic/go-licenser v0.4.1
@@ -164,7 +164,7 @@ require (
 	google.golang.org/api v0.128.0
 	google.golang.org/genproto v0.0.0-20230920204549-e6e6cdab5c13 // indirect
 	google.golang.org/grpc v1.58.3
-	google.golang.org/protobuf v1.31.0
+	google.golang.org/protobuf v1.32.0
 	gopkg.in/inf.v0 v0.9.1
 	gopkg.in/jcmturner/aescts.v1 v1.0.1 // indirect
 	gopkg.in/jcmturner/dnsutils.v1 v1.0.1 // indirect

go.sum

@@ -663,8 +663,8 @@ github.com/elastic/ebpfevents v0.3.2 h1:UJ8kW5jw2TpUR5MEMaZ1O62sK9JQ+5xTlj+YpQC6
 github.com/elastic/ebpfevents v0.3.2/go.mod h1:o21z5xup/9dK8u0Hg9bZRflSqqj1Zu5h2dg2hSTcUPQ=
 github.com/elastic/elastic-agent-autodiscover v0.6.7 h1:+KVjltN0rPsBrU8b156gV4lOTBgG/vt0efFCFARrf3g=
 github.com/elastic/elastic-agent-autodiscover v0.6.7/go.mod h1:hFeFqneS2r4jD0/QzGkrNk0YVdN0JGh7lCWdsH7zcI4=
-github.com/elastic/elastic-agent-client/v7 v7.6.0 h1:FEn6FjzynW4TIQo5G096Tr7xYK/P5LY9cSS6wRbXZTc=
-github.com/elastic/elastic-agent-client/v7 v7.6.0/go.mod h1:GlUKrbVd/O1CRAZonpBeN3J0RlVqP6VGcrBjFWca+aM=
+github.com/elastic/elastic-agent-client/v7 v7.8.0 h1:GHFzDJIWpdgI0qDk5EcqbQJGvwTsl2E2vQK3/xe+MYQ=
+github.com/elastic/elastic-agent-client/v7 v7.8.0/go.mod h1:ihtjqJzYiIltlRhNruaSSc0ogxIhqPD5hOMKq16cI1s=
 github.com/elastic/elastic-agent-libs v0.7.5 h1:4UMqB3BREvhwecYTs/L23oQp1hs/XUkcunPlmTZn5yg=
 github.com/elastic/elastic-agent-libs v0.7.5/go.mod h1:pGMj5myawdqu+xE+WKvM5FQzKQ/MonikkWOzoFTJxaU=
 github.com/elastic/elastic-agent-shipper-client v0.5.1-0.20230228231646-f04347b666f3 h1:sb+25XJn/JcC9/VL8HX4r4QXSUq4uTNzGS2kxOE7u1U=
@@ -2656,8 +2656,8 @@ google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp0
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
 google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
-google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8=
-google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+google.golang.org/protobuf v1.32.0 h1:pPC6BG5ex8PDFnkbrGU3EixyhKcQ2aDuBS36lqK/C7I=
+google.golang.org/protobuf v1.32.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/airbrake/gobrake.v2 v2.0.9/go.mod h1:/h5ZAUhDkGaJfjzjKLSjv6zCL6O0LLBxU4K+aSYdM/U=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

libbeat/cmd/instance/beat.go

@@ -838,10 +838,25 @@ func (b *Beat) configure(settings Settings) error {
 	}
 
 	// initialize config manager
-	b.Manager, err = management.NewManager(b.Config.Management, reload.RegisterV2)
+	m, err := management.NewManager(b.Config.Management, reload.RegisterV2)
 	if err != nil {
 		return err
 	}
+	b.Manager = m
+
+	if b.Manager.AgentInfo().Version != "" {
+		// During the manager initialization the client to connect to the agent is
+		// also initialized. That makes the beat to read information sent by the
+		// agent, which includes the AgentInfo with the agent's package version.
+		// Components running under agent should report the agent's package version
+		// as their own version.
+		// In order to do so b.Info.Version needs to be set to the version the agent
+		// sent. As this Beat instance is initialized much before the package
+		// version is received, it's overridden here. So far it's early enough for
+		// the whole beat to report the right version.
+		b.Info.Version = b.Manager.AgentInfo().Version
+		version.SetPackageVersion(b.Info.Version)
+	}
 
 	if err := b.Manager.CheckRawConfig(b.RawConfig); err != nil {
 		return err
@@ -1521,13 +1536,13 @@ func (bc *beatConfig) Validate() error {
 		if bc.Pipeline.Queue.IsSet() && outputPC.Queue.IsSet() {
 			return fmt.Errorf("top level queue and output level queue settings defined, only one is allowed")
 		}
-		//elastic-agent doesn't support disk queue yet
+		// elastic-agent doesn't support disk queue yet
 		if bc.Management.Enabled() && outputPC.Queue.Config().Enabled() && outputPC.Queue.Name() == diskqueue.QueueType {
 			return fmt.Errorf("disk queue is not supported when management is enabled")
 		}
 	}
 
-	//elastic-agent doesn't support disk queue yet
+	// elastic-agent doesn't support disk queue yet
 	if bc.Management.Enabled() && bc.Pipeline.Queue.Config().Enabled() && bc.Pipeline.Queue.Name() == diskqueue.QueueType {
 		return fmt.Errorf("disk queue is not supported when management is enabled")
 	}

libbeat/management/management.go

@@ -82,9 +82,12 @@ type Manager interface {
 	//
 	// Calls to 'CheckRawConfig()' or 'SetPayload()' will be ignored after calling stop.
 	//
-	// Note: Stop will not call 'UnregisterAction()' automaticallty.
+	// Note: Stop will not call 'UnregisterAction()' automatically.
 	Stop()
 
+	// AgentInfo returns the information of the agent to which the manager is connected.
+	AgentInfo() client.AgentInfo
+
 	// SetStopCallback accepts a function that need to be called when the manager want to shutdown the
 	// beats. This is needed when you want your beats to be gracefully shutdown remotely by the Elastic Agent
 	// when a policy doesn't need to run this beat.
@@ -190,6 +193,7 @@ func (n *fallbackManager) Stop() {
 // but that does not mean the Beat is being managed externally,
 // hence it will always return false.
 func (n *fallbackManager) Enabled() bool                         { return false }
+func (n *fallbackManager) AgentInfo() client.AgentInfo           { return client.AgentInfo{} }
 func (n *fallbackManager) Start() error                          { return nil }
 func (n *fallbackManager) CheckRawConfig(cfg *config.C) error    { return nil }
 func (n *fallbackManager) RegisterAction(action client.Action)   {}

libbeat/tests/integration/cmd_keystore_test.go

@@ -100,19 +100,23 @@ func TestKeystoreRemoveMultipleExistingKeys(t *testing.T) {
 	mockbeat.Stop()
 
 	mockbeat.Start("keystore", "add", "key1", "--stdin")
-	fmt.Fprintf(os.Stdin, "pass1")
+
+	fmt.Fprintf(mockbeat.stdin, "pass1")
+	require.NoError(t, mockbeat.stdin.Close(), "could not close mockbeat stdin")
 	procState, err := mockbeat.Process.Wait()
 	require.NoError(t, err)
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")
 
 	mockbeat.Start("keystore", "add", "key2", "--stdin")
-	fmt.Fprintf(os.Stdin, "pass2")
+	fmt.Fprintf(mockbeat.stdin, "pass2")
+	require.NoError(t, mockbeat.stdin.Close(), "could not close mockbeat stdin")
 	procState, err = mockbeat.Process.Wait()
 	require.NoError(t, err)
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")
 
 	mockbeat.Start("keystore", "add", "key3", "--stdin")
-	fmt.Fprintf(os.Stdin, "pass3")
+	fmt.Fprintf(mockbeat.stdin, "pass3")
+	require.NoError(t, mockbeat.stdin.Close(), "could not close mockbeat stdin")
 	procState, err = mockbeat.Process.Wait()
 	require.NoError(t, err)
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")
@@ -138,19 +142,22 @@ func TestKeystoreList(t *testing.T) {
 	mockbeat.Stop()
 
 	mockbeat.Start("keystore", "add", "key1", "--stdin")
-	fmt.Fprintf(os.Stdin, "pass1")
+	fmt.Fprintf(mockbeat.stdin, "pass1")
+	require.NoError(t, mockbeat.stdin.Close(), "could not close mockbeat stdin")
 	procState, err := mockbeat.Process.Wait()
 	require.NoError(t, err)
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")
 
 	mockbeat.Start("keystore", "add", "key2", "--stdin")
-	fmt.Fprintf(os.Stdin, "pass2")
+	fmt.Fprintf(mockbeat.stdin, "pass2")
+	require.NoError(t, mockbeat.stdin.Close(), "could not close mockbeat stdin")
 	procState, err = mockbeat.Process.Wait()
 	require.NoError(t, err)
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")
 
 	mockbeat.Start("keystore", "add", "key3", "--stdin")
-	fmt.Fprintf(os.Stdin, "pass3")
+	fmt.Fprintf(mockbeat.stdin, "pass3")
+	require.NoError(t, mockbeat.stdin.Close(), "could not close mockbeat stdin")
 	procState, err = mockbeat.Process.Wait()
 	require.NoError(t, err)
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")
@@ -186,7 +193,8 @@ func TestKeystoreAddSecretFromStdin(t *testing.T) {
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")
 
 	mockbeat.Start("keystore", "add", "key1", "--stdin")
-	fmt.Fprintf(os.Stdin, "pass1")
+	fmt.Fprintf(mockbeat.stdin, "pass1")
+	require.NoError(t, mockbeat.stdin.Close(), "could not close mockbeat stdin")
 	procState, err = mockbeat.Process.Wait()
 	require.NoError(t, err)
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")
@@ -202,13 +210,15 @@ func TestKeystoreUpdateForce(t *testing.T) {
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")
 
 	mockbeat.Start("keystore", "add", "key1", "--stdin")
-	fmt.Fprintf(os.Stdin, "pass1")
+	fmt.Fprintf(mockbeat.stdin, "pass1")
+	require.NoError(t, mockbeat.stdin.Close(), "could not close mockbeat stdin")
 	procState, err = mockbeat.Process.Wait()
 	require.NoError(t, err)
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")
 
 	mockbeat.Start("keystore", "add", "key1", "--force", "--stdin")
-	fmt.Fprintf(os.Stdin, "pass2")
+	fmt.Fprintf(mockbeat.stdin, "pass2")
+	require.NoError(t, mockbeat.stdin.Close(), "could not close mockbeat stdin")
 	procState, err = mockbeat.Process.Wait()
 	require.NoError(t, err)
 	require.Equal(t, 0, procState.ExitCode(), "incorrect exit code")

libbeat/tests/integration/framework.go

@@ -30,6 +30,7 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"regexp"
 	"strings"
@@ -55,6 +56,7 @@ type BeatProc struct {
 	logFileOffset       int64
 	t                   *testing.T
 	tempDir             string
+	stdin               io.WriteCloser
 	stdout              *os.File
 	stderr              *os.File
 	Process             *os.Process
@@ -90,18 +92,20 @@ type Total struct {
 	Value int `json:"value"`
 }
 
-// NewBeat createa a new Beat process from the system tests binary.
+// NewBeat creates a new Beat process from the system tests binary.
 // It sets some required options like the home path, logging, etc.
 // `tempDir` will be used as home and logs directory for the Beat
 // `args` will be passed as CLI arguments to the Beat
 func NewBeat(t *testing.T, beatName, binary string, args ...string) *BeatProc {
 	require.FileExistsf(t, binary, "beat binary must exists")
 	tempDir := createTempDir(t)
 	configFile := filepath.Join(tempDir, beatName+".yml")
+
 	stdoutFile, err := os.Create(filepath.Join(tempDir, "stdout"))
 	require.NoError(t, err, "error creating stdout file")
 	stderrFile, err := os.Create(filepath.Join(tempDir, "stderr"))
 	require.NoError(t, err, "error creating stderr file")
+
 	p := BeatProc{
 		Binary: binary,
 		baseArgs: append([]string{
@@ -213,15 +217,27 @@ func (b *BeatProc) Start(args ...string) {
 func (b *BeatProc) startBeat() {
 	b.cmdMutex.Lock()
 	defer b.cmdMutex.Unlock()
+
 	_, _ = b.stdout.Seek(0, 0)
 	_ = b.stdout.Truncate(0)
 	_, _ = b.stderr.Seek(0, 0)
 	_ = b.stderr.Truncate(0)
-	var procAttr os.ProcAttr
-	procAttr.Files = []*os.File{os.Stdin, b.stdout, b.stderr}
-	process, err := os.StartProcess(b.fullPath, b.Args, &procAttr)
+
+	cmd := exec.Cmd{
+		Path:   b.fullPath,
+		Args:   b.Args,
+		Stdout: b.stdout,
+		Stderr: b.stderr,
+	}
+
+	var err error
+	b.stdin, err = cmd.StdinPipe()
+	require.NoError(b.t, err, "could not get cmd StdinPipe")
+
+	err = cmd.Start()
 	require.NoError(b.t, err, "error starting beat process")
-	b.Process = process
+
+	b.Process = cmd.Process
 }
 
 // waitBeatToExit blocks until the Beat exits, it returns
@@ -515,6 +531,10 @@ func (b *BeatProc) LoadMeta() (Meta, error) {
 	return m, nil
 }
 
+func (b *BeatProc) Stdin() io.WriteCloser {
+	return b.stdin
+}
+
 func GetESURL(t *testing.T, scheme string) url.URL {
 	t.Helper()
 

libbeat/tests/integration/mockserver.go

@@ -38,18 +38,18 @@ type unitKey struct {
 }
 
 // NewMockServer creates a GRPC server to mock the Elastic-Agent.
-// On the first check in call it will send the first element of `unit`
+// On the first check-in call it will send the first element of `unit`
 // as the expected unit, on successive calls, if the Beat has reached
 // that state, it will move on to sending the next state.
 // It will also validate the features.
 //
 // if `observedCallback` is not nil, it will be called on every
-// check in receiving the `proto.CheckinObserved` sent by the
+// check-in receiving the `proto.CheckinObserved` sent by the
 // Beat and index from `units` that was last sent to the Beat.
 //
 // If `delay` is not zero, when the Beat state matches the last
 // sent units, the server will wait for `delay` before sending the
-// the next state. This will block the check in call from the Beat.
+// next state. This will block the check-in call from the Beat.
 func NewMockServer(
 	units [][]*proto.UnitExpected,
 	featuresIdxs []uint64,
@@ -58,7 +58,7 @@ func NewMockServer(
 	delay time.Duration,
 ) *mock.StubServerV2 {
 	i := 0
-	agentInfo := &proto.CheckinAgentInfo{
+	agentInfo := &proto.AgentInfo{
 		Id:       "elastic-agent-id",
 		Version:  version.GetDefaultVersion(),
 		Snapshot: true,