Register all simulators before starting controllers

rancher · Oct 3, 2020 · 99dd837 · 99dd837
1 parent ba03066
commit 99dd837
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 3 deletions.
diff --git a/modules/agent/pkg/agent/app.go b/modules/agent/pkg/agent/app.go
@@ -24,6 +24,7 @@ type Options struct {
 	ClusterID        string
 	NoLeaderElect    bool
 	CheckinInterval  time.Duration
+	StartAfter       <-chan struct{}
 }
 
 func Register(ctx context.Context, kubeConfig, namespace, clusterID string) error {
@@ -84,7 +85,8 @@ func Start(ctx context.Context, kubeConfig, namespace string, opts *Options) err
 		clientConfig,
 		fleetMapper,
 		mapper,
-		discovery)
+		discovery,
+		opts.StartAfter)
 }
 
 var (

diff --git a/modules/agent/pkg/controllers/controllers.go b/modules/agent/pkg/controllers/controllers.go
@@ -80,7 +80,8 @@ func Register(ctx context.Context, leaderElect bool,
 	checkinInterval time.Duration,
 	fleetConfig *rest.Config, clientConfig clientcmd.ClientConfig,
 	fleetMapper, mapper meta.RESTMapper,
-	discovery discovery.CachedDiscoveryInterface) error {
+	discovery discovery.CachedDiscoveryInterface,
+	startChan <-chan struct{}) error {
 	appCtx, err := newContext(fleetNamespace, agentNamespace, clusterNamespace, clusterName,
 		fleetConfig, clientConfig, fleetMapper, mapper, discovery)
 	if err != nil {
@@ -124,6 +125,11 @@ func Register(ctx context.Context, leaderElect bool,
 				logrus.Fatal(err)
 			}
 		})
+	} else if startChan != nil {
+		go func() {
+			<-startChan
+			logrus.Fatalf("failed to start: %v", appCtx.start(ctx))
+		}()
 	} else {
 		return appCtx.start(ctx)
 	}

diff --git a/modules/agent/pkg/simulator/simulator.go b/modules/agent/pkg/simulator/simulator.go
@@ -21,16 +21,21 @@ import (
 )
 
 var (
-	sem = semaphore.NewWeighted(50)
+	semMax = int64(50)
+	sem    = semaphore.NewWeighted(semMax)
 )
 
 func Simulate(ctx context.Context, count int, kubeConfig, namespace, defaultNamespace string, opts agent.Options) error {
 	logrus.Infof("Starting %d simulators", count)
 
+	startAfter := make(chan struct{})
+	opts.StartAfter = startAfter
+
 	eg, ctx := errgroup.WithContext(ctx)
 	for i := 0; i < count; i++ {
 		i := i
 		if err := sem.Acquire(ctx, 1); err != nil {
+			close(startAfter)
 			return err
 		}
 		logrus.Infof("STARING %s%05d", namespace, i)
@@ -44,6 +49,12 @@ func Simulate(ctx context.Context, count int, kubeConfig, namespace, defaultName
 		})
 	}
 
+	if err := sem.Acquire(ctx, semMax); err != nil {
+		close(startAfter)
+		return err
+	}
+	close(startAfter)
+
 	eg.Go(func() error {
 		// wait forever unless one of the simulators dies
 		<-ctx.Done()