Merge pull request #3200 from xyz-li/main

fix: output log even log is rotated

go.mod

@@ -37,6 +37,7 @@ require (
 	github.com/fahedouch/go-logrotate v0.2.1
 	github.com/fatih/color v1.17.0
 	github.com/fluent/fluent-logger-golang v1.9.0
+	github.com/fsnotify/fsnotify v1.7.0
 	github.com/ipfs/go-cid v0.4.1
 	github.com/klauspost/compress v1.17.9
 	github.com/mattn/go-isatty v0.0.20

go.sum

@@ -109,6 +109,8 @@ github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/fluent/fluent-logger-golang v1.9.0 h1:zUdY44CHX2oIUc7VTNZc+4m+ORuO/mldQDA7czhWXEg=
 github.com/fluent/fluent-logger-golang v1.9.0/go.mod h1:2/HCT/jTy78yGyeNGQLGQsjF3zzzAuy6Xlk6FCMV5eU=
+github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
+github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
 github.com/go-jose/go-jose/v4 v4.0.3 h1:o8aphO8Hv6RPmH+GfzVuyf7YXSBibp+8YyHdOoDESGo=
 github.com/go-jose/go-jose/v4 v4.0.3/go.mod h1:NKb5HO1EZccyMpiZNbdUw/14tiXNyUJh188dfnMCAfc=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=

pkg/logging/cri_logger.go

@@ -25,6 +25,7 @@ package logging
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -35,6 +36,7 @@ import (
 
 	"github.com/containerd/log"
 	"github.com/containerd/nerdctl/v2/pkg/logging/tail"
+	"github.com/fsnotify/fsnotify"
 )
 
 // LogStreamType is the type of the stream in CRI container log.
@@ -45,6 +47,9 @@ const (
 	Stdout LogStreamType = "stdout"
 	// Stderr is the stream type for stderr.
 	Stderr LogStreamType = "stderr"
+
+	// logForceCheckPeriod is the period to check for a new read
+	logForceCheckPeriod = 1 * time.Second
 )
 
 // LogTag is the tag of a log line in CRI container log.
@@ -89,7 +94,9 @@ func ReadLogs(opts *LogViewOptions, stdout, stderr io.Writer, stopChannel chan o
 	if err != nil {
 		return fmt.Errorf("failed to open log file %q: %v", logPath, err)
 	}
-	defer f.Close()
+	defer func() {
+		f.Close()
+	}()
 
 	// Search start point based on tail line.
 	start, err := tail.FindTailLineStartIndex(f, opts.Tail)
@@ -101,6 +108,8 @@ func ReadLogs(opts *LogViewOptions, stdout, stderr io.Writer, stopChannel chan o
 		return fmt.Errorf("failed to seek in log file %q: %v", logPath, err)
 	}
 
+	var watcher *fsnotify.Watcher
+
 	limitedMode := (opts.Tail > 0) && (!opts.Follow)
 	limitedNum := opts.Tail
 	// Start parsing the logs.
@@ -110,6 +119,9 @@ func ReadLogs(opts *LogViewOptions, stdout, stderr io.Writer, stopChannel chan o
 	isNewLine := true
 	writer := newLogWriter(stdout, stderr, opts)
 	msg := &logMessage{}
+	baseName := filepath.Base(logPath)
+	dir := filepath.Dir(logPath)
+
 	for {
 		select {
 		case <-stopChannel:
@@ -126,13 +138,48 @@ func ReadLogs(opts *LogViewOptions, stdout, stderr io.Writer, stopChannel chan o
 					return fmt.Errorf("failed to read log file %q: %v", logPath, err)
 				}
 				if opts.Follow {
-
 					// Reset seek so that if this is an incomplete line,
 					// it will be read again.
 					if _, err := f.Seek(-int64(len(l)), io.SeekCurrent); err != nil {
 						return fmt.Errorf("failed to reset seek in log file %q: %v", logPath, err)
 					}
 
+					if watcher == nil {
+						// Initialize the watcher if it has not been initialized yet.
+						if watcher, err = NewLogFileWatcher(dir); err != nil {
+							return err
+						}
+						defer watcher.Close()
+						// If we just created the watcher, try again to read as we might have missed
+						// the event.
+						continue
+					}
+
+					var recreated bool
+					// Wait until the next log change.
+					recreated, err = startTail(context.Background(), baseName, watcher)
+					if err != nil {
+						return err
+					}
+					if recreated {
+						newF, err := openFileShareDelete(logPath)
+						if err != nil {
+							if errors.Is(err, os.ErrNotExist) {
+								//If the user application outputs logs too quickly,
+								//There is a slight possibility that nerdctl has just rotated the log file,
+								//try opening it once more.
+								time.Sleep(10 * time.Millisecond)
+							}
+							newF, err = openFileShareDelete(logPath)
+							if err != nil {
+								return fmt.Errorf("failed to open cri logfile %q: %w", logPath, err)
+							}
+						}
+						f.Close()
+						f = newF
+						r = bufio.NewReader(f)
+					}
+
 					// If the container exited consume data until the next EOF
 					continue
 				}

pkg/logging/cri_logger_test.go

@@ -28,6 +28,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"path/filepath"
 	"reflect"
 	"testing"
 	"time"
@@ -230,3 +231,85 @@ func TestReadLogsLimitsWithTimestamps(t *testing.T) {
 		t.Errorf("should have two lines, lineCount= %d", lineCount)
 	}
 }
+
+func TestReadRotatedLog(t *testing.T) {
+	tmpDir := t.TempDir()
+	file, err := os.CreateTemp(tmpDir, "logfile")
+	if err != nil {
+		t.Errorf("unable to create temp file, error: %s", err.Error())
+	}
+	stdoutBuf := &bytes.Buffer{}
+	stderrBuf := &bytes.Buffer{}
+	containerStoped := make(chan os.Signal)
+	// Start to follow the container's log.
+	fileName := file.Name()
+	go func() {
+		lvOpts := &LogViewOptions{
+			Follow:  true,
+			LogPath: fileName,
+		}
+		_ = ReadLogs(lvOpts, stdoutBuf, stderrBuf, containerStoped)
+	}()
+
+	// log in stdout
+	expectedStdout := "line0line2line4line6line8"
+	// log in stderr
+	expectedStderr := "line1line3line5line7line9"
+
+	dir := filepath.Dir(file.Name())
+	baseName := filepath.Base(file.Name())
+
+	// Write 10 lines to log file.
+	// Let ReadLogs start.
+	time.Sleep(50 * time.Millisecond)
+
+	for line := 0; line < 10; line++ {
+		// Write the first three lines to log file
+		now := time.Now().Format(time.RFC3339Nano)
+		if line%2 == 0 {
+			file.WriteString(fmt.Sprintf(
+				"%s stdout P line%d\n", now, line))
+		} else {
+			file.WriteString(fmt.Sprintf(
+				"%s stderr P line%d\n", now, line))
+		}
+
+		time.Sleep(1 * time.Millisecond)
+
+		if line == 5 {
+			file.Close()
+			// Pretend to rotate the log.
+			rotatedName := fmt.Sprintf("%s.%s", baseName, time.Now().Format("220060102-150405"))
+			rotatedName = filepath.Join(dir, rotatedName)
+			if err := os.Rename(filepath.Join(dir, baseName), rotatedName); err != nil {
+				t.Errorf("failed to rotate log %q to %q, error: %s", file.Name(), rotatedName, err.Error())
+				return
+			}
+
+			time.Sleep(20 * time.Millisecond)
+			newF := filepath.Join(dir, baseName)
+			if file, err = os.Create(newF); err != nil {
+				t.Errorf("unable to create new log file, error: %s", err.Error())
+				return
+			}
+		}
+	}
+
+	// Finished writing into the file, close it, so we can delete it later.
+	err = file.Close()
+	if err != nil {
+		t.Errorf("could not close file, error: %s", err.Error())
+	}
+
+	time.Sleep(2 * time.Second)
+	// Make the function ReadLogs end.
+	close(containerStoped)
+
+	if expectedStdout != stdoutBuf.String() {
+		t.Errorf("expected: %s, acoutal: %s", expectedStdout, stdoutBuf.String())
+	}
+
+	if expectedStderr != stderrBuf.String() {
+		t.Errorf("expected: %s, acoutal: %s", expectedStderr, stderrBuf.String())
+	}
+}