Skip to content

Make kubectl-ate log and worker output filtered by atespace #458

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Zoe Zhao (zoez7) merged 2 commits into from
Jul 17, 2026
+83 −25
Merged

Make kubectl-ate log and worker output filtered by atespace #458

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
655286f
Make kubectl-ate log and worker output filtered by atespace
HavenXia Jul 17, 2026
dc9199f
Require non-empty atespace when filtering log lines
HavenXia Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions cmd/kubectl-ate/internal/cmd/logs_actors.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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()
Expand All @@ -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
Expand Down
85 changes: 69 additions & 16 deletions cmd/kubectl-ate/internal/cmd/logs_actors_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,86 +35,133 @@ func TestFilterAndDisplayLogLine(t *testing.T) {
tests := []struct {
name string
line string
targetAtespace string
targetActorName string
wantMatched bool
wantTime string
wantOutput string
}{
{
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",
Comment thread
HavenXia marked this conversation as resolved.
targetActorName: "act-1",
wantMatched: true,
wantTime: "2026-05-16T01:03:38.602878302Z",
wantOutput: `{"time":"2026-05-16T01:03:38.602878302Z","level":"info","msg":"Count"}`,
},
{
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",
wantOutput: `{"time":"2026-05-16T01:03:38Z","message":"Hello"}`,
},
{
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: "",
wantOutput: `{"level":"error","msg":"Failed"}`,
},
{
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",
wantOutput: `{"time":"2026-05-16T01:03:38.602878302Z","level":"info","msg":"Count"}`,
},
{
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",
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",
targetAtespace: "space-1",
targetActorName: "act-1",
wantMatched: false,
wantTime: "",
wantOutput: "",
},
{
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",
wantOutput: `{"time":"2026-05-16T01:03:38Z","err":"timeout","level":"info","msg":"Hello","traceID":"abc-123"}`,
},
{
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",
wantOutput: `{"time":"2026-05-16T01:03:38Z","custom_tag":"alert","message":"Disk full","severity":"error"}`,
},
{
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: "",
wantOutput: `{"code":401,"message":"login failed"}`,
},
{
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",
Expand All @@ -125,7 +172,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)
Expand Down Expand Up @@ -199,7 +246,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 {
Expand All @@ -215,6 +262,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,
Expand Down Expand Up @@ -258,6 +306,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,
Expand Down Expand Up @@ -311,7 +360,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()
Expand Down Expand Up @@ -342,6 +391,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,
Expand Down Expand Up @@ -391,6 +441,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,
Expand Down Expand Up @@ -473,7 +524,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()
Expand All @@ -490,13 +541,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,
Expand Down Expand Up @@ -591,7 +643,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()
Expand All @@ -602,13 +654,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,
Expand Down
4 changes: 2 additions & 2 deletions cmd/kubectl-ate/internal/printer/printer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,8 +127,8 @@ func PrintWorkersTo(out io.Writer, workers []*ateapipb.Worker, format string) er
assignedActor := "<none>"
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)
Expand Down
5 changes: 3 additions & 2 deletions cmd/kubectl-ate/internal/printer/printer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
},
},
},
Expand All @@ -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)
Expand Down
Loading