Skip to content

Commit 6be3d50

Browse files
teemowQuentinBisson
teemow
authored and
QuentinBisson
committed
fix(atenet,ateom): let an actor's egress through while it resumes — the golden boot fetches before readyz (#4)
* Let an actor's egress through while it resumes An actor's tunneled egress was armed only after every container served readyz, and atenet's egress gateway refused a CONNECT from any actor that was not RUNNING — a state the control plane commits only after the worker's Run returned, that is, after readyz. A workload that fetches what it needs to become ready — a model, a skill repository, packages — therefore could not: atunnel dropped the intercepted connection on the floor, the fetch died with "Broken pipe", readyz never came, and the actor was re-run every minute with one worker pinned to it. Nothing was logged at either hop; the workload's own error was the only trace. An ActorTemplate whose workload fetches at start-up never got its golden snapshot. The certificate the tunnel authenticates with is minted for exactly this placement before the workload starts (prepareActorEgress; the credential broker mints only for a worker's current assignment), so the identity is there from the first packet. Readiness gates ingress, not egress. * ateom (gVisor and micro-VM): arm egress right after the actor network is set up and the failure cleanup is registered, before the first container starts; ingress still waits for readyz. * atenet egress: a RESUMING actor is placed on a worker and may tunnel. SUSPENDED, PAUSED, CRASHED and DELETING are still refused: a certificate within its lifetime must not tunnel for an actor that has left its worker. The refusal is logged with the actor, its state and the destination. * atunnel: a dropped egress connection (no active egress, expired certificate) is logged with the peer address. * docs(fork): ledger row for the egress-while-resuming patch (#4) (cherry picked from commit 05de980)
1 parent b0599bb commit 6be3d50

10 files changed

Lines changed: 124 additions & 27 deletions

File tree

‎FORK.md‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,13 @@ Everything on `giantswarm` that is not in the pin (`git log v0.0.26..giantswarm`
3737
| Patch | Purpose | Fork commit | Upstream |
3838
|---|---|---|---|
3939
| Grant atelet cluster-wide read access to sandbox configs | atelet's sandbox-asset prewarm degraded on the second test cluster without the RBAC ([#37742](https://github.com/giantswarm/giantswarm/issues/37742) row 10) | `74b45f9e` (`git cherry-pick -x a7505e9c`) | [kagent-dev/substrate#33](https://github.com/kagent-dev/substrate/pull/33), merged 2026-09-08, not in v0.0.26 — falls away at the re-pin onto the first tag that contains it |
40+
| Let an actor's egress through while it resumes (ateom arms tunneled egress before the first container starts, atenet admits `RESUMING` actors, both hops log a refusal) | an actor whose workload fetches what it needs to become ready — kagent's Go ADK and Claude harnesses materialise git skills before readyz — never got its golden snapshot: atunnel dropped the fetch (`Broken pipe`), atenet would have refused a non-`RUNNING` actor, nothing was logged ([#37742](https://github.com/giantswarm/giantswarm/issues/37742) rows 8 and 13; acceptance test `agentlab skills-test`, [agentlab#137](https://github.com/giantswarm/agentlab/issues/137)) | [#4](https://github.com/giantswarm/substrate/pull/4) (squash; `git log --grep '(#4)'`) | to file: the upstream-shaped patch is branch [`upstream/atenet-egress-during-resume`](https://github.com/giantswarm/substrate/tree/upstream/atenet-egress-during-resume) here, rebased on the mirror `main`; a team member opens the kagent-dev/substrate pull request with DCO sign-off once #37742 has reviewed it |
4041
| Fork infrastructure: this file, the README pointer, `CODEOWNERS`, `.github/workflows/publish.yaml`, `.github/workflows/sync-upstream.yaml`, `.trivyignore`, and the branch triggers of `pr-workflow.yaml`, `helm-e2e.yaml`, `govulncheck.yaml` (`main` → `giantswarm`, govulncheck also on pull requests) | the line's CI, publishing and sync | the `giantswarm` branch history | not for upstream |
4142

42-
Nothing in the line changes Substrate's behaviour beyond what upstream has already merged. Giant Swarm
43-
specific wiring lives elsewhere: the CA/JWT pool bootstrap (`kubectl-ate admin make-ca-pool`/`make-jwt-pool`
43+
One patch changes Substrate's behaviour ahead of upstream — egress for an actor while it resumes, without
44+
which no skill-carrying agent of the platform boots; it is written for upstream and leaves at the first
45+
release that carries it. Everything else is what upstream has already merged. Giant Swarm specific wiring
46+
lives elsewhere: the CA/JWT pool bootstrap (`kubectl-ate admin make-ca-pool`/`make-jwt-pool`
4447
and the `ate-api-authentication` ConfigMap) is created by [agentlab](https://github.com/giantswarm/agentlab)
4548
and by meta chart 4.0; the `WorkerPool` the platform's Harnesses run on comes with the kagent chart
4649
(`kagent.substrateWorkerPool`); feature gates, Kyverno exceptions and network policies are cluster

‎cmd/atenet/internal/router/README.md‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,11 @@ Router has several responsibilities:
2626
* Authenticates actor identity on egress: on every CONNECT, the egress
2727
gateway's ext_proc handler re-verifies the actor's client certificate against
2828
the actor-identity CA, reads the `ActorIdentity` X.509 extension out of it,
29-
and checks the certified UID against the ATE API.
29+
and checks against the ATE API that the certified UID is the actor's and that
30+
the actor is placed on a worker (running, or resuming onto the worker that
31+
minted the certificate — what a workload fetches to become ready goes out
32+
before it serves readyz). Every denial is logged with the actor and the
33+
destination.
3034
* Authorizes egress against the actor's `EgressPolicy`: every request the
3135
gateway can read is decided on its `Host` and the address the actor dialed,
3236
rules in order, first match wins; what it cannot read is decided by address

‎cmd/atenet/internal/router/egress/egress.go‎

Lines changed: 28 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -161,7 +161,9 @@ func (h *Handler) handleConnect(ctx context.Context, md *extproc.RequestMetadata
161161
"egress denied: invalid actor certificate")
162162
}
163163

164-
if err := h.validateActor(ctx, actorRef); err != nil {
164+
// For a CONNECT the :authority is the actor's original destination
165+
// (IP:port).
166+
if err := h.validateActor(ctx, actorRef, md.Host); err != nil {
165167
return extproc.Result{}, err
166168
}
167169

@@ -253,9 +255,11 @@ func (h *Handler) lookupPolicy(ctx context.Context, leg string, ref resources.Ac
253255
}
254256

255257
// validateActor checks the actor certified by the certificate against the control
256-
// plane's current view of that actor: it still exists and it is running. Every
257-
// error it returns is already a client-facing ext_proc denial.
258-
func (h *Handler) validateActor(ctx context.Context, actorRef resources.ActorRef) error {
258+
// plane's current view of that actor: it still exists and it is placed on a
259+
// worker, running or resuming onto the worker that minted the certificate. Every
260+
// error it returns is already a client-facing ext_proc denial; destination is
261+
// the CONNECT's target, for the log.
262+
func (h *Handler) validateActor(ctx context.Context, actorRef resources.ActorRef, destination string) error {
259263
// Confirm the certified actor still exists.
260264
// TODO: this can cause heavy load on ate api server. Change it based on https://github.com/agent-substrate/substrate/issues/592.
261265
actor, err := h.apiClient.GetActor(ctx, &ateapipb.GetActorRequest{
@@ -265,12 +269,28 @@ func (h *Handler) validateActor(ctx context.Context, actorRef resources.ActorRef
265269
return mapEgressIdentityError(actorRef.Atespace, actorRef.Name, err)
266270
}
267271

268-
// The actor performing egress must actually be running.
269-
if actor.GetStatus().GetState() != ateapipb.ActorState_ACTOR_STATE_RUNNING {
272+
// The actor performing egress must be placed on a worker. RUNNING is the
273+
// steady state; RESUMING is the workload booting or restoring on the
274+
// worker that minted this certificate for exactly that placement (the
275+
// credential broker mints only for a worker's current assignment), and a
276+
// workload that fetches what it needs to become ready (models, skills,
277+
// packages) does so before it serves readyz. Every other state means the
278+
// actor has left its worker or is leaving it, and a certificate still
279+
// within its lifetime must not open a tunnel on its behalf.
280+
switch state := actor.GetStatus().GetState(); state {
281+
case ateapipb.ActorState_ACTOR_STATE_RUNNING, ateapipb.ActorState_ACTOR_STATE_RESUMING:
282+
return nil
283+
default:
284+
// Logged with the destination: from inside the sandbox this denial
285+
// is indistinguishable from a network failure.
286+
slog.WarnContext(ctx, "egress denied: actor is not placed on a worker",
287+
slog.String("atespace", actorRef.Atespace),
288+
slog.String("actor", actorRef.Name),
289+
slog.String("state", state.String()),
290+
slog.String("destination", destination))
270291
return extproc.NewReqError(envoy_type.StatusCode_Forbidden,
271-
"egress denied: actor %q/%q is %s, not running", actorRef.Atespace, actorRef.Name, actor.GetStatus().GetState())
292+
"egress denied: actor %q/%q is %s, not placed on a worker", actorRef.Atespace, actorRef.Name, state)
272293
}
273-
return nil
274294
}
275295

276296
// authenticateActorCertificate turns the mTLS peer certificate Envoy recorded

‎cmd/atenet/internal/router/egress/egress_test.go‎

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -449,6 +449,21 @@ func TestConnectLegWithoutRequestLegs(t *testing.T) {
449449
wantStatus(t, err, envoy_type.StatusCode_Forbidden)
450450
}
451451

452+
// A resuming actor is booting or restoring on the worker that minted its
453+
// certificate; what it fetches to become ready goes out before it serves
454+
// readyz, so the gateway lets it through.
455+
func TestHandleRequestHeadersAllowsResumingActor(t *testing.T) {
456+
ca := newTestCA(t, "actor-identity-ca")
457+
leaf := ca.issueActorCert(t, "spiffe://substrate-actor.local/ateom-for-actor/foo/bar", actorCertOptions{})
458+
resuming := runningActor()
459+
resuming.Status.State = ateapipb.ActorState_ACTOR_STATE_RESUMING
460+
h := egressHandler(ca.roots(), resuming, nil)
461+
462+
if _, err := h.HandleRequestHeaders(context.Background(), egressMetadata(xfccHeader(leaf))); err != nil {
463+
t.Fatalf("HandleRequestHeaders() error = %v, want nil", err)
464+
}
465+
}
466+
452467
func TestHandleRequestHeadersAllowsAgentgatewayCertificateAttribute(t *testing.T) {
453468
ca := newTestCA(t, "actor-identity-ca")
454469
leaf := ca.issueActorCert(t, "spiffe://substrate-actor.local/ateom-for-actor/foo/bar", actorCertOptions{})
@@ -689,7 +704,10 @@ func TestHandleRequestHeadersAuthorization(t *testing.T) {
689704
want envoy_type.StatusCode
690705
}{
691706
{
692-
name: "actor is not running",
707+
// Suspended, paused and crashed actors have left their worker;
708+
// a deleting one is leaving it. Their certificates are still
709+
// within their lifetime and must not open a tunnel.
710+
name: "actor is suspended",
693711
actor: &ateapipb.Actor{
694712
Metadata: &ateapipb.ResourceMetadata{
695713
Atespace: testEgressAtespace,
@@ -700,6 +718,18 @@ func TestHandleRequestHeadersAuthorization(t *testing.T) {
700718
},
701719
want: envoy_type.StatusCode_Forbidden,
702720
},
721+
{
722+
name: "actor is being deleted",
723+
actor: &ateapipb.Actor{
724+
Metadata: &ateapipb.ResourceMetadata{
725+
Atespace: testEgressAtespace,
726+
Name: testEgressActor,
727+
Uid: testEgressActorUID,
728+
},
729+
Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_DELETING},
730+
},
731+
want: envoy_type.StatusCode_Forbidden,
732+
},
703733
{
704734
name: "actor no longer exists",
705735
err: status.Error(codes.NotFound, "no such actor"),

‎cmd/ateom-gvisor/main.go‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -577,6 +577,11 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload
577577
}
578578
}
579579
}()
580+
// Egress is armed before the first container starts (see
581+
// ateomtunnel.Tunnel.ActivateEgress); ingress waits for readyz below.
582+
if err := s.tunnel.ActivateEgress(ateomstats.ActorAttributionFromRequest(req), egress); err != nil {
583+
return nil, err
584+
}
580585
// Create and start pause container. The bundle rootfs is composed here —
581586
// an overlay of the node's cached image layers plus the bundle's private
582587
// upper — because mounting is ateom's job (atelet runs with no
@@ -617,7 +622,7 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload
617622
if err := wakeupprobe.WaitAll(ctx, req.GetSpec().GetContainers(), ateomnet.ActorVethIP, wakeupprobe.DialFunc(s.sandboxDialer(req.GetActorUid()))); err != nil {
618623
return nil, fmt.Errorf("while waiting for container wakeup probe: %w", err)
619624
}
620-
if err := s.tunnel.Activate(ateomstats.ActorAttributionFromRequest(req), s.sandboxDialer(req.GetActorUid()), egress); err != nil {
625+
if err := s.tunnel.ActivateIngress(ateomstats.ActorAttributionFromRequest(req), s.sandboxDialer(req.GetActorUid())); err != nil {
621626
return nil, err
622627
}
623628

@@ -869,6 +874,10 @@ func (s *AteomService) RestoreWorkload(ctx context.Context, req *ateompb.Restore
869874
}
870875
}
871876
}()
877+
// As in RunWorkload: egress from the first packet, ingress after readyz.
878+
if err := s.tunnel.ActivateEgress(ateomstats.ActorAttributionFromRequest(req), egress); err != nil {
879+
return nil, err
880+
}
872881
checkpointDir := req.GetActorDirs().GetRestoreDir()
873882

874883
if hasDurableVolumes(req.GetSpec().GetContainers()) {
@@ -943,7 +952,7 @@ func (s *AteomService) RestoreWorkload(ctx context.Context, req *ateompb.Restore
943952
if err := wakeupprobe.WaitAll(ctx, req.GetSpec().GetContainers(), ateomnet.ActorVethIP, wakeupprobe.DialFunc(s.sandboxDialer(req.GetActorUid()))); err != nil {
944953
return nil, fmt.Errorf("while waiting for container wakeup probe: %w", err)
945954
}
946-
if err := s.tunnel.Activate(ateomstats.ActorAttributionFromRequest(req), s.sandboxDialer(req.GetActorUid()), egress); err != nil {
955+
if err := s.tunnel.ActivateIngress(ateomstats.ActorAttributionFromRequest(req), s.sandboxDialer(req.GetActorUid())); err != nil {
947956
return nil, err
948957
}
949958

‎cmd/ateom-microvm/restore.go‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -321,6 +321,10 @@ func (s *AteomService) restoreFullScope(ctx context.Context, p actorBootParams,
321321
}
322322
}
323323
}()
324+
// As in coldBootActor: egress from the first packet, ingress after readyz.
325+
if err := s.tunnel.ActivateEgress(p.attribution(), egress); err != nil {
326+
return err
327+
}
324328
netDevs, err := ch.SnapshotNetDevices(restoreDir)
325329
if err != nil {
326330
return fmt.Errorf("while reading snapshot net devices: %w", err)
@@ -478,7 +482,7 @@ func (s *AteomService) restoreFullScope(ctx context.Context, p actorBootParams,
478482
s.startActorLogForwarding(guestAC, attribution, c.GetName(), c.GetName())
479483
}
480484

481-
if err := s.tunnel.Activate(p.attribution(), s.sandboxDialer(p.actorUID), egress); err != nil {
485+
if err := s.tunnel.ActivateIngress(p.attribution(), s.sandboxDialer(p.actorUID)); err != nil {
482486
return err
483487
}
484488
s.setRunningVM(actorUID, ra)

‎cmd/ateom-microvm/run.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -415,6 +415,11 @@ func (s *AteomService) coldBootActor(ctx context.Context, p actorBootParams) (re
415415
}
416416
}
417417
}()
418+
// Egress is armed before the guest boots (see
419+
// ateomtunnel.Tunnel.ActivateEgress); ingress waits for readyz below.
420+
if err := s.tunnel.ActivateEgress(p.attribution(), egress); err != nil {
421+
return err
422+
}
418423

419424
// Guest sizing + agent kernel params.
420425
memMiB, vcpus, kparams := s.guestConfig()
@@ -590,7 +595,7 @@ func (s *AteomService) coldBootActor(ctx context.Context, p actorBootParams) (re
590595
slog.Duration("since_boot", time.Since(tBooted)))
591596

592597
ra := &runningActor{chCmd: chCmd, vfsdCmd: vfsdCmd, apiSocket: apiSocket, baseID: actorUID, guestAgent: ac, workloadIDs: workloadIDs(ctrs)}
593-
if err := s.tunnel.Activate(p.attribution(), s.sandboxDialer(p.actorUID), egress); err != nil {
598+
if err := s.tunnel.ActivateIngress(p.attribution(), s.sandboxDialer(p.actorUID)); err != nil {
594599
return err
595600
}
596601
s.setRunningVM(actorUID, ra)

‎internal/ateomtunnel/ateomtunnel.go‎

Lines changed: 15 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -218,12 +218,12 @@ func (t *Tunnel) PrepareEgress(ctx context.Context, actor resources.ActorAttribu
218218
return &ActorEgress{client: gatewayClient, certificateSource: certificateSource, expiresAt: expiresAt}, nil
219219
}
220220

221-
// Activate starts admitting the actor's traffic. Ingress reaches the actor
222-
// through dial; egress is activated only when it was prepared.
223-
func (t *Tunnel) Activate(actor resources.ActorAttribution, dial atunnel.DialFunc, egress *ActorEgress) error {
224-
if err := t.Ingress.Activate(actor.Ref.Atespace, actor.Ref.Name, actor.UID, dial); err != nil {
225-
return fmt.Errorf("while activating actor ingress: %w", err)
226-
}
221+
// ActivateEgress arms the actor's tunneled egress before its workload starts.
222+
// The certificate was minted for this placement in PrepareEgress, so the tunnel
223+
// carries the actor's identity from its first packet, and what a workload
224+
// fetches to become ready (models, skills, packages) goes out before it serves
225+
// readyz. A nil egress keeps the actor's egress on the masquerade path.
226+
func (t *Tunnel) ActivateEgress(actor resources.ActorAttribution, egress *ActorEgress) error {
227227
if egress == nil {
228228
return nil
229229
}
@@ -233,6 +233,15 @@ func (t *Tunnel) Activate(actor resources.ActorAttribution, dial atunnel.DialFun
233233
return nil
234234
}
235235

236+
// ActivateIngress admits inbound traffic, reaching the actor through dial, once
237+
// every readyz-enabled container of the workload reports ready.
238+
func (t *Tunnel) ActivateIngress(actor resources.ActorAttribution, dial atunnel.DialFunc) error {
239+
if err := t.Ingress.Activate(actor.Ref.Atespace, actor.Ref.Name, actor.UID, dial); err != nil {
240+
return fmt.Errorf("while activating actor ingress: %w", err)
241+
}
242+
return nil
243+
}
244+
236245
// Deactivate stops admitting the actor's traffic and drains its active streams,
237246
// before the actor network is torn down. It attempts both directions even if
238247
// one fails.

‎internal/ateomtunnel/ateomtunnel_test.go‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -275,8 +275,11 @@ func TestActivateAndDeactivate(t *testing.T) {
275275
actor := resources.ActorAttribution{Ref: resources.ActorRef{Atespace: "space", Name: "actor"}, UID: "uid"}
276276
dial := func(context.Context, string, string) (net.Conn, error) { return nil, net.ErrClosed }
277277

278-
if err := tunnel.Activate(actor, dial, nil); err != nil {
279-
t.Fatalf("Activate without egress: %v", err)
278+
if err := tunnel.ActivateEgress(actor, nil); err != nil {
279+
t.Fatalf("ActivateEgress without egress: %v", err)
280+
}
281+
if err := tunnel.ActivateIngress(actor, dial); err != nil {
282+
t.Fatalf("ActivateIngress: %v", err)
280283
}
281284
if err := tunnel.Deactivate(ctx, actor); err != nil {
282285
t.Fatalf("Deactivate: %v", err)
@@ -286,8 +289,8 @@ func TestActivateAndDeactivate(t *testing.T) {
286289
t.Fatalf("second Deactivate: %v", err)
287290
}
288291

289-
err := tunnel.Activate(resources.ActorAttribution{UID: "uid"}, dial, nil)
292+
err := tunnel.ActivateIngress(resources.ActorAttribution{UID: "uid"}, dial)
290293
if err == nil || !strings.Contains(err.Error(), "while activating actor ingress") {
291-
t.Fatalf("Activate with no actor reference = %v, want an ingress error", err)
294+
t.Fatalf("ActivateIngress with no actor reference = %v, want an ingress error", err)
292295
}
293296
}

‎internal/atunnel/egress.go‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -271,13 +271,23 @@ func (e *Egress) handle(downstream net.Conn, active *egressActivation) {
271271
e.mu.Lock()
272272
if active == nil || active.ctx.Err() != nil {
273273
e.mu.Unlock()
274+
// Dropped, and said so: from inside the sandbox a dropped connection
275+
// looks like a network failure, and this is the only place that knows
276+
// why. The original destination is deliberately not resolved for a
277+
// connection nobody vouches for; the peer address ties the line to
278+
// the actor's own log.
279+
slog.Warn("atunnel dropped an egress connection: no actor egress is active",
280+
slog.String("from", downstream.RemoteAddr().String()))
274281
_ = downstream.Close()
275282
return
276283
}
277-
if time.Now().Compare(active.expiresAt) >= 0 {
284+
if expiresAt := active.expiresAt; time.Now().Compare(expiresAt) >= 0 {
278285
// Expiry blocks only new tunnels. Connections admitted with a valid
279286
// certificate have completed mTLS and are allowed to drain normally.
280287
e.mu.Unlock()
288+
slog.WarnContext(active.ctx, "atunnel dropped an egress connection: the actor certificate has expired",
289+
slog.String("from", downstream.RemoteAddr().String()),
290+
slog.Time("expiredAt", expiresAt))
281291
_ = downstream.Close()
282292
return
283293
}

0 commit comments

Comments
 (0)