Change to start readLine start and stops updatePositions

Author: kalleep

internal/component/loki/source/file/decompresser.go

@@ -62,10 +62,6 @@ type decompressor struct {
 	cfg      DecompressionConfig
 
 	componentStopping func() bool
-
-	posquit chan struct{} // used by the readLine method to tell the updatePosition method to stop
-	posdone chan struct{} // used by the updatePosition method to notify when it stopped
-	done    chan struct{} // used by the readLine method to notify when it stopped
 }
 
 func newDecompressor(
@@ -166,38 +162,30 @@ func (d *decompressor) Run(ctx context.Context) {
 	handler := loki.AddLabelsMiddleware(labelsMiddleware).Wrap(loki.NewEntryHandler(d.receiver.Chan(), func() {}))
 	defer handler.Stop()
 
-	d.posquit = make(chan struct{})
-	d.posdone = make(chan struct{})
-	d.done = make(chan struct{})
-
 	d.metrics.filesActive.Add(1.)
-	go func() {
-		// updatePosition closes the channel t.posdone on exit
-		d.updatePosition()
-	}()
 
+	done := make(chan struct{})
 	ctx, cancel := context.WithCancel(ctx)
 	go func() {
-		// readLines closes the channels t.done and t.posquit on exit
-		d.readLines(handler)
+		// readLines closes the done on exit
+		d.readLines(handler, done)
 		cancel()
 	}()
 
 	d.running.Store(true)
 	defer d.running.Store(false)
 
 	<-ctx.Done()
-	d.stop()
+	d.stop(done)
 }
 
-func (d *decompressor) updatePosition() {
+func (d *decompressor) updatePosition(posquit chan struct{}) {
 	positionSyncPeriod := d.positions.SyncPeriod()
 	positionWait := time.NewTicker(positionSyncPeriod)
 	defer func() {
 		positionWait.Stop()
 		level.Info(d.logger).Log("msg", "position timer: exited", "path", d.path)
 		d.cleanupMetrics()
-		close(d.posdone)
 	}()
 
 	for {
@@ -207,7 +195,7 @@ func (d *decompressor) updatePosition() {
 				level.Error(d.logger).Log("msg", "position timer: error getting position and/or size, stopping decompressor", "path", d.path, "error", err)
 				return
 			}
-		case <-d.posquit:
+		case <-posquit:
 			return
 		}
 	}
@@ -218,18 +206,26 @@ func (d *decompressor) updatePosition() {
 // It first decompresses the file as a whole using a reader and then it will iterate
 // over its chunks, separated by '\n'.
 // During each iteration, the parsed and decoded log line is then sent to the API with the current timestamp.
-func (d *decompressor) readLines(handler loki.EntryHandler) {
+// done channel is closed when readlines exits.
+func (d *decompressor) readLines(handler loki.EntryHandler, done chan struct{}) {
 	level.Info(d.logger).Log("msg", "read lines routine: started", "path", d.path)
 
 	if d.cfg.InitialDelay > 0 {
 		level.Info(d.logger).Log("msg", "sleeping before starting decompression", "path", d.path, "duration", d.cfg.InitialDelay.String())
 		time.Sleep(d.cfg.InitialDelay)
 	}
 
+	posquit, posdone := make(chan struct{}), make(chan struct{})
+	go func() {
+		d.updatePosition(posquit)
+		close(posdone)
+	}()
+
 	defer func() {
 		level.Info(d.logger).Log("msg", "read lines routine finished", "path", d.path)
-		close(d.done)
-		close(d.posquit)
+		close(posquit)
+		<-posdone
+		close(done)
 	}()
 
 	entries := handler.Chan()
@@ -318,11 +314,9 @@ func (d *decompressor) markPositionAndSize() error {
 	return nil
 }
 
-func (d *decompressor) stop() {
+func (d *decompressor) stop(done chan struct{}) {
 	// Wait for readLines() to consume all the remaining messages and exit when the channel is closed
-	<-d.done
-	// Wait for the position marker thread to exit
-	<-d.posdone
+	<-done
 
 	level.Info(d.logger).Log("msg", "stopped decompressor", "path", d.path)
 

internal/component/loki/source/file/decompresser_test.go

@@ -50,6 +50,24 @@ func newNoopClient() *noopClient {
 	return c
 }
 
+var _ positions.Positions = (*noopPositions)(nil)
+
+type noopPositions struct{}
+
+func (n *noopPositions) Get(path string, labels string) (int64, error) { return 0, nil }
+
+func (n *noopPositions) GetString(path string, labels string) string { return "" }
+
+func (n *noopPositions) Put(path string, labels string, pos int64) {}
+
+func (n *noopPositions) PutString(path string, labels string, pos string) {}
+
+func (n *noopPositions) Remove(path string, labels string) {}
+
+func (n *noopPositions) Stop() {}
+
+func (n *noopPositions) SyncPeriod() time.Duration { return 10 * time.Second }
+
 func BenchmarkReadlines(b *testing.B) {
 	entryHandler := newNoopClient()
 
@@ -79,9 +97,9 @@ func BenchmarkReadlines(b *testing.B) {
 			for i := 0; i < b.N; i++ {
 				newDec := decBase
 				newDec.metrics = newMetrics(prometheus.NewRegistry())
-				newDec.done = make(chan struct{})
-				newDec.readLines(entryHandler)
-				<-newDec.done
+				done := make(chan struct{})
+				newDec.readLines(entryHandler, done)
+				<-done
 			}
 		})
 	}
@@ -93,21 +111,20 @@ func TestGigantiqueGunzipFile(t *testing.T) {
 	defer handler.Stop()
 
 	d := &decompressor{
-		logger:   log.NewNopLogger(),
-		running:  atomic.NewBool(false),
-		receiver: loki.NewLogsReceiver(),
-		path:     file,
-		done:     make(chan struct{}),
-		posquit:  make(chan struct{}),
-		metrics:  newMetrics(prometheus.NewRegistry()),
-		cfg:      DecompressionConfig{Format: "gz"},
+		logger:    log.NewNopLogger(),
+		running:   atomic.NewBool(false),
+		receiver:  loki.NewLogsReceiver(),
+		path:      file,
+		metrics:   newMetrics(prometheus.NewRegistry()),
+		cfg:       DecompressionConfig{Format: "gz"},
+		positions: &noopPositions{},
 	}
 
-	d.readLines(handler)
+	done := make(chan struct{})
+	d.readLines(handler, done)
+	<-done
 
-	<-d.done
 	time.Sleep(time.Millisecond * 200)
-
 	entries := handler.Received()
 	require.Equal(t, 100000, len(entries))
 }
@@ -124,21 +141,20 @@ func TestOnelineFiles(t *testing.T) {
 		defer handler.Stop()
 
 		d := &decompressor{
-			logger:   log.NewNopLogger(),
-			running:  atomic.NewBool(false),
-			receiver: loki.NewLogsReceiver(),
-			path:     file,
-			done:     make(chan struct{}),
-			posquit:  make(chan struct{}),
-			metrics:  newMetrics(prometheus.NewRegistry()),
-			cfg:      DecompressionConfig{Format: "gz"},
+			logger:    log.NewNopLogger(),
+			running:   atomic.NewBool(false),
+			receiver:  loki.NewLogsReceiver(),
+			path:      file,
+			metrics:   newMetrics(prometheus.NewRegistry()),
+			cfg:       DecompressionConfig{Format: "gz"},
+			positions: &noopPositions{},
 		}
 
-		d.readLines(handler)
+		done := make(chan struct{})
+		d.readLines(handler, done)
+		<-done
 
-		<-d.done
 		time.Sleep(time.Millisecond * 200)
-
 		entries := handler.Received()
 		require.Equal(t, 1, len(entries))
 		require.Equal(t, string(fileContent), entries[0].Line)
@@ -150,19 +166,19 @@ func TestOnelineFiles(t *testing.T) {
 		defer handler.Stop()
 
 		d := &decompressor{
-			logger:   log.NewNopLogger(),
-			running:  atomic.NewBool(false),
-			receiver: loki.NewLogsReceiver(),
-			path:     file,
-			done:     make(chan struct{}),
-			posquit:  make(chan struct{}),
-			metrics:  newMetrics(prometheus.NewRegistry()),
-			cfg:      DecompressionConfig{Format: "bz2"},
+			logger:    log.NewNopLogger(),
+			running:   atomic.NewBool(false),
+			receiver:  loki.NewLogsReceiver(),
+			path:      file,
+			metrics:   newMetrics(prometheus.NewRegistry()),
+			cfg:       DecompressionConfig{Format: "bz2"},
+			positions: &noopPositions{},
 		}
 
-		d.readLines(handler)
+		done := make(chan struct{})
+		d.readLines(handler, done)
+		<-done
 
-		<-d.done
 		time.Sleep(time.Millisecond * 200)
 
 		entries := handler.Received()
@@ -176,19 +192,19 @@ func TestOnelineFiles(t *testing.T) {
 		defer handler.Stop()
 
 		d := &decompressor{
-			logger:   log.NewNopLogger(),
-			running:  atomic.NewBool(false),
-			receiver: loki.NewLogsReceiver(),
-			path:     file,
-			done:     make(chan struct{}),
-			posquit:  make(chan struct{}),
-			metrics:  newMetrics(prometheus.NewRegistry()),
-			cfg:      DecompressionConfig{Format: "gz"},
+			logger:    log.NewNopLogger(),
+			running:   atomic.NewBool(false),
+			receiver:  loki.NewLogsReceiver(),
+			path:      file,
+			metrics:   newMetrics(prometheus.NewRegistry()),
+			cfg:       DecompressionConfig{Format: "gz"},
+			positions: &noopPositions{},
 		}
 
-		d.readLines(handler)
+		done := make(chan struct{})
+		d.readLines(handler, done)
 
-		<-d.done
+		<-done
 		time.Sleep(time.Millisecond * 200)
 
 		entries := handler.Received()

internal/component/loki/source/file/tailer.go

@@ -49,10 +49,6 @@ type tailer struct {
 	componentStopping func() bool
 
 	tail    *tail.Tail
-	posquit chan struct{} // used by the readLine method to tell the updatePosition method to stop
-	posdone chan struct{} // used by the updatePosition method to notify when it stopped
-	done    chan struct{} // used by the readLine method to notify when it stopped
-
 	decoder *encoding.Decoder
 }
 
@@ -159,23 +155,19 @@ func (t *tailer) Run(ctx context.Context) {
 
 	t.metrics.filesActive.Add(1.)
 
-	go func() {
-		// updatePosition closes the channel t.posdone on exit
-		t.updatePosition()
-	}()
-
+	done := make(chan struct{})
 	ctx, cancel := context.WithCancel(ctx)
 	go func() {
-		// readLines closes the channels t.done and t.posquit on exit
-		t.readLines(handler)
+		// readLines closes done on exit
+		t.readLines(handler, done)
 		cancel()
 	}()
 
 	t.running.Store(true)
 	defer t.running.Store(false)
 
 	<-ctx.Done()
-	t.stop()
+	t.stop(done)
 }
 
 func (t *tailer) initRun() (loki.EntryHandler, error) {
@@ -230,9 +222,6 @@ func (t *tailer) initRun() (loki.EntryHandler, error) {
 
 	labelsMiddleware := t.labels.Merge(model.LabelSet{filenameLabel: model.LabelValue(t.path)})
 	handler := loki.AddLabelsMiddleware(labelsMiddleware).Wrap(loki.NewEntryHandler(t.receiver.Chan(), func() {}))
-	t.posquit = make(chan struct{})
-	t.posdone = make(chan struct{})
-	t.done = make(chan struct{})
 
 	return handler, nil
 }
@@ -242,15 +231,14 @@ func (t *tailer) initRun() (loki.EntryHandler, error) {
 // an error it stops the tailer and exits, the tailer will be re-opened by the
 // backoff retry method if it still exists and will start reading from the
 // last successful entry in the positions file.
-func (t *tailer) updatePosition() {
+func (t *tailer) updatePosition(posquit chan struct{}) {
 	positionSyncPeriod := t.positions.SyncPeriod()
 	positionWait := time.NewTicker(positionSyncPeriod)
 	defer func() {
 		positionWait.Stop()
 		level.Info(t.logger).Log("msg", "position timer: exited", "path", t.path)
 		// NOTE: metrics must be cleaned up after the position timer exits, as markPositionAndSize() updates metrics.
 		t.cleanupMetrics()
-		close(t.posdone)
 	}()
 
 	for {
@@ -265,7 +253,7 @@ func (t *tailer) updatePosition() {
 				}
 				return
 			}
-		case <-t.posquit:
+		case <-posquit:
 			return
 		}
 	}
@@ -277,16 +265,23 @@ func (t *tailer) updatePosition() {
 // there are unread lines in this channel and the Stop method on the tailer is
 // called, the underlying tailer will never exit if there are unread lines in
 // the t.tail.Lines channel
-func (t *tailer) readLines(handler loki.EntryHandler) {
+func (t *tailer) readLines(handler loki.EntryHandler, done chan struct{}) {
 	level.Info(t.logger).Log("msg", "tail routine: started", "path", t.path)
 
+	posquit, posdone := make(chan struct{}), make(chan struct{})
+	go func() {
+		t.updatePosition(posquit)
+		close(posdone)
+	}()
+
 	// This function runs in a goroutine, if it exits this tailer will never do any more tailing.
 	// Clean everything up.
 	defer func() {
 		level.Info(t.logger).Log("msg", "tail routine: exited", "path", t.path)
-		close(t.done)
 		// Shut down the position marker thread
-		close(t.posquit)
+		close(posquit)
+		<-posdone
+		close(done)
 	}()
 
 	entries := handler.Chan()
@@ -355,7 +350,7 @@ func (t *tailer) markPositionAndSize() error {
 	return nil
 }
 
-func (t *tailer) stop() {
+func (t *tailer) stop(done chan struct{}) {
 	// Save the current position before shutting down tailer to ensure that if the file is tailed again
 	// it start where it left off.
 	err := t.markPositionAndSize()
@@ -378,9 +373,7 @@ func (t *tailer) stop() {
 	level.Debug(t.logger).Log("msg", "waiting for readline and position marker to exit", "path", t.path)
 
 	// Wait for readLines() to consume all the remaining messages and exit when the channel is closed
-	<-t.done
-	// Wait for the position marker thread to exit
-	<-t.posdone
+	<-done
 
 	level.Info(t.logger).Log("msg", "stopped tailing file", "path", t.path)