diff --git a/go/api/gen/kagent/api/v1/system.pb.go b/go/api/gen/kagent/api/v1/system.pb.go index 5ae0980ae8..0549b3c927 100644 --- a/go/api/gen/kagent/api/v1/system.pb.go +++ b/go/api/gen/kagent/api/v1/system.pb.go @@ -396,7 +396,8 @@ type GetSubstrateSummaryRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // Kubernetes namespace for workers and pools. Empty uses watched namespaces. Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` - // ATE atespace for actors and templates. Empty lists all atespaces. + // ATE atespace for actors and templates. Empty lists watched atespaces (all when + // the controller watches all namespaces). Atespace string `protobuf:"bytes,2,opt,name=atespace,proto3" json:"atespace,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache @@ -614,10 +615,12 @@ func (x *GetSubstrateSummaryResponse) GetActorTemplates() []*ateapipb.ActorTempl return nil } -// One upstream page of actors, in Substrate order. +// One page of actors. A controller that watches some namespaces reads its watched +// atespaces one at a time, in name order, and Substrate order inside each. type ListSubstrateActorsRequest struct { state protoimpl.MessageState `protogen:"open.v1"` - // ATE atespace of the actor itself. Empty lists all atespaces. + // ATE atespace of the actor itself. Empty lists watched atespaces (all when the + // controller watches all namespaces). Atespace string `protobuf:"bytes,6,opt,name=atespace,proto3" json:"atespace,omitempty"` // `limit` is capped at 100 in common.proto and zero means the server's default; // above the cap is refused rather than clamped. @@ -674,7 +677,9 @@ type ListSubstrateActorsResponse struct { state protoimpl.MessageState `protogen:"open.v1"` AteApiError string `protobuf:"bytes,2,opt,name=ate_api_error,json=ateApiError,proto3" json:"ate_api_error,omitempty"` Actors []*ateapipb.Actor `protobuf:"bytes,3,rep,name=actors,proto3" json:"actors,omitempty"` - // Upstream continuation token. A page can be empty even when another page exists. + // Continuation token. A controller that watches all namespaces passes Substrate's + // token through. A controller that watches some issues its own, which names only + // watched data. A page can be empty even when another page exists. Page *PageResponse `protobuf:"bytes,4,opt,name=page,proto3" json:"page,omitempty"` ComputedAt *timestamppb.Timestamp `protobuf:"bytes,5,opt,name=computed_at,json=computedAt,proto3" json:"computed_at,omitempty"` unknownFields protoimpl.UnknownFields @@ -739,8 +744,10 @@ func (x *ListSubstrateActorsResponse) GetComputedAt() *timestamppb.Timestamp { return nil } -// One upstream page of workers, in Substrate order. -// Namespace filtering applies to this page only; it may be empty and still have a next token. +// One page of workers. A controller that watches all namespaces filters one Substrate +// page, in Substrate order, which may be empty and still have a next token. A +// controller that watches some reads every Substrate worker and returns only those of +// its watched namespaces, in worker name order; only its last page can be empty. type ListSubstrateWorkersRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` diff --git a/go/core/internal/service/system/service.go b/go/core/internal/service/system/service.go index f3dd63e716..fd316d4a51 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -51,7 +51,7 @@ func NewService( ) *Service { return &Service{ kubeClient: kubeClient, - observedNamespaces: slices.Clone(observedNamespaces), + observedNamespaces: slices.Compact(slices.Sorted(slices.Values(observedNamespaces))), authorizer: authorizer, ateClient: ateClient, } @@ -148,16 +148,6 @@ func namespacesFromNames(names []string) []Namespace { return result } -func (s *Service) substrateNamespaces(requested string) []string { - if requested != "" { - return []string{requested} - } - if len(s.observedNamespaces) > 0 { - return slices.Clone(s.observedNamespaces) - } - return []string{""} -} - func (s *Service) listWorkerPools(ctx context.Context, namespace string) ([]atev1alpha1.WorkerPool, error) { var options []client.ListOption if namespace != "" { @@ -173,23 +163,25 @@ func (s *Service) listWorkerPools(ctx context.Context, namespace string) ([]atev } // substrateActorTemplates drains upstream pagination for the configuration-sized list. -func (s *Service) substrateActorTemplates(ctx context.Context, atespace string) ([]*ateapipb.ActorTemplate, error) { - templatesFromAPI, err := s.ateClient.ListActorTemplates(ctx, atespace) - if err != nil { - return nil, err - } - templates := make([]*ateapipb.ActorTemplate, 0, len(templatesFromAPI)) - for _, template := range templatesFromAPI { - if template == nil { - continue +func (s *Service) substrateActorTemplates(ctx context.Context, scope substrateScope) ([]*ateapipb.ActorTemplate, error) { + templates := []*ateapipb.ActorTemplate{} + for _, atespace := range scope.listNamespaces() { + templatesFromAPI, err := s.ateClient.ListActorTemplates(ctx, atespace) + if err != nil { + return nil, err + } + for _, template := range templatesFromAPI { + if template == nil { + continue + } + // Only expose inventory fields: containers can contain resolved credentials. + templates = append(templates, &ateapipb.ActorTemplate{ + Metadata: template.GetMetadata(), + Status: template.GetStatus(), + SandboxConfig: template.GetSandboxConfig(), + WorkerSelector: template.GetWorkerSelector(), + }) } - // Only expose inventory fields: containers can contain resolved credentials. - templates = append(templates, &ateapipb.ActorTemplate{ - Metadata: template.GetMetadata(), - Status: template.GetStatus(), - SandboxConfig: template.GetSandboxConfig(), - WorkerSelector: template.GetWorkerSelector(), - }) } slices.SortStableFunc(templates, func(left, right *ateapipb.ActorTemplate) int { diff --git a/go/core/internal/service/system/service_test.go b/go/core/internal/service/system/service_test.go index c6e19b1dd1..52768ac8bf 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -2,10 +2,10 @@ package system_test import ( "context" + "encoding/base64" + "encoding/json" "errors" - "fmt" - "strconv" - "strings" + "slices" "testing" atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" @@ -46,11 +46,14 @@ type fakeATEClient struct { pageSize int // How many page reads each list has taken, so a test can assert that the token // was followed rather than that the rows came back. - actorReads int - workerReads int + actorReads int + workerReads int + actorAtespaces []string + templateAtespaces []string } func (client *fakeATEClient) ListActorTemplates(_ context.Context, atespace string) ([]*ateapipb.ActorTemplate, error) { + client.templateAtespaces = append(client.templateAtespaces, atespace) templates := []*ateapipb.ActorTemplate{} for _, template := range client.templates { if atespace == "" || template.GetMetadata().GetAtespace() == atespace { @@ -62,10 +65,11 @@ func (client *fakeATEClient) ListActorTemplates(_ context.Context, atespace stri func (client *fakeATEClient) ListActorsPage(_ context.Context, atespace string, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { client.actorReads++ + client.actorAtespaces = append(client.actorAtespaces, atespace) if err := client.readError(client.actorReads); err != nil { return nil, "", err } - return fakePage(actorsInAtespace(client.actors, atespace), client.pageSize, pageSize, pageToken) + return fakePage(actorsInAtespace(client.actors, atespace), actorKey, client.pageSize, pageSize, pageToken) } func (client *fakeATEClient) ListWorkersPage(_ context.Context, pageSize int32, pageToken string) ([]*ateapipb.Worker, string, error) { @@ -73,42 +77,61 @@ func (client *fakeATEClient) ListWorkersPage(_ context.Context, pageSize int32, if err := client.readError(client.workerReads); err != nil { return nil, "", err } - return fakePage(client.workers, client.pageSize, pageSize, pageToken) + return fakePage(client.workers, workerKey, client.pageSize, pageSize, pageToken) } func (client *fakeATEClient) readError(int) error { return client.err } -// fakePage slices rows the way ate-api pages them: an opaque token, empty on the last -// page, and never more rows than its own ceiling however many were asked for. -func fakePage[T any](rows []T, ceiling int, requested int32, pageToken string) ([]T, string, error) { +type fakeCursor struct { + Last string `json:"last"` + Offset int `json:"offset"` +} + +// fakePage slices rows the way ate-api pages them: never more rows than its own +// ceiling however many were asked for, and a token that is empty on the last page and +// otherwise base64 JSON naming the page's last row, as ate-api's does. Carrying the +// row key lets a test see a token that would tell one tenant about another's rows. +func fakePage[T any](rows []T, key func(T) string, ceiling int, requested int32, pageToken string) ([]T, string, error) { pageSize := ceiling if requested > 0 && (ceiling <= 0 || int(requested) < ceiling) { pageSize = int(requested) } start := 0 if pageToken != "" { - parsed, err := strconv.Atoi(strings.TrimPrefix(pageToken, "upstream:")) - if err != nil { - return nil, "", fmt.Errorf("invalid page token %q", pageToken) + var cursor fakeCursor + raw, err := base64.StdEncoding.DecodeString(pageToken) + if err != nil || json.Unmarshal(raw, &cursor) != nil { + return nil, "", errors.New("invalid page token") } - start = parsed - } - if start > len(rows) { - start = len(rows) + start = max(cursor.Offset, 0) } + start = min(start, len(rows)) if pageSize <= 0 { return rows[start:], "", nil } end := min(start+pageSize, len(rows)) next := "" if end < len(rows) { - next = "upstream:" + strconv.Itoa(end) + next = fakePageToken(key(rows[end-1]), end) } return rows[start:end], next, nil } +func fakePageToken(last string, offset int) string { + raw, _ := json.Marshal(fakeCursor{Last: last, Offset: offset}) + return base64.StdEncoding.EncodeToString(raw) +} + +func actorKey(actor *ateapipb.Actor) string { + return actor.GetMetadata().GetAtespace() + "/" + actor.GetMetadata().GetName() +} + +func workerKey(worker *ateapipb.Worker) string { + return worker.GetMetadata().GetName() +} + func TestCurrentUser(t *testing.T) { service := system.NewService(nil, nil, nil, nil) claims := map[string]any{"sub": "user-1", "groups": []any{"admins"}} @@ -273,7 +296,7 @@ func TestListSubstrateWorkers(t *testing.T) { require.NoError(t, err) require.Len(t, page.Workers, 1) assert.Equal(t, "worker-0", page.Workers[0].WorkerPod) - assert.Equal(t, "upstream:2", page.NextPageToken) + assert.Equal(t, fakePageToken("", 2), page.NextPageToken) assert.Equal(t, 1, ateClient.workerReads) next, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "team", Page: &apiv1.PageRequest{Limit: 2, PageToken: page.NextPageToken}}) require.NoError(t, err) @@ -452,9 +475,13 @@ func TestGetSubstrateSummaryReadsAreIndependent(t *testing.T) { // counted while the pods they sit on cannot. type failingWorkersATEClient struct { fakeATEClient + workerErr error } func (client *failingWorkersATEClient) ListWorkersPage(context.Context, int32, string) ([]*ateapipb.Worker, string, error) { + if client.workerErr != nil { + return nil, "", client.workerErr + } return nil, "", errors.New("workers unavailable") } @@ -529,7 +556,7 @@ func TestSubstrateScopesAreIndependent(t *testing.T) { }, workers: []*ateapipb.Worker{{WorkerNamespace: "workers", WorkerPod: "pod"}}, } - service := system.NewService(kubeClient, []string{"workers"}, &pkgAuth.NoopAuthorizer{}, ateClient) + service := system.NewService(kubeClient, []string{"workers", "team-a", "shared"}, &pkgAuth.NoopAuthorizer{}, ateClient) for _, tc := range []struct { atespace string count int64 @@ -538,7 +565,6 @@ func TestSubstrateScopesAreIndependent(t *testing.T) { {atespace: "", count: 2, running: 1}, {atespace: "team-a", count: 1, running: 1}, {atespace: "shared", count: 1, running: 0}, - {atespace: "missing", count: 0, running: 0}, } { t.Run(tc.atespace, func(t *testing.T) { summary, err := service.GetSubstrateSummary(ctx, "", tc.atespace) @@ -593,16 +619,159 @@ func (c *emptyPageATEClient) ListWorkersPage(_ context.Context, size int32, toke func TestSubstrateEmptyPagesPreserveUpstreamTokens(t *testing.T) { ctx := pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}}) - service := system.NewService(nil, []string{"team"}, &pkgAuth.NoopAuthorizer{}, &emptyPageATEClient{t: t}) + service := system.NewService(nil, nil, &pkgAuth.NoopAuthorizer{}, &emptyPageATEClient{t: t}) page := &apiv1.PageRequest{Limit: 7, PageToken: "upstream:opaque/cursor="} actors, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "team", Page: page}) require.NoError(t, err) assert.Empty(t, actors.Actors) assert.Empty(t, actors.ATEAPIError) assert.Equal(t, "upstream:next/cursor=", actors.NextPageToken) - workers, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Page: page}) + workers, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "team", Page: page}) require.NoError(t, err) assert.Empty(t, workers.Workers) assert.Empty(t, workers.ATEAPIError) assert.Equal(t, "upstream:next/cursor=", workers.NextPageToken) } + +func TestSubstrateTenantIsolation(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, atev1alpha1.AddToScheme(scheme)) + ctx := pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}}) + ateClient := &fakeATEClient{ + pageSize: 1, + actors: []*ateapipb.Actor{ + substrateActor("session-b", "tenant-b", ateapipb.ActorState_ACTOR_STATE_RUNNING, "tenant-b", "pod-b"), + substrateActor("session-a", "tenant-a", ateapipb.ActorState_ACTOR_STATE_SUSPENDED, "", ""), + substrateActor("session-c", "tenant-c", ateapipb.ActorState_ACTOR_STATE_RUNNING, "tenant-c", "pod-c"), + substrateActor("session-b2", "tenant-b", ateapipb.ActorState_ACTOR_STATE_PAUSED, "", ""), + }, + templates: []*ateapipb.ActorTemplate{ + {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-a", Name: "agent-a"}}, + {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-b", Name: "agent-b"}}, + {Metadata: &ateapipb.ResourceMetadata{Atespace: "tenant-c", Name: "agent-c"}}, + }, + workers: []*ateapipb.Worker{ + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-a"}, WorkerNamespace: "tenant-a", WorkerPod: "pod-a"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-b"}, WorkerNamespace: "tenant-b", WorkerPod: "pod-b"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-c"}, WorkerNamespace: "tenant-c", WorkerPod: "pod-c"}, + {Metadata: &ateapipb.ResourceMetadata{Name: "uid-unlabelled"}, WorkerPod: "pod-unlabelled"}, + }, + } + newService := func(observedNamespaces []string) *system.Service { + return system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), observedNamespaces, &pkgAuth.NoopAuthorizer{}, ateClient) + } + tenantA := newService([]string{"tenant-a"}) + tenantsAC := newService([]string{"tenant-a", "tenant-c"}) + clusterWide := newService(nil) + + listAllActors := func(t *testing.T, service *system.Service, atespace string) []string { + var names []string + request := &apiv1.ListSubstrateActorsRequest{Atespace: atespace, Page: &apiv1.PageRequest{Limit: 1}} + for { + page, err := service.ListSubstrateActors(ctx, request) + require.NoError(t, err) + for _, actor := range page.Actors { + names = append(names, actor.GetMetadata().GetAtespace()+"/"+actor.GetMetadata().GetName()) + } + if page.NextPageToken == "" { + return names + } + request.Page.PageToken = page.NextPageToken + } + } + workerPods := func(t *testing.T, service *system.Service) []string { + var pods []string + request := &apiv1.ListSubstrateWorkersRequest{Page: &apiv1.PageRequest{}} + for { + page, err := service.ListSubstrateWorkers(ctx, request) + require.NoError(t, err) + for _, worker := range page.Workers { + pods = append(pods, worker.GetWorkerPod()) + } + if page.NextPageToken == "" { + return pods + } + request.Page.PageToken = page.NextPageToken + } + } + templateNames := func(summary system.SubstrateSummary) []string { + var names []string + for _, template := range summary.ActorTemplates { + names = append(names, template.GetMetadata().GetName()) + } + return names + } + + t.Run("lists only its own actors", func(t *testing.T) { + assert.Equal(t, []string{"tenant-a/session-a"}, listAllActors(t, tenantA, "")) + }) + + t.Run("summarises only its own atespace", func(t *testing.T) { + summary, err := tenantA.GetSubstrateSummary(ctx, "", "") + require.NoError(t, err) + require.Empty(t, summary.ATEAPIError) + assert.Equal(t, []string{"agent-a"}, templateNames(summary)) + assert.Equal(t, int64(1), summary.ActorCount) + assert.Equal(t, int64(0), summary.RunningActorCount) + assert.Equal(t, int64(1), summary.WorkerCount) + }) + + t.Run("hides a worker that reports no namespace", func(t *testing.T) { + assert.Equal(t, []string{"pod-a"}, workerPods(t, tenantA)) + assert.Equal(t, []string{"pod-a", "pod-c"}, workerPods(t, tenantsAC)) + }) + + t.Run("a controller watching two namespaces lists only those atespaces", func(t *testing.T) { + assert.Equal(t, []string{"tenant-a/session-a", "tenant-c/session-c"}, listAllActors(t, tenantsAC, "")) + }) + + t.Run("a controller watching two namespaces asks ate-api for only those atespaces", func(t *testing.T) { + ateClient.actorAtespaces, ateClient.templateAtespaces = nil, nil + summary, err := tenantsAC.GetSubstrateSummary(ctx, "", "") + require.NoError(t, err) + require.Empty(t, summary.ATEAPIError) + assert.Equal(t, int64(2), summary.ActorCount) + assert.Equal(t, int64(1), summary.RunningActorCount) + assert.ElementsMatch(t, []system.SubstrateActorStatusCount{ + {State: ateapipb.ActorState_ACTOR_STATE_RUNNING, Count: 1}, + {State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED, Count: 1}, + }, summary.ActorStatusCounts) + assert.Equal(t, []string{"agent-a", "agent-c"}, templateNames(summary)) + assert.Equal(t, int64(2), summary.WorkerCount) + + assert.Equal(t, []string{"tenant-a", "tenant-c"}, slices.Compact(slices.Sorted(slices.Values(ateClient.actorAtespaces)))) + assert.ElementsMatch(t, []string{"tenant-a", "tenant-c"}, ateClient.templateAtespaces) + }) + + t.Run("refuses another tenant's atespace or namespace", func(t *testing.T) { + _, err := tenantA.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"}) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) + assert.Contains(t, serviceerrors.MessageOf(err), "not watched by this controller", "the caller sees only the message, not the cause") + _, err = tenantA.GetSubstrateSummary(ctx, "", "tenant-b") + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) + _, err = tenantA.GetSubstrateSummary(ctx, "tenant-b", "") + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) + _, err = tenantA.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "tenant-b"}) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), err) + }) + + t.Run("a cluster-wide install still sees every tenant", func(t *testing.T) { + summary, err := clusterWide.GetSubstrateSummary(ctx, "", "") + require.NoError(t, err) + assert.Equal(t, int64(4), summary.ActorCount) + assert.Len(t, summary.ActorTemplates, 3) + assert.Equal(t, int64(4), summary.WorkerCount) + assert.Equal(t, []string{"pod-a", "pod-b", "pod-c", "pod-unlabelled"}, workerPods(t, clusterWide)) + assert.Equal(t, []string{"tenant-b/session-b", "tenant-b/session-b2"}, listAllActors(t, clusterWide, "tenant-b")) + }) + + t.Run("a cluster-wide install counts nothing in an atespace with no rows", func(t *testing.T) { + summary, err := clusterWide.GetSubstrateSummary(ctx, "", "missing") + require.NoError(t, err) + assert.Empty(t, summary.ATEAPIError) + assert.Zero(t, summary.ActorCount) + assert.Zero(t, summary.RunningActorCount) + assert.Empty(t, summary.ActorStatusCounts) + assert.Empty(t, summary.ActorTemplates) + }) +} diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index 33349bbff2..67a9d94798 100644 --- a/go/core/internal/service/system/substrate.go +++ b/go/core/internal/service/system/substrate.go @@ -2,6 +2,7 @@ package system import ( "context" + "encoding/base64" "fmt" "maps" "slices" @@ -10,6 +11,7 @@ import ( atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "google.golang.org/grpc/status" apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" "github.com/kagent-dev/kagent/go/core/internal/service/serviceerrors" @@ -76,7 +78,11 @@ type SubstrateActorStatusCount struct { // ate-api reports no totals, so every count costs a walk of its pages. The walk holds // one page at a time and keeps only tallies. func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, atespace string) (SubstrateSummary, error) { - namespaces, err := s.substrateScope(ctx, requestedNamespace) + workerScope, err := s.authorizeSubstrateScope(ctx, requestedNamespace) + if err != nil { + return SubstrateSummary{}, err + } + actorScope, err := s.resolveSubstrateScope(atespace) if err != nil { return SubstrateSummary{}, err } @@ -88,7 +94,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a ComputedAt: time.Now().UTC(), } - for _, namespace := range namespaces { + for _, namespace := range workerScope.listNamespaces() { workerPools, err := s.listWorkerPools(ctx, namespace) if err != nil { return SubstrateSummary{}, serviceerrors.NewInternal("Failed to list substrate resources from Kubernetes", err) @@ -99,18 +105,16 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a return strings.Compare(left.Namespace+"/"+left.Name, right.Namespace+"/"+right.Name) }) - allowAll, allowed := substrateScopeFilter(namespaces) - // Three independent reads: none gates the others, so one failure leaves the rest // counted rather than zeroing the whole summary. - if templates, err := s.substrateActorTemplates(ctx, atespace); err != nil { + if templates, err := s.substrateActorTemplates(ctx, actorScope); err != nil { result.recordATEError(ctx, err) } else { result.ActorTemplates = templates } statusCounts := map[ateapipb.ActorState]int64{} - if err := s.walkActors(ctx, atespace, func(actor *ateapipb.Actor) { + if err := s.walkActors(ctx, actorScope, func(actor *ateapipb.Actor) { if actor == nil { return } @@ -124,7 +128,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a } if err := s.walkWorkers(ctx, func(worker *ateapipb.Worker) { - if worker == nil || !allowedWorkerNamespace(worker.GetWorkerNamespace(), allowAll, allowed) { + if worker == nil || !workerScope.contains(worker.GetWorkerNamespace()) { return } result.WorkerCount++ @@ -145,40 +149,112 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a return result, nil } -// ListSubstrateActors returns one upstream page without changing its order or token. +// ListSubstrateActors returns one page of actors. func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubstrateActorsRequest) (SubstrateActorPage, error) { - if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: "Substrate"}); err != nil { + scope, err := s.authorizeSubstrateScope(ctx, input.GetAtespace()) + if err != nil { return SubstrateActorPage{}, err } + limit, token := substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken() + var actors []*ateapipb.Actor + var next string + if len(s.observedNamespaces) == 0 { + actors, next, err = s.ateClient.ListActorsPage(ctx, input.GetAtespace(), limit, token) + } else { + start, inner, tokenErr := parseScopedPageToken(scope.namespaces, token) + if tokenErr != nil { + return SubstrateActorPage{}, tokenErr + } + actors, next, err = s.listScopedActors(ctx, scope.namespaces[start:], inner, limit) + } result := SubstrateActorPage{ComputedAt: time.Now().UTC()} - actors, next, err := s.ateClient.ListActorsPage(ctx, input.GetAtespace(), substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api actors", "error", err) return result, nil } - result.Actors = actors + for _, actor := range actors { + if actor != nil && scope.contains(actor.GetMetadata().GetAtespace()) { + result.Actors = append(result.Actors, actor) + } + } result.NextPageToken = next return result, nil } -// ListSubstrateWorkers filters one upstream page by namespace and preserves its token. -// A page with no workers in scope may still have a next page. +const maxPageTokenBytes = 1024 + +// scopedPageToken is the cursor a controller that watches some namespaces hands out: +// the watched namespace a page ended in, then where to resume. +func scopedPageToken(namespace, cursor string) string { + return base64.RawURLEncoding.EncodeToString([]byte(namespace + "/" + cursor)) +} + +func parseScopedPageToken(namespaces []string, token string) (int, string, error) { + if token == "" { + return 0, "", nil + } + if len(token) > maxPageTokenBytes { + return 0, "", invalidPageToken() + } + decoded, err := base64.RawURLEncoding.DecodeString(token) + namespace, cursor, ok := strings.Cut(string(decoded), "/") + start := slices.Index(namespaces, namespace) + if err != nil || !ok || start < 0 { + return 0, "", invalidPageToken() + } + return start, cursor, nil +} + +func invalidPageToken() error { + return serviceerrors.NewInvalidArgument("Invalid page token", nil) +} + +func (s *Service) listScopedActors(ctx context.Context, atespaces []string, inner string, limit int32) ([]*ateapipb.Actor, string, error) { + var actors []*ateapipb.Actor + for i, atespace := range atespaces { + rows, next, err := s.ateClient.ListActorsPage(ctx, atespace, limit-int32(len(actors)), inner) + if err != nil { + return nil, "", err + } + actors = append(actors, rows...) + switch { + case next != "": + return actors, scopedPageToken(atespace, next), nil + case int32(len(actors)) >= limit && i+1 < len(atespaces): + return actors, scopedPageToken(atespaces[i+1], ""), nil + } + inner = "" + } + return actors, "", nil +} + +// ListSubstrateWorkers returns one page of workers. func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSubstrateWorkersRequest) (SubstrateWorkerPage, error) { - namespaces, err := s.substrateScope(ctx, input.GetNamespace()) + scope, err := s.authorizeSubstrateScope(ctx, input.GetNamespace()) if err != nil { return SubstrateWorkerPage{}, err } + limit, token := substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken() + if len(s.observedNamespaces) > 0 { + _, after, err := parseScopedPageToken(scope.namespaces, token) + if err == nil && token != "" && after == "" { + err = invalidPageToken() + } + if err != nil { + return SubstrateWorkerPage{}, err + } + return s.listWatchedWorkers(ctx, scope, limit, after), nil + } result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} - workers, next, err := s.ateClient.ListWorkersPage(ctx, substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) + workers, next, err := s.ateClient.ListWorkersPage(ctx, limit, token) if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api workers", "error", err) return result, nil } - allowAll, allowed := substrateScopeFilter(namespaces) for _, worker := range workers { - if worker != nil && allowedWorkerNamespace(worker.GetWorkerNamespace(), allowAll, allowed) { + if worker != nil && scope.contains(worker.GetWorkerNamespace()) { result.Workers = append(result.Workers, worker) } } @@ -186,17 +262,51 @@ func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSub return result, nil } -// walkActors visits every actor in the requested atespace, one page at a time. -func (s *Service) walkActors(ctx context.Context, atespace string, visit func(*ateapipb.Actor)) error { - read := func(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { - return s.ateClient.ListActorsPage(ctx, atespace, pageSize, pageToken) +func (s *Service) listWatchedWorkers(ctx context.Context, scope substrateScope, limit int32, after string) SubstrateWorkerPage { + result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} + var workers []*ateapipb.Worker + if err := s.walkWorkers(ctx, func(worker *ateapipb.Worker) { + if scope.contains(worker.GetWorkerNamespace()) && worker.GetMetadata().GetName() > after { + workers = append(workers, worker) + } + }); err != nil { + result.ATEAPIError = err.Error() + return result } - return walkSubstrate(ctx, read, visit) + slices.SortFunc(workers, func(left, right *ateapipb.Worker) int { + return strings.Compare(left.GetMetadata().GetName(), right.GetMetadata().GetName()) + }) + if int32(len(workers)) > limit { + workers = workers[:limit] + last := workers[limit-1] + result.NextPageToken = scopedPageToken(last.GetWorkerNamespace(), last.GetMetadata().GetName()) + } + result.Workers = workers + return result +} + +func (s *Service) walkActors(ctx context.Context, scope substrateScope, visit func(*ateapipb.Actor)) error { + for _, atespace := range scope.listNamespaces() { + read := func(ctx context.Context, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { + return s.ateClient.ListActorsPage(ctx, atespace, pageSize, pageToken) + } + if err := walkSubstrate(ctx, read, visit); err != nil { + return err + } + } + return nil } // walkWorkers calls visit for every worker ate-api holds, one page at a time. +// ate-api's ListWorkersRequest has no namespace filter, so a controller that watches +// some namespaces still reads every tenant's workers. func (s *Service) walkWorkers(ctx context.Context, visit func(*ateapipb.Worker)) error { - return walkSubstrate(ctx, s.ateClient.ListWorkersPage, visit) + err := walkSubstrate(ctx, s.ateClient.ListWorkersPage, visit) + if err != nil && len(s.observedNamespaces) > 0 { + logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api workers", "error", err) + return fmt.Errorf("ate-api could not list workers: %s", status.Code(err)) + } + return err } // walkSubstrate calls visit for every row ate-api holds, holding one page at a time @@ -225,36 +335,40 @@ func walkSubstrate[Row any]( return fmt.Errorf("ate-api did not finish paging after %d pages", maxATEPagesPerWalk) } -// substrateScope authorizes the caller and resolves the namespaces a read covers. -// ATE-only actor reads authorize independently of Kubernetes scope. -func (s *Service) substrateScope(ctx context.Context, requestedNamespace string) ([]string, error) { +type substrateScope struct { + all bool + namespaces []string +} + +func (s *Service) authorizeSubstrateScope(ctx context.Context, requested string) (substrateScope, error) { if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: "Substrate"}); err != nil { - return nil, err + return substrateScope{}, err } - return s.substrateNamespaces(requestedNamespace), nil + return s.resolveSubstrateScope(requested) } -// substrateScopeFilter turns resolved namespaces into the pair every row filter here -// takes: whether every namespace is in scope, and the set that is when it is not. -func substrateScopeFilter(namespaces []string) (bool, map[string]struct{}) { - allowAll := len(namespaces) == 1 && namespaces[0] == "" - allowed := make(map[string]struct{}, len(namespaces)) - for _, namespace := range namespaces { - if namespace != "" { - allowed[namespace] = struct{}{} - } +func (s *Service) resolveSubstrateScope(requested string) (substrateScope, error) { + switch { + case requested == "" && len(s.observedNamespaces) == 0: + return substrateScope{all: true}, nil + case requested == "": + return substrateScope{namespaces: slices.Clone(s.observedNamespaces)}, nil + case len(s.observedNamespaces) == 0 || slices.Contains(s.observedNamespaces, requested): + return substrateScope{namespaces: []string{requested}}, nil + default: + return substrateScope{}, serviceerrors.NewPermissionDenied(fmt.Sprintf("Namespace %q is not watched by this controller", requested), nil) } - return allowAll, allowed } -// allowedWorkerNamespace filters workers by their Kubernetes namespace. -func allowedWorkerNamespace(namespace string, allowAll bool, allowed map[string]struct{}) bool { - namespace = strings.TrimSpace(namespace) - if allowAll || namespace == "" { - return true +func (scope substrateScope) contains(namespace string) bool { + return scope.all || slices.Contains(scope.namespaces, namespace) +} + +func (scope substrateScope) listNamespaces() []string { + if scope.all { + return []string{""} } - _, ok := allowed[namespace] - return ok + return scope.namespaces } func substratePageSize(requested int32) int32 { diff --git a/go/core/internal/service/system/substrate_paging_test.go b/go/core/internal/service/system/substrate_paging_test.go new file mode 100644 index 0000000000..bcda6ef98e --- /dev/null +++ b/go/core/internal/service/system/substrate_paging_test.go @@ -0,0 +1,280 @@ +package system_test + +import ( + "context" + "encoding/base64" + "fmt" + "math/rand/v2" + "strings" + "testing" + + atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" + "github.com/agent-substrate/substrate/pkg/proto/ateapipb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1" + authimpl "github.com/kagent-dev/kagent/go/core/internal/httpserver/auth" + "github.com/kagent-dev/kagent/go/core/internal/service/serviceerrors" + "github.com/kagent-dev/kagent/go/core/internal/service/system" + pkgAuth "github.com/kagent-dev/kagent/go/core/pkg/auth" +) + +func pagingContext(t *testing.T) context.Context { + return pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}}) +} + +func pagingATEClient() *fakeATEClient { + client := &fakeATEClient{pageSize: 2} + for _, actor := range []string{"tenant-a/a-1", "tenant-b/b-1", "tenant-c/c-1", "tenant-a/a-2", "tenant-b/b-2", "tenant-c/c-2", "tenant-a/a-3"} { + atespace, name, _ := strings.Cut(actor, "/") + client.actors = append(client.actors, substrateActor(name, atespace, ateapipb.ActorState_ACTOR_STATE_RUNNING, "", "")) + } + for _, worker := range []string{"tenant-a/u4", "tenant-b/u2", "tenant-a/u1", "tenant-c/u3", "tenant-a/u6", "tenant-b/u5"} { + namespace, uid, _ := strings.Cut(worker, "/") + client.workers = append(client.workers, &ateapipb.Worker{Metadata: &ateapipb.ResourceMetadata{Name: uid + "-" + namespace}, WorkerNamespace: namespace}) + } + return client +} + +func pagingService(ateClient system.ATEClient) *system.Service { + return system.NewService(nil, []string{"tenant-c", "tenant-a", "tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) +} + +// tokenText is a page token with every base64 layer under it decoded, so a row key +// nested in an ate-api cursor inside a kagent cursor is still visible. +func tokenText(token string) string { + text := token + for _, encoding := range []*base64.Encoding{base64.RawURLEncoding, base64.StdEncoding} { + if raw, err := encoding.DecodeString(token); err == nil && len(raw) > 0 { + text += "\n" + tokenText(string(raw)) + } + } + if _, inner, ok := strings.Cut(token, "/"); ok { + text += "\n" + tokenText(inner) + } + return text +} + +func assertOwnToken(t *testing.T, rows int, pageToken string) { + t.Helper() + assert.NotContains(t, tokenText(pageToken), "tenant-b") + if rows == 0 { + assert.Empty(t, pageToken, "an empty page must not lead on") + } +} + +func walkActorNames(t *testing.T, service *system.Service, atespace string, limit int32) []string { + t.Helper() + names := []string{} + pageToken := "" + for range 100 { + page, err := service.ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Atespace: atespace, Page: &apiv1.PageRequest{Limit: limit, PageToken: pageToken}}) + require.NoError(t, err) + require.Empty(t, page.ATEAPIError) + assertOwnToken(t, len(page.Actors), page.NextPageToken) + names = append(names, actorNames(page.Actors)...) + if pageToken = page.NextPageToken; pageToken == "" { + return names + } + } + t.Fatal("the walk did not end") + return nil +} + +func TestListSubstrateActorsPagesWatchedAtespaces(t *testing.T) { + for atespace, want := range map[string][]string{ + "": {"a-1", "a-2", "a-3", "c-1", "c-2"}, + "tenant-c": {"c-1", "c-2"}, + } { + for _, limit := range []int32{1, 2, 3, 4, 100} { + t.Run(fmt.Sprintf("%q/%d", atespace, limit), func(t *testing.T) { + ateClient := pagingATEClient() + assert.Equal(t, want, walkActorNames(t, pagingService(ateClient), atespace, limit)) + assert.Subset(t, []string{"tenant-a", "tenant-c"}, ateClient.actorAtespaces, "every ate-api read is scoped to a watched atespace") + }) + } + } +} + +func actorNames(actors []*ateapipb.Actor) []string { + names := []string{} + for _, actor := range actors { + names = append(names, actor.GetMetadata().GetName()) + } + return names +} + +func scopedPageToken(raw string) string { + return base64.RawURLEncoding.EncodeToString([]byte(raw)) +} + +func TestListSubstrateActorsRefusesForeignTokens(t *testing.T) { + for name, tc := range map[string]struct { + atespace string + pageToken string + }{ + "not base64": {pageToken: "upstream:2"}, + "no separator": {pageToken: scopedPageToken("tenant-a")}, + "empty atespace": {pageToken: scopedPageToken("/upstream:2")}, + "unwatched atespace": {pageToken: scopedPageToken("tenant-b/upstream:2")}, + "outside the request": {atespace: "tenant-c", pageToken: scopedPageToken("tenant-a/upstream:2")}, + "ate-api actor token": {pageToken: scopedPageToken(`{"v":1,"kind":"actor","last":["tenant-b","b-1"]}`)}, + "over 1 KiB, else good": {pageToken: scopedPageToken("tenant-a/" + strings.Repeat("x", 800))}, + } { + t.Run(name, func(t *testing.T) { + ateClient := pagingATEClient() + _, err := pagingService(ateClient).ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Atespace: tc.atespace, Page: &apiv1.PageRequest{Limit: 2, PageToken: tc.pageToken}}) + require.Error(t, err) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + assert.NotContains(t, err.Error(), "tenant-b") + assert.NotContains(t, err.Error(), tc.pageToken) + assert.Zero(t, ateClient.actorReads) + }) + } +} + +func TestListSubstrateWorkersRefusesForeignTokens(t *testing.T) { + for name, tc := range map[string]struct { + namespace string + pageToken string + }{ + "not base64": {pageToken: "upstream:2"}, + "ate-api worker token": {pageToken: fakePageToken("u2-tenant-b", 2)}, + "bare worker name": {pageToken: scopedPageToken("u1-tenant-a")}, + "no worker name": {pageToken: scopedPageToken("tenant-a/")}, + "unwatched namespace": {pageToken: scopedPageToken("tenant-b/u2-tenant-b")}, + "outside the request": {namespace: "tenant-c", pageToken: scopedPageToken("tenant-a/u1-tenant-a")}, + "over 1 KiB, else good": {pageToken: scopedPageToken("tenant-a/" + strings.Repeat("x", 800))}, + } { + t.Run(name, func(t *testing.T) { + ateClient := pagingATEClient() + _, err := pagingService(ateClient).ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{Namespace: tc.namespace, Page: &apiv1.PageRequest{Limit: 2, PageToken: tc.pageToken}}) + require.Error(t, err) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + assert.NotContains(t, err.Error(), "tenant-b") + assert.NotContains(t, err.Error(), tc.pageToken) + assert.Zero(t, ateClient.workerReads) + }) + } +} + +func walkWorkerNames(t *testing.T, service *system.Service, namespace string, limit int32) []string { + t.Helper() + names := []string{} + pageToken := "" + for range 100 { + page, err := service.ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{Namespace: namespace, Page: &apiv1.PageRequest{Limit: limit, PageToken: pageToken}}) + require.NoError(t, err) + require.Empty(t, page.ATEAPIError) + assertOwnToken(t, len(page.Workers), page.NextPageToken) + for _, worker := range page.Workers { + names = append(names, worker.GetMetadata().GetName()) + } + if pageToken = page.NextPageToken; pageToken == "" { + return names + } + } + t.Fatal("the walk did not end") + return nil +} + +func TestListSubstrateWorkersPagesWatchedNamespaces(t *testing.T) { + for namespace, want := range map[string][]string{ + "": {"u1-tenant-a", "u3-tenant-c", "u4-tenant-a", "u6-tenant-a"}, + "tenant-a": {"u1-tenant-a", "u4-tenant-a", "u6-tenant-a"}, + } { + for _, limit := range []int32{1, 2, 3, 100} { + t.Run(fmt.Sprintf("%q/%d", namespace, limit), func(t *testing.T) { + assert.Equal(t, want, walkWorkerNames(t, pagingService(pagingATEClient()), namespace, limit)) + }) + } + } +} + +func TestListSubstrateWorkersHidesATEAPIErrorText(t *testing.T) { + ateClient := pagingATEClient() + ateClient.err = status.Error(codes.Internal, "worker u2-tenant-b is down") + + page, err := pagingService(ateClient).ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{}) + require.NoError(t, err) + assert.Equal(t, "ate-api could not list workers: Internal", page.ATEAPIError) + assert.Empty(t, page.Workers) + assert.Empty(t, page.NextPageToken) +} + +func TestGetSubstrateSummaryHidesATEAPIWorkerErrorText(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, atev1alpha1.AddToScheme(scheme)) + ateClient := &failingWorkersATEClient{fakeATEClient: *pagingATEClient()} + ateClient.workerErr = status.Error(codes.Internal, "worker u2-tenant-b is down") + service := system.NewService(fake.NewClientBuilder().WithScheme(scheme).Build(), []string{"tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) + + summary, err := service.GetSubstrateSummary(pagingContext(t), "", "") + require.NoError(t, err) + assert.Equal(t, "ate-api could not list workers: Internal", summary.ATEAPIError) + assert.Equal(t, int64(3), summary.ActorCount, "the actor walk is independent of the worker walk") +} + +func TestSubstrateForgedTokensStayInWatchedNamespaces(t *testing.T) { + forged := []string{ + scopedPageToken("tenant-b/"), + scopedPageToken("tenant-a/" + fakePageToken("tenant-b/b-1", 1)), + scopedPageToken("tenant-a/" + fakePageToken("", -5)), + scopedPageToken("tenant-a/" + fakePageToken("", 1<<40)), + scopedPageToken("/" + fakePageToken("", 0)), + fakePageToken("tenant-a/a-1", 1), + "!!!", + } + random := rand.New(rand.NewPCG(1, 1)) + for range 300 { + raw := make([]byte, random.IntN(80)) + for i := range raw { + raw[i] = "tenant-abc/xyz=+"[random.IntN(16)] + } + forged = append(forged, scopedPageToken(string(raw)), string(raw)) + } + + ateClient := pagingATEClient() + service := pagingService(ateClient) + watched := []string{"tenant-a", "tenant-c"} + for _, scope := range []string{"", "tenant-a", "tenant-c"} { + for _, pageToken := range forged { + actors, err := service.ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Atespace: scope, Page: &apiv1.PageRequest{Limit: 3, PageToken: pageToken}}) + if err != nil { + require.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + } + for _, actor := range actors.Actors { + require.Contains(t, watched, actor.GetMetadata().GetAtespace()) + } + require.NotContains(t, tokenText(actors.NextPageToken), "tenant-b") + + workers, err := service.ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{Namespace: scope, Page: &apiv1.PageRequest{Limit: 3, PageToken: pageToken}}) + if err != nil { + require.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), err) + } + for _, worker := range workers.Workers { + require.Contains(t, watched, worker.GetWorkerNamespace()) + } + require.NotContains(t, tokenText(workers.NextPageToken), "tenant-b") + } + } + assert.Subset(t, watched, ateClient.actorAtespaces, "every ate-api actor read is scoped to a watched atespace") +} + +func TestSubstrateLongestNamesFitThePageToken(t *testing.T) { + namespace := strings.Repeat("n", 63) + ateClient := &fakeATEClient{pageSize: 1} + for _, suffix := range []string{"0", "1"} { + ateClient.actors = append(ateClient.actors, substrateActor(strings.Repeat("a", 252)+suffix, namespace, ateapipb.ActorState_ACTOR_STATE_RUNNING, "", "")) + ateClient.workers = append(ateClient.workers, &ateapipb.Worker{Metadata: &ateapipb.ResourceMetadata{Name: strings.Repeat("w", 252) + suffix}, WorkerNamespace: namespace}) + } + service := system.NewService(nil, []string{namespace}, &pkgAuth.NoopAuthorizer{}, ateClient) + + assert.Len(t, walkActorNames(t, service, "", 1), 2) + assert.Len(t, walkWorkerNames(t, service, "", 1), 2) +} diff --git a/proto/kagent/api/v1/system.proto b/proto/kagent/api/v1/system.proto index b8e7cbecdf..64d4aadd1b 100644 --- a/proto/kagent/api/v1/system.proto +++ b/proto/kagent/api/v1/system.proto @@ -62,7 +62,8 @@ message GetSubstrateSummaryRequest { max_len: 63 pattern: "^([a-z0-9]([-a-z0-9]*[a-z0-9])?)?$" }]; - // ATE atespace for actors and templates. Empty lists all atespaces. + // ATE atespace for actors and templates. Empty lists watched atespaces (all when + // the controller watches all namespaces). string atespace = 2 [(buf.validate.field).string = { max_len: 63 pattern: "^([a-z0-9]([-a-z0-9]*[a-z0-9])?)?$" @@ -94,9 +95,11 @@ message GetSubstrateSummaryResponse { repeated ateapi.ActorTemplate actor_templates = 11; } -// One upstream page of actors, in Substrate order. +// One page of actors. A controller that watches some namespaces reads its watched +// atespaces one at a time, in name order, and Substrate order inside each. message ListSubstrateActorsRequest { - // ATE atespace of the actor itself. Empty lists all atespaces. + // ATE atespace of the actor itself. Empty lists watched atespaces (all when the + // controller watches all namespaces). string atespace = 6 [(buf.validate.field).string = { max_len: 63 pattern: "^([a-z0-9]([-a-z0-9]*[a-z0-9])?)?$" @@ -109,13 +112,17 @@ message ListSubstrateActorsRequest { message ListSubstrateActorsResponse { string ate_api_error = 2; repeated ateapi.Actor actors = 3; - // Upstream continuation token. A page can be empty even when another page exists. + // Continuation token. A controller that watches all namespaces passes Substrate's + // token through. A controller that watches some issues its own, which names only + // watched data. A page can be empty even when another page exists. PageResponse page = 4; google.protobuf.Timestamp computed_at = 5; } -// One upstream page of workers, in Substrate order. -// Namespace filtering applies to this page only; it may be empty and still have a next token. +// One page of workers. A controller that watches all namespaces filters one Substrate +// page, in Substrate order, which may be empty and still have a next token. A +// controller that watches some reads every Substrate worker and returns only those of +// its watched namespaces, in worker name order; only its last page can be empty. message ListSubstrateWorkersRequest { string namespace = 1 [(buf.validate.field).string = { max_len: 63 diff --git a/ui/playwright/DEFERRED.md b/ui/playwright/DEFERRED.md index be664c4f12..530fd90a6f 100644 --- a/ui/playwright/DEFERRED.md +++ b/ui/playwright/DEFERRED.md @@ -219,11 +219,13 @@ when it holds one page of them, and only the proto says which. | `ListSubstrateActors` | `ListSubstrateActorsRequest { atespace, page }` | a page | `string filter` and a sort field enum — **and ate-api has to grow them first** | | `ListSubstrateWorkers` | `ListSubstrateWorkersRequest { namespace, page }` | a page | the same two, on the same condition | -**Substrate actor and worker lists use upstream pagination.** Each list request makes -one ate-api call and preserves its order and continuation token. Actors use upstream -atespace filtering. Worker namespace filtering applies only to the returned page, so an -empty worker page may still have a next token. The UI labels counts as "on this page" -and offers no actor/worker text search or column sorting. +**Substrate actor and worker lists use upstream pagination.** A controller that watches +every namespace makes one ate-api call per list request and passes its order and +continuation token through. Worker namespace filtering then applies only to the returned +page, so an empty worker page may still have a next token. A controller that watches some +namespaces reads actors with one scoped ate-api list per watched atespace and returns its +own token. It reads every ate-api worker page and returns a worker-name cursor. The UI +labels counts as "on this page" and offers no actor/worker text search or column sorting. Two capabilities remain deferred until Substrate supports them: diff --git a/ui/src/api/domain/substrate.ts b/ui/src/api/domain/substrate.ts index 09e3d1d3d1..f13391950e 100644 --- a/ui/src/api/domain/substrate.ts +++ b/ui/src/api/domain/substrate.ts @@ -94,7 +94,6 @@ export interface Timed { computedAt?: string; } -/** One upstream page; an empty page may still have a continuation token. */ interface SubstratePage extends Timed { /** Upstream read failure. No rows or continuation token accompany it. */ ateApiError?: string; diff --git a/ui/src/generated/kagent/api/v1/system_pb.ts b/ui/src/generated/kagent/api/v1/system_pb.ts index 504a6cdba8..088a107c96 100644 --- a/ui/src/generated/kagent/api/v1/system_pb.ts +++ b/ui/src/generated/kagent/api/v1/system_pb.ts @@ -184,7 +184,8 @@ export type GetSubstrateSummaryRequest = Message<"kagent.api.v1.GetSubstrateSumm namespace: string; /** - * ATE atespace for actors and templates. Empty lists all atespaces. + * ATE atespace for actors and templates. Empty lists watched atespaces (all when + * the controller watches all namespaces). * * @generated from field: string atespace = 2; */ @@ -291,13 +292,15 @@ export const GetSubstrateSummaryResponseSchema: GenMessage & { /** - * ATE atespace of the actor itself. Empty lists all atespaces. + * ATE atespace of the actor itself. Empty lists watched atespaces (all when the + * controller watches all namespaces). * * @generated from field: string atespace = 6; */ @@ -334,7 +337,9 @@ export type ListSubstrateActorsResponse = Message<"kagent.api.v1.ListSubstrateAc actors: Actor[]; /** - * Upstream continuation token. A page can be empty even when another page exists. + * Continuation token. A controller that watches all namespaces passes Substrate's + * token through. A controller that watches some issues its own, which names only + * watched data. A page can be empty even when another page exists. * * @generated from field: kagent.api.v1.PageResponse page = 4; */ @@ -354,8 +359,10 @@ export const ListSubstrateActorsResponseSchema: GenMessage