Fix startup with failing configuration (#26126)

Fix startup with failing configuration (#26126)

x-pack/elastic-agent/CHANGELOG.next.asciidoc

@@ -68,6 +68,7 @@
 - Handle case where policy doesn't contain Fleet connection information {pull}25707[25707]
 - Fix fleet-server.yml spec to not overwrite existing keys {pull}25741[25741]
 - Agent sends wrong log level to Endpoint {issue}25583[25583]
+- Fix startup with failing configuration {pull}26057[26057]
 - Change timestamp in elatic-agent-json.log to use UTC {issue}25391[25391]
 
 ==== New features

x-pack/elastic-agent/pkg/agent/operation/operation_config.go

@@ -10,6 +10,7 @@ import (
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/configuration"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/errors"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/logger"
+ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/plugin/process"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/state"
 )
 
@@ -53,7 +54,13 @@ func (o *operationConfig) Check(_ context.Context, _ Application) (bool, error)
 func (o *operationConfig) Run(ctx context.Context, application Application) (err error) {
  defer func() {
  if err != nil {
- application.SetState(state.Failed, err.Error(), nil)
+ // application failed to apply config but is running.
+ s := state.Degraded
+ if errors.Is(err, process.ErrAppNotRunning) {
+ s = state.Failed
+ }
+
+ application.SetState(s, err.Error(), nil)
  }
  }()
  return application.Configure(ctx, o.cfg)

x-pack/elastic-agent/pkg/core/plugin/process/app.go

@@ -30,6 +30,7 @@ import (
 var (
  // ErrAppNotRunning is returned when configuration is performed on not running application.
  ErrAppNotRunning = errors.New("application is not running", errors.TypeApplication)
+ procExitTimeout = 10 * time.Second
 )
 
 // Application encapsulates a concrete application ran by elastic-agent e.g Beat.
@@ -47,6 +48,7 @@ type Application struct {
  tag app.Taggable
  state state.State
  reporter state.Reporter
+ watchClosers map[int]context.CancelFunc
 
  uid int
  gid int
@@ -105,6 +107,7 @@ func NewApplication(
  uid: uid,
  gid: gid,
  statusReporter: statusController.RegisterApp(id, appName),
+ watchClosers: make(map[int]context.CancelFunc),
  }, nil
 }
 
@@ -159,6 +162,8 @@ func (a *Application) Stop() {
 
  a.srvState = nil
  if a.state.ProcessInfo != nil {
+ // stop and clean watcher
+ a.stopWatcher(a.state.ProcessInfo)
  if err := a.state.ProcessInfo.Process.Signal(stopSig); err == nil {
  // no error on signal, so wait for it to stop
  _, _ = a.state.ProcessInfo.Process.Wait()
@@ -192,33 +197,52 @@ func (a *Application) watch(ctx context.Context, p app.Taggable, proc *process.I
  case ps := <-a.waitProc(proc.Process):
  procState = ps
  case <-a.bgContext.Done():
- a.Stop()
+ return
+ case <-ctx.Done():
+ // closer called
  return
  }
 
  a.appLock.Lock()
+ defer a.appLock.Unlock()
  if a.state.ProcessInfo != proc {
  // already another process started, another watcher is watching instead
- a.appLock.Unlock()
+ gracefulKill(proc)
+ return
+ }
+
+ // stop the watcher
+ a.stopWatcher(a.state.ProcessInfo)
+
+ // was already stopped by Stop, do not restart
+ if a.state.Status == state.Stopped {
  return
  }
+
  a.state.ProcessInfo = nil
  srvState := a.srvState
 
  if srvState == nil || srvState.Expected() == proto.StateExpected_STOPPING {
- a.appLock.Unlock()
  return
  }
 
  msg := fmt.Sprintf("exited with code: %d", procState.ExitCode())
- a.setState(state.Crashed, msg, nil)
+ a.setState(state.Restarting, msg, nil)
 
  // it was a crash
- a.start(ctx, p, cfg)
- a.appLock.Unlock()
+ a.start(ctx, p, cfg, true)
  }()
 }
 
+func (a *Application) stopWatcher(procInfo *process.Info) {
+ if procInfo != nil {
+ if closer, ok := a.watchClosers[procInfo.PID]; ok {
+ closer()
+ delete(a.watchClosers, procInfo.PID)
+ }
+ }
+}
+
 func (a *Application) waitProc(proc *os.Process) <-chan *os.ProcessState {
  resChan := make(chan *os.ProcessState)
 
@@ -250,3 +274,31 @@ func (a *Application) setState(s state.Status, msg string, payload map[string]in
 func (a *Application) cleanUp() {
  a.monitor.Cleanup(a.desc.Spec(), a.pipelineID)
 }
+
+func gracefulKill(proc *process.Info) {
+ if proc == nil || proc.Process == nil {
+ return
+ }
+
+ // send stop signal to request stop
+ proc.Stop()
+
+ var wg sync.WaitGroup
+ doneChan := make(chan struct{})
+ wg.Add(1)
+ go func() {
+ wg.Done()
+ _, _ = proc.Process.Wait()
+ close(doneChan)
+ }()
+
+ // wait for awaiter
+ wg.Wait()
+
+ // kill in case it's still running after timeout
+ select {
+ case <-doneChan:
+ case <-time.After(procExitTimeout):
+ _ = proc.Process.Kill()
+ }
+}

x-pack/elastic-agent/pkg/core/plugin/process/start.go

@@ -26,20 +26,37 @@ func (a *Application) Start(ctx context.Context, t app.Taggable, cfg map[string]
  a.appLock.Lock()
  defer a.appLock.Unlock()
 
- return a.start(ctx, t, cfg)
+ return a.start(ctx, t, cfg, false)
 }
 
 // Start starts the application without grabbing the lock.
-func (a *Application) start(ctx context.Context, t app.Taggable, cfg map[string]interface{}) (err error) {
+func (a *Application) start(ctx context.Context, t app.Taggable, cfg map[string]interface{}, isRestart bool) (err error) {
  defer func() {
  if err != nil {
  // inject App metadata
  err = errors.New(err, errors.M(errors.MetaKeyAppName, a.name), errors.M(errors.MetaKeyAppName, a.id))
  }
  }()
 
- // already started if not stopped or crashed
- if a.Started() {
+ // starting only if it's not running
+ // or if it is, then only in case it's restart and this call initiates from restart call
+ if a.Started() && a.state.Status != state.Restarting {
+ if a.state.ProcessInfo == nil {
+ // already started if not stopped or crashed
+ return nil
+ }
+
+ // in case app reported status it might still be running and failure timer
+ // in progress. Stop timer and stop failing process
+ a.stopFailedTimer()
+ a.stopWatcher(a.state.ProcessInfo)
+
+ // kill the process
+ _ = a.state.ProcessInfo.Process.Kill()
+ a.state.ProcessInfo = nil
+ }
+
+ if a.state.Status == state.Restarting && !isRestart {
  return nil
  }
 
@@ -69,7 +86,8 @@ func (a *Application) start(ctx context.Context, t app.Taggable, cfg map[string]
  if a.state.Status != state.Stopped {
  // restarting as it was previously in a different state
  a.setState(state.Restarting, "Restarting", nil)
- } else {
+ } else if a.state.Status != state.Restarting {
+ // keep restarting state otherwise it's starting
  a.setState(state.Starting, "Starting", nil)
  }
 
@@ -116,12 +134,15 @@ func (a *Application) start(ctx context.Context, t app.Taggable, cfg map[string]
  if err != nil {
  return err
  }
-
  // write connect info to stdin
  go a.writeToStdin(a.srvState, a.state.ProcessInfo.Stdin)
 
+ // create closer for watcher, used to terminate watcher without
+ // side effect of restarting process during shutdown
+ cancelCtx, cancel := context.WithCancel(ctx)
+ a.watchClosers[a.state.ProcessInfo.PID] = cancel
  // setup watcher
- a.watch(ctx, t, a.state.ProcessInfo, cfg)
+ a.watch(cancelCtx, t, a.state.ProcessInfo, cfg)
 
  return nil
 }

x-pack/elastic-agent/pkg/core/plugin/process/status.go

@@ -13,6 +13,7 @@ import (
 
  "github.com/elastic/elastic-agent-client/v7/pkg/proto"
 
+ "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/process"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/server"
  "github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/core/state"
 )
@@ -42,7 +43,8 @@ func (a *Application) OnStatusChange(s *server.ApplicationState, status proto.St
  _ = yaml.Unmarshal([]byte(s.Config()), &cfg)
 
  // start the failed timer
- a.startFailedTimer(cfg)
+ // pass process info to avoid killing new process spun up in a meantime
+ a.startFailedTimer(cfg, a.state.ProcessInfo)
  } else {
  a.stopFailedTimer()
  }
@@ -51,7 +53,7 @@ func (a *Application) OnStatusChange(s *server.ApplicationState, status proto.St
 // startFailedTimer starts a timer that will restart the application if it doesn't exit failed after a period of time.
 //
 // This does not grab the appLock, that must be managed by the caller.
-func (a *Application) startFailedTimer(cfg map[string]interface{}) {
+func (a *Application) startFailedTimer(cfg map[string]interface{}, proc *process.Info) {
  if a.restartCanceller != nil {
  // already have running failed timer; just update config
  a.restartConfig = cfg
@@ -74,7 +76,7 @@ func (a *Application) startFailedTimer(cfg map[string]interface{}) {
  case <-ctx.Done():
  return
  case <-t.C:
- a.restart()
+ a.restart(proc)
  }
  }()
 }
@@ -91,19 +93,31 @@ func (a *Application) stopFailedTimer() {
 }
 
 // restart restarts the application
-func (a *Application) restart() {
+func (a *Application) restart(proc *process.Info) {
  a.appLock.Lock()
  defer a.appLock.Unlock()
 
+ // stop the watcher
+ a.stopWatcher(proc)
+
  // kill the process
- if a.state.ProcessInfo != nil {
- _ = a.state.ProcessInfo.Process.Kill()
- a.state.ProcessInfo = nil
+ if proc != nil && proc.Process != nil {
+ _ = proc.Process.Kill()
+ }
+
+ if proc != a.state.ProcessInfo {
+ // we're restarting different process than actually running
+ // no need to start another one
+ return
  }
+
+ a.state.ProcessInfo = nil
+
  ctx := a.startContext
  tag := a.tag
 
- err := a.start(ctx, tag, a.restartConfig)
+ a.setState(state.Restarting, "", nil)
+ err := a.start(ctx, tag, a.restartConfig, true)
  if err != nil {
  a.setState(state.Crashed, fmt.Sprintf("failed to restart: %s", err), nil)
  }