more addressing

elastic · Mar 5, 2021 · 16db6e9 · 16db6e9
1 parent ecf73e8
commit 16db6e9
Show file tree

Hide file tree

Showing 3 changed files with 26 additions and 40 deletions.
diff --git a/filebeat/input/filestream/internal/input-logfile/harvester.go b/filebeat/input/filestream/internal/input-logfile/harvester.go
@@ -93,7 +93,7 @@ func (r *readerGroup) remove(id string) {
 	delete(r.table, id)
 }
 
-func (r *readerGroup) isIDAdded(id string) bool {
+func (r *readerGroup) hasID(id string) bool {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 

diff --git a/filebeat/input/filestream/internal/input-logfile/harvester_test.go b/filebeat/input/filestream/internal/input-logfile/harvester_test.go
@@ -91,11 +91,11 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 	source := &testSource{"/path/to/test"}
 
 	requireSourceAddedToBookkeeper := func(t *testing.T, hg *defaultHarvesterGroup, s Source) {
-		require.True(t, hg.readers.isIDAdded(s.Name()))
+		require.True(t, hg.readers.hasID(s.Name()))
 	}
 
 	requireSourceRemovedFromBookkeeper := func(t *testing.T, hg *defaultHarvesterGroup, s Source) {
-		require.False(t, hg.readers.isIDAdded(s.Name()))
+		require.False(t, hg.readers.hasID(s.Name()))
 	}
 
 	t.Run("assert a harvester is started in a goroutine", func(t *testing.T) {
@@ -129,16 +129,14 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 
 		hg.Start(input.Context{Logger: logp.L(), Cancelation: context.Background()}, source)
 
-		// run commands while harvester is running
-		gorountineChecker.RunFuncWhenNewGoroutinesAreStarted(func() {
-			// wait until harvester is started
-			if mockHarvester.getRunCount() == 1 {
-				requireSourceAddedToBookkeeper(t, hg, source)
-				// after started, stop it
-				hg.Stop(source)
-				gorountineChecker.WaitUntilOriginalCount()
-			}
-		})
+		gorountineChecker.WaitUntilIncreased(1)
+		// wait until harvester is started
+		if mockHarvester.getRunCount() == 1 {
+			requireSourceAddedToBookkeeper(t, hg, source)
+			// after started, stop it
+			hg.Stop(source)
+			gorountineChecker.WaitUntilOriginalCount()
+		}
 
 		requireSourceRemovedFromBookkeeper(t, hg, source)
 	})
@@ -154,12 +152,11 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 		hg.Start(inputCtx, source)
 		hg.Start(inputCtx, source)
 
-		gorountineChecker.RunFuncWhenNewGoroutinesAreStarted(func() {
-			// error is expected as a harvester group was expected to start twice for the same source
-			for !hg.readers.isIDAdded(source.Name()) {
-			}
-			time.Sleep(3 * time.Millisecond)
-		})
+		gorountineChecker.WaitUntilIncreased(2)
+		// error is expected as a harvester group was expected to start twice for the same source
+		for !hg.readers.hasID(source.Name()) {
+		}
+		time.Sleep(3 * time.Millisecond)
 
 		hg.Stop(source)
 
@@ -225,16 +222,15 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 		wg.Add(1)
 		hg.Start(inputCtx, source)
 
-		gorountineChecker.RunFuncWhenNewGoroutinesAreStarted(func() {
-			ok := false
-			for !ok {
-				// wait until harvester is added to the bookeeper
-				ok = hg.readers.isIDAdded(source.Name())
-				if ok {
-					releaseResource(r)
-				}
+		gorountineChecker.WaitUntilIncreased(1)
+		ok := false
+		for !ok {
+			// wait until harvester is added to the bookeeper
+			ok = hg.readers.hasID(source.Name())
+			if ok {
+				releaseResource(r)
 			}
-		})
+		}
 
 		// wait until harvester.Run is done
 		wg.Wait()
@@ -260,10 +256,8 @@ func TestDefaultHarvesterGroup(t *testing.T) {
 
 		hg.Start(inputCtx, source)
 
-		gorountineChecker.RunFuncWhenNewGoroutinesAreStarted(func() {
-			err := hg.StopGroup()
-			require.Error(t, err)
-		})
+		gorountineChecker.WaitUntilIncreased(1)
+		require.Error(t, hg.StopGroup())
 
 		require.Equal(t, 0, mockHarvester.getRunCount())
 	})

diff --git a/libbeat/tests/resources/goroutines.go b/libbeat/tests/resources/goroutines.go
@@ -90,16 +90,8 @@ func (c GoroutinesChecker) WaitUntilOriginalCount() int {
 	return after
 }
 
-func (c *GoroutinesChecker) RunFuncWhenNewGoroutinesAreStarted(f func()) {
-	for runtime.NumGoroutine() == c.before {
-		time.Sleep(10 * time.Millisecond)
-	}
-	f()
-}
-
 func (c *GoroutinesChecker) WaitUntilIncreased(n int) {
 	for runtime.NumGoroutine() < c.before+n {
-		fmt.Println("before", c.before+n, "now", runtime.NumGoroutine())
 		time.Sleep(10 * time.Millisecond)
 	}
 }