loki.source.docker: update to use shared scheduler

CHANGELOG.md

@@ -56,6 +56,10 @@ Main (unreleased)
 
 - Remove extraneous `output` stage from the `cri` stage pipeline in `loki.process`. (@kalleep)
 
+- Fix issue in `loki.source.docker` where scheduling containers to tail could take too long, this would increase the chance of Alloy missing shortlived jobs. (@kalleep)
+
+- Fix potential deadlock in `loki.source.docker` when component is shutting down. (@kalleep)
+
 v1.12.0
 -----------------
 

internal/component/common/loki/fanout.go

@@ -0,0 +1,63 @@
+package loki
+
+import (
+	"context"
+	"reflect"
+	"sync"
+)
+
+// NewFanout creates a new Fanout that will send log entries to the provided
+// list of LogsReceivers.
+func NewFanout(children []LogsReceiver) *Fanout {
+	return &Fanout{
+		children: children,
+	}
+}
+
+// Fanout distributes log entries to multiple LogsReceivers.
+// It is thread-safe and allows the list of receivers to be updated dynamically
+type Fanout struct {
+	mut      sync.RWMutex
+	children []LogsReceiver
+}
+
+// Send forwards a log entry to all registered receivers. It returns an error
+// if the context is cancelled while sending.
+func (f *Fanout) Send(ctx context.Context, entry Entry) error {
+	f.mut.RLock()
+	defer f.mut.RUnlock()
+	for _, recv := range f.children {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case recv.Chan() <- entry:
+		}
+	}
+	return nil
+}
+
+// UpdateChildren updates the list of receivers that will receive log entries.
+func (f *Fanout) UpdateChildren(children []LogsReceiver) {
+	f.mut.RLock()
+	if receiversChanged(f.children, children) {
+		// Upgrade lock to write.
+		f.mut.RUnlock()
+		f.mut.Lock()
+		f.children = children
+		f.mut.Unlock()
+	} else {
+		f.mut.RUnlock()
+	}
+}
+
+func receiversChanged(prev, next []LogsReceiver) bool {
+	if len(prev) != len(next) {
+		return true
+	}
+	for i := range prev {
+		if !reflect.DeepEqual(prev[i], next[i]) {
+			return true
+		}
+	}
+	return false
+}

internal/component/loki/source/consume.go

@@ -0,0 +1,27 @@
+package source
+
+import (
+	"context"
+
+	"github.com/grafana/alloy/internal/component/common/loki"
+)
+
+// Consume continuously reads log entries from recv and forwards them to the fanout f.
+// It runs until ctx is cancelled or an error occurs while sending to the fanout.
+//
+// This function is typically used in component Run methods to handle the forwarding
+// of log entries from a component's internal handler to downstream receivers.
+// The fanout allows entries to be sent to multiple receivers concurrently.
+func Consume(ctx context.Context, recv loki.LogsReceiver, f *loki.Fanout) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case entry := <-recv.Chan():
+			// NOTE: the only error we can get is context.Canceled.
+			if err := f.Send(ctx, entry); err != nil {
+				return
+			}
+		}
+	}
+}

internal/component/loki/source/consume_test.go

@@ -0,0 +1,45 @@
+package source
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"github.com/grafana/loki/pkg/push"
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/alloy/internal/component/common/loki"
+)
+
+func TestConsume(t *testing.T) {
+	consumer := loki.NewLogsReceiver()
+	producer := loki.NewLogsReceiver()
+	fanout := loki.NewFanout([]loki.LogsReceiver{consumer})
+
+	t.Run("should fanout any consumed entries", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+
+		wg := sync.WaitGroup{}
+		wg.Go(func() {
+			Consume(ctx, producer, fanout)
+		})
+
+		producer.Chan() <- loki.Entry{Entry: push.Entry{Line: "1"}}
+		e := <-consumer.Chan()
+		require.Equal(t, "1", e.Entry.Line)
+		cancel()
+		wg.Wait()
+	})
+
+	t.Run("should stop if context is canceled while trying to fanout", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		wg := sync.WaitGroup{}
+		wg.Go(func() {
+			Consume(ctx, producer, fanout)
+		})
+
+		producer.Chan() <- loki.Entry{Entry: push.Entry{Line: "1"}}
+		cancel()
+		wg.Wait()
+	})
+}