Promtail WAL support: Implement writer side

clients/pkg/promtail/promtail.go

@@ -178,7 +178,7 @@ func (p *Promtail) reloadConfig(cfg *config.Config) error {
 		entryHandlers = append(entryHandlers, p.walWriter)
 	}
 
-	p.entriesFanout = utils.NewEntryHandlerFanouter(entryHandlers...)
+	p.entriesFanout = utils.NewFanoutEntryHandler(entryHandlers...)
 
 	tms, err := targets.NewTargetManagers(p, p.reg, p.logger, cfg.PositionsConfig, p.entriesFanout, cfg.ScrapeConfig, &cfg.TargetConfig)
 	if err != nil {

clients/pkg/promtail/utils/entries.go

@@ -6,38 +6,51 @@ import (
 	"github.com/grafana/loki/clients/pkg/promtail/api"
 )
 
-// EntryHandlerFanouter implements api.EntryHandler, fanning out received entries to one or multiple channels.
-type EntryHandlerFanouter struct {
+// FanoutEntryHandler implements api.EntryHandler, fanning out received entries to one or multiple channels.
+type FanoutEntryHandler struct {
 	entries  chan api.Entry
 	handlers []api.EntryHandler
 
 	once sync.Once
 	wg   sync.WaitGroup
 }
 
-func NewEntryHandlerFanouter(handlers ...api.EntryHandler) *EntryHandlerFanouter {
-	multiplex := &EntryHandlerFanouter{
+func NewFanoutEntryHandler(handlers ...api.EntryHandler) *FanoutEntryHandler {
+	eh := &FanoutEntryHandler{
 		entries:  make(chan api.Entry),
 		handlers: handlers,
 	}
-	multiplex.wg.Add(1)
+	eh.start()
+	return eh
+}
+
+func (eh *FanoutEntryHandler) start() {
+	eh.wg.Add(1)
 	go func() {
-		defer multiplex.wg.Done()
-		for e := range multiplex.entries {
-			for _, handler := range multiplex.handlers {
-				handler.Chan() <- e
+		defer eh.wg.Done()
+		for e := range eh.entries {
+			// To prevent a single channel from blocking all others, we run each channel send in a separate go routine.
+			// This cause each entry to be sent in |eh.handlers| routines concurrently. When all finish, we know the entry
+			// has been fanned out properly, and we can read the next received entry.
+			var entryWG sync.WaitGroup
+			entryWG.Add(len(eh.handlers))
+			for _, handler := range eh.handlers {
+				go func(eh api.EntryHandler) {
+					defer entryWG.Done()
+					eh.Chan() <- e
+				}(handler)
 			}
+			entryWG.Wait()
 		}
 	}()
-	return multiplex
 }
 
-func (eh *EntryHandlerFanouter) Chan() chan<- api.Entry {
+func (eh *FanoutEntryHandler) Chan() chan<- api.Entry {
 	return eh.entries
 }
 
-// Stop only stops the channel EntryHandlerFanouter exposes, not the ones it fans out to.
-func (eh *EntryHandlerFanouter) Stop() {
+// Stop only stops the channel FanoutEntryHandler exposes, not the ones it fans out to.
+func (eh *FanoutEntryHandler) Stop() {
 	eh.once.Do(func() {
 		close(eh.entries)
 	})

clients/pkg/promtail/utils/entries_test.go

@@ -12,10 +12,16 @@ import (
 	"github.com/grafana/loki/pkg/logproto"
 )
 
-func TestEntryHandlerFanouter(t *testing.T) {
+func TestFanoutEntryHandler(t *testing.T) {
 	eh1 := newSavingEntryHandler()
 	eh2 := newSavingEntryHandler()
-	fanouter := NewEntryHandlerFanouter(eh1, eh2)
+	fanout := NewFanoutEntryHandler(eh1, eh2)
+
+	defer func() {
+		fanout.Stop()
+		eh2.Stop()
+		eh1.Stop()
+	}()
 
 	var expectedLines = []string{
 		"some line",
@@ -24,9 +30,9 @@ func TestEntryHandlerFanouter(t *testing.T) {
 	}
 
 	for _, line := range expectedLines {
-		fanouter.Chan() <- api.Entry{
+		fanout.Chan() <- api.Entry{
 			Labels: model.LabelSet{
-				"test": "fanouter",
+				"test": "fanout",
 			},
 			Entry: logproto.Entry{
 				Timestamp: time.Now(),
@@ -35,12 +41,9 @@ func TestEntryHandlerFanouter(t *testing.T) {
 		}
 	}
 
-	fanouter.Stop()
-	eh2.Stop()
-	eh1.Stop()
-
-	require.Len(t, eh1.Received, len(expectedLines))
-	require.Len(t, eh2.Received, len(expectedLines))
+	require.Eventually(t, func() bool {
+		return len(eh1.Received) == len(expectedLines) && len(eh2.Received) == len(expectedLines)
+	}, time.Second*10, time.Second, "expected entries to be received by fanned out channels")
 }
 
 type savingEntryHandler struct {
@@ -64,11 +67,11 @@ func newSavingEntryHandler() *savingEntryHandler {
 	return eh
 }
 
-func (x *savingEntryHandler) Chan() chan<- api.Entry {
-	return x.entries
+func (eh *savingEntryHandler) Chan() chan<- api.Entry {
+	return eh.entries
 }
 
-func (x *savingEntryHandler) Stop() {
-	close(x.entries)
-	x.wg.Wait()
+func (eh *savingEntryHandler) Stop() {
+	close(eh.entries)
+	eh.wg.Wait()
 }