test: add a stress test for WatchKind

This test isn't a regression, or a specific fix.

But I thought it'd be nice to keep it around as it excercises lots of
code paths, and it's a good oracle to see if something got broken.

Signed-off-by: Andrey Smirnov <[email protected]>

pkg/state/impl/etcd/watch_test.go

@@ -227,3 +227,120 @@ func TestWatchDeadCancel(t *testing.T) {
 		}
 	})
 }
+
+//nolint:gocognit,gocyclo,cyclop
+func TestWatchKindStress(t *testing.T) {
+	t.Cleanup(func() { goleak.VerifyNone(t, goleak.IgnoreCurrent()) })
+
+	withEtcd(t, func(s state.State) {
+		ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+		defer cancel()
+
+		const (
+			initialCreated    = 100
+			dummyWatches      = 3
+			iterations        = 20
+			numObjects        = 100
+			watchEventTimeout = 10 * time.Second
+		)
+
+		for i := range initialCreated {
+			require.NoError(t, s.Create(ctx, conformance.NewPathResource("default", fmt.Sprintf("path-%d", i))))
+		}
+
+		var watchChannels []chan state.Event
+
+		for iteration := range iterations {
+			// test setup:
+			// - on each iteration, set yet another watch with bootstrap contents
+			// - the new watch should receive 'initialContents'
+			// - add new watch to the list of watches
+			// - issue create, update, and destroy for numObjects resources
+			// - all watch channels should receive each update
+			t.Logf("iteration %d", iteration)
+
+			dummyCh := make(chan state.Event)
+
+			// these watches never receive any events, as they use a different namespace
+			for j := range dummyWatches {
+				require.NoError(t, s.WatchKind(ctx, conformance.NewPathResource(fmt.Sprintf("default-%d", j+iteration*dummyWatches), "").Metadata(), dummyCh, state.WithBootstrapContents(true)))
+			}
+
+			watchCh := make(chan state.Event, numObjects)
+
+			require.NoError(t, s.WatchKind(ctx, conformance.NewPathResource("default", "").Metadata(), watchCh, state.WithBootstrapContents(true)))
+
+			for range initialCreated {
+				select {
+				case <-time.After(watchEventTimeout):
+					t.Fatal("timeout waiting for event")
+				case ev := <-watchCh:
+					assert.Equal(t, state.Created, ev.Type)
+				}
+			}
+
+			select {
+			case <-time.After(watchEventTimeout):
+				t.Fatal("timeout waiting for event")
+			case ev := <-watchCh:
+				assert.Equal(t, state.Bootstrapped, ev.Type)
+			}
+
+			watchChannels = append(watchChannels, watchCh)
+
+			for i := range numObjects {
+				r := conformance.NewPathResource("default", fmt.Sprintf("o-%d", iteration*numObjects+i))
+
+				// add some metadata to make the object bigger
+				for j := range 5 {
+					r.Metadata().Labels().Set(fmt.Sprintf("label-%d", j), "prettybigvalueIwanttoputherejusttomakeitbig")
+				}
+
+				for j := range 5 {
+					r.Metadata().Finalizers().Add(fmt.Sprintf("finalizer-%d", j))
+				}
+
+				require.NoError(t, s.Create(ctx, r))
+
+				for j := range 5 {
+					r.Metadata().Finalizers().Remove(fmt.Sprintf("finalizer-%d", j))
+				}
+
+				require.NoError(t, s.Update(ctx, r))
+			}
+
+			for i := range numObjects {
+				require.NoError(t, s.Destroy(ctx, conformance.NewPathResource("default", fmt.Sprintf("o-%d", iteration*numObjects+i)).Metadata()))
+			}
+
+			for range numObjects {
+				for _, watchCh := range watchChannels {
+					select {
+					case <-time.After(watchEventTimeout):
+						t.Fatal("timeout waiting for event")
+					case ev := <-watchCh:
+						assert.Equal(t, state.Created, ev.Type)
+					}
+
+					select {
+					case <-time.After(watchEventTimeout):
+						t.Fatal("timeout waiting for event")
+					case ev := <-watchCh:
+						assert.Equal(t, state.Updated, ev.Type)
+					}
+				}
+			}
+
+			for range numObjects {
+				for _, watchCh := range watchChannels {
+					select {
+					case <-time.After(watchEventTimeout):
+						t.Fatal("timeout waiting for event")
+					case ev := <-watchCh:
+						assert.Equal(t, state.Destroyed, ev.Type)
+					}
+				}
+			}
+		}
+	})
+}