fix(promtail): remove flaky TestFileTarget_StopsTailersCleanly (#16473)

Author: cyriltovena

clients/pkg/promtail/targets/file/filetarget.go

@@ -50,15 +50,15 @@ type WatchConfig struct {
 	MaxPollFrequency time.Duration `mapstructure:"max_poll_frequency" yaml:"max_poll_frequency"`
 }
 
-var DefaultWatchConig = WatchConfig{
+var DefaultWatchConfig = WatchConfig{
 	MinPollFrequency: 250 * time.Millisecond,
 	MaxPollFrequency: 250 * time.Millisecond,
 }
 
 // RegisterFlags with prefix registers flags where every name is prefixed by
 // prefix. If prefix is a non-empty string, prefix should end with a period.
 func (cfg *WatchConfig) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
-	d := DefaultWatchConig
+	d := DefaultWatchConfig
 
 	f.DurationVar(&cfg.MinPollFrequency, prefix+"min_poll_frequency", d.MinPollFrequency, "Minimum period to poll for file changes")
 	f.DurationVar(&cfg.MaxPollFrequency, prefix+"max_poll_frequency", d.MaxPollFrequency, "Maximum period to poll for file changes")
@@ -247,7 +247,6 @@ func (t *FileTarget) sync() error {
 	} else {
 		// Gets current list of files to tail.
 		matches, err = doublestar.FilepathGlob(t.path)
-
 		if err != nil {
 			return errors.Wrap(err, "filetarget.sync.filepath.Glob")
 		}
@@ -257,7 +256,6 @@ func (t *FileTarget) sync() error {
 		matchesExcluded = []string{t.pathExclude}
 	} else {
 		matchesExcluded, err = doublestar.FilepathGlob(t.pathExclude)
-
 		if err != nil {
 			return errors.Wrap(err, "filetarget.sync.filepathexclude.Glob")
 		}

clients/pkg/promtail/targets/file/filetarget_test.go

@@ -8,7 +8,6 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
-	"sync"
 	"testing"
 	"time"
 
@@ -72,7 +71,7 @@ func TestFileTargetSync(t *testing.T) {
 	path := logDir1 + "/*.log"
 	target, err := NewFileTarget(metrics, logger, client, ps, path, "", nil, nil, &Config{
 		SyncPeriod: 1 * time.Minute, // assure the sync is not called by the ticker
-	}, DefaultWatchConig, nil, fakeHandler, "", nil)
+	}, DefaultWatchConfig, nil, fakeHandler, "", nil)
 	assert.NoError(t, err)
 
 	// Start with nothing watched.
@@ -84,7 +83,7 @@ func TestFileTargetSync(t *testing.T) {
 	}
 
 	// Create the base dir, still nothing watched.
-	err = os.MkdirAll(logDir1, 0750)
+	err = os.MkdirAll(logDir1, 0o750)
 	assert.NoError(t, err)
 
 	err = target.sync()
@@ -191,7 +190,7 @@ func TestFileTarget_StopsTailersCleanly(t *testing.T) {
 	registry := prometheus.NewRegistry()
 	target, err := NewFileTarget(NewMetrics(registry), logger, client, ps, pathToWatch, "", nil, nil, &Config{
 		SyncPeriod: 10 * time.Millisecond,
-	}, DefaultWatchConig, nil, fakeHandler, "", nil)
+	}, DefaultWatchConfig, nil, fakeHandler, "", nil)
 	assert.NoError(t, err)
 
 	_, err = os.Create(logFile)
@@ -247,95 +246,6 @@ func TestFileTarget_StopsTailersCleanly(t *testing.T) {
 	`), "promtail_files_active_total"))
 }
 
-func TestFileTarget_StopsTailersCleanly_Parallel(t *testing.T) {
-	w := log.NewSyncWriter(os.Stderr)
-	logger := log.NewLogfmtLogger(w)
-
-	tempDir := t.TempDir()
-	positionsFileName := filepath.Join(tempDir, "positions.yml")
-
-	ps, err := positions.New(logger, positions.Config{
-		SyncPeriod:    10 * time.Millisecond,
-		PositionsFile: positionsFileName,
-	})
-	require.NoError(t, err)
-
-	client := fake.New(func() {})
-	defer client.Stop()
-
-	pathToWatch := filepath.Join(tempDir, "*.log")
-	registry := prometheus.NewRegistry()
-	metrics := NewMetrics(registry)
-
-	// Increase this to several thousand to make the test more likely to fail when debugging a race condition
-	iterations := 500
-	fakeHandler := make(chan fileTargetEvent, 10*iterations)
-	for i := 0; i < iterations; i++ {
-		logFile := filepath.Join(tempDir, fmt.Sprintf("test_%d.log", i))
-
-		target, err := NewFileTarget(metrics, logger, client, ps, pathToWatch, "", nil, nil, &Config{
-			SyncPeriod: 10 * time.Millisecond,
-		}, DefaultWatchConig, nil, fakeHandler, "", nil)
-		assert.NoError(t, err)
-
-		file, err := os.Create(logFile)
-		assert.NoError(t, err)
-
-		// Write some data to the file
-		for j := 0; j < 5; j++ {
-			_, _ = file.WriteString(fmt.Sprintf("test %d\n", j))
-		}
-		require.NoError(t, file.Close())
-
-		requireEventually(t, func() bool {
-			return testutil.CollectAndCount(registry, "promtail_read_lines_total") == 1
-		}, "expected 1 read_lines_total metric")
-
-		requireEventually(t, func() bool {
-			return testutil.CollectAndCount(registry, "promtail_read_bytes_total") == 1
-		}, "expected 1 read_bytes_total metric")
-
-		requireEventually(t, func() bool {
-			return testutil.ToFloat64(metrics.readLines) == 5
-		}, "expected 5 read_lines_total")
-
-		requireEventually(t, func() bool {
-			return testutil.ToFloat64(metrics.totalBytes) == 35
-		}, "expected 35 total_bytes")
-
-		requireEventually(t, func() bool {
-			return testutil.ToFloat64(metrics.readBytes) == 35
-		}, "expected 35 read_bytes")
-
-		// Concurrently stop the target and remove the file
-		wg := sync.WaitGroup{}
-		wg.Add(2)
-		go func() {
-			sleepRandomDuration(time.Millisecond * 10)
-			target.Stop()
-			wg.Done()
-
-		}()
-		go func() {
-			sleepRandomDuration(time.Millisecond * 10)
-			_ = os.Remove(logFile)
-			wg.Done()
-		}()
-
-		wg.Wait()
-
-		requireEventually(t, func() bool {
-			return testutil.CollectAndCount(registry, "promtail_read_bytes_total") == 0
-		}, "expected read_bytes_total metric to be cleaned up")
-
-		requireEventually(t, func() bool {
-			return testutil.CollectAndCount(registry, "promtail_file_bytes_total") == 0
-		}, "expected file_bytes_total metric to be cleaned up")
-	}
-
-	ps.Stop()
-}
-
 // Make sure that Stop() doesn't hang if FileTarget is waiting on a channel send.
 func TestFileTarget_StopAbruptly(t *testing.T) {
 	w := log.NewSyncWriter(os.Stderr)
@@ -367,11 +277,11 @@ func TestFileTarget_StopAbruptly(t *testing.T) {
 	registry := prometheus.NewRegistry()
 	target, err := NewFileTarget(NewMetrics(registry), logger, client, ps, pathToWatch, "", nil, nil, &Config{
 		SyncPeriod: 10 * time.Millisecond,
-	}, DefaultWatchConig, nil, fakeHandler, "", nil)
+	}, DefaultWatchConfig, nil, fakeHandler, "", nil)
 	assert.NoError(t, err)
 
 	// Create a directory, still nothing is watched.
-	err = os.MkdirAll(logDir1, 0750)
+	err = os.MkdirAll(logDir1, 0o750)
 	assert.NoError(t, err)
 	_, err = os.Create(logfile1)
 	assert.NoError(t, err)
@@ -392,12 +302,12 @@ func TestFileTarget_StopAbruptly(t *testing.T) {
 
 	// Create two directories - one more than the buffer of fakeHandler,
 	// so that the file target hands until we call Stop().
-	err = os.MkdirAll(logDir2, 0750)
+	err = os.MkdirAll(logDir2, 0o750)
 	assert.NoError(t, err)
 	_, err = os.Create(logfile2)
 	assert.NoError(t, err)
 
-	err = os.MkdirAll(logDir3, 0750)
+	err = os.MkdirAll(logDir3, 0o750)
 	assert.NoError(t, err)
 	_, err = os.Create(logfile3)
 	assert.NoError(t, err)
@@ -479,7 +389,7 @@ func TestFileTargetPathExclusion(t *testing.T) {
 	pathExclude := filepath.Join(dirName, "log3", "*.log")
 	target, err := NewFileTarget(metrics, logger, client, ps, path, pathExclude, nil, nil, &Config{
 		SyncPeriod: 1 * time.Minute, // assure the sync is not called by the ticker
-	}, DefaultWatchConig, nil, fakeHandler, "", nil)
+	}, DefaultWatchConfig, nil, fakeHandler, "", nil)
 	assert.NoError(t, err)
 
 	// Start with nothing watched.
@@ -491,11 +401,11 @@ func TestFileTargetPathExclusion(t *testing.T) {
 	}
 
 	// Create the base directories, still nothing watched.
-	err = os.MkdirAll(logDir1, 0750)
+	err = os.MkdirAll(logDir1, 0o750)
 	assert.NoError(t, err)
-	err = os.MkdirAll(logDir2, 0750)
+	err = os.MkdirAll(logDir2, 0o750)
 	assert.NoError(t, err)
-	err = os.MkdirAll(logDir3, 0750)
+	err = os.MkdirAll(logDir3, 0o750)
 	assert.NoError(t, err)
 
 	err = target.sync()
@@ -571,7 +481,7 @@ func TestHandleFileCreationEvent(t *testing.T) {
 	logFile := filepath.Join(logDir, "test1.log")
 	logFileIgnored := filepath.Join(logDir, "test.donot.log")
 
-	if err := os.MkdirAll(logDir, 0750); err != nil {
+	if err := os.MkdirAll(logDir, 0o750); err != nil {
 		t.Fatal(err)
 	}
 
@@ -610,7 +520,7 @@ func TestHandleFileCreationEvent(t *testing.T) {
 	target, err := NewFileTarget(metrics, logger, client, ps, path, pathExclude, nil, nil, &Config{
 		// To handle file creation event from channel, set enough long time as sync period
 		SyncPeriod: 10 * time.Minute,
-	}, DefaultWatchConig, fakeFileHandler, fakeTargetHandler, "", nil)
+	}, DefaultWatchConfig, fakeFileHandler, fakeTargetHandler, "", nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -653,7 +563,6 @@ func TestToStopTailing(t *testing.T) {
 			t.Error("Results mismatch, expected", expected[i], "got", st[i])
 		}
 	}
-
 }
 
 func BenchmarkToStopTailing(b *testing.B) {
@@ -717,7 +626,6 @@ func TestMissing(t *testing.T) {
 	if _, ok := c["str3"]; !ok {
 		t.Error("Expected the set to contain str3 but it did not")
 	}
-
 }
 
 func requireEventually(t *testing.T, f func() bool, msg string) {

clients/pkg/promtail/targets/file/filetargetmanager_test.go

@@ -26,7 +26,7 @@ import (
 func newTestLogDirectories(t *testing.T) string {
 	tmpDir := t.TempDir()
 	logFileDir := filepath.Join(tmpDir, "logs")
-	err := os.MkdirAll(logFileDir, 0750)
+	err := os.MkdirAll(logFileDir, 0o750)
 	assert.NoError(t, err)
 	return logFileDir
 }
@@ -71,7 +71,7 @@ func newTestFileTargetManager(logger log.Logger, client api.EntryHandler, positi
 	}
 
 	metrics := NewMetrics(nil)
-	ftm, err := NewFileTargetManager(metrics, logger, positions, client, []scrapeconfig.Config{sc}, tc, DefaultWatchConig)
+	ftm, err := NewFileTargetManager(metrics, logger, positions, client, []scrapeconfig.Config{sc}, tc, DefaultWatchConfig)
 	if err != nil {
 		return nil, err
 	}
@@ -492,7 +492,7 @@ func TestDeadlockStartWatchingDuringSync(t *testing.T) {
 	go func() {
 		for i := 0; i < 10; i++ {
 			dir := filepath.Join(newLogDir, fmt.Sprintf("%d", i))
-			err := os.MkdirAll(dir, 0750)
+			err := os.MkdirAll(dir, 0o750)
 			assert.NoError(t, err)
 			time.Sleep(1 * time.Millisecond)
 			for j := 0; j < 100; j++ {
@@ -551,13 +551,13 @@ func TestLabelSetUpdate(t *testing.T) {
 		},
 	}
 
-	var target = model.LabelSet{
+	target := model.LabelSet{
 		hostLabel: "localhost",
 		pathLabel: "baz",
 		"job":     "foo",
 	}
 
-	var target2 = model.LabelSet{
+	target2 := model.LabelSet{
 		hostLabel: "localhost",
 		pathLabel: "baz",
 		"job":     "foo2",
@@ -593,7 +593,6 @@ func TestLabelSetUpdate(t *testing.T) {
 	}, targetEventHandler)
 	require.Equal(t, 0, len(syncer.targets))
 	require.Equal(t, 0, len(syncer.fileEventWatchers))
-
 }
 
 func TestFulfillKubePodSelector(t *testing.T) {