Add masquerade mode

Signed-off-by: lou-lan <loulan@loulan.me>

cmd/kube-vip.go

@@ -17,6 +17,7 @@ import (
 	"github.com/kube-vip/kube-vip/pkg/equinixmetal"
 	"github.com/kube-vip/kube-vip/pkg/kubevip"
 	"github.com/kube-vip/kube-vip/pkg/manager"
+	"github.com/kube-vip/kube-vip/pkg/sysctl"
 	"github.com/kube-vip/kube-vip/pkg/vip"
 )
 
@@ -350,6 +351,14 @@ var kubeVipManager = &cobra.Command{
 			}
 		}
 
+		if initConfig.LoadBalancerForwardingMethod == "masquerade" {
+			log.Infof("sysctl set net.ipv4.vs.conntrack to 1")
+			err := sysctl.WriteProcSys("/proc/sys/net/ipv4/vs/conntrack", "1")
+			if err != nil {
+				log.Fatalln(err)
+			}
+		}
+
 		// Define the new service manager
 		mgr, err := manager.New(configMap, &initConfig)
 		if err != nil {

pkg/cluster/cluster.go

@@ -50,7 +50,7 @@ func startNetworking(c *kubevip.Config) ([]vip.Network, error) {
 
 	networks := []vip.Network{}
 	for _, addr := range addresses {
-		network, err := vip.NewConfig(addr, c.Interface, c.VIPSubnet, c.DDNS, c.RoutingTableID, c.RoutingTableType, c.RoutingProtocol, c.DNSMode)
+		network, err := vip.NewConfig(addr, c.Interface, c.VIPSubnet, c.DDNS, c.RoutingTableID, c.RoutingTableType, c.RoutingProtocol, c.DNSMode, c.LoadBalancerForwardingMethod, c.IptablesBackend)
 		if err != nil {
 			return nil, err
 		}

pkg/cluster/service.go

@@ -80,7 +80,7 @@ func (cluster *Cluster) vipService(ctxArp, ctxDNS context.Context, c *kubevip.Co
 
 			log.Infof("Starting IPVS LoadBalancer")
 
-			lb, err := loadbalancer.NewIPVSLB(cluster.Network[i].IP(), c.LoadBalancerPort, c.LoadBalancerForwardingMethod)
+			lb, err := loadbalancer.NewIPVSLB(cluster.Network[i].IP(), c.LoadBalancerPort, c.LoadBalancerForwardingMethod, c.BackendHealthCheckInterval)
 			if err != nil {
 				log.Errorf("Error creating IPVS LoadBalancer [%s]", err)
 			}

pkg/iptables/iptables.go

@@ -64,8 +64,12 @@ const (
 )
 
 const (
-	TableFilter = "filter"
-	ChainInput  = "INPUT"
+	TableFilter      = "filter"
+	TableMangle      = "mangle"
+	TableNat         = "nat"
+	ChainInput       = "INPUT"
+	ChainPREROUTING  = "PREROUTING"
+	ChainPOSTROUTING = "POSTROUTING"
 )
 
 type IPTables struct {

pkg/iptables/version.go

@@ -0,0 +1,91 @@
+package iptables
+
+import (
+	"fmt"
+	"os/exec"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+type Version struct {
+	Major       int
+	Minor       int
+	Patch       int
+	BackendMode string
+}
+
+func (v Version) String() string {
+	return fmt.Sprintf("%d.%d.%d", v.Major, v.Minor, v.Patch)
+}
+
+func (v Version) Compare(other Version) int {
+	if v.Major != other.Major {
+		return v.Major - other.Major
+	}
+	if v.Minor != other.Minor {
+		return v.Minor - other.Minor
+	}
+	return v.Patch - other.Patch
+}
+
+func ParseVersion(versionString string) (Version, error) {
+	re := regexp.MustCompile(`v([0-9]+)\.([0-9]+)\.([0-9]+)`)
+	match := re.FindStringSubmatch(versionString)
+	if len(match) != 4 {
+		return Version{}, fmt.Errorf("invalid version string: %s", versionString)
+	}
+	major, _ := strconv.Atoi(match[1])
+	minor, _ := strconv.Atoi(match[2])
+	patch, _ := strconv.Atoi(match[3])
+	return Version{Major: major, Minor: minor, Patch: patch}, nil
+}
+
+func GetVersion() (Version, error) {
+	ver := Version{}
+	cmd := exec.Command("iptables", "--version")
+	out, err := cmd.Output()
+	if err != nil {
+		return ver, fmt.Errorf("run cmd 'iptables --version' wtith error: %v", err)
+	}
+
+	ver, err = ParseVersion(string(out))
+	if err != nil {
+		return ver, err
+	}
+
+	nft4 := getOutput("iptables-nft-save")
+	legacy4 := getOutput("iptables-legacy-save")
+
+	nft6 := getOutput("ip6tables-nft-save")
+	legacy6 := getOutput("ip6tables-legacy-save")
+
+	if strings.Contains(nft4, "KUBE-IPTABLES") ||
+		strings.Contains(nft6, "KUBE-IPTABLES") ||
+		strings.Contains(nft4, "KUBE-KUBELET") ||
+		strings.Contains(nft6, "KUBE-KUBELET") {
+		ver.BackendMode = "nft"
+	} else if strings.Contains(legacy4, "KUBE-IPTABLES") ||
+		strings.Contains(legacy6, "KUBE-IPTABLES") ||
+		strings.Contains(legacy4, "KUBE-KUBELET") ||
+		strings.Contains(legacy6, "KUBE-KUBELET") {
+		ver.BackendMode = "legacy"
+	} else {
+		nftCount := strings.Count(nft4, "\n") + strings.Count(nft6, "\n")
+		legacyCount := strings.Count(legacy4, "\n") + strings.Count(legacy6, "\n")
+
+		if nftCount >= legacyCount {
+			ver.BackendMode = "nft"
+		} else {
+			ver.BackendMode = "legacy"
+		}
+	}
+
+	return ver, nil
+}
+
+func getOutput(name string) string {
+	cmd := exec.Command(name)
+	out, _ := cmd.Output()
+	return string(out)
+}

pkg/kubevip/config_environment.go

@@ -162,6 +162,11 @@ func ParseEnvironment(c *Config) error {
 		c.DetectControlPlane = b
 	}
 
+	env = os.Getenv(kubernetesAddr)
+	if env != "" {
+		c.KubernetesAddr = env
+	}
+
 	// Find Services toggle
 	env = os.Getenv(svcEnable)
 	if env != "" {
@@ -570,5 +575,19 @@ func ParseEnvironment(c *Config) error {
 		c.EnableEndpointSlices = b
 	}
 
+	env = os.Getenv(iptablesBackend)
+	if env != "" {
+		c.IptablesBackend = env
+	}
+
+	env = os.Getenv(backendHealthCheckInterval)
+	if env != "" {
+		i, err := strconv.ParseInt(env, 10, 32)
+		if err != nil {
+			return err
+		}
+		c.BackendHealthCheckInterval = int(i)
+	}
+
 	return nil
 }

pkg/kubevip/config_envvar.go

@@ -151,6 +151,9 @@ const (
 	// cpDetect will attempt to automatically find a working address for the control plane from loopback
 	cpDetect = "cp_detect"
 
+	// kubernetesAddr，is the address of the Kubernetes API server on this machine
+	kubernetesAddr = "kubernetes_addr"
+
 	// svcEnable enables the Kubernetes service feature
 	svcEnable = "svc_enable"
 
@@ -190,7 +193,7 @@ const (
 	// vipConfigMap defines the configmap that kube-vip will watch for service definitions
 	// vipConfigMap = "vip_configmap"
 
-	//k8sConfigFile defines the path to the configfile used to speak with the API server
+	// k8sConfigFile defines the path to the configfile used to speak with the API server
 	k8sConfigFile = "k8s_config_file"
 
 	// dnsMode defines mode that DNS lookup will be performed with (first, ipv4, ipv6, dual)
@@ -201,4 +204,10 @@ const (
 
 	// enableEndpointSlices enables use of EndpointSlices instead of Endpoints
 	enableEndpointSlices = "enable_endpointslices"
+
+	// iptablesBackend iptables backend, can be specified as `nft` or `legacy`. If not set, it defaults to automatic detection.
+	iptablesBackend = "iptables_backend"
+
+	// backendHealthCheckInterval Interval in seconds for checking backend health.
+	backendHealthCheckInterval = "backend_health_check_interval"
 )

pkg/kubevip/config_generator.go

@@ -505,6 +505,23 @@ func generatePodSpec(c *Config, imageVersion string, inCluster bool) *corev1.Pod
 		newEnvironment = append(newEnvironment, disServiceUpdates...)
 	}
 
+	var securityContext *corev1.SecurityContext
+	if c.LoadBalancerForwardingMethod == "masquerade" {
+		var privileged = true
+		securityContext = &corev1.SecurityContext{
+			Privileged: &privileged,
+		}
+	} else {
+		securityContext = &corev1.SecurityContext{
+			Capabilities: &corev1.Capabilities{
+				Add: []corev1.Capability{
+					"NET_ADMIN",
+					"NET_RAW",
+				},
+			},
+		}
+	}
+
 	newManifest := &corev1.Pod{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "Pod",
@@ -520,14 +537,7 @@ func generatePodSpec(c *Config, imageVersion string, inCluster bool) *corev1.Pod
 					Name:            "kube-vip",
 					Image:           fmt.Sprintf("ghcr.io/kube-vip/kube-vip:%s", imageVersion),
 					ImagePullPolicy: corev1.PullAlways,
-					SecurityContext: &corev1.SecurityContext{
-						Capabilities: &corev1.Capabilities{
-							Add: []corev1.Capability{
-								"NET_ADMIN",
-								"NET_RAW",
-							},
-						},
-					},
+					SecurityContext: securityContext,
 					Args: []string{
 						command,
 					},

pkg/kubevip/config_types.go

@@ -27,6 +27,9 @@ type Config struct {
 	// DetectControlPlane, will attempt to find the control plane from loopback (127.0.0.1)
 	DetectControlPlane bool `yaml:"detectControlPlane"`
 
+	// KubernetesAddr，is the address of the Kubernetes API server on this machine
+	KubernetesAddr string `yaml:"kubernetesAddr"`
+
 	// EnableServices, will enable the services functionality (used for hybrid behaviour)
 	EnableServices bool `yaml:"enableServices"`
 
@@ -172,6 +175,12 @@ type Config struct {
 
 	// EnableEndpointSlices, if enabled, EndpointSlices will be used instead of Endpoints
 	EnableEndpointSlices bool `yaml:"enableEndpointSlices"`
+
+	// IptablesBackend iptables backend, can be specified as `nft` or `legacy`. If not set, it defaults to automatic detection.
+	IptablesBackend string `yaml:"iptablesBackend"`
+
+	// BackendHealthCheckInterval Interval in seconds for checking backend health.
+	BackendHealthCheckInterval int `yaml:"backendHealthCheckInterval"`
 }
 
 // KubernetesLeaderElection defines all of the settings for Kubernetes KubernetesLeaderElection

pkg/loadbalancer/ipvs.go

@@ -5,9 +5,12 @@ import (
 	"net"
 	"net/netip"
 	"strings"
+	"sync"
+	"time"
 
 	"github.com/cloudflare/ipvs"
 	"github.com/cloudflare/ipvs/netmask"
+	"github.com/kube-vip/kube-vip/pkg/k8s"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -33,14 +36,23 @@ const (
 	ROUNDROBIN = "rr"
 )
 
+type Backend struct {
+	Addr string
+	Port int
+}
+
 type IPVSLoadBalancer struct {
 	client              ipvs.Client
 	loadBalancerService ipvs.Service
 	Port                int
 	forwardingMethod    ipvs.ForwardType
+	backendMap          map[Backend]bool
+	interval            int
+	lock                sync.Mutex
+	stop                chan struct{}
 }
 
-func NewIPVSLB(address string, port int, forwardingMethod string) (*IPVSLoadBalancer, error) {
+func NewIPVSLB(address string, port int, forwardingMethod string, backendHealthCheckInterval int) (*IPVSLoadBalancer, error) {
 	// Create IPVS client
 	c, err := ipvs.New()
 	if err != nil {
@@ -88,17 +100,29 @@ func NewIPVSLB(address string, port int, forwardingMethod string) (*IPVSLoadBala
 		log.Warnf("unknown forwarding method. Defaulting to Local")
 	}
 
+	if backendHealthCheckInterval <= 0 {
+		backendHealthCheckInterval = 5
+	}
+
 	lb := &IPVSLoadBalancer{
 		Port:                port,
 		client:              c,
 		loadBalancerService: svc,
 		forwardingMethod:    m,
+		interval:            backendHealthCheckInterval,
+		backendMap:          make(map[Backend]bool),
 	}
+
+	if strings.ToLower(forwardingMethod) == "masquerade" {
+		go lb.healthCheck()
+	}
+
 	// Return our created load-balancer
 	return lb, nil
 }
 
 func (lb *IPVSLoadBalancer) RemoveIPVSLB() error {
+	close(lb.stop)
 	err := lb.client.RemoveService(lb.loadBalancerService)
 	if err != nil {
 		return fmt.Errorf("error removing existing IPVS service: %v", err)
@@ -107,6 +131,24 @@ func (lb *IPVSLoadBalancer) RemoveIPVSLB() error {
 }
 
 func (lb *IPVSLoadBalancer) AddBackend(address string, port int) error {
+	backend := Backend{Addr: address, Port: port}
+
+	lb.lock.Lock()
+	defer lb.lock.Unlock()
+	if _, ok := lb.backendMap[backend]; !ok {
+		isHealth := lb.checkBackend(backend)
+		if isHealth {
+			err := lb.addBackend(address, port)
+			if err != nil {
+				return err
+			}
+		}
+		lb.backendMap[backend] = isHealth
+	}
+	return nil
+}
+
+func (lb *IPVSLoadBalancer) addBackend(address string, port int) error {
 	// Check if this is the first backend
 	backends, err := lb.client.Destinations(lb.loadBalancerService)
 	if err != nil && strings.Contains(err.Error(), "file does not exist") {
@@ -166,6 +208,23 @@ func (lb *IPVSLoadBalancer) AddBackend(address string, port int) error {
 }
 
 func (lb *IPVSLoadBalancer) RemoveBackend(address string, port int) error {
+	backend := Backend{Addr: address, Port: port}
+
+	lb.lock.Lock()
+	defer lb.lock.Unlock()
+
+	if _, ok := lb.backendMap[backend]; ok {
+		err := lb.removeBackend(address, port)
+		if err != nil {
+			return err
+		}
+		delete(lb.backendMap, backend)
+	}
+
+	return nil
+}
+
+func (lb *IPVSLoadBalancer) removeBackend(address string, port int) error {
 	ip, family := ipAndFamily(address)
 	if family != lb.loadBalancerService.Family {
 		return nil
@@ -195,3 +254,59 @@ func ipAndFamily(address string) (netip.Addr, ipvs.AddressFamily) {
 	}
 	return netip.AddrFrom4([4]byte(ipAddr.To4())), ipvs.INET
 }
+
+func (lb *IPVSLoadBalancer) healthCheck() {
+	ticker := time.NewTicker(time.Second * time.Duration(lb.interval))
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-lb.stop:
+			ticker.Stop()
+			return
+		case <-ticker.C:
+			ticker.Stop()
+			lb.lock.Lock()
+			for backend, oldStatus := range lb.backendMap {
+				newStatus := lb.checkBackend(backend)
+				if newStatus {
+					// old status -> health
+					if !oldStatus {
+						err := lb.AddBackend(backend.Addr, backend.Port)
+						if err != nil {
+							log.Errorf("failed to add backend: %s", err)
+						}
+						lb.backendMap[backend] = newStatus
+					}
+				} else {
+					// old status -> not health
+					if oldStatus {
+						log.Infof("healthCheck failed for backend %s:%d, attempting to remove from load balancer", backend.Addr, backend.Port)
+						err := lb.removeBackend(backend.Addr, backend.Port)
+						if err != nil {
+							log.Errorf("failed to remove backend %s:%d: %s", backend.Addr, backend.Port, err)
+						}
+						lb.backendMap[backend] = newStatus
+					}
+				}
+			}
+			lb.lock.Unlock()
+			ticker.Reset(time.Second * time.Duration(lb.interval))
+		}
+	}
+}
+
+func (lb *IPVSLoadBalancer) checkBackend(backend Backend) bool {
+	adminConfigPath := "/etc/kubernetes/admin.conf"
+	client, err := k8s.NewClientset(adminConfigPath, false, fmt.Sprintf("%s:%v", backend.Addr, backend.Port))
+	if err != nil {
+		log.Infof("failed to new clientset: %s", err)
+		return false
+	}
+	_, err = client.DiscoveryClient.ServerVersion()
+	if err != nil {
+		log.Infof("failed check k8s server version: %s", err)
+		return false
+	}
+	return true
+}

pkg/manager/manager.go

@@ -71,7 +71,10 @@ func New(configMap string, config *kubevip.Config) (*Manager, error) {
 	case config.LeaderElectionType == "etcd":
 		// Do nothing, we don't construct a k8s client for etcd leader election
 	case fileExists(adminConfigPath):
-		if config.EnableControlPlane {
+		if config.KubernetesAddr != "" {
+			fmt.Println(config.KubernetesAddr)
+			clientset, err = k8s.NewClientset(adminConfigPath, false, config.KubernetesAddr)
+		} else if config.EnableControlPlane {
 			// If this is a control plane host it will likely have started as a static pod or won't have the
 			// VIP up before trying to connect to the API server, we set the API endpoint to this machine to
 			// ensure connectivity.

pkg/sysctl/sysctl.go

@@ -0,0 +1,29 @@
+package sysctl
+
+import (
+	"fmt"
+	"io"
+	"os"
+)
+
+func WriteProcSys(path, value string) error {
+	f, err := os.OpenFile(path, os.O_WRONLY, 0)
+	if err != nil {
+		return fmt.Errorf("failed to open file: %w", err)
+	}
+	defer func() {
+		if cErr := f.Close(); cErr != nil && err == nil {
+			err = fmt.Errorf("failed to close file: %w", cErr)
+		}
+	}()
+
+	n, err := f.Write([]byte(value))
+	if err != nil {
+		return fmt.Errorf("failed to write value: %w", err)
+	}
+	if n < len(value) {
+		return io.ErrShortWrite
+	}
+
+	return nil
+}

pkg/vip/address.go

@@ -21,6 +21,7 @@ import (
 const (
 	defaultValidLft                 = 60
 	iptablesComment                 = "%s kube-vip load balancer IP"
+	iptablesCommentMarkRule         = "kube-vip load balancer IP set mark for masquerade"
 	ignoreServiceSecurityAnnotation = "kube-vip.io/ignore-service-security"
 )
 
@@ -57,6 +58,9 @@ type network struct {
 	dnsName string
 	isDDNS  bool
 
+	forwardMethod   string
+	iptablesBackend string
+
 	routeTable       int
 	routingTableType int
 	routingProtocol  int
@@ -71,7 +75,7 @@ func netlinkParse(addr string) (*netlink.Addr, error) {
 }
 
 // NewConfig will attempt to provide an interface to the kernel network configuration
-func NewConfig(address string, iface string, subnet string, isDDNS bool, tableID int, tableType int, routingProtocol int, dnsMode string) ([]Network, error) {
+func NewConfig(address string, iface string, subnet string, isDDNS bool, tableID int, tableType int, routingProtocol int, dnsMode, forwardMethod, iptablesBackend string) ([]Network, error) {
 	networks := []Network{}
 
 	if IsIP(address) {
@@ -86,6 +90,8 @@ func NewConfig(address string, iface string, subnet string, isDDNS bool, tableID
 		result.routeTable = tableID
 		result.routingTableType = tableType
 		result.routingProtocol = routingProtocol
+		result.forwardMethod = forwardMethod
+		result.iptablesBackend = iptablesBackend
 
 		// Check if the subnet needs overriding
 		if subnet != "" {
@@ -243,6 +249,12 @@ func (configurator *network) AddIP() error {
 		}
 	}
 
+	if configurator.forwardMethod == "masquerade" {
+		if err := configurator.addIptablesRulesForMasquerade(); err != nil {
+			return errors.Wrap(err, "could not add iptables rules for masquerade")
+		}
+	}
+
 	return nil
 }
 
@@ -396,6 +408,72 @@ func (configurator *network) DeleteIP() error {
 		}
 	}
 
+	if configurator.forwardMethod == "masquerade" {
+		if err := configurator.removeIptablesRulesForMasquerade(); err != nil {
+			return errors.Wrap(err, "could not remove iptables masquerade rules ")
+		}
+	}
+
+	return nil
+}
+
+func (configurator *network) addIptablesRulesForMasquerade() error {
+	ver, err := iptables.GetVersion()
+	if err != nil {
+		return errors.Wrap(err, "could not get iptables version")
+	}
+
+	ipt, err := iptables.New(iptables.EnableNFTables(ver.BackendMode == "nft"))
+	if err != nil {
+		return errors.Wrap(err, "could not create iptables client")
+	}
+
+	vip := configurator.address.IP.String()
+	comment := fmt.Sprintf(iptablesComment, vip)
+	if err := addMasqueradeRuleForVIP(ipt, vip, comment); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// addIptablesRulesForMasquerade add iptables rules for MASQUERADE
+// insert example
+func (configurator *network) removeIptablesRulesForMasquerade() error {
+	ver, err := iptables.GetVersion()
+	if err != nil {
+		return errors.Wrap(err, "could not get iptables version")
+	}
+	ipt, err := iptables.New(iptables.EnableNFTables(ver.BackendMode == "nft"))
+	if err != nil {
+		return errors.Wrap(err, "could not create iptables client")
+	}
+	vip := configurator.address.IP.String()
+	comment := fmt.Sprintf(iptablesComment, configurator.serviceName)
+
+	err = delMasqueradeRuleForVIP(ipt, vip, comment)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func addMasqueradeRuleForVIP(ipt *iptables.IPTables, vip, comment string) error {
+	err := ipt.InsertUnique(iptables.TableNat, iptables.ChainPOSTROUTING,
+		1, "-m", "ipvs", "--vaddr", vip, "-j", "MASQUERADE", "-m", "comment", "--comment", comment)
+	if err != nil {
+		return fmt.Errorf("could not add masquerade rule for VIP %s: %v", vip, err)
+	}
+	return nil
+}
+
+func delMasqueradeRuleForVIP(ipt *iptables.IPTables, vip, comment string) error {
+	err := ipt.DeleteIfExists(iptables.TableNat, iptables.ChainPOSTROUTING,
+		"-d", "-m", "ipvs", "--vaddr", vip, "-j", "MASQUERADE", "-m", "comment", "--comment", comment)
+	if err != nil {
+		return fmt.Errorf("could not del masquerade rule for VIP %s: %v", vip, err)
+	}
 	return nil
 }