Embedded Server-Discovery mit Gossip
Um in Cloud Umgebungen dynamisch Anwendungen skalieren zu können oder ohne starre Konfiguration starten zu können, müssen sich die einzelnen Komponenten untereinander finden können. Hier kommen gerne Service-Discovery Lösungen wie Consul oder etcd zum Einsatz. Neben dem Verwalten von dynamischen Adresslisten können die Lösungen auch den Zustand überwachen und nicht verfügbare Dienste aussortieren.
Die fertigen Produkte sind millionenfach im Einsatz, sehr gut getestet, OpenSource und Kostenlos. Der große Nachteil, gerade für kleiner Projekte, besteht in dem zusätzlichen Aufwand für Betrieb und Pflege. Sie laufen meistens im eigenen Cluster und bringen so eine eigene Komplexität mit.
In Go kann hier eine embeded Server-Discovery helfen. Die von Consul intern verwendete Lösung Serf kann als Module leicht integriert werden. Die sehr leichtgewichtige Implementation benötigt fast keine Verstrickungen mit dem eigentlichen Service und kann später schnell durch externe oder eigene Implementationen ersetzt werden.
HashiCorp nutzt Serf u.a. in Consul oder Vault.
Serf kann auch als CLI Tool verwendet werden und neben Anwendungen laufen.
Protokoll link
Jeder Service hat einen Agent der das Protokoll handhabt. Dieser Agent hat eine Liste an Key/Value Tags. Mögliche ist z.B. die Region oder der eigentliche Service Port.
Gerne wird das Gossip Protokoll mit einer Virus Infektion verglichen. In regelmäßigen Abständen wählt jeder Agent bekannte andere Agents aus und tauscht sich mit diesen aus. Im Anschluss haben beide Agents die Tags und Mitglieder des jeweiligen anderen Agents. Falls Nachrichten oder Peers verloren gehen, schließt die Lücke der nächste Austausch mit einem anderen Node.
Serf Node link
Jeder Serf Node benötigt einen eindeutigen Namen zur Identifikation und eine Adresse auf der Serf horcht.
Über Key/Value Tags können Metainformationen über den Node verteilt werden. Beispiele wären der Service Port oder Service Type.
Für denn Start benötigt Serf eine Liste von Nodes wo eine Infektion mit dem Cluster wahrscheinlich stattfinden kann.
type Config struct {
NodeName string
BindAddr string
Tags map[string]string
StartJoinAddrs []string
}
Über einen Event Channel lässt sich Serf leicht in Projekte integrieren. Dazu später mehr.
type Node struct {
config *Config
server *serf.Serf
events chan serf.Event
handler MembershipHandler
}
Wie viele HashiCorp Module bringt auch Serf ein eigenes Config Format mit DefaultConfig mit.
func NewNode(config *Config) (*Node, error) {
addr, err := net.ResolveTCPAddr("tcp", config.BindAddr)
if err != nil {
return nil, err
}
node := &Node{
config: config,
}
serfConfig := serf.DefaultConfig()
serfConfig.Init()
serfConfig.MemberlistConfig.BindAddr = addr.IP.String()
serfConfig.MemberlistConfig.BindPort = addr.Port
node.events = make(chan serf.Event)
serfConfig.EventCh = node.events
serfConfig.Tags = node.config.Tags
serfConfig.NodeName = node.config.NodeName
node.server, err = serf.Create(serfConfig)
if err != nil {
return nil, err
}
return node, nil
}
Handle Events link
Im Event befindet sich eine Liste an geänderten Cluster Mitgliedern. Alle externen Änderungen können z.B. einfach an die eigentliche Anwendung weiter gegeben werden.
type MembershipHandler interface {
MemberJoin(nodeName string, nodeTags map[string]string)
MemberLeave(nodeName string, nodeTags map[string]string)
}
func (n *Node) eventHandler() {
for e := range n.events {
switch e.EventType() {
case serf.EventMemberJoin:
for _, member := range e.(serf.MemberEvent).Members {
if n.server.LocalMember().Name == member.Name {
continue
}
n.handler.MemberJoin(member.Name, member.Tags)
}
case serf.EventMemberLeave, serf.EventMemberFailed:
for _, member := range e.(serf.MemberEvent).Members {
if n.server.LocalMember().Name == member.Name {
continue
}
n.handler.MemberLeave(member.Name, member.Tags)
}
}
}
}
Join Cluster link
Der Übersichtlichkeit halber, habe ich den Start des SerfNodes in eine eigene Funktion ausgelagert. Der asynchrone Event Handler wird gestartet und dem Server eine Adressen zum verbinden gegeben. Ein Fehler wird nur zurück gegeben, wenn kein anderer Node gefunden wurde. Ist dieser Node der erste, ist dies kein Fehler und der Node sollte einfach warten!
func (n *Node) Start() error {
go n.eventHandler()
if n.config.StartJoinAddrs != nil {
_, err := n.server.Join(n.config.StartJoinAddrs, true)
return err
}
return nil
}
Cluster Management link
Für ein einfaches Arbeiten mit dem Cluster noch zwei hilfreiche Helfer Funktionen.
func (n *Node) Leave() error {
return n.server.Leave()
}
func (n *Node) Members() []serf.Member {
return n.server.Members()
}