Files
moby/daemon/libnetwork/service_iptables_linux.go
Cory Snider 01573380f9 libnetwork: add nftables load-balancer impl
Add a native nftables implementation of service load balancing which
achieves feature parity with the iptables firewall mode. This is
accomplished using a static nftables ruleset with dynamic data. NAT and
DSR modes are both supported. The backend server for each flow is chosen
randomly rather than round-robin like the iptables+IPVS implementation.

As the load balancer endpoint only sees one side of DSR flows we cannot
use netfilter's conntrack subsystem. And IPVS does not play well with
nftables -- nftables is missing an analogue for the `ipvs` xtables
matcher. DSR flows are made sticky by implementing one-sided connection
tracking directly in the nftables ruleset using maps that are updated in
the packet path.

A network's load-balancer tables have exactly one writer: each
loadBalancer records the data plane it wants, and the complete element set
is composed from every recorded contribution. nat-publish-port is an
interval map keyed on `l4proto . dport . numgen`, so the bucket space for
a published port has to be partitioned by a single owner -- and a service
whose published-port set changes has a separate binding, hence a separate
loadBalancer, per generation until the rolling update drains. Two of them
partitioning a shared port independently would emit overlapping intervals,
which nftables rejects. Composing them instead partitions the port once
over every backend that serves it, so a shared published port is
load-balanced across both generations. IPVS cannot do that: one fwMark
cannot select a different backend set per port.

Signed-off-by: Cory Snider <csnider@mirantis.com>
2026-11-03 20:59:49 -05:00

273 lines
9.7 KiB
Go

package libnetwork
import (
"context"
"errors"
"fmt"
"net"
"os"
"strconv"
"strings"
"syscall"
"github.com/containerd/log"
"github.com/moby/ipvs"
"github.com/moby/moby/v2/daemon/libnetwork/iptables"
"github.com/vishvananda/netlink/nl"
)
// addLBBackendIPTables adds a loadbalancer backend to the loadbalancer sandbox
// for the network. If needed add the service as well.
//
// It reports whether the service's data plane is in place - true also when the
// IPVS service already existed, so that a caller which failed a later step (such
// as publishing ingress ports) can retry that step on the next backend event.
func (n *Network) addLBBackendIPTables(ip net.IP, ep *Endpoint, sb *Sandbox, lb *loadBalancer) bool {
eIP := ep.Iface().Address()
i, err := ipvs.New(sb.Key())
if err != nil {
log.G(context.TODO()).Errorf("Failed to create an ipvs handle for sbox %.7s (%.7s,%s) for lb addition: %v", sb.ID(), sb.ContainerID(), sb.Key(), err)
return false
}
defer i.Close()
s := &ipvs.Service{
AddressFamily: nl.FAMILY_V4,
FWMark: lb.fwMark,
SchedName: ipvs.RoundRobin,
}
if !i.IsServicePresent(s) {
// Add IP alias for the VIP to the endpoint
ifName := findIfaceDstName(sb, ep)
if ifName == "" {
log.G(context.TODO()).Errorf("Failed find interface name for endpoint %s(%s) to create LB alias", ep.ID(), ep.Name())
return false
}
err := sb.osSbox.AddAliasIP(ifName, &net.IPNet{IP: lb.vip, Mask: net.CIDRMask(32, 32)})
if errors.Is(err, syscall.EEXIST) {
log.G(context.TODO()).Debugf("IP alias %s already exists on network %s LB endpoint interface %s", lb.vip, n.ID(), ifName)
} else if err != nil {
log.G(context.TODO()).Errorf("Failed add IP alias %s to network %s LB endpoint interface %s: %v", lb.vip, n.ID(), ifName, err)
return false
}
log.G(context.TODO()).Debugf("Creating service for vip %s fwMark %d ingressPorts %#v in sbox %.7s (%.7s)", lb.vip, lb.fwMark, lb.service.ingressPorts, sb.ID(), sb.ContainerID())
if err := sb.configureFWMarkIPTables(lb.vip, lb.fwMark, lb.service.ingressPorts, eIP, false, n.loadBalancerMode); err != nil {
log.G(context.TODO()).Errorf("Failed to add firewall mark rule in sbox %.7s (%.7s): %v", sb.ID(), sb.ContainerID(), err)
return false
}
if err := i.NewService(s); err != nil && !errors.Is(err, syscall.EEXIST) {
log.G(context.TODO()).Errorf("Failed to create a new service for vip %s fwmark %d in sbox %.7s (%.7s): %v", lb.vip, lb.fwMark, sb.ID(), sb.ContainerID(), err)
return false
}
}
// Remove the sched name before using the service to add
// destination.
s.SchedName = ""
var flags uint32
if n.loadBalancerMode == loadBalancerModeDSR {
flags = ipvs.ConnFwdDirectRoute
}
err = i.NewDestination(s, &ipvs.Destination{
AddressFamily: nl.FAMILY_V4,
Address: ip,
Weight: 1,
ConnectionFlags: flags,
})
if err != nil && !errors.Is(err, syscall.EEXIST) {
log.G(context.TODO()).Errorf("Failed to create real server %s for vip %s fwmark %d in sbox %.7s (%.7s): %v", ip, lb.vip, lb.fwMark, sb.ID(), sb.ContainerID(), err)
}
// Ensure that kernel tweaks are applied in case this is the first time
// we've initialized ip_vs
sb.osSbox.ApplyOSTweaks(sb.oslTypes)
return true
}
// rmLBBackendIPTables removes a loadbalancer backend the load balancing
// endpoint for this network. If 'rmService' is true, then remove the service
// entry as well. If 'fullRemove' is true then completely remove the entry,
// otherwise just deweight it for now.
func (n *Network) rmLBBackendIPTables(ip net.IP, ep *Endpoint, sb *Sandbox, lb *loadBalancer, rmService bool, fullRemove bool) {
eIP := ep.Iface().Address()
i, err := ipvs.New(sb.Key())
if err != nil {
log.G(context.TODO()).Errorf("Failed to create an ipvs handle for sbox %.7s (%.7s,%s) for lb removal: %v", sb.ID(), sb.ContainerID(), sb.Key(), err)
return
}
defer i.Close()
s := &ipvs.Service{
AddressFamily: nl.FAMILY_V4,
FWMark: lb.fwMark,
}
d := &ipvs.Destination{
AddressFamily: nl.FAMILY_V4,
Address: ip,
Weight: 1,
}
if n.loadBalancerMode == loadBalancerModeDSR {
d.ConnectionFlags = ipvs.ConnFwdDirectRoute
}
if fullRemove {
if err := i.DelDestination(s, d); err != nil && !errors.Is(err, syscall.ENOENT) {
log.G(context.TODO()).Errorf("Failed to delete real server %s for vip %s fwmark %d in sbox %.7s (%.7s): %v", ip, lb.vip, lb.fwMark, sb.ID(), sb.ContainerID(), err)
}
} else {
d.Weight = 0
if err := i.UpdateDestination(s, d); err != nil && !errors.Is(err, syscall.ENOENT) {
log.G(context.TODO()).Errorf("Failed to set LB weight of real server %s to 0 for vip %s fwmark %d in sbox %.7s (%.7s): %v", ip, lb.vip, lb.fwMark, sb.ID(), sb.ContainerID(), err)
}
}
if rmService {
s.SchedName = ipvs.RoundRobin
if err := i.DelService(s); err != nil && !errors.Is(err, syscall.ENOENT) {
log.G(context.TODO()).Errorf("Failed to delete service for vip %s fwmark %d in sbox %.7s (%.7s): %v", lb.vip, lb.fwMark, sb.ID(), sb.ContainerID(), err)
}
if err := sb.configureFWMarkIPTables(lb.vip, lb.fwMark, lb.service.ingressPorts, eIP, true, n.loadBalancerMode); err != nil {
log.G(context.TODO()).Errorf("Failed to delete firewall mark rule in sbox %.7s (%.7s): %v", sb.ID(), sb.ContainerID(), err)
}
// Remove IP alias from the VIP to the endpoint
ifName := findIfaceDstName(sb, ep)
if ifName == "" {
log.G(context.TODO()).Errorf("Failed find interface name for endpoint %s(%s) to create LB alias", ep.ID(), ep.Name())
return
}
err := sb.osSbox.RemoveAliasIP(ifName, &net.IPNet{IP: lb.vip, Mask: net.CIDRMask(32, 32)})
if err != nil {
log.G(context.TODO()).Errorf("Failed to remove IP alias %s from network %s LB endpoint interface %s: %v", lb.vip, n.ID(), ifName, err)
}
}
}
// configureFWMarkIPTables configures the sandbox firewall to mark vip destined packets
// with the firewall mark fwMark.
func (sb *Sandbox) configureFWMarkIPTables(vip net.IP, fwMark uint32, ingressPorts []*PortConfig, eIP *net.IPNet, isDelete bool, lbMode string) error {
// TODO IPv6 support
iptable := iptables.GetIptable(iptables.IPv4)
fwMarkStr := strconv.FormatUint(uint64(fwMark), 10)
addDelOpt := "-A"
if isDelete {
addDelOpt = "-D"
}
rules := make([][]string, 0, len(ingressPorts))
for _, iPort := range ingressPorts {
var (
protocol = strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)])
publishedPort = strconv.FormatUint(uint64(iPort.PublishedPort), 10)
)
rule := []string{"-t", "mangle", addDelOpt, "PREROUTING", "-p", protocol, "--dport", publishedPort, "-j", "MARK", "--set-mark", fwMarkStr}
rules = append(rules, rule)
}
var innerErr error
err := sb.ExecFunc(func() {
if !isDelete && lbMode == loadBalancerModeNAT {
subnet := net.IPNet{IP: eIP.IP.Mask(eIP.Mask), Mask: eIP.Mask}
ruleParams := []string{"-m", "ipvs", "--ipvs", "-d", subnet.String(), "-j", "SNAT", "--to-source", eIP.IP.String()}
if !iptable.Exists("nat", "POSTROUTING", ruleParams...) {
rule := append([]string{"-t", "nat", "-A", "POSTROUTING"}, ruleParams...)
rules = append(rules, rule)
err := os.WriteFile("/proc/sys/net/ipv4/vs/conntrack", []byte{'1', '\n'}, 0o644)
if err != nil {
innerErr = err
return
}
}
}
rule := []string{"-t", "mangle", addDelOpt, "INPUT", "-d", vip.String() + "/32", "-j", "MARK", "--set-mark", fwMarkStr}
rules = append(rules, rule)
for _, rule := range rules {
if err := iptable.RawCombinedOutputNative(rule...); err != nil {
innerErr = fmt.Errorf("set up rule failed, %v: %w", rule, err)
return
}
}
})
if err != nil {
return err
}
return innerErr
}
func (sb *Sandbox) addRedirectRulesIPTables(eIP *net.IPNet, ingressPorts []*PortConfig) error {
// TODO IPv6 support
iptable := iptables.GetIptable(iptables.IPv4)
ipAddr := eIP.IP.String()
rules := make([][]string, 0, len(ingressPorts)*3) // 3 rules per port
for _, iPort := range ingressPorts {
var (
protocol = strings.ToLower(PortConfig_Protocol_name[int32(iPort.Protocol)])
publishedPort = strconv.FormatUint(uint64(iPort.PublishedPort), 10)
targetPort = strconv.FormatUint(uint64(iPort.TargetPort), 10)
)
rules = append(rules,
[]string{"-t", "nat", "-A", "PREROUTING", "-d", ipAddr, "-p", protocol, "--dport", publishedPort, "-j", "REDIRECT", "--to-port", targetPort},
// Allow only incoming connections to exposed ports
[]string{"-I", "INPUT", "-d", ipAddr, "-p", protocol, "--dport", targetPort, "-m", "conntrack", "--ctstate", "NEW,ESTABLISHED", "-j", "ACCEPT"},
// Allow only outgoing connections from exposed ports
[]string{"-I", "OUTPUT", "-s", ipAddr, "-p", protocol, "--sport", targetPort, "-m", "conntrack", "--ctstate", "ESTABLISHED", "-j", "ACCEPT"},
)
}
var innerErr error
err := sb.ExecFunc(func() {
for _, rule := range rules {
if err := iptable.RawCombinedOutputNative(rule...); err != nil {
innerErr = fmt.Errorf("set up rule failed, %v: %w", rule, err)
return
}
}
if len(ingressPorts) == 0 {
return
}
// Ensure blocking rules for anything else in/to ingress network
for _, rule := range [][]string{
{"-d", ipAddr, "-p", "sctp", "-j", "DROP"},
{"-d", ipAddr, "-p", "udp", "-j", "DROP"},
{"-d", ipAddr, "-p", "tcp", "-j", "DROP"},
} {
if !iptable.ExistsNative(iptables.Filter, "INPUT", rule...) {
if err := iptable.RawCombinedOutputNative(append([]string{"-A", "INPUT"}, rule...)...); err != nil {
innerErr = fmt.Errorf("set up rule failed, %v: %w", rule, err)
return
}
}
rule[0] = "-s"
if !iptable.ExistsNative(iptables.Filter, "OUTPUT", rule...) {
if err := iptable.RawCombinedOutputNative(append([]string{"-A", "OUTPUT"}, rule...)...); err != nil {
innerErr = fmt.Errorf("set up rule failed, %v: %w", rule, err)
return
}
}
}
})
if err != nil {
return err
}
return innerErr
}