daemon: Avoid blocking during hive start

jrajahalme · jrajahalme · commit 9463a8684754 · 2024-06-21T21:16:39.000Z
Execute startDaemon() in a goroutine, which resolves daemon promise when
done. Change dependent start hooks to await for daemon promise in
goroutines so that the execution of start hooks is not stalled. This way
the hive may run concurrently with daemon start.

Move some early validation from startDaemon() to the start hook so that
we can fail out while we can simply return the error from the start
hook.

Move saving of daemon config to file also away from startDaemon, so that
we can resolve the DaemonConfig promise from the start hook itself. The
daemon promise will be resolved from the goroutine that runs
startDaemon(), or rejected if there was any error.

Signed-off-by: Jarno Rajahalme &lt;jarno@isovalent.com&gt;
diff --git a/daemon/cmd/daemon_main.go b/daemon/cmd/daemon_main.go
@@ -1706,7 +1706,15 @@ func newDaemonPromise(params daemonParams) (promise.Promise[*Daemon], promise.Pr
 	var wg sync.WaitGroup
 
 	params.Lifecycle.Append(cell.Hook{
-		OnStart: func(cell.HookContext) error {
+		OnStart: func(cell.HookContext) (err error) {
+			defer func() {
+				// Reject promises on error
+				if err != nil {
+					cfgResolver.Reject(err)
+					daemonResolver.Reject(err)
+				}
+			}()
+
 			d, restoredEndpoints, err := newDaemon(daemonCtx, cleaner, &params)
 			if err != nil {
 				cancelDaemonCtx()
@@ -1716,16 +1724,45 @@ func newDaemonPromise(params daemonParams) (promise.Promise[*Daemon], promise.Pr
 			daemon = d
 
 			if !option.Config.DryMode {
-				if err := startDaemon(daemon, restoredEndpoints, cleaner, params); err != nil {
-					daemonResolver.Reject(err)
-					cancelDaemonCtx()
-					cleaner.Clean()
-					cfgResolver.Reject(err)
-					return err
+				log.Info("Initializing daemon")
+
+				// This validation needs to be done outside of the agent until
+				// datapath.NodeAddressing is used consistently across the code base.
+				log.Info("Validating configured node address ranges")
+				if err := node.ValidatePostInit(); err != nil {
+					return fmt.Errorf("postinit failed: %w", err)
+				}
+
+				// Store config in file before resolving the DaemonConfig promise.
+				err = option.Config.StoreInFile(option.Config.StateDir)
+				if err != nil {
+					log.WithError(err).Error("Unable to store Cilium's configuration")
+				}
+
+				err = option.StoreViperInFile(option.Config.StateDir)
+				if err != nil {
+					log.WithError(err).Error("Unable to store Viper's configuration")
 				}
 			}
-			daemonResolver.Resolve(daemon)
+
+			// 'option.Config' is assumed to be stable at this point, execpt for
+			// 'option.Config.Opts' that are explicitly deemed to be runtime-changeable
 			cfgResolver.Resolve(option.Config)
+
+			if option.Config.DryMode {
+				daemonResolver.Resolve(daemon)
+			} else {
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					if err := startDaemon(daemon, restoredEndpoints, cleaner, params); err != nil {
+						log.WithError(err).Error("Daemon start failed")
+						daemonResolver.Reject(err)
+					} else {
+						daemonResolver.Resolve(daemon)
+					}
+				}()
+			}
 			return nil
 		},
 		OnStop: func(cell.HookContext) error {
@@ -1739,16 +1776,9 @@ func newDaemonPromise(params daemonParams) (promise.Promise[*Daemon], promise.Pr
 }
 
 // startDaemon starts the old unmodular part of the cilium-agent.
+// option.Config has already been exposed via *option.DaemonConfig promise,
+// so it may not be modified here
 func startDaemon(d *Daemon, restoredEndpoints *endpointRestoreState, cleaner *daemonCleanup, params daemonParams) error {
-	log.Info("Initializing daemon")
-
-	// This validation needs to be done outside of the agent until
-	// datapath.NodeAddressing is used consistently across the code base.
-	log.Info("Validating configured node address ranges")
-	if err := node.ValidatePostInit(); err != nil {
-		return fmt.Errorf("postinit failed: %w", err)
-	}
-
 	bootstrapStats.k8sInit.Start()
 	if params.Clientset.IsEnabled() {
 		// Wait only for certain caches, but not all!
@@ -1911,28 +1941,28 @@ func startDaemon(d *Daemon, restoredEndpoints *endpointRestoreState, cleaner *da
 	bootstrapStats.updateMetrics()
 	go d.launchHubble()
 
-	err = option.Config.StoreInFile(option.Config.StateDir)
-	if err != nil {
-		log.WithError(err).Error("Unable to store Cilium's configuration")
-	}
-
-	err = option.StoreViperInFile(option.Config.StateDir)
-	if err != nil {
-		log.WithError(err).Error("Unable to store Viper's configuration")
-	}
-
 	return nil
 }
 
 func registerEndpointStateResolver(lc cell.Lifecycle, daemonPromise promise.Promise[*Daemon], resolver promise.Resolver[endpointstate.Restorer]) {
+	var wg sync.WaitGroup
+
 	lc.Append(cell.Hook{
 		OnStart: func(ctx cell.HookContext) error {
-			daemon, err := daemonPromise.Await(context.Background())
-			if err != nil {
-				resolver.Reject(err)
-				return err
-			}
-			resolver.Resolve(daemon)
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				daemon, err := daemonPromise.Await(context.Background())
+				if err != nil {
+					resolver.Reject(err)
+				} else {
+					resolver.Resolve(daemon)
+				}
+			}()
+			return nil
+		},
+		OnStop: func(ctx cell.HookContext) error {
+			wg.Wait()
 			return nil
 		},
 	})
diff --git a/daemon/cmd/deletion_queue.go b/daemon/cmd/deletion_queue.go
@@ -7,6 +7,7 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"sync"
 
 	"github.com/cilium/hive/cell"
 
@@ -26,26 +27,39 @@ var deletionQueueCell = cell.Group(
 type deletionQueue struct {
 	lf            *lockfile.Lockfile
 	daemonPromise promise.Promise[*Daemon]
+	wg            sync.WaitGroup
 }
 
-func (dq *deletionQueue) Start(ctx cell.HookContext) error {
-	d, err := dq.daemonPromise.Await(ctx)
-	if err != nil {
-		return err
-	}
+func (dq *deletionQueue) Start(cell.HookContext) error {
+	dq.wg.Add(1)
+	go func() {
+		defer dq.wg.Done()
 
-	if err := dq.lock(ctx); err != nil {
-		return err
-	}
+		// hook context cancels when the start hooks have run, use context.Background()
+		// as we may be running after that.
+		d, err := dq.daemonPromise.Await(context.Background())
+		if err != nil {
+			log.WithError(err).Error("deletionQueue: Daemon promise failed")
+			return
+		}
 
-	bootstrapStats.deleteQueue.Start()
-	err = dq.processQueuedDeletes(d, ctx)
-	bootstrapStats.deleteQueue.EndError(err)
-	return err
+		if err := dq.lock(d.ctx); err != nil {
+			log.WithError(err).Error("deletionQueue: lock failed")
+			return
+		}
 
+		bootstrapStats.deleteQueue.Start()
+		err = dq.processQueuedDeletes(d, d.ctx)
+		bootstrapStats.deleteQueue.EndError(err)
+		if err != nil {
+			log.WithError(err).Error("deletionQueue: processQueuedDeletes failed")
+		}
+	}()
+	return nil
 }
 
-func (dq *deletionQueue) Stop(ctx cell.HookContext) error {
+func (dq *deletionQueue) Stop(cell.HookContext) error {
+	dq.wg.Wait()
 	return nil
 }
 
diff --git a/pkg/maps/nat/cell.go b/pkg/maps/nat/cell.go
@@ -49,11 +49,15 @@ var Cell = cell.Module(
 				ipv4Nat, ipv6Nat = GlobalMaps(cfg.EnableIPv4,
 					cfg.EnableIPv6, true)
 
-				// Maps are still created in daemon.initMaps(...) under the same circumstances
+				// Maps are still created before DaemonConfig promise is resolved in
+				// daemon.initMaps(...) under the same circumstances
 				// so we just open them here so they can be provided to hive.
 				//
 				// TODO: Refactor ctmap gc Enable() such that it can use the map descriptors from
 				// here so we can move all nat map creation logic into here.
+				// NOTE: This code runs concurrently with startDaemon(), so if any dependency to
+				// daemon having finished endpoint restore, for example, is added, we should
+				// await for an appropriate promise.
 				if cfg.EnableIPv4 {
 					if err := ipv4Nat.Open(); err != nil {
 						return fmt.Errorf("open IPv4 nat map: %w", err)
