Repository: metallb/metallb
Pull request #744 (2020-10-13 20:56:00)

Update manifests/metallb.yaml for kubernetes v1.19
According to kubernetes documentation, since v1.19 kubernetes stops using labels with "beta." prefix. Installation manifest manifests/metallb.yaml fixes nodeSelector to kubernetes.io/os.

Fixed   #743

russellb: lgtm, thanks!
champtar: What is the new minimum version of K8S supported ?
honza801: see `https://v1-15.docs.kubernetes.io/docs/setup/release/notes/`

```
The beta.kubernetes.io/os and beta.kubernetes.io/arch labels, deprecated since v1.14, are targeted for removal in v1.18.
```

i cannot find the earliest version with `kubernetes.io/os` support, because older doc than v1.15 does not exist anymore
champtar: ok perfect thanks
LGTM

Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 52742cc0..37c211ad 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -3,57 +3,47 @@
 version: 2
 jobs:
   test-1.13:
-    working_directory: /go/src/go.universe.tf/metallb
     docker:
-      - image: circleci/golang:1.13
+      - image: cimg/go:1.13
     steps:
       - checkout
-      - run: sudo apt-get install python-pip
-      - run: sudo pip install invoke semver pyyaml
-      - run: echo "export GO111MODULE=on" >>$BASH_ENV
       - run: go test -short ./...
       - run: go test -short -race ./...
       - run: cp manifests/metallb.yaml manifests/metallb.yaml.prev
   lint-1.13:
-    working_directory: /go/src/go.universe.tf/metallb
     docker:
-      - image: circleci/golang:1.13
+      - image: cimg/go:1.13
     steps:
       - checkout
-      - run: echo "export GO111MODULE=on" >>$BASH_ENV
       - run: curl -L https://git.io/vp6lP | sh # gometalinter
       - run: PATH=./bin:$PATH gometalinter --deadline=5m --disable-all --enable=gofmt --enable=vet --vendor ./...
-  deploy-controller:
-    working_directory: /go/src/go.universe.tf/metallb
+  publish-images:
     docker:
-      - image: circleci/golang:1.13
+      - image: cimg/go:1.13
     steps:
       - checkout
       - setup_remote_docker
-      - run: sudo apt-get install python-pip
-      - run: sudo pip install invoke semver pyyaml
+      - run: sudo apt-get update && sudo apt-get install -y python-pip
+      - run: pip install invoke semver pyyaml
       - run: docker login -u $DOCKER_USER -p $DOCKER_PASSWORD
       - run: mkdir -p ./bin
-      - run: wget -O ./bin/manifest-tool https://github.com/estesp/manifest-tool/releases/download/v1.0.0/manifest-tool-linux-amd64 && chmod +x ./bin/manifest-tool
-      - run: echo "export GO111MODULE=on" >>$BASH_ENV
-      - run: PATH=./bin:$PATH inv push-multiarch --binaries=controller --tag=${CIRCLE_BRANCH:-${CIRCLE_TAG}} --docker-user=metallb
-  deploy-speaker:
-    working_directory: /go/src/go.universe.tf/metallb
+      - run: wget -O ./bin/manifest-tool https://github.com/estesp/manifest-tool/releases/download/v1.0.3/manifest-tool-linux-amd64 && chmod +x ./bin/manifest-tool
+      - run: PATH=./bin:$PATH inv push-multiarch --binaries=controller --binaries=speaker --registry=docker.io --repo=metallb --tag=${CIRCLE_BRANCH:-${CIRCLE_TAG}}
+  publish-images-quay:
     docker:
-      - image: circleci/golang:1.13
+      - image: cimg/go:1.13
     steps:
       - checkout
       - setup_remote_docker
-      - run: sudo apt-get install python-pip
-      - run: sudo pip install invoke semver pyyaml
-      - run: docker login -u $DOCKER_USER -p $DOCKER_PASSWORD
+      - run: sudo apt-get update && sudo apt-get install -y python-pip
+      - run: pip install invoke semver pyyaml
+      - run: docker login quay.io -u $QUAY_USER -p $QUAY_PASSWORD
       - run: mkdir -p ./bin
-      - run: wget -O ./bin/manifest-tool https://github.com/estesp/manifest-tool/releases/download/v1.0.0/manifest-tool-linux-amd64 && chmod +x ./bin/manifest-tool
-      - run: echo "export GO111MODULE=on" >>$BASH_ENV
-      - run: PATH=./bin:$PATH inv push-multiarch --binaries=speaker --tag=${CIRCLE_BRANCH:-${CIRCLE_TAG}} --docker-user=metallb
+      - run: wget -O ./bin/manifest-tool https://github.com/estesp/manifest-tool/releases/download/v1.0.3/manifest-tool-linux-amd64 && chmod +x ./bin/manifest-tool
+      - run: PATH=./bin:$PATH inv push-multiarch --binaries=controller --binaries=speaker --registry=quay.io --repo=metallb --tag=${CIRCLE_BRANCH:-${CIRCLE_TAG}}
 workflows:
   version: 2
-  test-and-deploy:
+  test-and-publish:
     jobs:
       - test-1.13:
           filters:
@@ -63,7 +53,7 @@ workflows:
           filters:
             tags:
               only: /.*/
-      - deploy-controller:
+      - publish-images:
           filters:
             branches:
               only:
@@ -74,7 +64,7 @@ workflows:
           requires:
             - test-1.13
             - lint-1.13
-      - deploy-speaker:
+      - publish-images-quay:
           filters:
             branches:
               only:
diff --git a/internal/k8s/k8s.go b/internal/k8s/k8s.go
index 58b186c1..2a346c1f 100644
--- a/internal/k8s/k8s.go
+++ b/internal/k8s/k8s.go
@@ -259,8 +259,8 @@ func New(cfg *Config) (*Client, error) {
 	return c, nil
 }
 
-// GetPodsIPs get the IPs from all the pods matched by the labels string
-func (c *Client) GetPodsIPs(namespace, labels string) ([]string, error) {
+// PodIPs returns the IPs of all the pods matched by the labels string.
+func (c *Client) PodIPs(namespace, labels string) ([]string, error) {
 	pl, err := c.client.CoreV1().Pods(namespace).List(metav1.ListOptions{LabelSelector: labels})
 	if err != nil {
 		return nil, err
diff --git a/internal/layer2/announcer.go b/internal/layer2/announcer.go
index 1460490e..df9e91ec 100644
--- a/internal/layer2/announcer.go
+++ b/internal/layer2/announcer.go
@@ -20,6 +20,10 @@ type Announce struct {
 	ndps     map[int]*ndpResponder
 	ips      map[string]net.IP // svcName -> IP
 	ipRefcnt map[string]int    // ip.String() -> number of uses
+
+	// This channel can block - do not write to it while holding the mutex
+	// to avoid deadlocking.
+	spamCh chan net.IP
 }
 
 // New returns an initialized Announce.
@@ -30,8 +34,10 @@ func New(l log.Logger) (*Announce, error) {
 		ndps:     map[int]*ndpResponder{},
 		ips:      map[string]net.IP{},
 		ipRefcnt: map[string]int{},
+		spamCh:   make(chan net.IP, 1024),
 	}
 	go ret.interfaceScan()
+	go ret.spamLoop()
 
 	return ret, nil
 }
@@ -131,24 +137,52 @@ func (a *Announce) updateInterfaces() {
 	return
 }
 
-func (a *Announce) spam(name string) {
-	// TODO: should abort if we lose control of the IP mid-spam.
-	start := time.Now()
-	for time.Since(start) < 5*time.Second {
-		if err := a.gratuitous(name); err != nil {
-			a.logger.Log("op", "gratuitousAnnounce", "error", err, "service", name, "msg", "failed to make gratuitous IP announcement")
+func (a *Announce) spamLoop() {
+	// Map IP to spam stop time.
+	m := map[string]time.Time{}
+	// See https://github.com/metallb/metallb/issues/172 for the 1100 choice.
+	ticker := time.NewTicker(1100 * time.Millisecond)
+	for {
+		select {
+		case ip := <-a.spamCh:
+			ipStr := ip.String()
+			_, ok := m[ipStr]
+			// Set spam stop time to 5 seconds from now.
+			m[ipStr] = time.Now().Add(5 * time.Second)
+			if !ok {
+				// Spam right away to avoid waiting up to 1100 milliseconds even if
+				// it means we call spam() twice in a row in a short amount of time.
+				a.spam(ip)
+			}
+		case now := <-ticker.C:
+			for ipStr, until := range m {
+				if now.After(until) {
+					// We have spammed enough - remove the IP from the map.
+					delete(m, ipStr)
+				} else {
+					a.spam(net.ParseIP(ipStr))
+				}
+			}
 		}
-		time.Sleep(1100 * time.Millisecond)
 	}
 }
 
-func (a *Announce) gratuitous(name string) error {
-	a.Lock()
-	defer a.Unlock()
+func (a *Announce) doSpam(ip net.IP) {
+	a.spamCh <- ip
+}
 
-	ip, ok := a.ips[name]
-	if !ok {
-		// No IP means we've lost control of the IP, someone else is
+func (a *Announce) spam(ip net.IP) {
+	if err := a.gratuitous(ip); err != nil {
+		a.logger.Log("op", "gratuitousAnnounce", "error", err, "ip", ip, "msg", "failed to make gratuitous IP announcement")
+	}
+}
+
+func (a *Announce) gratuitous(ip net.IP) error {
+	a.RLock()
+	defer a.RUnlock()
+
+	if a.ipRefcnt[ip.String()] <= 0 {
+		// We've lost control of the IP, someone else is
 		// doing announcements.
 		return nil
 	}
@@ -181,6 +215,8 @@ func (a *Announce) shouldAnnounce(ip net.IP) dropReason {
 
 // SetBalancer adds ip to the set of announced addresses.
 func (a *Announce) SetBalancer(name string, ip net.IP) {
+	// Call doSpam at the end of the function without holding the lock
+	defer a.doSpam(ip)
 	a.Lock()
 	defer a.Unlock()
 
@@ -203,9 +239,6 @@ func (a *Announce) SetBalancer(name string, ip net.IP) {
 			a.logger.Log("op", "watchMulticastGroup", "error", err, "ip", ip, "msg", "failed to watch NDP multicast group for IP, NDP responder will not respond to requests for this address")
 		}
 	}
-
-	go a.spam(name)
-
 }
 
 // DeleteBalancer deletes an address from the set of addresses we should announce.
diff --git a/internal/speakerlist/speakerlist.go b/internal/speakerlist/speakerlist.go
new file mode 100644
index 00000000..7d60ad01
--- /dev/null
+++ b/internal/speakerlist/speakerlist.go
@@ -0,0 +1,319 @@
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package speakerlist
+
+import (
+	"crypto/sha256"
+	golog "log"
+	"strconv"
+	"sync"
+	"time"
+
+	"go.universe.tf/metallb/internal/k8s"
+
+	gokitlog "github.com/go-kit/kit/log"
+	"github.com/hashicorp/memberlist"
+)
+
+// SpeakerList represents a list of healthy speakers.
+type SpeakerList struct {
+	l         gokitlog.Logger
+	client    *k8s.Client
+	stopCh    chan struct{}
+	namespace string
+	labels    string
+
+	// The following fields are nil when memberlist is disabled.
+	mlEventCh chan memberlist.NodeEvent
+	ml        *memberlist.Memberlist
+	mlJoinCh  chan struct{}
+
+	mlMux        sync.Mutex // Mutex for mlSpeakerIPs.
+	mlSpeakerIPs []string   // Speaker pod IPs.
+}
+
+// New creates a new SpeakerList and returns a pointer to it.
+func New(logger gokitlog.Logger, nodeName, bindAddr, bindPort, secret, namespace, labels string, stopCh chan struct{}) (*SpeakerList, error) {
+	sl := SpeakerList{
+		l:         logger,
+		stopCh:    stopCh,
+		namespace: namespace,
+		labels:    labels,
+	}
+
+	if namespace == "" || labels == "" || bindAddr == "" {
+		logger.Log("op", "startup", "msg", "not starting fast dead node detection (memberlist), need ml-bindaddr / ml-labels / ml-namespace config")
+		return &sl, nil
+	}
+
+	mconfig := memberlist.DefaultLANConfig()
+
+	// mconfig.Name MUST be equal to the spec.nodeName field of the speaker pod as we match it
+	// against the nodeName field of Endpoint objects inside usableNodes().
+	mconfig.Name = nodeName
+	mconfig.BindAddr = bindAddr
+	if bindPort != "" {
+		mlport, err := strconv.Atoi(bindPort)
+		if err != nil {
+			logger.Log("op", "startup", "error", "unable to parse ml-bindport", "msg", err)
+			return nil, err
+		}
+		mconfig.BindPort = mlport
+		mconfig.AdvertisePort = mlport
+	}
+	loggerout := gokitlog.NewStdlibAdapter(gokitlog.With(sl.l, "component", "Memberlist"))
+	mconfig.Logger = golog.New(loggerout, "", golog.Lshortfile)
+	if secret == "" {
+		logger.Log("op", "startup", "warning", "no ml-secret-key set, memberlist traffic will not be encrypted")
+	} else {
+		sha := sha256.New()
+		mconfig.SecretKey = sha.Sum([]byte(secret))[:16]
+	}
+
+	// This channel is used by the Rejoin() method which runs on k8s node
+	// changes. A buffered channel is used to avoid blocking the main goroutine
+	// while waiting for the join operation to complete.
+	sl.mlJoinCh = make(chan struct{}, 1)
+
+	// ChannelEventDelegate hints that it should not block, so make mlEventCh
+	// 'big'.
+	// TODO: See https://github.com/metallb/metallb/issues/716
+	sl.mlEventCh = make(chan memberlist.NodeEvent, 1024)
+	mconfig.Events = &memberlist.ChannelEventDelegate{Ch: sl.mlEventCh}
+
+	ml, err := memberlist.Create(mconfig)
+	if err != nil {
+		logger.Log("op", "startup", "error", err, "msg", "failed to create memberlist")
+		return nil, err
+	}
+
+	sl.ml = ml
+
+	return &sl, nil
+}
+
+// Start initializes the SpeakerList. This functions must be called before using
+// other SpeakerList methods.
+func (sl *SpeakerList) Start(client *k8s.Client) {
+	// TODO: The k8s client parameter should ideally be a parameter to
+	// New(). However, that is not possible today because:
+	//
+	// 1. The newController() function takes the sList as param[1]
+	// 2. The newController() function uses the sList param to create the layer 2 controller[2]
+	// 3. The new Kubernetes client uses the created controller to set the callbacks[3]
+	//
+	// Then, we have a dependency cycle if we add the Kubernetes client to speakerlist.New():
+	//
+	// 1. To create a Kubernetes client we need a controller
+	// 2. The newController() function uses the sList param to create the layer 2 controller[2]
+	// 3. The new Kubernetes client uses the created controller to set the callbacks[3]
+	//
+	// We should probably move the speakerlist code to be an implementation
+	// detail of the Layer 2 controller and, when doing so, try to fix this
+	// problem.
+	//
+	// [1]: https://github.com/metallb/metallb/pull/662/files#diff-60053ad6fecb5a3cfabb6f3d9e720899R109
+	// [2]: https://github.com/metallb/metallb/pull/662/files#diff-60053ad6fecb5a3cfabb6f3d9e720899L232-L251
+	// [3]: https://github.com/metallb/metallb/pull/662/files#diff-60053ad6fecb5a3cfabb6f3d9e720899L160-L162
+
+	sl.client = client
+
+	if sl.ml == nil {
+		return
+	}
+
+	// Initialize sl.mlSpeakerIPs.
+	iplist, err := sl.mlSpeakers()
+	if err != nil {
+		sl.l.Log("op", "memberDiscovery", "error", err, "msg", "failed to get pod IPs")
+		iplist = nil
+	}
+
+	sl.mlMux.Lock()
+	sl.mlSpeakerIPs = iplist
+	sl.mlMux.Unlock()
+
+	// Update mlSpeakerIPs in the background.
+	go sl.updateSpeakerIPs()
+
+	go sl.memberlistWatchEvents()
+	go sl.joinMembers()
+}
+
+// updateSpeakerIPs runs forever updating the sl.mlSpeakerIPs slice with the
+// IPs of all the speaker pods. As the function queries the API server, it waits at
+// least 5 minutes between queries to avoid self-induced API server DoS (should
+// be treated with care).
+func (sl *SpeakerList) updateSpeakerIPs() {
+	for {
+		// This either stops the loop or makes sure to wait at least 5
+		// minutes before making another call to sl.mlSpeakers(), which
+		// queries the API server.
+		select {
+		case <-sl.stopCh:
+			return
+		case <-time.After(5 * time.Minute):
+			// This blocks until the API server responds.
+			iplist, err := sl.mlSpeakers()
+			if err != nil {
+				sl.l.Log("op", "memberDiscovery", "error", err, "msg", "failed to get pod IPs")
+				continue
+			}
+
+			sl.mlMux.Lock()
+			sl.mlSpeakerIPs = iplist
+			sl.mlMux.Unlock()
+		}
+	}
+}
+
+func (sl *SpeakerList) mlSpeakers() ([]string, error) {
+	// This call blocks until we get a response from the API server.
+	// In the client-go version we are using, there is no way to use a
+	// context. Newer versions of client-go support using a context
+	// in the call sl.client.PodIPs() is using.
+	//
+	// TODO: When updating client-go, this code can be simplified by using a
+	// context in the following way:
+	// If we use a context, we can specify a timeout and then there is no need to
+	// run mlUpdateSpeaker() in the background. We could then call
+	// this function from mlJoin() with a context (timeout) and use
+	// sl.mlSpeakerIPs on failure. On success, we could update
+	// sl.mlSpeakerIPs with the response and continue to do the join.
+	iplist, err := sl.client.PodIPs(sl.namespace, sl.labels)
+	if err != nil {
+		return nil, err
+	}
+
+	return iplist, nil
+}
+
+func (sl *SpeakerList) joinMembers() {
+	// Every one minute, try to rejoin.
+	// This joins nodes that leave the cluster for just a few seconds.
+	// Discovering new IPs (updated by sl.updateSpeakerIPs()) might take a while
+	// longer.
+	ticker := time.NewTicker(1 * time.Minute)
+
+	for {
+		select {
+		case <-sl.stopCh:
+			return
+		case <-ticker.C:
+			sl.mlJoin()
+		case <-sl.mlJoinCh:
+			sl.mlJoin()
+		}
+	}
+}
+
+func (sl *SpeakerList) members() map[string]struct{} {
+	members := make(map[string]struct{})
+
+	for _, node := range sl.ml.Members() {
+		ip := node.Addr.String()
+		members[ip] = struct{}{}
+	}
+
+	return members
+}
+
+// mlJoin joins speaker pods that are not members of this cluster
+// to the cluster. It performs a memberlist.Join() with the IPs in
+// mlSpeakerIPs that are not members of the cluster.
+func (sl *SpeakerList) mlJoin() {
+	// IPs to be joined to the memberlist cluster.
+	var joinIPs []string
+
+	members := sl.members()
+
+	sl.mlMux.Lock()
+	defer sl.mlMux.Unlock()
+
+	for _, ip := range sl.mlSpeakerIPs {
+		// If an IP is not a member of the cluster, add it to joinIPs.
+		if _, isMember := members[ip]; !isMember {
+			joinIPs = append(joinIPs, ip)
+		}
+	}
+
+	if len(joinIPs) == 0 {
+		// Not logging here to avoid spamming the logs.
+		return
+	}
+
+	nr, err := sl.ml.Join(joinIPs)
+	if err != nil || nr != len(joinIPs) {
+		sl.l.Log("op", "memberDiscovery", "msg", "partial join", "joined", nr, "expected", len(joinIPs), "error", err)
+	} else {
+		sl.l.Log("op", "Member detection", "msg", "memberlist join succesfully", "number of other nodes", nr)
+	}
+}
+
+// Rejoin initiates a discovery and joins all the speakers to the memberlist
+// cluster.
+func (sl *SpeakerList) Rejoin() {
+	if sl.ml == nil {
+		return
+	}
+
+	select {
+	case sl.mlJoinCh <- struct{}{}:
+		sl.l.Log("op", "memberDiscovery", "msg", "triggering discovery")
+	default:
+		sl.l.Log("op", "memberDiscovery", "msg", "previous discovery in progress - doing nothing")
+	}
+
+	return
+}
+
+// UsableSpeakers returns a map of usable speaker nodes.
+func (sl *SpeakerList) UsableSpeakers() map[string]bool {
+	if sl.ml == nil {
+		return nil
+	}
+	activeNodes := map[string]bool{}
+	for _, n := range sl.ml.Members() {
+		activeNodes[n.Name] = true
+	}
+	return activeNodes
+}
+
+// Stop stops the SpeakerList.
+func (sl *SpeakerList) Stop() {
+	if sl.ml == nil {
+		return
+	}
+
+	sl.l.Log("op", "shutdown", "msg", "leaving memberlist cluster")
+	err := sl.ml.Leave(time.Second)
+	sl.l.Log("op", "shutdown", "msg", "left memberlist cluster", "error", err)
+	err = sl.ml.Shutdown()
+	sl.l.Log("op", "shutdown", "msg", "memberlist shutdown", "error", err)
+}
+
+func event2String(e memberlist.NodeEventType) string {
+	return []string{"NodeJoin", "NodeLeave", "NodeUpdate"}[e]
+}
+
+func (sl *SpeakerList) memberlistWatchEvents() {
+	for {
+		select {
+		case e := <-sl.mlEventCh:
+			sl.l.Log("msg", "node event - forcing sync", "node addr", e.Node.Addr, "node name", e.Node.Name, "node event", event2String(e.Event))
+			sl.client.ForceSync()
+		case <-sl.stopCh:
+			return
+		}
+	}
+}
diff --git a/manifests/metallb.yaml b/manifests/metallb.yaml
index 7742e8c6..70192018 100644
--- a/manifests/metallb.yaml
+++ b/manifests/metallb.yaml
@@ -335,7 +335,7 @@ spec:
           readOnlyRootFilesystem: true
       hostNetwork: true
       nodeSelector:
-        beta.kubernetes.io/os: linux
+        kubernetes.io/os: linux
       serviceAccountName: speaker
       terminationGracePeriodSeconds: 2
       tolerations:
@@ -386,7 +386,7 @@ spec:
             - all
           readOnlyRootFilesystem: true
       nodeSelector:
-        beta.kubernetes.io/os: linux
+        kubernetes.io/os: linux
       securityContext:
         runAsNonRoot: true
         runAsUser: 65534
diff --git a/speaker/layer2_controller.go b/speaker/layer2_controller.go
index 2209b563..83e3e101 100644
--- a/speaker/layer2_controller.go
+++ b/speaker/layer2_controller.go
@@ -21,7 +21,6 @@ import (
 	"sort"
 
 	"github.com/go-kit/kit/log"
-	"github.com/hashicorp/memberlist"
 	"go.universe.tf/metallb/internal/config"
 	"go.universe.tf/metallb/internal/layer2"
 	"k8s.io/api/core/v1"
@@ -30,7 +29,7 @@ import (
 type layer2Controller struct {
 	announcer *layer2.Announce
 	myNode    string
-	mList     *memberlist.Memberlist
+	sList     SpeakerList
 }
 
 func (c *layer2Controller) SetConfig(log.Logger, *config.Config) error {
@@ -39,23 +38,18 @@ func (c *layer2Controller) SetConfig(log.Logger, *config.Config) error {
 
 // usableNodes returns all nodes that have at least one fully ready
 // endpoint on them.
-func usableNodes(eps *v1.Endpoints, mList *memberlist.Memberlist) []string {
-	var activeNodes map[string]bool
-	if mList != nil {
-		activeNodes = map[string]bool{}
-		for _, n := range mList.Members() {
-			activeNodes[n.Name] = true
-		}
-	}
-
+// The speakers parameter is a map with the node name as key and the readiness
+// status as value (true means ready, false means not ready).
+// If the speakers map is nil, it is ignored.
+func usableNodes(eps *v1.Endpoints, speakers map[string]bool) []string {
 	usable := map[string]bool{}
 	for _, subset := range eps.Subsets {
 		for _, ep := range subset.Addresses {
 			if ep.NodeName == nil {
 				continue
 			}
-			if activeNodes != nil {
-				if _, ok := activeNodes[*ep.NodeName]; !ok {
+			if speakers != nil {
+				if ready, ok := speakers[*ep.NodeName]; !ok || !ready {
 					continue
 				}
 			}
@@ -76,7 +70,7 @@ func usableNodes(eps *v1.Endpoints, mList *memberlist.Memberlist) []string {
 }
 
 func (c *layer2Controller) ShouldAnnounce(l log.Logger, name string, svc *v1.Service, eps *v1.Endpoints) string {
-	nodes := usableNodes(eps, c.mList)
+	nodes := usableNodes(eps, c.sList.UsableSpeakers())
 	// Sort the slice by the hash of node + service name. This
 	// produces an ordering of ready nodes that is unique to this
 	// service.
@@ -110,5 +104,6 @@ func (c *layer2Controller) DeleteBalancer(l log.Logger, name, reason string) err
 }
 
 func (c *layer2Controller) SetNode(log.Logger, *v1.Node) error {
+	c.sList.Rejoin()
 	return nil
 }
diff --git a/speaker/main.go b/speaker/main.go
index 4f1089aa..a68804e5 100644
--- a/speaker/main.go
+++ b/speaker/main.go
@@ -15,27 +15,23 @@
 package main
 
 import (
-	"crypto/sha256"
 	"flag"
 	"fmt"
-	golog "log"
 	"net"
 	"os"
 	"os/signal"
-	"strconv"
 	"syscall"
-	"time"
 
 	"go.universe.tf/metallb/internal/bgp"
 	"go.universe.tf/metallb/internal/config"
 	"go.universe.tf/metallb/internal/k8s"
 	"go.universe.tf/metallb/internal/layer2"
 	"go.universe.tf/metallb/internal/logging"
+	"go.universe.tf/metallb/internal/speakerlist"
 	"go.universe.tf/metallb/internal/version"
 	v1 "k8s.io/api/core/v1"
 
 	gokitlog "github.com/go-kit/kit/log"
-	"github.com/hashicorp/memberlist"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -101,44 +97,16 @@ func main() {
 	}()
 	defer logger.Log("op", "shutdown", "msg", "done")
 
-	var mlist *memberlist.Memberlist
-	var eventCh chan memberlist.NodeEvent
-	if *mlNamespace == "" || *mlLabels == "" || *mlBindAddr == "" {
-		logger.Log("op", "startup", "msg", "Not starting fast dead node detection (MemberList), need ml-bindaddr / ml-labels / ml-namespace config")
-	} else {
-		mconfig := memberlist.DefaultLANConfig()
-		// mconfig.Name MUST be spec.nodeName, as we will match it against Enpoints nodeName in usableNodes()
-		mconfig.Name = *myNode
-		mconfig.BindAddr = *mlBindAddr
-		if *mlBindPort != "" {
-			mlport, err := strconv.Atoi(*mlBindPort)
-			if err != nil {
-				logger.Log("op", "startup", "error", "unable to parse ml-bindport", "msg", err)
-				os.Exit(1)
-			}
-			mconfig.BindPort = mlport
-			mconfig.AdvertisePort = mlport
-		}
-		loggerout := gokitlog.NewStdlibAdapter(gokitlog.With(logger, "component", "MemberList"))
-		mconfig.Logger = golog.New(loggerout, "", golog.Lshortfile)
-		if *mlSecret != "" {
-			sha := sha256.New()
-			mconfig.SecretKey = sha.Sum([]byte(*mlSecret))[:16]
-		}
-		eventCh = make(chan memberlist.NodeEvent, 16)
-		mconfig.Events = &memberlist.ChannelEventDelegate{Ch: eventCh}
-		mlist, err = memberlist.Create(mconfig)
-		if err != nil {
-			logger.Log("op", "startup", "error", err, "msg", "failed to create memberlist")
-			os.Exit(1)
-		}
+	sList, err := speakerlist.New(logger, *myNode, *mlBindAddr, *mlBindPort, *mlSecret, *mlNamespace, *mlLabels, stopCh)
+	if err != nil {
+		os.Exit(1)
 	}
 
 	// Setup all clients and speakers, config decides what is being done runtime.
 	ctrl, err := newController(controllerConfig{
 		MyNode: *myNode,
 		Logger: logger,
-		MList:  mlist,
+		SList:  sList,
 	})
 	if err != nil {
 		logger.Log("op", "startup", "error", err, "msg", "failed to create MetalLB controller")
@@ -167,47 +135,14 @@ func main() {
 	}
 	ctrl.client = client
 
-	if mlist != nil {
-		go watchMemberListEvents(logger, eventCh, stopCh, client)
-
-		iplist, err := client.GetPodsIPs(*mlNamespace, *mlLabels)
-		if err != nil {
-			logger.Log("op", "startup", "error", err, "msg", "failed to get PodsIPs")
-			os.Exit(1)
-		}
-		n, err := mlist.Join(iplist)
-		logger.Log("op", "startup", "msg", "Memberlist join", "nb joigned", n, "error ?", err)
-		defer func() {
-			logger.Log("op", "shutdown", "msg", "leaving MemberList cluster")
-			err = mlist.Leave(time.Second)
-			logger.Log("op", "shutdown", "msg", "left MemberList cluster", "error ?", err)
-			mlist.Shutdown()
-			logger.Log("op", "shutdown", "msg", "MemberList shutdown", "error ?", err)
-		}()
-	}
+	sList.Start(client)
+	defer sList.Stop()
 
 	if err := client.Run(stopCh); err != nil {
 		logger.Log("op", "startup", "error", err, "msg", "failed to run k8s client")
 	}
 }
 
-func event2String(e memberlist.NodeEventType) string {
-	return [...]string{"NodeJoin", "NodeLeave", "NodeUpdate"}[e]
-}
-
-func watchMemberListEvents(logger gokitlog.Logger, eventCh chan memberlist.NodeEvent, stopCh chan struct{}, client *k8s.Client) {
-	for {
-		select {
-		case e := <-eventCh:
-			logger.Log("msg", "Node event", "node addr", e.Node.Addr, "node name", e.Node.Name, "node event", event2String(e.Event))
-			logger.Log("msg", "Call Force Sync")
-			client.ForceSync()
-		case <-stopCh:
-			return
-		}
-	}
-}
-
 type controller struct {
 	myNode string
 
@@ -222,7 +157,7 @@ type controller struct {
 type controllerConfig struct {
 	MyNode string
 	Logger gokitlog.Logger
-	MList  *memberlist.Memberlist
+	SList  SpeakerList
 
 	// For testing only, and will be removed in a future release.
 	// See: https://github.com/google/metallb/issues/152.
@@ -246,7 +181,7 @@ func newController(cfg controllerConfig) (*controller, error) {
 		protocols[config.Layer2] = &layer2Controller{
 			announcer: a,
 			myNode:    cfg.MyNode,
-			mList:     cfg.MList,
+			sList:     cfg.SList,
 		}
 	}
 
@@ -429,3 +364,9 @@ type Protocol interface {
 	DeleteBalancer(gokitlog.Logger, string, string) error
 	SetNode(gokitlog.Logger, *v1.Node) error
 }
+
+// Speakerlist represents a list of healthy speakers.
+type SpeakerList interface {
+	UsableSpeakers() map[string]bool
+	Rejoin()
+}
diff --git a/tasks.py b/tasks.py
index 0a46e012..ff205576 100644
--- a/tasks.py
+++ b/tasks.py
@@ -63,10 +63,11 @@ def _make_build_dirs():
       help={
           "binaries": "binaries to build. One or more of {}, or 'all'".format(", ".join(sorted(all_binaries))),
           "architectures": "architectures to build. One or more of {}, or 'all'".format(", ".join(sorted(all_architectures))),
-          "tag": "docker image tag prefix to use. Actual tag will be <tag>-<arch>. Default 'dev'.",
-          "docker-user": "docker user under which to tag the images. Default 'metallb'.",
+          "registry": "Docker registry under which to tag the images. Default 'docker.io'.",
+          "repo": "Docker repository under which to tag the images. Default 'metallb'.",
+          "tag": "Docker image tag prefix to use. Actual tag will be <tag>-<arch>. Default 'dev'.",
       })
-def build(ctx, binaries, architectures, tag="dev", docker_user="metallb"):
+def build(ctx, binaries, architectures, registry="docker.io", repo="metallb", tag="dev"):
     """Build MetalLB docker images."""
     binaries = _check_binaries(binaries)
     architectures = _check_architectures(architectures)
@@ -95,9 +96,10 @@ def build(ctx, binaries, architectures, tag="dev", docker_user="metallb"):
                 env=env,
                 echo=True)
             run("docker build "
-                "-t {user}/{bin}:{tag}-{arch} "
+                "-t {registry}/{repo}/{bin}:{tag}-{arch} "
                 "-f {bin}/Dockerfile build/{arch}/{bin}".format(
-                    user=docker_user,
+                    registry=registry,
+                    repo=repo,
                     bin=bin,
                     tag=tag,
                     arch=arch),
@@ -107,19 +109,21 @@ def build(ctx, binaries, architectures, tag="dev", docker_user="metallb"):
       help={
           "binaries": "binaries to build. One or more of {}, or 'all'".format(", ".join(sorted(all_binaries))),
           "architectures": "architectures to build. One or more of {}, or 'all'".format(", ".join(sorted(all_architectures))),
-          "tag": "docker image tag prefix to use. Actual tag will be <tag>-<arch>. Default 'dev'.",
-          "docker-user": "docker user under which to tag the images. Default 'metallb'.",
+          "registry": "Docker registry under which to tag the images. Default 'docker.io'.",
+          "repo": "Docker repository under which to tag the images. Default 'metallb'.",
+          "tag": "Docker image tag prefix to use. Actual tag will be <tag>-<arch>. Default 'dev'.",
       })
-def push(ctx, binaries, architectures, tag="dev", docker_user="metallb"):
+def push(ctx, binaries, architectures, registry="docker.io", repo="metallb", tag="dev"):
     """Build and push docker images to registry."""
     binaries = _check_binaries(binaries)
     architectures = _check_architectures(architectures)
 
     for arch in architectures:
         for bin in binaries:
-            build(ctx, binaries=[bin], architectures=[arch], tag=tag, docker_user=docker_user)
-            run("docker push {user}/{bin}:{tag}-{arch}".format(
-                user=docker_user,
+            build(ctx, binaries=[bin], architectures=[arch], registry=registry, repo=repo, tag=tag)
+            run("docker push {registry}/{repo}/{bin}:{tag}-{arch}".format(
+                registry=registry,
+                repo=repo,
                 bin=bin,
                 arch=arch,
                 tag=tag),
@@ -128,23 +132,25 @@ def push(ctx, binaries, architectures, tag="dev", docker_user="metallb"):
 @task(iterable=["binaries"],
       help={
           "binaries": "binaries to build. One or more of {}, or 'all'".format(", ".join(sorted(all_binaries))),
-          "tag": "docker image tag prefix to use. Actual tag will be <tag>-<arch>. Default 'dev'.",
-          "docker-user": "docker user under which to tag the images. Default 'metallb'.",
+          "registry": "Docker registry under which to tag the images. Default 'docker.io'.",
+          "repo": "Docker repository under which to tag the images. Default 'metallb'.",
+          "tag": "Docker image tag prefix to use. Actual tag will be <tag>-<arch>. Default 'dev'.",
       })
-def push_multiarch(ctx, binaries, tag="dev", docker_user="metallb"):
+def push_multiarch(ctx, binaries, registry="docker.io", repo="metallb", tag="dev"):
     """Build and push multi-architecture docker images to registry."""
     binaries = _check_binaries(binaries)
     architectures = _check_architectures(["all"])
-    push(ctx, binaries=binaries, architectures=architectures, tag=tag, docker_user=docker_user)
+    push(ctx, binaries=binaries, architectures=architectures, registry=registry, repo=repo, tag=tag)
 
     platforms = ",".join("linux/{}".format(arch) for arch in architectures)
     for bin in binaries:
         run("manifest-tool push from-args "
             "--platforms {platforms} "
-            "--template {user}/{bin}:{tag}-ARCH "
-            "--target {user}/{bin}:{tag}".format(
+            "--template {registry}/{repo}/{bin}:{tag}-ARCH "
+            "--target {registry}/{repo}/{bin}:{tag}".format(
                 platforms=platforms,
-                user=docker_user,
+                registry=registry,
+                repo=repo,
                 bin=bin,
                 tag=tag),
             echo=True)
diff --git a/website/content/release-notes/_index.md b/website/content/release-notes/_index.md
index 0159ef6b..8821a734 100644
--- a/website/content/release-notes/_index.md
+++ b/website/content/release-notes/_index.md
@@ -3,9 +3,30 @@ title: Release Notes
 weight: 8
 ---
 
-## Version 0.9.3
+## Version 0.9.4
 [Documentation for this release](https://metallb.universe.tf)
 
+New features:
+
+- Make Memberlist bind port configurable ([#582](https://github.com/metallb/metallb/pull/582)).
+
+Bugfixes:
+
+- Improve speaker log output ([#587](https://github.com/metallb/metallb/pull/587)).
+- Add "other" exec permission to binaries in Docker images ([#644](https://github.com/metallb/metallb/pull/644)).
+- Fix wrong behavior of the addresses_in_use_total metric under certain conditions ([#627](https://github.com/metallb/metallb/pull/627)).
+- Layer 2: Fix Memberlist convergence following a network partition ([#662](https://github.com/metallb/metallb/pull/662)).
+- Layer 2: Send gratuitous ARP / unsolicited NDP neighbor advertisements following a network partition ([#736](https://github.com/metallb/metallb/pull/736)).
+
+This release includes contributions from Andrew Grosser, Chance Carey, Damien
+TOURDE, Etienne Champetier, Johannes Liebermann, Jumpy Squirrel, Lars Ekman,
+Rodrigo Campos, Russell Bryant, Sebastien Dionne, Steven Follis, sumarsono
+Thorsten Schifferdecker, toby cabot and Yuan Liu. Thanks to all of them for
+making MetalLB better!
+
+## Version 0.9.3
+[Documentation for this release](https://v0-9-3--metallb.netlify.com)
+
 Bugfixes:
 
 - Fix manifests to use container image version `v0.9.3` instead of `main`. Users


Test patch:
diff --git a/internal/layer2/announcer_test.go b/internal/layer2/announcer_test.go
index 20deb0b2..737cc398 100644
--- a/internal/layer2/announcer_test.go
+++ b/internal/layer2/announcer_test.go
@@ -9,6 +9,7 @@ func Test_SetBalancer_AddsToAnnouncedServices(t *testing.T) {
 	announce := &Announce{
 		ips:      map[string]net.IP{},
 		ipRefcnt: map[string]int{},
+		spamCh:   make(chan net.IP, 1),
 	}
 
 	services := []struct {
@@ -27,6 +28,8 @@ func Test_SetBalancer_AddsToAnnouncedServices(t *testing.T) {
 
 	for _, service := range services {
 		announce.SetBalancer(service.name, service.ip)
+		// We need to empty spamCh as spamLoop() is not started.
+		<-announce.spamCh
 
 		if !announce.AnnounceName(service.name) {
 			t.Fatalf("service %v is not anounced", service.name)
diff --git a/speaker/layer2_controller_test.go b/speaker/layer2_controller_test.go
index 384fe8b5..48e3acf5 100644
--- a/speaker/layer2_controller_test.go
+++ b/speaker/layer2_controller_test.go
@@ -13,6 +13,18 @@ import (
 	v1 "k8s.io/api/core/v1"
 )
 
+type fakeSpeakerList struct {
+	speakers map[string]bool
+}
+
+func (sl *fakeSpeakerList) UsableSpeakers() map[string]bool {
+	return sl.speakers
+}
+
+func (sl *fakeSpeakerList) Rejoin() {
+	return
+}
+
 func compareUseableNodesReturnedValue(a, b []string) bool {
 	if &a == &b {
 		return true
@@ -43,6 +55,8 @@ func TestUsableNodes(t *testing.T) {
 
 		eps *v1.Endpoints
 
+		usableSpeakers map[string]bool
+
 		cExpectedResult []string
 	}{
 		{
@@ -63,6 +77,7 @@ func TestUsableNodes(t *testing.T) {
 					},
 				},
 			},
+			usableSpeakers:  map[string]bool{"iris1": true, "iris2": true},
 			cExpectedResult: []string{"iris1", "iris2"},
 		},
 
@@ -84,6 +99,7 @@ func TestUsableNodes(t *testing.T) {
 					},
 				},
 			},
+			usableSpeakers:  map[string]bool{"iris1": true, "iris2": true},
 			cExpectedResult: []string{"iris1"},
 		},
 
@@ -107,12 +123,13 @@ func TestUsableNodes(t *testing.T) {
 					},
 				},
 			},
+			usableSpeakers:  map[string]bool{"iris1": true, "iris2": true},
 			cExpectedResult: []string{"iris1"},
 		},
 	}
 
 	for _, test := range tests {
-		response := usableNodes(test.eps, nil)
+		response := usableNodes(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)
@@ -121,9 +138,16 @@ func TestUsableNodes(t *testing.T) {
 }
 
 func TestShouldAnnounce(t *testing.T) {
+	fakeSL := &fakeSpeakerList{
+		speakers: map[string]bool{
+			"iris1": true,
+			"iris2": true,
+		},
+	}
 	c1, err := newController(controllerConfig{
 		MyNode: "iris1",
 		Logger: log.NewLogfmtLogger(log.NewSyncWriter(os.Stderr)),
+		SList:  fakeSL,
 	})
 	if err != nil {
 		t.Fatalf("creating controller: %s", err)
@@ -133,6 +157,7 @@ func TestShouldAnnounce(t *testing.T) {
 	c2, err := newController(controllerConfig{
 		MyNode: "iris2",
 		Logger: log.NewLogfmtLogger(log.NewSyncWriter(os.Stderr)),
+		SList:  fakeSL,
 	})
 	if err != nil {
 		t.Fatalf("creating controller: %s", err)