Fix compilation errors and update to current agentsdk API

- Replace deprecated AgentLogWriter with agentsdk.Client and PatchLogs
- Fix struct field names in tests (namespace -> namespaces)
- Add missing isEmpty method to tokenCache
- Update setPodToken and setReplicaSetToken to return values
- Fix agentLog struct literals in tests
- Add PatchLogs endpoint to fake API for tests
- Update test expectations to match current log message format

Co-authored-by: kylecarbs <7122116+kylecarbs@users.noreply.github.com>

Author: blink-so[bot]

logger.go

@@ -371,23 +371,24 @@ type tokenCache struct {
 	replicaSets map[string][]string
 }
 
-func (tc *tokenCache) setPodToken(name, token string) {
+func (tc *tokenCache) setPodToken(name, token string) []string {
 	tc.mu.Lock()
 	defer tc.mu.Unlock()
 
 	tokens, ok := tc.pods[name]
 	if !ok {
 		tc.pods[name] = []string{token}
-		return
+		return []string{token}
 	}
 
 	for _, t := range tokens {
 		if t == token {
-			return
+			return append([]string(nil), tokens...)
 		}
 	}
 
 	tc.pods[name] = append(tokens, token)
+	return append([]string(nil), tc.pods[name]...)
 }
 
 func (tc *tokenCache) deletePodToken(name string) []string {
@@ -415,23 +416,24 @@ func (tc *tokenCache) getPodTokens(name string) []string {
 	return append([]string(nil), tokens...)
 }
 
-func (tc *tokenCache) setReplicaSetToken(name, token string) {
+func (tc *tokenCache) setReplicaSetToken(name, token string) []string {
 	tc.mu.Lock()
 	defer tc.mu.Unlock()
 
 	tokens, ok := tc.replicaSets[name]
 	if !ok {
 		tc.replicaSets[name] = []string{token}
-		return
+		return []string{token}
 	}
 
 	for _, t := range tokens {
 		if t == token {
-			return
+			return append([]string(nil), tokens...)
 		}
 	}
 
 	tc.replicaSets[name] = append(tokens, token)
+	return append([]string(nil), tc.replicaSets[name]...)
 }
 
 func (tc *tokenCache) deleteReplicaSetToken(name string) []string {
@@ -459,6 +461,13 @@ func (tc *tokenCache) getReplicaSetTokens(name string) []string {
 	return append([]string(nil), tokens...)
 }
 
+func (tc *tokenCache) isEmpty() bool {
+	tc.mu.RLock()
+	defer tc.mu.RUnlock()
+
+	return len(tc.pods) == 0 && len(tc.replicaSets) == 0
+}
+
 type agentLog struct {
 	name   string
 	token  string
@@ -505,7 +514,8 @@ func (lc *logCache) flush(token string) []agentsdk.Log {
 }
 
 type agentLoggerLifecycle struct {
-	logger agentsdk.AgentLogWriter
+	client *agentsdk.Client
+	sourceID uuid.UUID
 	timer  *quartz.Timer
 }
 
@@ -536,38 +546,47 @@ func (lq *logQueuer) ensureLogger(ctx context.Context, token string) {
 		return
 	}
 
-	client := codersdk.New(lq.coderURL)
-	client.SetSessionToken(token)
-
-	logger := client.AgentLogWriter(ctx, uuid.New())
+	coderClient := codersdk.New(lq.coderURL)
+	coderClient.SetSessionToken(token)
+	agentClient := agentsdk.New(lq.coderURL)
+	agentClient.SetSessionToken(token)
+
+	// Create a log source for this agent
+	sourceID := agentsdk.ExternalLogSourceID
+	_, err := agentClient.PostLogSource(ctx, agentsdk.PostLogSourceRequest{
+		ID:          sourceID,
+		DisplayName: "Kubernetes",
+		Icon:        "/icon/k8s.png",
+	})
+	if err != nil {
+		// Log source might already exist, which is fine
+		lq.logger.Debug(ctx, "failed to create log source", slog.Error(err))
+	}
 
 	timer := lq.clock.AfterFunc(lq.loggerTTL, func() {
 		lq.deleteLogger(token)
 	})
 
 	lq.loggers[token] = agentLoggerLifecycle{
-		logger: logger,
-		timer:  timer,
+		client:   agentClient,
+		sourceID: sourceID,
+		timer:    timer,
 	}
 
 	go func() {
-		defer func() {
-			err := logger.Close()
-			if err != nil {
-				lq.logger.Error(ctx, "close agent logger", slog.Error(err))
-			}
-		}()
-
 		for {
 			logs := lq.logCache.flush(token)
 			if len(logs) == 0 {
 				time.Sleep(time.Second)
 				continue
 			}
 
-			err := logger.Write(ctx, logs...)
+			err := agentClient.PatchLogs(ctx, agentsdk.PatchLogs{
+				LogSourceID: sourceID,
+				Logs:        logs,
+			})
 			if err != nil {
-				lq.logger.Error(ctx, "write agent logs", slog.Error(err))
+				lq.logger.Error(ctx, "patch agent logs", slog.Error(err))
 				return
 			}
 		}

logger_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/coder/coder/v2/codersdk/agentsdk"
 	"github.com/coder/coder/v2/testutil"
 	"github.com/coder/quartz"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 func TestReplicaSetEvents(t *testing.T) {
@@ -41,13 +42,14 @@ func TestReplicaSetEvents(t *testing.T) {
 	agentURL, err := url.Parse(api.server.URL)
 	require.NoError(t, err)
 	namespace := "test-namespace"
+	sourceUUID := agentsdk.ExternalLogSourceID
 	client := fake.NewSimpleClientset()
 
 	cMock := quartz.NewMock(t)
 	reporter, err := newPodEventLogger(ctx, podEventLoggerOptions{
 		client:      client,
 		coderURL:    agentURL,
-		namespace:   namespace,
+		namespaces:  namespace,
 		logger:      slogtest.Make(t, nil).Leveled(slog.LevelDebug),
 		logDebounce: 5 * time.Second,
 		clock:       cMock,
@@ -89,7 +91,7 @@ func TestReplicaSetEvents(t *testing.T) {
 
 	logs := testutil.RequireRecvCtx(ctx, t, api.logs)
 	require.Len(t, logs, 1)
-	require.Contains(t, logs[0].Output, "Queued pod from ReplicaSet")
+	require.Contains(t, logs[0].Output, "Created replicaset")
 
 	event := &corev1.Event{
 		ObjectMeta: v1.ObjectMeta{
@@ -117,7 +119,7 @@ func TestReplicaSetEvents(t *testing.T) {
 
 	logs = testutil.RequireRecvCtx(ctx, t, api.logs)
 	require.Len(t, logs, 1)
-	require.Contains(t, logs[0].Output, "Deleted ReplicaSet")
+	require.Contains(t, logs[0].Output, "Deleted replicaset")
 
 	require.Eventually(t, func() bool {
 		return reporter.tc.isEmpty()
@@ -138,13 +140,14 @@ func TestPodEvents(t *testing.T) {
 	agentURL, err := url.Parse(api.server.URL)
 	require.NoError(t, err)
 	namespace := "test-namespace"
+	sourceUUID := agentsdk.ExternalLogSourceID
 	client := fake.NewSimpleClientset()
 
 	cMock := quartz.NewMock(t)
 	reporter, err := newPodEventLogger(ctx, podEventLoggerOptions{
 		client:      client,
 		coderURL:    agentURL,
-		namespace:   namespace,
+		namespaces:  namespace,
 		logger:      slogtest.Make(t, nil).Leveled(slog.LevelDebug),
 		logDebounce: 5 * time.Second,
 		clock:       cMock,
@@ -305,9 +308,9 @@ func Test_logQueuer(t *testing.T) {
 		go lq.work(ctx)
 
 		ch <- agentLog{
-			op:           opLog,
-			resourceName: "mypod",
-			agentToken:   "0b42fa72-7f1a-4b59-800d-69d67f56ed8b",
+			name:   "mypod",
+			token:  "0b42fa72-7f1a-4b59-800d-69d67f56ed8b",
+			delete: false,
 			log: agentsdk.Log{
 				CreatedAt: time.Now(),
 				Output:    "This is a log.",
@@ -321,9 +324,9 @@ func Test_logQueuer(t *testing.T) {
 		require.Len(t, logs, 1)
 
 		ch <- agentLog{
-			op:           opLog,
-			resourceName: "mypod",
-			agentToken:   "0b42fa72-7f1a-4b59-800d-69d67f56ed8b",
+			name:   "mypod",
+			token:  "0b42fa72-7f1a-4b59-800d-69d67f56ed8b",
+			delete: false,
 			log: agentsdk.Log{
 				CreatedAt: time.Now(),
 				Output:    "This is a log too.",
@@ -361,6 +364,10 @@ func newFakeAgentAPI(t *testing.T) *fakeAgentAPI {
 			fakeAPI.PostLogSource(w, r)
 		})
 
+		rtr.Patch("/api/v2/workspaceagents/me/logs", func(w http.ResponseWriter, r *http.Request) {
+			fakeAPI.PatchLogs(w, r)
+		})
+
 		rtr.Get("/api/v2/workspaceagents/me/rpc", func(w http.ResponseWriter, r *http.Request) {
 			defer func() {
 				fakeAPI.disconnect <- struct{}{}
@@ -460,6 +467,29 @@ func (f *fakeAgentAPI) PostLogSource(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (f *fakeAgentAPI) PatchLogs(w http.ResponseWriter, r *http.Request) {
+	var req agentsdk.PatchLogs
+	err := json.NewDecoder(r.Body).Decode(&req)
+	if err != nil {
+		fmt.Println("failed to decode patch logs:", err.Error())
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	
+	// Convert agentsdk.Log to proto.Log for the channel
+	protoLogs := make([]*proto.Log, len(req.Logs))
+	for i, log := range req.Logs {
+		protoLogs[i] = &proto.Log{
+			CreatedAt: timestamppb.New(log.CreatedAt),
+			Output:    log.Output,
+			Level:     proto.Log_Level(proto.Log_Level_value[string(log.Level)]),
+		}
+	}
+	
+	f.logs <- protoLogs
+	w.WriteHeader(http.StatusOK)
+}
+
 func TestParseNamespaces(t *testing.T) {
 	tests := []struct {
 		name     string