Fix promtail deadlock

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## Main
 
+* [5170](https://github.com/grafana/loki/pull/5170) **chaudum** Fix deadlock in Promtail caused when targets got removed from a target group by the discovery manager.
 * [5163](https://github.com/grafana/loki/pull/5163) **chaudum** Fix regression in fluentd plugin introduced with #5107 that caused `NoMethodError` when parsing non-string values of log lines.
 * [5144](https://github.com/grafana/loki/pull/5144) **dannykopping** Ruler: fix remote write basic auth credentials.
 * [5107](https://github.com/grafana/loki/pull/5107) **chaudum** Fix bug in fluentd plugin that caused log lines containing non UTF-8 characters to be dropped.

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

@@ -319,6 +319,15 @@ func (s *targetSyncer) sync(groups []*targetgroup.Group, targetEventHandler chan
 			target.Stop()
 			s.metrics.targetsActive.Add(-1.)
 			delete(s.targets, key)
+
+			// close related file event watcher
+			k := target.path
+			if _, ok := s.fileEventWatchers[k]; ok {
+				close(s.fileEventWatchers[k])
+				delete(s.fileEventWatchers, k)
+			} else {
+				level.Warn(s.log).Log("msg", "failed to remove file event watcher", "path", k)
+			}
 		}
 	}
 	s.droppedTargets = dropped
@@ -376,6 +385,7 @@ func (s *targetSyncer) ready() bool {
 	}
 	return false
 }
+
 func (s *targetSyncer) stop() {
 	s.mtx.Lock()
 	defer s.mtx.Unlock()
@@ -385,6 +395,7 @@ func (s *targetSyncer) stop() {
 		target.Stop()
 		delete(s.targets, key)
 	}
+
 	for key, watcher := range s.fileEventWatchers {
 		close(watcher)
 		delete(s.fileEventWatchers, key)

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

@@ -6,11 +6,13 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/go-kit/log"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/discovery"
 	"github.com/prometheus/prometheus/discovery/targetgroup"
+	"gopkg.in/fsnotify.v1"
 
 	"github.com/grafana/loki/clients/pkg/promtail/api"
 	"github.com/grafana/loki/clients/pkg/promtail/client/fake"
@@ -447,3 +449,54 @@ func TestGlobWithMultipleFiles(t *testing.T) {
 		t.Error("Handler did not receive the correct number of messages, expected 20 received", len(client.Received()))
 	}
 }
+
+func TestDeadlockTargetManager(t *testing.T) {
+	client := fake.New(func() {})
+	defer client.Stop()
+
+	targetEventHandler := make(chan fileTargetEvent)
+	defer func() {
+		close(targetEventHandler)
+	}()
+
+	syncer := &targetSyncer{
+		metrics:           NewMetrics(nil),
+		log:               log.NewLogfmtLogger(log.NewSyncWriter(os.Stderr)),
+		positions:         nil,
+		entryHandler:      client,
+		hostname:          "localhost",
+		fileEventWatchers: make(map[string]chan fsnotify.Event),
+		targets:           make(map[string]*FileTarget),
+		targetConfig: &Config{
+			SyncPeriod: time.Hour,
+		},
+	}
+
+	syncer.sync([]*targetgroup.Group{
+		{
+			Targets: []model.LabelSet{
+				{
+					hostLabel: "localhost",
+					pathLabel: "baz",
+					"job":     "bar",
+				},
+			},
+		},
+	}, targetEventHandler)
+
+	require.Equal(t, len(syncer.targets), 1)
+	require.Equal(t, len(syncer.fileEventWatchers), 1)
+
+	syncer.sync([]*targetgroup.Group{
+		{
+			Targets: []model.LabelSet{
+				{},
+			},
+		},
+	}, targetEventHandler)
+
+	syncer.sendFileCreateEvent(fsnotify.Event{Name: "baz"})
+
+	require.Equal(t, len(syncer.targets), 0)
+	require.Equal(t, len(syncer.fileEventWatchers), 0)
+}

clients/pkg/promtail/targets/manager.go

@@ -9,7 +9,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/grafana/loki/clients/pkg/promtail/api"
-	"github.com/grafana/loki/clients/pkg/promtail/client"
 	"github.com/grafana/loki/clients/pkg/promtail/positions"
 	"github.com/grafana/loki/clients/pkg/promtail/scrapeconfig"
 	"github.com/grafana/loki/clients/pkg/promtail/targets/cloudflare"
@@ -59,21 +58,19 @@ func NewTargetManagers(
 	client api.EntryHandler,
 	scrapeConfigs []scrapeconfig.Config,
 	targetConfig *file.Config,
-	clientConfigs ...client.Config,
 ) (*TargetManagers, error) {
-	var targetManagers []targetManager
-	targetScrapeConfigs := make(map[string][]scrapeconfig.Config, 4)
-
 	if targetConfig.Stdin {
 		level.Debug(logger).Log("msg", "configured to read from stdin")
 		stdin, err := stdin.NewStdinTargetManager(reg, logger, app, client, scrapeConfigs)
 		if err != nil {
 			return nil, err
 		}
-		targetManagers = append(targetManagers, stdin)
-		return &TargetManagers{targetManagers: targetManagers}, nil
+		return &TargetManagers{targetManagers: []targetManager{stdin}}, nil
 	}
 
+	var targetManagers []targetManager
+	targetScrapeConfigs := make(map[string][]scrapeconfig.Config, 4)
+
 	for _, cfg := range scrapeConfigs {
 		switch {
 		case cfg.HasServiceDiscoveryConfig():