Improve embedded executor agent startup time

Start CRIs and Kubelet in goroutine, and wait for CRI startup before running kubelet

Signed-off-by: Brad Davidson <[email protected]>

Author: brandond

pkg/agent/containerd/watcher.go

@@ -259,8 +259,7 @@ func importAndWatchImages(ctx context.Context, cfg *config.Node) error {
 
 	// wait for the workqueue to empty before returning
 	for w.workqueue.Len() > 0 {
-		logrus.Debugf("Waiting for initial import of images from %s", cfg.Images)
-		time.Sleep(time.Second * 2)
+		time.Sleep(500 * time.Millisecond)
 	}
 
 	// prune unseen entries from last run once all existing files have been processed

pkg/daemons/executor/embed.go

@@ -80,6 +80,7 @@ func (e *Embedded) Kubelet(ctx context.Context, args []string) error {
 	command.SetArgs(args)
 
 	go func() {
+		<-e.CRIReadyChan()
 		<-e.APIServerReadyChan()
 		defer func() {
 			if err := recover(); err != nil {
@@ -227,7 +228,6 @@ func (*Embedded) CloudControllerManager(ctx context.Context, ccmRBACReady <-chan
 		}
 		signals.RequestShutdown(nil)
 	}()
-
 	return nil
 }
 
@@ -236,19 +236,40 @@ func (e *Embedded) CurrentETCDOptions() (InitialOptions, error) {
 }
 
 func (e *Embedded) Containerd(ctx context.Context, cfg *daemonconfig.Node) error {
-	return closeIfNilErr(containerd.Run(ctx, cfg), e.criReady)
+	go func() {
+		if err := containerd.Run(ctx, cfg); err != nil {
+			signals.RequestShutdown(pkgerrors.WithMessage(err, "failed to start containerd"))
+		} else {
+			close(e.criReady)
+		}
+	}()
+	return nil
 }
 
 func (e *Embedded) Docker(ctx context.Context, cfg *daemonconfig.Node) error {
-	return closeIfNilErr(cridockerd.Run(ctx, cfg), e.criReady)
+	go func() {
+		if err := cridockerd.Run(ctx, cfg); err != nil {
+			signals.RequestShutdown(pkgerrors.WithMessage(err, "failed to start containerd"))
+		} else {
+			close(e.criReady)
+		}
+	}()
+	return nil
 }
 
 func (e *Embedded) CRI(ctx context.Context, cfg *daemonconfig.Node) error {
 	// agentless sets cri socket path to /dev/null to indicate no CRI is needed
 	if cfg.ContainerRuntimeEndpoint != "/dev/null" {
-		return closeIfNilErr(cri.WaitForService(ctx, cfg.ContainerRuntimeEndpoint, "CRI"), e.criReady)
+		go func() {
+			if err := cri.WaitForService(ctx, cfg.ContainerRuntimeEndpoint, "CRI"); err != nil {
+				signals.RequestShutdown(pkgerrors.WithMessage(err, "failed to wait for CRI startup"))
+			} else {
+				close(e.criReady)
+			}
+		}()
 	}
-	return closeIfNilErr(nil, e.criReady)
+	close(e.criReady)
+	return nil
 }
 
 func (e *Embedded) APIServerReadyChan() <-chan struct{} {
@@ -271,10 +292,3 @@ func (e *Embedded) CRIReadyChan() <-chan struct{} {
 	}
 	return e.criReady
 }
-
-func closeIfNilErr(err error, ch chan struct{}) error {
-	if err == nil {
-		close(ch)
-	}
-	return err
-}