Cluster Discovery with Serf — Gossip-Based Membership and Failure Detection

Cluster Discovery with Serf — Gossip-Based Membership and Failure Detection

Nodes in a distributed log must discover each other without a static configuration file. In this section, you'll implement cluster membership using HashiCorp Serf — a gossip-based discovery system. When a node joins, the cluster learns its RPC and Raft addresses. When a node fails, the cluster detects it and reacts. All code lives in broker/cluster/discovery/discovery.go.

Step 1: Understand why Serf

Benefits of gossip-based discovery

You could hardcode peer addresses, but that breaks when nodes are added, removed, or moved. Serf solves this:

  • No central coordinator — no single point of failure for discovery.
  • Gossip protocol — each node periodically tells a few random peers about the cluster state. Information spreads exponentially.
  • Failure detection — Serf probes members and marks unresponsive nodes as failed.
  • Custom tags — each node attaches metadata (RPC address, Raft address) that is spread with membership.

Step 2: Define the handler interface

Handler interface for membership changes

Define what the rest of the system needs when membership changes:

// broker/cluster/discovery/discovery.go
package discovery

type Handler interface {
    Join(id, raftAddr, rpcAddr string) error
    Leave(name string) error
}

When Serf detects a new member, Join is called with the node's ID, Raft address, and RPC address. broker/cluster.Cluster implements this interface — Join adds the node to the Raft cluster's voter configuration, and Leave removes it.

Note the parameter order: raftAddr comes before rpcAddr. This matches how Cluster uses them — Raft is the primary cluster communication channel.

Step 3: Create the Membership struct

Membership struct overview

The Membership struct embeds config.Config directly, giving it access to all configuration fields without extra plumbing:

type Membership struct {
    config.Config
    handler Handler
    serf    *serf.Serf
    events  chan serf.Event
    logger  *zap.Logger
}

By embedding Config, any code with a *Membership can access m.BindAddr, m.NodeConfig, m.RaftConfig, etc. directly.

Step 4: Create the constructor

Creating a Membership instance

The constructor is named New (not NewMembership) — following Go convention since the package is discovery:

func New(handler Handler, config config.Config) (*Membership, error) {
    c := &Membership{
        Config:  config,
        handler: handler,
        logger:  zap.L().Named("discovery"),
    }
    if err := c.setupSerf(); err != nil {
        return nil, err
    }
    return c, nil
}

handler is the *cluster.Cluster instance that implements the Handler interface. config comes from the server startup code and contains bind addresses, node identity, and peer addresses.

Step 5: Set up Serf

Serf initialization and configuration

setupSerf configures the Serf agent, starts the event handler goroutine, and optionally joins existing peers:

func (m *Membership) setupSerf() (err error) {
    addr, err := net.ResolveTCPAddr("tcp", m.BindAddr)
    if err != nil {
        return err
    }
    config := serf.DefaultConfig()
    config.Init()
    config.MemberlistConfig.BindAddr = addr.IP.String()
    config.MemberlistConfig.BindPort = addr.Port
    config.MemberlistConfig.LogOutput = io.Discard
    m.events = make(chan serf.Event)
    config.EventCh = m.events
    rpcAddr, err := m.Config.RPCAddr()
    if err != nil {
        return err
    }
    config.Tags = map[string]string{
        "rpc_addr":  rpcAddr,
        "raft_addr": m.RaftConfig.Address,
    }
    config.NodeName = m.NodeConfig.ID
    m.serf, err = serf.Create(config)
    if err != nil {
        return err
    }
    go m.eventHandler()
    if m.StartJoinAddrs != nil {
        _, err = m.serf.Join(m.StartJoinAddrs, true)
        if err != nil {
            m.logger.Warn("serf join failed", zap.Error(err), zap.Strings("addrs", m.StartJoinAddrs))
        }
    }
    return nil
}

Walk through step by step:

1. Resolve the bind address. m.BindAddr is a string like "127.0.0.1:7946". Split it into IP and port for Serf's memberlist configuration.

2. Create default config and initialize. serf.DefaultConfig() returns sensible defaults for gossip intervals, probe timeouts, etc. config.Init() initializes internal fields.

3. Suppress memberlist logging. Serf's memberlist is chatty. Routing its log output to io.Discard keeps the application's own logs clean — events are handled explicitly instead.

4. Create the event channel. The channel is unbuffered (make(chan serf.Event)). This means Serf blocks until the event handler reads each event, which is fine because events are processed quickly.

5. Compute the RPC address. m.Config.RPCAddr() builds the address from the advertise address (or bind address) plus the RPC port:

// broker/config/config.go
func (c Config) RPCAddr() (string, error) {
    if c.AdvertiseAddr != "" {
        return fmt.Sprintf("%s:%d", c.AdvertiseAddr, c.NodeConfig.RPCPort), nil
    }
    host, _, err := net.SplitHostPort(c.BindAddr)
    if err != nil {
        return "", err
    }
    return fmt.Sprintf("%s:%d", host, c.NodeConfig.RPCPort), nil
}

If an AdvertiseAddr is set (common in Docker, where a node binds to 0.0.0.0 but advertises its container hostname), it uses that. Otherwise it extracts the host from BindAddr.

6. Set tags. Tags are key-value metadata that Serf gossips to all members. Two addresses are stored:

  • rpc_addr — where clients and the replication engine connect (e.g., node1:9091)
  • raft_addr — where Raft communicates (e.g., node1:9093)

7. Create Serf and start the event handler. serf.Create() starts the Serf agent. The event handler goroutine begins listening immediately.

8. Join existing peers. If StartJoinAddrs is set (from the --peer flag), join those addresses. The second argument true tells Serf to replay member events for nodes already in the cluster. A failed join is not fatal — this node might be the first one starting.

Note the ordering: the event handler starts before the join, so no events from the join process itself are missed.

Key configuration summary:

Field Purpose
NodeName Unique identifier for this node (e.g., node1). Must be stable across restarts.
BindAddr / BindPort Address for gossip traffic. All nodes must be reachable on this port.
Tags Key-value metadata: rpc_addr and raft_addr.
EventCh Channel where Serf pushes membership events.
LogOutput Set to io.Discard to suppress memberlist's internal logging.

Step 6: Implement the event handler

Event handler goroutine

The event handler goroutine reads from the event channel and dispatches join and leave events:

func (m *Membership) eventHandler() {
    for e := range m.events {
        switch e.EventType() {
        case serf.EventMemberJoin:
            for _, member := range e.(serf.MemberEvent).Members {
                if m.isLocal(member) {
                    continue
                }
                m.handleJoin(member)
            }
        case serf.EventMemberLeave, serf.EventMemberFailed:
            for _, member := range e.(serf.MemberEvent).Members {
                if m.isLocal(member) {
                    continue
                }
                m.handleLeave(member)
            }
        }
    }
}

A few things to notice:

  • Switch on EventType(), not a type assertion. Serf events have different types — only member events matter here.
  • Type-assert to serf.MemberEvent to access the Members slice. Each event can contain multiple members (e.g., if several nodes joined simultaneously).
  • Skip the local node. A node shouldn't add itself to its own Raft cluster or try to replicate to itself.
  • Both EventMemberLeave and EventMemberFailed trigger the same handleLeave path. A graceful departure and a crash are handled the same way — remove the node as a Raft voter.

Handling join events

Extract the RPC and Raft addresses from the member's tags and call the handler. There is fallback logic: if one tag is missing, use the other:

func (m *Membership) handleJoin(member serf.Member) {
    raftAddr := member.Tags["raft_addr"]
    rpcAddr := member.Tags["rpc_addr"]
    if raftAddr == "" {
        raftAddr = rpcAddr
    }
    if rpcAddr == "" {
        rpcAddr = raftAddr
    }
    err := m.handler.Join(member.Name, raftAddr, rpcAddr)
    if err != nil {
        m.logError(err, "failed to join", member)
    }
}

The fallback (if raftAddr == "" { raftAddr = rpcAddr }) is a safety net. In practice both tags are always set, but having one address is better than having none if a node is misconfigured.

Handling leave and failure events

func (m *Membership) handleLeave(member serf.Member) {
    if err := m.handler.Leave(member.Name); err != nil {
        m.logError(err, "failed to leave", member)
    }
}

Leave only needs the node ID — Cluster.Leave uses it to remove the node from Raft's voter configuration.

Error logging with level awareness

Not every error deserves the same log level. If a non-leader node receives a join/leave event and tries to modify the Raft cluster, Raft returns raft.ErrNotLeader. This is expected — only the Raft leader can add or remove voters. The helper logs these at Info instead of Error:

func (m *Membership) logError(err error, msg string, member serf.Member) {
    log := m.logger.Error
    if err == raft.ErrNotLeader {
        log = m.logger.Info
    }
    log(
        msg,
        zap.Error(err),
        zap.String("name", member.Name),
        zap.String("rpc_addr", member.Tags["rpc_addr"]),
    )
}

This matters because in a 3-node cluster, every node's Serf agent fires join/leave events, but only the Raft leader can act on them. Without this helper, non-leader nodes would flood the logs with errors that aren't actually problems.

Checking if a member is local

func (m *Membership) isLocal(member serf.Member) bool {
    return m.serf.LocalMember().Name == member.Name
}

Step 7: Query the member list

Members, AliveMembers, and AliveNodeDetails

Other parts of the system need to query the current membership — not just "who joined" but "who is alive right now, and at what address":

func (m *Membership) Members() []serf.Member {
    return m.serf.Members()
}

// AliveMembers returns the names of Serf members currently in the Alive state.
func (m *Membership) AliveMembers() []string {
    var names []string
    for _, member := range m.serf.Members() {
        if member.Status == serf.StatusAlive {
            names = append(names, member.Name)
        }
    }
    return names
}

// NodeInfo holds identity and address information for a cluster member.
type NodeInfo struct {
    Name     string
    RaftAddr string
    RpcAddr  string
}

// AliveNodeDetails returns identity and address info for Serf members currently alive.
func (m *Membership) AliveNodeDetails() []NodeInfo {
    var nodes []NodeInfo
    for _, member := range m.serf.Members() {
        if member.Status == serf.StatusAlive {
            nodes = append(nodes, NodeInfo{
                Name:     member.Name,
                RaftAddr: member.Tags["raft_addr"],
                RpcAddr:  member.Tags["rpc_addr"],
            })
        }
    }
    return nodes
}

Members() returns all members, including those in StatusLeft or StatusFailed state — callers that only want live nodes use AliveMembers/AliveNodeDetails instead. These two methods are exactly the shape of the cluster.MemberLister interface from the previous chapter: *Membership is handed to Cluster.SetMemberLister right after both are constructed, which is how Cluster.AliveNodeIDs() and Cluster.NodeRPCAddr() get their data without broker/cluster importing anything about Serf's own types.

Step 8: Implement graceful shutdown

Leaving the cluster cleanly

When the node shuts down, leave the cluster gracefully so other nodes are notified immediately (instead of waiting for Serf's failure detection timeout, which can take several seconds):

func (m *Membership) Leave() error {
    return m.serf.Leave()
}

Step 9: Understand the integration flow

How Serf integrates with the rest of the system

Node starts
    │
    ├── Create Serf with tags (rpc_addr, raft_addr)
    │
    ├── Start event handler goroutine
    │
    ├── Join existing peers (if any)
    │
    └── Event handler processes events
            │
            ├── Member joins → handler.Join(id, raftAddr, rpcAddr)
            │                    └── Cluster adds node as a Raft voter
            │
            └── Member leaves/fails → handler.Leave(id)
                                       └── Cluster removes node as a Raft voter
                                       └── watchPeerChanges reacts once the removal commits

Startup sequence:

  1. The first node starts with no peers. Serf creates a one-node cluster.
  2. The second node starts with --start-join-addr node1:gossip_port. Serf gossips the join.
  3. Node 1's event handler fires Join(node2, raftAddr, rpcAddr). Cluster (if the Raft leader) adds node 2 as a voter.
  4. The third node joins similarly. Now all three nodes know about each other.

Failure handling:

  1. Node 2 crashes. Serf detects it via failed probes (within seconds).
  2. Node 1 and node 3's event handlers fire Leave(node2).
  3. Cluster removes node 2 as a Raft voter; the resulting configuration change is what triggers leader reassignment for any topic node 2 was leading (covered in the topic management chapter).

Summary

Component Purpose
Serf Gossip-based cluster membership. No central coordinator.
Tags Per-node metadata (rpc_addr, raft_addr) spread with membership.
Event handler Goroutine that processes join/leave/fail events and calls the Handler interface.
Handler interface Join(id, raftAddr, rpcAddr) and Leave(name) — implemented by broker/cluster.Cluster.
Members / AliveMembers / AliveNodeDetails Query current membership; the latter two double as the cluster.MemberLister interface.
logError Downgrades raft.ErrNotLeader to Info — only the Raft leader can modify cluster membership.

With discovery in place, the next page integrates Raft consensus — so the cluster can agree on metadata like topic assignments and leader elections.