implement maxRetries for log processing

Author: kacpersaw

helm/templates/service.yaml

@@ -120,6 +120,10 @@ spec:
             - name: SSL_CERT_DIR
               value: {{ .Values.image.sslCertDir }}
             {{- end }}
+            {{- if .Values.maxRetries }}
+            - name: CODER_MAX_RETRIES
+              value: {{ .Values.maxRetries }}
+            {{- end }}
           {{- with .Values.securityContext }}
           securityContext:
             {{- toYaml . | nindent 12 }}

helm/values.yaml

@@ -5,6 +5,9 @@ url: ""
 # If unspecified or empty it will watch all namespaces.
 namespaces: []
 
+# maxRetries -- Maximum retry attempts for failed log sends (logs are discarded after this limit)
+maxRetries: 10
+
 # volumes -- A list of extra volumes to add to the coder-logstream pod.
 volumes:
   #   emptyDir: {}

logger.go

@@ -34,6 +34,7 @@ type podEventLoggerOptions struct {
 
 	logger      slog.Logger
 	logDebounce time.Duration
+	maxRetries  int
 
 	// The following fields are optional!
 	namespaces    []string
@@ -52,6 +53,10 @@ func newPodEventLogger(ctx context.Context, opts podEventLoggerOptions) (*podEve
 		opts.clock = quartz.NewReal()
 	}
 
+	if opts.maxRetries == 0 {
+		opts.maxRetries = 10
+	}
+
 	logCh := make(chan agentLog, 512)
 	ctx, cancelFunc := context.WithCancel(ctx)
 	reporter := &podEventLogger{
@@ -75,6 +80,7 @@ func newPodEventLogger(ctx context.Context, opts podEventLoggerOptions) (*podEve
 			logCache: logCache{
 				logs: map[string][]agentsdk.Log{},
 			},
+			maxRetries: opts.maxRetries,
 		},
 	}
 
@@ -408,7 +414,8 @@ type logQueuer struct {
 	loggers   map[string]agentLoggerLifecycle
 	logCache  logCache
 
-	retries map[string]*retryState
+	retries    map[string]*retryState
+	maxRetries int
 }
 
 func (l *logQueuer) work(ctx context.Context) {
@@ -588,8 +595,9 @@ func (l *agentLoggerLifecycle) resetCloseTimer(ttl time.Duration) {
 
 // retryState tracks exponential backoff for an agent token.
 type retryState struct {
-	delay time.Duration
-	timer *quartz.Timer
+	delay      time.Duration
+	timer      *quartz.Timer
+	retryCount int
 }
 
 func (l *logQueuer) scheduleRetry(ctx context.Context, token string) {
@@ -603,6 +611,18 @@ func (l *logQueuer) scheduleRetry(ctx context.Context, token string) {
 		l.retries[token] = rs
 	}
 
+	rs.retryCount++
+
+	// If we've reached the max retries, clear the retry state and delete the log cache.
+	if rs.retryCount >= l.maxRetries {
+		l.logger.Error(ctx, "max retries exceeded",
+			slog.F("retryCount", rs.retryCount),
+			slog.F("maxRetries", l.maxRetries))
+		l.clearRetry(token)
+		l.logCache.delete(token)
+		return
+	}
+
 	if rs.timer != nil {
 		return
 	}
@@ -613,7 +633,9 @@ func (l *logQueuer) scheduleRetry(ctx context.Context, token string) {
 		rs.delay = 30 * time.Second
 	}
 
-	l.logger.Info(ctx, "scheduling retry", slog.F("delay", rs.delay.String()))
+	l.logger.Info(ctx, "scheduling retry",
+		slog.F("delay", rs.delay.String()),
+		slog.F("retryCount", rs.retryCount))
 
 	rs.timer = l.clock.AfterFunc(rs.delay, func() {
 		l.mu.Lock()

logger_test.go

@@ -594,24 +594,30 @@ func Test_logQueuer(t *testing.T) {
 			logCache: logCache{
 				logs: map[string][]agentsdk.Log{},
 			},
+			maxRetries: 10,
 		}
 
 		ctx := context.Background()
 		token := "test-token"
 
 		// Set up a retry state with a large delay
 		lq.retries = make(map[string]*retryState)
-		lq.retries[token] = &retryState{delay: 20 * time.Second}
+		lq.retries[token] = &retryState{
+			delay:      20 * time.Second,
+			retryCount: 0,
+		}
 
 		// Schedule a retry - should cap at 30 seconds
 		lq.scheduleRetry(ctx, token)
 
 		rs := lq.retries[token]
+		require.NotNil(t, rs)
 		require.Equal(t, 30*time.Second, rs.delay)
 
 		// Schedule another retry - should stay at 30 seconds
 		lq.scheduleRetry(ctx, token)
 		rs = lq.retries[token]
+		require.NotNil(t, rs)
 		require.Equal(t, 30*time.Second, rs.delay)
 	})
 
@@ -627,6 +633,7 @@ func Test_logQueuer(t *testing.T) {
 			logCache: logCache{
 				logs: map[string][]agentsdk.Log{},
 			},
+			maxRetries: 2,
 		}
 
 		ctx := context.Background()
@@ -640,6 +647,64 @@ func Test_logQueuer(t *testing.T) {
 		lq.clearRetry(token)
 		require.Nil(t, lq.retries[token])
 	})
+
+	t.Run("MaxRetries", func(t *testing.T) {
+		t.Parallel()
+
+		// Create a failing API that will reject connections
+		failingAPI := newFailingAgentAPI(t)
+		agentURL, err := url.Parse(failingAPI.server.URL)
+		require.NoError(t, err)
+		clock := quartz.NewMock(t)
+		ttl := time.Second
+
+		ch := make(chan agentLog, 10)
+		logger := slogtest.Make(t, &slogtest.Options{
+			IgnoreErrors: true,
+		})
+		lq := &logQueuer{
+			logger:    logger,
+			clock:     clock,
+			q:         ch,
+			coderURL:  agentURL,
+			loggerTTL: ttl,
+			loggers:   map[string]agentLoggerLifecycle{},
+			logCache: logCache{
+				logs: map[string][]agentsdk.Log{},
+			},
+			maxRetries: 2,
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		go lq.work(ctx)
+
+		token := "max-retry-token"
+		ch <- agentLog{
+			op:           opLog,
+			resourceName: "hello",
+			agentToken:   token,
+			log: agentsdk.Log{
+				CreatedAt: time.Now(),
+				Output:    "This is a log.",
+				Level:     codersdk.LogLevelInfo,
+			},
+		}
+
+		// Wait for retry state to be cleared after exceeding maxRetries
+		require.Eventually(t, func() bool {
+			lq.mu.Lock()
+			defer lq.mu.Unlock()
+			rs := lq.retries[token]
+			return rs == nil
+		}, testutil.WaitShort, testutil.IntervalFast)
+
+		// Verify cache is also cleared
+		lq.mu.Lock()
+		cachedLogs := lq.logCache.get(token)
+		lq.mu.Unlock()
+		require.Nil(t, cachedLogs)
+	})
 }
 
 func Test_logCache(t *testing.T) {

main.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/url"
 	"os"
+	"strconv"
 	"strings"
 
 	"cdr.dev/slog"
@@ -30,6 +31,7 @@ func root() *cobra.Command {
 		kubeConfig    string
 		namespacesStr string
 		labelSelector string
+		maxRetriesStr string
 	)
 	cmd := &cobra.Command{
 		Use:   "coder-logstream-kube",
@@ -72,13 +74,19 @@ func root() *cobra.Command {
 				}
 			}
 
+			maxRetries, err := strconv.Atoi(maxRetriesStr)
+			if err != nil {
+				return fmt.Errorf("parse max retries: %w", err)
+			}
+
 			reporter, err := newPodEventLogger(cmd.Context(), podEventLoggerOptions{
 				coderURL:      parsedURL,
 				client:        client,
 				namespaces:    namespaces,
 				fieldSelector: fieldSelector,
 				labelSelector: labelSelector,
 				logger:        slog.Make(sloghuman.Sink(cmd.ErrOrStderr())).Leveled(slog.LevelDebug),
+				maxRetries:    maxRetries,
 			})
 			if err != nil {
 				return fmt.Errorf("create pod event reporter: %w", err)
@@ -97,6 +105,7 @@ func root() *cobra.Command {
 	cmd.Flags().StringVarP(&namespacesStr, "namespaces", "n", os.Getenv("CODER_NAMESPACES"), "List of namespaces to use when listing pods")
 	cmd.Flags().StringVarP(&fieldSelector, "field-selector", "f", "", "Field selector to use when listing pods")
 	cmd.Flags().StringVarP(&labelSelector, "label-selector", "l", "", "Label selector to use when listing pods")
+	cmd.Flags().StringVarP(&maxRetriesStr, "max-retries", "m", os.Getenv("CODER_MAX_RETRIES"), "Maximum retry attempts for failed log sends (logs are discarded after this limit)")
 
 	return cmd
 }