RSDK-11248 RSDK-11266 RSDK-11900 RSDK-11901 Add restart checking logic

cmd/viam-agent/main.go

@@ -213,7 +213,8 @@ func setupExitSignalHandling() (context.Context, context.CancelFunc) {
 			case syscall.SIGABRT:
 				fallthrough
 			case syscall.SIGTERM:
-				globalLogger.Info("exiting")
+				exitMsg := fmt.Sprintf("Signal received. %s will now exit to be restarted by service manager", agent.SubsystemName)
+				globalLogger.Infow(exitMsg, "signal", sig)
 				signal.Ignore(os.Interrupt, syscall.SIGTERM, syscall.SIGABRT) // keeping SIGQUIT for stack trace debugging
 				return
 

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/jessevdk/go-flags v1.6.1
 	github.com/nightlyone/lockfile v1.0.0
 	github.com/pkg/errors v0.9.1
+	github.com/samber/mo v1.16.0
 	github.com/schollz/progressbar/v3 v3.18.0
 	github.com/sergeymakinen/go-systemdconf/v2 v2.0.2
 	github.com/tidwall/jsonc v0.3.2
@@ -29,7 +30,7 @@ require (
 require (
 	github.com/cenkalti/backoff v2.2.1+incompatible // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect
 	github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f // indirect
 	github.com/dgottlieb/smarty-assertions v1.2.6 // indirect
@@ -72,15 +73,15 @@ require (
 	github.com/pion/stun v0.6.1 // indirect
 	github.com/pion/transport/v2 v2.2.10 // indirect
 	github.com/pion/turn/v2 v2.1.6 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
 	github.com/rs/cors v1.11.1 // indirect
 	github.com/saltosystems/winrt-go v0.0.0-20240509164145-4f7860a3bd2b // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/soypat/cyw43439 v0.0.0-20241116210509-ae1ce0e084c5 // indirect
 	github.com/soypat/seqs v0.0.0-20240527012110-1201bab640ef // indirect
 	github.com/srikrsna/protoc-gen-gotag v0.6.2 // indirect
-	github.com/stretchr/testify v1.10.0 // indirect
+	github.com/stretchr/testify v1.11.1 // indirect
 	github.com/tinygo-org/cbgo v0.0.4 // indirect
 	github.com/tinygo-org/pio v0.0.0-20231216154340-cd888eb58899 // indirect
 	github.com/viamrobotics/webrtc/v3 v3.99.10 // indirect

go.sum

@@ -98,8 +98,9 @@ github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7Do
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/daixiang0/gci v0.2.8/go.mod h1:+4dZ7TISfSmqfAGv59ePaHfNzgGtIkHAhhdKggP1JAc=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
+github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/decred/dcrd/crypto/blake256 v1.0.1/go.mod h1:2OfgNZ5wDpcsFmHmCK5gZTPcCXqlm2ArzUIkw9czNJo=
 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 h1:8UrgZ3GkP4i/CLijOJx79Yu+etlyjdBU4sfcs2WYQMs=
 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0/go.mod h1:v57UDF4pDQJcEfFUCRop3lJL149eHGSe9Jvczhzjo/0=
@@ -539,8 +540,9 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA=
 github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZI=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
+github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/polyfloyd/go-errorlint v0.0.0-20201127212506-19bd8db6546f/go.mod h1:wi9BfjxjF/bwiZ701TzmfKu6UKC357IOAtNr0Td0Lvw=
 github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
@@ -593,6 +595,8 @@ github.com/ryanrolds/sqlclosecheck v0.3.0/go.mod h1:1gREqxyTGR3lVtpngyFo3hZAgk0K
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/saltosystems/winrt-go v0.0.0-20240509164145-4f7860a3bd2b h1:du3zG5fd8snsFN6RBoLA7fpaYV9ZQIsyH9snlk2Zvik=
 github.com/saltosystems/winrt-go v0.0.0-20240509164145-4f7860a3bd2b/go.mod h1:CIltaIm7qaANUIvzr0Vmz71lmQMAIbGJ7cvgzX7FMfA=
+github.com/samber/mo v1.16.0 h1:qpEPCI63ou6wXlsNDMLE0IIN8A+devbGX/K1xdgr4b4=
+github.com/samber/mo v1.16.0/go.mod h1:DlgzJ4SYhOh41nP1L9kh9rDNERuf8IqWSAs+gj2Vxag=
 github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/sanposhiho/wastedassign v0.1.3/go.mod h1:LGpq5Hsv74QaqM47WtIsRSF/ik9kqk07kchgv66tLVE=
 github.com/schollz/progressbar/v3 v3.18.0 h1:uXdoHABRFmNIjUfte/Ex7WtuyVslrw2wVPQmCN62HpA=
@@ -661,8 +665,8 @@ github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o
 github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
-github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
 github.com/tdakkota/asciicheck v0.0.0-20200416200610-e657995f937b/go.mod h1:yHp0ai0Z9gUljN3o0xMhYJnH/IcvkdTBOX2fmJ93JEM=
 github.com/tetafro/godot v1.4.4/go.mod h1:FVDd4JuKliW3UgjswZfJfHq4vAx0bD/Jd5brJjGeaz4=

manager.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/url"
 	"os"
 	"regexp"
@@ -21,13 +22,18 @@ import (
 	"github.com/viamrobotics/agent/subsystems/viamserver"
 	"github.com/viamrobotics/agent/utils"
 	pb "go.viam.com/api/app/agent/v1"
+	apppb "go.viam.com/api/app/v1"
 	"go.viam.com/rdk/logging"
 	goutils "go.viam.com/utils"
 	"go.viam.com/utils/rpc"
 )
 
 const (
-	minimalCheckInterval  = time.Second * 5
+	// The minimal (and default) interval for checking for config updates via DeviceAgentConfig.
+	minimalDeviceAgentConfigCheckInterval = time.Second * 5
+	// The minimal (and default) interval for checking whether agent needs to be restarted.
+	minimalNeedsRestartCheckInterval = time.Second * 1
+
 	defaultNetworkTimeout = time.Second * 15
 	// stopAllTimeout must be lower than systemd subsystems/viamagent/viam-agent.service timeout of 4mins
 	// and higher than subsystems/viamserver/viamserver.go timeout of 2mins.
@@ -42,7 +48,6 @@ type Manager struct {
 
 	connMu      sync.RWMutex
 	conn        rpc.ClientConn
-	client      pb.AgentDeviceServiceClient
 	cloudConfig *logging.CloudConfig
 
 	logger      logging.Logger
@@ -209,7 +214,7 @@ func (m *Manager) SubsystemUpdates(ctx context.Context) {
 			m.logger.Warn(err)
 		}
 		if m.viamAgentNeedsRestart {
-			m.Exit()
+			m.Exit(fmt.Sprintf("A new version of %s has been installed", SubsystemName))
 			return
 		}
 	} else {
@@ -221,17 +226,19 @@ func (m *Manager) SubsystemUpdates(ctx context.Context) {
 		needRestartConfigChange := m.viamServer.Update(ctx, m.cfg)
 
 		if needRestart || needRestartConfigChange || m.viamServerNeedsRestart || m.viamAgentNeedsRestart {
-			if m.viamServer.(viamserver.RestartCheck).SafeToRestart(ctx) {
+			if m.viamServer.Property(ctx, viamserver.RestartPropertyRestartAllowed) {
+				m.logger.Infof("%s has allowed a restart; will restart", viamserver.SubsysName)
 				if err := m.viamServer.Stop(ctx); err != nil {
 					m.logger.Warn(err)
 				} else {
 					m.viamServerNeedsRestart = false
 				}
 				if m.viamAgentNeedsRestart {
-					m.Exit()
+					m.Exit(fmt.Sprintf("A new version of %s has been installed", SubsystemName))
 					return
 				}
 			} else {
+				m.logger.Warnf("%s has NOT allowed a restart; will NOT restart", viamserver.SubsysName)
 				m.viamServerNeedsRestart = true
 			}
 		}
@@ -280,26 +287,26 @@ func (m *Manager) SubsystemUpdates(ctx context.Context) {
 // CheckUpdates retrieves an updated config from the cloud, and then passes it to SubsystemUpdates().
 func (m *Manager) CheckUpdates(ctx context.Context) time.Duration {
 	defer utils.Recover(m.logger, nil)
-	m.logger.Debug("Checking cloud for update")
-	interval, err := m.GetConfig(ctx)
+	m.logger.Debug("Checking cloud for device agent config updates")
+	deviceAgentConfigCheckInterval, err := m.GetConfig(ctx)
 
-	if interval < minimalCheckInterval {
-		interval = minimalCheckInterval
+	if deviceAgentConfigCheckInterval < minimalDeviceAgentConfigCheckInterval {
+		deviceAgentConfigCheckInterval = minimalDeviceAgentConfigCheckInterval
 	}
 
 	// randomly fuzz the interval by +/- 5%
-	interval = utils.FuzzTime(interval, 0.05)
+	deviceAgentConfigCheckInterval = utils.FuzzTime(deviceAgentConfigCheckInterval, 0.05)
 
 	// we already log in all error cases inside GetConfig, so
 	// no need to log again.
 	if err != nil {
-		return interval
+		return deviceAgentConfigCheckInterval
 	}
 
 	// update and (re)start subsystems
 	m.SubsystemUpdates(ctx)
 
-	return interval
+	return deviceAgentConfigCheckInterval
 }
 
 func (m *Manager) setDebug(debug bool) {
@@ -380,13 +387,51 @@ func (m *Manager) SubsystemHealthChecks(ctx context.Context) {
 	}
 }
 
+// CheckIfNeedsRestart returns the check restart interval and whether the agent (and
+// therefore all its subsystems) has been forcibly restarted by app.
+func (m *Manager) CheckIfNeedsRestart(ctx context.Context) (time.Duration, bool) {
+	m.logger.Debug("Checking cloud for forced restarts")
+	if m.cloudConfig == nil {
+		m.logger.Warn("can't CheckIfNeedsRestart until successful config load")
+		return minimalNeedsRestartCheckInterval, false
+	}
+
+	// Only continue this check if viam-server does not handle restart checking itself
+	// (return early if viamserver _does_ handle restart checking).
+	if !m.viamServer.Property(ctx, viamserver.RestartPropertyDoesNotHandleNeedsRestart) {
+		return minimalNeedsRestartCheckInterval, false
+	}
+
+	m.logger.Debug("Checking cloud for forced restarts")
+	timeoutCtx, cancelFunc := context.WithTimeout(ctx, defaultNetworkTimeout)
+	defer cancelFunc()
+
+	if err := m.dial(timeoutCtx); err != nil {
+		m.logger.Warn(errw.Wrapf(err, "dialing to check if restart needed"))
+		return minimalNeedsRestartCheckInterval, false
+	}
+
+	robotServiceClient := apppb.NewRobotServiceClient(m.conn)
+	req := &apppb.NeedsRestartRequest{Id: m.cloudConfig.ID}
+	res, err := robotServiceClient.NeedsRestart(timeoutCtx, req)
+	if err != nil {
+		m.logger.Warn(errw.Wrapf(err, "checking if restart needed"))
+		return minimalNeedsRestartCheckInterval, false
+	}
+
+	return res.GetRestartCheckInterval().AsDuration(), res.GetMustRestart()
+}
+
 // CloseAll stops all subsystems and closes the cloud connection.
 func (m *Manager) CloseAll() {
 	ctx, cancel := context.WithCancel(context.Background())
 
 	// Use a slow goroutine watcher to log and continue if shutdown is taking too long.
 	slowWatcher, slowWatcherCancel := goutils.SlowGoroutineWatcher(
-		stopAllTimeout, "Agent is taking a while to shut down,", m.logger)
+		stopAllTimeout,
+		fmt.Sprintf("Viam agent subsystems and/or background workers failed to shut down within %v", stopAllTimeout),
+		m.logger,
+	)
 
 	slowTicker := time.NewTicker(10 * time.Second)
 	defer slowTicker.Stop()
@@ -430,7 +475,6 @@ func (m *Manager) CloseAll() {
 			}
 		}
 
-		m.client = nil
 		m.conn = nil
 	})
 
@@ -479,7 +523,8 @@ func (m *Manager) CloseAll() {
 	}
 }
 
-// StartBackgroundChecks kicks off a go routine that loops on a timer to check for updates and health checks.
+// StartBackgroundChecks kicks off go routines that loop on a timerr to check for updates,
+// health checks, and restarts.
 func (m *Manager) StartBackgroundChecks(ctx context.Context) {
 	if ctx.Err() != nil {
 		return
@@ -495,18 +540,18 @@ func (m *Manager) StartBackgroundChecks(ctx context.Context) {
 		})
 		defer m.activeBackgroundWorkers.Done()
 
-		checkInterval := minimalCheckInterval
+		deviceAgentConfigCheckInterval := minimalDeviceAgentConfigCheckInterval
 		m.cfgMu.RLock()
 		wait := m.cfg.AdvancedSettings.WaitForUpdateCheck.Get()
 		m.cfgMu.RUnlock()
 		if wait {
-			checkInterval = m.CheckUpdates(ctx)
+			deviceAgentConfigCheckInterval = m.CheckUpdates(ctx)
 		} else {
 			// premptively start things before we go into the regular update/check/restart
 			m.SubsystemHealthChecks(ctx)
 		}
 
-		timer := time.NewTimer(checkInterval)
+		timer := time.NewTimer(deviceAgentConfigCheckInterval)
 		defer timer.Stop()
 		for {
 			if ctx.Err() != nil {
@@ -516,9 +561,39 @@ func (m *Manager) StartBackgroundChecks(ctx context.Context) {
 			case <-ctx.Done():
 				return
 			case <-timer.C:
-				checkInterval = m.CheckUpdates(ctx)
+				deviceAgentConfigCheckInterval = m.CheckUpdates(ctx)
 				m.SubsystemHealthChecks(ctx)
-				timer.Reset(checkInterval)
+				timer.Reset(deviceAgentConfigCheckInterval)
+			}
+		}
+	}()
+
+	m.activeBackgroundWorkers.Add(1)
+	go func() {
+		defer m.activeBackgroundWorkers.Done()
+
+		timer := time.NewTimer(minimalNeedsRestartCheckInterval)
+		defer timer.Stop()
+		for {
+			if ctx.Err() != nil {
+				return
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case <-timer.C:
+				needsRestartCheckInterval, needsRestart := m.CheckIfNeedsRestart(ctx)
+				if needsRestartCheckInterval < minimalNeedsRestartCheckInterval {
+					needsRestartCheckInterval = minimalNeedsRestartCheckInterval
+				}
+				if needsRestart {
+					// Do not mark m.agentNeedsRestart and instead Exit immediately; we do not want
+					// to wait for viam-server to allow a restart as it may be in a bad state.
+					m.Exit(fmt.Sprintf("A restart of %s was requested from app", SubsystemName))
+				}
+				// As with the device agent config check interval, randomly fuzz the interval by
+				// +/- 5%.
+				timer.Reset(utils.FuzzTime(needsRestartCheckInterval, 0.05))
 			}
 		}
 	}()
@@ -531,11 +606,11 @@ func (m *Manager) dial(ctx context.Context) error {
 		return ctx.Err()
 	}
 	if m.cloudConfig == nil {
-		return errors.New("cannot dial() until successful LoadConfig")
+		return errors.New("cannot dial() until successful config load")
 	}
 	m.connMu.Lock()
 	defer m.connMu.Unlock()
-	if m.client != nil {
+	if m.conn != nil {
 		return nil
 	}
 
@@ -564,7 +639,6 @@ func (m *Manager) dial(ctx context.Context) error {
 		return err
 	}
 	m.conn = conn
-	m.client = pb.NewAgentDeviceServiceClient(m.conn)
 
 	if m.netAppender != nil {
 		m.netAppender.SetConn(conn, true)
@@ -577,27 +651,28 @@ func (m *Manager) dial(ctx context.Context) error {
 // GetConfig retrieves the configuration from the cloud.
 func (m *Manager) GetConfig(ctx context.Context) (time.Duration, error) {
 	if m.cloudConfig == nil {
-		err := errors.New("can't GetConfig until successful LoadConfig")
+		err := errors.New("can't GetConfig until successful config load")
 		m.logger.Warn(err)
-		return minimalCheckInterval, err
+		return minimalDeviceAgentConfigCheckInterval, err
 	}
 	timeoutCtx, cancelFunc := context.WithTimeout(ctx, defaultNetworkTimeout)
 	defer cancelFunc()
 
 	if err := m.dial(timeoutCtx); err != nil {
-		m.logger.Warn(errw.Wrapf(err, "fetching %s config", SubsystemName))
-		return minimalCheckInterval, err
+		m.logger.Warn(errw.Wrapf(err, "dialing to fetch %s config", SubsystemName))
+		return minimalDeviceAgentConfigCheckInterval, err
 	}
 
+	agentDeviceServiceClient := pb.NewAgentDeviceServiceClient(m.conn)
 	req := &pb.DeviceAgentConfigRequest{
 		Id:          m.cloudConfig.ID,
 		HostInfo:    m.getHostInfo(),
 		VersionInfo: m.getVersions(),
 	}
-	resp, err := m.client.DeviceAgentConfig(timeoutCtx, req)
+	resp, err := agentDeviceServiceClient.DeviceAgentConfig(timeoutCtx, req)
 	if err != nil {
 		m.logger.Warn(errw.Wrapf(err, "fetching %s config", SubsystemName))
-		return minimalCheckInterval, err
+		return minimalDeviceAgentConfigCheckInterval, err
 	}
 	fixWindowsPaths(resp)
 
@@ -699,7 +774,7 @@ func (m *Manager) getVersions() *pb.VersionInfo {
 	return vers
 }
 
-func (m *Manager) Exit() {
-	m.logger.Info("A new viam-agent has been installed. Will now exit to be restarted by service manager.")
+func (m *Manager) Exit(reason string) {
+	m.logger.Infow(fmt.Sprintf("%s will now exit to be restarted by service manager", SubsystemName), "reason", reason)
 	m.globalCancel()
 }

subsystems/networking/networking_linux.go

@@ -425,3 +425,8 @@ func (n *Networking) writeWifiPowerSave(ctx context.Context) error {
 
 	return nil
 }
+
+// Property is a noop for the networking subsystem.
+func (n *Networking) Property(_ context.Context, _ string) bool {
+	return false
+}