Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions cmd/atelet/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -1449,7 +1449,7 @@ func (s *AteomHerder) prepareOCIBundles(
[]string{"/pause"},
nil,
nil,
ateompath.AteomNetNSPath(targetAteomUid),
ateompath.ActorNetNSPath(actorUID),
nil, // pause is sandbox infra; it mounts no volumes.
nil,
nil, // pause only reaps; it needs no capabilities.
Expand Down Expand Up @@ -1477,7 +1477,7 @@ func (s *AteomHerder) prepareOCIBundles(
ctr.GetCommand(),
ctr.GetArgs(),
envs,
ateompath.AteomNetNSPath(targetAteomUid),
ateompath.ActorNetNSPath(actorUID),
spec.GetVolumes(),
ctr.GetVolumeMounts(),
resolveCapabilities(ctr.GetSecurityContext().GetCapabilities()),
Expand Down
99 changes: 38 additions & 61 deletions cmd/ateom-gvisor/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,6 @@ import (
"github.com/agent-substrate/substrate/internal/sizing"
"github.com/agent-substrate/substrate/internal/version"
"github.com/spf13/pflag"
"github.com/vishvananda/netns"
"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
"golang.org/x/sys/unix"
"google.golang.org/grpc"
Expand Down Expand Up @@ -195,26 +194,30 @@ func do(ctx context.Context) error {
return fmt.Errorf("while opening unix socket: %w", err)
}

// Create a new network namespace that we will pass to gVisor. gVisor will
// read the addresses and routes off of every link in the namespace, then
// remove all the addresses and handle injecting packets into the interfaces
// using AF_PACKET.
interiorNetNS, err := ateomnet.CreateNetNSWithoutSwitching(ateompath.AteomNetNSName(*podUID))
if err != nil {
return fmt.Errorf("while creating ateom-interior netns: %w", err)
}

actorLogger := actorlog.NewActorLogger(syncedWriter, metadata.OnGCE())
upstream, err := url.Parse(actorHTTPUpstream)
if err != nil {
return fmt.Errorf("while parsing atunnel upstream: %w", err)
}
atunnelIngress, atunnelEgress, atunnelEgressPort, err := runAtunnel(ctx, upstream)
// Use the pod's resolvers for cluster DNS access.
nameservers, err := atunnel.ResolvConfNameservers("/etc/resolv.conf")
if err != nil {
return err
return fmt.Errorf("while reading the worker pod resolvers: %w", err)
}
dnsRelay, err := atunnel.NewDNSRelay(nameservers)
if err != nil {
return fmt.Errorf("while building the actor DNS relay: %w", err)
}
slog.InfoContext(ctx, "Actor DNS relay ready", slog.Any("upstreams", nameservers))

ateomService := NewService(interiorNetNS, actorLogger, atunnelIngress, atunnelEgress, atunnelEgressPort, *workerCredentialBundle, *podIdentityTrustBundle, *egressGatewayTrustBundle)
// Construct the service first so atunnel can use its namespace dialer.
ateomService := NewService(dnsRelay, actorLogger, *workerCredentialBundle, *podIdentityTrustBundle, *egressGatewayTrustBundle)

atunnelIngress, atunnelEgress, atunnelEgressPort, err := runAtunnel(ctx, upstream, ateomService.sandbox.Dialer())
if err != nil {
return err
}
ateomService.attachAtunnel(atunnelIngress, atunnelEgress, atunnelEgressPort)

svr := grpc.NewServer(
grpc.StatsHandler(otelgrpc.NewServerHandler()),
Expand Down Expand Up @@ -266,12 +269,13 @@ func do(ctx context.Context) error {
return nil
}

func runAtunnel(ctx context.Context, upstream *url.URL) (*atunnel.Server, *atunnel.Egress, uint16, error) {
func runAtunnel(ctx context.Context, upstream *url.URL, dial atunnel.DialFunc) (*atunnel.Server, *atunnel.Egress, uint16, error) {
atunnelIngress, err := atunnel.NewServer(atunnel.Config{
CredentialBundlePath: *workerCredentialBundle,
TrustBundlePath: *podIdentityTrustBundle,
AllowedClientID: *atunnelClientIdentity,
Upstream: upstream,
Dial: dial,
})
if err != nil {
return nil, nil, 0, fmt.Errorf("while configuring atunnel: %w", err)
Expand Down Expand Up @@ -301,22 +305,11 @@ func runAtunnel(ctx context.Context, upstream *url.URL) (*atunnel.Server, *atunn
if err != nil {
return nil, nil, 0, fmt.Errorf("while configuring atunnel egress: %w", err)
}
egressListener, err := net.Listen("tcp", *atunnelEgressListenAddress)
// Bind egress only in sandbox namespaces.
atunnelEgressPort, err := atunnel.EgressPort(*atunnelEgressListenAddress)
if err != nil {
return nil, nil, 0, fmt.Errorf("while opening atunnel egress listener: %w", err)
}
egressTCPAddr, ok := egressListener.Addr().(*net.TCPAddr)
if !ok || egressTCPAddr.Port < 1 || egressTCPAddr.Port > 65535 {
_ = egressListener.Close()
return nil, nil, 0, fmt.Errorf("atunnel egress listener has invalid address %q", egressListener.Addr())
return nil, nil, 0, err
}
atunnelEgressPort := uint16(egressTCPAddr.Port)
go func() {
if err := atunnelEgress.Serve(ctx, egressListener); err != nil {
serverboot.Fatal(ctx, "Failed to serve actor egress", err)
}
}()
slog.InfoContext(ctx, "atunnel egress serving", slog.String("address", *atunnelEgressListenAddress))
return atunnelIngress, atunnelEgress, atunnelEgressPort, nil
}

Expand Down Expand Up @@ -374,11 +367,16 @@ type AteomService struct {
// subcommands are probably not safe to call concurrently.
lock *cancelableMutex

interiorNetNS netns.NsHandle
// sandbox is the network of the actor this worker is serving.
sandbox ateomnet.SessionHolder

actorLogger *actorlog.ActorLogger
atunnelIngress *atunnel.Server
atunnelEgress *atunnel.Egress

// dnsRelay answers the sandbox's DNS from inside its own namespace.
dnsRelay *atunnel.DNSRelay

// atunnelEgressPort is the local atunnel listener used as the target of the
// actor network's transparent TCP redirect.
atunnelEgressPort uint16
Expand Down Expand Up @@ -440,14 +438,11 @@ type AteomService struct {
var _ ateompb.AteomServer = (*AteomService)(nil)

// NewService creates a new AteomService.
func NewService(interiorNetNS netns.NsHandle, actorLogger *actorlog.ActorLogger, atunnelIngress *atunnel.Server, atunnelEgress *atunnel.Egress, atunnelEgressPort uint16, workerCredentialBundlePath, podIdentityTrustBundlePath, egressGatewayTrustBundlePath string) *AteomService {
func NewService(dnsRelay *atunnel.DNSRelay, actorLogger *actorlog.ActorLogger, workerCredentialBundlePath, podIdentityTrustBundlePath, egressGatewayTrustBundlePath string) *AteomService {
return &AteomService{
lock: newCancelableMutex(),
interiorNetNS: interiorNetNS,
dnsRelay: dnsRelay,
actorLogger: actorLogger,
atunnelIngress: atunnelIngress,
atunnelEgress: atunnelEgress,
atunnelEgressPort: atunnelEgressPort,
workerCredentialBundlePath: workerCredentialBundlePath,
podIdentityTrustBundlePath: podIdentityTrustBundlePath,
egressGatewayTrustBundlePath: egressGatewayTrustBundlePath,
Expand Down Expand Up @@ -666,15 +661,11 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload
if err != nil {
return nil, err
}
if err := ateomnet.SetupActorNetwork(ctx, ateomnet.NetworkConfig{
InteriorNetNS: s.interiorNetNS,
DumpNetInfo: true,
EgressRedirectPort: s.egressRedirectPort(req.GetEgressGateway() != nil),
}); err != nil {
if err := s.prepareSandboxNetwork(ctx, req.GetActorUid()); err != nil {
// Cleared here as well as in the deferred cleanup below, because that
// defer is not registered until after this check.
s.activeActor.Store(nil)
return nil, fmt.Errorf("while setting up actor network: %w", err)
return nil, err
}
rcmd := &runsc{
path: req.GetRunscPath(),
Expand All @@ -700,7 +691,7 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload
slog.WarnContext(ctx, "Failed to unmount bundle rootfs overlays after Run failure",
"actorUID", req.GetActorUid(), "err", err)
}
if err := ateomnet.CleanupActorNetwork(cleanupCtx, s.interiorNetNS); err != nil {
if err := s.releaseSandboxNetwork(cleanupCtx); err != nil {
slog.WarnContext(cleanupCtx, "Failed to clean up actor network after Run failure", slog.Any("err", err))
}
}
Expand Down Expand Up @@ -742,7 +733,7 @@ func (s *AteomService) RunWorkload(ctx context.Context, req *ateompb.RunWorkload
}

// Block until every readyz-enabled container reports 200.
if err := readyz.WaitAll(ctx, req.GetSpec().GetContainers(), ateomnet.ActorVethIP); err != nil {
if err := readyz.WaitAll(ctx, req.GetSpec().GetContainers(), ateomnet.ActorVethIP, readyz.DialFunc(s.sandbox.Dialer())); err != nil {
return nil, fmt.Errorf("while waiting for container readyz: %w", err)
}
if err := s.activateActorNetworking(req.GetAtespace(), req.GetActorName(), egress); err != nil {
Expand Down Expand Up @@ -965,14 +956,10 @@ func (s *AteomService) RestoreWorkload(ctx context.Context, req *ateompb.Restore
if err != nil {
return nil, err
}
if err := ateomnet.SetupActorNetwork(ctx, ateomnet.NetworkConfig{
InteriorNetNS: s.interiorNetNS,
DumpNetInfo: true,
EgressRedirectPort: s.egressRedirectPort(req.GetEgressGateway() != nil),
}); err != nil {
if err := s.prepareSandboxNetwork(ctx, req.GetActorUid()); err != nil {
// Same as the Run path: the defer below is not registered yet.
s.activeActor.Store(nil)
return nil, fmt.Errorf("while setting up actor network: %w", err)
return nil, err
}
rcmd := &runsc{
path: req.GetRunscPath(),
Expand All @@ -995,7 +982,7 @@ func (s *AteomService) RestoreWorkload(ctx context.Context, req *ateompb.Restore
slog.WarnContext(ctx, "Failed to unmount bundle rootfs overlays after Restore failure",
"actorUID", req.GetActorUid(), "err", err)
}
if err := ateomnet.CleanupActorNetwork(cleanupCtx, s.interiorNetNS); err != nil {
if err := s.releaseSandboxNetwork(cleanupCtx); err != nil {
slog.WarnContext(cleanupCtx, "Failed to clean up actor network after Restore failure", slog.Any("err", err))
}
}
Expand Down Expand Up @@ -1071,7 +1058,7 @@ func (s *AteomService) RestoreWorkload(ctx context.Context, req *ateompb.Restore
}

// Block until every readyz-enabled container reports 200.
if err := readyz.WaitAll(ctx, req.GetSpec().GetContainers(), ateomnet.ActorVethIP); err != nil {
if err := readyz.WaitAll(ctx, req.GetSpec().GetContainers(), ateomnet.ActorVethIP, readyz.DialFunc(s.sandbox.Dialer())); err != nil {
return nil, fmt.Errorf("while waiting for container readyz: %w", err)
}
if err := s.activateActorNetworking(req.GetAtespace(), req.GetActorName(), egress); err != nil {
Expand Down Expand Up @@ -1182,7 +1169,7 @@ func (s *AteomService) terminateWorkload(ctx context.Context, actorRef resources
errs = append(errs, fmt.Errorf("while unmounting bundle rootfs overlays: %w", err))
}

if err := ateomnet.CleanupActorNetwork(ctx, s.interiorNetNS); err != nil {
if err := s.releaseSandboxNetwork(ctx); err != nil {
errs = append(errs, fmt.Errorf("while cleaning up actor network: %w", err))
}

Expand Down Expand Up @@ -1221,16 +1208,6 @@ func (s *AteomService) deactivateActorNetworking(ctx context.Context) error {
return nil
}

// egressRedirectPort returns the local atunnel egress listener port when the
// activation arms tunneled egress, and zero otherwise, which leaves the
// prerouting redirect uninstalled and actor egress on the masquerade path.
func (s *AteomService) egressRedirectPort(redirectEgress bool) uint16 {
if !redirectEgress {
return 0
}
return s.atunnelEgressPort
}

// setupCgroupDelegation prepares the worker pod's cgroup so runsc can create a
// per-actor-container leaf under it with real cpu/memory/pids accounting.
//
Expand Down
1 change: 1 addition & 0 deletions cmd/ateom-gvisor/runsc.go
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ func (r *runsc) shapeSpec(containerName string) error {
ContainerName: containerName,
DurableVolumes: r.durableVolumes,
Size: r.size,
ResolvConf: ateompath.ActorResolvConfPath(r.actorUID),
})
return ocispec.Save(bundle, spec)
}
Expand Down
89 changes: 89 additions & 0 deletions cmd/ateom-gvisor/sandboxnet.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
//go:build linux

// Copyright 2026 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package main

import (
"context"
"errors"
"fmt"
"io/fs"
"log/slog"
"os"
"path/filepath"

"github.com/agent-substrate/substrate/internal/ateomnet"
"github.com/agent-substrate/substrate/internal/ateompath"
"github.com/agent-substrate/substrate/internal/atunnel"
)

// prepareSandboxNetwork builds the actor's network and starts serving it.
func (s *AteomService) prepareSandboxNetwork(ctx context.Context, actorUID string) error {
if err := s.releaseSandboxNetwork(ctx); err != nil {
return err
}
session, err := ateomnet.ServeSandbox(ctx, ateomnet.SandboxNetworkConfig{

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Preserve the documented behavior when no egress gateway is configured.

gVisor setup and microVM setup now always redirect TCP into atunnel, but prepareActorEgress still returns nil for an absent gateway and activateActorNetworking then skips activating egress. The inactive handler closes every intercepted connection. The API server still defaults the gateway address to empty, and the protocol contract explicitly promises direct egress in that case.

Could we preserve this supported mode, or explicitly require a gateway and update the configuration contract? At present an actor can start successfully with all external TCP connectivity broken. This was traced through both runtimes; I did not run a complete boot without a gateway.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

After some side discussions: I think we want to change this to clearly state that egress is not supported without a gateway, but a gateway is not strictly required (perhaps your actor does not need egress).

If we really need it later we can add a no-capture mode, but it greatly simplifies things to always use atunnel, and atunnel will need an egress gateway for egress.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

plus 1 to ben

ActorUID: actorUID,
Veth: true,
EgressPort: s.atunnelEgressPort,
DNSPort: atunnel.DNSPort,
}, s.atunnelEgress, s.dnsRelay)
if err != nil {
return fmt.Errorf("while setting up the sandbox network: %w", err)
}

// Point the sandbox resolver at its gateway.
if _, err := actorResolvConf(actorUID); err != nil {
_ = session.Close(ctx)
return err
}

return s.sandbox.Replace(ctx, session)
}

// releaseSandboxNetwork stops serving the actor and takes its network down,
// along with the resolv.conf atelet's per-activation reset leaves behind.
func (s *AteomService) releaseSandboxNetwork(ctx context.Context) error {
Comment thread
BenTheElder marked this conversation as resolved.
if session := s.sandbox.Session(); session != nil {
if err := os.Remove(ateompath.ActorResolvConfPath(session.Network.ActorUID)); err != nil && !errors.Is(err, fs.ErrNotExist) {
slog.WarnContext(ctx, "Failed to remove the actor resolv.conf", slog.Any("err", err))
}
}
return s.sandbox.Close(ctx)
}

// actorResolvConf writes the resolver bind source outside the actor's rootfs.
func actorResolvConf(actorUID string) (string, error) {
pod, err := os.ReadFile("/etc/resolv.conf")
if err != nil {
return "", fmt.Errorf("reading the worker pod resolv.conf: %w", err)
}
path := ateompath.ActorResolvConfPath(actorUID)
if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
return "", fmt.Errorf("creating the actor directory: %w", err)
}
if err := os.WriteFile(path, ateomnet.SandboxResolvConf(pod), 0o644); err != nil {
return "", fmt.Errorf("writing the actor resolv.conf: %w", err)
}
return path, nil
}

// attachAtunnel completes setup after atunnel receives the service's dialer.
func (s *AteomService) attachAtunnel(ingress *atunnel.Server, egress *atunnel.Egress, egressPort uint16) {
s.atunnelIngress = ingress
s.atunnelEgress = egress
s.atunnelEgressPort = egressPort
}
4 changes: 1 addition & 3 deletions cmd/ateom-microvm/checkpoint.go
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,6 @@ import (
"path/filepath"
"time"

"github.com/agent-substrate/substrate/internal/ateomnet"

"github.com/agent-substrate/substrate/cmd/ateom-microvm/internal/ch"
"github.com/agent-substrate/substrate/cmd/ateom-microvm/internal/kata"
"github.com/agent-substrate/substrate/internal/ateompath"
Expand Down Expand Up @@ -398,7 +396,7 @@ func (s *AteomService) terminateWorkload(ctx context.Context, actorUID string) e
// the two views of "is an actor here" from disagreeing.
s.activeActor.Store(nil)

if err := ateomnet.CleanupActorNetwork(ctx, s.interiorNetNS); err != nil {
if err := s.releaseSandboxNetwork(ctx); err != nil {
errs = append(errs, fmt.Errorf("while cleaning up actor network: %w", err))
}
return errors.Join(errs...)
Expand Down
Loading
Loading