Repository: metallb/metallb
Pull request #662 (2020-07-09 16:53:30)

layer2: Make nodes re-join memberlist cluster
This PR makes the nodes join the memberlist cluster when they left for some reason (network outage, etc.).

It is based on the ideas from the RFC PR #647 but storing the Speaker IPs to also be able to recover when the API server is down (before we relied on the API server to gives us the APIs, so if it was down we couldn't join members), based on @champtar suggestion on that PR.

It consists of two commits: the first one is cherry-picked from #595 and just moves the memberlist code from `speaker/main.go` to it's own package, so it is easier to reason and abstract. The second one just adds a go routine that will try to re-join the nodes whenever it is needed.

The semantic changes in the second commit aim to be simple to follow and reason about, without any major rework. More reworks can be done in the future, in another minor release, taking into accounts the ideas from #595. This is aimed at v0.9.4, therefore the change tries to be as simple and correct as possible, avoiding major semantic changes.

I've tested this locally with kind (with `inv dev-env`) and blocking traffic with iptables. It worked well and as expected on all situations (API server down and recovers, one other node down and recovers, etc.).

Let me know what you think. 

TODO:
- [ ] Add tests. Memberlist is quite untested currently.

I'll c&p the commit message here from the rejoin commit, as it is the core of this PR. To properly understand what I refer to here, please look just at this commit and not the whole diff of the PR:

Currently, nodes join the memberlist cluster only on startup and if they are removed (e.g. small network downtime and they recover) they never join again. This means the node is alive, but not taken into account as a possible speaker, making it pointless to run the MetalLB speaker.

This commit adds a go routine to do memberlist joins in a loop. This was missing in the commit that introduced memberlist (3c5fc63 "Use hashicorp/memberlist to speedup dead node detection") and fixes issue #589.

This routine tries to join nodes in two situations: (a) every 1 minute, (b) when the current k8s node object changes and the SetNode() callback is called.

The case (b) is used when the node loses connectivity for long enough for Kubernetes to mark it in a Not Ready state. When the node is Ready again (e.g network is restored for the node), the callback is invoked and the node tries to join the cluster.

The case (a), periodic checks, is needed for downtimes that are smaller than the time Kubernetes takes to mark a node as Not Ready (about 45 sec by default). For example, if a node has a small downtime (~15 seconds) memberlist will remove it from the cluster but as this is short enough for Kubernetes to not change the node state to Not Ready, the SetNode() callback won't be invoked (as the node didn't change). In these cases, the periodic join every 1 minute will make those node join the
memberlist cluster again.

The internal kubernetes watcher already has this logic to notify when the current node changes. In the Layer2 case, though, it was just a "return nil". Now, it queues the message to process a re-join.

This routine tries to join only the nodes that are not currently part of the cluster, as it was advised by Etienne Champetier that the memberlist Join() operation is not a no-op for nodes that are already members and can take a considerable amount of time (specially when only a small group of nodes need to re-join the cluster).

Furthermore, the IPs to join are stored in the SpeakerList struct to be able to recover nodes even if the API server is down. This means we don't need to query the API server and just use the IPs cached in the struct. This was also very good advise from Etienne.

There is another go routine introduced here to update the IPs to join stored in the struct. This go routine updates them by quering the API server, every 5 minutes only to not overload the API server (something very easy to do by mistake). This is in a separate go routine because, as I mention in the code, the current version of client-go we are using doesn't support a context or a timeout so that query can block. To avoid this blocking other operations, we use a separate go routine. When we update to new client-go versions, this will of course not be needed and we can just run with a timeout if we want to.

champtar: @johananl that is why I wanted to review each commit separately, I can't address any of your reviews on my commit
@rata I'll let you fix my commit
rata: @champtar 
>  This is worse than that, this is a split brain and you might have multiple nodes responding to ARP requests causing the tcp connection to switch nodes and get tcp reset

Yes, true. Didn't go into the details as I linked the issue. Tried to explain but the commit message was already big and explaining that in more detail was complicated while being clear, so I just explained that bit and let the link to the issue do the rest :)
champtar: > @champtar
> 
> > This is worse than that, this is a split brain and you might have multiple nodes responding to ARP requests causing the tcp connection to switch nodes and get tcp reset
> 
> Yes, true. Didn't go into the details as I linked the issue. Tried to explain but the commit message was already big and explaining that in more detail was complicated while being clear, so I just explained that bit and let the link to the issue do the rest :)

No need to go into detail, but I read your message as 1 node is unusable, when we should read the full service might be unusable, kind of day and night
rata: @liuyuan10 thanks a lot for the review! :)
rata: @johananl @champtar PTAL.

Just fixed most of the comments. There are a few open that need more discussion :)
champtar: I'm fine with having code review, but any plan to ship something at some point ?
- we have bug reports that might be fixed by this PR
- a good part of this code is changed/going away in my other commits in #595 
rata: > I'm fine with having code review, but any plan to ship something at some point ?

Yes, of course :). I'd like to release in the next few weeks. I'll try to address @johananl review and merge when it is ready.
rata: @johananl thanks for the review. PTAL

I think I addressed all the comments. Thanks a lot for the suggestions, I applied them all and amended them to the proper commit.
rata: @johananl PTAL. I think I address your last remaining comments.

FYI, I amended your github suggestions to the respective commits that introduced them. That is why you won't see a commit doing them, but they are addressed :)
johananl: >FYI, I amended your github suggestions to the respective commits that introduced them. That is why you won't see a commit doing them, but they are addressed :)

That's fine, I never actually use the "commit" UI option. I use the suggestions feature just for the diff display.

Patch:
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/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/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()
+}


Test patch:
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)