feat: use agentv2 API for build logs (#78)

.github/workflows/ci.yaml

@@ -37,13 +37,13 @@ jobs:
       # Install Go!
       - uses: actions/setup-go@v3
         with:
-          go-version: "~1.20"
+          go-version: "~1.22"
 
       # Check for Go linting errors!
       - name: Lint Go
-        uses: golangci/golangci-lint-action@v3.3.1
+        uses: golangci/golangci-lint-action@v6.0.1
         with:
-          version: v1.51.0
+          version: v1.59.1
           args: "--out-${NO_FUTURE}format colored-line-number"
 
       - name: Lint shell scripts
@@ -98,7 +98,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: "~1.20"
+          go-version: "~1.22"
 
       # Sadly the new "set output" syntax (of writing env vars to
       # $GITHUB_OUTPUT) does not work on both powershell and bash so we use the
@@ -136,7 +136,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: "1.20.5"
+          go-version: "~1.22"
 
       # Sadly the new "set output" syntax (of writing env vars to
       # $GITHUB_OUTPUT) does not work on both powershell and bash so we use the
@@ -170,7 +170,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-go@v3
         with:
-          go-version: "1.20.5"
+          go-version: "~1.22"
 
       - name: Go Cache Paths
         id: go-cache-paths
@@ -223,7 +223,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v3
         with:
-          go-version: "~1.20"
+          go-version: "~1.22"
 
       - name: Go Cache Paths
         id: go-cache-paths
@@ -254,7 +254,7 @@ jobs:
 
       - uses: actions/setup-go@v3
         with:
-          go-version: "1.20.5"
+          go-version: "~1.22"
 
       - name: build image
         run: make -j build/image/envbox

.golangci.yaml

@@ -54,7 +54,6 @@ linters-settings:
       # - importShadow
       - indexAlloc
       - initClause
-      - ioutilDeprecated
       - mapKey
       - methodExprCall
       # - nestingReduce

background/process.go

@@ -233,6 +233,7 @@ func scanIntoLog(ctx context.Context, log slog.Logger, scanner *bufio.Scanner, b
 			logFn = log.Info
 		)
 
+		//nolint:gocritic
 		if strings.Contains(line, "level=debug") {
 			logFn = log.Debug
 		} else if strings.Contains(line, "level=info") {

buildlog/coder.go

@@ -3,10 +3,19 @@ package buildlog
 import (
 	"context"
 	"fmt"
+	"io"
+	"net/url"
 	"time"
 
+	"github.com/google/uuid"
+	"golang.org/x/xerrors"
+	"storj.io/drpc"
+
 	"cdr.dev/slog"
-	"github.com/coder/coder/codersdk/agentsdk"
+	"github.com/coder/coder/v2/agent/proto"
+	"github.com/coder/coder/v2/codersdk"
+	"github.com/coder/coder/v2/codersdk/agentsdk"
+	"github.com/coder/retry"
 )
 
 const (
@@ -22,30 +31,94 @@ type StartupLog struct {
 }
 
 type CoderClient interface {
-	PatchStartupLogs(ctx context.Context, req agentsdk.PatchStartupLogs) error
+	Send(level codersdk.LogLevel, log string) error
+	io.Closer
 }
 
-type CoderLogger struct {
-	ctx     context.Context
-	cancel  context.CancelFunc
-	client  CoderClient
-	logger  slog.Logger
-	logChan chan string
-	err     error
+type coderClient struct {
+	ctx    context.Context
+	cancel context.CancelFunc
+	source uuid.UUID
+	ls     *agentsdk.LogSender
+	sl     agentsdk.ScriptLogger
+	log    slog.Logger
 }
 
-func OpenCoderLogger(ctx context.Context, client CoderClient, log slog.Logger) Logger {
-	ctx, cancel := context.WithCancel(ctx)
+func (c *coderClient) Send(level codersdk.LogLevel, log string) error {
+	err := c.sl.Send(c.ctx, agentsdk.Log{
+		CreatedAt: time.Now(),
+		Output:    log,
+		Level:     level,
+	})
+	if err != nil {
+		return xerrors.Errorf("send build log: %w", err)
+	}
+	return nil
+}
 
-	coder := &CoderLogger{
-		ctx:     ctx,
-		cancel:  cancel,
-		client:  client,
-		logger:  log,
-		logChan: make(chan string),
+func (c *coderClient) Close() error {
+	defer c.cancel()
+	c.ls.Flush(c.source)
+	err := c.ls.WaitUntilEmpty(c.ctx)
+	if err != nil {
+		return xerrors.Errorf("wait until empty: %w", err)
 	}
+	return nil
+}
 
-	go coder.processLogs()
+func OpenCoderClient(ctx context.Context, accessURL *url.URL, logger slog.Logger, token string) (CoderClient, error) {
+	client := agentsdk.New(accessURL)
+	client.SetSessionToken(token)
+
+	cctx, cancel := context.WithCancel(ctx)
+	uid := uuid.New()
+	ls := agentsdk.NewLogSender(logger)
+	sl := ls.GetScriptLogger(uid)
+
+	var conn drpc.Conn
+	var err error
+	for r := retry.New(10*time.Millisecond, time.Second); r.Wait(ctx); {
+		conn, err = client.ConnectRPC(ctx)
+		if err != nil {
+			logger.Error(ctx, "connect err", slog.Error(err))
+			continue
+		}
+		break
+	}
+	if conn == nil {
+		cancel()
+		return nil, xerrors.Errorf("connect rpc: %w", err)
+	}
+
+	arpc := proto.NewDRPCAgentClient(conn)
+	go func() {
+		err := ls.SendLoop(ctx, arpc)
+		if err != nil {
+			logger.Error(ctx, "send loop", slog.Error(err))
+		}
+	}()
+
+	return &coderClient{
+		ctx:    cctx,
+		cancel: cancel,
+		source: uid,
+		ls:     ls,
+		sl:     sl,
+		log:    logger,
+	}, nil
+}
+
+type CoderLogger struct {
+	ctx    context.Context
+	client CoderClient
+	logger slog.Logger
+}
+
+func OpenCoderLogger(client CoderClient, log slog.Logger) Logger {
+	coder := &CoderLogger{
+		client: client,
+		logger: log,
+	}
 
 	return coder
 }
@@ -55,84 +128,31 @@ func (c *CoderLogger) Infof(format string, a ...any) {
 }
 
 func (c *CoderLogger) Info(output string) {
-	c.log(output)
+	c.log(codersdk.LogLevelInfo, output)
 }
 
 func (c *CoderLogger) Errorf(format string, a ...any) {
 	c.Error(fmt.Sprintf(format, a...))
 }
 
 func (c *CoderLogger) Error(output string) {
-	c.log("ERROR: " + output)
+	c.log(codersdk.LogLevelError, output)
 }
 
-func (c *CoderLogger) log(output string) {
-	if c.err != nil {
-		return
+func (c *CoderLogger) log(level codersdk.LogLevel, output string) {
+	if err := c.client.Send(level, output); err != nil {
+		c.logger.Error(c.ctx, "send build log",
+			slog.F("log", output),
+			slog.Error(err),
+		)
 	}
-	c.logChan <- output
 }
 
 func (c *CoderLogger) Write(p []byte) (int, error) {
 	c.Info(string(p))
 	return len(p), nil
 }
 
-func (c *CoderLogger) Close() {
-	c.cancel()
-}
-
-func (c *CoderLogger) processLogs() {
-	for {
-		var (
-			line string
-			logs = make([]agentsdk.StartupLog, 0, CoderLoggerMaxLogs)
-		)
-
-		select {
-		case line = <-c.logChan:
-			lines := cutString(line, MaxCoderLogSize)
-
-			for _, output := range lines {
-				logs = append(logs, agentsdk.StartupLog{
-					CreatedAt: time.Now(),
-					Output:    output,
-				})
-			}
-
-		case <-c.ctx.Done():
-			close(c.logChan)
-			return
-		}
-
-		// Send the logs in a goroutine so that we can avoid blocking
-		// too long on the channel.
-		cpLogs := logs
-		go func(startupLogs []agentsdk.StartupLog) {
-			err := c.client.PatchStartupLogs(c.ctx, agentsdk.PatchStartupLogs{
-				Logs: startupLogs,
-			})
-			if err != nil {
-				c.logger.Error(c.ctx, "send startup logs", slog.Error(err))
-			}
-		}(cpLogs)
-	}
-}
-
-// cutString cuts a string up into smaller strings that have a len no greater
-// than the provided max size.
-// If the string is less than the max size the return slice has one
-// element with a value of the provided string.
-func cutString(s string, maxSize int) []string {
-	if len(s) <= maxSize {
-		return []string{s}
-	}
-
-	toks := []string{}
-	for len(s) > maxSize {
-		toks = append(toks, s[:maxSize])
-		s = s[maxSize:]
-	}
-
-	return append(toks, s)
+func (c *CoderLogger) Close() error {
+	return c.client.Close()
 }