Wait until stores are filled before processing service account token events

pkg/controller/framework/controller.go

@@ -17,6 +17,7 @@ limitations under the License.
 package framework
 
 import (
+	"sync"
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/client/cache"
@@ -61,7 +62,9 @@ type ProcessFunc func(obj interface{}) error
 
 // Controller is a generic controller framework.
 type Controller struct {
-	config Config
+	config         Config
+	reflector      *cache.Reflector
+	reflectorMutex sync.RWMutex
 }
 
 // New makes a new Controller from the given Config.
@@ -77,16 +80,32 @@ func New(c *Config) *Controller {
 // Run blocks; call via go.
 func (c *Controller) Run(stopCh <-chan struct{}) {
 	defer util.HandleCrash()
-	cache.NewReflector(
+	r := cache.NewReflector(
 		c.config.ListerWatcher,
 		c.config.ObjectType,
 		c.config.Queue,
 		c.config.FullResyncPeriod,
-	).RunUntil(stopCh)
+	)
+
+	c.reflectorMutex.Lock()
+	c.reflector = r
+	c.reflectorMutex.Unlock()
+
+	r.RunUntil(stopCh)
 
 	util.Until(c.processLoop, time.Second, stopCh)
 }
 
+// Returns true once this controller has completed an initial resource listing
+func (c *Controller) HasSynced() bool {
+	c.reflectorMutex.RLock()
+	defer c.reflectorMutex.RUnlock()
+	if c.reflector == nil {
+		return false
+	}
+	return c.reflector.LastSyncResourceVersion() != ""
+}
+
 // processLoop drains the work queue.
 // TODO: Consider doing the processing in parallel. This will require a little thought
 // to make sure that we don't end up processing the same object multiple times

pkg/controller/framework/controller_test.go

@@ -214,10 +214,20 @@ func TestHammerController(t *testing.T) {
 		},
 	)
 
+	if controller.HasSynced() {
+		t.Errorf("Expected HasSynced() to return false before we started the controller")
+	}
+
 	// Run the controller and run it until we close stop.
 	stop := make(chan struct{})
 	go controller.Run(stop)
 
+	// Let's wait for the controller to do its initial sync
+	time.Sleep(100 * time.Millisecond)
+	if !controller.HasSynced() {
+		t.Errorf("Expected HasSynced() to return true after the initial sync")
+	}
+
 	wg := sync.WaitGroup{}
 	const threads = 3
 	wg.Add(threads)