[Ingest Manager] Fix failing installation on windows 7

libbeat/common/backoff/exponential.go

@@ -21,8 +21,8 @@ import (
 	"time"
 )
 
-// ExpBackoff exponential backoff, will wait an initial time and exponentialy
-// increases the wait time up to a predefined maximun. Resetting Backoff will reset the next sleep
+// ExpBackoff exponential backoff, will wait an initial time and exponentially
+// increases the wait time up to a predefined maximum. Resetting Backoff will reset the next sleep
 // timer to the initial backoff duration.
 type ExpBackoff struct {
 	duration time.Duration

libbeat/common/file/helper_windows.go

@@ -19,6 +19,7 @@ package file
 
 import (
 	"os"
+	"path/filepath"
 )
 
 // SafeFileRotate safely rotates an existing file under path and replaces it with the tempfile
@@ -40,5 +41,13 @@ func SafeFileRotate(path, tempfile string) error {
 	if e = os.Rename(tempfile, path); e != nil {
 		return e
 	}
+
+	// sync all files
+	parent := filepath.Dir(path)
+	if f, err := os.OpenFile(parent, os.O_SYNC|os.O_RDWR, 0755); err == nil {
+		f.Sync()
+		f.Close()
+	}
+
 	return nil
 }

x-pack/elastic-agent/CHANGELOG.asciidoc

@@ -38,6 +38,7 @@
 - Fix: Successfully installed and enrolled agent running standalone{pull}[24128]24128
 - Make installer atomic on windows {pull}[24253]24253
 - Remove installed services on agent uninstall {pull}[24151]24151
+- Fix failing installation on windows 7 {pull}[24387]24387
 - Fix windows installer during enroll {pull}[24343]24343
 
 ==== New features

x-pack/elastic-agent/pkg/agent/application/enroll_cmd.go

@@ -22,7 +22,6 @@ import (
 	"github.com/elastic/beats/v7/libbeat/common/backoff"
 
 	"github.com/elastic/beats/v7/libbeat/common/transport/tlscommon"
-	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/info"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/paths"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/control/client"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/control/proto"
@@ -234,8 +233,9 @@ func (c *EnrollCmd) fleetServerBootstrap(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	if err := c.configStore.Save(reader); err != nil {
-		return errors.New(err, "could not save fleet server bootstrap information", errors.TypeFilesystem)
+
+	if err := safelyStoreAgentInfo(c.configStore, reader); err != nil {
+		return err
 	}
 
 	if agentRunning {
@@ -405,11 +405,7 @@ func (c *EnrollCmd) enroll(ctx context.Context) error {
 		return err
 	}
 
-	if err := c.configStore.Save(reader); err != nil {
-		return errors.New(err, "could not save enrollment information", errors.TypeFilesystem)
-	}
-
-	if _, err := info.NewAgentInfo(); err != nil {
+	if err := safelyStoreAgentInfo(c.configStore, reader); err != nil {
 		return err
 	}
 
@@ -557,3 +553,31 @@ func getAppFromStatus(status *client.AgentStatus, name string) *client.Applicati
 	}
 	return nil
 }
+
+func safelyStoreAgentInfo(s saver, reader io.Reader) error {
+	err := storeAgentInfo(s, reader)
+	signal := make(chan struct{})
+	backExp := backoff.NewExpBackoff(signal, 200*time.Millisecond, 15*time.Second)
+
+	for err != nil {
+		backExp.Wait()
+		err = storeAgentInfo(s, reader)
+	}
+
+	close(signal)
+	return err
+}
+
+func storeAgentInfo(s saver, reader io.Reader) error {
+	fileLock := paths.AgentConfigFileLock()
+	if err := fileLock.TryLock(); err != nil {
+		return err
+	}
+	defer fileLock.Unlock()
+
+	if err := s.Save(reader); err != nil {
+		return errors.New(err, "could not save enrollment information", errors.TypeFilesystem)
+	}
+
+	return nil
+}

x-pack/elastic-agent/pkg/agent/application/locker.go → x-pack/elastic-agent/pkg/agent/application/filelock/locker.go

@@ -2,7 +2,7 @@
 // or more contributor license agreements. Licensed under the Elastic License;
 // you may not use this file except in compliance with the Elastic License.
 
-package application
+package filelock
 
 import (
 	"fmt"

x-pack/elastic-agent/pkg/agent/application/locker_test.go → x-pack/elastic-agent/pkg/agent/application/filelock/locker_test.go

@@ -2,7 +2,7 @@
 // or more contributor license agreements. Licensed under the Elastic License;
 // you may not use this file except in compliance with the Elastic License.
 
-package application
+package filelock
 
 import (
 	"io/ioutil"

x-pack/elastic-agent/pkg/agent/application/info/agent_id.go

@@ -8,10 +8,12 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"time"
 
 	"github.com/gofrs/uuid"
 	"gopkg.in/yaml.v2"
 
+	"github.com/elastic/beats/v7/libbeat/common/backoff"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/paths"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/errors"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/storage"
@@ -34,7 +36,7 @@ type ioStore interface {
 
 // updateLogLevel updates log level and persists it to disk.
 func updateLogLevel(level string) error {
-	ai, err := loadAgentInfo(false, defaultLogLevel)
+	ai, err := loadAgentInfoWithBackoff(false, defaultLogLevel)
 	if err != nil {
 		return err
 	}
@@ -60,31 +62,6 @@ func generateAgentID() (string, error) {
 	return uid.String(), nil
 }
 
-func loadAgentInfo(forceUpdate bool, logLevel string) (*persistentAgentInfo, error) {
-	agentConfigFile := paths.AgentConfigFile()
-	s := storage.NewDiskStore(agentConfigFile)
-
-	agentinfo, err := getInfoFromStore(s, logLevel)
-	if err != nil {
-		return nil, err
-	}
-
-	if agentinfo != nil && !forceUpdate && agentinfo.ID != "" {
-		return agentinfo, nil
-	}
-
-	agentinfo.ID, err = generateAgentID()
-	if err != nil {
-		return nil, err
-	}
-
-	if err := updateAgentInfo(s, agentinfo); err != nil {
-		return nil, errors.New(err, "storing generated agent id", errors.TypeFilesystem)
-	}
-
-	return agentinfo, nil
-}
-
 func getInfoFromStore(s ioStore, logLevel string) (*persistentAgentInfo, error) {
 	agentConfigFile := paths.AgentConfigFile()
 	reader, err := s.Load()
@@ -150,6 +127,19 @@ func updateAgentInfo(s ioStore, agentInfo *persistentAgentInfo) error {
 		return errors.New(err, "failed to unpack stored config to map")
 	}
 
+	// best effort to keep the ID
+	if agentInfoSubMap, found := configMap[agentInfoKey]; found {
+		if cc, err := config.NewConfigFrom(agentInfoSubMap); err == nil {
+			pid := &persistentAgentInfo{}
+			err := cc.Unpack(&pid)
+			if err == nil && pid.ID != agentInfo.ID {
+				// if our id is different (we just generated it)
+				// keep the one present in the file
+				agentInfo.ID = pid.ID
+			}
+		}
+	}
+
 	configMap[agentInfoKey] = agentInfo
 
 	r, err := yamlToReader(configMap)
@@ -167,3 +157,57 @@ func yamlToReader(in interface{}) (io.Reader, error) {
 	}
 	return bytes.NewReader(data), nil
 }
+
+func loadAgentInfoWithBackoff(forceUpdate bool, logLevel string) (*persistentAgentInfo, error) {
+	ai, err := loadAgentInfo(forceUpdate, logLevel)
+	signal := make(chan struct{})
+	backExp := backoff.NewExpBackoff(signal, 200*time.Millisecond, 5*time.Second)
+
+	for err != nil {
+		backExp.Wait()
+		ai, err = loadAgentInfo(forceUpdate, logLevel)
+	}
+
+	close(signal)
+	return ai, err
+}
+
+func loadAgentInfo(forceUpdate bool, logLevel string) (*persistentAgentInfo, error) {
+	idLock := paths.AgentConfigFileLock()
+	if err := idLock.TryLock(); err != nil {
+		return nil, err
+	}
+	defer idLock.Unlock()
+
+	agentConfigFile := paths.AgentConfigFile()
+	s := storage.NewDiskStore(agentConfigFile)
+
+	agentinfo, err := getInfoFromStore(s, logLevel)
+	if err != nil {
+		return nil, err
+	}
+
+	if agentinfo != nil && !forceUpdate && agentinfo.ID != "" {
+		return agentinfo, nil
+	}
+
+	if err := updateID(agentinfo, s); err != nil {
+		return nil, err
+	}
+
+	return agentinfo, nil
+}
+
+func updateID(agentInfo *persistentAgentInfo, s ioStore) error {
+	var err error
+	agentInfo.ID, err = generateAgentID()
+	if err != nil {
+		return err
+	}
+
+	if err := updateAgentInfo(s, agentInfo); err != nil {
+		return errors.New(err, "storing generated agent id", errors.TypeFilesystem)
+	}
+
+	return nil
+}

x-pack/elastic-agent/pkg/agent/application/info/agent_info.go

@@ -21,7 +21,7 @@ type AgentInfo struct {
 // If agent config file does not exist it gets created.
 // Initiates log level to predefined value.
 func NewAgentInfoWithLog(level string) (*AgentInfo, error) {
-	agentInfo, err := loadAgentInfo(false, level)
+	agentInfo, err := loadAgentInfoWithBackoff(false, level)
 	if err != nil {
 		return nil, err
 	}
@@ -51,6 +51,16 @@ func (i *AgentInfo) LogLevel(level string) error {
 	return nil
 }
 
+// ReloadID reloads agent info ID from configuration file.
+func (i *AgentInfo) ReloadID() error {
+	newInfo, err := NewAgentInfoWithLog(i.logLevel)
+	if err != nil {
+		return err
+	}
+	i.agentID = newInfo.agentID
+	return nil
+}
+
 // AgentID returns an agent identifier.
 func (i *AgentInfo) AgentID() string {
 	return i.agentID

x-pack/elastic-agent/pkg/agent/application/managed_mode.go

@@ -278,6 +278,11 @@ func (m *Managed) Start() error {
 		return nil
 	}
 
+	// reload ID because of win7 sync issue
+	if err := m.agentInfo.ReloadID(); err != nil {
+		return err
+	}
+
 	err := m.upgrader.Ack(m.bgContext)
 	if err != nil {
 		m.log.Warnf("failed to ack update %v", err)

x-pack/elastic-agent/pkg/agent/application/paths/files.go

@@ -5,7 +5,10 @@
 package paths
 
 import (
+	"fmt"
 	"path/filepath"
+
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/filelock"
 )
 
 // defaultAgentConfigFile is a name of file used to store agent information
@@ -21,6 +24,14 @@ func AgentConfigFile() string {
 	return filepath.Join(Config(), defaultAgentConfigFile)
 }
 
+// AgentConfigFileLock is a locker for agent config file updates.
+func AgentConfigFileLock() *filelock.AppLocker {
+	return filelock.NewAppLocker(
+		Config(),
+		fmt.Sprintf("%s.lock", defaultAgentConfigFile),
+	)
+}
+
 // AgentCapabilitiesPath is a name of file used to store agent capabilities
 func AgentCapabilitiesPath() string {
 	return filepath.Join(Config(), defaultAgentCapabilitiesFile)

x-pack/elastic-agent/pkg/agent/cmd/install.go

@@ -9,12 +9,11 @@ import (
 	"os"
 	"os/exec"
 
-	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application"
-	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/paths"
-
 	"github.com/spf13/cobra"
 
 	c "github.com/elastic/beats/v7/libbeat/common/cli"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/filelock"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/paths"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/install"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/warn"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/cli"
@@ -58,9 +57,9 @@ func installCmd(streams *cli.IOStreams, cmd *cobra.Command, flags *globalFlags,
 	}
 
 	// check the lock to ensure that elastic-agent is not already running in this directory
-	locker := application.NewAppLocker(paths.Data(), agentLockFileName)
+	locker := filelock.NewAppLocker(paths.Data(), agentLockFileName)
 	if err := locker.TryLock(); err != nil {
-		if err == application.ErrAppAlreadyRunning {
+		if err == filelock.ErrAppAlreadyRunning {
 			return fmt.Errorf("cannot perform installation as Elastic Agent is already running from this directory")
 		}
 		return err

x-pack/elastic-agent/pkg/agent/cmd/run.go

@@ -26,6 +26,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/beats/v7/libbeat/service"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/filelock"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/info"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/paths"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/reexec"
@@ -63,7 +64,7 @@ func run(flags *globalFlags, streams *cli.IOStreams) error { // Windows: Mark se
 	// This must be the first deferred cleanup task (last to execute).
 	defer service.NotifyTermination()
 
-	locker := application.NewAppLocker(paths.Data(), agentLockFileName)
+	locker := filelock.NewAppLocker(paths.Data(), agentLockFileName)
 	if err := locker.TryLock(); err != nil {
 		return err
 	}

x-pack/elastic-agent/pkg/agent/cmd/watch.go

@@ -15,7 +15,7 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/filelock"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/paths"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/application/upgrade"
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/configuration"
@@ -67,9 +67,9 @@ func watchCmd(streams *cli.IOStreams, cmd *cobra.Command, flags *globalFlags, ar
 		return nil
 	}
 
-	locker := application.NewAppLocker(paths.Top(), watcherLockFile)
+	locker := filelock.NewAppLocker(paths.Top(), watcherLockFile)
 	if err := locker.TryLock(); err != nil {
-		if err == application.ErrAppAlreadyRunning {
+		if err == filelock.ErrAppAlreadyRunning {
 			log.Debugf("exiting, lock already exists")
 			return nil
 		}