Retry starting/installing service upon failure (#9313)

* Retry starting/installing service upon failure

* Add service restart delay

* Adding CHANGELOG entry

* [WIP] Debugging

* Add mutex to prevent data race

* [WIP] Debugging

* Remove max service restart attempts

* Attempt to (re)start service when timer fires as well

* Move serviceRestartDelay to struct

Author: ycombinator

changelog/fragments/1755283748-service-retry-install.yaml

@@ -0,0 +1,32 @@
+# 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: bug-fix
+
+# Change summary; a 80ish characters long description of the change.
+summary: Retry service start command upon failure with 30-second delay.
+
+# 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:
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# 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/9313
+
+# 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

pkg/component/runtime/conn_info_server.go

@@ -43,6 +43,7 @@ func newConnInfoServer(log *logger.Logger, comm Communicator, address string) (*
 	if err != nil {
 		return nil, fmt.Errorf("failed to start connection credentials listener: %w", err)
 	}
+	log.Debugf("started connection info server listener on %s", address)
 
 	s := &connInfoServer{log: log, listener: listener, stopTimeout: defaultStopTimeout}
 

pkg/component/runtime/conn_info_server_test.go

@@ -73,7 +73,7 @@ func (c *mockCommunicator) CheckinObserved() <-chan *proto.CheckinObserved {
 const testPort = 6788
 
 // Test Elastic Agent Connection Info sock
-const testSock = ".teaci.sock"
+const testSock = ".cteaci.sock"
 
 func getAddress(dir string, isLocal bool) string {
 	if isLocal {

pkg/component/runtime/service.go

@@ -58,6 +58,16 @@ type serviceRuntime struct {
 	executeServiceCommandImpl executeServiceCommandFunc
 
 	isLocal bool // true if rpc is domain socket, or named pipe
+
+	// The most recent mode received on actionCh. The mode will be either
+	// actionStart (indicating the process should be running, and should be
+	// created if it is not), or actionStop or actionTeardown (indicating that
+	// it should terminate).
+	actionState actionModeSigned
+
+	// serviceRestartDelay is the time duration the service runtime will wait before
+	// attempting to restart a service that failed to start.
+	serviceRestartDelay time.Duration
 }
 
 // newServiceRuntime creates a new command runtime for the provided component.
@@ -81,6 +91,7 @@ func newServiceRuntime(comp component.Component, logger *logger.Logger, isLocal
 		state:                     state,
 		executeServiceCommandImpl: executeServiceCommand,
 		isLocal:                   isLocal,
+		serviceRestartDelay:       30 * time.Second,
 	}
 
 	// Set initial state as STOPPED
@@ -208,6 +219,7 @@ func (s *serviceRuntime) Run(ctx context.Context, comm Communicator) (err error)
 			return ctx.Err()
 		case as := <-s.actionCh:
 			s.log.Debugf("got action %v for %s service", as.actionMode, s.name())
+			s.actionState = as
 			switch as.actionMode {
 			case actionStart:
 				// Initial state on start
@@ -253,8 +265,14 @@ func (s *serviceRuntime) Run(ctx context.Context, comm Communicator) (err error)
 						}
 					}
 
-					cisStop()
-					break
+					// Error is due to a fatal exit code. Schedule a restart of the service after a delay by resending
+					// the start action on the action channel
+					delay := s.serviceRestartDelay
+					s.log.Errorf("failed to start %s service, err: %v, restarting after waiting for %v", s.name(), err, delay)
+					time.AfterFunc(delay, func() {
+						s.actionCh <- as
+					})
+					continue
 				}
 
 				// Start check-in timer
@@ -290,8 +308,19 @@ func (s *serviceRuntime) Run(ctx context.Context, comm Communicator) (err error)
 				s.processCheckin(checkin, comm, &lastCheckin)
 			}
 		case <-checkinTimer.C:
-			s.checkStatus(s.checkinPeriod(), &lastCheckin, &missedCheckins)
-			checkinTimer.Reset(s.checkinPeriod())
+			if s.actionState.actionMode == actionStart {
+				if !s.isRunning() {
+					// not running, but should be running
+					if err := s.start(ctx); err != nil {
+						s.forceCompState(client.UnitStateFailed, fmt.Sprintf("service %s failed to start: %v", s.name(), err))
+						continue
+					}
+				} else {
+					// running and should be running
+					s.checkStatus(s.checkinPeriod(), &lastCheckin, &missedCheckins)
+					checkinTimer.Reset(s.checkinPeriod())
+				}
+			}
 		case <-teardownCheckinTimer.C:
 			s.log.Debugf("got tearing down timeout for %s service", s.name())
 			// Teardown timed out

pkg/component/runtime/service_test.go

@@ -280,7 +280,7 @@ func TestCISKeepsRunningOnNonFatalExitCodeFromStart(t *testing.T) {
 	log, logObs := loggertest.New("test")
 	const nonFatalExitCode = 99
 	const cisPort = 9999
-	const cisSocket = ".teaci.sock"
+	const cisSocket = ".steaci.sock"
 
 	// Make an Endpoint component for testing
 	endpoint := makeEndpointComponent(t, map[string]interface{}{})
@@ -371,6 +371,82 @@ func TestCISKeepsRunningOnNonFatalExitCodeFromStart(t *testing.T) {
 	}, 2*time.Second, 200*time.Millisecond)
 }
 
+// TestServiceStartRetry tests that the service runtime will
+// retry the service start command if it fails
+func TestServiceStartRetry(t *testing.T) {
+	log, logObs := loggertest.New("test")
+	const cisPort = 9999
+	const cisSocket = ".rteaci.sock"
+
+	// Make an Endpoint component for testing
+	endpoint := makeEndpointComponent(t, map[string]interface{}{})
+	endpoint.InputSpec.Spec.Service = &component.ServiceSpec{
+		CPort:   cisPort,
+		CSocket: cisSocket,
+		Log:     nil,
+		Operations: component.ServiceOperationsSpec{
+			Check:   &component.ServiceOperationsCommandSpec{},
+			Install: &component.ServiceOperationsCommandSpec{},
+		},
+		Timeouts: component.ServiceTimeoutSpec{},
+	}
+
+	// Create binary mocking Endpoint such that executing it will return
+	// the non-fatal exit code from the spec above.
+	endpoint.InputSpec.BinaryPath = mockEndpointBinary(t, 99)
+	endpoint.InputSpec.BinaryName = "endpoint"
+
+	t.Logf("mock binary path: %s\n", endpoint.InputSpec.BinaryPath)
+
+	// Create new service runtime with component
+	service, err := newServiceRuntime(endpoint, log, true)
+	require.NoError(t, err)
+
+	// Shorten service restart delay for testing
+	service.serviceRestartDelay = 2 * time.Second
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	comm := newMockCommunicator("")
+
+	// Observe component state
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-service.ch:
+			}
+		}
+	}()
+
+	// Run the service runtime
+	go func() {
+		err := service.Run(ctx, comm)
+		require.EqualError(t, err, context.Canceled.Error())
+	}()
+
+	// Start the service
+	service.actionCh <- actionModeSigned{
+		actionMode: actionStart,
+	}
+
+	expectedRestartLogMsg := fmt.Sprintf(
+		"failed to start endpoint service, err: %s, restarting after waiting for %v",
+		"failed install endpoint service: exit status 99", service.serviceRestartDelay,
+	)
+	require.Eventually(t, func() bool {
+		logs := logObs.TakeAll()
+		for _, l := range logs {
+			t.Logf("[%s] %s", l.Level, l.Message)
+			if l.Level == zapcore.ErrorLevel && l.Message == expectedRestartLogMsg {
+				return true
+			}
+		}
+		return false
+	}, service.serviceRestartDelay+1*time.Second, 500*time.Millisecond)
+}
+
 func mockEndpointBinary(t *testing.T, exitCode int) string {
 	// Build a mock Endpoint binary that can return a specific exit code.
 	outPath := filepath.Join(t.TempDir(), "mock_endpoint")