Send package.version to components

changelog/fragments/1707143167-send-package-version-to-components.yaml

@@ -0,0 +1,34 @@
+# 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: Agent information is sent to components together with the connection information.
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+description: When starting a component the Elastic Agent sends its information,
+including its the package version, which the componets should report as their
+own version.
+
+# Affected component; a word indicating the component this changeset affects.
+component: component runtime
+
+# PR URL; 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: https://github.com/elastic/elastic-agent/pull/4024
+
+# Issue URL; 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: https://github.com/owner/repo/1234

internal/pkg/agent/application/actions/handlers/handler_action_policy_change.go

@@ -37,7 +37,7 @@ const (
 // PolicyChangeHandler is a handler for POLICY_CHANGE action.
 type PolicyChangeHandler struct {
 	log       *logger.Logger
-	agentInfo *info.AgentInfo
+	agentInfo info.Agent
 	config    *configuration.Configuration
 	store     storage.Store
 	ch        chan coordinator.ConfigChange
@@ -52,7 +52,7 @@ type PolicyChangeHandler struct {
 // NewPolicyChangeHandler creates a new PolicyChange handler.
 func NewPolicyChangeHandler(
 	log *logger.Logger,
-	agentInfo *info.AgentInfo,
+	agentInfo info.Agent,
 	config *configuration.Configuration,
 	store storage.Store,
 	ch chan coordinator.ConfigChange,
@@ -264,7 +264,7 @@ func clientEqual(k1 remote.Config, k2 remote.Config) bool {
 	return true
 }
 
-func fleetToReader(agentInfo *info.AgentInfo, cfg *configuration.Configuration) (io.Reader, error) {
+func fleetToReader(agentInfo info.Agent, cfg *configuration.Configuration) (io.Reader, error) {
 	configToStore := map[string]interface{}{
 		"fleet": cfg.Fleet,
 		"agent": map[string]interface{}{

internal/pkg/agent/application/actions/handlers/handler_action_policy_change_test.go

@@ -34,10 +34,7 @@ func TestPolicyChange(t *testing.T) {
 	log, _ := logger.New("", false)
 	ack := noopacker.New()
 
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	agentInfo, _ := info.NewAgentInfo(ctx, true)
+	agentInfo := &info.AgentInfo{}
 	nullStore := &storage.NullStore{}
 
 	t.Run("Receive a config change and successfully emits a raw configuration", func(t *testing.T) {
@@ -63,10 +60,8 @@ func TestPolicyChange(t *testing.T) {
 
 func TestPolicyAcked(t *testing.T) {
 	log, _ := logger.New("", false)
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
 
-	agentInfo, _ := info.NewAgentInfo(ctx, true)
+	agentInfo := &info.AgentInfo{}
 	nullStore := &storage.NullStore{}
 
 	t.Run("Config change should ACK", func(t *testing.T) {

internal/pkg/agent/application/actions/handlers/handler_action_settings.go

@@ -20,14 +20,14 @@ import (
 // Settings handles settings change coming from fleet and updates log level.
 type Settings struct {
 	log       *logger.Logger
-	agentInfo *info.AgentInfo
+	agentInfo info.Agent
 	coord     *coordinator.Coordinator
 }
 
 // NewSettings creates a new Settings handler.
 func NewSettings(
 	log *logger.Logger,
-	agentInfo *info.AgentInfo,
+	agentInfo info.Agent,
 	coord *coordinator.Coordinator,
 ) *Settings {
 	return &Settings{

internal/pkg/agent/application/actions/handlers/handler_action_upgrade_test.go

@@ -63,7 +63,8 @@ func TestUpgradeHandler(t *testing.T) {
 	defer cancel()
 
 	log, _ := logger.New("", false)
-	agentInfo, _ := info.NewAgentInfo(ctx, true)
+
+	agentInfo := &info.AgentInfo{}
 	msgChan := make(chan string)
 
 	// Create and start the coordinator
@@ -95,7 +96,8 @@ func TestUpgradeHandlerSameVersion(t *testing.T) {
 	defer cancel()
 
 	log, _ := logger.New("", false)
-	agentInfo, _ := info.NewAgentInfo(ctx, true)
+
+	agentInfo := &info.AgentInfo{}
 	msgChan := make(chan string)
 
 	// Create and start the Coordinator
@@ -129,7 +131,8 @@ func TestUpgradeHandlerNewVersion(t *testing.T) {
 	defer cancel()
 
 	log, _ := logger.New("", false)
-	agentInfo, _ := info.NewAgentInfo(ctx, true)
+
+	agentInfo := &info.AgentInfo{}
 	msgChan := make(chan string)
 
 	// Create and start the Coordinator

internal/pkg/agent/application/application.go

@@ -41,7 +41,7 @@ func New(
 	log *logger.Logger,
 	baseLogger *logger.Logger,
 	logLevel logp.Level,
-	agentInfo *info.AgentInfo,
+	agentInfo info.Agent,
 	reexec coordinator.ReExecManager,
 	tracer *apm.Tracer,
 	testingMode bool,

internal/pkg/agent/application/coordinator/coordinator.go

@@ -173,7 +173,7 @@ type configReloader interface {
 // All configuration changes, update variables, and upgrade actions are managed and controlled by the coordinator.
 type Coordinator struct {
 	logger    *logger.Logger
-	agentInfo *info.AgentInfo
+	agentInfo info.Agent
 	isManaged bool
 
 	cfg   *configuration.Configuration
@@ -322,7 +322,7 @@ type UpdateComponentChange struct {
 }
 
 // New creates a new coordinator.
-func New(logger *logger.Logger, cfg *configuration.Configuration, logLevel logp.Level, agentInfo *info.AgentInfo, specs component.RuntimeSpecs, reexecMgr ReExecManager, upgradeMgr UpgradeManager, runtimeMgr RuntimeManager, configMgr ConfigManager, varsMgr VarsManager, caps capabilities.Capabilities, monitorMgr MonitorManager, isManaged bool, modifiers ...ComponentsModifier) *Coordinator {
+func New(logger *logger.Logger, cfg *configuration.Configuration, logLevel logp.Level, agentInfo info.Agent, specs component.RuntimeSpecs, reexecMgr ReExecManager, upgradeMgr UpgradeManager, runtimeMgr RuntimeManager, configMgr ConfigManager, varsMgr VarsManager, caps capabilities.Capabilities, monitorMgr MonitorManager, isManaged bool, modifiers ...ComponentsModifier) *Coordinator {
 	var fleetState cproto.State
 	var fleetMessage string
 	if !isManaged {

internal/pkg/agent/application/coordinator/diagnostics_test.go

@@ -409,10 +409,9 @@ components:
 	assert.YAMLEq(t, expected, string(result), "components-actual diagnostic returned unexpected value")
 }
 
+// TestDiagnosticState creates a coordinator with a test state and verify that
+// the state diagnostic reports it.
 func TestDiagnosticState(t *testing.T) {
-	// Create a coordinator with a test state and verify that the state
-	// diagnostic reports it
-
 	now := time.Now().UTC()
 	state := State{
 		State:        agentclient.Starting,
@@ -427,7 +426,8 @@ func TestDiagnosticState(t *testing.T) {
 					State:   client.UnitStateDegraded,
 					Message: "degraded message",
 					VersionInfo: runtime.ComponentVersionInfo{
-						Name: "version name",
+						Name:      "version name",
+						BuildHash: "a-build-hash",
 					},
 				},
 			},
@@ -461,6 +461,7 @@ components:
       units: {}
       version_info:
         name: "version name"
+        build_hash: "a-build-hash"
 upgrade_details:
   target_version: 8.12.0
   state: UPG_DOWNLOADING
@@ -503,7 +504,8 @@ func TestDiagnosticStateForAPM(t *testing.T) {
 					State:   client.UnitStateDegraded,
 					Message: "degraded message",
 					VersionInfo: runtime.ComponentVersionInfo{
-						Name: "version name",
+						Name:      "version name",
+						BuildHash: "a-build-hash",
 					},
 					Component: &proto.Component{
 						ApmConfig: &proto.APMConfig{
@@ -540,6 +542,7 @@ components:
       units: {}
       version_info:
         name: "version name"
+        build_hash: "a-build-hash"
       component:
         apmconfig:
           elastic:

internal/pkg/agent/application/fleet_server_bootstrap.go

@@ -102,7 +102,7 @@ func FleetServerComponentModifier(serverCfg *configuration.FleetServerConfig) co
 
 // InjectFleetConfigComponentModifier The modifier that injects the fleet configuration for the components
 // that need to be able to connect to fleet server.
-func InjectFleetConfigComponentModifier(fleetCfg *configuration.FleetAgentConfig, agentInfo *info.AgentInfo) coordinator.ComponentsModifier {
+func InjectFleetConfigComponentModifier(fleetCfg *configuration.FleetAgentConfig, agentInfo info.Agent) coordinator.ComponentsModifier {
 	return func(comps []component.Component, cfg map[string]interface{}) ([]component.Component, error) {
 		hostsStr := fleetCfg.Client.GetHosts()
 		fleetHosts := make([]interface{}, 0, len(hostsStr))

internal/pkg/agent/application/info/agent_info.go

@@ -11,6 +11,29 @@ import (
 	"github.com/elastic/elastic-agent/pkg/core/logger"
 )
 
+type Agent interface {
+	// AgentID returns an agent identifier.
+	AgentID() string
+
+	// Headers returns custom headers used to communicate with elasticsearch.
+	Headers() map[string]string
+
+	// LogLevel retrieves a log level.
+	LogLevel() string
+
+	// ReloadID reloads agent info ID from configuration file.
+	ReloadID(ctx context.Context) error
+
+	// SetLogLevel updates log level of agent.
+	SetLogLevel(ctx context.Context, level string) error
+
+	// Snapshot returns if this version is a snapshot.
+	Snapshot() bool
+
+	// Version returns the version for this Agent.
+	Version() string
+}
+
 // AgentInfo is a collection of information about agent.
 type AgentInfo struct {
 	agentID  string

internal/pkg/agent/application/managed_mode.go

@@ -41,7 +41,7 @@ const dispatchFlushInterval = time.Minute * 5
 
 type managedConfigManager struct {
 	log                  *logger.Logger
-	agentInfo            *info.AgentInfo
+	agentInfo            info.Agent
 	cfg                  *configuration.Configuration
 	client               *remote.Client
 	store                storage.Store
@@ -60,7 +60,7 @@ type managedConfigManager struct {
 func newManagedConfigManager(
 	ctx context.Context,
 	log *logger.Logger,
-	agentInfo *info.AgentInfo,
+	agentInfo info.Agent,
 	cfg *configuration.Configuration,
 	storeSaver storage.Store,
 	runtime *runtime.Manager,

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

@@ -69,15 +69,15 @@ type BeatsMonitor struct {
 	enabled         bool // feature flag disabling whole v1 monitoring story
 	config          *monitoringConfig
 	operatingSystem string
-	agentInfo       *info.AgentInfo
+	agentInfo       info.Agent
 }
 
 type monitoringConfig struct {
 	C *monitoringCfg.MonitoringConfig `config:"agent.monitoring"`
 }
 
 // New creates a new BeatsMonitor instance.
-func New(enabled bool, operatingSystem string, cfg *monitoringCfg.MonitoringConfig, agentInfo *info.AgentInfo) *BeatsMonitor {
+func New(enabled bool, operatingSystem string, cfg *monitoringCfg.MonitoringConfig, agentInfo info.Agent) *BeatsMonitor {
 	return &BeatsMonitor{
 		enabled: enabled,
 		config: &monitoringConfig{
@@ -914,7 +914,7 @@ func (b *BeatsMonitor) injectMetricsInput(cfg map[string]interface{}, componentI
 	return nil
 }
 
-func createProcessorsForJSONInput(name string, compID, monitoringNamespace string, agentInfo *info.AgentInfo) []interface{} {
+func createProcessorsForJSONInput(name string, compID, monitoringNamespace string, agentInfo info.Agent) []interface{} {
 	return []interface{}{
 		map[string]interface{}{
 			"add_fields": map[string]interface{}{

internal/pkg/agent/application/upgrade/upgrade.go

@@ -53,7 +53,7 @@ var ErrSameVersion = errors.New("upgrade did not occur because its the same vers
 type Upgrader struct {
 	log            *logger.Logger
 	settings       *artifact.Config
-	agentInfo      *info.AgentInfo
+	agentInfo      info.Agent
 	upgradeable    bool
 	fleetServerURI string
 	markerWatcher  MarkerWatcher
@@ -67,7 +67,7 @@ func IsUpgradeable() bool {
 }
 
 // NewUpgrader creates an upgrader which is capable of performing upgrade operation
-func NewUpgrader(log *logger.Logger, settings *artifact.Config, agentInfo *info.AgentInfo) (*Upgrader, error) {
+func NewUpgrader(log *logger.Logger, settings *artifact.Config, agentInfo info.Agent) (*Upgrader, error) {
 	return &Upgrader{
 		log:           log,
 		settings:      settings,

internal/pkg/agent/storage/encrypted_disk_store.go

@@ -164,7 +164,7 @@ func (d *EncryptedDiskStore) Load() (rc io.ReadCloser, err error) {
 	fd, err := os.OpenFile(d.target, os.O_RDONLY, perms)
 	if err != nil {
 		if errors.Is(err, os.ErrNotExist) {
-			// If file doesn't exists, return empty reader closer
+			// If file doesn't exist, return empty reader closer
 			return io.NopCloser(bytes.NewReader([]byte{})), nil
 		}
 		return nil, errors.New(err,

internal/pkg/agent/vault/vault_notdarwin.go

@@ -96,18 +96,25 @@ func New(ctx context.Context, path string, opts ...OptionFunc) (v *Vault, err er
 func (v *Vault) Set(ctx context.Context, key string, data []byte) (err error) {
 	enc, err := v.encrypt(data)
 	if err != nil {
-		return err
+		return fmt.Errorf("vault Set: could not encrypt key: %w", err)
 	}
 
 	err = v.tryLock(ctx)
 	if err != nil {
-		return err
+		return fmt.Errorf("vault Set: could acquire lock: %w", err)
 	}
 	defer func() {
 		err = v.unlockAndJoinErrors(err)
+		if err != nil {
+			err = fmt.Errorf("vault Set: unlockAndJoinErrors failed: %w", err)
+		}
 	}()
 
-	return writeFile(v.filepathFromKey(key), enc)
+	err = writeFile(v.filepathFromKey(key), enc)
+	if err != nil {
+		return fmt.Errorf("vaukt: could not write key to file: %w", err)
+	}
+	return nil
 }
 
 // Get retrieves the key from the vault store

pkg/component/runtime/command.go

@@ -420,7 +420,7 @@ func (c *commandRuntime) stop(ctx context.Context) error {
 
 func (c *commandRuntime) startWatcher(info *process.Info, comm Communicator) {
 	go func() {
-		err := comm.WriteConnInfo(info.Stdin)
+		err := comm.WriteStartUpInfo(info.Stdin)
 		if err != nil {
 			_, _ = c.logErr.Write([]byte(fmt.Sprintf("Failed: failed to provide connection information to spawned pid '%d': %s", info.PID, err)))
 			// kill instantly

pkg/component/runtime/conn_info_server.go

@@ -46,7 +46,7 @@ func newConnInfoServer(log *logger.Logger, comm Communicator, port int) (*connIn
 				break
 			}
 			log.Debugf("client connected, sending connection info")
-			err = comm.WriteConnInfo(conn)
+			err = comm.WriteStartUpInfo(conn)
 			if err != nil {
 				if !errors.Is(err, io.EOF) {
 					log.Errorf("failed write conn info: %v", err)

pkg/component/runtime/conn_info_server_test.go

@@ -44,7 +44,7 @@ func newMockCommunicator() *mockCommunicator {
 	}
 }
 
-func (c *mockCommunicator) WriteConnInfo(w io.Writer, services ...client.Service) error {
+func (c *mockCommunicator) WriteStartUpInfo(w io.Writer, services ...client.Service) error {
 	infoBytes, err := protobuf.Marshal(c.startupInfo)
 	if err != nil {
 		return fmt.Errorf("failed to marshal connection information: %w", err)