Embedded Service Discovery mit Memberlist
In verteilten Systemen stehen wir oft vor der Herausforderung, dass sich Services dynamisch finden und über Änderungen im Cluster informiert werden müssen. Ich habe bereits über das HashiCorp Go Package Serf geschrieben.
Memberlist
Klassische Service-Discovery-Lösungen wie Consul, etcd oder Zookeeper sind sehr bewährt, bringen aber erhebliche Komplexität mit sich. Sie müssen in separaten Clustern administriert und betrieben werden und sind gerade für kleinere Projekte oft größer als das eigentliche Projekt.
Memberlist ist ein reines Go-Package, das direkt in die Anwendung eingebettet wird und keinerlei externe Dependencies benötigt. Die Bibliothek implementiert das “SWIM: Scalable Weakly-consistent Infection-style Process Group Membership Protocol” und ermöglicht es Nodes, sich selbst zu organisieren und über Cluster-Änderungen zu informieren.
Memberlist vs. Serf
Während Serf sehr viele Features bietet, die für eine vollständige Service-Discovery-Lösung notwendig sind, konzentriert sich Memberlist auf Cluster-Membership und failure detection.
Serf bietet daneben z.B. auch ein Query System um Nodes zu finden oder kann Events austauschen. Memberlist ist somit die richtige Wahl, wenn man nur wissen möchte, welche Nodes im Cluster sind und über Join/Leave-Events informiert werden will - ohne den Overhead von Serf.
Einfaches Beispiel
Der Einstieg in Memberlist ist sehr einfach. Das folgende Beispiel aus der Memberlist Readme zeigt, wie schnell der Start funktioniert.
Zunächst muss eine Memberlist Instanz erstellt werden.
list, err := memberlist.Create(memberlist.DefaultLocalConfig())
Anschließend muss der Memberlist mindestens ein anderer Node mitgeteilt werden, über den sich die Memberlist mit den restlichen Nodes verbinden kann. Wenn die Liste länger ist, kann das System sich sicherer verbinden, auch wenn einzelne Nodes nicht erreichbar sind.
_, err = list.Join([]string{"1.2.3.4"})
Im Anschluss kann ich die anderen Mitglieder abrufen.
for _, member := range list.Members() {
fmt.Printf("Member: %s %s\n", member.Name, member.Addr)
}
Das ist eigentlich schon alles. In der Configuration kann noch einiges mehr eingestellt werden. Neben Local, LAN oder WAN auch Intervalle oder Empfindlichkeit.
Events
Über die Konfiguration kann ich auch einen Channel angeben, der über Änderungen im Cluster informiert.
memberListEvents := make(chan memberlist.NodeEvent, 100)
memberListConfig := memberlist.DefaultLANConfig()
memberListConfig.Name = "my-node"
memberListConfig.Events = &memberlist.ChannelEventDelegate{
Ch: memberListEvents,
}
Über diesen Channel kann ich im Anschluss Logik implementieren, die auf Join/Leave-Events reagiert. NodeEvent enthält im Feld Event einen EventTyp, der angibt, ob ein Node beigetreten, verlassen oder aktualisiert wurde. Das Feld Node enthält den jeweiligen Node.
go func() {
for {
case event := <-memberListEvents:
switch event.Event {
case memberlist.NodeJoin:
fmt.Printf("Node joined: %s %s\n", event.Node.Name, event.Node.Addr)
handleNewNode(event.Node)
case memberlist.NodeLeave:
fmt.Printf("Node left: %s %s\n", event.Node.Name, event.Node.Addr)
handleNodeLeave(event.Node)
case memberlist.NodeUpdate:
fmt.Printf("Node updated: %s %s\n", event.Node.Name, event.Node.Addr)
}
case <-ctx.Done():
fmt.Println("Stopping memberlist event handler")
return
}
}
}()
Metadaten
Ein großer Vorteil von Memberlist ist die Möglichkeit, strukturierte Metadaten pro Node zu übertragen. In meinem Beispiel möchte ich zu jedem Node den jeweiligen GRPC Endpunkt und den Node Type wissen. So kann ich später über die Metadaten nicht nur wissen, wo der Service läuft, sondern auch, was er anbietet.
type NodeMetadata struct {
Name string `json:"-"`
Type string `json:"type"` // "broker", "worker"
GRPCAddr string `json:"grpc"` // GRPC-Adresse für Service-Calls
}
func (n *NodeMetadata) AsBytes() []byte {
data, err := json.Marshal(n)
if err != nil {
return []byte{}
}
return data
}
Memberlist arbeitet mit Delegates, die es ermöglichen, eigene Logik für Events oder die Metadaten zu implementieren. Es gibt hier noch deutlich mehr Möglichkeiten, aber für mein Beispiel implementiere ich nur das minimum um die Metadaten zu übertragen.
// NodeDelegate implements the memberlist.Delegate interface
type NodeDelegate struct {
Meta []byte
}
// NodeMeta is used to retrieve meta-data about the current node when advertising using memberlist
func (d *NodeDelegate) NodeMeta(limit int) []byte {
if len(d.Meta) > limit {
return d.Meta[:limit]
}
return d.Meta
}
func (d *NodeDelegate) NotifyMsg([]byte) {}
func (d *NodeDelegate) GetBroadcasts(overhead, limit int) [][]byte {
return nil
}
func (d *NodeDelegate) LocalState(join bool) []byte {
return nil
}
func (d *NodeDelegate) MergeRemoteState(buf []byte, join bool) {}
Die Funktion NodeMeta wird aufgerufen, wenn Memberlist die Metadaten des Nodes an andere Nodes überträgt. Anschließend kann ich die Metadaten initialisieren und bei der Konfiguration der Memberlist angeben.
metadataPayload := &NodeMetadata{
Type: "worker",
GRPCAddr: "localhost:50051",
}
memberListConfig := memberlist.DefaultLANConfig()
memberListConfig.Delegate = &NodeDelegate{
Meta: metadataPayload.AsBytes(),
}
In anderen Nodes hängen die Metadaten dann automatisch im Feld Meta an der Node-Instanz.
func handleNewNode(node *memberlist.Node) {
metadata := &NodeMetadata{}
if err := json.Unmarshal(node.Meta, &metadata); err != nil {
fmt.Printf("Error unmarshalling node metadata: %v\n", err)
return
}
fmt.Printf("New node: %s, Type: %s, GRPC Address: %s\n", metadata.Name, metadata.Type, metadata.GRPCAddr)
}
Weitere Möglichkeiten
Memberlist bietet über die Delegate Schnittstellen noch viele weitere Möglichkeiten, weitere Funktionen zu implementieren. So kann ich z.B. auch Broadcast-Nachrichten an andere Nodes senden, die dann über die NotifyMsg-Funktion empfangen werden können. Ich nutze Memberlist, damit Broker alle APIs der Worker Nodes finden, und verwende dort eher die API, um Nachrichten zu verschicken.