fix(promtail): Handle docker logs when a log is split in multiple frames

clients/pkg/promtail/promtail.go

@@ -184,7 +184,7 @@ func (p *Promtail) reloadConfig(cfg *config.Config) error {
 	entryHandlers = append(entryHandlers, p.client)
 	p.entriesFanout = utils.NewFanoutEntryHandler(timeoutUntilFanoutHardStop, entryHandlers...)
 
-	tms, err := targets.NewTargetManagers(p, p.reg, p.logger, cfg.PositionsConfig, p.entriesFanout, cfg.ScrapeConfig, &cfg.TargetConfig, cfg.Global.FileWatch)
+	tms, err := targets.NewTargetManagers(p, p.reg, p.logger, cfg.PositionsConfig, p.entriesFanout, cfg.ScrapeConfig, &cfg.TargetConfig, cfg.Global.FileWatch, &cfg.LimitsConfig)
 	if err != nil {
 		return err
 	}

clients/pkg/promtail/targets/docker/target.go

@@ -34,6 +34,7 @@ type Target struct {
 	labels        model.LabelSet
 	relabelConfig []*relabel.Config
 	metrics       *Metrics
+	maxLineSize   int
 
 	cancel  context.CancelFunc
 	client  client.APIClient
@@ -51,6 +52,7 @@ func NewTarget(
 	labels model.LabelSet,
 	relabelConfig []*relabel.Config,
 	client client.APIClient,
+	maxLineSize int,
 ) (*Target, error) {
 
 	pos, err := position.Get(positions.CursorKey(containerName))
@@ -71,6 +73,7 @@ func NewTarget(
 		labels:        labels,
 		relabelConfig: relabelConfig,
 		metrics:       metrics,
+		maxLineSize:   maxLineSize,
 
 		client:  client,
 		running: atomic.NewBool(false),
@@ -107,8 +110,8 @@ func (t *Target) processLoop(ctx context.Context) {
 	}
 
 	// Start transferring
-	cstdout := make(chan []byte, 1)
-	cstderr := make(chan []byte, 1)
+	cstdout := make(chan []byte)
+	cstderr := make(chan []byte)
 	t.wg.Add(1)
 	go func() {
 		defer func() {
@@ -161,16 +164,21 @@ func (t *Target) process(frames chan []byte, logStream string) {
 		t.wg.Done()
 	}()
 
-	// softMaxBuffer is to prevent unlimited memory growth.
-	// In principle this introduces the same issue that we are compensating for,
-	// i.e. split messages, however there should be some limit and so this number
-	// is chosen to be "very high" just not "OOM high" (which admittedly is not
-	// a universally defined threshold).
-	const softMaxBuffer = 16 * 1024 * 1024
 	var (
-		payloadAcc = new(strings.Builder)
-		curTs      = time.Now()
+		sizeLimit            = t.maxLineSize
+		discardRemainingLine = false
+		payloadAcc           = new(strings.Builder)
+		curTs                = time.Now()
 	)
+
+	// If max_line_size is disabled (set to 0), we can in theory have infinite buffer growth.
+	// We can't guarantee that there's any bound on Docker logs, they could be an infinite stream
+	// without newlines for all we know. To protect promtail from OOM in that case, we introduce
+	// this safety limit.
+	if sizeLimit == 0 {
+		sizeLimit = 256 * 1024
+	}
+
 	for frame := range frames {
 		// Split frame into timestamp and payload
 		ts, payload, err := extractTs(string(frame))
@@ -185,34 +193,50 @@ func (t *Target) process(frames chan []byte, logStream string) {
 			ts = curTs
 		}
 
+		// If time has changed, we are looking at a new event (although we should have seen a new line..),
+		// so flush the buffer if we have one.
+		if ts != curTs {
+			discardRemainingLine = false
+			if payloadAcc.Len() > 0 {
+				t.handleOutput(logStream, curTs, payloadAcc.String())
+				payloadAcc = new(strings.Builder)
+			}
+		}
+
+		// Check if we have the end of the event
+		var isEol = strings.HasSuffix(payload, "\n")
+
+		// If we are currently discarding a line (due to size limits), skip ahead, but don't skip the next
+		// frame if we saw the end of the line.
+		if discardRemainingLine {
+			discardRemainingLine = !isEol
+			continue
+		}
+
+		// Strip newline ending if we have it
+		payload = strings.TrimRight(payload, "\r\n")
+
 		// Fast path: Most log lines are a single frame. If we have a full line in frame and buffer is empty,
 		// then don't use the buffer at all.
-		if payloadAcc.Len() == 0 && strings.HasSuffix(payload, "\n") {
+		if payloadAcc.Len() == 0 && isEol {
 			t.handleOutput(logStream, ts, payload)
 			continue
 		}
 
-		// If time has changed, we are looking at a different event (although we should have seen a new line..),
-		// so flush the buffer.
-		if payloadAcc.Len() > 0 && ts != curTs {
-			t.handleOutput(logStream, curTs, payloadAcc.String())
-			payloadAcc = new(strings.Builder)
-		}
 		// Add to buffer
 		payloadAcc.WriteString(payload)
 		curTs = ts
+
 		// Send immediately if line ended or we built a very large event
-		if strings.HasSuffix(payload, "\n") || payloadAcc.Len() >= softMaxBuffer {
+		if isEol || payloadAcc.Len() > sizeLimit {
+			discardRemainingLine = !isEol
 			t.handleOutput(logStream, curTs, payloadAcc.String())
 			payloadAcc = new(strings.Builder)
 		}
 	}
 }
 
 func (t *Target) handleOutput(logStream string, ts time.Time, payload string) {
-	// We don't output trailing newlines
-	payload = strings.TrimSuffix(payload, "\n")
-	payload = strings.TrimSuffix(payload, "\r")
 	// Add all labels from the config, relabel and filter them.
 	lb := labels.NewBuilder(nil)
 	for k, v := range t.labels {

clients/pkg/promtail/targets/docker/target_group.go

@@ -36,6 +36,7 @@ type targetGroup struct {
 	httpClientConfig config.HTTPClientConfig
 	client           client.APIClient
 	refreshInterval  model.Duration
+	maxLineSize      int
 
 	mtx     sync.Mutex
 	targets map[string]*Target
@@ -120,6 +121,7 @@ func (tg *targetGroup) addTarget(id string, discoveredLabels model.LabelSet) err
 		discoveredLabels.Merge(tg.defaultLabels),
 		tg.relabelConfig,
 		tg.client,
+		tg.maxLineSize,
 	)
 	if err != nil {
 		return err

clients/pkg/promtail/targets/docker/target_test.go

@@ -85,6 +85,7 @@ func Test_DockerTarget(t *testing.T) {
 		model.LabelSet{"job": "docker"},
 		[]*relabel.Config{},
 		client,
+		0,
 	)
 	require.NoError(t, err)
 
@@ -149,6 +150,7 @@ func doTestPartial(t *testing.T, tty bool) {
 		model.LabelSet{"job": "docker"},
 		[]*relabel.Config{},
 		client,
+		0,
 	)
 	require.NoError(t, err)
 

clients/pkg/promtail/targets/docker/targetmanager.go

@@ -43,6 +43,7 @@ func NewTargetManager(
 	positions positions.Positions,
 	pushClient api.EntryHandler,
 	scrapeConfigs []scrapeconfig.Config,
+	maxLineSize int,
 ) (*TargetManager, error) {
 	noopRegistry := util.NoopRegistry{}
 	noopSdMetrics, err := discovery.CreateAndRegisterSDMetrics(noopRegistry)
@@ -94,6 +95,7 @@ func NewTargetManager(
 						host:             sdConfig.Host,
 						httpClientConfig: sdConfig.HTTPClientConfig,
 						refreshInterval:  sdConfig.RefreshInterval,
+						maxLineSize:      maxLineSize,
 					}
 				}
 				configs[syncerKey] = append(configs[syncerKey], sdConfig)

clients/pkg/promtail/targets/docker/targetmanager_test.go

@@ -95,6 +95,7 @@ func Test_TargetManager(t *testing.T) {
 		ps,
 		entryHandler,
 		cfgs,
+		0,
 	)
 	require.NoError(t, err)
 	require.True(t, ta.Ready())

clients/pkg/promtail/targets/manager.go

@@ -9,6 +9,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 
 	"github.com/grafana/loki/v3/clients/pkg/promtail/api"
+	"github.com/grafana/loki/v3/clients/pkg/promtail/limit"
 	"github.com/grafana/loki/v3/clients/pkg/promtail/positions"
 	"github.com/grafana/loki/v3/clients/pkg/promtail/scrapeconfig"
 	"github.com/grafana/loki/v3/clients/pkg/promtail/targets/azureeventhubs"
@@ -76,6 +77,7 @@ func NewTargetManagers(
 	scrapeConfigs []scrapeconfig.Config,
 	targetConfig *file.Config,
 	watchConfig file.WatchConfig,
+	limitsConfig *limit.Config,
 ) (*TargetManagers, error) {
 	if targetConfig.Stdin {
 		level.Debug(logger).Log("msg", "configured to read from stdin")
@@ -273,7 +275,7 @@ func NewTargetManagers(
 			if err != nil {
 				return nil, err
 			}
-			cfTargetManager, err := docker.NewTargetManager(dockerMetrics, logger, pos, client, scrapeConfigs)
+			cfTargetManager, err := docker.NewTargetManager(dockerMetrics, logger, pos, client, scrapeConfigs, limitsConfig.MaxLineSize.Val())
 			if err != nil {
 				return nil, errors.Wrap(err, "failed to make Docker service discovery target manager")
 			}