Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
da9e1c0
----- DV work begins here
thockin Aug 9, 2026
962210f
--- Prefactoring
thockin Aug 9, 2026
71639b9
Revert errant change
thockin Aug 9, 2026
d899140
Add .gitattributes
thockin Aug 4, 2026
94378a1
--- Pin deps on k8s stuff
thockin Aug 9, 2026
2c64795
Do not reformat code in *any* third_party dir
thockin Aug 2, 2026
bc463bf
Pin k8s codegen deps to v0.37.0-rc.0 in tools
thockin Aug 15, 2026
ab497bb
Run hack/update/go-generate.sh with new deps
thockin Aug 15, 2026
6e78025
Fork k8s.io/code-generator & apimachinery in tools
thockin Aug 15, 2026
c648223
Carry k8s PR 141395 as a patch in tools
thockin Aug 15, 2026
170436a
Pin k8s apimachinery deps to v0.37.0-rc.0 in root
thockin Aug 15, 2026
429304f
Fork k8s.io/code-generator in root
thockin Aug 15, 2026
e7b8cdc
Carry k8s PR 141395 as a patch in root
thockin Aug 15, 2026
bd4f780
--- Main commits: create
thockin Aug 9, 2026
9c92924
Enable validation-gen as a tool
thockin Aug 1, 2026
6827bae
Call validation-gen (no usage yet)
thockin Aug 2, 2026
167974b
Add 2 required/optional tags to ResourceMetadata
thockin Aug 2, 2026
e3196bb
Add DV tags for ResourceMetadata.*
thockin Aug 2, 2026
afe85fd
Add testing for ResourceMetadata validation
thockin Aug 2, 2026
6056822
Add first DV tag to CreateActorRequest
thockin Aug 2, 2026
5581ad3
Enable DV for CreateActor metadata
thockin Aug 2, 2026
cc4926b
Add DV for Actor.status
thockin Aug 3, 2026
600dadd
Move generated validation code
thockin Aug 10, 2026
dac9018
Change "go generate" to a script
thockin Aug 15, 2026
1034e39
WIP: how create might actually work
thockin Aug 14, 2026
e357113
WIP: use ifEnabled/ifDisabled
thockin Aug 15, 2026
cc882bf
DNM: Remove noise from CreateActor in storage
thockin Aug 16, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
The table of contents is too big for display.
Diff view
Diff view
  •  
  •  
  •  
The diff you're trying to view is too large. We only load the first 3000 changed files.
4 changes: 4 additions & 0 deletions .gitattributes
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
# Always check-out / check-in files with LF line endings.
* text=auto eol=lf

**/zz_generated.*.go linguist-generated=true
128 changes: 88 additions & 40 deletions cmd/ateapi/internal/controlapi/actor.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,32 +26,47 @@ import (
"github.com/agent-substrate/substrate/internal/resources"
atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
"github.com/google/uuid"
"go.opentelemetry.io/otel/attribute"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/timestamppb"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/operation"
"k8s.io/apimachinery/pkg/api/validate/content"
"k8s.io/apimachinery/pkg/util/validation/field"
)

func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequest) (created *ateapipb.Actor, err error) {
if errs := validateCreateActorRequest(req); len(errs) > 0 {
// First scrub any fields that users are not allowed to set.
inActor := req.Actor
if inActor != nil { // otherwise validation will flag it
scrubActor(inActor)
}

// Validate the request, including the object within it.
if errs := validateCreateActorRequest(ctx, req); len(errs) > 0 {
return nil, toGRPCStatusError(errs)
}

//
// Handle the request
//

start := time.Now()
in := req.GetActor()
// Recorded only after validation, so every operation uniformly measures a
// validated request; malformed ones stay visible in rpc.server.call.duration.
defer func() {
s.instruments.recordLifecycleOp(ctx, ateattr.OperationCreate, start, err,
ateattr.TemplateNameKey.String(in.GetActorTemplateName()),
ateattr.TemplateNamespaceKey.String(in.GetActorTemplateNamespace()),
ateattr.TemplateNameKey.String(inActor.GetActorTemplateName()),
ateattr.TemplateNamespaceKey.String(inActor.GetActorTemplateNamespace()),
)
}()
templateNamespace := in.GetActorTemplateNamespace()
templateName := in.GetActorTemplateName()
templateNamespace := inActor.GetActorTemplateNamespace()
templateName := inActor.GetActorTemplateName()

setSpanActorRefAttributes(ctx, resources.ActorRefFromActor(in))
setSpanActorRefAttributes(ctx, resources.ActorRefFromActor(inActor))

template, err := s.actorTemplateLister.ActorTemplates(templateNamespace).Get(templateName)
if err != nil {
Expand All @@ -62,15 +77,15 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
}

var sourceSnapshotInfo *ateapipb.ActorSnapshotSource
if src := in.GetSourceSnapshot(); src != nil {
sourceSnapshotInfo, err = s.resolveSnapshotSource(ctx, in.GetMetadata().GetAtespace(), src, template)
if src := inActor.GetSourceSnapshot(); src != nil {
sourceSnapshotInfo, err = s.resolveSnapshotSource(ctx, inActor.GetMetadata().GetAtespace(), src, template)
if err != nil {
return nil, err
}
}

atespace := in.GetMetadata().GetAtespace()
name := in.GetMetadata().GetName()
atespace := inActor.GetMetadata().GetAtespace()
name := inActor.GetMetadata().GetName()

// The atespace must already exist.
exists, err := s.persistence.AtespaceExists(ctx, atespace)
Expand All @@ -87,20 +102,18 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
return nil, err
}

actor := &ateapipb.Actor{
Metadata: &ateapipb.ResourceMetadata{
Atespace: atespace,
Name: name,
},
Status: ateapipb.Actor_STATUS_SUSPENDED,
ActorTemplateNamespace: templateNamespace,
ActorTemplateName: templateName,
WorkerSelector: in.GetWorkerSelector(),
ActorVolumes: initVols,
LatestSnapshot: sourceSnapshotInfo.GetSnapshot(),
SourceSnapshot: sourceSnapshotInfo,
}
stored, err := s.persistence.CreateActor(ctx, actor)
// Verify that the result is properly valid before storing it.
outActor := proto.CloneOf(inActor)
outActor.Status = ateapipb.Actor_STATUS_SUSPENDED
outActor.ActorVolumes = initVols
outActor.LatestSnapshot = sourceSnapshotInfo.GetSnapshot()
outActor.SourceSnapshot = sourceSnapshotInfo
if errs := validateActorUpdate(ctx, outActor, inActor); len(errs) > 0 {
return nil, toGRPCInternalError(errs)
}

// Save the data in the storage layer.
stored, err := s.persistence.CreateActor(ctx, outActor)
if err != nil {
if errors.Is(err, store.ErrAlreadyExists) {
return nil, status.Errorf(codes.AlreadyExists, "Actor %s already exists", name)
Expand All @@ -112,6 +125,40 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
return stored, nil
}

// scrubActor removes any fields from the request that clients are not allowed
// to set.
func scrubActor(actor *ateapipb.Actor) {
// TODO: find a way to do this automatically - proto tags or codegen or something
//FIXME: this is obviously wrong for update
scrubResourceMetadata(actor.Metadata)
actor.Status = 0
actor.WorkerAssignment = nil
actor.InProgressSnapshotName = ""
actor.LatestSnapshot = nil
actor.LocalSnapshotInfo = nil
actor.InProgressSnapshotSourceActorVersion = 0
actor.ActorVolumes = nil
actor.InProgressLocalSnapshotName = ""
// FIXME: SourceSnapshot has mixed input/output fields, so we need smarter scrub
}

// FIXME: put this in a common place for all resources.
// TODO: find a way to do this automatically - proto tags or codegen or something
func scrubResourceMetadata(in *ateapipb.ResourceMetadata) {
if in == nil {
return // validation will flag it
}
now := timestamppb.Now()
*in = ateapipb.ResourceMetadata{
Atespace: in.Atespace,
Name: in.Name,
Uid: uuid.NewString(),
Version: 1,
CreateTime: now,
UpdateTime: now,
}
}

// resolveSnapshotSource resolves a CreateActor request's source snapshot tag
// and checks that its scope and ActorSnapshot are compatible with creating
// an Actor in actorAtespace from template.
Expand Down Expand Up @@ -161,29 +208,20 @@ func (s *Service) resolveSnapshotSource(ctx context.Context, actorAtespace strin
}, nil
}

func validateCreateActorRequest(req *ateapipb.CreateActorRequest) field.ErrorList {
func validateCreateActorRequest(ctx context.Context, req *ateapipb.CreateActorRequest) field.ErrorList {
var fldPath *field.Path
var errs field.ErrorList

// Call the generated validation.
op := operation.Operation{Type: operation.Create, Options: map[string]bool{"validateOutput": false}}
errs := Validate_CreateActorRequest(ctx, op, nil, req, nil)

actor := req.GetActor()
actorPath := fldPath.Child("actor")
if actor == nil {
errs = append(errs, field.Required(actorPath, ""))
// handled by DV
return errs
}

metaPath := actorPath.Child("metadata")
if val, p := actor.GetMetadata().GetAtespace(), metaPath.Child("atespace"); val == "" {
errs = append(errs, field.Required(p, ""))
} else {
errs = append(errs, resources.ValidateResourceName(val, p)...)
}
if val, p := actor.GetMetadata().GetName(), metaPath.Child("name"); val == "" {
errs = append(errs, field.Required(p, ""))
} else {
errs = append(errs, resources.ValidateResourceName(val, p)...)
}

if val, p := actor.GetActorTemplateNamespace(), actorPath.Child("actor_template_namespace"); val == "" {
errs = append(errs, field.Required(p, ""))
} else {
Expand Down Expand Up @@ -468,6 +506,16 @@ func validateSuspendActorRequest(req *ateapipb.SuspendActorRequest) field.ErrorL
return errs
}

func validateActorUpdate(ctx context.Context, newVal, oldVal *ateapipb.Actor) field.ErrorList {
var fldPath *field.Path

// Call the generated validation.
op := operation.Operation{Type: operation.Update, Options: map[string]bool{"validateOutput": true}}
errs := Validate_Actor(ctx, op, fldPath, newVal, oldVal)

return errs
}

func validateSelector(sel *ateapipb.Selector, fldPath *field.Path) field.ErrorList {
var errs field.ErrorList

Expand Down
30 changes: 25 additions & 5 deletions cmd/ateapi/internal/controlapi/actor_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -98,38 +98,58 @@ func TestValidateCreateActorRequest(t *testing.T) {
"missing actor",
&ateapipb.CreateActorRequest{},
field.ErrorList{field.Required(field.NewPath("actor"), "")},
}, {
"missing actor.metadata",
validActor(func(a *ateapipb.Actor) { a.Metadata = nil }),
field.ErrorList{field.Required(field.NewPath("actor", "metadata"), "")},
}, {
"missing actor.metadata.atespace",
validActor(func(a *ateapipb.Actor) { a.Metadata.Atespace = "" }),
field.ErrorList{field.Required(field.NewPath("actor", "metadata", "atespace"), "")},
}, {
"invalid actor.metadata.atespace",
validActor(func(a *ateapipb.Actor) { a.Metadata.Atespace = "NS1" }),
field.ErrorList{field.Invalid(field.NewPath("actor", "metadata", "atespace"), "NS1", "")},
field.ErrorList{field.Invalid(field.NewPath("actor", "metadata", "atespace"), nil, "").WithOrigin("format=k8s-short-name")},
}, {
"missing actor.metadata.name",
validActor(func(a *ateapipb.Actor) { a.Metadata.Name = "" }),
field.ErrorList{field.Required(field.NewPath("actor", "metadata", "name"), "")},
}, {
"invalid actor.metadata.name",
validActor(func(a *ateapipb.Actor) { a.Metadata.Name = "ID1" }),
field.ErrorList{field.Invalid(field.NewPath("actor", "metadata", "name"), "ID1", "")},
field.ErrorList{field.Invalid(field.NewPath("actor", "metadata", "name"), nil, "").WithOrigin("format=k8s-short-name")},
}, {
"missing actor_template_namespace",
validActor(func(a *ateapipb.Actor) { a.ActorTemplateNamespace = "" }),
field.ErrorList{field.Required(field.NewPath("actor", "actor_template_namespace"), "")},
}, {
"invalid actor_template_namespace",
validActor(func(a *ateapipb.Actor) { a.ActorTemplateNamespace = "invalid value" }),
field.ErrorList{field.Invalid(field.NewPath("actor", "actor_template_namespace"), "invalid value", "")},
field.ErrorList{field.Invalid(field.NewPath("actor", "actor_template_namespace"), nil, "")},
}, {
"missing actor_template_name",
validActor(func(a *ateapipb.Actor) { a.ActorTemplateName = "" }),
field.ErrorList{field.Required(field.NewPath("actor", "actor_template_name"), "")},
}, {
"invalid actor_template_name",
validActor(func(a *ateapipb.Actor) { a.ActorTemplateName = "invalid value" }),
field.ErrorList{field.Invalid(field.NewPath("actor", "actor_template_name"), "invalid value", "")},
field.ErrorList{field.Invalid(field.NewPath("actor", "actor_template_name"), nil, "")},
}, {
"unspecified actor.status",
validActor(func(a *ateapipb.Actor) { a.Status = 0 }),
nil,
}, {
"negative actor.status",
validActor(func(a *ateapipb.Actor) { a.Status = -1 }),
field.ErrorList{field.Forbidden(field.NewPath("actor", "status"), "")},
}, {
"valid actor.status",
validActor(func(a *ateapipb.Actor) { a.Status = ateapipb.Actor_STATUS_RUNNING }),
field.ErrorList{field.Forbidden(field.NewPath("actor", "status"), "")},
}, {
"invalid actor.status",
validActor(func(a *ateapipb.Actor) { a.Status = 1234567890 }),
field.ErrorList{field.Forbidden(field.NewPath("actor", "status"), "")},
}, {
"worker_selector with nil match_labels",
validActor(func(a *ateapipb.Actor) { a.WorkerSelector = &ateapipb.Selector{} }),
Expand Down Expand Up @@ -167,7 +187,7 @@ func TestValidateCreateActorRequest(t *testing.T) {
}}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assertValidateErr(t, validateCreateActorRequest(tt.req), tt.want)
assertValidateErr(t, validateCreateActorRequest(context.Background(), tt.req), tt.want)
})
}
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,12 @@
// See the License for the specific language governing permissions and
// limitations under the License.

package ateompb
// Kubernetes codegen tools required this to be in doc.go, no other name will
// work.

//go:generate bash -c "../../../hack/protoc.sh --plugin=protoc-gen-go=$(bash ../../../hack/run-tool.sh --print-bin-path protoc-gen-go) --plugin=protoc-gen-go-grpc=$(bash ../../../hack/run-tool.sh --print-bin-path protoc-gen-go-grpc) --go_out=paths=source_relative:. --go-grpc_out=paths=source_relative:. ateom.proto"
// +k8s:validation-gen=TypesWithSuffix=Request
// +k8s:validation-gen-input=github.com/agent-substrate/substrate/pkg/proto/ateapipb
// +k8s:validation-gen-scheme-registry=nil
// +k8s:validation-gen-deep-equal-func=protoDeepEqual

package controlapi
10 changes: 3 additions & 7 deletions cmd/ateapi/internal/controlapi/functional_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -828,12 +828,8 @@ func TestCreateActor_Success(t *testing.T) {

createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Actor: &ateapipb.Actor{
Metadata: &ateapipb.ResourceMetadata{
Atespace: testAtespace,
Name: "id1",
Uid: "caller-supplied-uid",
Version: 999,
CreateTime: timestamppb.New(time.Unix(1, 0)),
UpdateTime: timestamppb.New(time.Unix(1, 0)),
Atespace: testAtespace,
Name: "id1",
},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
Expand Down Expand Up @@ -3494,7 +3490,7 @@ func TestDeleteAtespace_NotFound(t *testing.T) {

func assertValidateErr(t *testing.T, got field.ErrorList, want field.ErrorList) {
t.Helper()
field.ErrorMatcher{}.ByType().ByField().ByValue().Test(t, want, got)
field.ErrorMatcher{}.ByType().ByField().ByOrigin().Test(t, want, got)
}

// TestSuspendActor_FromPaused suspends a PAUSED actor end-to-end: instead of
Expand Down
17 changes: 17 additions & 0 deletions cmd/ateapi/internal/controlapi/validate.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,26 @@ package controlapi
import (
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/proto"
"k8s.io/apimachinery/pkg/util/validation/field"
)

func toGRPCStatusError(errs field.ErrorList) error {
return status.Error(codes.InvalidArgument, errs.ToAggregate().Error())
}

func toGRPCInternalError(errs field.ErrorList) error {
return status.Error(codes.Internal, errs.ToAggregate().Error())
}

func protoDeepEqual[T any](a, b T) bool {
pa, ok := any(a).(proto.Message)
if !ok {
panic("protoDeepEqual: a is not a proto.Message")
}
pb, ok := any(b).(proto.Message)
if !ok {
panic("protoDeepEqual: b is not a proto.Message")
}
return proto.Equal(pa, pb)
}
Loading
Loading