Skip to content

Commit 37b5006

Browse files
dberkovdberkov
authored
DurableDir support (agent-substrate#295)
Implement a layer allowing to take durableDir snapshot only, without taking snapshot of entire memory. This is a part of the agent-substrate#119 feature. The "durableDir" support is introduced via adding Volumes support to ActorTemplate. ## Knowns issues / limitations 1. No MicroVM support for homedir yet. Will be implemented in following PR. 2. Resume from SNAPSHOT_SCOPE_DATA temporary behavior Current behavior of resume from durableStorage returns immediately after gVisor process is started, it means container's bootstrap is not completed by the time resume API returned to the caller. It causes the issue that first curl request for suspended actor throws an HTTP exception, since webserver listening on the port inside container is not running yet. PR agent-substrate#330 fixes the issue, for a meantime the workaround is to call ` kubectl ate resume actor` and after it call to the curl itself. - [X ] Unit tests pass - [X ] E2E tests are updated and pass
1 parent 3d53de8 commit 37b5006

27 files changed

Lines changed: 2347 additions & 419 deletions

‎cmd/ateapi/internal/controlapi/converter.go‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
// Copyright 2026 Google LLC
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package controlapi
16+
17+
import (
18+
"log/slog"
19+
20+
"github.com/agent-substrate/substrate/internal/proto/ateletpb"
21+
atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
22+
)
23+
24+
// convert atev1alpha1.SnapshotScope to ateletpb.SnapshotScope
25+
func toAteletSnapshotScope(in atev1alpha1.SnapshotScope) ateletpb.SnapshotScope {
26+
switch in {
27+
case atev1alpha1.SnapshotScopeFull:
28+
return ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL
29+
case atev1alpha1.SnapshotScopeData:
30+
return ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA
31+
default:
32+
slog.Warn("unknown SnapshotScope; falling back to Full", "scope", string(in))
33+
return ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL
34+
}
35+
}

‎cmd/ateapi/internal/controlapi/converter_test.go‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
// Copyright 2026 Google LLC
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package controlapi
16+
17+
import (
18+
"testing"
19+
20+
"github.com/agent-substrate/substrate/internal/proto/ateletpb"
21+
atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1"
22+
)
23+
24+
func TestToAteletSnapshotScope(t *testing.T) {
25+
tests := []struct {
26+
name string
27+
in atev1alpha1.SnapshotScope
28+
expected ateletpb.SnapshotScope
29+
}{
30+
{
31+
name: "Full scope",
32+
in: atev1alpha1.SnapshotScopeFull,
33+
expected: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL,
34+
},
35+
{
36+
name: "Data scope",
37+
in: atev1alpha1.SnapshotScopeData,
38+
expected: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_DATA,
39+
},
40+
{
41+
name: "Default scope (empty)",
42+
in: "",
43+
expected: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL,
44+
},
45+
{
46+
name: "Default scope (unknown)",
47+
in: "unknown",
48+
expected: ateletpb.SnapshotScope_SNAPSHOT_SCOPE_FULL,
49+
},
50+
}
51+
52+
for _, tt := range tests {
53+
t.Run(tt.name, func(t *testing.T) {
54+
result := toAteletSnapshotScope(tt.in)
55+
if result != tt.expected {
56+
t.Errorf("toAteletSnapshotScope(%q) = %v, want %v", tt.in, result, tt.expected)
57+
}
58+
})
59+
}
60+
}

‎cmd/ateapi/internal/controlapi/workflow_pause.go‎

Lines changed: 6 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,8 @@ func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, st
114114
}
115115
client := ateletpb.NewAteomHerderClient(ateletConn)
116116

117+
workloadSpec := workloadSpecFromActorTemplate(state.ActorTemplate)
118+
117119
// Checkpoint does not carry the sandbox config: atelet uses the version the
118120
// actor is currently running (recorded on-node at Run/Restore) and pins it
119121
// into the snapshot manifest.
@@ -122,36 +124,16 @@ func (s *CallAteletPauseStep) Execute(ctx context.Context, input *PauseInput, st
122124
ActorTemplateNamespace: state.Actor.GetActorTemplateNamespace(),
123125
ActorTemplateName: state.Actor.GetActorTemplateName(),
124126
ActorId: state.Actor.GetActorId(),
125-
Spec: &ateletpb.WorkloadSpec{
126-
PauseImage: state.ActorTemplate.Spec.PauseImage,
127-
},
128-
Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL,
127+
Spec: workloadSpec,
128+
Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL,
129129
Config: &ateletpb.CheckpointRequest_LocalConfig{
130130
LocalConfig: &ateletpb.LocalCheckpointConfiguration{
131131
SnapshotPrefix: state.Actor.InProgressSnapshot,
132132
},
133133
},
134+
Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnPause),
134135
}
135-
for _, ctr := range state.ActorTemplate.Spec.Containers {
136-
ateletCtr := &ateletpb.Container{
137-
Name: ctr.Name,
138-
Image: ctr.Image,
139-
Command: ctr.Command,
140-
Readyz: toAteletReadyz(ctr.Readyz),
141-
}
142-
for _, env := range ctr.Env {
143-
var val string
144-
if env.Value != nil {
145-
val = *env.Value
146-
}
147-
ateletEnv := &ateletpb.EnvEntry{
148-
Name: env.Name,
149-
Value: val,
150-
}
151-
ateletCtr.Env = append(ateletCtr.Env, ateletEnv)
152-
}
153-
req.Spec.Containers = append(req.Spec.Containers, ateletCtr)
154-
}
136+
155137
_, err = client.Checkpoint(ctx, req)
156138
if err != nil {
157139
return fmt.Errorf("while checkpointing workload: %w", err)

‎cmd/ateapi/internal/controlapi/workflow_resume.go‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -252,7 +252,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
252252
}
253253
client := ateletpb.NewAteomHerderClient(ateletConn)
254254

255-
workloadSpec, err := workloadSpecFromActorTemplate(ctx, s.kubeClient, s.secretCache, state.ActorTemplate)
255+
workloadSpec, err := workloadSpecFromActorTemplateWithEnv(ctx, s.kubeClient, s.secretCache, state.ActorTemplate)
256256
if err != nil {
257257
return err
258258
}
@@ -275,13 +275,15 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
275275
SnapshotPrefix: state.Actor.GetLatestSnapshotInfo().GetLocal().SnapshotPrefix,
276276
},
277277
}
278+
req.Scope = toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnPause)
278279
case ateapipb.SnapshotType_SNAPSHOT_TYPE_EXTERNAL:
279280
req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL
280281
req.Config = &ateletpb.RestoreRequest_ExternalConfig{
281282
ExternalConfig: &ateletpb.ExternalCheckpointConfiguration{
282283
SnapshotUriPrefix: state.Actor.GetLatestSnapshotInfo().GetExternal().SnapshotUriPrefix,
283284
},
284285
}
286+
req.Scope = toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnCommit)
285287
default:
286288
return fmt.Errorf("unsupported snapshot type: %v", state.Actor.GetLatestSnapshotInfo().GetType())
287289
}
@@ -308,6 +310,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
308310
SnapshotUriPrefix: snapshot,
309311
},
310312
},
313+
Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnCommit),
311314
}
312315
_, err = client.Restore(ctx, req)
313316
if err != nil {

‎cmd/ateapi/internal/controlapi/workflow_suspend.go‎

Lines changed: 6 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,8 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
116116
}
117117
client := ateletpb.NewAteomHerderClient(ateletConn)
118118

119+
workloadSpec := workloadSpecFromActorTemplate(state.ActorTemplate)
120+
119121
// Checkpoint does not carry the sandbox config: atelet uses the version the
120122
// actor is currently running (recorded on-node at Run/Restore) and pins it
121123
// into the snapshot manifest.
@@ -124,36 +126,16 @@ func (s *CallAteletSuspendStep) Execute(ctx context.Context, input *SuspendInput
124126
ActorTemplateNamespace: state.Actor.GetActorTemplateNamespace(),
125127
ActorTemplateName: state.Actor.GetActorTemplateName(),
126128
ActorId: state.Actor.GetActorId(),
127-
Spec: &ateletpb.WorkloadSpec{
128-
PauseImage: state.ActorTemplate.Spec.PauseImage,
129-
},
130-
Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL,
129+
Spec: workloadSpec,
130+
Type: ateletpb.CheckpointType_CHECKPOINT_TYPE_EXTERNAL,
131131
Config: &ateletpb.CheckpointRequest_ExternalConfig{
132132
ExternalConfig: &ateletpb.ExternalCheckpointConfiguration{
133133
SnapshotUriPrefix: state.Actor.GetInProgressSnapshot(),
134134
},
135135
},
136+
Scope: toAteletSnapshotScope(state.ActorTemplate.Spec.SnapshotsConfig.OnCommit),
136137
}
137-
for _, ctr := range state.ActorTemplate.Spec.Containers {
138-
ateletCtr := &ateletpb.Container{
139-
Name: ctr.Name,
140-
Image: ctr.Image,
141-
Command: ctr.Command,
142-
Readyz: toAteletReadyz(ctr.Readyz),
143-
}
144-
for _, env := range ctr.Env {
145-
var val string
146-
if env.Value != nil {
147-
val = *env.Value
148-
}
149-
ateletEnv := &ateletpb.EnvEntry{
150-
Name: env.Name,
151-
Value: val,
152-
}
153-
ateletCtr.Env = append(ateletCtr.Env, ateletEnv)
154-
}
155-
req.Spec.Containers = append(req.Spec.Containers, ateletCtr)
156-
}
138+
157139
_, err = client.Checkpoint(ctx, req)
158140
if err != nil {
159141
return fmt.Errorf("while checkpointing workload: %w", err)

‎cmd/ateapi/internal/controlapi/workload_spec.go‎

Lines changed: 43 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -32,14 +32,26 @@ import (
3232

3333
const envSecretCacheTTL = 30 * time.Second
3434

35-
func workloadSpecFromActorTemplate(ctx context.Context, kubeClient kubernetes.Interface, secretCache *envSecretCache, actorTemplate *atev1alpha1.ActorTemplate) (*ateletpb.WorkloadSpec, error) {
35+
// workloadSpecFromActorTemplate builds a WorkloadSpec without resolving
36+
// container env vars. Use this when downstream consumers (e.g. checkpoint
37+
// requests) don't need env entries materialized.
38+
func workloadSpecFromActorTemplate(actorTemplate *atev1alpha1.ActorTemplate) *ateletpb.WorkloadSpec {
3639
workloadSpec := &ateletpb.WorkloadSpec{
3740
PauseImage: actorTemplate.Spec.PauseImage,
3841
}
39-
resolver := envResolver{
40-
kubeClient: kubeClient,
41-
namespace: actorTemplate.Namespace,
42-
cache: secretCache,
42+
43+
// add volumes
44+
for _, vol := range actorTemplate.Spec.Volumes {
45+
// volume is durable-dir type
46+
if vol.VolumeSource.DurableDir != nil {
47+
workloadSpec.Volumes = append(workloadSpec.Volumes, &ateletpb.Volume{
48+
Name: vol.Name,
49+
Type: ateletpb.VolumeType_VOLUME_TYPE_DURABLE_DIR,
50+
Source: &ateletpb.Volume_DurableDir{
51+
DurableDir: &ateletpb.DurableDirVolume{},
52+
},
53+
})
54+
}
4355
}
4456

4557
for _, ctr := range actorTemplate.Spec.Containers {
@@ -49,16 +61,40 @@ func workloadSpecFromActorTemplate(ctx context.Context, kubeClient kubernetes.In
4961
Command: ctr.Command,
5062
Readyz: toAteletReadyz(ctr.Readyz),
5163
}
64+
for _, mount := range ctr.VolumeMounts {
65+
ateletCtr.VolumeMounts = append(ateletCtr.VolumeMounts, &ateletpb.VolumeMount{
66+
Name: mount.Name,
67+
MountPath: mount.MountPath,
68+
})
69+
}
70+
workloadSpec.Containers = append(workloadSpec.Containers, ateletCtr)
71+
}
72+
73+
return workloadSpec
74+
}
75+
76+
// workloadSpecFromActorTemplateWithEnv builds a WorkloadSpec and resolves each
77+
// container's env vars against the cluster. kubeClient must be non-nil;
78+
// secretCache is optional and, when supplied, deduplicates Secret reads.
79+
func workloadSpecFromActorTemplateWithEnv(ctx context.Context, kubeClient kubernetes.Interface, secretCache *envSecretCache, actorTemplate *atev1alpha1.ActorTemplate) (*ateletpb.WorkloadSpec, error) {
80+
workloadSpec := workloadSpecFromActorTemplate(actorTemplate)
81+
82+
resolver := envResolver{
83+
kubeClient: kubeClient,
84+
namespace: actorTemplate.Namespace,
85+
cache: secretCache,
86+
}
87+
88+
for i, ctr := range actorTemplate.Spec.Containers {
5289
for _, env := range ctr.Env {
5390
ateletEnv, err := resolver.resolve(ctx, ctr.Name, env)
5491
if err != nil {
5592
return nil, err
5693
}
5794
if ateletEnv != nil {
58-
ateletCtr.Env = append(ateletCtr.Env, ateletEnv)
95+
workloadSpec.Containers[i].Env = append(workloadSpec.Containers[i].Env, ateletEnv)
5996
}
6097
}
61-
workloadSpec.Containers = append(workloadSpec.Containers, ateletCtr)
6298
}
6399

64100
return workloadSpec, nil

0 commit comments

Comments
 (0)