diff --git a/go/api/gen/kagent/api/v1/system.pb.go b/go/api/gen/kagent/api/v1/system.pb.go index 5ae0980ae8..13e953fa5d 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. Only a last page can be empty. 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, in worker name order. A controller that watches all namespaces +// filters one Substrate page, 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; 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..95bd09903f 100644 --- a/go/core/internal/service/system/service.go +++ b/go/core/internal/service/system/service.go @@ -32,7 +32,9 @@ type ATEClient interface { } type Service struct { - kubeClient client.Client + kubeClient client.Client + // Sorted and unique: the order Substrate pages are read in. A repeat must not read + // an atespace twice. Empty means every namespace. observedNamespaces []string authorizer auth.Authorizer ateClient ATEClient @@ -51,7 +53,7 @@ func NewService( ) *Service { return &Service{ kubeClient: kubeClient, - observedNamespaces: slices.Clone(observedNamespaces), + observedNamespaces: slices.Compact(slices.Sorted(slices.Values(observedNamespaces))), authorizer: authorizer, ateClient: ateClient, } @@ -172,24 +174,27 @@ func (s *Service) listWorkerPools(ctx context.Context, namespace string) ([]atev return workerPoolList.Items, nil } -// substrateActorTemplates drains upstream pagination for the configuration-sized list. +// substrateActorTemplates lists the templates of the requested atespace, or of each +// watched one. The list is configuration-sized. 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 + templates := []*ateapipb.ActorTemplate{} + for _, atespace := range s.substrateNamespaces(atespace) { + page, err := s.ateClient.ListActorTemplates(ctx, atespace) + if err != nil { + return nil, err + } + for _, template := range page { + // ate-api scoped the list already. This is a second guard. + if template != nil && (atespace == "" || template.GetMetadata().GetAtespace() == atespace) { + // 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..b493b825a0 100644 --- a/go/core/internal/service/system/service_test.go +++ b/go/core/internal/service/system/service_test.go @@ -1,9 +1,11 @@ package system_test import ( + "cmp" "context" "errors" "fmt" + "slices" "strconv" "strings" "testing" @@ -12,6 +14,8 @@ import ( "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" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -44,40 +48,96 @@ type fakeATEClient struct { // for, so a fake that always fills the request would hide a caller that stops // following the token as soon as it has enough rows. pageSize int + // The ceiling of a worker page, when it differs from pageSize. + workerPageSize 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 + // Every call in order, so a test can assert on what reached ate-api. + calls []ateCall + // ignoreAtespace answers an actor or template list with the rows of every + // atespace, as an ate-api that drops the filter would. + ignoreAtespace bool + // failCall fails the call with this 1-based number with failErr. + failCall int + failErr error +} + +// ateCall is one ate-api call: the method, the atespace, the token and the page size. +type ateCall struct { + method, scope, token string + size int32 +} + +// record notes a call and answers with the error it is set up to fail with. +func (client *fakeATEClient) record(method, scope, token string, size int32) error { + client.calls = append(client.calls, ateCall{method: method, scope: scope, token: token, size: size}) + if client.failCall == len(client.calls) { + return client.failErr + } + return client.err } func (client *fakeATEClient) ListActorTemplates(_ context.Context, atespace string) ([]*ateapipb.ActorTemplate, error) { + if err := client.record("ListActorTemplates", atespace, "", 0); err != nil { + return nil, err + } templates := []*ateapipb.ActorTemplate{} for _, template := range client.templates { - if atespace == "" || template.GetMetadata().GetAtespace() == atespace { + if atespace == "" || client.ignoreAtespace || template.GetMetadata().GetAtespace() == atespace { templates = append(templates, template) } } - return templates, client.err + return templates, nil } func (client *fakeATEClient) ListActorsPage(_ context.Context, atespace string, pageSize int32, pageToken string) ([]*ateapipb.Actor, string, error) { client.actorReads++ - if err := client.readError(client.actorReads); err != nil { + if err := client.record("ListActorsPage", atespace, pageToken, pageSize); err != nil { return nil, "", err } - return fakePage(actorsInAtespace(client.actors, atespace), client.pageSize, pageSize, pageToken) + scope := atespace + if client.ignoreAtespace { + scope = "" + } + return actorPage(actorsInAtespace(client.actors, scope), atespace, client.pageSize, pageSize, pageToken) } func (client *fakeATEClient) ListWorkersPage(_ context.Context, pageSize int32, pageToken string) ([]*ateapipb.Worker, string, error) { client.workerReads++ - if err := client.readError(client.workerReads); err != nil { + if err := client.record("ListWorkersPage", "", pageToken, pageSize); err != nil { return nil, "", err } - return fakePage(client.workers, client.pageSize, pageSize, pageToken) + return fakePage(client.workers, cmp.Or(client.workerPageSize, client.pageSize), pageSize, pageToken) } -func (client *fakeATEClient) readError(int) error { - return client.err +// actorPage pages actors the way ate-api does: in atespace and name order, after the +// last row of the token, with a token bound to the atespace it was issued for. A token +// sent with another atespace is refused, as ate-api refuses it. +func actorPage(actors []*ateapipb.Actor, atespace string, ceiling int, requested int32, pageToken string) ([]*ateapipb.Actor, string, error) { + key := func(actor *ateapipb.Actor) string { + return actor.GetMetadata().GetAtespace() + "/" + actor.GetMetadata().GetName() + } + rows := slices.SortedStableFunc(slices.Values(actors), func(left, right *ateapipb.Actor) int { + return strings.Compare(key(left), key(right)) + }) + prefix := "upstream:" + atespace + ":" + if pageToken != "" { + last, ok := strings.CutPrefix(pageToken, prefix) + if !ok { + return nil, "", status.Error(codes.InvalidArgument, "invalid page_token") + } + rows = slices.DeleteFunc(rows, func(actor *ateapipb.Actor) bool { return key(actor) <= last }) + } + pageSize := ceiling + if requested > 0 && (ceiling <= 0 || int(requested) < ceiling) { + pageSize = int(requested) + } + if pageSize <= 0 || pageSize >= len(rows) { + return rows, "", nil + } + return rows[:pageSize], prefix + key(rows[pageSize-1]), nil } // fakePage slices rows the way ate-api pages them: an opaque token, empty on the last @@ -266,6 +326,7 @@ func TestListSubstrateWorkers(t *testing.T) { {WorkerNamespace: "team", WorkerPool: "pool", WorkerPod: "worker-0", Status: &ateapipb.WorkerStatus{Allocated: &ateapipb.WorkerResources{Actors: 2}}}, {WorkerNamespace: "other", WorkerPool: "pool", WorkerPod: "worker-1"}, {WorkerNamespace: "team", WorkerPool: "pool", WorkerPod: "worker-2"}, + {WorkerPool: "pool", WorkerPod: "names-no-namespace"}, }} service := system.NewService(nil, nil, &pkgAuth.NoopAuthorizer{}, ateClient) @@ -529,7 +590,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, nil, &pkgAuth.NoopAuthorizer{}, ateClient) for _, tc := range []struct { atespace string count int64 @@ -573,36 +634,72 @@ func TestSubstrateScopesAreIndependent(t *testing.T) { assert.Same(t, actor, page.Actors[0]) } -type emptyPageATEClient struct { - fakeATEClient - t *testing.T -} - -func (c *emptyPageATEClient) ListActorsPage(_ context.Context, atespace string, size int32, token string) ([]*ateapipb.Actor, string, error) { - assert.Equal(c.t, "team", atespace) - assert.Equal(c.t, int32(7), size) - assert.Equal(c.t, "upstream:opaque/cursor=", token) - return nil, "upstream:next/cursor=", nil +// A controller that watches namespaces must not read another tenant's namespace or +// atespace, and must not ask ate-api or Kubernetes for it. +func TestSubstrateRefusesUnwatchedNamespaces(t *testing.T) { + ctx := pkgAuth.AuthSessionTo(t.Context(), &authimpl.SimpleSession{P: pkgAuth.Principal{User: pkgAuth.User{ID: "user"}}}) + for _, tc := range []struct { + name string + read func(*system.Service) error + }{ + {"actors in another atespace", func(service *system.Service) error { + _, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-b"}) + return err + }}, + {"actors in a watched atespace with other case", func(service *system.Service) error { + _, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "Tenant-A"}) + return err + }}, + {"actors in a watched atespace with a trailing space", func(service *system.Service) error { + _, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: "tenant-a "}) + return err + }}, + {"a summary of another atespace", func(service *system.Service) error { + _, err := service.GetSubstrateSummary(ctx, "", "tenant-b") + return err + }}, + {"a summary of another namespace", func(service *system.Service) error { + _, err := service.GetSubstrateSummary(ctx, "tenant-b", "") + return err + }}, + {"a summary of a watched namespace and another atespace", func(service *system.Service) error { + _, err := service.GetSubstrateSummary(ctx, "tenant-a", "tenant-b") + return err + }}, + {"workers in another namespace", func(service *system.Service) error { + _, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: "tenant-b"}) + return err + }}, + } { + t.Run(tc.name, func(t *testing.T) { + kubeLists := 0 + ateClient := pagingATEClient() + kubeClient := pagingKubeClient(t, interceptor.Funcs{List: func(ctx context.Context, c ctrlclient.WithWatch, list ctrlclient.ObjectList, opts ...ctrlclient.ListOption) error { + kubeLists++ + return c.List(ctx, list, opts...) + }}) + service := system.NewService(kubeClient, []string{"tenant-a"}, &pkgAuth.NoopAuthorizer{}, ateClient) + + err := tc.read(service) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodePermissionDenied), "err = %v", err) + assert.Empty(t, ateClient.calls, "a refused read must not reach ate-api") + assert.Zero(t, kubeLists, "a refused read must not list WorkerPools") + }) + } } -func (c *emptyPageATEClient) ListWorkersPage(_ context.Context, size int32, token string) ([]*ateapipb.Worker, string, error) { - assert.Equal(c.t, int32(7), size) - assert.Equal(c.t, "upstream:opaque/cursor=", token) - return []*ateapipb.Worker{{WorkerNamespace: "other"}}, "upstream:next/cursor=", nil +func actorNames(actors []*ateapipb.Actor) []string { + names := []string{} + for _, actor := range actors { + names = append(names, actor.GetMetadata().GetName()) + } + return names } -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}) - 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}) - require.NoError(t, err) - assert.Empty(t, workers.Workers) - assert.Empty(t, workers.ATEAPIError) - assert.Equal(t, "upstream:next/cursor=", workers.NextPageToken) +func templateNames(templates []*ateapipb.ActorTemplate) []string { + names := []string{} + for _, template := range templates { + names = append(names, template.GetMetadata().GetName()) + } + return names } diff --git a/go/core/internal/service/system/substrate.go b/go/core/internal/service/system/substrate.go index 33349bbff2..c3059286dc 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,7 @@ 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) + namespaces, err := s.substrateScope(ctx, requestedNamespace, atespace) if err != nil { return SubstrateSummary{}, err } @@ -99,8 +101,6 @@ 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 { @@ -111,9 +111,6 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a statusCounts := map[ateapipb.ActorState]int64{} if err := s.walkActors(ctx, atespace, func(actor *ateapipb.Actor) { - if actor == nil { - return - } result.ActorCount++ statusCounts[actor.GetStatus().GetState()]++ if actor.GetStatus().GetState() == ateapipb.ActorState_ACTOR_STATE_RUNNING { @@ -123,10 +120,7 @@ func (s *Service) GetSubstrateSummary(ctx context.Context, requestedNamespace, a result.recordATEError(ctx, err) } - if err := s.walkWorkers(ctx, func(worker *ateapipb.Worker) { - if worker == nil || !allowedWorkerNamespace(worker.GetWorkerNamespace(), allowAll, allowed) { - return - } + if err := s.walkWorkers(ctx, namespaces, func(worker *ateapipb.Worker) { result.WorkerCount++ if worker.GetStatus().GetAllocated().GetActors() > 0 { result.BusyWorkerCount++ @@ -145,13 +139,39 @@ 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. A controller that watches every +// namespace passes ate-api's page and token through. A controller that watches some +// reads each of its atespaces with its own scoped list, so ate-api reads, and no page +// token names, an actor of a namespace it does not watch. 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 { + if _, err := s.substrateScope(ctx, "", input.GetAtespace()); err != nil { return SubstrateActorPage{}, err } result := SubstrateActorPage{ComputedAt: time.Now().UTC()} - actors, next, err := s.ateClient.ListActorsPage(ctx, input.GetAtespace(), substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken()) + limit, token := substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken() + var actors []*ateapipb.Actor + var next string + var err error + if len(s.observedNamespaces) == 0 { + actors, next, err = s.ateClient.ListActorsPage(ctx, input.GetAtespace(), limit, token) + } else { + atespaces := s.substrateNamespaces(input.GetAtespace()) + start, inner := 0, "" + if token != "" { + // A token without a watched atespace is refused rather than read as page one. + if len(token) > maxPageTokenBytes { + return SubstrateActorPage{}, errInvalidPageToken + } + decoded, decodeErr := base64.RawURLEncoding.DecodeString(token) + atespace, rest, ok := strings.Cut(string(decoded), "/") + start = slices.Index(atespaces, atespace) + if decodeErr != nil || !ok || atespace == "" || start < 0 { + return SubstrateActorPage{}, errInvalidPageToken + } + inner = rest + } + actors, next, err = s.listScopedActors(ctx, atespaces[start:], inner, limit) + } if err != nil { result.ATEAPIError = err.Error() logging.FromContext(ctx).ErrorContext(ctx, "failed to list ate-api actors", "error", err) @@ -162,23 +182,67 @@ func (s *Service) ListSubstrateActors(ctx context.Context, input *apiv1.ListSubs 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. +// actorPageToken is the page token of a controller that watches some namespaces: the +// atespace to read next, and ate-api's token inside it. An atespace holds no "/". +func actorPageToken(atespace, inner string) string { + return base64.RawURLEncoding.EncodeToString([]byte(atespace + "/" + inner)) +} + +// The longest page token this controller accepts. Its own tokens are far shorter. +const maxPageTokenBytes = 1024 + +// errInvalidPageToken carries no cause: the token is the caller's, and its text can +// name anything. +var errInvalidPageToken = serviceerrors.NewInvalidArgument("Invalid page token", nil) + +// listScopedActors fills one page from the atespaces in order, each with its own scoped +// list, starting at ate-api's inner token in the first. +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 + } + for _, actor := range rows { + // ate-api scoped the list already. This is a second guard. + if actor.GetMetadata().GetAtespace() == atespace { + actors = append(actors, actor) + } + } + switch { + case next != "": + return actors, actorPageToken(atespace, next), nil + case int32(len(actors)) >= limit && i+1 < len(atespaces): + return actors, actorPageToken(atespaces[i+1], ""), nil + } + inner = "" + } + return actors, "", nil +} + +// ListSubstrateWorkers returns one page of workers. A controller that watches every +// namespace filters ate-api's page and passes its token through. A controller that +// watches some reads every ate-api worker and returns only the workers of watched +// namespaces, so no page token names a worker of a namespace it does not watch. func (s *Service) ListSubstrateWorkers(ctx context.Context, input *apiv1.ListSubstrateWorkersRequest) (SubstrateWorkerPage, error) { - namespaces, err := s.substrateScope(ctx, input.GetNamespace()) + namespaces, err := s.substrateScope(ctx, input.GetNamespace(), "") if err != nil { return SubstrateWorkerPage{}, err } + limit, token := substratePageSize(input.GetPage().GetLimit()), input.GetPage().GetPageToken() + if len(s.observedNamespaces) > 0 { + return s.listOwnWorkers(ctx, namespaces, limit, token) + } 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 inNamespace(worker, namespaces) { result.Workers = append(result.Workers, worker) } } @@ -186,17 +250,66 @@ 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. +// listOwnWorkers fills one page with the workers of the namespaces, in worker name +// order. The token is the name of the last worker on the previous page: a cursor only, +// never sent to ate-api. +func (s *Service) listOwnWorkers(ctx context.Context, namespaces []string, limit int32, token string) (SubstrateWorkerPage, error) { + result := SubstrateWorkerPage{ComputedAt: time.Now().UTC()} + var workers []*ateapipb.Worker + if err := s.walkWorkers(ctx, namespaces, func(worker *ateapipb.Worker) { + if worker.GetMetadata().GetName() > token { + workers = append(workers, worker) + } + }); err != nil { + result.ATEAPIError = err.Error() + return result, nil + } + // ate-api sends workers in name order already. The cursor must not rely on it. + 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] + result.NextPageToken = workers[limit-1].GetMetadata().GetName() + } + result.Workers = workers + return result, nil +} + +// walkActors visits every actor in the requested atespace, or in each watched one. 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) + for _, atespace := range s.substrateNamespaces(atespace) { + 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, func(actor *ateapipb.Actor) { + // ate-api scoped the list already. This is a second guard. + if actor != nil && (atespace == "" || actor.GetMetadata().GetAtespace() == atespace) { + visit(actor) + } + }); err != nil { + return err + } } - return walkSubstrate(ctx, read, visit) + return nil } -// walkWorkers calls visit for every worker ate-api holds, one page at a time. -func (s *Service) walkWorkers(ctx context.Context, visit func(*ateapipb.Worker)) error { - return walkSubstrate(ctx, s.ateClient.ListWorkersPage, visit) +// walkWorkers visits every worker in the namespaces. A controller that watches some +// reads every ate-api worker and visits only the workers of watched namespaces. +func (s *Service) walkWorkers(ctx context.Context, namespaces []string, visit func(*ateapipb.Worker)) error { + // NOTE: this drains every ate-api worker page per request. Replace it with an + // ate-api worker namespace filter when one lands. + err := walkSubstrate(ctx, s.ateClient.ListWorkersPage, func(worker *ateapipb.Worker) { + if inNamespace(worker, namespaces) { + visit(worker) + } + }) + if err != nil && len(s.observedNamespaces) > 0 { + // ate-api error text can name a worker of any namespace. The user gets the code. + 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 @@ -226,35 +339,30 @@ func walkSubstrate[Row any]( } // 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) { +// A controller that watches namespaces refuses a namespace or atespace it does not watch: +// kagent names each atespace after its namespace, and ate-api serves every tenant. +func (s *Service) substrateScope(ctx context.Context, requestedNamespace, requestedAtespace string) ([]string, error) { if err := s.authorize(ctx, auth.VerbGet, auth.Resource{Type: "Substrate"}); err != nil { return nil, err } + for _, requested := range []string{requestedNamespace, requestedAtespace} { + if requested != "" && !s.watches(requested) { + return nil, serviceerrors.NewPermissionDenied("Not authorized", fmt.Errorf("namespace %q is not watched", requested)) + } + } return s.substrateNamespaces(requestedNamespace), nil } -// 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{}{} - } - } - return allowAll, allowed +// watches reports whether this controller manages the namespace, or the atespace of +// the same name. +func (s *Service) watches(namespace string) bool { + return len(s.observedNamespaces) == 0 || slices.Contains(s.observedNamespaces, namespace) } -// 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 - } - _, ok := allowed[namespace] - return ok +// inNamespace reports whether a worker is in one of the namespaces, or in any when the +// first is empty. A controller that watches every namespace reads exactly one, maybe empty. +func inNamespace(worker *ateapipb.Worker, namespaces []string) bool { + return worker != nil && (namespaces[0] == "" || slices.Contains(namespaces, worker.GetWorkerNamespace())) } 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..b6a84f49d9 --- /dev/null +++ b/go/core/internal/service/system/substrate_paging_test.go @@ -0,0 +1,547 @@ +package system_test + +import ( + "context" + "encoding/base64" + "errors" + "fmt" + "slices" + "strconv" + "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" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + ctrlclient "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + 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"}}}) +} + +// pagingATEClient holds the paging tests' data: three tenants whose rows mix in +// ate-api's order. Every tenant-b name holds "tenant-b", so a test can look for it in +// any text. +func pagingATEClient() *fakeATEClient { + client := &fakeATEClient{ + workers: []*ateapipb.Worker{ + ateWorker("tenant-a", "u1-tenant-a", 2), + ateWorker("tenant-b", "u2-tenant-b", 1), + ateWorker("tenant-c", "u3-tenant-c", 1), + ateWorker("tenant-a", "u4-tenant-a", 0), + ateWorker("tenant-b", "u5-tenant-b", 0), + ateWorker("tenant-a", "u6-tenant-a", 0), + }, + // Two workers a page, so a drain of the workers takes three reads. + workerPageSize: 2, + } + for _, tenant := range []struct { + atespace string + actors int + }{{"tenant-a", 5}, {"tenant-b", 4}, {"tenant-c", 3}} { + for i := 1; i <= tenant.actors; i++ { + client.actors = append(client.actors, namedActor(tenant.atespace, tenant.atespace+"-actor-"+strconv.Itoa(i))) + } + client.templates = append(client.templates, &ateapipb.ActorTemplate{ + Metadata: &ateapipb.ResourceMetadata{Atespace: tenant.atespace, Name: tenant.atespace + "-template"}, + }) + } + return client +} + +// ownWorkerUIDs are the workers of tenant-a and tenant-c, in name order. +var ownWorkerUIDs = []string{"u1-tenant-a", "u3-tenant-c", "u4-tenant-a", "u6-tenant-a"} + +// ateWorker is a Worker as ate-api holds it: named after its pod's UID. +func ateWorker(namespace, uid string, allocatedActors int32) *ateapipb.Worker { + return &ateapipb.Worker{ + Metadata: &ateapipb.ResourceMetadata{Name: uid}, + WorkerNamespace: namespace, + WorkerPod: "pod-" + uid, + WorkerPodUid: uid, + Status: &ateapipb.WorkerStatus{Allocated: &ateapipb.WorkerResources{Actors: allocatedActors}}, + } +} + +// pagingKubeClient holds a WorkerPool in each tenant. +func pagingKubeClient(t *testing.T, funcs interceptor.Funcs) ctrlclient.WithWatch { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + require.NoError(t, atev1alpha1.AddToScheme(scheme)) + objects := []ctrlclient.Object{} + for _, tenant := range []string{"tenant-a", "tenant-b", "tenant-c"} { + objects = append(objects, &atev1alpha1.WorkerPool{ObjectMeta: metav1.ObjectMeta{Namespace: tenant, Name: "pool"}}) + } + return fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).WithInterceptorFuncs(funcs).Build() +} + +// ownWorkerService watches tenant-a and tenant-c. It has no Kubernetes client: the +// workers view reads ate-api only. +func ownWorkerService(ateClient system.ATEClient) *system.Service { + return system.NewService(nil, []string{"tenant-a", "tenant-c"}, &pkgAuth.NoopAuthorizer{}, ateClient) +} + +// walkWorkers reads every page of workers and fails the test on a hard or soft error. +func walkWorkers(t *testing.T, ctx context.Context, service *system.Service, namespace string, limit int32) []system.SubstrateWorkerPage { + t.Helper() + pages := []system.SubstrateWorkerPage{} + pageToken := "" + for range 100 { + page, err := service.ListSubstrateWorkers(ctx, &apiv1.ListSubstrateWorkersRequest{Namespace: namespace, Page: &apiv1.PageRequest{Limit: limit, PageToken: pageToken}}) + require.NoError(t, err) + require.Empty(t, page.ATEAPIError) + pages = append(pages, page) + if pageToken = page.NextPageToken; pageToken == "" { + return pages + } + } + t.Fatal("the walk did not end") + return nil +} + +// listWorkers reads one page of workers in every watched namespace. +func listWorkers(t *testing.T, service *system.Service, limit int32, pageToken string) system.SubstrateWorkerPage { + t.Helper() + page, err := service.ListSubstrateWorkers(pagingContext(t), &apiv1.ListSubstrateWorkersRequest{Page: &apiv1.PageRequest{Limit: limit, PageToken: pageToken}}) + require.NoError(t, err) + return page +} + +func walkedWorkerUIDs(pages []system.SubstrateWorkerPage) []string { + uids := []string{} + for _, page := range pages { + for _, worker := range page.Workers { + uids = append(uids, worker.GetWorkerPodUid()) + } + } + return uids +} + +// ownActorNames is every actor of the watched atespaces, in the order a walk returns them. +func ownActorNames(client *fakeATEClient, watched []string) []string { + names := []string{} + for _, atespace := range slices.Sorted(slices.Values(watched)) { + for _, actor := range client.actors { + if actor.GetMetadata().GetAtespace() == atespace && !slices.Contains(names, actor.GetMetadata().GetName()) { + names = append(names, actor.GetMetadata().GetName()) + } + } + } + return names +} + +// walkActors reads every page of actors and fails the test on a hard or soft error. +func walkActors(t *testing.T, ctx context.Context, service *system.Service, atespace string, limit int32) []system.SubstrateActorPage { + t.Helper() + pages := []system.SubstrateActorPage{} + pageToken := "" + for range 100 { + page, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: atespace, Page: &apiv1.PageRequest{Limit: limit, PageToken: pageToken}}) + require.NoError(t, err) + require.Empty(t, page.ATEAPIError) + pages = append(pages, page) + if pageToken = page.NextPageToken; pageToken == "" { + return pages + } + } + t.Fatal("the walk did not end") + return nil +} + +func walkedActorNames(pages []system.SubstrateActorPage) []string { + names := []string{} + for _, page := range pages { + names = append(names, actorNames(page.Actors)...) + } + return names +} + +func namedActor(atespace, name string) *ateapipb.Actor { + return substrateActor(name, atespace, ateapipb.ActorState_ACTOR_STATE_RUNNING, "", "") +} + +// tokenAtespace reads the atespace a kagent actor page token names. +func tokenAtespace(t *testing.T, token string) string { + raw, err := base64.RawURLEncoding.DecodeString(token) + require.NoError(t, err) + atespace, _, _ := strings.Cut(string(raw), "/") + return atespace +} + +// A page that runs out of rows in one watched atespace goes on in the next, and its +// token names where it stopped. +func TestListSubstrateActorsFanOut(t *testing.T) { + ctx := pagingContext(t) + ateClient := &fakeATEClient{actors: []*ateapipb.Actor{ + namedActor("tenant-a", "a-1"), namedActor("tenant-b", "b-1"), namedActor("tenant-c", "c-1"), + namedActor("tenant-a", "a-2"), namedActor("tenant-c", "c-2"), namedActor("tenant-a", "a-3"), + }} + service := system.NewService(nil, []string{"tenant-a", "tenant-c"}, &pkgAuth.NoopAuthorizer{}, ateClient) + + first, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 2}}) + require.NoError(t, err) + assert.Equal(t, []string{"a-1", "a-2"}, actorNames(first.Actors)) + assert.Equal(t, []ateCall{{method: "ListActorsPage", scope: "tenant-a", size: 2}}, ateClient.calls) + + second, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 2, PageToken: first.NextPageToken}}) + require.NoError(t, err) + assert.Equal(t, []string{"a-3", "c-1"}, actorNames(second.Actors)) + assert.Equal(t, []ateCall{ + {method: "ListActorsPage", scope: "tenant-a", token: "upstream:tenant-a:tenant-a/a-2", size: 2}, + {method: "ListActorsPage", scope: "tenant-c", size: 1}, + }, ateClient.calls[1:]) + assert.Equal(t, "tenant-c", tokenAtespace(t, second.NextPageToken)) +} + +func token(raw string) string { + return base64.RawURLEncoding.EncodeToString([]byte(raw)) +} + +// A token this controller did not issue is a hard InvalidArgument. It never reaches +// ate-api, never starts a walk again at page one, and its error names no tenant-b row. +func TestListSubstrateActorsStrictToken(t *testing.T) { + for name, pageToken := range map[string]string{ + "not base64": "not base64!", + "no separator": token("tenant-a"), + "empty atespace": token("/x"), + "unwatched atespace": token("tenant-b/x"), + // A Substrate actor token, as the released controller passed it through. + "substrate token": token(`{"v":1,"kind":"actor","scope":"","last":["tenant-b","session-1"]}`), + "longer than 1 KiB, else valid": token("tenant-a/" + strings.Repeat("x", 800)), + } { + t.Run(name, func(t *testing.T) { + ateClient := &fakeATEClient{} + service := system.NewService(nil, []string{"tenant-a", "tenant-c"}, &pkgAuth.NoopAuthorizer{}, ateClient) + _, err := service.ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 2, PageToken: pageToken}}) + require.Error(t, err) + assert.True(t, serviceerrors.IsCode(err, serviceerrors.CodeInvalidArgument), "err = %v", err) + assert.NotContains(t, err.Error(), "tenant-b") + assert.Empty(t, ateClient.calls, "a bad token must not reach ate-api") + }) + } +} + +// The token names its atespace, not its place in the watched list, so it resumes in the +// same atespace after the list grows. +func TestListSubstrateActorsTokenHoldsAtespaceName(t *testing.T) { + ctx := pagingContext(t) + first := system.NewService(nil, []string{"tenant-a", "tenant-c"}, &pkgAuth.NoopAuthorizer{}, pagingATEClient()) + // tenant-a has five actors, so a page of six ends one row into tenant-c. + page, err := first.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 6}}) + require.NoError(t, err) + + ateClient := pagingATEClient() + wider := system.NewService(nil, []string{"tenant-a", "tenant-b", "tenant-c"}, &pkgAuth.NoopAuthorizer{}, ateClient) + next, err := wider.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 6, PageToken: page.NextPageToken}}) + require.NoError(t, err) + assert.Equal(t, []string{"tenant-c-actor-2", "tenant-c-actor-3"}, actorNames(next.Actors)) + assert.Equal(t, []string{"tenant-c"}, calledScopes(ateClient, "ListActorsPage")) +} + +func TestListSubstrateActorsNamedAtespace(t *testing.T) { + ctx := pagingContext(t) + ateClient := pagingATEClient() + service := system.NewService(nil, []string{"tenant-a", "tenant-c"}, &pkgAuth.NoopAuthorizer{}, ateClient) + + pages := walkActors(t, ctx, service, "tenant-c", 1) + assert.Equal(t, []string{"tenant-c-actor-1", "tenant-c-actor-2", "tenant-c-actor-3"}, walkedActorNames(pages)) + assert.Equal(t, []string{"tenant-c"}, calledScopes(ateClient, "ListActorsPage")) + for _, page := range pages[:len(pages)-1] { + assert.Equal(t, "tenant-c", tokenAtespace(t, page.NextPageToken)) + } +} + +// A page whose second read fails returns none of the first read's rows: a short page +// beside a warning would look like the end of the list. +func TestListSubstrateActorsLaterCallFails(t *testing.T) { + ateClient := &fakeATEClient{ + actors: []*ateapipb.Actor{namedActor("tenant-a", "a-1"), namedActor("tenant-c", "c-1")}, + failCall: 2, + failErr: errors.New("ate-api unreachable"), + } + service := system.NewService(nil, []string{"tenant-a", "tenant-c"}, &pkgAuth.NoopAuthorizer{}, ateClient) + + page, err := service.ListSubstrateActors(pagingContext(t), &apiv1.ListSubstrateActorsRequest{Page: &apiv1.PageRequest{Limit: 2}}) + require.NoError(t, err) + assert.Equal(t, "ate-api unreachable", page.ATEAPIError) + assert.Empty(t, page.Actors) + assert.Empty(t, page.NextPageToken) +} + +// The ate-api reads of a full walk, counted by hand from the paging data: tenant-a +// has five actors and tenant-c three. +func TestListSubstrateActorsCallCounts(t *testing.T) { + for limit, want := range map[int32]int{1: 8, 2: 5, 3: 4, 5: 2, 50: 2} { + t.Run(strconv.Itoa(int(limit)), func(t *testing.T) { + ateClient := pagingATEClient() + service := system.NewService(nil, []string{"tenant-a", "tenant-c"}, &pkgAuth.NoopAuthorizer{}, ateClient) + walkActors(t, pagingContext(t), service, "", limit) + assert.Len(t, ateClient.calls, want) + }) + } +} + +// ate-api error text can name a worker of another namespace. The user sees a fixed text. +func TestListSubstrateWorkersFixedErrorText(t *testing.T) { + ateClient := pagingATEClient() + ateClient.failCall = 2 // the second page of the drain + ateClient.failErr = status.Error(codes.Internal, "worker u2-tenant-b is on a node that failed") + + page := listWorkers(t, ownWorkerService(ateClient), 0, "") + assert.Equal(t, "ate-api could not list workers: Internal", page.ATEAPIError) + assert.Empty(t, page.Workers, "no partial page") + assert.Empty(t, page.NextPageToken) +} + +// A named namespace narrows the view. A watched namespace with no worker is one empty +// page with no token. +func TestListSubstrateWorkersNamedNamespace(t *testing.T) { + for namespace, want := range map[string][]string{ + "tenant-a": {"u1-tenant-a", "u4-tenant-a", "u6-tenant-a"}, + "tenant-d": {}, + } { + t.Run(namespace, func(t *testing.T) { + service := system.NewService(nil, []string{"tenant-a", "tenant-c", "tenant-d"}, &pkgAuth.NoopAuthorizer{}, pagingATEClient()) + pages := walkWorkers(t, pagingContext(t), service, namespace, 1) + assert.Equal(t, want, walkedWorkerUIDs(pages)) + assert.Len(t, pages, max(1, len(want))) + }) + } +} + +// endlessATEClient answers each worker page with no rows and the next token it is given. +type endlessATEClient struct { + fakeATEClient + next func(read int) string +} + +func (client *endlessATEClient) ListWorkersPage(context.Context, int32, string) ([]*ateapipb.Worker, string, error) { + client.workerReads++ + return nil, client.next(client.workerReads), nil +} + +// ate-api faults that stop a drain: a repeated token, which the drain error quotes, and +// a token that never ends. +func TestListSubstrateWorkersEndlessPages(t *testing.T) { + for name, tc := range map[string]struct { + next func(read int) string + reads int + }{ + "repeated token": {func(int) string { return "upstream:u2-tenant-b" }, 2}, + "endless tokens": {func(read int) string { return "upstream:" + strconv.Itoa(read) }, 10_000}, + } { + t.Run(name, func(t *testing.T) { + ateClient := &endlessATEClient{next: tc.next} + page := listWorkers(t, ownWorkerService(ateClient), 0, "") + assert.Equal(t, "ate-api could not list workers: Unknown", page.ATEAPIError) + assert.Empty(t, page.NextPageToken) + assert.Equal(t, tc.reads, ateClient.workerReads) + }) + } +} + +// A page holds 50 workers by default, in name order even when ate-api's is not. +func TestListSubstrateWorkersDefaultPageSize(t *testing.T) { + ateClient := &fakeATEClient{workerPageSize: 2} + for i := 59; i >= 0; i-- { + ateClient.workers = append(ateClient.workers, ateWorker("tenant-a", fmt.Sprintf("w%02d-tenant-a", i), 0)) + } + page := listWorkers(t, ownWorkerService(ateClient), 0, "") + require.Len(t, page.Workers, 50) + assert.Equal(t, "w00-tenant-a", page.Workers[0].GetWorkerPodUid()) + assert.Equal(t, "w49-tenant-a", page.NextPageToken) +} + +// Any token is a cursor: an actor token, a foreign worker name or a name past the end +// moves it, and no row, token or error names tenant-b. A worker added or deleted +// between pages moves it the same way. +func TestListSubstrateWorkersTokenIsCursor(t *testing.T) { + for name, tc := range map[string]struct { + pageToken string + want []string + }{ + "actor token": {token("tenant-a/upstream:tenant-b:tenant-b/tenant-b-actor-1"), ownWorkerUIDs}, + "foreign worker": {"u2-tenant-b", []string{"u3-tenant-c", "u4-tenant-a", "u6-tenant-a"}}, + "after every row": {"zzz", []string{}}, + } { + t.Run(name, func(t *testing.T) { + page := listWorkers(t, ownWorkerService(pagingATEClient()), 50, tc.pageToken) + assert.Equal(t, tc.want, walkedWorkerUIDs([]system.SubstrateWorkerPage{page})) + assert.Empty(t, page.NextPageToken) + assert.Empty(t, page.ATEAPIError) + }) + } +} + +// newPagingWorkerService is ownWorkerService with WorkerPools for the summary. +func newPagingWorkerService(t *testing.T, ateClient system.ATEClient) *system.Service { + return system.NewService(pagingKubeClient(t, interceptor.Funcs{}), []string{"tenant-a", "tenant-c"}, &pkgAuth.NoopAuthorizer{}, ateClient) +} + +// calledScopes lists the atespace of each call to the method, once each. +func calledScopes(client *fakeATEClient, method string) []string { + scopes := []string{} + for _, call := range client.calls { + if call.method == method && !slices.Contains(scopes, call.scope) { + scopes = append(scopes, call.scope) + } + } + return scopes +} + +// The summary and the actor list read only watched atespaces, and keep only their rows +// even from an ate-api that drops the atespace filter. +func TestSubstrateReadsOnlyWatched(t *testing.T) { + for _, ignoreAtespace := range []bool{false, true} { + t.Run("ignoreAtespace="+strconv.FormatBool(ignoreAtespace), func(t *testing.T) { + ctx := pagingContext(t) + ateClient := pagingATEClient() + ateClient.ignoreAtespace = ignoreAtespace + service := newPagingWorkerService(t, ateClient) + + summary, err := service.GetSubstrateSummary(ctx, "", "") + require.NoError(t, err) + assert.Empty(t, summary.ATEAPIError) + assert.Equal(t, int64(8), summary.ActorCount) + assert.Equal(t, int64(8), summary.RunningActorCount) + assert.Equal(t, []string{"tenant-a-template", "tenant-c-template"}, templateNames(summary.ActorTemplates)) + assert.Equal(t, int64(4), summary.WorkerCount) + assert.Equal(t, int64(2), summary.BusyWorkerCount) + assert.Equal(t, []string{"tenant-a", "tenant-c"}, calledScopes(ateClient, "ListActorsPage")) + assert.Equal(t, []string{"tenant-a", "tenant-c"}, calledScopes(ateClient, "ListActorTemplates")) + assert.Equal(t, 3, ateClient.workerReads, "one drain") + + assert.Equal(t, ownActorNames(ateClient, []string{"tenant-a", "tenant-c"}), walkedActorNames(walkActors(t, ctx, service, "", 100))) + }) + } +} + +func TestGetSubstrateSummaryInputs(t *testing.T) { + ateClient := pagingATEClient() + service := newPagingWorkerService(t, ateClient) + + summary, err := service.GetSubstrateSummary(pagingContext(t), "tenant-a", "tenant-c") + require.NoError(t, err) + assert.Equal(t, int64(3), summary.ActorCount) + assert.Equal(t, int64(3), summary.WorkerCount) + assert.Equal(t, []string{"tenant-c"}, calledScopes(ateClient, "ListActorsPage")) + assert.Equal(t, []string{"tenant-c"}, calledScopes(ateClient, "ListActorTemplates")) + + summary, err = service.GetSubstrateSummary(pagingContext(t), "tenant-c", "") + require.NoError(t, err) + assert.Equal(t, int64(1), summary.WorkerCount) + assert.Equal(t, int64(1), summary.BusyWorkerCount) +} + +// A failed second template list drops the first one's templates too. The other reads +// still count. +func TestGetSubstrateSummaryPartialFailure(t *testing.T) { + ateClient := pagingATEClient() + ateClient.failCall = 2 // the tenant-c template list + ateClient.failErr = errors.New("templates unavailable") + service := newPagingWorkerService(t, ateClient) + + summary, err := service.GetSubstrateSummary(pagingContext(t), "", "") + require.NoError(t, err) + assert.Equal(t, "templates unavailable", summary.ATEAPIError) + assert.Empty(t, summary.ActorTemplates) + assert.Equal(t, int64(8), summary.ActorCount) + assert.Equal(t, int64(4), summary.WorkerCount) +} + +// Walks every RPC with each watched list and page size of the paging data. A walk +// reads, and its tokens name, only watched data. It returns each own row once, and no +// page is empty beside a token: that would tell the caller a hidden row exists. The +// lists cover a repeat, which must not read an atespace twice, and a watched atespace +// with no rows. +func TestSubstrateWalksStayInWatchedNamespaces(t *testing.T) { + for _, watched := range [][]string{{"tenant-a", "tenant-c"}, {"tenant-a"}, {"tenant-c", "tenant-a", "tenant-a"}, {"tenant-a", "empty", "tenant-c"}} { + for _, limit := range []int32{1, 2, 3, 4, 5, 6, 7, 100} { + t.Run(fmt.Sprintf("%v/%d", watched, limit), func(t *testing.T) { + ctx := pagingContext(t) + ateClient := pagingATEClient() + kubeClient := pagingKubeClient(t, interceptor.Funcs{}) + service := system.NewService(kubeClient, watched, &pkgAuth.NoopAuthorizer{}, ateClient) + ownWorkerIDs := []string{} + for _, uid := range ownWorkerUIDs { + if slices.ContainsFunc(watched, func(namespace string) bool { return strings.HasSuffix(uid, namespace) }) { + ownWorkerIDs = append(ownWorkerIDs, uid) + } + } + actorPages := walkActors(t, ctx, service, "", limit) + workerPages := walkWorkers(t, ctx, service, "", limit) + summary, err := service.GetSubstrateSummary(ctx, "", "") + require.NoError(t, err) + + for _, call := range ateClient.calls { + switch call.method { + case "ListActorsPage", "ListActorTemplates": + assert.Contains(t, watched, call.scope, "%+v", call) + case "ListWorkersPage": + // A drain sends only ate-api's own tokens. + assert.Contains(t, []string{"", "upstream:2", "upstream:4"}, call.token, "%+v", call) + default: + t.Errorf("unexpected call %+v", call) + } + } + checkPage := func(rows int, pageToken string) { + // An actor token holds ate-api's inner token as base64 text. + raw, _ := base64.RawURLEncoding.DecodeString(pageToken) + assert.NotContains(t, pageToken+string(raw), "tenant-b") + if rows == 0 { + assert.Empty(t, pageToken) + } + } + for _, page := range actorPages { + checkPage(len(page.Actors), page.NextPageToken) + } + for _, page := range workerPages { + checkPage(len(page.Workers), page.NextPageToken) + } + assert.Equal(t, ownActorNames(ateClient, watched), walkedActorNames(actorPages)) + assert.Equal(t, ownWorkerIDs, walkedWorkerUIDs(workerPages)) + assert.Equal(t, int64(len(ownActorNames(ateClient, watched))), summary.ActorCount) + assert.Equal(t, int64(len(ownWorkerIDs)), summary.WorkerCount) + }) + } + } +} + +// A controller that watches every namespace (the app passes an empty list) answers as +// it did before: ate-api's page, token and error text unchanged. +func TestSubstratePassThroughWhenWatchingAll(t *testing.T) { + ctx := pagingContext(t) + ateClient := pagingATEClient() + service := system.NewService(nil, []string{}, &pkgAuth.NoopAuthorizer{}, ateClient) + + for _, atespace := range []string{"", "tenant-b"} { + page, err := service.ListSubstrateActors(ctx, &apiv1.ListSubstrateActorsRequest{Atespace: atespace, Page: &apiv1.PageRequest{Limit: 1}}) + require.NoError(t, err) + calls := ateClient.calls + assert.Equal(t, ateCall{method: "ListActorsPage", scope: atespace, size: 1}, calls[len(calls)-1]) + assert.Equal(t, "upstream:"+atespace+":"+page.Actors[0].GetMetadata().GetAtespace()+"/"+page.Actors[0].GetMetadata().GetName(), page.NextPageToken) + } + + page := listWorkers(t, service, 1, "upstream:2") + assert.Equal(t, []string{"u3-tenant-c"}, walkedWorkerUIDs([]system.SubstrateWorkerPage{page})) + assert.Equal(t, "upstream:3", page.NextPageToken) + assert.Equal(t, ateCall{method: "ListWorkersPage", token: "upstream:2", size: 1}, ateClient.calls[len(ateClient.calls)-1]) + + ateClient.err = errors.New("worker u2-tenant-b is down") + assert.Equal(t, "worker u2-tenant-b is down", listWorkers(t, service, 1, "").ATEAPIError) +} diff --git a/proto/kagent/api/v1/system.proto b/proto/kagent/api/v1/system.proto index b8e7cbecdf..0eb8c62bbb 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. Only a last page can be empty. 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, in worker name order. A controller that watches all namespaces +// filters one Substrate page, 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; 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..2afe547b0b 100644 --- a/ui/src/api/domain/substrate.ts +++ b/ui/src/api/domain/substrate.ts @@ -94,7 +94,11 @@ export interface Timed { computedAt?: string; } -/** One upstream page; an empty page may still have a continuation token. */ +/** + * One page. Only a last page is empty, except a workers page from a controller that + * watches all namespaces: it filters one upstream page, so it can be empty and 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..492410d558 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. Only a last page can be empty. * * @generated from field: kagent.api.v1.PageResponse page = 4; */ @@ -354,8 +359,10 @@ export const ListSubstrateActorsResponseSchema: GenMessage