Merge pull request #3 from hogklint/feat/retry-failed-tails

Resume logs on failure and restart

Author: hogklint

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/spf13/cobra v1.8.1
 	github.com/spf13/pflag v1.0.5
 	golang.org/x/sync v0.8.0
+	golang.org/x/time v0.7.0
 	gopkg.in/yaml.v3 v3.0.1
 	k8s.io/klog/v2 v2.130.1
 )
@@ -48,7 +49,6 @@ require (
 	go.opentelemetry.io/otel/trace v1.27.0 // indirect
 	golang.org/x/net v0.30.0 // indirect
 	golang.org/x/sys v0.26.0 // indirect
-	golang.org/x/time v0.7.0 // indirect
 	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gotest.tools/v3 v3.5.1 // indirect

stern/docker_stern.go

@@ -10,7 +10,7 @@ import (
 
 	dockerclient "github.com/docker/docker/client"
 	"golang.org/x/sync/errgroup"
-	"k8s.io/klog/v2"
+	"golang.org/x/time/rate"
 )
 
 func RunDocker(ctx context.Context, client *dockerclient.Client, config *DockerConfig) error {
@@ -19,7 +19,7 @@ func RunDocker(ctx context.Context, client *dockerclient.Client, config *DockerC
 			Timestamps:      config.Timestamps,
 			TimestampFormat: config.TimestampFormat,
 			Location:        config.Location,
-			DockerSinceTime: time.Now().Add(-config.Since),
+			DockerSinceTime: time.Now().Add(-config.Since).Format(time.RFC3339),
 			Exclude:         config.Exclude,
 			Include:         config.Include,
 			Highlight:       config.Highlight,
@@ -35,9 +35,6 @@ func RunDocker(ctx context.Context, client *dockerclient.Client, config *DockerC
 			target.Name,
 			target.ComposeProject,
 			target.Tty,
-			target.StartedAt,
-			target.FinishedAt,
-			target.SeenPreviously,
 			config.Template,
 			config.Out,
 			config.ErrOut,
@@ -50,26 +47,16 @@ func RunDocker(ctx context.Context, client *dockerclient.Client, config *DockerC
 		return tail.Start()
 	}
 
-	filter := newDockerTargetFilter(dockerTargetFilterConfig{
-		containerFilter:        config.ContainerQuery,
-		containerExcludeFilter: config.ExcludeContainerQuery,
-		composeProjectFilter:   config.ComposeProjectQuery,
-		imageFilter:            config.ImageQuery,
-	},
+	filter := newDockerTargetFilter(
+		dockerTargetFilterConfig{
+			containerFilter:        config.ContainerQuery,
+			containerExcludeFilter: config.ExcludeContainerQuery,
+			composeProjectFilter:   config.ComposeProjectQuery,
+			imageFilter:            config.ImageQuery,
+		},
 		max(config.MaxLogRequests*2, 100),
 	)
 
-	tailTarget := func(target *DockerTarget) error {
-		tail := newTail(target)
-		defer tail.Close()
-		err := tail.Start(ctx)
-		if err != nil && filter.isActive(target) {
-			fmt.Fprintf(config.ErrOut, "failed to tail %s: %v\n", target.Name, err)
-			return err
-		}
-		return nil
-	}
-
 	if !config.Follow {
 		containers, err := FilteredContainerGenerator(ctx, config, client, filter)
 		if err != nil {
@@ -81,7 +68,14 @@ func RunDocker(ctx context.Context, client *dockerclient.Client, config *DockerC
 		for target := range containers {
 			target := target
 			eg.Go(func() error {
-				return tailTarget(target)
+				tail := newTail(target)
+				defer tail.Close()
+				err := tail.Start(ctx)
+				if err != nil && filter.isActive(target) {
+					fmt.Fprintf(config.ErrOut, "failed to tail %s: %v\n", target.Name, err)
+					return err
+				}
+				return nil
 			})
 		}
 		return eg.Wait()
@@ -93,6 +87,39 @@ func RunDocker(ctx context.Context, client *dockerclient.Client, config *DockerC
 		return err
 	}
 
+	tailTarget := func(target *DockerTarget) {
+		limiter := rate.NewLimiter(rate.Every(time.Second*20), 2)
+		resumeRequest := target.ResumeRequest
+		for {
+			if err := limiter.Wait(ctx); err != nil {
+				fmt.Fprintf(config.ErrOut, "failed to retry: %v\n", err)
+				return
+			}
+			tail := newTail(target)
+			var err error
+			if resumeRequest == nil {
+				err = tail.Start(ctx)
+			} else {
+				err = tail.Resume(ctx, resumeRequest)
+			}
+			tail.Close()
+
+			if err == nil {
+				filter.setResumeRequest(target.Id, tail.GetResumeRequest())
+				return
+			}
+			if !filter.isActive(target) {
+				filter.setResumeRequest(target.Id, tail.GetResumeRequest())
+				fmt.Fprintf(config.ErrOut, "failed to tail: %v\n", err)
+				return
+			}
+			fmt.Fprintf(config.ErrOut, "failed to tail: %v, will retry\n", err)
+			if resumeReq := tail.GetResumeRequest(); resumeReq != nil {
+				resumeRequest = resumeReq
+			}
+		}
+	}
+
 	var numRequests atomic.Int64
 	for {
 		target := <-added
@@ -103,10 +130,7 @@ func RunDocker(ctx context.Context, client *dockerclient.Client, config *DockerC
 				config.MaxLogRequests)
 		}
 		go func() {
-			err := tailTarget(target)
-			if err != nil {
-				klog.V(7).ErrorS(err, "Error tailing container", "id", target.Id, "name", target.Name)
-			}
+			tailTarget(target)
 			numRequests.Add(-1)
 		}()
 	}

stern/docker_tail.go

@@ -9,10 +9,10 @@ import (
 	"io"
 	"strings"
 	"text/template"
-	"time"
 
 	"github.com/docker/docker/api/types/container"
 	dockerclient "github.com/docker/docker/client"
+	"github.com/docker/docker/errdefs"
 	"github.com/fatih/color"
 	"k8s.io/klog/v2"
 )
@@ -23,16 +23,18 @@ type DockerTail struct {
 	ContainerName  string
 	ComposeProject string
 	Tty            bool
-	StartedAt      time.Time
-	FinishedAt     string
-	SeenPreviously bool
 	composeColor   *color.Color
 	containerColor *color.Color
 	Options        *TailOptions
 	tmpl           *template.Template
 	closed         chan struct{}
-	out            io.Writer
-	errOut         io.Writer
+	last           struct {
+		timestamp string // RFC3339 timestamp (not RFC3339Nano)
+		lines     int    // the number of lines seen during this timestamp
+	}
+	resumeRequest *ResumeRequest
+	out           io.Writer
+	errOut        io.Writer
 }
 
 func NewDockerTail(
@@ -41,9 +43,6 @@ func NewDockerTail(
 	containerName string,
 	composeProject string,
 	tty bool,
-	startedAt time.Time,
-	finishedAt string,
-	seenPreviously bool,
 	tmpl *template.Template,
 	out, errOut io.Writer,
 	options *TailOptions,
@@ -56,9 +55,6 @@ func NewDockerTail(
 		ContainerName:  containerName,
 		ComposeProject: composeProject,
 		Tty:            tty,
-		StartedAt:      startedAt,
-		FinishedAt:     finishedAt,
-		SeenPreviously: seenPreviously,
 		Options:        options,
 		composeColor:   composeColor,
 		containerColor: containerColor,
@@ -86,43 +82,6 @@ func (t *DockerTail) Start(ctx context.Context) error {
 
 	t.printStarting()
 
-	err := t.consumeRequest(ctx)
-	if err != nil {
-		klog.V(7).ErrorS(err, "Error fetching logs for container", "name", t.ContainerName, "id", t.ContainerId)
-		if errors.Is(err, context.Canceled) {
-			return nil
-		}
-	}
-
-	return err
-}
-
-func (t *DockerTail) Close() {
-	t.printStopping()
-	close(t.closed)
-}
-
-// Since log streams end when containers terminate tailfin must keep track of when a container has been previously
-// tailed and use a "since"-time from last finish/start. Otherwise it will include logs from previous starts thus
-// printing logs already printed.
-func (t *DockerTail) getSinceTime() time.Time {
-	if !t.SeenPreviously ||
-		t.StartedAt.Before(t.Options.DockerSinceTime) {
-		return t.Options.DockerSinceTime
-	}
-
-	// If there's no finish time it should mean the container only started once so it's safe to use the options time.
-	finished, err := time.Parse(time.RFC3339, t.FinishedAt)
-	if err != nil {
-		return t.Options.DockerSinceTime
-	} else if finished.Before(t.StartedAt) {
-		// Sometimes early logs are missing if StartedAt is used
-		return finished
-	}
-	return t.StartedAt
-}
-
-func (t *DockerTail) consumeRequest(ctx context.Context) error {
 	logs, err := t.client.ContainerLogs(
 		ctx,
 		t.ContainerId,
@@ -131,7 +90,7 @@ func (t *DockerTail) consumeRequest(ctx context.Context) error {
 			ShowStderr: true,
 			Follow:     t.Options.Follow,
 			Timestamps: true,
-			Since:      t.getSinceTime().Format(time.RFC3339),
+			Since:      t.Options.DockerSinceTime,
 			Tail:       t.Options.DockerTailLines,
 		},
 	)
@@ -140,6 +99,30 @@ func (t *DockerTail) consumeRequest(ctx context.Context) error {
 	}
 	defer logs.Close()
 
+	err = t.consumeStream(ctx, logs)
+	if err != nil {
+		klog.V(7).ErrorS(err, "Error fetching logs for container", "name", t.ContainerName, "id", t.ContainerId)
+		if errors.Is(err, context.Canceled) || errdefs.IsConflict(err) {
+			return nil
+		}
+	}
+
+	return err
+}
+
+func (t *DockerTail) Close() {
+	t.printStopping()
+	close(t.closed)
+}
+
+func (t *DockerTail) Resume(ctx context.Context, resumeRequest *ResumeRequest) error {
+	t.resumeRequest = resumeRequest
+	t.Options.DockerSinceTime = resumeRequest.Timestamp
+	t.Options.DockerTailLines = "-1"
+	return t.Start(ctx)
+}
+
+func (t *DockerTail) consumeStream(ctx context.Context, logs io.Reader) error {
 	r := bufio.NewReader(logs)
 	for {
 		line, err := r.ReadBytes('\n')
@@ -162,6 +145,13 @@ func (t *DockerTail) consumeLine(line string) {
 		return
 	}
 
+	rfc3339 := removeSubsecond(rfc3339Nano)
+	t.rememberLastTimestamp(rfc3339)
+	if t.resumeRequest.shouldSkip(rfc3339) {
+		return
+	}
+	t.resumeRequest = nil
+
 	if t.Options.IsExclude(content) || !t.Options.IsInclude(content) {
 		return
 	}
@@ -236,6 +226,21 @@ func trimLeadingChars(line string, tty bool) string {
 		klog.V(7).InfoS("Invalid log line format received", "line", line)
 		return ""
 	}
-
 	return line[8:]
 }
+
+func (t *DockerTail) rememberLastTimestamp(timestamp string) {
+	if t.last.timestamp == timestamp {
+		t.last.lines++
+		return
+	}
+	t.last.timestamp = timestamp
+	t.last.lines = 1
+}
+
+func (t *DockerTail) GetResumeRequest() *ResumeRequest {
+	if t.last.timestamp == "" {
+		return nil
+	}
+	return &ResumeRequest{Timestamp: t.last.timestamp, LinesToSkip: t.last.lines}
+}