daemon/logger: replace flaky TestFollowLogsHandleDecodeErr

Signed-off-by: Kazuyoshi Kato <katokazu@amazon.com>

daemon/logger/loggerutils/follow_test.go

@@ -0,0 +1,37 @@
+package loggerutils // import "github.com/docker/docker/daemon/logger/loggerutils"
+
+import (
+	"io"
+	"os"
+	"testing"
+
+	"gotest.tools/v3/assert"
+)
+
+func TestHandleDecoderErr(t *testing.T) {
+	f, err := os.CreateTemp("", t.Name())
+	assert.NilError(t, err)
+	defer os.Remove(f.Name())
+
+	_, err = f.Write([]byte("hello"))
+	assert.NilError(t, err)
+
+	pos, err := f.Seek(0, io.SeekCurrent)
+	assert.NilError(t, err)
+	assert.Assert(t, pos != 0)
+
+	dec := &testDecoder{}
+
+	// Simulate "turncate" case, where the file was bigger before.
+	fl := &follow{file: f, dec: dec, oldSize: 100}
+	err = fl.handleDecodeErr(io.EOF)
+	assert.NilError(t, err)
+
+	// handleDecodeErr seeks to zero.
+	pos, err = f.Seek(0, io.SeekCurrent)
+	assert.NilError(t, err)
+	assert.Equal(t, int64(0), pos)
+
+	// Reset is called.
+	assert.Equal(t, 1, dec.resetCount)
+}

daemon/logger/loggerutils/logfile_test.go

@@ -4,13 +4,11 @@ import (
 	"bufio"
 	"bytes"
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"testing"
 	"text/tabwriter"
@@ -24,8 +22,9 @@ import (
 )
 
 type testDecoder struct {
-	rdr     io.Reader
-	scanner *bufio.Scanner
+	rdr        io.Reader
+	scanner    *bufio.Scanner
+	resetCount int
 }
 
 func (d *testDecoder) Decode() (*logger.Message, error) {
@@ -42,6 +41,7 @@ func (d *testDecoder) Decode() (*logger.Message, error) {
 func (d *testDecoder) Reset(rdr io.Reader) {
 	d.rdr = rdr
 	d.scanner = bufio.NewScanner(rdr)
+	d.resetCount++
 }
 
 func (d *testDecoder) Close() {
@@ -246,77 +246,6 @@ func TestFollowLogsProducerGone(t *testing.T) {
 	}
 }
 
-type lineDecoder struct {
-	r          *bufio.Reader
-	resetCount int
-}
-
-func (d *lineDecoder) Decode() (*logger.Message, error) {
-	line, err := d.r.ReadString('\n')
-	if err != nil {
-		return nil, err
-	}
-	m := logger.NewMessage()
-	m.Line = []byte(line)
-	return m, nil
-}
-
-func (d *lineDecoder) Reset(r io.Reader) {
-	d.r = bufio.NewReader(r)
-	d.resetCount++
-}
-
-func (d *lineDecoder) Close() {
-}
-
-func TestFollowLogsHandleDecodeErr(t *testing.T) {
-	lw := logger.NewLogWatcher()
-	defer lw.ConsumerGone()
-
-	fw, err := os.CreateTemp("", t.Name())
-	assert.NilError(t, err)
-	defer os.Remove(fw.Name())
-
-	fr, err := os.Open(fw.Name())
-	assert.NilError(t, err)
-
-	dec := &lineDecoder{}
-	dec.Reset(fr)
-
-	var since, until time.Time
-	rotate := make(chan interface{})
-	evict := make(chan interface{})
-
-	var wg sync.WaitGroup
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		followLogs(fr, lw, rotate, evict, dec, since, until)
-	}()
-
-	sendReceive := func(f io.Writer, message string) {
-		_, err = f.Write([]byte(message))
-		assert.NilError(t, err)
-		m := <-lw.Msg
-		assert.Equal(t, message, string(m.Line))
-	}
-
-	sendReceive(fw, "log1\n")
-	sendReceive(fw, "log2\n")
-
-	ft, err := os.OpenFile(fw.Name(), os.O_WRONLY|os.O_TRUNC, 0600)
-	assert.NilError(t, err)
-
-	sendReceive(ft, "log3\n")
-
-	evict <- errors.New("stop followLogs")
-	wg.Wait()
-
-	// followLogs calls Reset() in the beginning,
-	// each 3 writes result Reset(), then handleDecodeErr() calles Reset().
-	assert.Equal(t, 5, dec.resetCount)
-}
-
 func TestCheckCapacityAndRotate(t *testing.T) {
 	dir, err := os.MkdirTemp("", t.Name())
 	assert.NilError(t, err)