Make UpdateActor method to follow API guidelines (#742)

#732  

* `UpdateActorRequest` now carries the resource itself + `update_mask`
* `worker_selector` is now applied via the update mask (making it
possible to clear the worker selector)
* `update_mask` is validated against an allowlist of paths. Currently,
only `worker_selector` is allowed.
* Added `uid` and `version` as optional guards. 

I'll update UpdateActorSnapshotTag in a follow-up PR


- [x] Tests pass
- [x] Appropriate changes to documentation are included in the PR
This commit is contained in:
Luiz Oliveira
2026-08-05 16:28:59 -04:00
committed by GitHub
parent 7a9bb4b1f2
commit b11c6e07ff
6 changed files with 614 additions and 328 deletions
+115 -11
View File
@@ -50,6 +50,7 @@ import (
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
"google.golang.org/protobuf/proto" "google.golang.org/protobuf/proto"
"google.golang.org/protobuf/testing/protocmp" "google.golang.org/protobuf/testing/protocmp"
"google.golang.org/protobuf/types/known/fieldmaskpb"
"google.golang.org/protobuf/types/known/timestamppb" "google.golang.org/protobuf/types/known/timestamppb"
corev1 "k8s.io/api/core/v1" corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors" apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -2301,10 +2302,15 @@ func TestUpdateActor_Success(t *testing.T) {
} }
updateResp, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{ updateResp, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{
Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "id1"}, Actor: &ateapipb.Actor{
WorkerSelector: &ateapipb.Selector{ Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "id1"},
MatchLabels: map[string]string{"tier": "paid"}, WorkerSelector: &ateapipb.Selector{
MatchLabels: map[string]string{"tier": "paid"},
},
// Output-only fields outside the mask are ignored.
Status: ateapipb.Actor_STATUS_RUNNING,
}, },
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"worker_selector"}},
}) })
if err != nil { if err != nil {
t.Fatalf("UpdateActor failed: %v", err) t.Fatalf("UpdateActor failed: %v", err)
@@ -2319,8 +2325,7 @@ func TestUpdateActor_Success(t *testing.T) {
MatchLabels: map[string]string{"tier": "paid"}, MatchLabels: map[string]string{"tier": "paid"},
}, },
} }
wantUpdateResp := &ateapipb.UpdateActorResponse{Actor: wantActor} if diff := cmp.Diff(wantActor, updateResp, protocmp.Transform(), ignoreUID, ignoreTimestamps); diff != "" {
if diff := cmp.Diff(wantUpdateResp, updateResp, protocmp.Transform(), ignoreUID, ignoreTimestamps); diff != "" {
t.Errorf("UpdateActor response mismatch (-want +got):\n%s", diff) t.Errorf("UpdateActor response mismatch (-want +got):\n%s", diff)
} }
@@ -2334,12 +2339,105 @@ func TestUpdateActor_Success(t *testing.T) {
} }
} }
// TestUpdateActor_Preconditions verifies the optional version and uid guards
// carried in the embedded resource's metadata.
func TestUpdateActor_Preconditions(t *testing.T) {
ns := namespaceForTest("ns-update-preconditions")
tc := setupTest(t, ns)
defer tc.cleanup()
createTemplate(t, tc, ns)
ctx := context.Background()
createActor := func() *ateapipb.Actor {
t.Helper()
actor, err := tc.client.CreateActor(ctx, &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{
Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: testActorID},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
}})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
}
return actor
}
update := func(meta *ateapipb.ResourceMetadata, tier string) (*ateapipb.Actor, error) {
meta.Atespace, meta.Name = testAtespace, testActorID
return tc.client.UpdateActor(ctx, &ateapipb.UpdateActorRequest{
Actor: &ateapipb.Actor{
Metadata: meta,
WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": tier}},
},
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"worker_selector"}},
})
}
// Delete and recreate the same atespace/name actor, so the first lifecycle's uid
// becomes stale.
staleUID := createActor().GetMetadata().GetUid()
if _, err := tc.client.DeleteActor(ctx, &ateapipb.DeleteActorRequest{
Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: testActorID},
}); err != nil {
t.Fatalf("DeleteActor failed: %v", err)
}
created := createActor()
staleVersion := created.GetMetadata().GetVersion()
uid := created.GetMetadata().GetUid()
if uid == staleUID {
t.Fatalf("recreated actor reused uid %s, want a fresh one", uid)
}
// The uid from the deleted lifecycle must be rejected, even though the
// atespace/name it was observed under still resolves.
_, err := update(&ateapipb.ResourceMetadata{Uid: staleUID}, "other-lifecycle")
assertGrpcError(t, err, codes.Aborted, fmt.Sprintf("Actor %s/%s has uid %s, not %s", testAtespace, testActorID, uid, staleUID))
// An unguarded update is last-writer-wins, and moves the resource past the
// version observed above.
unguarded, err := update(&ateapipb.ResourceMetadata{}, "free")
if err != nil {
t.Fatalf("UpdateActor(no guards) failed: %v", err)
}
currentVersion := unguarded.GetMetadata().GetVersion()
if currentVersion <= staleVersion {
t.Fatalf("version = %d, want greater than %d after an update", currentVersion, staleVersion)
}
if got := unguarded.GetWorkerSelector().GetMatchLabels()["tier"]; got != "free" {
t.Errorf("worker_selector[tier] = %q, want free", got)
}
// The version observed before that write is now stale: rejected rather than
// silently overwriting the concurrent change.
_, err = update(&ateapipb.ResourceMetadata{Version: staleVersion}, "stale")
assertGrpcError(t, err, codes.Aborted, "concurrent update conflict, please retry")
// Both uid and version matching the observed state: the update goes through.
updated, err := update(&ateapipb.ResourceMetadata{Uid: uid, Version: currentVersion}, "paid")
if err != nil {
t.Fatalf("UpdateActor(matching guards) failed: %v", err)
}
if got := updated.GetWorkerSelector().GetMatchLabels()["tier"]; got != "paid" {
t.Errorf("worker_selector[tier] = %q, want paid", got)
}
if updated.GetMetadata().GetVersion() <= currentVersion {
t.Errorf("version = %d, want greater than %d", updated.GetMetadata().GetVersion(), currentVersion)
}
// The guard the client just satisfied is now stale in turn.
_, err = update(&ateapipb.ResourceMetadata{Version: currentVersion}, "free")
assertGrpcError(t, err, codes.Aborted, "concurrent update conflict, please retry")
}
func TestUpdateActor_NotFound(t *testing.T) { func TestUpdateActor_NotFound(t *testing.T) {
ns := namespaceForTest("ns-update-actor-notfound") ns := namespaceForTest("ns-update-actor-notfound")
tc := setupTest(t, ns) tc := setupTest(t, ns)
defer tc.cleanup() defer tc.cleanup()
_, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: "does-not-exist"}}) _, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{
Actor: &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: "does-not-exist"}},
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"worker_selector"}},
})
assertGrpcError(t, err, codes.NotFound, "Actor test-atespace/does-not-exist not found") assertGrpcError(t, err, codes.NotFound, "Actor test-atespace/does-not-exist not found")
} }
@@ -2391,8 +2489,11 @@ func TestResumeActor_ReleasesStaleWorkerWhenPoolBecomesIneligible(t *testing.T)
tc.fakeAtelet.FailRun = nil tc.fakeAtelet.FailRun = nil
if _, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{ if _, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{
Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}, Actor: &ateapipb.Actor{
WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "b"}}, Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: name},
WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "b"}},
},
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"worker_selector"}},
}); err != nil { }); err != nil {
t.Fatalf("UpdateActor failed: %v", err) t.Fatalf("UpdateActor failed: %v", err)
} }
@@ -2605,10 +2706,13 @@ func TestUpdateActor_ReassignsPoolAcrossSuspendResume(t *testing.T) {
} }
if _, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{ if _, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{
Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: name}, Actor: &ateapipb.Actor{
WorkerSelector: &ateapipb.Selector{ Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: name},
MatchLabels: map[string]string{"tier": "b"}, WorkerSelector: &ateapipb.Selector{
MatchLabels: map[string]string{"tier": "b"},
},
}, },
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"worker_selector"}},
}); err != nil { }); err != nil {
t.Fatalf("UpdateActor failed: %v", err) t.Fatalf("UpdateActor failed: %v", err)
} }
+81 -11
View File
@@ -18,20 +18,32 @@ import (
"context" "context"
"errors" "errors"
"fmt" "fmt"
"maps"
"slices"
"github.com/agent-substrate/substrate/cmd/ateapi/internal/store" "github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
"github.com/agent-substrate/substrate/internal/resources" "github.com/agent-substrate/substrate/internal/resources"
"github.com/agent-substrate/substrate/pkg/proto/ateapipb" "github.com/agent-substrate/substrate/pkg/proto/ateapipb"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/fieldmaskpb"
"k8s.io/apimachinery/pkg/util/validation/field" "k8s.io/apimachinery/pkg/util/validation/field"
) )
func (s *Service) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequest) (*ateapipb.UpdateActorResponse, error) { // actorMutableFields maps the Actor field paths a client may name in an
// UpdateActor update_mask to the setter that applies them.
// Every other field is either output-only (server-managed), immutable or
// unsupported (e.g. '*'), and naming one is an error.
var actorMutableFields = map[string]func(dst, src *ateapipb.Actor){
"worker_selector": func(dst, src *ateapipb.Actor) { dst.WorkerSelector = src.GetWorkerSelector() },
}
func (s *Service) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequest) (*ateapipb.Actor, error) {
if errs := validateUpdateActorRequest(req); len(errs) > 0 { if errs := validateUpdateActorRequest(req); len(errs) > 0 {
return nil, toGRPCStatusError(errs) return nil, toGRPCStatusError(errs)
} }
actorRef := resources.ActorRefFromObjectRef(req.GetActor()) in := req.GetActor()
actorRef := resources.ActorRefFromActor(in)
setSpanActorRefAttributes(ctx, actorRef) setSpanActorRefAttributes(ctx, actorRef)
actor, err := s.persistence.GetActor(ctx, actorRef) actor, err := s.persistence.GetActor(ctx, actorRef)
@@ -41,9 +53,20 @@ func (s *Service) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequ
} }
return nil, fmt.Errorf("while getting actor: %w", err) return nil, fmt.Errorf("while getting actor: %w", err)
} }
actor.WorkerSelector = req.GetWorkerSelector()
updated, err := s.persistence.UpdateActor(ctx, actor, actor.GetMetadata().GetVersion()) // UID and version preconditions
if uid := in.GetMetadata().GetUid(); uid != "" && uid != actor.GetMetadata().GetUid() {
return nil, status.Errorf(codes.Aborted, "Actor %s has uid %s, not %s", actorRef, actor.GetMetadata().GetUid(), uid)
}
expectedVersion := actor.GetMetadata().GetVersion()
if version := in.GetMetadata().GetVersion(); version != 0 {
expectedVersion = version
}
applyActorUpdateMask(actor, in, req.GetUpdateMask())
updated, err := s.persistence.UpdateActor(ctx, actor, expectedVersion)
if err != nil { if err != nil {
if errors.Is(err, store.ErrVersionConflict) { if errors.Is(err, store.ErrVersionConflict) {
return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry") return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry")
@@ -52,21 +75,68 @@ func (s *Service) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequ
} }
setSpanActorAttributes(ctx, updated) setSpanActorAttributes(ctx, updated)
return &ateapipb.UpdateActorResponse{Actor: updated}, nil return updated, nil
}
// applyActorUpdateMask copies the masked fields from src onto dst. Fields set on
// src but absent from the mask are ignored, and a masked field that is unset on
// src is cleared on dst.
func applyActorUpdateMask(dst, src *ateapipb.Actor, mask *fieldmaskpb.FieldMask) {
for _, path := range mask.GetPaths() {
apply, ok := actorMutableFields[path]
if ok {
apply(dst, src)
}
}
} }
func validateUpdateActorRequest(req *ateapipb.UpdateActorRequest) field.ErrorList { func validateUpdateActorRequest(req *ateapipb.UpdateActorRequest) field.ErrorList {
var fldPath *field.Path var fldPath *field.Path
var errs field.ErrorList var errs field.ErrorList
if val, fldPath := req.Actor, fldPath.Child("actor"); val == nil { actor := req.GetActor()
errs = append(errs, field.Required(fldPath, "")) actorPath := fldPath.Child("actor")
} else { if actor == nil {
errs = append(errs, resources.ValidateObjectRef(val, fldPath)...) return field.ErrorList{field.Required(actorPath, "")}
} }
if val := req.WorkerSelector; val != nil { // atespace and name identify the resource to update; uid and version are
errs = append(errs, validateSelector(val, fldPath.Child("worker_selector"))...) // optional preconditions.
metaPath := actorPath.Child("metadata")
if atespace, p := actor.GetMetadata().GetAtespace(), metaPath.Child("atespace"); atespace == "" {
errs = append(errs, field.Required(p, ""))
} else {
errs = append(errs, resources.ValidateResourceName(atespace, p)...)
}
if name, p := actor.GetMetadata().GetName(), metaPath.Child("name"); name == "" {
errs = append(errs, field.Required(p, ""))
} else {
errs = append(errs, resources.ValidateResourceName(name, p)...)
}
if uid, p := actor.GetMetadata().GetUid(), metaPath.Child("uid"); uid != "" {
errs = append(errs, resources.ValidateUUID(uid, p)...)
}
if version, p := actor.GetMetadata().GetVersion(), metaPath.Child("version"); version < 0 {
errs = append(errs, field.Invalid(p, version, "must not be negative"))
}
maskPath := fldPath.Child("update_mask")
if paths := req.GetUpdateMask().GetPaths(); len(paths) == 0 {
errs = append(errs, field.Required(maskPath, "must name at least one field to update"))
} else {
supportedMutableFields := slices.Sorted(maps.Keys(actorMutableFields))
for _, path := range paths {
if _, ok := actorMutableFields[path]; !ok {
errs = append(errs, field.NotSupported(maskPath, path, supportedMutableFields))
}
}
}
if selector := actor.GetWorkerSelector(); selector != nil {
errs = append(errs, validateSelector(selector, actorPath.Child("worker_selector"))...)
} }
return errs return errs
@@ -18,76 +18,112 @@ import (
"context" "context"
"testing" "testing"
"github.com/google/go-cmp/cmp"
"go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/attribute"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
"google.golang.org/protobuf/testing/protocmp"
"google.golang.org/protobuf/types/known/fieldmaskpb"
"k8s.io/apimachinery/pkg/util/validation/field" "k8s.io/apimachinery/pkg/util/validation/field"
"github.com/agent-substrate/substrate/cmd/ateapi/internal/store/storetest"
"github.com/agent-substrate/substrate/internal/ateattr" "github.com/agent-substrate/substrate/internal/ateattr"
"github.com/agent-substrate/substrate/pkg/proto/ateapipb" "github.com/agent-substrate/substrate/pkg/proto/ateapipb"
) )
func TestValidateUpdateActorRequest(t *testing.T) { func TestValidateUpdateActorRequest(t *testing.T) {
mutableFields := []string{"worker_selector"}
tests := []struct { tests := []struct {
name string name string
req *ateapipb.UpdateActorRequest req *ateapipb.UpdateActorRequest
want field.ErrorList want field.ErrorList
}{{ }{{
"valid", "valid",
&ateapipb.UpdateActorRequest{Actor: &ateapipb.ObjectRef{Atespace: "ns1", Name: "id1"}}, updateActorReq(),
nil, nil,
}, { }, {
"missing actor", "missing actor",
&ateapipb.UpdateActorRequest{}, &ateapipb.UpdateActorRequest{UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"worker_selector"}}},
field.ErrorList{field.Required(field.NewPath("actor"), "")}, field.ErrorList{field.Required(field.NewPath("actor"), "")},
}, { }, {
"missing actor.atespace", "missing actor.metadata.atespace",
&ateapipb.UpdateActorRequest{Actor: &ateapipb.ObjectRef{Name: "id1"}}, updateActorReq(withMetadata(func(m *ateapipb.ResourceMetadata) { m.Atespace = "" })),
field.ErrorList{field.Required(field.NewPath("actor", "atespace"), "")}, field.ErrorList{field.Required(field.NewPath("actor", "metadata", "atespace"), "")},
}, { }, {
"invalid actor.atespace", "invalid actor.metadata.atespace",
&ateapipb.UpdateActorRequest{Actor: &ateapipb.ObjectRef{Atespace: "NS1", Name: "id1"}}, updateActorReq(withMetadata(func(m *ateapipb.ResourceMetadata) { m.Atespace = "NS1" })),
field.ErrorList{field.Invalid(field.NewPath("actor", "atespace"), "NS1", "")}, field.ErrorList{field.Invalid(field.NewPath("actor", "metadata", "atespace"), "NS1", "")},
}, { }, {
"missing actor.name", "missing actor.metadata.name",
&ateapipb.UpdateActorRequest{Actor: &ateapipb.ObjectRef{Atespace: "ns1"}}, updateActorReq(withMetadata(func(m *ateapipb.ResourceMetadata) { m.Name = "" })),
field.ErrorList{field.Required(field.NewPath("actor", "name"), "")}, field.ErrorList{field.Required(field.NewPath("actor", "metadata", "name"), "")},
}, { }, {
"invalid actor.name", "invalid actor.metadata.name",
&ateapipb.UpdateActorRequest{Actor: &ateapipb.ObjectRef{Atespace: "ns1", Name: "ID1"}}, updateActorReq(withMetadata(func(m *ateapipb.ResourceMetadata) { m.Name = "ID1" })),
field.ErrorList{field.Invalid(field.NewPath("actor", "name"), "ID1", "")}, field.ErrorList{field.Invalid(field.NewPath("actor", "metadata", "name"), "ID1", "")},
}, {
"valid actor.metadata.uid precondition",
updateActorReq(withMetadata(func(m *ateapipb.ResourceMetadata) {
m.Uid = "2a5f8c1e-9b3d-4f7a-8e6c-1d0b4a7f2e93"
})),
nil,
}, {
"invalid actor.metadata.uid precondition",
updateActorReq(withMetadata(func(m *ateapipb.ResourceMetadata) { m.Uid = "not-a-uuid" })),
field.ErrorList{field.Invalid(field.NewPath("actor", "metadata", "uid"), "not-a-uuid", "")},
}, {
"valid actor.metadata.version precondition",
updateActorReq(withMetadata(func(m *ateapipb.ResourceMetadata) { m.Version = 7 })),
nil,
}, {
"negative actor.metadata.version precondition",
updateActorReq(withMetadata(func(m *ateapipb.ResourceMetadata) { m.Version = -1 })),
field.ErrorList{field.Invalid(field.NewPath("actor", "metadata", "version"), int64(-1), "")},
}, {
"missing update_mask",
updateActorReq(func(req *ateapipb.UpdateActorRequest) { req.UpdateMask = nil }),
field.ErrorList{field.Required(field.NewPath("update_mask"), "")},
}, {
"empty update_mask",
updateActorReq(withMaskPaths()),
field.ErrorList{field.Required(field.NewPath("update_mask"), "")},
}, {
"wildcard update_mask",
updateActorReq(withMaskPaths("*")),
field.ErrorList{field.NotSupported(field.NewPath("update_mask"), "*", mutableFields)},
}, {
"output-only field in update_mask",
updateActorReq(withMaskPaths("status")),
field.ErrorList{field.NotSupported(field.NewPath("update_mask"), "status", mutableFields)},
}, {
"immutable field in update_mask",
updateActorReq(withMaskPaths("metadata.name")),
field.ErrorList{field.NotSupported(field.NewPath("update_mask"), "metadata.name", mutableFields)},
}, {
"nested path in update_mask",
updateActorReq(withMaskPaths("worker_selector.match_labels")),
field.ErrorList{field.NotSupported(field.NewPath("update_mask"), "worker_selector.match_labels", mutableFields)},
}, { }, {
"nil worker_selector", "nil worker_selector",
&ateapipb.UpdateActorRequest{Actor: &ateapipb.ObjectRef{Atespace: "ns1", Name: "id1"}, WorkerSelector: nil}, updateActorReq(),
nil, nil,
}, { }, {
"valid worker_selector", "valid worker_selector",
&ateapipb.UpdateActorRequest{ updateActorReq(withSelector(map[string]string{"tier": "1"})),
Actor: &ateapipb.ObjectRef{Atespace: "ns1", Name: "id1"},
WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "1"}},
},
nil, nil,
}, { }, {
"invalid worker_selector label key", "invalid worker_selector label key",
&ateapipb.UpdateActorRequest{ updateActorReq(withSelector(map[string]string{"bad key!": "1"})),
Actor: &ateapipb.ObjectRef{Atespace: "ns1", Name: "id1"}, field.ErrorList{field.Invalid(field.NewPath("actor", "worker_selector", "match_labels").Key("bad key!"), "bad key!", "")},
WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"bad key!": "1"}},
},
field.ErrorList{field.Invalid(field.NewPath("worker_selector", "match_labels").Key("bad key!"), "bad key!", "")},
}, { }, {
"invalid worker_selector label value", "invalid worker_selector label value",
&ateapipb.UpdateActorRequest{ updateActorReq(withSelector(map[string]string{"tier": "not valid!"})),
Actor: &ateapipb.ObjectRef{Atespace: "ns1", Name: "id1"}, field.ErrorList{field.Invalid(field.NewPath("actor", "worker_selector", "match_labels").Key("tier"), "not valid!", "")},
WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "not valid!"}},
},
field.ErrorList{field.Invalid(field.NewPath("worker_selector", "match_labels").Key("tier"), "not valid!", "")},
}, { }, {
"too many worker_selector.match_labels", "too many worker_selector.match_labels",
&ateapipb.UpdateActorRequest{ updateActorReq(withSelector(selectorLabelsOfSize(11))),
Actor: &ateapipb.ObjectRef{Atespace: "ns1", Name: "id1"}, field.ErrorList{field.TooMany(field.NewPath("actor", "worker_selector", "match_labels"), 11, 10)},
WorkerSelector: &ateapipb.Selector{MatchLabels: selectorLabelsOfSize(11)},
},
field.ErrorList{field.TooMany(field.NewPath("worker_selector", "match_labels"), 11, 10)},
}} }}
for _, tt := range tests { for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {
@@ -96,6 +132,74 @@ func TestValidateUpdateActorRequest(t *testing.T) {
} }
} }
func TestApplyActorUpdateMask(t *testing.T) {
selector := &ateapipb.Selector{MatchLabels: map[string]string{"tier": "paid"}}
tests := []struct {
name string
src *ateapipb.Actor
dst *ateapipb.Actor
paths []string
want *ateapipb.Actor
}{{
name: "sets a masked field",
src: &ateapipb.Actor{WorkerSelector: selector},
dst: &ateapipb.Actor{},
paths: []string{"worker_selector"},
want: &ateapipb.Actor{WorkerSelector: selector},
}, {
name: "clears a masked field left unset on src",
src: &ateapipb.Actor{},
dst: &ateapipb.Actor{WorkerSelector: selector},
paths: []string{"worker_selector"},
want: &ateapipb.Actor{},
}, {
name: "ignores fields set on src but absent from the mask",
src: &ateapipb.Actor{Status: ateapipb.Actor_STATUS_RUNNING, WorkerSelector: selector},
dst: &ateapipb.Actor{Status: ateapipb.Actor_STATUS_SUSPENDED},
paths: []string{"worker_selector"},
want: &ateapipb.Actor{Status: ateapipb.Actor_STATUS_SUSPENDED, WorkerSelector: selector},
}, {
// Unreachable through the RPC, which rejects the path during validation.
name: "skips a path outside the mutable set",
src: &ateapipb.Actor{Status: ateapipb.Actor_STATUS_RUNNING},
dst: &ateapipb.Actor{Status: ateapipb.Actor_STATUS_SUSPENDED},
paths: []string{"status"},
want: &ateapipb.Actor{Status: ateapipb.Actor_STATUS_SUSPENDED},
}}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
applyActorUpdateMask(tt.dst, tt.src, &fieldmaskpb.FieldMask{Paths: tt.paths})
if diff := cmp.Diff(tt.want, tt.dst, protocmp.Transform()); diff != "" {
t.Errorf("actor mismatch (-want +got):\n%s", diff)
}
})
}
}
// TestUpdateActor_ClearsMaskedField verifies that naming a field in the mask
// while leaving it unset on the request clears it, which is the whole point of
// requiring an explicit mask.
func TestUpdateActor_ClearsMaskedField(t *testing.T) {
svc, _ := serviceWithActor(t, &ateapipb.Actor{
Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: testActorID},
ActorTemplateNamespace: "ns1",
ActorTemplateName: "tmpl1",
WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "free"}},
})
updated, err := svc.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{
Actor: &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: testActorID}},
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"worker_selector"}},
})
if err != nil {
t.Fatalf("UpdateActor failed: %v", err)
}
if got := updated.GetWorkerSelector(); got != nil {
t.Errorf("worker_selector = %v, want nil after masked clear", got)
}
}
func TestUpdateActor_StampsFullSpanIdentity(t *testing.T) { func TestUpdateActor_StampsFullSpanIdentity(t *testing.T) {
ns := namespaceForTest("ns-span-update") ns := namespaceForTest("ns-span-update")
tc := setupTest(t, ns) tc := setupTest(t, ns)
@@ -114,10 +218,13 @@ func TestUpdateActor_StampsFullSpanIdentity(t *testing.T) {
attrs := recordRootSpanAttrs(t, func(ctx context.Context) { attrs := recordRootSpanAttrs(t, func(ctx context.Context) {
if _, err := tc.service.UpdateActor(ctx, &ateapipb.UpdateActorRequest{ if _, err := tc.service.UpdateActor(ctx, &ateapipb.UpdateActorRequest{
Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: testActorID}, Actor: &ateapipb.Actor{
WorkerSelector: &ateapipb.Selector{ Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: testActorID},
MatchLabels: map[string]string{"env": "prod"}, WorkerSelector: &ateapipb.Selector{
MatchLabels: map[string]string{"env": "prod"},
},
}, },
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"worker_selector"}},
}); err != nil { }); err != nil {
t.Fatalf("UpdateActor: %v", err) t.Fatalf("UpdateActor: %v", err)
} }
@@ -142,7 +249,8 @@ func TestUpdateActor_FailedLookupStampsRefIdentityOnly(t *testing.T) {
attrs := recordRootSpanAttrs(t, func(ctx context.Context) { attrs := recordRootSpanAttrs(t, func(ctx context.Context) {
if _, err := tc.service.UpdateActor(ctx, &ateapipb.UpdateActorRequest{ if _, err := tc.service.UpdateActor(ctx, &ateapipb.UpdateActorRequest{
Actor: &ateapipb.ObjectRef{Atespace: testAtespace, Name: testActorID}, Actor: &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Atespace: testAtespace, Name: testActorID}},
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"worker_selector"}},
}); status.Code(err) != codes.NotFound { }); status.Code(err) != codes.NotFound {
t.Fatalf("UpdateActor(missing) error = %v, want code NotFound", err) t.Fatalf("UpdateActor(missing) error = %v, want code NotFound", err)
} }
@@ -156,3 +264,44 @@ func TestUpdateActor_FailedLookupStampsRefIdentityOnly(t *testing.T) {
} }
} }
} }
// updateActorReq builds a minimal valid UpdateActorRequest, then applies the
// given mutations.
func updateActorReq(mutate ...func(*ateapipb.UpdateActorRequest)) *ateapipb.UpdateActorRequest {
req := &ateapipb.UpdateActorRequest{
Actor: &ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Atespace: "ns1", Name: "id1"}},
UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"worker_selector"}},
}
for _, m := range mutate {
m(req)
}
return req
}
func withMetadata(mutate func(*ateapipb.ResourceMetadata)) func(*ateapipb.UpdateActorRequest) {
return func(req *ateapipb.UpdateActorRequest) { mutate(req.GetActor().GetMetadata()) }
}
func withMaskPaths(paths ...string) func(*ateapipb.UpdateActorRequest) {
return func(req *ateapipb.UpdateActorRequest) { req.UpdateMask = &fieldmaskpb.FieldMask{Paths: paths} }
}
func withSelector(labels map[string]string) func(*ateapipb.UpdateActorRequest) {
return func(req *ateapipb.UpdateActorRequest) {
req.GetActor().WorkerSelector = &ateapipb.Selector{MatchLabels: labels}
}
}
// serviceWithActor seeds one actor in a miniredis-backed store and returns a
// Service over it.
func serviceWithActor(t *testing.T, actor *ateapipb.Actor) (*Service, *ateapipb.Actor) {
t.Helper()
persistence, cleanup := storetest.SetupTestStore(t)
t.Cleanup(cleanup)
created, err := persistence.CreateActor(context.Background(), actor)
if err != nil {
t.Fatalf("Failed to CreateActor: %v", err)
}
return &Service{persistence: persistence}, created
}
File diff suppressed because it is too large Load Diff
+16 -15
View File
@@ -16,6 +16,7 @@ syntax = "proto3";
package ateapi; package ateapi;
import "google/protobuf/field_mask.proto";
import "google/protobuf/timestamp.proto"; import "google/protobuf/timestamp.proto";
option go_package = "github.com/agent-substrate/substrate/pkg/proto/ateapipb"; option go_package = "github.com/agent-substrate/substrate/pkg/proto/ateapipb";
@@ -29,7 +30,7 @@ service Control {
rpc CreateActor(CreateActorRequest) returns (Actor) {} rpc CreateActor(CreateActorRequest) returns (Actor) {}
// Update mutable fields on an existing Actor. // Update mutable fields on an existing Actor.
rpc UpdateActor(UpdateActorRequest) returns (UpdateActorResponse) {} rpc UpdateActor(UpdateActorRequest) returns (Actor) {}
// Suspend a given actor to a new snapshot. // Suspend a given actor to a new snapshot.
rpc SuspendActor(SuspendActorRequest) returns (SuspendActorResponse) {} rpc SuspendActor(SuspendActorRequest) returns (SuspendActorResponse) {}
@@ -47,20 +48,17 @@ service Control {
rpc GetActorSnapshot(GetActorSnapshotRequest) returns (ActorSnapshot) {} rpc GetActorSnapshot(GetActorSnapshotRequest) returns (ActorSnapshot) {}
// List ActorSnapshots. // List ActorSnapshots.
rpc ListActorSnapshots(ListActorSnapshotsRequest) rpc ListActorSnapshots(ListActorSnapshotsRequest) returns (ListActorSnapshotsResponse) {}
returns (ListActorSnapshotsResponse) {}
// Add an Atespace-owned, stable name for an ActorSnapshot. // Add an Atespace-owned, stable name for an ActorSnapshot.
rpc TagActorSnapshot(TagActorSnapshotRequest) returns (ActorSnapshotTag) {} rpc TagActorSnapshot(TagActorSnapshotRequest) returns (ActorSnapshotTag) {}
// Publish or unpublish an ActorSnapshot tag without changing its address. // Publish or unpublish an ActorSnapshot tag without changing its address.
rpc UpdateActorSnapshotTag(UpdateActorSnapshotTagRequest) rpc UpdateActorSnapshotTag(UpdateActorSnapshotTagRequest) returns (ActorSnapshotTag) {}
returns (ActorSnapshotTag) {}
// Delete an ActorSnapshot tag. The snapshot becomes garbage-collectable when // Delete an ActorSnapshot tag. The snapshot becomes garbage-collectable when
// its final tag is deleted. // its final tag is deleted.
rpc DeleteActorSnapshotTag(DeleteActorSnapshotTagRequest) rpc DeleteActorSnapshotTag(DeleteActorSnapshotTagRequest) returns (ActorSnapshotTag) {}
returns (ActorSnapshotTag) {}
// List Workers. // List Workers.
rpc ListWorkers(ListWorkersRequest) returns (ListWorkersResponse) {} rpc ListWorkers(ListWorkersRequest) returns (ListWorkersResponse) {}
@@ -318,15 +316,18 @@ message CreateActorRequest {
// May be called regardless of the actor's current status. // May be called regardless of the actor's current status.
// Changes take effect on the next ResumeActor call. // Changes take effect on the next ResumeActor call.
message UpdateActorRequest { message UpdateActorRequest {
ObjectRef actor = 1; // The actor to update.
// actor.metadata.atespace and actor.metadata.name identify which resource to
// worker_selector replaces the actor's current placement constraint. // update.
// Takes effect on the next ResumeActor call. // actor.metadata.version and actor.metadata.uid are optional preconditions and
Selector worker_selector = 2; // zero values skip the check.
}
message UpdateActorResponse {
Actor actor = 1; Actor actor = 1;
// The set of fields to update. Required.
//
// Only the following fields are supported:
// - worker_selector
google.protobuf.FieldMask update_mask = 2;
} }
message SuspendActorRequest { message SuspendActorRequest {
+5 -5
View File
@@ -64,7 +64,7 @@ type ControlClient interface {
// Create a new Actor deriving from a given ActorTemplate. // Create a new Actor deriving from a given ActorTemplate.
CreateActor(ctx context.Context, in *CreateActorRequest, opts ...grpc.CallOption) (*Actor, error) CreateActor(ctx context.Context, in *CreateActorRequest, opts ...grpc.CallOption) (*Actor, error)
// Update mutable fields on an existing Actor. // Update mutable fields on an existing Actor.
UpdateActor(ctx context.Context, in *UpdateActorRequest, opts ...grpc.CallOption) (*UpdateActorResponse, error) UpdateActor(ctx context.Context, in *UpdateActorRequest, opts ...grpc.CallOption) (*Actor, error)
// Suspend a given actor to a new snapshot. // Suspend a given actor to a new snapshot.
SuspendActor(ctx context.Context, in *SuspendActorRequest, opts ...grpc.CallOption) (*SuspendActorResponse, error) SuspendActor(ctx context.Context, in *SuspendActorRequest, opts ...grpc.CallOption) (*SuspendActorResponse, error)
// Pause a given actor and keep its snapshots on node VM. // Pause a given actor and keep its snapshots on node VM.
@@ -127,9 +127,9 @@ func (c *controlClient) CreateActor(ctx context.Context, in *CreateActorRequest,
return out, nil return out, nil
} }
func (c *controlClient) UpdateActor(ctx context.Context, in *UpdateActorRequest, opts ...grpc.CallOption) (*UpdateActorResponse, error) { func (c *controlClient) UpdateActor(ctx context.Context, in *UpdateActorRequest, opts ...grpc.CallOption) (*Actor, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(UpdateActorResponse) out := new(Actor)
err := c.cc.Invoke(ctx, Control_UpdateActor_FullMethodName, in, out, cOpts...) err := c.cc.Invoke(ctx, Control_UpdateActor_FullMethodName, in, out, cOpts...)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -298,7 +298,7 @@ type ControlServer interface {
// Create a new Actor deriving from a given ActorTemplate. // Create a new Actor deriving from a given ActorTemplate.
CreateActor(context.Context, *CreateActorRequest) (*Actor, error) CreateActor(context.Context, *CreateActorRequest) (*Actor, error)
// Update mutable fields on an existing Actor. // Update mutable fields on an existing Actor.
UpdateActor(context.Context, *UpdateActorRequest) (*UpdateActorResponse, error) UpdateActor(context.Context, *UpdateActorRequest) (*Actor, error)
// Suspend a given actor to a new snapshot. // Suspend a given actor to a new snapshot.
SuspendActor(context.Context, *SuspendActorRequest) (*SuspendActorResponse, error) SuspendActor(context.Context, *SuspendActorRequest) (*SuspendActorResponse, error)
// Pause a given actor and keep its snapshots on node VM. // Pause a given actor and keep its snapshots on node VM.
@@ -347,7 +347,7 @@ func (UnimplementedControlServer) GetActor(context.Context, *GetActorRequest) (*
func (UnimplementedControlServer) CreateActor(context.Context, *CreateActorRequest) (*Actor, error) { func (UnimplementedControlServer) CreateActor(context.Context, *CreateActorRequest) (*Actor, error) {
return nil, status.Error(codes.Unimplemented, "method CreateActor not implemented") return nil, status.Error(codes.Unimplemented, "method CreateActor not implemented")
} }
func (UnimplementedControlServer) UpdateActor(context.Context, *UpdateActorRequest) (*UpdateActorResponse, error) { func (UnimplementedControlServer) UpdateActor(context.Context, *UpdateActorRequest) (*Actor, error) {
return nil, status.Error(codes.Unimplemented, "method UpdateActor not implemented") return nil, status.Error(codes.Unimplemented, "method UpdateActor not implemented")
} }
func (UnimplementedControlServer) SuspendActor(context.Context, *SuspendActorRequest) (*SuspendActorResponse, error) { func (UnimplementedControlServer) SuspendActor(context.Context, *SuspendActorRequest) (*SuspendActorResponse, error) {