From 655286fba2d49d4196595b0a1e5d3fa55c2b4285 Mon Sep 17 00:00:00 2001 From: Haven Xia Date: Fri, 17 Jul 2026 13:40:18 -0700 Subject: [PATCH 1/2] Make kubectl-ate log and worker output filtered by atespace The current implementation still only use actor name to filter, which could leads to interleaved information from actors with same name in different atespaces. --- cmd/kubectl-ate/internal/cmd/logs_actors.go | 14 ++-- .../internal/cmd/logs_actors_test.go | 67 ++++++++++++++----- cmd/kubectl-ate/internal/printer/printer.go | 4 +- .../internal/printer/printer_test.go | 5 +- 4 files changed, 65 insertions(+), 25 deletions(-) diff --git a/cmd/kubectl-ate/internal/cmd/logs_actors.go b/cmd/kubectl-ate/internal/cmd/logs_actors.go index 5c78f422d..5badf1ba1 100644 --- a/cmd/kubectl-ate/internal/cmd/logs_actors.go +++ b/cmd/kubectl-ate/internal/cmd/logs_actors.go @@ -136,7 +136,7 @@ func (r *LogsActorRunner) runOneShot(ctx context.Context, actorName string) erro scanner.Buffer(buf, 1024*1024) // Support up to 1MB lines for scanner.Scan() { line := scanner.Text() - filterAndDisplayLogLine(line, actorName, r.stdout) + filterAndDisplayLogLine(line, r.atespace, actorName, r.stdout) } if err := scanner.Err(); err != nil { return fmt.Errorf("error reading log stream: %w", err) @@ -213,7 +213,7 @@ func (r *LogsActorRunner) runFollow(ctx context.Context, actorName string) error scanner.Buffer(buf, 1024*1024) // Support up to 1MB lines for scanner.Scan() { line := scanner.Text() - logTime, _ := filterAndDisplayLogLine(line, actorName, r.stdout) + logTime, _ := filterAndDisplayLogLine(line, r.atespace, actorName, r.stdout) if !logTime.IsZero() { lastSeenTime = logTime } @@ -306,7 +306,7 @@ func runLogsActor(cmd *cobra.Command, args []string) error { return runner.Run(ctx, actorName) } -func filterAndDisplayLogLine(line, targetActorName string, w io.Writer) (time.Time, bool) { +func filterAndDisplayLogLine(line, targetAtespace, targetActorName string, w io.Writer) (time.Time, bool) { var m map[string]any dec := json.NewDecoder(strings.NewReader(line)) dec.UseNumber() @@ -323,19 +323,23 @@ func filterAndDisplayLogLine(line, targetActorName string, w io.Writer) (time.Ti } } - var actorName string + var atespace, actorName string for _, labelKey := range []string{"logging.googleapis.com/labels", "labels"} { if labelsAny, ok := m[labelKey]; ok { if labels, ok := labelsAny.(map[string]any); ok { if name, ok := labels["ate.dev/actor_name"].(string); ok && name != "" { actorName = name + atespace, _ = labels["ate.dev/actor_atespace"].(string) break } } } } - matched := (actorName != "" && actorName == targetActorName) + // Actor names are only unique within an atespace, and a worker pod can host + // actors from different atespaces over time, so match on both. + matched := (actorName != "" && actorName == targetActorName && + targetAtespace != "" && atespace == targetAtespace) if !matched { return logTime, false diff --git a/cmd/kubectl-ate/internal/cmd/logs_actors_test.go b/cmd/kubectl-ate/internal/cmd/logs_actors_test.go index ddafa354d..30b82b650 100644 --- a/cmd/kubectl-ate/internal/cmd/logs_actors_test.go +++ b/cmd/kubectl-ate/internal/cmd/logs_actors_test.go @@ -35,6 +35,7 @@ func TestFilterAndDisplayLogLine(t *testing.T) { tests := []struct { name string line string + targetAtespace string targetActorName string wantMatched bool wantTime string @@ -42,7 +43,8 @@ func TestFilterAndDisplayLogLine(t *testing.T) { }{ { name: "matching actor, JSON log with RFC3339Nano", - line: `{"time":"2026-05-16T01:03:38.602878302Z","level":"info","msg":"Count","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-1"}}`, + line: `{"time":"2026-05-16T01:03:38.602878302Z","level":"info","msg":"Count","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-1"}}`, + targetAtespace: "space-1", targetActorName: "act-1", wantMatched: true, wantTime: "2026-05-16T01:03:38.602878302Z", @@ -50,7 +52,8 @@ func TestFilterAndDisplayLogLine(t *testing.T) { }, { name: "matching actor, plain text log", - line: `{"time":"2026-05-16T01:03:38Z","message":"Hello","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-1"}}`, + line: `{"time":"2026-05-16T01:03:38Z","message":"Hello","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-1"}}`, + targetAtespace: "space-1", targetActorName: "act-1", wantMatched: true, wantTime: "2026-05-16T01:03:38Z", @@ -58,7 +61,8 @@ func TestFilterAndDisplayLogLine(t *testing.T) { }, { name: "matching actor, JSON log with no timestamp fallback", - line: `{"level":"error","msg":"Failed","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-1"}}`, + line: `{"level":"error","msg":"Failed","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-1"}}`, + targetAtespace: "space-1", targetActorName: "act-1", wantMatched: true, wantTime: "", @@ -66,7 +70,8 @@ func TestFilterAndDisplayLogLine(t *testing.T) { }, { name: "matching actor, fallback to standard labels key", - line: `{"time":"2026-05-16T01:03:38.602878302Z","level":"info","msg":"Count","labels":{"ate.dev/actor_name":"act-1"}}`, + line: `{"time":"2026-05-16T01:03:38.602878302Z","level":"info","msg":"Count","labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-1"}}`, + targetAtespace: "space-1", targetActorName: "act-1", wantMatched: true, wantTime: "2026-05-16T01:03:38.602878302Z", @@ -74,7 +79,26 @@ func TestFilterAndDisplayLogLine(t *testing.T) { }, { name: "non-matching actor", - line: `{"time":"2026-05-16T01:03:38Z","message":"Hello world","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-2"}}`, + line: `{"time":"2026-05-16T01:03:38Z","message":"Hello world","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-2"}}`, + targetAtespace: "space-1", + targetActorName: "act-1", + wantMatched: false, + wantTime: "2026-05-16T01:03:38Z", + wantOutput: "", + }, + { + name: "same actor name in a different atespace", + line: `{"time":"2026-05-16T01:03:38Z","message":"Hello world","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-2","ate.dev/actor_name":"act-1"}}`, + targetAtespace: "space-1", + targetActorName: "act-1", + wantMatched: false, + wantTime: "2026-05-16T01:03:38Z", + wantOutput: "", + }, + { + name: "matching actor name without atespace label", + line: `{"time":"2026-05-16T01:03:38Z","message":"Hello world","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-1"}}`, + targetAtespace: "space-1", targetActorName: "act-1", wantMatched: false, wantTime: "2026-05-16T01:03:38Z", @@ -83,6 +107,7 @@ func TestFilterAndDisplayLogLine(t *testing.T) { { name: "invalid json line", line: "not a json line", + targetAtespace: "space-1", targetActorName: "act-1", wantMatched: false, wantTime: "", @@ -90,7 +115,8 @@ func TestFilterAndDisplayLogLine(t *testing.T) { }, { name: "matching actor, flat JSON log", - line: `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"Hello","traceID":"abc-123","err":"timeout","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-1"}}`, + line: `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"Hello","traceID":"abc-123","err":"timeout","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-1"}}`, + targetAtespace: "space-1", targetActorName: "act-1", wantMatched: true, wantTime: "2026-05-16T01:03:38Z", @@ -98,7 +124,8 @@ func TestFilterAndDisplayLogLine(t *testing.T) { }, { name: "matching actor, severity and message keys", - line: `{"time":"2026-05-16T01:03:38Z","severity":"error","message":"Disk full","custom_tag":"alert","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-1"}}`, + line: `{"time":"2026-05-16T01:03:38Z","severity":"error","message":"Disk full","custom_tag":"alert","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-1"}}`, + targetAtespace: "space-1", targetActorName: "act-1", wantMatched: true, wantTime: "2026-05-16T01:03:38Z", @@ -106,7 +133,8 @@ func TestFilterAndDisplayLogLine(t *testing.T) { }, { name: "matching actor, 2-field structured log without time", - line: `{"message":"login failed","code":401,"logging.googleapis.com/labels":{"ate.dev/actor_name":"act-1"}}`, + line: `{"message":"login failed","code":401,"logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-1"}}`, + targetAtespace: "space-1", targetActorName: "act-1", wantMatched: true, wantTime: "", @@ -114,7 +142,8 @@ func TestFilterAndDisplayLogLine(t *testing.T) { }, { name: "matching actor, JSON log with custom application labels", - line: `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"Hello","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-1","app":"my-app"}}`, + line: `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"Hello","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-1","app":"my-app"}}`, + targetAtespace: "space-1", targetActorName: "act-1", wantMatched: true, wantTime: "2026-05-16T01:03:38Z", @@ -125,7 +154,7 @@ func TestFilterAndDisplayLogLine(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { var buf bytes.Buffer - logTime, matched := filterAndDisplayLogLine(tc.line, tc.targetActorName, &buf) + logTime, matched := filterAndDisplayLogLine(tc.line, tc.targetAtespace, tc.targetActorName, &buf) if matched != tc.wantMatched { t.Errorf("got matched = %v, want %v", matched, tc.wantMatched) @@ -199,7 +228,7 @@ func TestLogsActorRunner_Run_OneShotSuccess(t *testing.T) { }, } - logLine := `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"Hello world","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-123"}}` + logLine := `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"Hello world","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-123"}}` mockStreamer := &mockPodLogsStreamer{ StreamLogsFunc: func(ctx context.Context, ns, name string, opts *corev1.PodLogOptions) (io.ReadCloser, error) { if ns != namespace || name != podName { @@ -215,6 +244,7 @@ func TestLogsActorRunner_Run_OneShotSuccess(t *testing.T) { var stdout, stderr bytes.Buffer runner := &LogsActorRunner{ apiClient: mockAPI, + atespace: "space-1", streamer: mockStreamer, stdout: &stdout, stderr: &stderr, @@ -258,6 +288,7 @@ func TestLogsActorRunner_Run_OneShot_ActorNotRunning(t *testing.T) { var stdout, stderr bytes.Buffer runner := &LogsActorRunner{ apiClient: mockAPI, + atespace: "space-1", streamer: mockStreamer, stdout: &stdout, stderr: &stderr, @@ -311,7 +342,7 @@ func TestLogsActorRunner_Run_Follow_SuspendedToRunning(t *testing.T) { }, } - logLine := `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"Follow hello","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-123"}}` + logLine := `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"Follow hello","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-123"}}` ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -342,6 +373,7 @@ func TestLogsActorRunner_Run_Follow_SuspendedToRunning(t *testing.T) { var stdout, stderr bytes.Buffer runner := &LogsActorRunner{ apiClient: mockAPI, + atespace: "space-1", streamer: mockStreamer, stdout: &stdout, stderr: &stderr, @@ -391,6 +423,7 @@ func TestLogsActorRunner_Run_Follow_NotFoundActor(t *testing.T) { var stdout, stderr bytes.Buffer runner := &LogsActorRunner{ apiClient: mockAPI, + atespace: "space-1", streamer: mockStreamer, stdout: &stdout, stderr: &stderr, @@ -473,7 +506,7 @@ func TestLogsActorRunner_Run_Follow_ActorMigration(t *testing.T) { pr, pw := io.Pipe() go func() { // write one line and then keep it open - fmt.Fprintln(pw, `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"line 1 from pod-1","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-migrate"}}`) + fmt.Fprintln(pw, `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"line 1 from pod-1","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-migrate"}}`) close(lineRead) // guaranteed to have been read because io.Pipe is unbuffered! // wait until context is cancelled <-streamCtx.Done() @@ -490,13 +523,14 @@ func TestLogsActorRunner_Run_Follow_ActorMigration(t *testing.T) { // Now we can cancel the main context to exit the follow loop cancel() - return io.NopCloser(strings.NewReader(`{"time":"2026-05-16T01:03:39Z","level":"info","msg":"line 1 from pod-2","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-migrate"}}` + "\n")), nil + return io.NopCloser(strings.NewReader(`{"time":"2026-05-16T01:03:39Z","level":"info","msg":"line 1 from pod-2","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-migrate"}}` + "\n")), nil }, } var stdout, stderr bytes.Buffer runner := &LogsActorRunner{ apiClient: mockAPI, + atespace: "space-1", streamer: mockStreamer, stdout: &stdout, stderr: &stderr, @@ -591,7 +625,7 @@ func TestLogsActorRunner_Run_Follow_ActorSuspendedMidStream(t *testing.T) { if streamCalls == 1 { pr, pw := io.Pipe() go func() { - fmt.Fprintln(pw, `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"before suspend","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-suspended-mid"}}`) + fmt.Fprintln(pw, `{"time":"2026-05-16T01:03:38Z","level":"info","msg":"before suspend","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-suspended-mid"}}`) close(lineRead) // guaranteed to have been read! <-streamCtx.Done() pw.Close() @@ -602,13 +636,14 @@ func TestLogsActorRunner_Run_Follow_ActorSuspendedMidStream(t *testing.T) { // Second stream (after resuming): cancel context to stop test cancel() - return io.NopCloser(strings.NewReader(`{"time":"2026-05-16T01:03:40Z","level":"info","msg":"after resume","logging.googleapis.com/labels":{"ate.dev/actor_name":"act-suspended-mid"}}` + "\n")), nil + return io.NopCloser(strings.NewReader(`{"time":"2026-05-16T01:03:40Z","level":"info","msg":"after resume","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":"act-suspended-mid"}}` + "\n")), nil }, } var stdout, stderr bytes.Buffer runner := &LogsActorRunner{ apiClient: mockAPI, + atespace: "space-1", streamer: mockStreamer, stdout: &stdout, stderr: &stderr, diff --git a/cmd/kubectl-ate/internal/printer/printer.go b/cmd/kubectl-ate/internal/printer/printer.go index 606f760d1..bd5b23481 100644 --- a/cmd/kubectl-ate/internal/printer/printer.go +++ b/cmd/kubectl-ate/internal/printer/printer.go @@ -127,8 +127,8 @@ func PrintWorkersTo(out io.Writer, workers []*ateapipb.Worker, format string) er assignedActor := "" if wass := worker.Assignment; wass != nil { status = "ASSIGNED" - assignedActor = fmt.Sprintf("%s/%s/%s", - wass.ActorTemplate.Namespace, wass.ActorTemplate.Name, wass.Actor.Name) + assignedActor = fmt.Sprintf("%s/%s/%s/%s", + wass.ActorTemplate.Namespace, wass.ActorTemplate.Name, wass.Actor.Atespace, wass.Actor.Name) } fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", ns, pool, pod, status, assignedActor) diff --git a/cmd/kubectl-ate/internal/printer/printer_test.go b/cmd/kubectl-ate/internal/printer/printer_test.go index b84826bdd..26873b143 100644 --- a/cmd/kubectl-ate/internal/printer/printer_test.go +++ b/cmd/kubectl-ate/internal/printer/printer_test.go @@ -214,7 +214,8 @@ func TestPrintWorkersTo_Table(t *testing.T) { Name: "template-1", }, Actor: &ateapipb.ObjectRef{ - Name: "id-1", + Atespace: "space-1", + Name: "id-1", }, }, }, @@ -226,7 +227,7 @@ func TestPrintWorkersTo_Table(t *testing.T) { output := buf.String() expected := `NAMESPACE POOL POD STATUS ASSIGNED ACTOR -default pool-1 pod-1 ASSIGNED default/template-1/id-1 +default pool-1 pod-1 ASSIGNED default/template-1/space-1/id-1 ` if diff := cmp.Diff(expected, output); diff != "" { t.Errorf("output mismatch (-want +got):\n%s", diff) From dc9199f2dea20e3f39f1138da280147b69d838f0 Mon Sep 17 00:00:00 2001 From: Haven Xia Date: Fri, 17 Jul 2026 14:05:13 -0700 Subject: [PATCH 2/2] Require non-empty atespace when filtering log lines --- .../internal/cmd/logs_actors_test.go | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/cmd/kubectl-ate/internal/cmd/logs_actors_test.go b/cmd/kubectl-ate/internal/cmd/logs_actors_test.go index 30b82b650..5ff52fddc 100644 --- a/cmd/kubectl-ate/internal/cmd/logs_actors_test.go +++ b/cmd/kubectl-ate/internal/cmd/logs_actors_test.go @@ -104,6 +104,24 @@ func TestFilterAndDisplayLogLine(t *testing.T) { wantTime: "2026-05-16T01:03:38Z", wantOutput: "", }, + { + name: "empty target atespace does not match empty atespace label", + line: `{"time":"2026-05-16T01:03:38Z","message":"Hello world","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"","ate.dev/actor_name":"act-1"}}`, + targetAtespace: "", + targetActorName: "act-1", + wantMatched: false, + wantTime: "2026-05-16T01:03:38Z", + wantOutput: "", + }, + { + name: "empty target actor name does not match empty name label", + line: `{"time":"2026-05-16T01:03:38Z","message":"Hello world","logging.googleapis.com/labels":{"ate.dev/actor_atespace":"space-1","ate.dev/actor_name":""}}`, + targetAtespace: "space-1", + targetActorName: "", + wantMatched: false, + wantTime: "2026-05-16T01:03:38Z", + wantOutput: "", + }, { name: "invalid json line", line: "not a json line",