diff --git a/orbit/changes/20397-fix-orbit-interrupt-services b/orbit/changes/20397-fix-orbit-interrupt-services new file mode 100644 index 0000000000..1844a56568 --- /dev/null +++ b/orbit/changes/20397-fix-orbit-interrupt-services @@ -0,0 +1,2 @@ +* Fixed a startup bug by performing an early restart of orbit if an agent options setting has changed. +* Implemented a small refactor of orbit subsystems. diff --git a/orbit/cmd/orbit/orbit.go b/orbit/cmd/orbit/orbit.go index 6cc9fe9f59..cf24f3c0ef 100644 --- a/orbit/cmd/orbit/orbit.go +++ b/orbit/cmd/orbit/orbit.go @@ -463,10 +463,12 @@ func main() { // Setting up the system service management early on the process lifetime appDoneCh = make(chan struct{}) - // Initializing service runner and system service manager - systemChecker := newSystemChecker() - g.Add(systemChecker.Execute, systemChecker.Interrupt) - go osservice.SetupServiceManagement(constant.SystemServiceName, systemChecker.svcInterruptCh, appDoneCh) + // Initializing windows service runner and system service manager. + if runtime.GOOS == "windows" { + systemChecker := newSystemChecker() + addSubsystem(&g, "system checker", systemChecker) + go osservice.SetupServiceManagement(constant.SystemServiceName, systemChecker.svcInterruptCh, appDoneCh) + } // sofwareupdated is a macOS daemon that automatically updates Apple software. if c.Bool("disable-kickstart-softwareupdated") && runtime.GOOS == "darwin" { @@ -537,7 +539,7 @@ func main() { return nil } - g.Add(updateRunner.Execute, updateRunner.Interrupt) + addSubsystem(&g, "update runner", updateRunner) // if getting any of the targets fails, keep on // retrying, the `updater.Get` method has built-in backoff functionality. @@ -716,8 +718,8 @@ func main() { return fmt.Errorf("create TLS proxy: %w", err) } - g.Add( - func() error { + addSubsystem(&g, "insecure proxy", &wrapSubsystem{ + execute: func() error { log.Info(). Str("addr", fmt.Sprintf("localhost:%d", proxy.Port)). Str("target", c.String("fleet-url")). @@ -725,12 +727,12 @@ func main() { err := proxy.InsecureServeTLS() return err }, - func(error) { + interrupt: func(err error) { if err := proxy.Close(); err != nil { log.Error().Err(err).Msg("close proxy") } }, - ) + }) // Directory to store proxy related assets proxyDirectory := filepath.Join(c.String("root-dir"), "proxy") @@ -854,7 +856,6 @@ func main() { windowsMDMBitlockerCommandFrequency = time.Hour ) - orbitClient.RegisterConfigReceiver(update.ApplyRenewEnrollmentProfileConfigFetcherMiddleware(orbitClient, renewEnrollmentProfileCommandFrequency, fleetURL)) scriptConfigReceiver, scriptsEnabledFn := update.ApplyRunScriptsConfigFetcherMiddleware( c.Bool("enable-scripts"), orbitClient, ) @@ -862,7 +863,9 @@ func main() { switch runtime.GOOS { case "darwin": - // add middleware to handle nudge installation and updates + orbitClient.RegisterConfigReceiver(update.ApplyRenewEnrollmentProfileConfigFetcherMiddleware( + orbitClient, renewEnrollmentProfileCommandFrequency, fleetURL, + )) const nudgeLaunchInterval = 30 * time.Minute orbitClient.RegisterConfigReceiver(update.ApplyNudgeConfigReceiverMiddleware(update.NudgeConfigFetcherOptions{ UpdateRunner: updateRunner, RootDir: c.String("root-dir"), Interval: nudgeLaunchInterval, @@ -874,10 +877,10 @@ func main() { orbitClient.RegisterConfigReceiver(update.ApplyWindowsMDMBitlockerFetcherMiddleware(windowsMDMBitlockerCommandFrequency, orbitClient)) } - flagUpdateReciver := update.NewFlagReceiver(orbitClient.ReceiverUpdateCancelFunc, update.FlagUpdateOptions{ + flagUpdateReceiver := update.NewFlagReceiver(orbitClient.TriggerOrbitRestart, update.FlagUpdateOptions{ RootDir: c.String("root-dir"), }) - orbitClient.RegisterConfigReceiver(flagUpdateReciver) + orbitClient.RegisterConfigReceiver(flagUpdateReceiver) if !c.Bool("disable-updates") { serverOverridesReceiver := newServerOverridesReceiver( @@ -887,7 +890,7 @@ func main() { DesktopPath: desktopPath, }, c.Bool("fleet-desktop"), - orbitClient.ReceiverUpdateCancelFunc, + orbitClient.TriggerOrbitRestart, ) orbitClient.RegisterConfigReceiver(serverOverridesReceiver) @@ -899,7 +902,7 @@ func main() { if !c.Bool("disable-updates") || c.Bool("dev-mode") { extRunner := update.NewExtensionConfigUpdateRunner(update.ExtensionUpdateOptions{ RootDir: c.String("root-dir"), - }, updateRunner, orbitClient.ReceiverUpdateCancelFunc) + }, updateRunner, orbitClient.TriggerOrbitRestart) // call UpdateAction on the updateRunner after we have fetched extensions from Fleet _, err := updateRunner.UpdateAction() @@ -930,11 +933,26 @@ func main() { orbitClient.RegisterConfigReceiver(extRunner) } + // Run a early check of fleetd configuration to check if orbit needs to + // restart before proceeding to start the sub-systems. + // + // E.g. the administrator has updated the following agent options for this device: + // - `update_channels` + // - `extensions` were removed/unset + // - `command_line_flags` (osquery startup flags) if err := orbitClient.RunConfigReceivers(); err != nil { log.Error().Msgf("failed initial config fetch: %s", err) + } else { + if orbitClient.RestartTriggered() { + log.Info().Msg("exiting after early config fetch") + return nil + } } - g.Add(orbitClient.ExecuteConfigReceivers, orbitClient.InterruptConfigReceivers) + addSubsystem(&g, "config receivers", &wrapSubsystem{ + execute: orbitClient.ExecuteConfigReceivers, + interrupt: orbitClient.InterruptConfigReceivers, + }) var trw *token.ReadWriter if c.Bool("fleet-desktop") { @@ -1083,7 +1101,7 @@ func main() { if err != nil { return fmt.Errorf("create osquery runner: %w", err) } - g.Add(r.Execute, r.Interrupt) + addSubsystem(&g, "osqueryd runner", r) // rootDir string, addr string, rootCA string, insecureSkipVerify bool, enrollSecret, uuid string checkerClient, err := service.NewOrbitClient( @@ -1109,7 +1127,7 @@ func main() { capabilitiesChecker := newCapabilitiesChecker(checkerClient) // We populate the known capabilities so that the capability checker does not need to do the initial check on startup. checkerClient.GetServerCapabilities().Copy(orbitClient.GetServerCapabilities()) - g.Add(capabilitiesChecker.actor()) + addSubsystem(&g, "capabilities checker", capabilitiesChecker) var desktopVersion string if c.Bool("fleet-desktop") { @@ -1160,7 +1178,7 @@ func main() { c.String("fleet-desktop-alternative-browser-host"), opt.RootDirectory, ) - g.Add(desktopRunner.actor()) + addSubsystem(&g, "desktop runner", desktopRunner) } // --end-user-email is only supported on Windows and Linux (for macOS it gets the @@ -1205,7 +1223,11 @@ func main() { // Install a signal handler ctx, cancel := context.WithCancel(context.Background()) defer cancel() - g.Add(signalHandler(ctx)) + signalHandlerExecute, signalHandlerInterrupt := signalHandler(ctx) + addSubsystem(&g, "signal handler", &wrapSubsystem{ + execute: signalHandlerExecute, + interrupt: signalHandlerInterrupt, + }) go sigusrListener(c.String("root-dir")) @@ -1312,7 +1334,7 @@ func getFleetdComponentPaths( func registerExtensionRunner(g *run.Group, extSockPath string, opts ...table.Opt) { ext := table.NewRunner(extSockPath, opts...) - g.Add(ext.Execute, ext.Interrupt) + addSubsystem(g, "osqueryd extension runner", ext) } // desktopRunner runs the Fleet Desktop application. @@ -1366,10 +1388,6 @@ func newDesktopRunner( } } -func (d *desktopRunner) actor() (func() error, func(error)) { - return d.execute, d.interrupt -} - // execute makes sure the fleet-desktop application is running. // // We have to support the scenario where the user closes its sessions (log out). @@ -1379,7 +1397,7 @@ func (d *desktopRunner) actor() (func() error, func(error)) { // closes all its sessions). // // NOTE(lucas): This logic could be improved to detect if there's a valid session or not first. -func (d *desktopRunner) execute() error { +func (d *desktopRunner) Execute() error { defer close(d.executeDoneCh) log.Info().Msg("killing any pre-existing fleet-desktop instances") @@ -1489,9 +1507,7 @@ func retry(d time.Duration, waitFirst bool, done chan struct{}, fn func() bool) } } -func (d *desktopRunner) interrupt(err error) { - log.Debug().Err(err).Msg("interrupt desktopRunner") - +func (d *desktopRunner) Interrupt(err error) { close(d.interruptCh) // Signal execute to return. <-d.executeDoneCh // Wait for execute to return. @@ -1604,7 +1620,6 @@ func (s *serviceChecker) Execute() error { } func (s *serviceChecker) Interrupt(err error) { - log.Error().Err(err).Msg("interrupt serviceChecker") close(s.localInterruptCh) // Signal execute to return. } @@ -1626,15 +1641,11 @@ func newCapabilitiesChecker(client *service.OrbitClient) *capabilitiesChecker { } } -func (f *capabilitiesChecker) actor() (func() error, func(error)) { - return f.execute, f.interrupt -} - // execute will poll the server for capabilities and emit a stop signal to restart // Orbit if certain capabilities are enabled. // // You need to add an explicit check for each capability you want to watch for -func (f *capabilitiesChecker) execute() error { +func (f *capabilitiesChecker) Execute() error { defer close(f.executeDoneCh) capabilitiesCheckTicker := time.NewTicker(5 * time.Minute) @@ -1678,8 +1689,7 @@ func (f *capabilitiesChecker) execute() error { } } -func (f *capabilitiesChecker) interrupt(err error) { - log.Debug().Err(err).Msg("interrupt capabilitiesChecker") +func (f *capabilitiesChecker) Interrupt(err error) { close(f.interruptCh) // Signal execute to return. <-f.executeDoneCh // Wait for execute to return. } @@ -1706,11 +1716,11 @@ func writeSecret(enrollSecret string, orbitRoot string) error { // serverOverridesRunner is a oklog.Group runner that polls for configuration overrides from Fleet. type serverOverridesRunner struct { - rootDir string - fallbackCfg fallbackServerOverridesConfig - desktopEnabled bool - cancel chan struct{} - queueOrbitRestart context.CancelFunc + rootDir string + fallbackCfg fallbackServerOverridesConfig + desktopEnabled bool + cancel chan struct{} + triggerOrbitRestart func(reason string) } // newServerOverridesReveiver creates a runner for updating server overrides configuration with values fetched from Fleet. @@ -1718,14 +1728,14 @@ func newServerOverridesReceiver( rootDir string, fallbackCfg fallbackServerOverridesConfig, desktopEnabled bool, - queueOrbitRestart context.CancelFunc, + triggerOrbitRestart func(reason string), ) *serverOverridesRunner { return &serverOverridesRunner{ - rootDir: rootDir, - fallbackCfg: fallbackCfg, - desktopEnabled: desktopEnabled, - cancel: make(chan struct{}), - queueOrbitRestart: queueOrbitRestart, + rootDir: rootDir, + fallbackCfg: fallbackCfg, + desktopEnabled: desktopEnabled, + cancel: make(chan struct{}), + triggerOrbitRestart: triggerOrbitRestart, } } @@ -1745,7 +1755,7 @@ func (r *serverOverridesRunner) Run(orbitCfg *fleet.OrbitConfig) error { if err := r.updateServerOverrides(orbitCfg); err != nil { return err } - r.queueOrbitRestart() + r.triggerOrbitRestart("server overrides updated") return nil } @@ -1853,3 +1863,42 @@ func loadServerOverrides(rootDir string) (*serverOverridesConfig, error) { } return &cfg, nil } + +// subSystem is an interface that implements the methods needed for oklog/run.Group. +type subSystem interface { + // Execute partially implements the interface needed for oklog/run.Group.Add. + Execute() error + // Interrupt partially implements the interface needed for oklog/run.Group.Add. + Interrupt(err error) +} + +// addSubsystem adds a new subsystem to the oklog/run.Group. +func addSubsystem(g *run.Group, name string, s subSystem) { + g.Add( + func() error { + log.Debug().Msgf("start %s", name) + + return s.Execute() + }, func(err error) { + log.Info().Err(err).Msgf("interrupt %s", name) + + s.Interrupt(err) + }, + ) +} + +// wrapSubsystem wraps functions to implement the subSystem interface. +type wrapSubsystem struct { + execute func() error + interrupt func(err error) +} + +// Execute partially implements subSystem. +func (w *wrapSubsystem) Execute() error { + return w.execute() +} + +// Interrupt partially implements subSystem. +func (w *wrapSubsystem) Interrupt(err error) { + w.interrupt(err) +} diff --git a/orbit/pkg/osquery/osquery.go b/orbit/pkg/osquery/osquery.go index 75340d4393..8b003d5bc8 100644 --- a/orbit/pkg/osquery/osquery.go +++ b/orbit/pkg/osquery/osquery.go @@ -25,9 +25,11 @@ type Runner struct { proc *process.Process cmd *exec.Cmd dataPath string - cancelMu sync.Mutex - cancel func() singleQuery bool + + ctxMu sync.Mutex // protects the ctx and cancel + ctx context.Context + cancel func() } type Option func(*Runner) error @@ -164,9 +166,7 @@ func (r *Runner) Execute() error { return fmt.Errorf("start osqueryd shell: %w", err) } } else { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - r.setCancel(cancel) + ctx, _ := r.getContextAndCancel() if err := r.proc.Start(); err != nil { return fmt.Errorf("start osqueryd: %w", err) @@ -182,8 +182,7 @@ func (r *Runner) Execute() error { // Runner interrupts the running osquery process. func (r *Runner) Interrupt(err error) { - log.Error().Err(err).Msg("interrupt osquery") - if cancel := r.getCancel(); cancel != nil { + if _, cancel := r.getContextAndCancel(); cancel != nil { cancel() } } @@ -199,16 +198,15 @@ func (r *Runner) ExtensionSocketPath() string { return filepath.Join(r.dataPath, extensionSocketName) } -func (r *Runner) setCancel(c func()) { - r.cancelMu.Lock() - defer r.cancelMu.Unlock() +func (r *Runner) getContextAndCancel() (context.Context, func()) { + r.ctxMu.Lock() + defer r.ctxMu.Unlock() - r.cancel = c -} - -func (r *Runner) getCancel() func() { - r.cancelMu.Lock() - defer r.cancelMu.Unlock() - - return r.cancel + if r.ctx != nil { + return r.ctx, r.cancel + } + ctx, cancel := context.WithCancel(context.Background()) + r.ctx = ctx + r.cancel = cancel + return r.ctx, r.cancel } diff --git a/orbit/pkg/table/extension.go b/orbit/pkg/table/extension.go index cc7ddaccb3..45f0c71255 100644 --- a/orbit/pkg/table/extension.go +++ b/orbit/pkg/table/extension.go @@ -25,10 +25,12 @@ import ( type Runner struct { socket string tableExtensions []Extension + executeDone chan struct{} - // mu protects access to srv and cancel in Execute and Interrupt. + // mu protects access to srv, ctx and cancel in Execute and Interrupt. mu sync.Mutex srv *osquery.ExtensionManagerServer + ctx context.Context cancel func() } @@ -59,7 +61,10 @@ func WithExtension(t Extension) Opt { // NewRunner creates an extension runner. func NewRunner(socket string, opts ...Opt) *Runner { - r := &Runner{socket: socket} + r := &Runner{ + socket: socket, + executeDone: make(chan struct{}), + } for _, fn := range opts { fn(r) } @@ -68,14 +73,13 @@ func NewRunner(socket string, opts ...Opt) *Runner { // Execute creates an osquery extension manager server and registers osquery plugins. func (r *Runner) Execute() error { - log.Debug().Msg("start osquery extension") + defer close(r.executeDone) if err := waitExtensionSocket(r.socket, 1*time.Minute); err != nil { return err } - ctx, cancel := context.WithCancel(context.Background()) - r.setCancel(cancel) + ctx, _ := r.getContextAndCancel() ticker := time.NewTicker(200 * time.Millisecond) for { @@ -159,10 +163,10 @@ func OrbitDefaultTables() []osquery.OsqueryPlugin { // Interrupt shuts down the osquery manager server. func (r *Runner) Interrupt(err error) { - log.Error().Err(err).Msg("interrupt osquery extension") - if cancel := r.getCancel(); cancel != nil { + if _, cancel := r.getContextAndCancel(); cancel != nil { cancel() } + <-r.executeDone if srv := r.getSrv(); srv != nil { if err := srv.Shutdown(context.Background()); err != nil { log.Debug().Err(err).Msg("shutdown extension") @@ -250,13 +254,6 @@ func (r *Runner) setSrv(s *osquery.ExtensionManagerServer) { r.srv = s } -func (r *Runner) setCancel(c func()) { - r.mu.Lock() - defer r.mu.Unlock() - - r.cancel = c -} - func (r *Runner) getSrv() *osquery.ExtensionManagerServer { r.mu.Lock() defer r.mu.Unlock() @@ -264,9 +261,15 @@ func (r *Runner) getSrv() *osquery.ExtensionManagerServer { return r.srv } -func (r *Runner) getCancel() func() { +func (r *Runner) getContextAndCancel() (context.Context, func()) { r.mu.Lock() defer r.mu.Unlock() - return r.cancel + if r.ctx != nil { + return r.ctx, r.cancel + } + ctx, cancel := context.WithCancel(context.Background()) + r.ctx = ctx + r.cancel = cancel + return r.ctx, r.cancel } diff --git a/orbit/pkg/update/flag_runner.go b/orbit/pkg/update/flag_runner.go index 19acdccfe8..f8b3db8bf1 100644 --- a/orbit/pkg/update/flag_runner.go +++ b/orbit/pkg/update/flag_runner.go @@ -1,7 +1,6 @@ package update import ( - "context" "encoding/json" "errors" "fmt" @@ -23,8 +22,8 @@ import ( // It uses an OrbitConfigFetcher (which may be the OrbitClient with additional middleware), along // with FlagUpdateOptions to connect to Fleet type FlagRunner struct { - queueOrbitRestart context.CancelFunc - opt FlagUpdateOptions + triggerOrbitRestart func(reason string) + opt FlagUpdateOptions } // FlagUpdateOptions is options provided for the flag update runner @@ -35,10 +34,10 @@ type FlagUpdateOptions struct { // NewFlagRunner creates a new runner with provided options // The runner must be started with Execute -func NewFlagReceiver(queueOrbitRestart context.CancelFunc, opt FlagUpdateOptions) *FlagRunner { +func NewFlagReceiver(triggerOrbitRestart func(reason string), opt FlagUpdateOptions) *FlagRunner { return &FlagRunner{ - queueOrbitRestart: queueOrbitRestart, - opt: opt, + triggerOrbitRestart: triggerOrbitRestart, + opt: opt, } } @@ -79,7 +78,7 @@ func (r *FlagRunner) Run(config *fleet.OrbitConfig) error { return fmt.Errorf("error writing flags to disk: %w", err) } - r.queueOrbitRestart() + r.triggerOrbitRestart("osquery flags updated") return nil } @@ -89,9 +88,9 @@ func (r *FlagRunner) Run(config *fleet.OrbitConfig) error { // It uses an an OrbitConfigFetcher (which may be the OrbitClient with additional middleware), along // with ExtensionUpdateOptions and updateRunner to connect to Fleet. type ExtensionRunner struct { - opt ExtensionUpdateOptions - updateRunner *Runner - queueOrbitRestart context.CancelFunc + opt ExtensionUpdateOptions + updateRunner *Runner + triggerOrbitRestart func(reason string) } // ExtensionUpdateOptions is options provided for the extensions fetch/update runner @@ -102,19 +101,18 @@ type ExtensionUpdateOptions struct { // NewExtensionConfigUpdateRunner creates a new runner with provided options // The runner must be started with Execute -func NewExtensionConfigUpdateRunner(opt ExtensionUpdateOptions, updateRunner *Runner, queueOrbitRestart context.CancelFunc) *ExtensionRunner { +func NewExtensionConfigUpdateRunner(opt ExtensionUpdateOptions, updateRunner *Runner, triggerOrbitRestart func(reason string)) *ExtensionRunner { return &ExtensionRunner{ - opt: opt, - updateRunner: updateRunner, - queueOrbitRestart: queueOrbitRestart, + opt: opt, + updateRunner: updateRunner, + triggerOrbitRestart: triggerOrbitRestart, } } // DoExtensionConfigUpdate calls the /config API endpoint to grab extensions from Fleet // It parses the extensions, computes the local hash, and writes the binary path to extension.load file // -// It returns a (bool, error), where bool indicates whether orbit should restart -// It only returns (true, nil) when extensions were previously configured and now are cleared +// It will only trigger a orbit restart when extensions were previously configured and now are cleared. func (r *ExtensionRunner) Run(config *fleet.OrbitConfig) error { extensionAutoLoadFile := filepath.Join(r.opt.RootDir, "extensions.load") if len(config.Extensions) == 0 { @@ -126,7 +124,6 @@ func (r *ExtensionRunner) Run(config *fleet.OrbitConfig) error { // Handle case 1, where our autoload file does not exist, so there is nothing to update and no error case errors.Is(err, os.ErrNotExist): log.Debug().Msg(extensionAutoLoadFile + " not found, nothing to update") - // we do not want orbit to restart return nil case err == nil: // handle case 2: create/truncate the extensions.load file and let the runner interrupt, so that @@ -135,19 +132,14 @@ func (r *ExtensionRunner) Run(config *fleet.OrbitConfig) error { if stat.Size() > 0 { err := os.WriteFile(extensionAutoLoadFile, []byte(""), constant.DefaultFileMode) if err != nil { - // we do not want orbit to restart return fmt.Errorf("extensionsUpdate: error creating file %s, %w", extensionAutoLoadFile, err) } - // we want to return true here, and restart with the empty extensions.load file - // so that we "unload" the previously loaded - // extensions - r.queueOrbitRestart() + // Restart with the empty extensions.load file so that we "unload" the previously loaded extensions. + r.triggerOrbitRestart("unloading extensions") return nil } - // we do not want orbit to restart return nil default: - // we do not want orbit to restart, just log the error return fmt.Errorf("stat file: %s", extensionAutoLoadFile) } } @@ -157,7 +149,6 @@ func (r *ExtensionRunner) Run(config *fleet.OrbitConfig) error { var extensions fleet.Extensions err := json.Unmarshal(config.Extensions, &extensions) if err != nil { - // we do not want orbit to restart return fmt.Errorf("error unmarshing json extensions config from fleet: %w", err) } @@ -205,7 +196,6 @@ func (r *ExtensionRunner) Run(config *fleet.OrbitConfig) error { } if err := r.updateRunner.StoreLocalHash(targetName); err != nil { - // we do not want orbit to restart return fmt.Errorf("unable to lookup metadata for target: %s, %w", targetName, err) } @@ -215,8 +205,6 @@ func (r *ExtensionRunner) Run(config *fleet.OrbitConfig) error { return fmt.Errorf("error writing extensions autoload file: %w", err) } - // we do not want orbit to restart - // runner.UpdateAction() will fetch the new targets and restart for us if needed return nil } diff --git a/orbit/pkg/update/flag_runner_test.go b/orbit/pkg/update/flag_runner_test.go index d5e366ab6b..90ff2b5558 100644 --- a/orbit/pkg/update/flag_runner_test.go +++ b/orbit/pkg/update/flag_runner_test.go @@ -90,7 +90,7 @@ func TestDoFlagsUpdateWithEmptyFlags(t *testing.T) { } var restartQueued bool - queueOrbitRestart := func() { restartQueued = true } + queueOrbitRestart := func(string) { restartQueued = true } fr := NewFlagReceiver(queueOrbitRestart, FlagUpdateOptions{ RootDir: rootDir, diff --git a/orbit/pkg/update/runner.go b/orbit/pkg/update/runner.go index d1551ac419..361d69278b 100644 --- a/orbit/pkg/update/runner.go +++ b/orbit/pkg/update/runner.go @@ -165,8 +165,6 @@ func randomizeDuration(max time.Duration) (time.Duration, error) { // Execute begins a loop checking for updates. func (r *Runner) Execute() error { - log.Debug().Msg("start updater") - // Randomize the initial interval so that all agents don't synchronize their updates initialInterval := r.opt.CheckInterval // Developers use a shorter update interval (10s), so they need a faster first update check @@ -322,7 +320,6 @@ func (r *Runner) updateTarget(target string) error { func (r *Runner) Interrupt(err error) { r.cancel <- struct{}{} - log.Error().Err(err).Msg("interrupt updater") } // compareVersion compares the old and new versions of a binary and prints the appropriate message. diff --git a/server/service/orbit_client.go b/server/service/orbit_client.go index 5435f41c92..fb4fe0aa2f 100644 --- a/server/service/orbit_client.go +++ b/server/service/orbit_client.go @@ -53,11 +53,10 @@ type OrbitClient struct { ConfigReceivers []fleet.OrbitConfigReceiver // How frequently a new config will be fetched ReceiverUpdateInterval time.Duration - // Cancelable context used by ExecuteConfigReceivers to cancel the - // update loop - ReceiverUpdateContext context.Context - // ReceiverUpdateCancelFunc will be called when ReceiverUpdateContext is cancelled - ReceiverUpdateCancelFunc context.CancelFunc + // receiverUpdateContext used by ExecuteConfigReceivers to cancel the update loop. + receiverUpdateContext context.Context + // receiverUpdateCancelFunc is used to cancel receiverUpdateContext. + receiverUpdateCancelFunc context.CancelFunc } // time-to-live for config cache @@ -165,11 +164,27 @@ func NewOrbitClient( onGetConfigErrFns: onGetConfigErrFns, lastIdleConnectionsCleanup: time.Now(), ReceiverUpdateInterval: defaultOrbitConfigReceiverInterval, - ReceiverUpdateContext: ctx, - ReceiverUpdateCancelFunc: cancelFunc, + receiverUpdateContext: ctx, + receiverUpdateCancelFunc: cancelFunc, }, nil } +// TriggerOrbitRestart triggers a orbit process restart. +func (oc *OrbitClient) TriggerOrbitRestart(reason string) { + log.Info().Msgf("orbit restart triggered: %s", reason) + oc.receiverUpdateCancelFunc() +} + +// RestartTriggered returns true if any of the config receivers triggered an orbit restart. +func (oc *OrbitClient) RestartTriggered() bool { + select { + case <-oc.receiverUpdateContext.Done(): + return true + default: + return false + } +} + // closeIdleConnections attempts to close idle connections from the pool // every 55 minutes. // @@ -252,7 +267,7 @@ func (oc *OrbitClient) ExecuteConfigReceivers() error { for { select { - case <-oc.ReceiverUpdateContext.Done(): + case <-oc.receiverUpdateContext.Done(): return nil case <-ticker.C: if err := oc.RunConfigReceivers(); err != nil { @@ -263,8 +278,7 @@ func (oc *OrbitClient) ExecuteConfigReceivers() error { } func (oc *OrbitClient) InterruptConfigReceivers(err error) { - log.Error().Err(err).Msg("interrupt config receivers") - oc.ReceiverUpdateCancelFunc() + oc.receiverUpdateCancelFunc() } // GetConfig returns the Orbit config fetched from Fleet server for this instance of OrbitClient. diff --git a/server/service/orbit_client_test.go b/server/service/orbit_client_test.go index 8f76595a60..789bdaf670 100644 --- a/server/service/orbit_client_test.go +++ b/server/service/orbit_client_test.go @@ -39,8 +39,8 @@ func TestGetConfig(t *testing.T) { func clientWithConfig(cfg *fleet.OrbitConfig) *OrbitClient { ctx, cancel := context.WithCancel(context.Background()) oc := &OrbitClient{ - ReceiverUpdateContext: ctx, - ReceiverUpdateCancelFunc: cancel, + receiverUpdateContext: ctx, + receiverUpdateCancelFunc: cancel, } oc.configCache.config = cfg oc.configCache.lastUpdated = time.Now().Add(1 * time.Hour) @@ -127,7 +127,7 @@ func TestExecuteConfigReceiversCancel(t *testing.T) { cfunc := fleet.OrbitConfigReceiverFunc(func(cfg *fleet.OrbitConfig) error { calls1++ if calls1 == requiredCalls { - client.ReceiverUpdateCancelFunc() + client.receiverUpdateCancelFunc() } return nil }) @@ -149,7 +149,7 @@ func TestExecuteConfigReceiversCancel(t *testing.T) { func TestExecuteConfigReceiversInterrupt(t *testing.T) { client := clientWithConfig(&fleet.OrbitConfig{}) - defer client.ReceiverUpdateCancelFunc() + defer client.receiverUpdateCancelFunc() client.ReceiverUpdateInterval = 100 * time.Millisecond @@ -168,7 +168,7 @@ func TestExecuteConfigReceiversInterrupt(t *testing.T) { go func() { time.Sleep(500 * time.Millisecond) - client.ReceiverUpdateCancelFunc() + client.receiverUpdateCancelFunc() }() select { diff --git a/tools/tuf/test/README.md b/tools/tuf/test/README.md index 6bf057e602..55249fcbd4 100644 --- a/tools/tuf/test/README.md +++ b/tools/tuf/test/README.md @@ -31,7 +31,9 @@ MSI_FLEET_URL=https://host.docker.internal:8080 \ MSI_TUF_URL=http://host.docker.internal:8081 \ GENERATE_PKG=1 \ GENERATE_DEB=1 \ +GENERATE_DEB_ARM64=1 \ GENERATE_RPM=1 \ +GENERATE_RPM_ARM64=1 \ GENERATE_MSI=1 \ ENROLL_SECRET=6/EzU/+jPkxfTamWnRv1+IJsO4T9Etju \ FLEET_DESKTOP=1 \ diff --git a/tools/tuf/test/create_repository.sh b/tools/tuf/test/create_repository.sh index 82c751e3d9..86fa0f5e17 100755 --- a/tools/tuf/test/create_repository.sh +++ b/tools/tuf/test/create_repository.sh @@ -91,7 +91,6 @@ for system in $SYSTEMS; do # Apple keychain, some tables, etc), if this is the case, compile an # universal binary. # - # NOTE(lucas): Cross-compiling orbit for arm64 from Intel macOS currently fails (CGO error). if [ $system == "macos" ] && [ "$(uname -s)" = "Darwin" ] && [ "$(uname -m)" = "arm64" ]; then CGO_ENABLED=1 \ CODESIGN_IDENTITY=$CODESIGN_IDENTITY \ @@ -99,7 +98,23 @@ for system in $SYSTEMS; do ORBIT_BINARY_PATH=$orbit_target \ go run ./orbit/tools/build/build.go else - CGO_ENABLED=0 GOOS=$goose_value GOARCH=$goarch_value go build -ldflags="-X github.com/fleetdm/fleet/v4/orbit/pkg/build.Version=42" -o $orbit_target ./orbit/cmd/orbit + race_value=false + # Enable race on macOS Intel at least. + # + # For cross-compiling to Windows with `-race` we need CGO_ENABLED=1 but we cannot + # do cross-compilation with CGO_ENABLED=1. + if [ "$goose_value" = "darwin" ] && [ "$(uname -s)" = "Darwin" ] && [ "$(uname -m)" = "x86_64" ]; then + race_value=true + fi + # NOTE(lucas): Cross-compiling orbit for arm64 from Intel macOS currently fails (CGO error), + # thus on Intel we do not build an universal binary. + CGO_ENABLED=0 \ + GOOS=$goose_value \ + GOARCH=$goarch_value \ + go build \ + -race=$race_value \ + -ldflags="-X github.com/fleetdm/fleet/v4/orbit/pkg/build.Version=42" \ + -o $orbit_target ./orbit/cmd/orbit fi ./build/fleetctl updates add \ diff --git a/tools/tuf/test/gen_pkgs.sh b/tools/tuf/test/gen_pkgs.sh index f021835aea..f1fcddeb58 100755 --- a/tools/tuf/test/gen_pkgs.sh +++ b/tools/tuf/test/gen_pkgs.sh @@ -82,7 +82,9 @@ if [ -n "$GENERATE_DEB" ]; then ${FLEET_DESKTOP_ALTERNATIVE_BROWSER_HOST:+--fleet-desktop-alternative-browser-host=$FLEET_DESKTOP_ALTERNATIVE_BROWSER_HOST} \ ${ENABLE_SCRIPTS:+--enable-scripts} \ --update-url=$DEB_TUF_URL +fi +if [ -n "$GENERATE_DEB_ARM64" ]; then echo "Generating deb (arm64)..." ./build/fleetctl package \ --type=deb \ @@ -128,7 +130,9 @@ if [ -n "$GENERATE_RPM" ]; then ${FLEET_DESKTOP_ALTERNATIVE_BROWSER_HOST:+--fleet-desktop-alternative-browser-host=$FLEET_DESKTOP_ALTERNATIVE_BROWSER_HOST} \ ${ENABLE_SCRIPTS:+--enable-scripts} \ --update-url=$RPM_TUF_URL +fi +if [ -n "$GENERATE_RPM_ARM64" ]; then echo "Generating rpm (arm64)..." ./build/fleetctl package \ --type=rpm \ diff --git a/tools/tuf/test/main.sh b/tools/tuf/test/main.sh index 2f5d8f20a5..6d7d4f5097 100755 --- a/tools/tuf/test/main.sh +++ b/tools/tuf/test/main.sh @@ -29,6 +29,6 @@ if [ -z "$SKIP_SERVER" ]; then ./tools/tuf/test/run_server.sh fi -if [ -n "$GENERATE_PKG" ] || [ -n "$GENERATE_DEB" ] || [ -n "$GENERATE_RPM" ] || [ -n "$GENERATE_MSI" ]; then +if [ -n "$GENERATE_PKG" ] || [ -n "$GENERATE_DEB" ] || [ -n "$GENERATE_RPM" ] || [ -n "$GENERATE_MSI" ] || [ -n "$GENERATE_DEB_ARM64" ] || [ -n "$GENERATE_RPM_ARM64" ]; then bash ./tools/tuf/test/gen_pkgs.sh fi