coder
diff --git a/‎agent/logs.go
Lines changed: 50 additions & 28 deletions b/‎agent/logs.go
Lines changed: 50 additions & 28 deletions
@@ -14,9 +14,14 @@ import (
 )
 
 const (
-	flushInterval     = time.Second
-	logOutputMaxBytes = 1 << 20 // 1MiB
-	overheadPerLog    = 21      // found by testing
+	flushInterval    = time.Second
+	maxBytesPerBatch = 1 << 20 // 1MiB
+	overheadPerLog   = 21      // found by testing
+
+	// maxBytesQueued is the maximum length of logs we will queue in memory.  The number is taken
+	// from dump.sql `max_logs_length` constraint, as there is no point queuing more logs than we'll
+	// accept in the database.
+	maxBytesQueued = 1048576
 )
 
 type logQueue struct {
@@ -33,6 +38,7 @@ type logSender struct {
 	queues           map[uuid.UUID]*logQueue
 	logger           slog.Logger
 	exceededLogLimit bool
+	outputLen        int
 }
 
 type logDest interface {
@@ -47,37 +53,50 @@ func newLogSender(logger slog.Logger) *logSender {
 	}
 }
 
-func (l *logSender) enqueue(src uuid.UUID, logs ...agentsdk.Log) error {
+func (l *logSender) enqueue(src uuid.UUID, logs ...agentsdk.Log) {
 	logger := l.logger.With(slog.F("log_source_id", src))
 	if len(logs) == 0 {
 		logger.Debug(context.Background(), "enqueue called with no logs")
-		return nil
+		return
 	}
 	l.L.Lock()
 	defer l.L.Unlock()
 	if l.exceededLogLimit {
 		logger.Warn(context.Background(), "dropping enqueued logs because we have reached the server limit")
 		// don't error, as we also write to file and don't want the overall write to fail
-		return nil
+		return
 	}
 	defer l.Broadcast()
 	q, ok := l.queues[src]
 	if !ok {
 		q = &logQueue{}
 		l.queues[src] = q
 	}
-	for _, log := range logs {
+	for k, log := range logs {
+		// Here we check the queue size before adding a log because we want to queue up slightly
+		// more logs than the database would store to ensure we trigger "logs truncated" at the
+		// database layer.  Otherwise, the end user wouldn't know logs are truncated unless they
+		// examined the Coder agent logs.
+		if l.outputLen > maxBytesQueued {
+			logger.Warn(context.Background(), "log queue full; truncating new logs", slog.F("new_logs", k), slog.F("queued_logs", len(q.logs)))
+			return
+		}
 		pl, err := agentsdk.ProtoFromLog(log)
 		if err != nil {
-			return xerrors.Errorf("failed to convert log: %w", err)
+			logger.Critical(context.Background(), "failed to convert log", slog.Error(err))
+			return
+		}
+		if len(pl.Output)+overheadPerLog > maxBytesPerBatch {
+			logger.Warn(context.Background(), "dropping log line that exceeds our limit", slog.F("len", len(pl.Output)))
+			continue
 		}
 		q.logs = append(q.logs, pl)
+		l.outputLen += len(pl.Output)
 	}
 	logger.Debug(context.Background(), "enqueued agent logs", slog.F("new_logs", len(logs)), slog.F("queued_logs", len(q.logs)))
-	return nil
 }
 
-func (l *logSender) flush(src uuid.UUID) error {
+func (l *logSender) flush(src uuid.UUID) {
 	l.L.Lock()
 	defer l.L.Unlock()
 	defer l.Broadcast()
@@ -87,13 +106,21 @@ func (l *logSender) flush(src uuid.UUID) error {
 	}
 	// queue might not exist because it's already been flushed and removed from
 	// the map.
-	return nil
 }
 
 // sendLoop sends any pending logs until it hits an error or the context is canceled.  It does not
 // retry as it is expected that a higher layer retries establishing connection to the agent API and
 // calls sendLoop again.
 func (l *logSender) sendLoop(ctx context.Context, dest logDest) error {
+	l.L.Lock()
+	defer l.L.Unlock()
+	if l.exceededLogLimit {
+		l.logger.Debug(ctx, "aborting sendLoop because log limit is already exceeded")
+		// no point in keeping this loop going, if log limit is exceeded, but don't return an
+		// error because we're already handled it
+		return nil
+	}
+
 	ctxDone := false
 	defer l.logger.Debug(ctx, "sendLoop exiting")
 
@@ -119,42 +146,36 @@ func (l *logSender) sendLoop(ctx context.Context, dest logDest) error {
 		}
 	}()
 
-	l.L.Lock()
-	defer l.L.Unlock()
 	for {
-		for !ctxDone && !l.exceededLogLimit && !l.hasPendingWorkLocked() {
+		for !ctxDone && !l.hasPendingWorkLocked() {
 			l.Wait()
 		}
 		if ctxDone {
 			return nil
 		}
-		if l.exceededLogLimit {
-			l.logger.Debug(ctx, "aborting sendLoop because log limit is already exceeded")
-			// no point in keeping this loop going, if log limit is exceeded, but don't return an
-			// error because we're already handled it
-			return nil
-		}
+
 		src, q := l.getPendingWorkLocked()
 		logger := l.logger.With(slog.F("log_source_id", src))
 		q.flushRequested = false // clear flag since we're now flushing
 		req := &proto.BatchCreateLogsRequest{
 			LogSourceId: src[:],
 		}
-		o := 0
+
+		// outputToSend keeps track of the size of the protobuf message we send, while
+		// outputToRemove keeps track of the size of the output we'll remove from the queues on
+		// success.  They are different because outputToSend also counts protocol message overheads.
+		outputToSend := 0
+		outputToRemove := 0
 		n := 0
 		for n < len(q.logs) {
 			log := q.logs[n]
-			if len(log.Output) > logOutputMaxBytes {
-				logger.Warn(ctx, "dropping log line that exceeds our limit")
-				n++
-				continue
-			}
-			o += len(log.Output) + overheadPerLog
-			if o > logOutputMaxBytes {
+			outputToSend += len(log.Output) + overheadPerLog
+			if outputToSend > maxBytesPerBatch {
 				break
 			}
 			req.Logs = append(req.Logs, log)
 			n++
+			outputToRemove += len(log.Output)
 		}
 
 		l.L.Unlock()
@@ -181,6 +202,7 @@ func (l *logSender) sendLoop(ctx context.Context, dest logDest) error {
 			q.logs[i] = nil
 		}
 		q.logs = q.logs[n:]
+		l.outputLen -= outputToRemove
 		if len(q.logs) == 0 {
 			// no empty queues
 			delete(l.queues, src)