Repository: metallb/metallb
Pull request #2692 (2025-02-14 14:00:44)

Reprocess services if shareability of a given service changes
<!-- Thanks for sending a pull request!
1. If this is your first time, please read the [contributing guide](https://metallb.universe.tf/community/#contributing)
2. For non-trivial pull requests, please [file an
   issue](https://github.com/metallb/metallb/issues/new) first, and get
   agreement that the change is a good idea, and a general guideline
   for how it should be implemented, before sending code. Large PRs
   that weren't first discussed and agreed upon in an issue won't be
   accepted.
3. If the PR fixes a particular bug, please include the words "Fixed
   #<issue number>" in the PR text, so that the bug auto-closes when
   the PR is merged.
-->

**Is this a BUG FIX or a FEATURE ?**:

> Uncomment only one, leave it on its own line:
>
/kind bug
> /kind cleanup
> /kind feature
> /kind design
> /kind flake
> /kind failing
> /kind documentation
> /kind regression

**What this PR does / why we need it**:

If the sharing criteria (ETP, endpoint selector) of a given service change, it might mean that the IP can be shared with other services.
Because of this,  we reprocess all the services.

**Special notes for your reviewer**:

**Release note**:
<!--  Write your release note:
1. Enter your extended release note in the below block. If the PR requires additional action from users switching to the new release, include the string "action required".
2. Follow the instructions for writing a release note from k8s: https://git.k8s.io/community/contributors/guide/release-notes.md
3. If no release note is required, just write "NONE".
-->

```release-note
Reprocess the services and reassign IPs if the sharing criteria of a given service changes (ie, the external traffic policy changes and suddenly another service in pending becomes compatible).
```

Fixes https://github.com/metallb/metallb/issues/2403


Patch:
diff --git a/charts/metallb/README.md b/charts/metallb/README.md
index 7a104ba0..2737c20e 100644
--- a/charts/metallb/README.md
+++ b/charts/metallb/README.md
@@ -138,7 +138,7 @@ Kubernetes: `>= 1.19.0-0`
 | speaker.livenessProbe.successThreshold | int | `1` |  |
 | speaker.livenessProbe.timeoutSeconds | int | `1` |  |
 | speaker.logLevel | string | `"info"` | Speaker log level. Must be one of: `all`, `debug`, `info`, `warn`, `error` or `none` |
-| speaker.memberlist.enabled | bool | `true` |  |
+| speaker.memberlist.enabled | bool | `true` | When enabled: false, the speaker pods must run on all nodes |
 | speaker.memberlist.mlBindAddrOverride | string | `""` |  |
 | speaker.memberlist.mlBindPort | int | `7946` |  |
 | speaker.memberlist.mlSecretKeyPath | string | `"/etc/ml_secret_key"` |  |
diff --git a/charts/metallb/templates/speaker.yaml b/charts/metallb/templates/speaker.yaml
index e70743ce..9d9297a8 100644
--- a/charts/metallb/templates/speaker.yaml
+++ b/charts/metallb/templates/speaker.yaml
@@ -8,6 +8,10 @@
 {{- fail "frrk8s.enabled frrk8s.external are mutually exclusive!" }}
 {{- end }}
 
+{{- if and (not .Values.speaker.memberlist.enabled) .Values.speaker.nodeSelector }}
+{{- fail "nodeSelector must be empty when memberlist is disabled" }}
+{{- end }}
+
 {{- if .Values.speaker.frr.enabled }}
 
 # FRR expects to have these files owned by frr:frr on startup.
diff --git a/charts/metallb/values.yaml b/charts/metallb/values.yaml
index 50d26bca..365125ed 100644
--- a/charts/metallb/values.yaml
+++ b/charts/metallb/values.yaml
@@ -262,6 +262,7 @@ speaker:
   logLevel: info
   tolerateMaster: true
   memberlist:
+    # --  When enabled: false, the speaker pods must run on all nodes
     enabled: true
     mlBindPort: 7946
     mlBindAddrOverride: ""
diff --git a/controller/main.go b/controller/main.go
index bfcb9588..5597f8fb 100644
--- a/controller/main.go
+++ b/controller/main.go
@@ -79,11 +79,19 @@ func (c *controller) SetBalancer(l log.Logger, name string, svcRo *v1.Service, _
 	syncStateRes := controllers.SyncStateSuccess
 
 	prevIPs := c.ips.IPs(name)
+	prevAllocKey := c.ips.AllocationKey(name)
 
 	if c.convergeBalancer(l, name, svc) != nil {
 		syncStateRes = controllers.SyncStateErrorNoRetry
 	}
 
+	newAllocKey := c.ips.AllocationKey(name)
+
+	if prevAllocKey != newAllocKey {
+		level.Debug(l).Log("event", "allocation key changed", "msg", "allocation changed for shared service, reprocessing")
+		syncStateRes = controllers.SyncStateReprocessAll
+	}
+
 	if reflect.DeepEqual(svcRo, svc) {
 		level.Debug(l).Log("event", "noChange", "msg", "service converged, no change")
 		return syncStateRes
diff --git a/internal/allocator/allocator.go b/internal/allocator/allocator.go
index 971ff70c..5358cd48 100644
--- a/internal/allocator/allocator.go
+++ b/internal/allocator/allocator.go
@@ -560,6 +560,13 @@ func (a *Allocator) IPs(svc string) []net.IP {
 	return nil
 }
 
+func (a *Allocator) AllocationKey(svc string) string {
+	if alloc := a.allocated[svc]; alloc != nil {
+		return alloc.key.backend + alloc.key.sharing
+	}
+	return ""
+}
+
 // PoolForIP returns the pool structure associated with an IP.
 func (a *Allocator) PoolForIP(ips []net.IP) *config.Pool {
 	return poolFor(a.pools.ByName, ips)
diff --git a/internal/speakerlist/speakerlist.go b/internal/speakerlist/speakerlist.go
index 21e03e80..09d590ef 100644
--- a/internal/speakerlist/speakerlist.go
+++ b/internal/speakerlist/speakerlist.go
@@ -25,6 +25,13 @@ import (
 	"github.com/hashicorp/memberlist"
 )
 
+// SpeakerListInfo contains information about available speaker nodes.
+type SpeakerListInfo struct {
+	Nodes map[string]bool
+	// Disabled true indicates that the memberlist protocol is off
+	Disabled bool
+}
+
 // SpeakerList represents a list of healthy speakers.
 type SpeakerList struct {
 	l         log.Logger
@@ -33,6 +40,7 @@ type SpeakerList struct {
 	namespace string
 	labels    string
 
+	disabled bool
 	// The following fields are nil when memberlist is disabled.
 	mlEventCh chan memberlist.NodeEvent
 	ml        *memberlist.Memberlist
@@ -53,6 +61,8 @@ func New(logger log.Logger, nodeName, bindAddr, bindPort, secret, namespace, lab
 
 	if labels == "" || bindAddr == "" {
 		level.Info(logger).Log("op", "startup", "msg", "not starting fast dead node detection (memberlist), need ml-bindaddr / ml-labels config")
+		level.Info(logger).Log("op", "startup", "msg", "when memberlist is disabled, the speaker pods must run on all nodes")
+		sl.disabled = true
 		return &sl, nil
 	}
 
@@ -278,15 +288,23 @@ func (sl *SpeakerList) Rejoin() {
 }
 
 // UsableSpeakers returns a map of usable speaker nodes.
-func (sl *SpeakerList) UsableSpeakers() map[string]bool {
+func (sl *SpeakerList) UsableSpeakers() SpeakerListInfo {
 	if sl.ml == nil {
-		return nil
+		return SpeakerListInfo{
+			Nodes:    nil,
+			Disabled: true,
+		}
 	}
+
 	activeNodes := map[string]bool{}
 	for _, n := range sl.ml.Members() {
 		activeNodes[n.Name] = true
 	}
-	return activeNodes
+
+	return SpeakerListInfo{
+		Nodes:    activeNodes,
+		Disabled: false,
+	}
 }
 
 // Stop stops the SpeakerList.
diff --git a/speaker/layer2_controller.go b/speaker/layer2_controller.go
index 6636f487..f433e65e 100644
--- a/speaker/layer2_controller.go
+++ b/speaker/layer2_controller.go
@@ -17,6 +17,7 @@ package main
 import (
 	"bytes"
 	"crypto/sha256"
+	"maps"
 	"net"
 	"sort"
 
@@ -46,11 +47,9 @@ func (c *layer2Controller) SetConfig(log.Logger, *config.Config) error {
 	return nil
 }
 
-// usableNodes returns all nodes that have at least one fully ready
-// endpoint on them.
-// The speakers parameter is a map containing all the nodes with active speakers.
-// If the speakers map is nil, it is ignored.
-func usableNodes(eps []discovery.EndpointSlice, speakers map[string]bool) []string {
+// nodesWithEndpoint returns all nodes that have at least one fully ready
+// endpoint on them and that have a speaker according to the speaker map param.
+func nodesWithEndpoint(eps []discovery.EndpointSlice, speakers map[string]bool) []string {
 	usable := map[string]bool{}
 	for _, slice := range eps {
 		for _, ep := range slice.Endpoints {
@@ -61,10 +60,8 @@ func usableNodes(eps []discovery.EndpointSlice, speakers map[string]bool) []stri
 				continue
 			}
 			nodeName := *ep.NodeName
-			if speakers != nil {
-				if hasSpeaker := speakers[nodeName]; !hasSpeaker {
-					continue
-				}
+			if hasSpeaker := speakers[nodeName]; !hasSpeaker {
+				continue
 			}
 			if _, ok := usable[nodeName]; !ok {
 				usable[nodeName] = true
@@ -93,13 +90,10 @@ func (c *layer2Controller) ShouldAnnounce(l log.Logger, name string, toAnnounce
 		return "notOwner"
 	}
 
-	// we select the nodes with at least one matching l2 advertisement
-	forPool := c.speakersForPool(l, name, pool, nodes)
-	var availableNodes []string
+	speakerMap := c.speakersForPool(l, name, pool, nodes)
+	availableNodes := nodesWithActiveSpeakers(speakerMap)
 	if svc.Spec.ExternalTrafficPolicy == v1.ServiceExternalTrafficPolicyTypeLocal {
-		availableNodes = usableNodes(eps, forPool)
-	} else {
-		availableNodes = nodesWithActiveSpeakers(forPool)
+		availableNodes = nodesWithEndpoint(eps, speakerMap)
 	}
 
 	if len(availableNodes) == 0 {
@@ -223,8 +217,15 @@ func poolMatchesNodeL2(pool *config.Pool, node string) bool {
 }
 
 func (c *layer2Controller) speakersForPool(l log.Logger, name string, pool *config.Pool, nodes map[string]*v1.Node) map[string]bool {
+	sl := c.sList.UsableSpeakers()
+	eligibleNodes := maps.Keys(sl.Nodes)
+	if sl.Disabled {
+		// when memberlist is disabled we consider all nodes eligible under the assumption that
+		// there is a speaker pod running on all nodes.
+		eligibleNodes = maps.Keys(nodes)
+	}
 	res := map[string]bool{}
-	for s := range c.sList.UsableSpeakers() {
+	for s := range eligibleNodes {
 		if err := k8snodes.IsNodeAvailable(nodes[s]); err != nil {
 			level.Debug(l).Log("event", "skipping should announce l2", "service", name, "reason", "speaker's node has NodeNetworkUnavailable condition")
 			continue
diff --git a/speaker/main.go b/speaker/main.go
index ba82511c..708e3b13 100644
--- a/speaker/main.go
+++ b/speaker/main.go
@@ -630,6 +630,6 @@ type Protocol interface {
 
 // Speakerlist represents a list of healthy speakers.
 type SpeakerList interface {
-	UsableSpeakers() map[string]bool
+	UsableSpeakers() speakerlist.SpeakerListInfo
 	Rejoin()
 }


Test patch:
diff --git a/e2etest/bgptests/bgp.go b/e2etest/bgptests/bgp.go
index 02da5339..e89cf3e0 100644
--- a/e2etest/bgptests/bgp.go
+++ b/e2etest/bgptests/bgp.go
@@ -389,7 +389,6 @@ var _ = ginkgo.Describe("BGP", func() {
 			func(svc *corev1.Service) {
 				svc.Spec.LoadBalancerIP = serviceIP
 				svc.Annotations = map[string]string{"metallb.io/allow-shared-ip": "foo"}
-				svc.Spec.Ports[0].Port = int32(testservice.TestServicePort)
 			})
 		defer testservice.Delete(cs, svc)
 		svc1, _ := testservice.CreateWithBackendPort(cs, testNamespace, "second-service",
@@ -397,7 +396,6 @@ var _ = ginkgo.Describe("BGP", func() {
 			func(svc *corev1.Service) {
 				svc.Spec.LoadBalancerIP = serviceIP
 				svc.Annotations = map[string]string{"metallb.io/allow-shared-ip": "foo"}
-				svc.Spec.Ports[0].Port = int32(testservice.TestServicePort + 1)
 			})
 		defer testservice.Delete(cs, svc1)
 
diff --git a/e2etest/l2tests/assignment.go b/e2etest/l2tests/assignment.go
index d74bc876..f63d02a1 100644
--- a/e2etest/l2tests/assignment.go
+++ b/e2etest/l2tests/assignment.go
@@ -20,6 +20,7 @@ import (
 	metallbv1beta1 "go.universe.tf/metallb/api/v1beta1"
 
 	jigservice "go.universe.tf/e2etest/pkg/jigservice"
+	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -29,7 +30,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-const secondNamespace = "test-namespace"
+const (
+	secondNamespace         = "test-namespace"
+	allowSharedIPAnnotation = "metallb.io" + "/" + "allow-shared-ip"
+)
 
 var (
 	firstNsLabels = map[string]string{
@@ -223,6 +227,106 @@ var _ = ginkgo.Describe("IP Assignment", func() {
 				restartAndAssert()
 			}
 		})
+
+		ginkgo.It("should reconsider services when sharing the ip and conditions change", func() {
+			ip, err := config.GetIPFromRangeByIndex(IPV4ServiceRange, 0)
+			Expect(err).NotTo(HaveOccurred())
+
+			resources := config.Resources{
+				Pools: []metallbv1beta1.IPAddressPool{
+					{
+						ObjectMeta: metav1.ObjectMeta{
+							Name: "singleip-pool",
+						},
+						Spec: metallbv1beta1.IPAddressPoolSpec{
+							Addresses: []string{
+								fmt.Sprintf("%s/32", ip),
+							},
+						},
+					},
+				},
+			}
+			err = ConfigUpdater.Update(resources)
+			Expect(err).NotTo(HaveOccurred())
+
+			checkHasLBIP := func(svc *corev1.Service, ip string) {
+				Eventually(func() string {
+					s, err := cs.CoreV1().Services(svc.Namespace).Get(context.Background(), svc.Name, metav1.GetOptions{})
+					Expect(err).NotTo(HaveOccurred())
+					if len(s.Status.LoadBalancer.Ingress) == 0 {
+						return ""
+					}
+					return s.Status.LoadBalancer.Ingress[0].IP
+				}, time.Minute, 1*time.Second).Should(Equal(ip))
+			}
+
+			checkLBIPGetsRemoved := func(svc *corev1.Service) {
+				Eventually(func() int {
+					s, err := cs.CoreV1().Services(svc.Namespace).Get(context.Background(), svc.Name, metav1.GetOptions{})
+					Expect(err).NotTo(HaveOccurred())
+					return len(s.Status.LoadBalancer.Ingress)
+				}, time.Minute, 1*time.Second).Should(BeZero())
+			}
+
+			svc, jig := service.CreateWithBackendPort(cs, testNamespace, "changesharing", 80, func(svc *corev1.Service) {
+				service.TrafficPolicyCluster(svc)
+				svc.Annotations = map[string]string{allowSharedIPAnnotation: "share"}
+			})
+
+			ginkgo.By("Creating another service")
+			svc1, jig1 := service.CreateWithBackendPort(cs, testNamespace, "changesharing1", 81, func(svc *corev1.Service) {
+				service.TrafficPolicyCluster(svc)
+				svc.Annotations = map[string]string{allowSharedIPAnnotation: "share"}
+				svc.Spec.Selector = jig.Labels // assigning the same selector to the two services
+			})
+
+			Expect(err).NotTo(HaveOccurred())
+			defer func() {
+				service.Delete(cs, svc)
+				service.Delete(cs, svc1)
+			}()
+
+			checkHasLBIP(svc, ip)
+			checkHasLBIP(svc1, ip)
+
+			ginkgo.By("changing etp policy of the second service to local")
+			jig1.UpdateService(context.Background(), func(svc *corev1.Service) {
+				svc.Spec.ExternalTrafficPolicy = corev1.ServiceExternalTrafficPolicyLocal
+			})
+
+			ginkgo.By("checking the second service loses its ip")
+			checkLBIPGetsRemoved(svc1)
+
+			ginkgo.By("checking the first service keeps its ip")
+			Consistently(func() string {
+				toCheck, err := cs.CoreV1().Services(svc.Namespace).Get(context.Background(), svc.Name, metav1.GetOptions{})
+				Expect(err).NotTo(HaveOccurred())
+				return toCheck.Status.LoadBalancer.Ingress[0].IP
+			}, 2*time.Minute, 1*time.Second).Should(Equal(ip))
+
+			ginkgo.By("changing etp policy of the first service to local")
+			jig.UpdateService(context.Background(), func(svc *corev1.Service) {
+				svc.Spec.ExternalTrafficPolicy = corev1.ServiceExternalTrafficPolicyLocal
+			})
+
+			ginkgo.By("checking both services now got their ips")
+			checkHasLBIP(svc, ip)
+			checkHasLBIP(svc1, ip)
+
+			ginkgo.By("changing the selector of the first service")
+			jig.UpdateService(context.Background(), func(svc *corev1.Service) {
+				svc.Spec.Selector = map[string]string{"foo": "bar"}
+			})
+			ginkgo.By("checking the first service loses its ip")
+			checkLBIPGetsRemoved(svc)
+
+			ginkgo.By("aligning the selector of the second service")
+			jig1.UpdateService(context.Background(), func(svc *corev1.Service) {
+				svc.Spec.Selector = map[string]string{"foo": "bar"}
+			})
+			checkHasLBIP(svc, ip)
+			checkHasLBIP(svc1, ip)
+		})
 	})
 
 	ginkgo.Context("IPV4 removing pools", func() {
diff --git a/e2etest/pkg/service/create.go b/e2etest/pkg/service/create.go
index 219f878d..192bf483 100644
--- a/e2etest/pkg/service/create.go
+++ b/e2etest/pkg/service/create.go
@@ -50,6 +50,7 @@ func Delete(cs clientset.Interface, svc *corev1.Service) {
 }
 
 func tweakServicePort(svc *corev1.Service, port int) {
+	svc.Spec.Ports[0].Port = int32(port)
 	svc.Spec.Ports[0].TargetPort = intstr.FromInt(port)
 }
 
diff --git a/speaker/layer2_controller_test.go b/speaker/layer2_controller_test.go
index 572fc878..d7320118 100644
--- a/speaker/layer2_controller_test.go
+++ b/speaker/layer2_controller_test.go
@@ -12,6 +12,7 @@ import (
 	"go.universe.tf/metallb/internal/config"
 	"go.universe.tf/metallb/internal/k8s/controllers"
 	"go.universe.tf/metallb/internal/layer2"
+	"go.universe.tf/metallb/internal/speakerlist"
 
 	"github.com/go-kit/log"
 	v1 "k8s.io/api/core/v1"
@@ -25,8 +26,11 @@ type fakeSpeakerList struct {
 	speakers map[string]bool
 }
 
-func (sl *fakeSpeakerList) UsableSpeakers() map[string]bool {
-	return sl.speakers
+func (sl *fakeSpeakerList) UsableSpeakers() speakerlist.SpeakerListInfo {
+	return speakerlist.SpeakerListInfo{
+		Nodes:    sl.speakers,
+		Disabled: sl.speakers == nil,
+	}
 }
 
 func (sl *fakeSpeakerList) Rejoin() {}
@@ -46,6 +50,93 @@ func compareUseableNodesReturnedValue(a, b []string) bool {
 	return true
 }
 
+func TestUsableNodesEPSlicesWithSpeakerlistDisabled(t *testing.T) {
+	// Create a speaker response with a null list as expected when speaker is disabled
+	fakeSL := &fakeSpeakerList{
+		speakers: nil,
+	}
+	c1, err := newController(controllerConfig{
+		MyNode:  "iris1",
+		Logger:  log.NewLogfmtLogger(log.NewSyncWriter(os.Stderr)),
+		SList:   fakeSL,
+		bgpType: bgpNative,
+	})
+	if err != nil {
+		t.Fatalf("creating controller: %s", err)
+	}
+	c1.client = &testK8S{t: t}
+	allNodes := map[string]*v1.Node{
+		"iris1": {},
+		"iris2": {},
+	}
+
+	advertisementsForNode := []*config.L2Advertisement{
+		{
+			Nodes: map[string]bool{
+				"iris1": true,
+				"iris2": true,
+			},
+		},
+	}
+
+	conf := &config.Config{
+		Pools: &config.Pools{ByName: map[string]*config.Pool{
+			"default": {
+				CIDR:             []*net.IPNet{ipnet("10.20.30.0/24")},
+				L2Advertisements: advertisementsForNode,
+			},
+		}},
+	}
+
+	svc := &v1.Service{
+		Spec: v1.ServiceSpec{
+			Type:                  "LoadBalancer",
+			ExternalTrafficPolicy: v1.ServiceExternalTrafficPolicyTypeLocal,
+		},
+		Status: statusAssigned("10.20.30.1"),
+	}
+	eps := map[string][]discovery.EndpointSlice{
+		"10.20.30.1": {
+			{
+				Endpoints: []discovery.Endpoint{
+					{
+						Addresses: []string{
+							"2.3.4.5",
+						},
+						NodeName: ptr.To("iris1"),
+						Conditions: discovery.EndpointConditions{
+							Ready: ptr.To(true),
+						},
+					},
+					{
+						Addresses: []string{
+							"2.3.4.15",
+						},
+						NodeName: ptr.To("iris1"),
+						Conditions: discovery.EndpointConditions{
+							Ready: ptr.To(true),
+						},
+					},
+				},
+			},
+		},
+	}
+
+	lbIP := net.ParseIP(svc.Status.LoadBalancer.Ingress[0].IP)
+	lbIPStr := lbIP.String()
+	l := log.NewNopLogger()
+	response := c1.protocolHandlers[config.Layer2].ShouldAnnounce(l,
+		"test1",
+		[]net.IP{lbIP},
+		conf.Pools.ByName["default"],
+		svc,
+		eps[lbIPStr],
+		allNodes)
+	if response != "" {
+		t.Errorf("Expecting fallback when speakers list is not configured but got: %v", response)
+	}
+}
+
 func TestUsableNodesEPSlices(t *testing.T) {
 	b := &fakeBGP{
 		t: t,
@@ -256,7 +347,7 @@ func TestUsableNodesEPSlices(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		response := usableNodes(test.eps, test.usableSpeakers)
+		response := nodesWithEndpoint(test.eps, test.usableSpeakers)
 		sort.Strings(response)
 		if !compareUseableNodesReturnedValue(response, test.cExpectedResult) {
 			t.Errorf("%q: shouldAnnounce for controller returned incorrect result, expected '%s', but received '%s'", test.desc, test.cExpectedResult, response)