trending/substrate
1
0
Fork 0
mirror of https://github.com/agent-substrate/substrate.git synced 2026-11-13 10:43:28 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fold actor name+atespace into ActorRef across actor RPCs

Browse Source 
An actor name is only unique within its atespace, so the two always travel together. Add ActorRef{atespace, name} and use it across the Get/Create/Update/Suspend/Pause/Resume/Delete actor requests in place of separate actor_id + atespace fields. The stored Actor object and Redis key layout are unchanged.
This commit is contained in:
Haven Xia
2026-06-29 14:57:54 -07:00
committed by Tim Hockin
parent b855c3a045
commit db97c5ea13
25 changed files with 520 additions and 602 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/ateapi/internal/controlapi/create_actor.go
+9 -9
View File
@@ -42,15 +42,15 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
}
// The atespace must already exist.
exists, err := s.persistence.AtespaceExists(ctx, req.GetAtespace())
exists, err := s.persistence.AtespaceExists(ctx, req.GetActorRef().GetAtespace())
if err != nil {
return nil, fmt.Errorf("while checking atespace: %w", err)
}
if !exists {
return nil, status.Errorf(codes.FailedPrecondition, "Atespace %s not found", req.GetAtespace())
return nil, status.Errorf(codes.FailedPrecondition, "Atespace %s not found", req.GetActorRef().GetAtespace())
}
id := req.GetActorId()
id := req.GetActorRef().GetName()
actor := &ateapipb.Actor{
ActorId: id,
Version: 1,
@@ -58,7 +58,7 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
ActorTemplateNamespace: req.GetActorTemplateNamespace(),
ActorTemplateName: req.GetActorTemplateName(),
WorkerSelector: req.GetWorkerSelector(),
Atespace: req.GetAtespace(),
Atespace: req.GetActorRef().GetAtespace(),
}
err = s.persistence.CreateActor(ctx, actor)
if err != nil {
@@ -68,7 +68,7 @@ func (s *Service) CreateActor(ctx context.Context, req *ateapipb.CreateActorRequ
return nil, fmt.Errorf("while recording actor: %w", err)
}
storedActor, err := s.persistence.GetActor(ctx, req.GetAtespace(), id)
storedActor, err := s.persistence.GetActor(ctx, req.GetActorRef().GetAtespace(), id)
if err != nil {
return nil, fmt.Errorf("while fetching recorded actor from DB: %w", err)
}
@@ -85,16 +85,16 @@ func validateCreateActorRequest(req *ateapipb.CreateActorRequest) error {
if req.GetActorTemplateName() == "" {
return status.Error(codes.InvalidArgument, "actor_template_name is required")
}
if req.GetActorId() == "" {
if req.GetActorRef().GetName() == "" {
return status.Error(codes.InvalidArgument, "actor_id is required")
}
if err := resources.ValidateActorID(req.GetActorId()); err != nil {
if err := resources.ValidateActorID(req.GetActorRef().GetName()); err != nil {
return status.Error(codes.InvalidArgument, err.Error())
}
if req.GetAtespace() == "" {
if req.GetActorRef().GetAtespace() == "" {
return status.Error(codes.InvalidArgument, "atespace is required")
}
if err := resources.ValidateAtespace(req.GetAtespace()); err != nil {
if err := resources.ValidateAtespace(req.GetActorRef().GetAtespace()); err != nil {
return status.Error(codes.InvalidArgument, err.Error())
}
if err := validateSelector(req.GetWorkerSelector()); err != nil {
cmd/ateapi/internal/controlapi/delete_actor.go
+8 -8
View File
@@ -31,16 +31,16 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ
return nil, err
}
if err := s.persistence.DeleteActor(ctx, req.GetAtespace(), req.GetActorId()); err != nil {
if err := s.persistence.DeleteActor(ctx, req.GetActorRef().GetAtespace(), req.GetActorRef().GetName()); err != nil {
if errors.Is(err, store.ErrNotFound) {
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorId())
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorRef().GetName())
}
if errors.Is(err, store.ErrFailedPrecondition) {
actor, getErr := s.persistence.GetActor(ctx, req.GetAtespace(), req.GetActorId())
actor, getErr := s.persistence.GetActor(ctx, req.GetActorRef().GetAtespace(), req.GetActorRef().GetName())
if getErr == nil {
return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended (status: %v)", req.GetActorId(), actor.GetStatus())
return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended (status: %v)", req.GetActorRef().GetName(), actor.GetStatus())
}
return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended", req.GetActorId())
return nil, status.Errorf(codes.FailedPrecondition, "Actor %s is not suspended", req.GetActorRef().GetName())
}
if errors.Is(err, store.ErrPersistenceRetry) {
return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry")
@@ -52,13 +52,13 @@ func (s *Service) DeleteActor(ctx context.Context, req *ateapipb.DeleteActorRequ
}
func validateDeleteActorRequest(req *ateapipb.DeleteActorRequest) error {
if req.GetActorId() == "" {
if req.GetActorRef().GetName() == "" {
return status.Error(codes.InvalidArgument, "actor_id is required")
}
if err := resources.ValidateActorID(req.GetActorId()); err != nil {
if err := resources.ValidateActorID(req.GetActorRef().GetName()); err != nil {
return status.Error(codes.InvalidArgument, err.Error())
}
if req.GetAtespace() == "" {
if req.GetActorRef().GetAtespace() == "" {
return status.Error(codes.InvalidArgument, "atespace is required")
}
return nil
cmd/ateapi/internal/controlapi/functional_test.go
+106 -158
View File
@@ -638,10 +638,9 @@ func TestCreateActor_Success(t *testing.T) {
createTemplate(t, tc, ns)
createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "free"}},
})
if err != nil {
@@ -672,10 +671,9 @@ func TestCreateActor_TemplateNotFound(t *testing.T) {
defer tc.cleanup()
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "non-existent",
ActorId: "id1",
})
assertGrpcError(t, err, codes.FailedPrecondition, fmt.Sprintf("ActorTemplate %s/non-existent not found", ns))
}
@@ -689,20 +687,18 @@ func TestCreateActor_Duplicate(t *testing.T) {
createTemplate(t, tc, ns)
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("first CreateActor failed: %v", err)
}
_, err = tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
assertGrpcError(t, err, codes.AlreadyExists, "Actor id1 already exists")
}
@@ -716,10 +712,9 @@ func TestGetActor_Found(t *testing.T) {
createTemplate(t, tc, ns)
createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
@@ -728,8 +723,7 @@ func TestGetActor_Found(t *testing.T) {
id := createResp.GetActor().GetActorId()
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
@@ -754,8 +748,7 @@ func TestGetActor_NotFound(t *testing.T) {
defer tc.cleanup()
_, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{
Atespace: testAtespace,
ActorId: "non-existent",
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "non-existent"},
})
assertGrpcError(t, err, codes.NotFound, "Actor non-existent not found")
}
@@ -774,19 +767,17 @@ func TestListActors(t *testing.T) {
createTemplate(t, tc, ns)
resp1, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor 1 failed: %v", err)
}
resp2, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id2"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id2",
})
if err != nil {
t.Fatalf("CreateActor 2 failed: %v", err)
@@ -832,10 +823,9 @@ func TestListActors_ByAtespace(t *testing.T) {
create := func(id, atespace string) *ateapipb.Actor {
resp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: atespace, Name: id},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: id,
Atespace: atespace,
})
if err != nil {
t.Fatalf("CreateActor(%s, atespace=%q) failed: %v", id, atespace, err)
@@ -870,10 +860,10 @@ func TestListActors_ByAtespace(t *testing.T) {
}
// Get is scoped: the right atespace hits, the empty atespace misses (deny-across by key).
if _, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ActorId: "id1", Atespace: "team-a"}); err != nil {
if _, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: "team-a", Name: "id1"}}); err != nil {
t.Errorf("GetActor(id1, team-a) failed: %v", err)
}
_, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{Atespace: testAtespace, ActorId: "id1"})
_, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}})
assertGrpcError(t, err, codes.NotFound, "Actor id1 not found")
}
@@ -890,10 +880,9 @@ func TestListActors_AllAtespaces(t *testing.T) {
create := func(id, atespace string) {
if _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: atespace, Name: id},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: id,
Atespace: atespace,
}); err != nil {
t.Fatalf("CreateActor(%s, atespace=%q) failed: %v", id, atespace, err)
}
@@ -929,10 +918,9 @@ func TestListActors_Pagination(t *testing.T) {
var want []*ateapipb.Actor
for i := 0; i < 5; i++ {
resp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: fmt.Sprintf("id%d", i)},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: fmt.Sprintf("id%d", i),
})
if err != nil {
t.Fatalf("CreateActor %d failed: %v", i, err)
@@ -1060,10 +1048,9 @@ func TestResumeActor(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
@@ -1071,8 +1058,7 @@ func TestResumeActor(t *testing.T) {
id := "id1"
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("ResumeActor failed: %v", err)
@@ -1083,8 +1069,7 @@ func TestResumeActor(t *testing.T) {
}
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
@@ -1182,17 +1167,15 @@ func TestResumeActorResolvesValueFromEnv(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
_, err = tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
}
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: "id1",
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
})
if err != nil {
t.Fatalf("ResumeActor failed: %v", err)
@@ -1238,10 +1221,9 @@ func TestResumeActor_NoWorkers(t *testing.T) {
createTemplate(t, tc, ns)
createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
@@ -1250,8 +1232,7 @@ func TestResumeActor_NoWorkers(t *testing.T) {
id := createResp.GetActor().GetActorId()
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
assertGrpcError(t, err, codes.FailedPrecondition, "no free workers available")
}
@@ -1269,17 +1250,17 @@ func TestResumeActor_NoEligiblePool(t *testing.T) {
MatchLabels: map[string]string{"nonexistent": ns},
})
createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Atespace: testAtespace,
createResp, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
}
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Atespace: testAtespace,
ActorId: createResp.GetActor().GetActorId(),
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: createResp.GetActor().GetActorId()},
})
assertGrpcError(t, err, codes.FailedPrecondition, "no worker pool matches the template's sandboxClass and the template/actor selectors")
}
@@ -1301,10 +1282,10 @@ func TestResumeActor_MultiPoolSelector(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-a", "node1", "pool-a")
createWorkerPod(t, tc, ns, "worker-b", "node1", "pool-b")
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Atespace: testAtespace,
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
WorkerSelector: &ateapipb.Selector{
MatchLabels: map[string]string{"tier": "b"},
},
@@ -1313,12 +1294,12 @@ func TestResumeActor_MultiPoolSelector(t *testing.T) {
t.Fatalf("CreateActor failed: %v", err)
}
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Atespace: testAtespace, ActorId: "id1"})
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}})
if err != nil {
t.Fatalf("ResumeActor failed: %v", err)
}
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{Atespace: testAtespace, ActorId: "id1"})
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
}
@@ -1352,10 +1333,10 @@ func TestResumeActor_RequiresBothSelectorsToMatch(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-template-only", "node1", "pool-template-only")
createWorkerPod(t, tc, ns, "worker-actor-only", "node1", "pool-actor-only")
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Atespace: testAtespace,
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
WorkerSelector: &ateapipb.Selector{
MatchLabels: map[string]string{"tier": "b"},
},
@@ -1364,11 +1345,11 @@ func TestResumeActor_RequiresBothSelectorsToMatch(t *testing.T) {
t.Fatalf("CreateActor failed: %v", err)
}
if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Atespace: testAtespace, ActorId: "id1"}); err != nil {
if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}}); err != nil {
t.Fatalf("ResumeActor failed: %v", err)
}
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{Atespace: testAtespace, ActorId: "id1"})
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
}
@@ -1398,10 +1379,9 @@ func TestResumeActor_Reentrancy(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
@@ -1412,8 +1392,7 @@ func TestResumeActor_Reentrancy(t *testing.T) {
tc.fakeAtelet.FailRestore = fmt.Errorf("mock atelet failure")
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err == nil {
t.Fatalf("expected ResumeActor to fail due to atelet error")
@@ -1433,8 +1412,7 @@ func TestResumeActor_Reentrancy(t *testing.T) {
tc.fakeAtelet.RestoreCalled = false // reset for verification
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("ResumeActor failed on retry: %v", err)
@@ -1474,10 +1452,9 @@ func TestSuspendActor(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
@@ -1486,8 +1463,7 @@ func TestSuspendActor(t *testing.T) {
// Resume first to make it running
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("ResumeActor failed: %v", err)
@@ -1495,8 +1471,7 @@ func TestSuspendActor(t *testing.T) {
// Suspend
_, err = tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("SuspendActor failed: %v", err)
@@ -1507,8 +1482,7 @@ func TestSuspendActor(t *testing.T) {
}
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
@@ -1563,10 +1537,9 @@ func TestPauseActor(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
@@ -1575,8 +1548,7 @@ func TestPauseActor(t *testing.T) {
// Resume first to make it running
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("ResumeActor failed: %v", err)
@@ -1584,8 +1556,7 @@ func TestPauseActor(t *testing.T) {
// Pause
_, err = tc.client.PauseActor(context.Background(), &ateapipb.PauseActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("PauseActor failed: %v", err)
@@ -1596,8 +1567,7 @@ func TestPauseActor(t *testing.T) {
}
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
@@ -1642,10 +1612,10 @@ func TestUpdateActor_Success(t *testing.T) {
createTemplate(t, tc, ns)
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Atespace: testAtespace,
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
WorkerSelector: &ateapipb.Selector{
MatchLabels: map[string]string{"tier": "free"},
},
@@ -1654,8 +1624,8 @@ func TestUpdateActor_Success(t *testing.T) {
t.Fatalf("CreateActor failed: %v", err)
}
updateResp, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{Atespace: testAtespace,
ActorId: "id1",
updateResp, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
WorkerSelector: &ateapipb.Selector{
MatchLabels: map[string]string{"tier": "paid"},
},
@@ -1680,7 +1650,7 @@ func TestUpdateActor_Success(t *testing.T) {
t.Errorf("UpdateActor response mismatch (-want +got):\n%s", diff)
}
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{Atespace: testAtespace, ActorId: "id1"})
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
}
@@ -1695,7 +1665,7 @@ func TestUpdateActor_NotFound(t *testing.T) {
tc := setupTest(t, ns)
defer tc.cleanup()
_, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{Atespace: testAtespace, ActorId: "does-not-exist"})
_, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "does-not-exist"}})
assertGrpcError(t, err, codes.NotFound, "Actor does-not-exist not found")
}
@@ -1727,10 +1697,10 @@ func TestResumeActor_ReleasesStaleWorkerWhenPoolBecomesIneligible(t *testing.T)
createWorkerPod(t, tc, ns, "worker-b", "node1", "pool-b")
id := "id1"
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Atespace: testAtespace,
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: id,
WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "a"}},
})
if err != nil {
@@ -1738,24 +1708,24 @@ func TestResumeActor_ReleasesStaleWorkerWhenPoolBecomesIneligible(t *testing.T)
}
tc.fakeAtelet.FailRun = fmt.Errorf("mock atelet failure")
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Atespace: testAtespace, ActorId: id})
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id}})
if err == nil {
t.Fatalf("expected first ResumeActor (onto worker-a) to fail")
}
tc.fakeAtelet.FailRun = nil
if _, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{Atespace: testAtespace,
ActorId: id,
if _, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "b"}},
}); err != nil {
t.Fatalf("UpdateActor failed: %v", err)
}
if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Atespace: testAtespace, ActorId: id}); err != nil {
if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id}}); err != nil {
t.Fatalf("second ResumeActor failed: %v", err)
}
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{Atespace: testAtespace, ActorId: id})
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id}})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
}
@@ -1815,10 +1785,10 @@ func TestUpdateActor_ReassignsPoolAcrossSuspendResume(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-b", "node1", "pool-b")
id := "id1"
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{Atespace: testAtespace,
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: id,
WorkerSelector: &ateapipb.Selector{
MatchLabels: map[string]string{"tier": "a"},
},
@@ -1827,11 +1797,11 @@ func TestUpdateActor_ReassignsPoolAcrossSuspendResume(t *testing.T) {
t.Fatalf("CreateActor failed: %v", err)
}
if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Atespace: testAtespace, ActorId: id}); err != nil {
if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id}}); err != nil {
t.Fatalf("first ResumeActor failed: %v", err)
}
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{Atespace: testAtespace, ActorId: id})
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id}})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
}
@@ -1842,8 +1812,8 @@ func TestUpdateActor_ReassignsPoolAcrossSuspendResume(t *testing.T) {
t.Fatalf("expected actor to first resume onto worker-a, got ateom_pod_name=%q", got)
}
if _, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{Atespace: testAtespace,
ActorId: id,
if _, err := tc.client.UpdateActor(context.Background(), &ateapipb.UpdateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
WorkerSelector: &ateapipb.Selector{
MatchLabels: map[string]string{"tier": "b"},
},
@@ -1851,14 +1821,14 @@ func TestUpdateActor_ReassignsPoolAcrossSuspendResume(t *testing.T) {
t.Fatalf("UpdateActor failed: %v", err)
}
if _, err := tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{Atespace: testAtespace, ActorId: id}); err != nil {
if _, err := tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id}}); err != nil {
t.Fatalf("SuspendActor failed: %v", err)
}
if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{Atespace: testAtespace, ActorId: id}); err != nil {
if _, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id}}); err != nil {
t.Fatalf("second ResumeActor failed: %v", err)
}
getResp, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{Atespace: testAtespace, ActorId: id})
getResp, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id}})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
}
@@ -1891,37 +1861,37 @@ func TestValidation(t *testing.T) {
}{
{
"missing namespace",
&ateapipb.CreateActorRequest{Atespace: testAtespace, ActorTemplateName: "tmpl1", ActorId: "id1"},
&ateapipb.CreateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}, ActorTemplateName: "tmpl1"},
"actor_template_namespace is required"},
{
"missing template name",
&ateapipb.CreateActorRequest{Atespace: testAtespace, ActorTemplateNamespace: "ns1", ActorId: "id1"},
&ateapipb.CreateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}, ActorTemplateNamespace: "ns1"},
"actor_template_name is required"},
{
"missing actor id",
&ateapipb.CreateActorRequest{Atespace: testAtespace, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1"},
&ateapipb.CreateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace}, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1"},
"actor_id is required"},
{
"invalid actor id (capitals)",
&ateapipb.CreateActorRequest{Atespace: testAtespace, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1", ActorId: "ID1"},
&ateapipb.CreateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "ID1"}, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1"},
"invalid actor_id: must start and end with a lower case alphanumeric character, and consist only of lower case alphanumeric characters or '-'"},
{
"invalid actor id (special chars)",
&ateapipb.CreateActorRequest{Atespace: testAtespace, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1", ActorId: "id_1"},
&ateapipb.CreateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id_1"}, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1"},
"invalid actor_id: must start and end with a lower case alphanumeric character, and consist only of lower case alphanumeric characters or '-'"},
{
"invalid worker_selector label key",
&ateapipb.CreateActorRequest{Atespace: testAtespace, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1", ActorId: "id1", WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"bad key!": "x"}}},
&ateapipb.CreateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1", WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"bad key!": "x"}}},
`invalid worker_selector label key "bad key!": name part must consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character (e.g. 'MyName', or 'my.name', or '123-abc', regex used for validation is '([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9]')`,
},
{
"invalid worker_selector label value",
&ateapipb.CreateActorRequest{Atespace: testAtespace, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1", ActorId: "id1", WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "not valid!"}}},
&ateapipb.CreateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1", WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "not valid!"}}},
`invalid worker_selector label value "not valid!" for key "tier": a valid label must be an empty string or consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character (e.g. 'MyValue', or 'my_value', or '12345', regex used for validation is '(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])?')`,
},
{
"too many worker_selector match_labels",
&ateapipb.CreateActorRequest{Atespace: testAtespace, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1", ActorId: "id1", WorkerSelector: &ateapipb.Selector{MatchLabels: selectorLabelsOfSize(11)}},
&ateapipb.CreateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}, ActorTemplateNamespace: "ns1", ActorTemplateName: "tmpl1", WorkerSelector: &ateapipb.Selector{MatchLabels: selectorLabelsOfSize(11)}},
"worker_selector has 11 match_labels entries, exceeding the limit of 10",
},
}
@@ -1939,7 +1909,7 @@ func TestValidation(t *testing.T) {
req *ateapipb.GetActorRequest
wantMsg string
}{
{"missing id", &ateapipb.GetActorRequest{Atespace: testAtespace}, "id is required"},
{"missing id", &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace}}, "id is required"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
@@ -1955,7 +1925,7 @@ func TestValidation(t *testing.T) {
req *ateapipb.ResumeActorRequest
wantMsg string
}{
{"missing id", &ateapipb.ResumeActorRequest{Atespace: testAtespace}, "id is required"},
{"missing id", &ateapipb.ResumeActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace}}, "id is required"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
@@ -1971,7 +1941,7 @@ func TestValidation(t *testing.T) {
req *ateapipb.SuspendActorRequest
wantMsg string
}{
{"missing id", &ateapipb.SuspendActorRequest{Atespace: testAtespace}, "id is required"},
{"missing id", &ateapipb.SuspendActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace}}, "id is required"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
@@ -1987,20 +1957,20 @@ func TestValidation(t *testing.T) {
req *ateapipb.UpdateActorRequest
wantMsg string
}{
{"missing id", &ateapipb.UpdateActorRequest{Atespace: testAtespace}, "actor_id is required"},
{"missing id", &ateapipb.UpdateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace}}, "actor_id is required"},
{
"invalid worker_selector label key",
&ateapipb.UpdateActorRequest{Atespace: testAtespace, ActorId: "id1", WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"bad key!": "x"}}},
&ateapipb.UpdateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}, WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"bad key!": "x"}}},
`invalid worker_selector label key "bad key!": name part must consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character (e.g. 'MyName', or 'my.name', or '123-abc', regex used for validation is '([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9]')`,
},
{
"invalid worker_selector label value",
&ateapipb.UpdateActorRequest{Atespace: testAtespace, ActorId: "id1", WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "not valid!"}}},
&ateapipb.UpdateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}, WorkerSelector: &ateapipb.Selector{MatchLabels: map[string]string{"tier": "not valid!"}}},
`invalid worker_selector label value "not valid!" for key "tier": a valid label must be an empty string or consist of alphanumeric characters, '-', '_' or '.', and must start and end with an alphanumeric character (e.g. 'MyValue', or 'my_value', or '12345', regex used for validation is '(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])?')`,
},
{
"too many worker_selector match_labels",
&ateapipb.UpdateActorRequest{Atespace: testAtespace, ActorId: "id1", WorkerSelector: &ateapipb.Selector{MatchLabels: selectorLabelsOfSize(11)}},
&ateapipb.UpdateActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"}, WorkerSelector: &ateapipb.Selector{MatchLabels: selectorLabelsOfSize(11)}},
"worker_selector has 11 match_labels entries, exceeding the limit of 10",
},
}
@@ -2018,9 +1988,9 @@ func TestValidation(t *testing.T) {
req *ateapipb.DeleteActorRequest
wantMsg string
}{
{"missing id", &ateapipb.DeleteActorRequest{Atespace: testAtespace}, "actor_id is required"},
{"invalid actor id (capitals)", &ateapipb.DeleteActorRequest{Atespace: testAtespace, ActorId: "ID1"}, "invalid actor_id: must start and end with a lower case alphanumeric character, and consist only of lower case alphanumeric characters or '-'"},
{"invalid actor id (special chars)", &ateapipb.DeleteActorRequest{Atespace: testAtespace, ActorId: "id_1"}, "invalid actor_id: must start and end with a lower case alphanumeric character, and consist only of lower case alphanumeric characters or '-'"},
{"missing id", &ateapipb.DeleteActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace}}, "actor_id is required"},
{"invalid actor id (capitals)", &ateapipb.DeleteActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "ID1"}}, "invalid actor_id: must start and end with a lower case alphanumeric character, and consist only of lower case alphanumeric characters or '-'"},
{"invalid actor id (special chars)", &ateapipb.DeleteActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id_1"}}, "invalid actor_id: must start and end with a lower case alphanumeric character, and consist only of lower case alphanumeric characters or '-'"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
@@ -2041,10 +2011,9 @@ func TestResumeActor_LockConflict(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
@@ -2058,8 +2027,7 @@ func TestResumeActor_LockConflict(t *testing.T) {
errChan := make(chan error, 1)
go func() {
_, err := tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
errChan <- err
}()
@@ -2069,8 +2037,7 @@ func TestResumeActor_LockConflict(t *testing.T) {
// Launch Request B (should fail due to lock conflict)
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
assertGrpcError(t, err, codes.Aborted, "another operation is in progress for this actor")
@@ -2091,10 +2058,9 @@ func TestResumeActor_DanglingWorker(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-a", "node1", "pool1")
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
@@ -2106,8 +2072,7 @@ func TestResumeActor_DanglingWorker(t *testing.T) {
// 3. Call ResumeActor -> Expect failure
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err == nil {
t.Fatalf("expected ResumeActor to fail due to atelet error")
@@ -2115,8 +2080,7 @@ func TestResumeActor_DanglingWorker(t *testing.T) {
// Verify actor state is RESUMING with worker A assigned
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
@@ -2140,8 +2104,7 @@ func TestResumeActor_DanglingWorker(t *testing.T) {
// 8. Call ResumeActor again -> Expect success and picking Worker B!
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("ResumeActor failed on retry: %v", err)
@@ -2175,10 +2138,9 @@ func TestSuspendActor_DanglingWorker(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
@@ -2187,8 +2149,7 @@ func TestSuspendActor_DanglingWorker(t *testing.T) {
// Resume first to make it running
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("ResumeActor failed: %v", err)
@@ -2204,8 +2165,7 @@ func TestSuspendActor_DanglingWorker(t *testing.T) {
}
_, err = tc.client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("SuspendActor failed: %v", err)
@@ -2213,8 +2173,7 @@ func TestSuspendActor_DanglingWorker(t *testing.T) {
// 4. Verify it becomes SUSPENDED in Redis
getResp, err := tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{
Atespace: testAtespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: id},
})
if err != nil {
t.Fatalf("GetActor failed: %v", err)
@@ -2235,26 +2194,23 @@ func TestDeleteActor_Success(t *testing.T) {
createTemplate(t, tc, ns)
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
}
_, err = tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{
Atespace: testAtespace,
ActorId: "id1",
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
})
if err != nil {
t.Fatalf("DeleteActor failed: %v", err)
}
_, err = tc.client.GetActor(context.Background(), &ateapipb.GetActorRequest{
Atespace: testAtespace,
ActorId: "id1",
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
})
assertGrpcError(t, err, codes.NotFound, "Actor id1 not found")
}
@@ -2268,26 +2224,23 @@ func TestDeleteActor_NotSuspended(t *testing.T) {
createWorkerPod(t, tc, ns, "worker-1", "node1", "pool1")
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
Atespace: testAtespace,
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
})
if err != nil {
t.Fatalf("CreateActor failed: %v", err)
}
_, err = tc.client.ResumeActor(context.Background(), &ateapipb.ResumeActorRequest{
Atespace: testAtespace,
ActorId: "id1",
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
})
if err != nil {
t.Fatalf("ResumeActor failed: %v", err)
}
_, err = tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{
Atespace: testAtespace,
ActorId: "id1",
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "id1"},
})
assertGrpcError(t, err, codes.FailedPrecondition, "Actor id1 is not suspended (status: STATUS_RUNNING)")
}
@@ -2298,8 +2251,7 @@ func TestDeleteActor_NotFound(t *testing.T) {
defer tc.cleanup()
_, err := tc.client.DeleteActor(context.Background(), &ateapipb.DeleteActorRequest{
Atespace: testAtespace,
ActorId: "non-existent",
ActorRef: &ateapipb.ActorRef{Atespace: testAtespace, Name: "non-existent"},
})
assertGrpcError(t, err, codes.NotFound, "Actor non-existent not found")
}
@@ -2330,10 +2282,9 @@ func TestCreateActor_AtespaceNotFound(t *testing.T) {
// The template exists, but "missing-as" was never created. The template
// check fires first, so reaching this error proves the atespace check ran.
_, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: "missing-as", Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
Atespace: "missing-as",
})
assertGrpcError(t, err, codes.FailedPrecondition, "Atespace missing-as not found")
}
@@ -2355,10 +2306,9 @@ func TestCreateAtespace_Success(t *testing.T) {
// An actor can now be created into the new atespace.
if _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: "team-a", Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
Atespace: "team-a",
}); err != nil {
t.Errorf("CreateActor into freshly created atespace failed: %v", err)
}
@@ -2472,10 +2422,9 @@ func TestDeleteAtespace_NonEmpty_Rejected(t *testing.T) {
t.Fatalf("CreateAtespace failed: %v", err)
}
if _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: "team-a", Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
Atespace: "team-a",
}); err != nil {
t.Fatalf("CreateActor failed: %v", err)
}
@@ -2500,10 +2449,9 @@ func TestDeleteAtespace_ScopedToTargetAtespace(t *testing.T) {
// Actor only in team-b.
if _, err := tc.client.CreateActor(context.Background(), &ateapipb.CreateActorRequest{
ActorRef: &ateapipb.ActorRef{Atespace: "team-b", Name: "id1"},
ActorTemplateNamespace: ns,
ActorTemplateName: "tmpl1",
ActorId: "id1",
Atespace: "team-b",
}); err != nil {
t.Fatalf("CreateActor failed: %v", err)
}
cmd/ateapi/internal/controlapi/get_actor.go
+4 -4
View File
@@ -29,9 +29,9 @@ func (s *Service) GetActor(ctx context.Context, req *ateapipb.GetActorRequest) (
if err := validateGetActorRequest(req); err != nil {
return nil, err
}
actor, err := s.persistence.GetActor(ctx, req.GetAtespace(), req.GetActorId())
actor, err := s.persistence.GetActor(ctx, req.GetActorRef().GetAtespace(), req.GetActorRef().GetName())
if errors.Is(err, store.ErrNotFound) {
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorId())
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorRef().GetName())
} else if err != nil {
return nil, fmt.Errorf("while getting actor from DB: %w", err)
}
@@ -41,10 +41,10 @@ func (s *Service) GetActor(ctx context.Context, req *ateapipb.GetActorRequest) (
}
func validateGetActorRequest(req *ateapipb.GetActorRequest) error {
if req.GetActorId() == "" {
if req.GetActorRef().GetName() == "" {
return status.Error(codes.InvalidArgument, "id is required")
}
if req.GetAtespace() == "" {
if req.GetActorRef().GetAtespace() == "" {
return status.Error(codes.InvalidArgument, "atespace is required")
}
return nil
cmd/ateapi/internal/controlapi/pause_actor.go
+4 -4
View File
@@ -29,13 +29,13 @@ func (s *Service) PauseActor(ctx context.Context, req *ateapipb.PauseActorReques
return nil, err
}
actor, err := s.actorWorkflow.PauseActor(ctx, req.GetAtespace(), req.GetActorId())
actor, err := s.actorWorkflow.PauseActor(ctx, req.GetActorRef().GetAtespace(), req.GetActorRef().GetName())
if err != nil {
if errors.Is(err, store.ErrPersistenceRetry) {
return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry")
}
if errors.Is(err, store.ErrNotFound) {
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorId())
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorRef().GetName())
}
return nil, err
}
@@ -44,10 +44,10 @@ func (s *Service) PauseActor(ctx context.Context, req *ateapipb.PauseActorReques
}
func validatePauseActorRequest(req *ateapipb.PauseActorRequest) error {
if req.GetActorId() == "" {
if req.GetActorRef().GetName() == "" {
return status.Error(codes.InvalidArgument, "id is required")
}
if req.GetAtespace() == "" {
if req.GetActorRef().GetAtespace() == "" {
return status.Error(codes.InvalidArgument, "atespace is required")
}
return nil
cmd/ateapi/internal/controlapi/resume_actor.go
+4 -4
View File
@@ -29,13 +29,13 @@ func (s *Service) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequ
return nil, err
}
actor, err := s.actorWorkflow.ResumeActor(ctx, req.GetAtespace(), req.GetActorId(), req.GetBoot())
actor, err := s.actorWorkflow.ResumeActor(ctx, req.GetActorRef().GetAtespace(), req.GetActorRef().GetName(), req.GetBoot())
if err != nil {
if errors.Is(err, store.ErrPersistenceRetry) {
return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry")
}
if errors.Is(err, store.ErrNotFound) {
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorId())
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorRef().GetName())
}
return nil, err
}
@@ -44,10 +44,10 @@ func (s *Service) ResumeActor(ctx context.Context, req *ateapipb.ResumeActorRequ
}
func validateResumeActorRequest(req *ateapipb.ResumeActorRequest) error {
if req.GetActorId() == "" {
if req.GetActorRef().GetName() == "" {
return status.Error(codes.InvalidArgument, "id is required")
}
if req.GetAtespace() == "" {
if req.GetActorRef().GetAtespace() == "" {
return status.Error(codes.InvalidArgument, "atespace is required")
}
return nil
cmd/ateapi/internal/controlapi/suspend_actor.go
+4 -4
View File
@@ -29,13 +29,13 @@ func (s *Service) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRe
return nil, err
}
actor, err := s.actorWorkflow.SuspendActor(ctx, req.GetAtespace(), req.GetActorId())
actor, err := s.actorWorkflow.SuspendActor(ctx, req.GetActorRef().GetAtespace(), req.GetActorRef().GetName())
if err != nil {
if errors.Is(err, store.ErrPersistenceRetry) {
return nil, status.Error(codes.Aborted, "concurrent update conflict, please retry")
}
if errors.Is(err, store.ErrNotFound) {
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorId())
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorRef().GetName())
}
return nil, err
}
@@ -44,10 +44,10 @@ func (s *Service) SuspendActor(ctx context.Context, req *ateapipb.SuspendActorRe
}
func validateSuspendActorRequest(req *ateapipb.SuspendActorRequest) error {
if req.GetActorId() == "" {
if req.GetActorRef().GetName() == "" {
return status.Error(codes.InvalidArgument, "id is required")
}
if req.GetAtespace() == "" {
if req.GetActorRef().GetAtespace() == "" {
return status.Error(codes.InvalidArgument, "atespace is required")
}
return nil
cmd/ateapi/internal/controlapi/update_actor.go
+5 -5
View File
@@ -31,10 +31,10 @@ func (s *Service) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequ
return nil, err
}
actor, err := s.persistence.GetActor(ctx, req.GetAtespace(), req.GetActorId())
actor, err := s.persistence.GetActor(ctx, req.GetActorRef().GetAtespace(), req.GetActorRef().GetName())
if err != nil {
if errors.Is(err, store.ErrNotFound) {
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorId())
return nil, status.Errorf(codes.NotFound, "Actor %s not found", req.GetActorRef().GetName())
}
return nil, fmt.Errorf("while getting actor: %w", err)
}
@@ -51,13 +51,13 @@ func (s *Service) UpdateActor(ctx context.Context, req *ateapipb.UpdateActorRequ
}
func validateUpdateActorRequest(req *ateapipb.UpdateActorRequest) error {
if req.GetActorId() == "" {
if req.GetActorRef().GetName() == "" {
return status.Error(codes.InvalidArgument, "actor_id is required")
}
if req.GetAtespace() == "" {
if req.GetActorRef().GetAtespace() == "" {
return status.Error(codes.InvalidArgument, "atespace is required")
}
if err := resources.ValidateAtespace(req.GetAtespace()); err != nil {
if err := resources.ValidateAtespace(req.GetActorRef().GetAtespace()); err != nil {
return status.Error(codes.InvalidArgument, err.Error())
}
if err := validateSelector(req.GetWorkerSelector()); err != nil {
cmd/atecontroller/internal/controllers/actortemplate_controller.go
+3 -6
View File
@@ -88,8 +88,7 @@ func (r *ActorTemplateReconciler) Reconcile(ctx context.Context, req ctrl.Reques
}
createReq := &ateapipb.CreateActorRequest{
ActorId: actorID,
Atespace: at.ObjectMeta.Namespace,
ActorRef: &ateapipb.ActorRef{Atespace: at.ObjectMeta.Namespace, Name: actorID},
ActorTemplateNamespace: at.ObjectMeta.Namespace,
ActorTemplateName: at.ObjectMeta.Name,
}
@@ -118,8 +117,7 @@ func (r *ActorTemplateReconciler) Reconcile(ctx context.Context, req ctrl.Reques
// TODO: Maybe this should go through a different RPC dedicated to
// booting an actor from scratch.
resumeReq := &ateapipb.ResumeActorRequest{
ActorId: at.Status.GoldenActorID,
Atespace: at.ObjectMeta.Namespace,
ActorRef: &ateapipb.ActorRef{Atespace: at.ObjectMeta.Namespace, Name: at.Status.GoldenActorID},
}
_, err := r.AteClient.ResumeActor(ctx, resumeReq)
if err != nil {
@@ -145,8 +143,7 @@ func (r *ActorTemplateReconciler) Reconcile(ctx context.Context, req ctrl.Reques
// from it.
req := &ateapipb.SuspendActorRequest{
ActorId: at.Status.GoldenActorID,
Atespace: at.ObjectMeta.Namespace,
ActorRef: &ateapipb.ActorRef{Atespace: at.ObjectMeta.Namespace, Name: at.Status.GoldenActorID},
}
resp, err := r.AteClient.SuspendActor(ctx, req)
if err != nil {
cmd/atenet/internal/router/extproc_test.go
+2 -2
View File
@@ -179,8 +179,8 @@ func TestExtProcHeadersEvaluation(t *testing.T) {
t.Run(tc.name, func(t *testing.T) {
clientMock := &mockClient{
resumeFn: func(ctx context.Context, in *ateapipb.ResumeActorRequest, opts ...grpc.CallOption) (*ateapipb.ResumeActorResponse, error) {
if in.GetActorId() != testUUID {
t.Errorf("unexpected identifier parsed in test context: %s", in.GetActorId())
if in.GetActorRef().GetName() != testUUID {
t.Errorf("unexpected identifier parsed in test context: %s", in.GetActorRef().GetName())
}
if tc.resumeErr != nil {
return nil, tc.resumeErr
cmd/atenet/internal/router/resumer.go
+1 -2
View File
@@ -70,8 +70,7 @@ func (r *ActorResumer) ResumeActor(ctx context.Context, atespace, actorID string
err := wait.ExponentialBackoffWithContext(bgCtx, backoff, func(ctx context.Context) (bool, error) {
var err error
resumeResp, err = r.apiClient.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
ActorId: actorID,
Atespace: atespace,
ActorRef: &ateapipb.ActorRef{Atespace: atespace, Name: actorID},
})
if err == nil {
return true, nil
cmd/kubectl-ate/internal/cmd/create_actor.go
+1 -2
View File
@@ -48,8 +48,7 @@ var createActorCmd = &cobra.Command{
resp, err := apiClient.CreateActor(ctx, &ateapipb.CreateActorRequest{
ActorTemplateNamespace: parts[0],
ActorTemplateName: parts[1],
ActorId: actorID,
Atespace: atespaceFlag,
ActorRef: &ateapipb.ActorRef{Atespace: atespaceFlag, Name: actorID},
})
if err != nil {
return fmt.Errorf("failed to create actor: %w", err)
cmd/kubectl-ate/internal/cmd/delete_actor.go
+1 -2
View File
@@ -38,8 +38,7 @@ var deleteActorCmd = &cobra.Command{
id := args[0]
_, err = c.ControlClient.DeleteActor(ctx, &ateapipb.DeleteActorRequest{
ActorId: id,
Atespace: deleteAtespaceFlag,
ActorRef: &ateapipb.ActorRef{Atespace: deleteAtespaceFlag, Name: id},
})
if err != nil {
return err
cmd/kubectl-ate/internal/cmd/get_actors.go
+1 -1
View File
@@ -52,7 +52,7 @@ var getActorsCmd = &cobra.Command{
if getActorsAtespaceFlag == "" {
return fmt.Errorf("--atespace is required when getting a specific actor")
}
resp, err := apiClient.GetActor(ctx, &ateapipb.GetActorRequest{ActorId: args[0], Atespace: getActorsAtespaceFlag})
resp, err := apiClient.GetActor(ctx, &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: getActorsAtespaceFlag, Name: args[0]}})
if err != nil {
return fmt.Errorf("failed to get actor: %w", err)
}
cmd/kubectl-ate/internal/cmd/logs_actors.go
+3 -3
View File
@@ -109,7 +109,7 @@ func (r *LogsActorRunner) Run(ctx context.Context, actorID string) error {
}
func (r *LogsActorRunner) runOneShot(ctx context.Context, actorID string) error {
actorResp, err := r.apiClient.GetActor(ctx, &ateapipb.GetActorRequest{ActorId: actorID, Atespace: r.atespace})
actorResp, err := r.apiClient.GetActor(ctx, &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: r.atespace, Name: actorID}})
if err != nil {
return fmt.Errorf("failed to get actor: %w", err)
}
@@ -156,7 +156,7 @@ func (r *LogsActorRunner) runFollow(ctx context.Context, actorID string) error {
default:
}
actorResp, err := r.apiClient.GetActor(ctx, &ateapipb.GetActorRequest{ActorId: actorID, Atespace: r.atespace})
actorResp, err := r.apiClient.GetActor(ctx, &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: r.atespace, Name: actorID}})
if err != nil {
if status.Code(err) == codes.NotFound {
return fmt.Errorf("actor %s not found: %w", actorID, err)
@@ -264,7 +264,7 @@ func (r *LogsActorRunner) startMigrationMonitor(
case <-ctx.Done():
return
case <-ticker.C:
resp, err := r.apiClient.GetActor(ctx, &ateapipb.GetActorRequest{ActorId: actorID, Atespace: r.atespace})
resp, err := r.apiClient.GetActor(ctx, &ateapipb.GetActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: r.atespace, Name: actorID}})
if err == nil {
act := resp.GetActor()
if act.GetStatus() != ateapipb.Actor_STATUS_RUNNING || act.GetAteomPodName() != currentPod {
cmd/kubectl-ate/internal/cmd/logs_actors_test.go
+2 -2
View File
@@ -187,8 +187,8 @@ func TestLogsActorRunner_Run_OneShotSuccess(t *testing.T) {
mockAPI := &mockAteAPIClient{
GetActorFunc: func(ctx context.Context, in *ateapipb.GetActorRequest, opts ...grpc.CallOption) (*ateapipb.GetActorResponse, error) {
if in.ActorId != actorID {
return nil, fmt.Errorf("unexpected actor ID: %s", in.ActorId)
if in.GetActorRef().GetName() != actorID {
return nil, fmt.Errorf("unexpected actor ID: %s", in.GetActorRef().GetName())
}
return &ateapipb.GetActorResponse{
Actor: &ateapipb.Actor{
cmd/kubectl-ate/internal/cmd/pause_actor.go
+1 -2
View File
@@ -38,8 +38,7 @@ var pauseActorCmd = &cobra.Command{
defer apiClient.Close()
resp, err := apiClient.PauseActor(ctx, &ateapipb.PauseActorRequest{
ActorId: args[0],
Atespace: pauseAtespaceFlag,
ActorRef: &ateapipb.ActorRef{Atespace: pauseAtespaceFlag, Name: args[0]},
})
if err != nil {
return fmt.Errorf("failed to pause actor: %w", err)
cmd/kubectl-ate/internal/cmd/resume_actor.go
+1 -2
View File
@@ -39,9 +39,8 @@ var resumeActorCmd = &cobra.Command{
defer apiClient.Close()
resp, err := apiClient.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
ActorId: args[0],
ActorRef: &ateapipb.ActorRef{Atespace: resumeAtespaceFlag, Name: args[0]},
Boot: bootFlag,
Atespace: resumeAtespaceFlag,
})
if err != nil {
return fmt.Errorf("failed to resume actor: %w", err)
cmd/kubectl-ate/internal/cmd/suspend_actor.go
+1 -2
View File
@@ -38,8 +38,7 @@ var suspendActorCmd = &cobra.Command{
defer apiClient.Close()
resp, err := apiClient.SuspendActor(ctx, &ateapipb.SuspendActorRequest{
ActorId: args[0],
Atespace: suspendAtespaceFlag,
ActorRef: &ateapipb.ActorRef{Atespace: suspendAtespaceFlag, Name: args[0]},
})
if err != nil {
return fmt.Errorf("failed to suspend actor: %w", err)
demos/agent-secret/main.go
+1 -1
View File
@@ -131,7 +131,7 @@ func suspendSelf(id string) {
client := ateapipb.NewControlClient(conn)
log.Printf("Yielding compute. Requesting self-suspension for actor %s...", id)
_, err = client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{ActorId: id})
_, err = client.SuspendActor(context.Background(), &ateapipb.SuspendActorRequest{ActorRef: &ateapipb.ActorRef{Name: id}})
if err != nil {
log.Printf("Failed to self-suspend: %v", err)
}
demos/sandbox/client/main.go
+2 -2
View File
@@ -94,7 +94,7 @@ func main() {
defer conn.Close()
log.Printf("Resuming actor %s...", *actorID)
_, err = cli.ResumeActor(ctx, &ateapipb.ResumeActorRequest{ActorId: *actorID, Atespace: *atespace})
_, err = cli.ResumeActor(ctx, &ateapipb.ResumeActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: *atespace, Name: *actorID}})
if err != nil {
log.Fatalf("Failed to resume actor: %v", err)
}
@@ -104,7 +104,7 @@ func main() {
defer func() {
log.Printf("Suspending actor %s...", *actorID)
suspendCtx := context.Background()
_, err := cli.SuspendActor(suspendCtx, &ateapipb.SuspendActorRequest{ActorId: *actorID, Atespace: *atespace})
_, err := cli.SuspendActor(suspendCtx, &ateapipb.SuspendActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: *atespace, Name: *actorID}})
if err != nil {
log.Printf("Failed to suspend actor: %v", err)
} else {
internal/e2e/suites/demo/demo_test.go
+30 -44
View File
@@ -141,6 +141,9 @@ func TestDurableDirLifecycle(t *testing.T) {
ctx := context.Background()
clients := e2e.GetClients()
// CreateActor requires the atespace to exist first.
_, _ = clients.SubstrateAPI.CreateAtespace(ctx, &ateapipb.CreateAtespaceRequest{Name: demoAtespace})
// Create actor template.
at, err := createActorTemplate(ctx, t, clients, nsObj, tc.onCommit, tc.onPause)
if err != nil {
@@ -154,7 +157,7 @@ func TestDurableDirLifecycle(t *testing.T) {
t.Logf("Creating Actor %q using Substrate API...", actorID)
createResp, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
ActorTemplateNamespace: nsObj.Name,
ActorTemplateName: at.Name,
})
@@ -164,20 +167,20 @@ func TestDurableDirLifecycle(t *testing.T) {
t.Logf("Successfully created Actor: %s", createResp.GetActor().GetActorId())
defer func() {
clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
})
}()
// Resuming the actor
t.Logf("Resuming Actor %q...", actorID)
if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to resume Actor: %v", err)
}
waitForActorStatus(ctx, t, clients, actorID, ateapipb.Actor_STATUS_RUNNING)
resp, err := callActor(t, actorID)
resp, err := callActor(t, demoAtespace, actorID)
if err != nil {
t.Fatalf("failed to call actor: %v", err)
}
@@ -188,7 +191,7 @@ func TestDurableDirLifecycle(t *testing.T) {
//
t.Logf("Pausing Actor %q...", actorID)
if _, err := clients.SubstrateAPI.PauseActor(ctx, &ateapipb.PauseActorRequest{
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to pause Actor: %v", err)
}
@@ -197,13 +200,13 @@ func TestDurableDirLifecycle(t *testing.T) {
// Resuming the actor
t.Logf("Resuming Actor %q again...", actorID)
if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to resume Actor again: %v", err)
}
waitForActorStatus(ctx, t, clients, actorID, ateapipb.Actor_STATUS_RUNNING)
resp, err = callActor(t, actorID)
resp, err = callActor(t, demoAtespace, actorID)
if err != nil {
t.Fatalf("failed to call actor again: %v", err)
}
@@ -214,7 +217,7 @@ func TestDurableDirLifecycle(t *testing.T) {
//
t.Logf("Suspending Actor %q...", actorID)
if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to suspend Actor: %v", err)
}
@@ -223,13 +226,13 @@ func TestDurableDirLifecycle(t *testing.T) {
// Resuming the actor
t.Logf("Resuming Actor %q again...", actorID)
if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to resume Actor again: %v", err)
}
waitForActorStatus(ctx, t, clients, actorID, ateapipb.Actor_STATUS_RUNNING)
resp, err = callActor(t, actorID)
resp, err = callActor(t, demoAtespace, actorID)
if err != nil {
t.Fatalf("failed to call actor again: %v", err)
}
@@ -256,8 +259,7 @@ func createActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj
t.Logf("Creating Actor %q using Substrate API...", actorID)
createResp, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
ActorTemplateNamespace: nsObj.Name,
ActorTemplateName: at.Name,
})
@@ -267,8 +269,7 @@ func createActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj
t.Logf("Successfully created Actor: %s", createResp.GetActor().GetActorId())
defer func() {
clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
})
}()
@@ -312,8 +313,7 @@ func pauseActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj *
// Creating an actor
t.Logf("Creating Actor %q...", actorID)
if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
ActorTemplateNamespace: nsObj.Name,
ActorTemplateName: at.Name,
}); err != nil {
@@ -324,8 +324,7 @@ func pauseActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj *
// Resuming the actor
t.Logf("Resuming Actor %q...", actorID)
if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to resume Actor: %v", err)
}
@@ -345,8 +344,7 @@ func pauseActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj *
// Pausing the actor
t.Logf("Pausing Actor %q...", actorID)
if _, err := clients.SubstrateAPI.PauseActor(ctx, &ateapipb.PauseActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to pause Actor: %v", err)
}
@@ -355,8 +353,7 @@ func pauseActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj *
// Resuming the actor again
t.Logf("Resuming Actor %q again...", actorID)
if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to resume Actor again: %v", err)
}
@@ -375,8 +372,7 @@ func pauseActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj *
// Suspending the actor before deletion
t.Logf("Suspending Actor %q before deletion...", actorID)
if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to suspend Actor: %v", err)
}
@@ -385,15 +381,13 @@ func pauseActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj *
// Deleting the actor
t.Logf("Deleting Actor %q...", actorID)
if _, err := clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to delete Actor: %v", err)
}
// Verify deletion
if _, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err == nil {
t.Fatalf("expected actor %q to be deleted, but it still exists", actorID)
}
@@ -407,8 +401,7 @@ func suspendActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj
// Creating an actor
t.Logf("Creating Actor %q...", actorID)
if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
ActorTemplateNamespace: nsObj.Name,
ActorTemplateName: at.Name,
}); err != nil {
@@ -419,8 +412,7 @@ func suspendActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj
// Resuming the actor
t.Logf("Resuming Actor %q...", actorID)
if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to resume Actor: %v", err)
}
@@ -439,8 +431,7 @@ func suspendActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj
// Suspending the actor
t.Logf("Suspending Actor %q...", actorID)
if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to suspend Actor: %v", err)
}
@@ -449,8 +440,7 @@ func suspendActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj
// Resuming the actor again
t.Logf("Resuming Actor %q again...", actorID)
if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to resume Actor again: %v", err)
}
@@ -469,8 +459,7 @@ func suspendActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj
// Suspending the actor before deletion
t.Logf("Suspending Actor %q before deletion...", actorID)
if _, err := clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to suspend Actor: %v", err)
}
@@ -479,15 +468,13 @@ func suspendActor(ctx context.Context, t *testing.T, clients *e2e.Clients, nsObj
// Deleting the actor
t.Logf("Deleting Actor %q...", actorID)
if _, err := clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err != nil {
t.Fatalf("failed to delete Actor: %v", err)
}
// Verify deletion
if _, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
}); err == nil {
t.Fatalf("expected actor %q to be deleted, but it still exists", actorID)
}
@@ -619,8 +606,7 @@ func waitForActorStatus(ctx context.Context, t *testing.T, clients *e2e.Clients,
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
resp, err := clients.SubstrateAPI.GetActor(ctx, &ateapipb.GetActorRequest{
Atespace: demoAtespace,
ActorId: actorID,
ActorRef: &ateapipb.ActorRef{Atespace: demoAtespace, Name: actorID},
})
if err == nil {
if resp.GetActor().GetStatus() == expectedStatus {
internal/e2e/suites/identity/identity_test.go
+4 -5
View File
@@ -162,8 +162,7 @@ func createAndResumeActor(t *testing.T, ctx context.Context, clients *e2e.Client
// CreateActor requires the atespace to exist first.
_, _ = clients.SubstrateAPI.CreateAtespace(ctx, &ateapipb.CreateAtespaceRequest{Name: probeNamespace})
if _, err := clients.SubstrateAPI.CreateActor(ctx, &ateapipb.CreateActorRequest{
Atespace: probeNamespace,
ActorId: id,
ActorRef: &ateapipb.ActorRef{Atespace: probeNamespace, Name: id},
ActorTemplateNamespace: probeNamespace,
ActorTemplateName: probeTemplate,
}); err != nil {
@@ -171,12 +170,12 @@ func createAndResumeActor(t *testing.T, ctx context.Context, clients *e2e.Client
}
t.Cleanup(func() {
// DeleteActor requires the actor to be suspended.
_, _ = clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{Atespace: probeNamespace, ActorId: id})
_, _ = clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{Atespace: probeNamespace, ActorId: id})
_, _ = clients.SubstrateAPI.SuspendActor(ctx, &ateapipb.SuspendActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: probeNamespace, Name: id}})
_, _ = clients.SubstrateAPI.DeleteActor(ctx, &ateapipb.DeleteActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: probeNamespace, Name: id}})
})
// Resume from the golden snapshot (the restore path, not --boot).
if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{Atespace: probeNamespace, ActorId: id}); err != nil {
if _, err := clients.SubstrateAPI.ResumeActor(ctx, &ateapipb.ResumeActorRequest{ActorRef: &ateapipb.ActorRef{Atespace: probeNamespace, Name: id}}); err != nil {
t.Fatalf("ResumeActor %q: %v", id, err)
}
}
pkg/proto/ateapipb/ateapi.pb.go
+308 -309
View File
File diff suppressed because it is too large Load Diff
pkg/proto/ateapipb/ateapi.proto
+14 -19
View File
@@ -148,6 +148,12 @@ message Atespace {
string name = 1;
}
// ActorRef identifies an actor: a name is only unique within its atespace.
message ActorRef {
string atespace = 1;
string name = 2;
}
message CreateAtespaceRequest {
string name = 1;
}
@@ -173,8 +179,7 @@ message DeleteAtespaceRequest {
message DeleteAtespaceResponse {}
message GetActorRequest {
string actor_id = 1;
string atespace = 2;
ActorRef actor_ref = 1;
}
message GetActorResponse {
@@ -183,8 +188,8 @@ message GetActorResponse {
// Request to create a new Actor.
message CreateActorRequest {
// The custom identifier for the new Actor. Must be a valid DNS-1123 label.
string actor_id = 1;
// The new actor's atespace and actor_id (actor_id must be a valid DNS-1123 label).
ActorRef actor_ref = 1;
// The namespace of the ActorTemplate to derive from.
string actor_template_namespace = 2;
@@ -195,9 +200,6 @@ message CreateActorRequest {
// worker_selector sets the actor's placement constraint at creation time.
// If empty, the actor matches any pool admitted by the template's selector.
Selector worker_selector = 4;
// The atespace to create the actor into.
string atespace = 5;
}
message CreateActorResponse {
@@ -208,14 +210,11 @@ message CreateActorResponse {
// May be called regardless of the actor's current status.
// Changes take effect on the next ResumeActor call.
message UpdateActorRequest {
string actor_id = 1;
ActorRef actor_ref = 1;
// worker_selector replaces the actor's current placement constraint.
// Takes effect on the next ResumeActor call.
Selector worker_selector = 2;
// The atespace the actor lives in.
string atespace = 3;
}
message UpdateActorResponse {
@@ -223,8 +222,7 @@ message UpdateActorResponse {
}
message SuspendActorRequest {
string actor_id = 1;
string atespace = 2;
ActorRef actor_ref = 1;
}
message SuspendActorResponse {
@@ -232,8 +230,7 @@ message SuspendActorResponse {
}
message PauseActorRequest {
string actor_id = 1;
string atespace = 2;
ActorRef actor_ref = 1;
}
message PauseActorResponse {
@@ -241,11 +238,10 @@ message PauseActorResponse {
}
message ResumeActorRequest {
string actor_id = 1;
ActorRef actor_ref = 1;
// If true, skip golden snapshot and boot the workload from scratch.
bool boot = 2;
string atespace = 3;
}
message ResumeActorResponse {
@@ -253,8 +249,7 @@ message ResumeActorResponse {
}
message DeleteActorRequest {
string actor_id = 1;
string atespace = 2;
ActorRef actor_ref = 1;
}
message DeleteActorResponse {}