Skip to content

revert: Restore the edge gateway and NAT on bonded hosts #853

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
privateip wants to merge 1 commit into from
Closed

revert: Restore the edge gateway and NAT on bonded hosts #853

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
9 changes: 4 additions & 5 deletions cmd/galactic-gateway/dispatch_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ import (
"go.datum.net/galactic/internal/plumbing/ebpf/natattach"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpattach"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpdispatch"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpmode"
)

func requireRoot(t *testing.T) {
Expand Down Expand Up @@ -95,7 +94,7 @@ type gatewayProcess struct {
func startGateway(t *testing.T, d pinDirs) gatewayProcess {
t.Helper()
ctx, cancel := context.WithCancel(context.Background())
objs, dispatcher, public, _, leaseDone, err := joinDispatcher(ctx, d.dispatch, d.edge, xdpmode.Native, false, nil)
objs, dispatcher, public, _, leaseDone, err := joinDispatcher(ctx, d.dispatch, d.edge, false, nil)
if err != nil {
cancel()
t.Fatalf("gateway joinDispatcher: %v", err)
Expand Down Expand Up @@ -127,7 +126,7 @@ func startShard(t *testing.T, d pinDirs, uplink string) func() error {
if err := natattach.PopulateProgArray(objs); err != nil {
t.Fatalf("populate nat_progs: %v", err)
}
dispatcher, err := xdpdispatch.Open(context.Background(), d.dispatch, xdpmode.Native)
dispatcher, err := xdpdispatch.Open(context.Background(), d.dispatch)
if err != nil {
t.Fatalf("shard open dispatcher: %v", err)
}
Expand Down Expand Up @@ -166,7 +165,7 @@ func TestGatewayAndShardShareAnUplink(t *testing.T) {
shardCovered := startShard(t, d, uplink)
root := xdpProgID(t, uplink)

probe, err := xdpdispatch.Open(context.Background(), d.dispatch, xdpmode.Native)
probe, err := xdpdispatch.Open(context.Background(), d.dispatch)
if err != nil {
t.Fatalf("open dispatcher: %v", err)
}
Expand Down Expand Up @@ -203,7 +202,7 @@ func TestAttachDirect_RefusesAnUplinkTheShardRunsOn(t *testing.T) {
startShard(t, d, uplink)
root := xdpProgID(t, uplink)

_, _, _, err := attachDirect(context.Background(), d.dispatch, d.edge, xdpmode.Native, []string{uplink}, nil)
_, _, _, err := attachDirect(context.Background(), d.dispatch, d.edge, []string{uplink}, nil)
if !errors.Is(err, xdpattach.ErrDispatcherInUse) {
t.Fatalf("attachDirect over a live shard slot = %v, want ErrDispatcherInUse", err)
}
Expand Down
45 changes: 13 additions & 32 deletions cmd/galactic-gateway/gateway.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@ import (
"go.datum.net/galactic/internal/plumbing/ebpf/edgeprog"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpattach"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpdispatch"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpmode"
"go.datum.net/galactic/internal/plumbing/sysctl"
)

Expand Down Expand Up @@ -110,20 +109,6 @@ func setupGatewayDatapath(
overlap, publicInterface)
}

// One attach mode for every XDP attachment on the host, the public and
// internal targets alike (xdpmode). A target that does not exist, or a
// NIC whose driver cannot be read, fails startup rather than settling the
// mode without it.
modeDecision, err := xdpmode.Select(append(slices.Clone(targets), returnTargets...)...)
if err != nil {
return nil, fmt.Errorf("select the XDP attach mode for this host: %w", err)
}
mode := modeDecision.Mode
slog.Info("Selected the XDP attach mode for this host", "mode", mode.String(), "reason", modeDecision.Reason)
for _, skipped := range modeDecision.Skipped {
slog.Warn("Left a NIC with no driver out of the XDP attach mode decision", "reason", skipped)
}

// Required for the FIB lookup in the datapath's header push to succeed
// on the interface the program actually runs on. The lookup uses the
// ingress interface, meaning whichever resolved target the packet
Expand Down Expand Up @@ -166,7 +151,7 @@ func setupGatewayDatapath(
return nil, fmt.Errorf("register dispatcher lease metrics: %w", err)
}
objs, dispatcher, publicSet, returnSet, leaseDone, err = joinDispatcher(leaseCtx, xdpdispatch.PinDir,
edgeattach.PinDir, mode, len(returnTargets) > 0, lease)
edgeattach.PinDir, len(returnTargets) > 0, lease)
if err != nil {
stopLeases()
return nil, err
Expand All @@ -180,7 +165,7 @@ func setupGatewayDatapath(
}
}
} else {
objs, publicSet, returnSet, err = attachDirect(ctx, xdpdispatch.PinDir, edgeattach.PinDir, mode, targets,
objs, publicSet, returnSet, err = attachDirect(ctx, xdpdispatch.PinDir, edgeattach.PinDir, targets,
returnTargets)
}
if err != nil {
Expand Down Expand Up @@ -239,14 +224,11 @@ func (s dispatchSet) reconcile(ctx context.Context, targets []string) []string {
// them to the targets themselves, unpinned, after releasing an idle
// dispatcher from those targets. A dispatcher still serving the egress shard
// there is refused rather than taken: detaching it would cut egress.
// Attachment is all-or-nothing, as it has always been in this mode, and every
// attach is in mode.
func attachDirect(ctx context.Context, dispatchDir, edgeDir string, mode xdpmode.Mode,
targets, returnTargets []string,
) (
// Attachment is all-or-nothing, as it has always been in this mode.
func attachDirect(ctx context.Context, dispatchDir, edgeDir string, targets, returnTargets []string) (
*edgeprog.EdgedsrObjects, targetSet, targetSet, error,
) {
if err := xdpattach.ReleaseIdleDispatcher(ctx, dispatchDir, mode, append(slices.Clone(targets), returnTargets...),
if err := xdpattach.ReleaseIdleDispatcher(ctx, dispatchDir, append(slices.Clone(targets), returnTargets...),
xdpdispatch.SlotGatewayLB, xdpdispatch.SlotGatewayReturn); err != nil {
return nil, nil, nil, fmt.Errorf("%w; set %s=%s", err, config.EnvGatewayXDPAttach,
config.GatewayXDPAttachDispatch)
Expand All @@ -257,12 +239,12 @@ func attachDirect(ctx context.Context, dispatchDir, edgeDir string, mode xdpmode
return nil, nil, nil, fmt.Errorf("load edge gateway eBPF datapath: %w", err)
}

xdpLinks, err := edgeattach.Attach(objs.EdgeLb, targets, mode)
xdpLinks, err := edgeattach.Attach(objs.EdgeLb, targets)
if err != nil {
_ = objs.Close()
return nil, nil, nil, fmt.Errorf("attach edge gateway datapath to public interfaces %v: %w", targets, err)
}
publicSet, err := xdpattach.NewSet(objs.EdgeLb, targets, xdpLinks, mode)
publicSet, err := xdpattach.NewSet(objs.EdgeLb, targets, xdpLinks)
if err != nil {
closeAll(xdpLinks)
_ = objs.Close()
Expand All @@ -272,14 +254,14 @@ func attachDirect(ctx context.Context, dispatchDir, edgeDir string, mode xdpmode
return objs, directSet{publicSet}, nil, nil
}

returnLinks, err := edgeattach.Attach(objs.EdgeReturn, returnTargets, mode)
returnLinks, err := edgeattach.Attach(objs.EdgeReturn, returnTargets)
if err != nil {
closeAll(xdpLinks)
_ = objs.Close()
return nil, nil, nil, fmt.Errorf("attach edge gateway return datapath to internal interfaces %v: %w",
returnTargets, err)
}
returnSet, err := xdpattach.NewSet(objs.EdgeReturn, returnTargets, returnLinks, mode)
returnSet, err := xdpattach.NewSet(objs.EdgeReturn, returnTargets, returnLinks)
if err != nil {
closeAll(append(xdpLinks, returnLinks...))
_ = objs.Close()
Expand All @@ -295,14 +277,13 @@ func attachDirect(ctx context.Context, dispatchDir, edgeDir string, mode xdpmode
// without the role on any interface it never runs. Nothing is attached here;
// the returned sets put the dispatcher on the targets and read coverage back.
// The returned channel closes once both lease loops have stopped, which a
// caller waits on before closing the dispatcher. mode is the host's XDP attach
// mode, which the dispatcher attaches in and migrates its links to.
func joinDispatcher(ctx context.Context, dispatchDir, edgeDir string, mode xdpmode.Mode, withReturn bool,
// caller waits on before closing the dispatcher.
func joinDispatcher(ctx context.Context, dispatchDir, edgeDir string, withReturn bool,
lease *xdpdispatch.LeaseMetrics,
) (
*edgeprog.EdgedsrObjects, *xdpdispatch.Dispatcher, targetSet, targetSet, <-chan struct{}, error,
) {
dispatcher, err := xdpdispatch.Open(ctx, dispatchDir, mode)
dispatcher, err := xdpdispatch.Open(ctx, dispatchDir)
if err != nil {
return nil, nil, nil, nil, nil, fmt.Errorf("open the node's XDP dispatcher: %w", err)
}
Expand Down Expand Up @@ -363,7 +344,7 @@ func joinDispatcher(ctx context.Context, dispatchDir, edgeDir string, mode xdpmo
leases.Wait()
close(leaseDone)
}()
slog.Info("Edge gateway datapath installed in the XDP dispatcher", "pinDir", dispatchDir, "mode", mode.String())
slog.Info("Edge gateway datapath installed in the XDP dispatcher", "pinDir", dispatchDir)
return objs, dispatcher, dispatchSet{public}, ret, leaseDone, nil
}

Expand Down
9 changes: 4 additions & 5 deletions cmd/galactic-nat/dispatch_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ import (
"go.datum.net/galactic/internal/plumbing/ebpf/natattach"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpattach"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpdispatch"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpmode"
)

// testPinDirs returns per-test bpffs directories for the dispatcher and the
Expand Down Expand Up @@ -95,7 +94,7 @@ func joinForTest(t *testing.T, dispatchDir string, program *ebpf.Program) (*xdpd
) {
t.Helper()
ctx, cancel := context.WithCancel(context.Background())
d, set, leaseDone, err := joinDispatcher(ctx, dispatchDir, xdpmode.Native, program, nil)
d, set, leaseDone, err := joinDispatcher(ctx, dispatchDir, program, nil)
if err != nil {
cancel()
t.Fatalf("joinDispatcher: %v", err)
Expand Down Expand Up @@ -157,13 +156,13 @@ func TestReleaseIdleDispatcher_FreesTheUplinkForADirectAttach(t *testing.T) {
t.Fatalf("missing %v", missing)
}

if err := releaseIdleDispatcher(context.Background(), dispatchDir, xdpmode.Native, []string{uplink}); err != nil {
if err := releaseIdleDispatcher(context.Background(), dispatchDir, []string{uplink}); err != nil {
t.Fatalf("releaseIdleDispatcher: %v", err)
}
if got := xdpProgID(t, uplink); got != 0 {
t.Fatalf("uplink still runs program %d after release", got)
}
links, err := natattach.Attach(program, []string{uplink}, xdpmode.Native)
links, err := natattach.Attach(program, []string{uplink})
if err != nil {
t.Fatalf("direct attach after release: %v", err)
}
Expand Down Expand Up @@ -194,7 +193,7 @@ func TestReleaseIdleDispatcher_RefusesWhileAnotherSlotIsLive(t *testing.T) {
}
root := xdpProgID(t, uplink)

if err := releaseIdleDispatcher(context.Background(), dispatchDir, xdpmode.Native, []string{uplink}); err == nil {
if err := releaseIdleDispatcher(context.Background(), dispatchDir, []string{uplink}); err == nil {
t.Fatal("releaseIdleDispatcher succeeded with the gateway's slot live")
}
if got := xdpProgID(t, uplink); got != root {
Expand Down
39 changes: 10 additions & 29 deletions cmd/galactic-nat/nat.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@ import (
"go.datum.net/galactic/internal/plumbing/ebpf/natprog"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpattach"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpdispatch"
"go.datum.net/galactic/internal/plumbing/ebpf/xdpmode"
"go.datum.net/galactic/internal/plumbing/nicstats"
"go.datum.net/galactic/internal/plumbing/sysctl"
)
Expand Down Expand Up @@ -286,20 +285,6 @@ func setupNatDatapath(ctx context.Context, cfg *config.NATConfig,
return nil, err
}

// One attach mode for every XDP attachment on the host, shared with the
// edge gateway's (xdpmode). An uplink that does not exist, or a NIC whose
// driver cannot be read, fails startup rather than settling the mode
// without it.
modeDecision, err := xdpmode.Select(uplinks...)
if err != nil {
return nil, fmt.Errorf("select the XDP attach mode for this host: %w", err)
}
mode := modeDecision.Mode
slog.Info("Selected the XDP attach mode for this host", "mode", mode.String(), "reason", modeDecision.Reason)
for _, skipped := range modeDecision.Skipped {
slog.Warn("Left a NIC with no driver out of the XDP attach mode decision", "reason", skipped)
}

// Required for the FIB lookup in both the forward and return paths to
// succeed on the interface the program actually runs on. The lookup uses
// the ingress interface, meaning whichever uplink the packet arrived on, so
Expand Down Expand Up @@ -349,23 +334,22 @@ func setupNatDatapath(ctx context.Context, cfg *config.NATConfig,
_ = objs.Close()
return nil, fmt.Errorf("register dispatcher lease metrics: %w", err)
}
if dispatcher, dispatchSet, _, err = joinDispatcher(ctx, xdpdispatch.PinDir, mode, objs.NatIngress,
lease); err != nil {
if dispatcher, dispatchSet, _, err = joinDispatcher(ctx, xdpdispatch.PinDir, objs.NatIngress, lease); err != nil {
_ = objs.Close()
return nil, err
}
d.missing = dispatchSet.Reconcile(ctx, uplinks)
default:
if err := releaseIdleDispatcher(ctx, xdpdispatch.PinDir, mode, uplinks); err != nil {
if err := releaseIdleDispatcher(ctx, xdpdispatch.PinDir, uplinks); err != nil {
_ = objs.Close()
return nil, err
}
xdpLinks, err = natattach.Attach(objs.NatIngress, uplinks, mode)
xdpLinks, err = natattach.Attach(objs.NatIngress, uplinks)
if err != nil {
_ = objs.Close()
return nil, fmt.Errorf("attach egress translation datapath to uplink interfaces %v: %w", uplinks, err)
}
if set, err = xdpattach.NewSet(objs.NatIngress, uplinks, xdpLinks, mode); err != nil {
if set, err = xdpattach.NewSet(objs.NatIngress, uplinks, xdpLinks); err != nil {
closeAll(xdpLinks)
_ = objs.Close()
return nil, err
Expand Down Expand Up @@ -430,14 +414,11 @@ func registerUplinkQueueStats(metricsReg prometheus.Registerer, d *natDatapath)
// each renewal in lease, which may be nil. It
// returns the set that keeps the dispatcher on the uplinks, and a channel
// closed once the lease loop has stopped, which a caller waits on before
// closing the dispatcher. mode is the host's XDP attach mode, which the
// dispatcher attaches in and migrates its links to.
func joinDispatcher(ctx context.Context, pinDir string, mode xdpmode.Mode, program *ebpf.Program,
lease *xdpdispatch.LeaseMetrics,
) (
// closing the dispatcher.
func joinDispatcher(ctx context.Context, pinDir string, program *ebpf.Program, lease *xdpdispatch.LeaseMetrics) (
*xdpdispatch.Dispatcher, *xdpattach.DispatchSet, <-chan struct{}, error,
) {
dispatcher, err := xdpdispatch.Open(ctx, pinDir, mode)
dispatcher, err := xdpdispatch.Open(ctx, pinDir)
if err != nil {
return nil, nil, nil, fmt.Errorf("open the node's XDP dispatcher: %w", err)
}
Expand Down Expand Up @@ -468,15 +449,15 @@ func joinDispatcher(ctx context.Context, pinDir string, mode xdpmode.Mode, progr
"the dispatcher skips the slot once it lapses", "err", err)
}))
}()
slog.Info("Egress translation datapath installed in the XDP dispatcher", "pinDir", pinDir, "mode", mode.String())
slog.Info("Egress translation datapath installed in the XDP dispatcher", "pinDir", pinDir)
return dispatcher, set, leaseDone, nil
}

// releaseIdleDispatcher detaches an idle XDP dispatcher from uplinks before a
// direct attach (xdpattach.ReleaseIdleDispatcher), refusing while another
// datapath's slot is live there.
func releaseIdleDispatcher(ctx context.Context, pinDir string, mode xdpmode.Mode, uplinks []string) error {
if err := xdpattach.ReleaseIdleDispatcher(ctx, pinDir, mode, uplinks, xdpdispatch.SlotNAT); err != nil {
func releaseIdleDispatcher(ctx context.Context, pinDir string, uplinks []string) error {
if err := xdpattach.ReleaseIdleDispatcher(ctx, pinDir, uplinks, xdpdispatch.SlotNAT); err != nil {
return fmt.Errorf("%w; set %s=%s", err, config.EnvNATXDPAttach, config.NATXDPAttachDispatch)
}
return nil
Expand Down
2 changes: 1 addition & 1 deletion config/galactic-nat/kustomization.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
# deliberately does NOT include base/. A shard's identity -- its SID and
# per-family masquerade addresses -- comes from its EgressShard's spec, and
# its uplinks are auto-detected, so base/daemonset.yaml no longer needs any
# per-node value to start. But auto-detection attaches XDP to every
# per-node value to start. But auto-detection attaches native XDP to every
# interface carrying the IPv6 default route or a BGP-learned route, and on a
# node where that includes a management NIC the deployer has to name the
# fabric uplinks with GALACTIC_NAT_UPLINK_INTERFACES instead, which only
Expand Down
Loading
Loading