Skip to content

Commit ac97429

Browse files
blink-so[bot]kylecarbs
blink-so[bot]
and
kylecarbs
committed
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>
1 parent efe2bac commit ac97429

File tree

2 files changed

+81
-32
lines changed

2 files changed

+81
-32
lines changed

‎logger.go‎

Lines changed: 41 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -371,23 +371,24 @@ type tokenCache struct {
371371
replicaSets map[string][]string
372372
}
373373

374-
func (tc *tokenCache) setPodToken(name, token string) {
374+
func (tc *tokenCache) setPodToken(name, token string) []string {
375375
tc.mu.Lock()
376376
defer tc.mu.Unlock()
377377

378378
tokens, ok := tc.pods[name]
379379
if !ok {
380380
tc.pods[name] = []string{token}
381-
return
381+
return []string{token}
382382
}
383383

384384
for _, t := range tokens {
385385
if t == token {
386-
return
386+
return append([]string(nil), tokens...)
387387
}
388388
}
389389

390390
tc.pods[name] = append(tokens, token)
391+
return append([]string(nil), tc.pods[name]...)
391392
}
392393

393394
func (tc *tokenCache) deletePodToken(name string) []string {
@@ -415,23 +416,24 @@ func (tc *tokenCache) getPodTokens(name string) []string {
415416
return append([]string(nil), tokens...)
416417
}
417418

418-
func (tc *tokenCache) setReplicaSetToken(name, token string) {
419+
func (tc *tokenCache) setReplicaSetToken(name, token string) []string {
419420
tc.mu.Lock()
420421
defer tc.mu.Unlock()
421422

422423
tokens, ok := tc.replicaSets[name]
423424
if !ok {
424425
tc.replicaSets[name] = []string{token}
425-
return
426+
return []string{token}
426427
}
427428

428429
for _, t := range tokens {
429430
if t == token {
430-
return
431+
return append([]string(nil), tokens...)
431432
}
432433
}
433434

434435
tc.replicaSets[name] = append(tokens, token)
436+
return append([]string(nil), tc.replicaSets[name]...)
435437
}
436438

437439
func (tc *tokenCache) deleteReplicaSetToken(name string) []string {
@@ -459,6 +461,13 @@ func (tc *tokenCache) getReplicaSetTokens(name string) []string {
459461
return append([]string(nil), tokens...)
460462
}
461463

464+
func (tc *tokenCache) isEmpty() bool {
465+
tc.mu.RLock()
466+
defer tc.mu.RUnlock()
467+
468+
return len(tc.pods) == 0 && len(tc.replicaSets) == 0
469+
}
470+
462471
type agentLog struct {
463472
name string
464473
token string
@@ -505,7 +514,8 @@ func (lc *logCache) flush(token string) []agentsdk.Log {
505514
}
506515

507516
type agentLoggerLifecycle struct {
508-
logger agentsdk.AgentLogWriter
517+
client *agentsdk.Client
518+
sourceID uuid.UUID
509519
timer *quartz.Timer
510520
}
511521

@@ -536,38 +546,47 @@ func (lq *logQueuer) ensureLogger(ctx context.Context, token string) {
536546
return
537547
}
538548

539-
client := codersdk.New(lq.coderURL)
540-
client.SetSessionToken(token)
541-
542-
logger := client.AgentLogWriter(ctx, uuid.New())
549+
coderClient := codersdk.New(lq.coderURL)
550+
coderClient.SetSessionToken(token)
551+
agentClient := agentsdk.New(lq.coderURL)
552+
agentClient.SetSessionToken(token)
553+
554+
// Create a log source for this agent
555+
sourceID := agentsdk.ExternalLogSourceID
556+
_, err := agentClient.PostLogSource(ctx, agentsdk.PostLogSourceRequest{
557+
ID: sourceID,
558+
DisplayName: "Kubernetes",
559+
Icon: "/icon/k8s.png",
560+
})
561+
if err != nil {
562+
// Log source might already exist, which is fine
563+
lq.logger.Debug(ctx, "failed to create log source", slog.Error(err))
564+
}
543565

544566
timer := lq.clock.AfterFunc(lq.loggerTTL, func() {
545567
lq.deleteLogger(token)
546568
})
547569

548570
lq.loggers[token] = agentLoggerLifecycle{
549-
logger: logger,
550-
timer: timer,
571+
client: agentClient,
572+
sourceID: sourceID,
573+
timer: timer,
551574
}
552575

553576
go func() {
554-
defer func() {
555-
err := logger.Close()
556-
if err != nil {
557-
lq.logger.Error(ctx, "close agent logger", slog.Error(err))
558-
}
559-
}()
560-
561577
for {
562578
logs := lq.logCache.flush(token)
563579
if len(logs) == 0 {
564580
time.Sleep(time.Second)
565581
continue
566582
}
567583

568-
err := logger.Write(ctx, logs...)
584+
err := agentClient.PatchLogs(ctx, agentsdk.PatchLogs{
585+
LogSourceID: sourceID,
586+
Logs: logs,
587+
})
569588
if err != nil {
570-
lq.logger.Error(ctx, "write agent logs", slog.Error(err))
589+
lq.logger.Error(ctx, "patch agent logs", slog.Error(err))
571590
return
572591
}
573592
}

‎logger_test.go‎

Lines changed: 40 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import (
3030
"github.com/coder/coder/v2/codersdk/agentsdk"
3131
"github.com/coder/coder/v2/testutil"
3232
"github.com/coder/quartz"
33+
"google.golang.org/protobuf/types/known/timestamppb"
3334
)
3435

3536
func TestReplicaSetEvents(t *testing.T) {
@@ -41,13 +42,14 @@ func TestReplicaSetEvents(t *testing.T) {
4142
agentURL, err := url.Parse(api.server.URL)
4243
require.NoError(t, err)
4344
namespace := "test-namespace"
45+
sourceUUID := agentsdk.ExternalLogSourceID
4446
client := fake.NewSimpleClientset()
4547

4648
cMock := quartz.NewMock(t)
4749
reporter, err := newPodEventLogger(ctx, podEventLoggerOptions{
4850
client: client,
4951
coderURL: agentURL,
50-
namespace: namespace,
52+
namespaces: namespace,
5153
logger: slogtest.Make(t, nil).Leveled(slog.LevelDebug),
5254
logDebounce: 5 * time.Second,
5355
clock: cMock,
@@ -89,7 +91,7 @@ func TestReplicaSetEvents(t *testing.T) {
8991

9092
logs := testutil.RequireRecvCtx(ctx, t, api.logs)
9193
require.Len(t, logs, 1)
92-
require.Contains(t, logs[0].Output, "Queued pod from ReplicaSet")
94+
require.Contains(t, logs[0].Output, "Created replicaset")
9395

9496
event := &corev1.Event{
9597
ObjectMeta: v1.ObjectMeta{
@@ -117,7 +119,7 @@ func TestReplicaSetEvents(t *testing.T) {
117119

118120
logs = testutil.RequireRecvCtx(ctx, t, api.logs)
119121
require.Len(t, logs, 1)
120-
require.Contains(t, logs[0].Output, "Deleted ReplicaSet")
122+
require.Contains(t, logs[0].Output, "Deleted replicaset")
121123

122124
require.Eventually(t, func() bool {
123125
return reporter.tc.isEmpty()
@@ -138,13 +140,14 @@ func TestPodEvents(t *testing.T) {
138140
agentURL, err := url.Parse(api.server.URL)
139141
require.NoError(t, err)
140142
namespace := "test-namespace"
143+
sourceUUID := agentsdk.ExternalLogSourceID
141144
client := fake.NewSimpleClientset()
142145

143146
cMock := quartz.NewMock(t)
144147
reporter, err := newPodEventLogger(ctx, podEventLoggerOptions{
145148
client: client,
146149
coderURL: agentURL,
147-
namespace: namespace,
150+
namespaces: namespace,
148151
logger: slogtest.Make(t, nil).Leveled(slog.LevelDebug),
149152
logDebounce: 5 * time.Second,
150153
clock: cMock,
@@ -305,9 +308,9 @@ func Test_logQueuer(t *testing.T) {
305308
go lq.work(ctx)
306309

307310
ch <- agentLog{
308-
op: opLog,
309-
resourceName: "mypod",
310-
agentToken: "0b42fa72-7f1a-4b59-800d-69d67f56ed8b",
311+
name: "mypod",
312+
token: "0b42fa72-7f1a-4b59-800d-69d67f56ed8b",
313+
delete: false,
311314
log: agentsdk.Log{
312315
CreatedAt: time.Now(),
313316
Output: "This is a log.",
@@ -321,9 +324,9 @@ func Test_logQueuer(t *testing.T) {
321324
require.Len(t, logs, 1)
322325

323326
ch <- agentLog{
324-
op: opLog,
325-
resourceName: "mypod",
326-
agentToken: "0b42fa72-7f1a-4b59-800d-69d67f56ed8b",
327+
name: "mypod",
328+
token: "0b42fa72-7f1a-4b59-800d-69d67f56ed8b",
329+
delete: false,
327330
log: agentsdk.Log{
328331
CreatedAt: time.Now(),
329332
Output: "This is a log too.",
@@ -361,6 +364,10 @@ func newFakeAgentAPI(t *testing.T) *fakeAgentAPI {
361364
fakeAPI.PostLogSource(w, r)
362365
})
363366

367+
rtr.Patch("/api/v2/workspaceagents/me/logs", func(w http.ResponseWriter, r *http.Request) {
368+
fakeAPI.PatchLogs(w, r)
369+
})
370+
364371
rtr.Get("/api/v2/workspaceagents/me/rpc", func(w http.ResponseWriter, r *http.Request) {
365372
defer func() {
366373
fakeAPI.disconnect <- struct{}{}
@@ -460,6 +467,29 @@ func (f *fakeAgentAPI) PostLogSource(w http.ResponseWriter, r *http.Request) {
460467
}
461468
}
462469

470+
func (f *fakeAgentAPI) PatchLogs(w http.ResponseWriter, r *http.Request) {
471+
var req agentsdk.PatchLogs
472+
err := json.NewDecoder(r.Body).Decode(&req)
473+
if err != nil {
474+
fmt.Println("failed to decode patch logs:", err.Error())
475+
w.WriteHeader(http.StatusBadRequest)
476+
return
477+
}
478+
479+
// Convert agentsdk.Log to proto.Log for the channel
480+
protoLogs := make([]*proto.Log, len(req.Logs))
481+
for i, log := range req.Logs {
482+
protoLogs[i] = &proto.Log{
483+
CreatedAt: timestamppb.New(log.CreatedAt),
484+
Output: log.Output,
485+
Level: proto.Log_Level(proto.Log_Level_value[string(log.Level)]),
486+
}
487+
}
488+
489+
f.logs <- protoLogs
490+
w.WriteHeader(http.StatusOK)
491+
}
492+
463493
func TestParseNamespaces(t *testing.T) {
464494
tests := []struct {
465495
name string

0 commit comments

Comments
 (0)