From 60c786b53725361a77e9405f3678a448dc0d1995 Mon Sep 17 00:00:00 2001 From: Gabriele Pennacchia Date: Tue, 14 Jul 2026 09:47:34 +0200 Subject: [PATCH] fix(egress): isolate nftables tables by instance Signed-off-by: Gabriele Pennacchia --- CHANGELOG.md | 1 + cmd/kube-vip.go | 1 + pkg/egress/egress.go | 4 +- pkg/kubevip/annotations.go | 3 + pkg/kubevip/config_environment.go | 14 +- pkg/kubevip/config_envvar.go | 3 + pkg/kubevip/config_file_test.go | 26 ++-- pkg/kubevip/config_generator.go | 6 + pkg/kubevip/config_generator_test.go | 65 +++++++++ pkg/kubevip/config_types.go | 3 + pkg/kubevip/config_validation.go | 38 +++++ pkg/kubevip/config_validation_test.go | 38 +++++ pkg/manager/manager.go | 3 +- pkg/manager/manager_dump.go | 2 + pkg/nftables/nftables.go | 200 +++++++++++++++++++++----- pkg/nftables/nftables_test.go | 109 ++++++++++++++ pkg/services/egress.go | 70 ++++++++- pkg/services/processor.go | 9 ++ pkg/services/services.go | 77 +++++++++- 19 files changed, 613 insertions(+), 59 deletions(-) create mode 100644 pkg/nftables/nftables_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index ef04c67f..000feefb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Fixed +- Add a configurable kube-vip instance name and use it to isolate internal nftables egress tables, persist table ownership on Services, and migrate per-Service chains without affecting other deployments. Fixes #1634. - Retry on 403 Forbidden and 401 Unauthorized in `ServicesWatcher` at startup with exponential backoff. Fixes #1464. - Reintroduce BGP config via node annotations. Fixes #1488. - Fail fast in runtime `manager` and `service` paths when legacy `vip_address` is used without `vip_subnet` in control-plane ARP, BGP, or Routing Table mode. diff --git a/cmd/kube-vip.go b/cmd/kube-vip.go index 503dc421..d8e721bd 100644 --- a/cmd/kube-vip.go +++ b/cmd/kube-vip.go @@ -167,6 +167,7 @@ func init() { // Configuration file flag kubeVipCmd.PersistentFlags().StringVar(&initConfig.ConfigFile, "config-file", "", "Path to a JSON/YAML configuration file to load settings from") + kubeVipCmd.PersistentFlags().StringVar(&initConfig.InstanceName, "instanceName", "", "Unique name for this kube-vip instance (currently used to isolate nftables egress tables)") kubeVipCmd.PersistentFlags().BoolVar(&initConfig.EgressWithNftables, "egressWithNftables", true, "Use nftables-based egress implementation") diff --git a/pkg/egress/egress.go b/pkg/egress/egress.go index 07e3ad49..af45ded0 100644 --- a/pkg/egress/egress.go +++ b/pkg/egress/egress.go @@ -19,15 +19,13 @@ func Teardown(podIP, vipIP, namespace, serviceUUID string, annotations map[strin internalEgress := annotations[kubevip.EgressInternal] protocol := iptables.ProtocolIPv4 - IPv6 := false if utils.IsIPv6(podIP) { protocol = iptables.ProtocolIPv6 - IPv6 = true } // Use the internal egress implementation if internalEgress != "" || useNftables { - return nftables.DeleteSNAT(IPv6, serviceUUID) + return nftables.DeleteSNATFromAllTables(serviceUUID) } i, err := vip.CreateIptablesClient(useNftables, namespace, protocol) diff --git a/pkg/kubevip/annotations.go b/pkg/kubevip/annotations.go index 3956d26f..a1c2defd 100644 --- a/pkg/kubevip/annotations.go +++ b/pkg/kubevip/annotations.go @@ -37,6 +37,9 @@ const ( // The current active endpoint(pod) for the Egress VIP (v6) ActiveEndpointIPv6 = "kube-vip.io/active-endpoint-ipv6" + // The nftables egress table base name that owns this Service's SNAT chain + EgressNftablesTable = "kube-vip.io/egress-nftables-table" + // Flush the conntrack rules (remove existing sessions) once Egress is configured FlushContrack = "kube-vip.io/flush-conntrack" diff --git a/pkg/kubevip/config_environment.go b/pkg/kubevip/config_environment.go index 1c8a56a5..8182a974 100644 --- a/pkg/kubevip/config_environment.go +++ b/pkg/kubevip/config_environment.go @@ -32,6 +32,13 @@ func ParseEnvironment(c *Config) error { c.Logging = int32(logLevel) } + if env = os.Getenv(instanceName); env == "" { + env = os.Getenv(strings.ToUpper(instanceName)) + } + if env != "" { + c.InstanceName = env + } + // Find interface env = os.Getenv(vipInterface) if env != "" { @@ -908,7 +915,6 @@ func mergeConfigValues(baseConfig, fileConfig *Config) { if baseConfig.ServicesLeaseName == "" && fileConfig.ServicesLeaseName != "" { baseConfig.ServicesLeaseName = fileConfig.ServicesLeaseName } - // LoadBalancer configuration if baseConfig.LoadBalancerPort == 0 && fileConfig.LoadBalancerPort != 0 { baseConfig.LoadBalancerPort = fileConfig.LoadBalancerPort @@ -970,6 +976,11 @@ func mergeConfigValues(baseConfig, fileConfig *Config) { baseConfig.HealthCheckPort = fileConfig.HealthCheckPort } + // Instance configuration + if baseConfig.InstanceName == "" && fileConfig.InstanceName != "" { + baseConfig.InstanceName = fileConfig.InstanceName + } + // Egress configuration if baseConfig.EgressPodCidr == "" && fileConfig.EgressPodCidr != "" { baseConfig.EgressPodCidr = fileConfig.EgressPodCidr @@ -977,7 +988,6 @@ func mergeConfigValues(baseConfig, fileConfig *Config) { if baseConfig.EgressServiceCidr == "" && fileConfig.EgressServiceCidr != "" { baseConfig.EgressServiceCidr = fileConfig.EgressServiceCidr } - // Mirror configuration if baseConfig.MirrorDestInterface == "" && fileConfig.MirrorDestInterface != "" { baseConfig.MirrorDestInterface = fileConfig.MirrorDestInterface diff --git a/pkg/kubevip/config_envvar.go b/pkg/kubevip/config_envvar.go index a549bf30..a8c41f5d 100644 --- a/pkg/kubevip/config_envvar.go +++ b/pkg/kubevip/config_envvar.go @@ -33,6 +33,9 @@ const ( // vipLogLevel - defines the level of logging to produce (5 being the most verbose) vipLogLevel = "vip_loglevel" + // instanceName identifies this kube-vip deployment when naming host-global resources + instanceName = "instance_name" + // vipInterface - defines the interface that the vip should bind too vipInterface = "vip_interface" diff --git a/pkg/kubevip/config_file_test.go b/pkg/kubevip/config_file_test.go index 8792f87a..10152252 100644 --- a/pkg/kubevip/config_file_test.go +++ b/pkg/kubevip/config_file_test.go @@ -34,6 +34,7 @@ address: "192.168.1.100" port: 6443 interface: "eth0" namespace: "kube-system" +instanceName: "release_a" vipSubnet: "192.168.1.0/24" leaseName: "test-lease" leaseDuration: 15 @@ -50,6 +51,7 @@ prometheusHTTPServer: ":2112" Port: 6443, Interface: "eth0", Namespace: "kube-system", + InstanceName: "release_a", VIPSubnet: "192.168.1.0/24", PrometheusHTTPServer: ":2112", KubernetesLeaderElection: KubernetesLeaderElection{ @@ -239,6 +241,9 @@ bgpConfig: if config.Interface != tt.expectedConfig.Interface { t.Errorf("Interface = %v, expected %v", config.Interface, tt.expectedConfig.Interface) } + if config.InstanceName != tt.expectedConfig.InstanceName { + t.Errorf("InstanceName = %v, expected %v", config.InstanceName, tt.expectedConfig.InstanceName) + } // Test BGP config if present if tt.expectedConfig.EnableBGP { @@ -420,16 +425,18 @@ func TestMergeConfigValues(t *testing.T) { Port: 0, // Should be overridden }, fileConfig: &Config{ - Logging: 2, - Port: 6443, - Interface: "eth0", - Address: "192.168.1.100", + Logging: 2, + Port: 6443, + Interface: "eth0", + Address: "192.168.1.100", + InstanceName: "release_a", }, expectedBase: &Config{ - Logging: 5, // From base (non-zero) - Port: 6443, // From file (base was zero) - Interface: "eth0", // From file (base was empty) - Address: "192.168.1.100", // From file (base was empty) + Logging: 5, // From base (non-zero) + Port: 6443, // From file (base was zero) + Interface: "eth0", // From file (base was empty) + Address: "192.168.1.100", // From file (base was empty) + InstanceName: "release_a", // From file (base was empty) }, }, { @@ -516,6 +523,9 @@ func TestMergeConfigValues(t *testing.T) { if tt.baseConfig.Interface != tt.expectedBase.Interface { t.Errorf("Interface = %v, expected %v", tt.baseConfig.Interface, tt.expectedBase.Interface) } + if tt.baseConfig.InstanceName != tt.expectedBase.InstanceName { + t.Errorf("InstanceName = %v, expected %v", tt.baseConfig.InstanceName, tt.expectedBase.InstanceName) + } if tt.baseConfig.EnableARP != tt.expectedBase.EnableARP { t.Errorf("EnableARP = %v, expected %v", tt.baseConfig.EnableARP, tt.expectedBase.EnableARP) } diff --git a/pkg/kubevip/config_generator.go b/pkg/kubevip/config_generator.go index 344bab32..91c352c3 100644 --- a/pkg/kubevip/config_generator.go +++ b/pkg/kubevip/config_generator.go @@ -202,6 +202,12 @@ func generatePodSpec(c *Config, image, imageVersion string, inCluster bool) (*co }, }, } + if c.InstanceName != "" { + newEnvironment = append(newEnvironment, corev1.EnvVar{ + Name: instanceName, + Value: c.InstanceName, + }) + } // If we're specifically saying which interface to use then add it to the manifest if c.Interface != "" { diff --git a/pkg/kubevip/config_generator_test.go b/pkg/kubevip/config_generator_test.go index bea1af6a..73324759 100644 --- a/pkg/kubevip/config_generator_test.go +++ b/pkg/kubevip/config_generator_test.go @@ -2,6 +2,7 @@ package kubevip import ( "os" + "strings" "testing" ) @@ -25,6 +26,70 @@ func TestParseEnvironment(t *testing.T) { } } +func TestParseEnvironmentInstanceName(t *testing.T) { + tests := []struct { + name string + lowercase string + uppercase string + want string + }{ + {name: "lowercase", lowercase: "release_a", want: "release_a"}, + {name: "uppercase fallback", uppercase: "release_b", want: "release_b"}, + {name: "lowercase takes precedence", lowercase: "release_a", uppercase: "release_b", want: "release_a"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Setenv(instanceName, tt.lowercase) + t.Setenv(strings.ToUpper(instanceName), tt.uppercase) + + config := &Config{} + if err := ParseEnvironment(config); err != nil { + t.Fatalf("ParseEnvironment() error = %v", err) + } + if config.InstanceName != tt.want { + t.Fatalf("InstanceName = %q, want %q", config.InstanceName, tt.want) + } + }) + } +} + +func TestGeneratePodSpecInstanceName(t *testing.T) { + tests := []struct { + name string + instanceName string + wantPresent bool + }{ + {name: "configured", instanceName: "release_a", wantPresent: true}, + {name: "empty", wantPresent: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + pod, err := generatePodSpec(&Config{InstanceName: tt.instanceName}, "ghcr.io/kube-vip/kube-vip", "v0.0.0", true) + if err != nil { + t.Fatalf("generatePodSpec() error = %v", err) + } + + var value string + found := false + for _, env := range pod.Spec.Containers[0].Env { + if env.Name == instanceName { + found = true + value = env.Value + break + } + } + if found != tt.wantPresent { + t.Fatalf("instance_name present = %t, want %t", found, tt.wantPresent) + } + if found && value != tt.instanceName { + t.Fatalf("instance_name = %q, want %q", value, tt.instanceName) + } + }) + } +} + func TestParseEnvironmentConfigFile(t *testing.T) { // Save original environment originalConfigFile := os.Getenv("config_file") diff --git a/pkg/kubevip/config_types.go b/pkg/kubevip/config_types.go index 6f4efb4a..0efbe694 100644 --- a/pkg/kubevip/config_types.go +++ b/pkg/kubevip/config_types.go @@ -91,6 +91,9 @@ type Config struct { // Namespace will define which namespace the control plane pods will run in Namespace string `yaml:"namespace"` + // InstanceName identifies this kube-vip deployment when naming host-global resources. + InstanceName string `yaml:"instanceName"` + // Namespace will define which namespace the control plane pods will run in ServiceNamespace string `yaml:"serviceNamespace"` diff --git a/pkg/kubevip/config_validation.go b/pkg/kubevip/config_validation.go index 04f4594e..dc36767f 100644 --- a/pkg/kubevip/config_validation.go +++ b/pkg/kubevip/config_validation.go @@ -6,16 +6,54 @@ import ( "strings" ) +const ( + // nftables object names are limited to 255 bytes. Reserve space for the + // prefix and address-family suffix added to the instance name. + nftablesNameMaxLength = 255 + egressNftablesTablePrefix = "kube_vip_" + egressNftablesTableSuffix = "_v4" + instanceNameMaxLength = nftablesNameMaxLength - len(egressNftablesTablePrefix) - len(egressNftablesTableSuffix) +) + // Validate runs configuration checks that are independent of host state. // This should be called after all config sources (flags, file, env vars) are merged. func (c *Config) Validate() error { if err := validateHealthCheckAddress(c.ControlPlaneHealthCheck.Address); err != nil { return err } + if err := validateInstanceName(c.InstanceName); err != nil { + return err + } return nil } +func validateInstanceName(name string) error { + if name == "" { + return nil + } + if len(name) > instanceNameMaxLength { + return fmt.Errorf("instance_name is %d bytes, must not exceed %d bytes so the %q prefix and %q or %q suffix fit within the nftables %d-byte name limit", + len(name), instanceNameMaxLength, egressNftablesTablePrefix, "_v4", "_v6", nftablesNameMaxLength) + } + + for position, char := range name { + if isValidNftablesNameCharacter(char) { + continue + } + return fmt.Errorf("instance_name %q contains invalid character %q at byte %d; only ASCII letters, digits, '.', '-' and '_' are allowed", + name, char, position) + } + return nil +} + +func isValidNftablesNameCharacter(char rune) bool { + return char >= 'a' && char <= 'z' || + char >= 'A' && char <= 'Z' || + char >= '0' && char <= '9' || + char == '_' || char == '-' || char == '.' +} + func validateHealthCheckAddress(address string) error { if address == "" { return nil diff --git a/pkg/kubevip/config_validation_test.go b/pkg/kubevip/config_validation_test.go index cd2039a2..217e011c 100644 --- a/pkg/kubevip/config_validation_test.go +++ b/pkg/kubevip/config_validation_test.go @@ -1,6 +1,7 @@ package kubevip import ( + "strings" "testing" ) @@ -30,3 +31,40 @@ func TestValidate_HealthCheckAddress(t *testing.T) { }) } } + +func TestValidate_InstanceName(t *testing.T) { + tests := []struct { + name string + instanceName string + wantErr bool + }{ + {name: "empty uses legacy default", instanceName: "", wantErr: false}, + {name: "letters digits and separators", instanceName: "release_01.prod-a", wantErr: false}, + {name: "exact maximum length", instanceName: strings.Repeat("a", instanceNameMaxLength), wantErr: false}, + {name: "exceeds maximum length", instanceName: strings.Repeat("a", instanceNameMaxLength+1), wantErr: true}, + {name: "space", instanceName: "release a", wantErr: true}, + {name: "slash", instanceName: "namespace/release", wantErr: true}, + {name: "dollar sign", instanceName: "release$a", wantErr: true}, + {name: "at sign", instanceName: "release@a", wantErr: true}, + {name: "newline", instanceName: "release\na", wantErr: true}, + {name: "null byte", instanceName: "release\x00a", wantErr: true}, + {name: "unicode", instanceName: "rilascio-à", wantErr: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + config := &Config{InstanceName: tt.instanceName} + err := config.Validate() + if (err != nil) != tt.wantErr { + t.Fatalf("Validate() error = %v, wantErr %t", err, tt.wantErr) + } + }) + } +} + +func TestInstanceNameLimitReservesNftablesPrefixAndFamilySuffix(t *testing.T) { + name := strings.Repeat("a", instanceNameMaxLength) + if got := len(egressNftablesTablePrefix + name + egressNftablesTableSuffix); got != nftablesNameMaxLength { + t.Fatalf("family-specific table name length = %d, want %d", got, nftablesNameMaxLength) + } +} diff --git a/pkg/manager/manager.go b/pkg/manager/manager.go index adb34aa9..abd51a49 100644 --- a/pkg/manager/manager.go +++ b/pkg/manager/manager.go @@ -388,7 +388,8 @@ func (sm *Manager) startMode(ctx context.Context) error { if sm.config.EnableServices { // This will tidy any dangling kube-vip iptables rules if sm.config.EgressClean { - err := nftables.ClearTables() + tableName := nftables.EgressTableBaseNameForInstance(sm.config.InstanceName) + err := nftables.ClearTablesWithName(tableName) if err != nil { log.Warn("[egress]", "mode", "nftables-internal", "clearing error", err) } else { diff --git a/pkg/manager/manager_dump.go b/pkg/manager/manager_dump.go index 239e26b5..e78aa6b7 100644 --- a/pkg/manager/manager_dump.go +++ b/pkg/manager/manager_dump.go @@ -46,6 +46,7 @@ func (sm *Manager) dumpConfigSection() { fmt.Printf("VIP Subnet: %s\n", sm.config.VIPSubnet) fmt.Printf("Port: %d\n", sm.config.Port) fmt.Printf("Namespace: %s\n", sm.config.Namespace) + fmt.Printf("Instance Name: %s\n", sm.config.InstanceName) fmt.Printf("Service Namespace: %s\n", sm.config.ServiceNamespace) fmt.Printf("Interface: %s\n", sm.config.Interface) fmt.Printf("Services Interface: %s\n", sm.config.ServicesInterface) @@ -218,6 +219,7 @@ func (sm *Manager) dumpRuntimeSection() { fmt.Printf("Egress Clean Enabled: %t\n", sm.config.EgressClean) if sm.config.EgressClean { fmt.Printf("Egress with nftables: %t\n", sm.config.EgressWithNftables) + fmt.Printf("Egress nftables table name: %s\n", nftables.EgressTableBaseNameForInstance(sm.config.InstanceName)) fmt.Printf("Egress Pod CIDR: %s\n", sm.config.EgressPodCidr) fmt.Printf("Egress Service CIDR: %s\n", sm.config.EgressServiceCidr) } diff --git a/pkg/nftables/nftables.go b/pkg/nftables/nftables.go index 6fb754ae..4cb18687 100644 --- a/pkg/nftables/nftables.go +++ b/pkg/nftables/nftables.go @@ -19,8 +19,9 @@ import ( ) const ( - NatTable = "kube_vip_%s" - SNatChain = "kube_vip_snat_%s" + NatTable = "kube_vip_%s" + SNatChain = "kube_vip_snat_%s" + DefaultEgressTableName = "kube_vip" ) const ( @@ -54,33 +55,33 @@ const ( const connmarkOffset = 0x10000 // 65536 - added to listenPort to get our connmark value func ApplySNAT(podIP, vipIP, service, destinationPorts string, ignoreCIDR []string, allowCIDR []string, IPv6 bool) error { + return ApplySNATWithTable(podIP, vipIP, service, destinationPorts, ignoreCIDR, allowCIDR, IPv6, "") +} + +// ApplySNATWithTable applies an egress SNAT rule in the configured table. +func ApplySNATWithTable(podIP, vipIP, service, destinationPorts string, ignoreCIDR []string, allowCIDR []string, IPv6 bool, tableName string) error { conn, err := nftables.New() if err != nil { return err } - var tableName string - if IPv6 { - tableName = fmt.Sprintf(NatTable, "v6") - } else { - tableName = fmt.Sprintf(NatTable, "v4") - } + resolvedTableName := egressTableName(tableName, IPv6) // Look up the table - if t, err := FilterTable(conn, tableName, IPv6); err != nil { + if t, err := FilterTable(conn, resolvedTableName, IPv6); err != nil { if t == nil { // If it doesn't exist then create it - slog.Debug("[egress]", "Creating Table", tableName) - conn.AddTable(GetTable(IPv6)) + slog.Debug("[egress]", "Creating Table", resolvedTableName) + conn.AddTable(GetEgressTable(IPv6, tableName)) } } slog.Debug("[egress]", "Creating Chain for service", service, utils.IPv6Family, IPv6) - conn.AddChain(GetSNatChain(IPv6, service)) + conn.AddChain(GetSNatChainForTable(IPv6, service, tableName)) err = conn.Flush() if err != nil { return err } - portExpressions, err := portSet(conn, IPv6, destinationPorts) + portExpressions, err := portSet(conn, IPv6, destinationPorts, tableName) if err != nil { return err } @@ -90,7 +91,7 @@ func ApplySNAT(podIP, vipIP, service, destinationPorts string, ignoreCIDR []stri // Create our nftables rule if len(allowCIDR) == 0 { // No allowed CIDRs - rule, err = CreateRule(podIP, vipIP, service, ignoreCIDR, "", conn, IPv6, portExpressions[x]) + rule, err = CreateRuleForTable(podIP, vipIP, service, ignoreCIDR, "", conn, IPv6, portExpressions[x], tableName) if err != nil { return err } @@ -99,7 +100,7 @@ func ApplySNAT(podIP, vipIP, service, destinationPorts string, ignoreCIDR []stri } else { // Create a rule for each allowed CIDR for y := range allowCIDR { - rule, err = CreateRule(podIP, vipIP, service, ignoreCIDR, allowCIDR[y], conn, IPv6, portExpressions[x]) + rule, err = CreateRuleForTable(podIP, vipIP, service, ignoreCIDR, allowCIDR[y], conn, IPv6, portExpressions[x], tableName) if err != nil { return err } @@ -113,7 +114,7 @@ func ApplySNAT(podIP, vipIP, service, destinationPorts string, ignoreCIDR []stri // Create our nftables rule if len(allowCIDR) == 0 { // No allowed CIDRs - rule, err = CreateRule(podIP, vipIP, service, ignoreCIDR, "", conn, IPv6, nil) + rule, err = CreateRuleForTable(podIP, vipIP, service, ignoreCIDR, "", conn, IPv6, nil, tableName) if err != nil { return err } @@ -122,7 +123,7 @@ func ApplySNAT(podIP, vipIP, service, destinationPorts string, ignoreCIDR []stri } else { // Create a rule for each allowed CIDR for y := range allowCIDR { - rule, err = CreateRule(podIP, vipIP, service, ignoreCIDR, allowCIDR[y], conn, IPv6, nil) + rule, err = CreateRuleForTable(podIP, vipIP, service, ignoreCIDR, allowCIDR[y], conn, IPv6, nil, tableName) if err != nil { return err } @@ -140,10 +141,10 @@ func ApplySNAT(podIP, vipIP, service, destinationPorts string, ignoreCIDR []stri return conn.CloseLasting() // Close out any remaining netlink communication } -func portSet(conn *nftables.Conn, IPv6 bool, destinationPorts string) (setExpression [][]expr.Any, err error) { +func portSet(conn *nftables.Conn, IPv6 bool, destinationPorts, tableName string) (setExpression [][]expr.Any, err error) { // If we filter on ports protocols then parse them if destinationPorts != "" { - table := GetTable(IPv6) + table := GetEgressTable(IPv6, tableName) fixedPorts := strings.Split(destinationPorts, ",") // Create an element using our pod IP @@ -303,6 +304,11 @@ func portSet(conn *nftables.Conn, IPv6 bool, destinationPorts string) (setExpres } func DeleteSNAT(IPv6 bool, service string) error { + return DeleteSNATFromTable(IPv6, service, "") +} + +// DeleteSNATFromTable deletes an egress SNAT chain from the configured table. +func DeleteSNATFromTable(IPv6 bool, service, tableName string) error { conn, err := nftables.New() if err != nil { return err @@ -311,7 +317,7 @@ func DeleteSNAT(IPv6 bool, service string) error { var chainName = fmt.Sprintf(SNatChain, service) slog.Info("[egress]", "Looking for", chainName) - chain, err := conn.ListChain(GetTable(IPv6), chainName) + chain, err := conn.ListChain(GetEgressTable(IPv6, tableName), chainName) if err != nil { return err } @@ -325,13 +331,96 @@ func DeleteSNAT(IPv6 bool, service string) error { return fmt.Errorf("unable to find chain [%s]", chainName) } -func GetTable(IPv6 bool) *nftables.Table { - var tableName string - if IPv6 { - tableName = fmt.Sprintf(NatTable, "v6") - } else { - tableName = fmt.Sprintf(NatTable, "v4") +// DeleteSNATFromTableIfExists deletes a Service's egress SNAT chain when it is +// present. It is used while migrating a Service between instance-owned tables. +func DeleteSNATFromTableIfExists(IPv6 bool, service, tableName string) error { + conn, err := nftables.New() + if err != nil { + return err } + + table := GetEgressTable(IPv6, tableName) + chainName := fmt.Sprintf(SNatChain, service) + chains, err := conn.ListChainsOfTableFamily(table.Family) + if err != nil { + return fmt.Errorf("failed to list nftables chains: %w", err) + } + + for _, chain := range chains { + if chain.Table != nil && chain.Table.Name == table.Name && chain.Name == chainName { + slog.Info("[egress] deleting service chain", "table", table.Name, "chain", chainName) + conn.DelChain(chain) + return conn.Flush() + } + } + + return nil +} + +// DeleteSNATFromOtherTables removes a Service's SNAT chain from every table in +// one address family except the table currently owned by this kube-vip +// instance. Chain names include the Service UID, so cleanup remains +// service-scoped even when the legacy table is shared by multiple instances. +func DeleteSNATFromOtherTables(IPv6 bool, service, keepTableName string) error { + return deleteSNATFromTables(IPv6, service, EgressTableBaseName(keepTableName)) +} + +// DeleteSNATFromAllTables removes a Service's SNAT chain from all IPv4 and +// IPv6 tables. This does not depend on endpoint data and is safe for teardown +// of IPv4-only, IPv6-only, and dual-stack Services. +func DeleteSNATFromAllTables(service string) error { + return errors.Join( + deleteSNATFromTables(false, service, ""), + deleteSNATFromTables(true, service, ""), + ) +} + +func deleteSNATFromTables(IPv6 bool, service, keepTableName string) error { + conn, err := nftables.New() + if err != nil { + return err + } + + family := nftables.TableFamilyIPv4 + if IPv6 { + family = nftables.TableFamilyIPv6 + } + keepTable := "" + if keepTableName != "" { + keepTable = GetEgressTable(IPv6, keepTableName).Name + } + chainName := fmt.Sprintf(SNatChain, service) + chains, err := conn.ListChainsOfTableFamily(family) + if err != nil { + return fmt.Errorf("failed to list nftables chains: %w", err) + } + + deleted := false + for _, chain := range chains { + if !shouldDeleteSNATChain(chain, chainName, keepTable) { + continue + } + slog.Info("[egress] deleting stale service chain", "table", chain.Table.Name, "chain", chainName) + conn.DelChain(chain) + deleted = true + } + if deleted { + return conn.Flush() + } + return nil +} + +func shouldDeleteSNATChain(chain *nftables.Chain, chainName, keepTable string) bool { + return chain.Table != nil && chain.Name == chainName && chain.Table.Name != keepTable +} + +func GetTable(IPv6 bool) *nftables.Table { + return GetEgressTable(IPv6, "") +} + +// GetEgressTable returns the family-specific egress table. +func GetEgressTable(IPv6 bool, tableName string) *nftables.Table { + tableName = egressTableName(tableName, IPv6) // Default to IPv4 table := &nftables.Table{ Family: nftables.TableFamilyIPv4, @@ -345,12 +434,42 @@ func GetTable(IPv6 bool) *nftables.Table { return table } +func egressTableName(baseName string, IPv6 bool) string { + family := "v4" + if IPv6 { + family = "v6" + } + return EgressTableBaseName(baseName) + "_" + family +} + +// EgressTableBaseName returns the effective table base name. An empty +// configured value retains the legacy kube_vip table names. +func EgressTableBaseName(baseName string) string { + if baseName == "" { + return DefaultEgressTableName + } + return baseName +} + +// EgressTableBaseNameForInstance returns the egress table base owned by a +// kube-vip instance. An empty instance name retains the legacy table names. +func EgressTableBaseNameForInstance(instanceName string) string { + if instanceName == "" { + return DefaultEgressTableName + } + return DefaultEgressTableName + "_" + instanceName +} + func GetSNatChain(IPv6 bool, service string) *nftables.Chain { + return GetSNatChainForTable(IPv6, service, "") +} + +func GetSNatChainForTable(IPv6 bool, service, tableName string) *nftables.Chain { var chainName = fmt.Sprintf(SNatChain, service) policy := nftables.ChainPolicyAccept return &nftables.Chain{ Name: chainName, - Table: GetTable(IPv6), + Table: GetEgressTable(IPv6, tableName), Type: nftables.ChainTypeNAT, Hooknum: nftables.ChainHookPostrouting, Priority: nftables.ChainPriorityNATSource, @@ -367,33 +486,42 @@ func FilterTable(conn *nftables.Conn, tableName string, IPv6 bool) (*nftables.Ta // ClearTable will remove the original tables and create new empty ones func ClearTables() error { + return ClearTablesWithName("") +} + +// ClearTablesWithName clears only the egress tables owned by this kube-vip instance. +func ClearTablesWithName(tableName string) error { conn, err := nftables.New() if err != nil { return err } - tableName := fmt.Sprintf(NatTable, "v6") - if t, err := FilterTable(conn, tableName, false); err != nil { + ipv6TableName := egressTableName(tableName, true) + if t, err := FilterTable(conn, ipv6TableName, true); err != nil { slog.Debug("[egress]", "Cleaning IPv6 finding tables error", err) } else if t != nil { conn.DelTable(t) } // These don't return errors, so not 100% sure how to guarantee things were created - conn.AddTable(GetTable(true)) - tableName = fmt.Sprintf(NatTable, "v4") - if t, err := FilterTable(conn, tableName, true); err != nil { + conn.AddTable(GetEgressTable(true, tableName)) + ipv4TableName := egressTableName(tableName, false) + if t, err := FilterTable(conn, ipv4TableName, false); err != nil { slog.Debug("[egress]", "Cleaning IPv4 finding tables error", err) } else if t != nil { conn.DelTable(t) } // These don't return errors, so not 100% sure how to guarantee things were created - conn.AddTable(GetTable(false)) - return nil + conn.AddTable(GetEgressTable(false, tableName)) + return conn.Flush() } // Create our nftables rule func CreateRule(podIP, vipIP, service string, ignoreCIDR []string, allowCIDR string, conn *nftables.Conn, IPv6 bool, portExpression []expr.Any) (*nftables.Rule, error) { + return CreateRuleForTable(podIP, vipIP, service, ignoreCIDR, allowCIDR, conn, IPv6, portExpression, "") +} + +func CreateRuleForTable(podIP, vipIP, service string, ignoreCIDR []string, allowCIDR string, conn *nftables.Conn, IPv6 bool, portExpression []expr.Any, tableName string) (*nftables.Rule, error) { // Validate pod IP if net.ParseIP(podIP) == nil { @@ -406,7 +534,7 @@ func CreateRule(podIP, vipIP, service string, ignoreCIDR []string, allowCIDR str } // Get the kube-vip table - table := GetTable(IPv6) + table := GetEgressTable(IPv6, tableName) // Create our rule rule := &nftables.Rule{ @@ -414,7 +542,7 @@ func CreateRule(podIP, vipIP, service string, ignoreCIDR []string, allowCIDR str Exprs: []expr.Any{}, } // Set the correct chain - rule.Chain = GetSNatChain(IPv6, service) + rule.Chain = GetSNatChainForTable(IPv6, service, tableName) // Create a set for our original/source address set := &nftables.Set{ diff --git a/pkg/nftables/nftables_test.go b/pkg/nftables/nftables_test.go new file mode 100644 index 00000000..9956ea2b --- /dev/null +++ b/pkg/nftables/nftables_test.go @@ -0,0 +1,109 @@ +package nftables + +import ( + "testing" + + googlenftables "github.com/google/nftables" +) + +func TestEgressTableName(t *testing.T) { + tests := []struct { + name string + baseName string + ipv6 bool + want string + }{ + {name: "default IPv4", want: "kube_vip_v4"}, + {name: "default IPv6", ipv6: true, want: "kube_vip_v6"}, + {name: "release IPv4", baseName: "kube_vip_release_a", want: "kube_vip_release_a_v4"}, + {name: "namespace IPv6", baseName: "kube-vip-networking", ipv6: true, want: "kube-vip-networking_v6"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := egressTableName(tt.baseName, tt.ipv6); got != tt.want { + t.Fatalf("egressTableName(%q, %t) = %q, want %q", tt.baseName, tt.ipv6, got, tt.want) + } + }) + } +} + +func TestGetEgressTable(t *testing.T) { + table := GetEgressTable(false, "release_a") + if table.Name != "release_a_v4" { + t.Fatalf("table name = %q, want %q", table.Name, "release_a_v4") + } +} + +func TestEgressTableBaseName(t *testing.T) { + if got := EgressTableBaseName(""); got != DefaultEgressTableName { + t.Fatalf("EgressTableBaseName(\"\") = %q, want %q", got, DefaultEgressTableName) + } + if got := EgressTableBaseName("release_a"); got != "release_a" { + t.Fatalf("EgressTableBaseName(\"release_a\") = %q, want %q", got, "release_a") + } +} + +func TestEgressTableBaseNameForInstance(t *testing.T) { + if got := EgressTableBaseNameForInstance(""); got != DefaultEgressTableName { + t.Fatalf("EgressTableBaseNameForInstance(\"\") = %q, want %q", got, DefaultEgressTableName) + } + if got := EgressTableBaseNameForInstance("release_a"); got != "kube_vip_release_a" { + t.Fatalf("EgressTableBaseNameForInstance(\"release_a\") = %q, want %q", got, "kube_vip_release_a") + } +} + +func TestEgressTableNameForInstance(t *testing.T) { + baseName := EgressTableBaseNameForInstance("release_a") + if got := egressTableName(baseName, false); got != "kube_vip_release_a_v4" { + t.Fatalf("IPv4 table name = %q, want %q", got, "kube_vip_release_a_v4") + } + if got := egressTableName(baseName, true); got != "kube_vip_release_a_v6" { + t.Fatalf("IPv6 table name = %q, want %q", got, "kube_vip_release_a_v6") + } +} + +func TestShouldDeleteSNATChain(t *testing.T) { + tests := []struct { + name string + chain *googlenftables.Chain + keepTable string + want bool + }{ + { + name: "stale table for matching Service UID", + chain: &googlenftables.Chain{ + Name: "kube_vip_snat_service-a", + Table: &googlenftables.Table{Name: "release_a_v4"}, + }, + keepTable: "release_b_v4", + want: true, + }, + { + name: "current table for matching Service UID", + chain: &googlenftables.Chain{ + Name: "kube_vip_snat_service-a", + Table: &googlenftables.Table{Name: "release_b_v4"}, + }, + keepTable: "release_b_v4", + want: false, + }, + { + name: "different Service UID in stale table", + chain: &googlenftables.Chain{ + Name: "kube_vip_snat_service-b", + Table: &googlenftables.Table{Name: "release_a_v4"}, + }, + keepTable: "release_b_v4", + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := shouldDeleteSNATChain(tt.chain, "kube_vip_snat_service-a", tt.keepTable); got != tt.want { + t.Fatalf("shouldDeleteSNATChain() = %t, want %t", got, tt.want) + } + }) + } +} diff --git a/pkg/services/egress.go b/pkg/services/egress.go index 68032cc9..5928d09b 100644 --- a/pkg/services/egress.go +++ b/pkg/services/egress.go @@ -3,6 +3,7 @@ package services import ( "bufio" "context" + "errors" "fmt" "net" "os" @@ -15,7 +16,9 @@ import ( "github.com/kube-vip/kube-vip/pkg/nftables" "github.com/kube-vip/kube-vip/pkg/utils" "github.com/kube-vip/kube-vip/pkg/vip" - v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/util/retry" ) // DEBUG @@ -143,7 +146,10 @@ func checkCIDR(ip, cidr string) (string, error) { return "", nil } -func (p *Processor) configureEgress(ctx context.Context, vipIP, podIP, namespace, serviceUUID string, annotations map[string]string) error { +func (p *Processor) configureEgress(ctx context.Context, vipIP, podIP, namespace, serviceUUID string, annotations map[string]string, applied *bool) error { + if applied != nil { + *applied = false + } var podCidr, serviceCidr string var autoServiceCIDR, autoPodCIDR string var discoverErr error @@ -234,11 +240,18 @@ func (p *Processor) configureEgress(ctx context.Context, vipIP, podIP, namespace allowCIDRs = append(allowCIDRs, networks...) } + currentTableName := nftables.EgressTableBaseNameForInstance(p.config.InstanceName) // Apply the SNAT rules - err := nftables.ApplySNAT(podIP, vipIP, serviceUUID, destinationPorts, ignoreCIDRs, allowCIDRs, utils.IsIPv6(vipIP)) + err := nftables.ApplySNATWithTable(podIP, vipIP, serviceUUID, destinationPorts, ignoreCIDRs, allowCIDRs, utils.IsIPv6(vipIP), currentTableName) if err != nil { return fmt.Errorf("error performing netlink nftables [%s]", err) } + if err := nftables.DeleteSNATFromOtherTables(utils.IsIPv6(vipIP), serviceUUID, currentTableName); err != nil { + return fmt.Errorf("cleaning stale nftables egress chains: %w", err) + } + if applied != nil { + *applied = true + } return nil } @@ -346,9 +359,17 @@ func (p *Processor) configureEgress(ctx context.Context, vipIP, podIP, namespace return nil } +func (p *Processor) prepareEgressNftablesTable(serviceUID string, ipv6 bool) error { + tableName := nftables.EgressTableBaseNameForInstance(p.config.InstanceName) + if err := nftables.DeleteSNATFromTableIfExists(ipv6, serviceUID, tableName); err != nil { + return fmt.Errorf("preparing nftables table %q for egress reconciliation: %w", tableName, err) + } + return nil +} + func (p *Processor) AutoDiscoverCIDRs(ctx context.Context) (serviceCIDR, podCIDR string, err error) { log.Debug("Trying to automatically discover Service and Pod CIDRs") - options := v1.ListOptions{ + options := metav1.ListOptions{ LabelSelector: "component=kube-controller-manager", } podList, err := p.clientSet.CoreV1().Pods("kube-system").List(ctx, options) @@ -374,3 +395,44 @@ func (p *Processor) AutoDiscoverCIDRs(ctx context.Context) (serviceCIDR, podCIDR return } + +func (p *Processor) updateEgressNftablesTableAnnotation(ctx context.Context, service *corev1.Service) error { + tableName := nftables.EgressTableBaseNameForInstance(p.config.InstanceName) + err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + currentService, err := p.clientSet.CoreV1().Services(service.Namespace).Get(ctx, service.Name, metav1.GetOptions{}) + if err != nil { + return err + } + if currentService.Annotations[kubevip.EgressNftablesTable] == tableName { + return nil + } + + serviceCopy := currentService.DeepCopy() + if serviceCopy.Annotations == nil { + serviceCopy.Annotations = make(map[string]string) + } + serviceCopy.Annotations[kubevip.EgressNftablesTable] = tableName + _, err = p.clientSet.CoreV1().Services(service.Namespace).Update(ctx, serviceCopy, metav1.UpdateOptions{}) + return err + }) + if err != nil { + return fmt.Errorf("updating nftables table annotation on Service %s/%s: %w", service.Namespace, service.Name, err) + } + return nil +} + +func (p *Processor) cleanupStaleEgressNftablesChains(service *corev1.Service) error { + if service.Annotations[kubevip.Egress] != "true" || + (service.Annotations[kubevip.EgressInternal] == "" && !p.config.EgressWithNftables) { + return nftables.DeleteSNATFromAllTables(string(service.UID)) + } + + tableName := nftables.EgressTableBaseNameForInstance(p.config.InstanceName) + if service.Annotations[kubevip.EgressNftablesTable] != tableName { + return nil + } + return errors.Join( + nftables.DeleteSNATFromOtherTables(false, string(service.UID), tableName), + nftables.DeleteSNATFromOtherTables(true, string(service.UID), tableName), + ) +} diff --git a/pkg/services/processor.go b/pkg/services/processor.go index fd379fca..f80700aa 100644 --- a/pkg/services/processor.go +++ b/pkg/services/processor.go @@ -116,6 +116,15 @@ func (p *Processor) AddOrModify(ctx context.Context, event watch.Event, serviceF return nil } + // The Service annotation is cluster-wide while nftables state is local to + // each node. Reconcile stale per-Service chains on every node after a table + // migration, even when this kube-vip pod is not the Service leader. + if svc.Annotations[kubevip.EgressNftablesTable] != "" { + if err := p.cleanupStaleEgressNftablesChains(svc); err != nil { + log.Warn("failed to clean stale nftables egress chains", "service", svc.Name, "namespace", svc.Namespace, "err", err) + } + } + svcAddresses, svcHostnames := instance.FetchServiceAddresses(svc) // We only care about LoadBalancer services that have been allocated an address diff --git a/pkg/services/services.go b/pkg/services/services.go index 42781517..2ae9749e 100644 --- a/pkg/services/services.go +++ b/pkg/services/services.go @@ -27,6 +27,7 @@ import ( "github.com/kube-vip/kube-vip/pkg/instance" "github.com/kube-vip/kube-vip/pkg/kubevip" "github.com/kube-vip/kube-vip/pkg/lease" + "github.com/kube-vip/kube-vip/pkg/nftables" "github.com/kube-vip/kube-vip/pkg/servicecontext" "github.com/kube-vip/kube-vip/pkg/upnp" "github.com/kube-vip/kube-vip/pkg/utils" @@ -305,6 +306,9 @@ func (p *Processor) configureService(ctx context.Context, inst *instance.Instanc } var podIP string errList := []error{} + configuredRules := 0 + useInternalNftables := svc.Annotations[kubevip.EgressInternal] != "" || p.config.EgressWithNftables + preparedFamilies := map[bool]bool{} // Should egress be IPv6 if svc.Annotations[kubevip.EgressIPv6] == "true" { @@ -314,11 +318,21 @@ func (p *Processor) configureService(ctx context.Context, inst *instance.Instanc if !p.config.EnableEndpoints && utils.IsIPv6(serviceIP) { podIP = svc.Annotations[kubevip.ActiveEndpointIPv6] + if useInternalNftables && !preparedFamilies[true] { + if err := p.prepareEgressNftablesTable(string(svc.UID), true); err != nil { + errList = append(errList, err) + continue + } + preparedFamilies[true] = true + } - err := p.configureEgress(ctx, serviceIP, podIP, svc.Namespace, string(svc.UID), svc.Annotations) + applied := false + err := p.configureEgress(ctx, serviceIP, podIP, svc.Namespace, string(svc.UID), svc.Annotations, &applied) if err != nil { errList = append(errList, err) log.Warn("[service] configuring egress IPv6", "service", svc.Name, "namespace", svc.Namespace, "err", err) + } else if applied { + configuredRules++ } } } @@ -329,14 +343,31 @@ func (p *Processor) configureService(ctx context.Context, inst *instance.Instanc if !p.config.EnableEndpoints && utils.IsIPv6(serviceIP) { podIPs = svc.Annotations[kubevip.ActiveEndpointIPv6] } - err := p.configureEgress(ctx, serviceIP, podIPs, svc.Namespace, string(svc.UID), svc.Annotations) + ipv6 := utils.IsIPv6(serviceIP) + if useInternalNftables && !preparedFamilies[ipv6] { + if err := p.prepareEgressNftablesTable(string(svc.UID), ipv6); err != nil { + errList = append(errList, err) + continue + } + preparedFamilies[ipv6] = true + } + applied := false + err := p.configureEgress(ctx, serviceIP, podIPs, svc.Namespace, string(svc.UID), svc.Annotations, &applied) if err != nil { errList = append(errList, err) log.Warn("[service] configuring egress IPv4", "service", svc.Name, "namespace", svc.Namespace, "err", err) + } else if applied { + configuredRules++ } } } if len(errList) == 0 { + if configuredRules > 0 && useInternalNftables { + if err := p.updateEgressNftablesTableAnnotation(ctx, svc); err != nil { + return err + } + } + var provider providers.Provider if p.config.EnableEndpoints { provider = providers.NewEndpoints() @@ -431,6 +462,13 @@ func (p *Processor) deleteService(ctx context.Context, uid types.UID) error { } } + internalNftablesEgress := serviceInstance.ServiceSnapshot.Annotations[kubevip.EgressInternal] != "" || p.config.EgressWithNftables + if serviceInstance.ServiceSnapshot.Annotations[kubevip.Egress] == "true" && internalNftablesEgress { + if err := nftables.DeleteSNATFromAllTables(string(serviceInstance.ServiceSnapshot.UID)); err != nil { + log.Error("[service] nftables egress teardown", "service", serviceInstance.ServiceSnapshot.Name, "err", err) + } + } + if !shared { for x := range serviceInstance.Clusters { serviceInstance.Clusters[x].Stop() @@ -469,7 +507,7 @@ func (p *Processor) deleteService(ctx context.Context, uid types.UID) error { } // We will need to tear down the egress - if serviceInstance.ServiceSnapshot.Annotations[kubevip.Egress] == "true" { + if serviceInstance.ServiceSnapshot.Annotations[kubevip.Egress] == "true" && !internalNftablesEgress { if serviceInstance.ServiceSnapshot.Annotations[kubevip.ActiveEndpoint] != "" { log.Info("[service] egress re-write enabled", "service", serviceInstance.ServiceSnapshot.Name) err := egress.Teardown(serviceInstance.ServiceSnapshot.Annotations[kubevip.ActiveEndpoint], serviceInstance.ServiceSnapshot.Spec.LoadBalancerIP, serviceInstance.ServiceSnapshot.Namespace, string(serviceInstance.ServiceSnapshot.UID), serviceInstance.ServiceSnapshot.Annotations, p.config.EgressWithNftables) @@ -545,6 +583,9 @@ func (p *Processor) updateEgressConfiguration(ctx context.Context, svc *v1.Servi // Apply new egress rules with updated endpoint serviceIPs, _ := instance.FetchServiceAddresses(svc) errList := []error{} + configuredRules := 0 + useInternalNftables := svc.Annotations[kubevip.EgressInternal] != "" || p.config.EgressWithNftables + preparedFamilies := map[bool]bool{} // Check if egress should be IPv6 if svc.Annotations[kubevip.EgressIPv6] == "true" { @@ -553,10 +594,20 @@ func (p *Processor) updateEgressConfiguration(ctx context.Context, svc *v1.Servi for _, serviceIP := range serviceIPs { if !p.config.EnableEndpoints && utils.IsIPv6(serviceIP) { podIP := newIPv6 - err := p.configureEgress(ctx, serviceIP, podIP, svc.Namespace, string(svc.UID), svc.Annotations) + if useInternalNftables && !preparedFamilies[true] { + if err := p.prepareEgressNftablesTable(string(svc.UID), true); err != nil { + errList = append(errList, err) + continue + } + preparedFamilies[true] = true + } + applied := false + err := p.configureEgress(ctx, serviceIP, podIP, svc.Namespace, string(svc.UID), svc.Annotations, &applied) if err != nil { errList = append(errList, err) log.Warn("[service] configuring egress IPv6", "service", svc.Name, "namespace", svc.Namespace, "err", err) + } else if applied { + configuredRules++ } } } @@ -567,10 +618,21 @@ func (p *Processor) updateEgressConfiguration(ctx context.Context, svc *v1.Servi if !p.config.EnableEndpoints && utils.IsIPv6(serviceIP) { podIPs = newIPv6 } - err := p.configureEgress(ctx, serviceIP, podIPs, svc.Namespace, string(svc.UID), svc.Annotations) + ipv6 := utils.IsIPv6(serviceIP) + if useInternalNftables && !preparedFamilies[ipv6] { + if err := p.prepareEgressNftablesTable(string(svc.UID), ipv6); err != nil { + errList = append(errList, err) + continue + } + preparedFamilies[ipv6] = true + } + applied := false + err := p.configureEgress(ctx, serviceIP, podIPs, svc.Namespace, string(svc.UID), svc.Annotations, &applied) if err != nil { errList = append(errList, err) log.Warn("[service] configuring egress IPv4", "service", svc.Name, "namespace", svc.Namespace, "err", err) + } else if applied { + configuredRules++ } } } @@ -578,6 +640,11 @@ func (p *Processor) updateEgressConfiguration(ctx context.Context, svc *v1.Servi if len(errList) > 0 { return fmt.Errorf("errors configuring egress: %v", errList) } + if configuredRules > 0 && useInternalNftables { + if err := p.updateEgressNftablesTableAnnotation(ctx, svc); err != nil { + return err + } + } // Update the service snapshot to reflect the new state // NOTE: Do NOT call UpdateServiceAnnotation here - the annotation was already updated