Make sure all existing objects are announced when a watch is created (#89)

#88

Signed-off-by: Karol Zadora-Przylecki <[email protected]>

Author: karolz-ms

pkg/storage/filepath/jsonfile_rest.go

@@ -549,33 +549,40 @@ func (f *filepathREST) Watch(ctx context.Context, options *metainternalversion.L
 	p := newSelectionPredicate(options)
 	jw := f.watchSet.newWatch()
 
-	// On initial watch, send all the existing objects
-	list, err := f.List(ctx, options)
-	if err != nil {
-		return nil, err
-	}
-
-	danger := reflect.ValueOf(list).Elem()
-	items := danger.FieldByName("Items")
-
-	initEvents := []watch.Event{}
-	for i := 0; i < items.Len(); i++ {
-		obj := items.Index(i).Addr().Interface().(runtime.Object)
-		ok, err := p.Matches(obj)
+	getInitEvents := func() ([]watch.Event, error) {
+		// On initial watch, send all the existing objects.
+		// We may receive duplicated "Added" events for some objects via the watch updata channel,
+		// and we may report them twice, but that is much better than not reporting them at all,
+		// and having clients left unaware that an object they might be interested in was created.
+		// See https://github.com/tilt-dev/tilt-apiserver/issues/88
+
+		list, err := f.List(ctx, options)
 		if err != nil {
 			return nil, err
 		}
-		if !ok {
-			continue
+		danger := reflect.ValueOf(list).Elem()
+		items := danger.FieldByName("Items")
+
+		initEvents := []watch.Event{}
+		for i := 0; i < items.Len(); i++ {
+			obj := items.Index(i).Addr().Interface().(runtime.Object)
+			ok, err := p.Matches(obj)
+			if err != nil {
+				return nil, err
+			}
+			if !ok {
+				continue
+			}
+			initEvents = append(initEvents, watch.Event{
+				Type:   watch.Added,
+				Object: obj,
+			})
 		}
-		initEvents = append(initEvents, watch.Event{
-			Type:   watch.Added,
-			Object: obj,
-		})
+		return initEvents, nil
 	}
-	jw.Start(p, initEvents)
 
-	return jw, nil
+	startErr := jw.Start(p, getInitEvents)
+	return jw, startErr
 }
 
 func (f *filepathREST) conflictErr(name string) error {

pkg/storage/filepath/jsonfile_rest_test.go

@@ -19,6 +19,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apimachinery/pkg/watch"
 	genericapirequest "k8s.io/apiserver/pkg/endpoints/request"
 	"k8s.io/apiserver/pkg/registry/generic"
@@ -315,6 +316,99 @@ func TestFilepathREST_UpdateIdentical(t *testing.T) {
 	require.Equal(t, "2", actual.ResourceVersion)
 }
 
+func TestFilepathREST_ParallelCreateAndWatch(t *testing.T) {
+	f := newRESTFixture(t)
+	defer f.tearDown()
+
+	// Record unique object creation events
+	created := sync.Map{}
+
+	eventRecorder := func() {
+		sendInitialEvents := true
+		w, werr := f.watcher().Watch(f.rootCtx, &metainternalversion.ListOptions{
+			Watch:             true,
+			SendInitialEvents: &sendInitialEvents,
+		})
+		require.NoError(t, werr)
+		defer w.Stop()
+
+		for {
+			select {
+			case <-f.rootCtx.Done():
+				return
+			case e, isOpen := <-w.ResultChan():
+				if !isOpen {
+					return
+				}
+				if e.Type == watch.Added {
+					created.Store(e.Object.(metav1.Object).GetName(), true)
+				}
+			}
+		}
+	}
+
+	// Spawn a bunch of workers that create objects
+	const TotalObjects = 500
+	const Workers = 20
+	const objNamePrefix = "parallel-create-and-watch-"
+	var wready, wdone sync.WaitGroup
+	wready.Add(Workers)
+	wdone.Add(Workers)
+	objCreator := f.creater()
+	start := make(chan struct{})
+
+	for i := 0; i < Workers; i++ {
+		go func(worker int) {
+			wready.Done()
+			<-start // Start all workers at the same time
+
+			objectsToCreate := TotalObjects / Workers
+
+			for j := 0; j < objectsToCreate; j++ {
+				// When the first worker is in the middle of creating their objects,
+				// start the watcher/event counter.
+				if worker == 0 && j == objectsToCreate/2 {
+					go eventRecorder()
+				}
+
+				objName := fmt.Sprintf("%s-%d-%d", objNamePrefix, worker, j)
+				obj := &v1alpha1.Manifest{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: objName,
+					},
+					Spec: v1alpha1.ManifestSpec{
+						Message: objName,
+					},
+				}
+				_, err := objCreator.Create(f.rootCtx, obj, nil, nil)
+				require.NoError(t, err)
+			}
+
+			wdone.Done()
+		}(i)
+	}
+
+	wready.Wait()
+	close(start)
+	wdone.Wait()
+
+	// Wait for all Added events for the objects we created
+	count := 0
+	waitErr := wait.PollUntilContextTimeout(f.rootCtx, 200*time.Millisecond, 10*time.Second, true, func(ctx context.Context) (bool, error) {
+		count = 0
+		created.Range(func(key, value interface{}) bool {
+			count++
+			return true
+		})
+
+		if count == TotalObjects {
+			return true, nil
+		}
+		return false, nil
+	})
+	require.NoError(t, waitErr, "Did not receive expected number of Added events (received %d, expected %d)", count, TotalObjects)
+}
+
 type restOptionsGetter struct {
 	codec runtime.Codec
 }

pkg/storage/filepath/watchset.go

@@ -53,10 +53,14 @@ type watchNode struct {
 }
 
 // Start sending events to this watch.
-func (w *watchNode) Start(p storage.SelectionPredicate, initEvents []watch.Event) {
+func (w *watchNode) Start(p storage.SelectionPredicate, initEventFactory func() ([]watch.Event, error)) error {
 	w.s.mu.Lock()
+	defer w.s.mu.Unlock()
 	w.s.nodes[w.id] = w
-	w.s.mu.Unlock()
+	initEvents, err := initEventFactory()
+	if err != nil {
+		return err
+	}
 
 	go func() {
 		for _, e := range initEvents {
@@ -76,6 +80,8 @@ func (w *watchNode) Start(p storage.SelectionPredicate, initEvents []watch.Event
 		}
 		close(w.outCh)
 	}()
+
+	return nil
 }
 
 func (w *watchNode) Stop() {