raft.go

package raft

import (
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"time"

	"github.com/ipfs/ipfs-cluster/state"

	host "github.com/libp2p/go-libp2p-core/host"
	peer "github.com/libp2p/go-libp2p-core/peer"
	p2praft "github.com/libp2p/go-libp2p-raft"

	hraft "github.com/hashicorp/raft"
	raftboltdb "github.com/hashicorp/raft-boltdb"
	"go.opencensus.io/trace"
)

// ErrWaitingForSelf is returned when we are waiting for ourselves to depart
// the peer set, which won't happen
var errWaitingForSelf = errors.New("waiting for ourselves to depart")

// RaftMaxSnapshots indicates how many snapshots to keep in the consensus data
// folder.
// TODO: Maybe include this in Config. Not sure how useful it is to touch
// this anyways.
var RaftMaxSnapshots = 5

// RaftLogCacheSize is the maximum number of logs to cache in-memory.
// This is used to reduce disk I/O for the recently committed entries.
var RaftLogCacheSize = 512

// How long we wait for updates during shutdown before snapshotting
var waitForUpdatesShutdownTimeout = 5 * time.Second
var waitForUpdatesInterval = 400 * time.Millisecond

// How many times to retry snapshotting when shutting down
var maxShutdownSnapshotRetries = 5

// raftWrapper wraps the hraft.Raft object and related things like the
// different stores used or the hraft.Configuration.
// Its methods provide functionality for working with Raft.
type raftWrapper struct {
	ctx           context.Context
	cancel        context.CancelFunc
	raft          *hraft.Raft
	config        *Config
	host          host.Host
	serverConfig  hraft.Configuration
	transport     *hraft.NetworkTransport
	snapshotStore hraft.SnapshotStore
	logStore      hraft.LogStore
	stableStore   hraft.StableStore
	boltdb        *raftboltdb.BoltStore
	staging       bool
}

// newRaftWrapper creates a Raft instance and initializes
// everything leaving it ready to use. Note, that Bootstrap() should be called
// to make sure the raft instance is usable.
func newRaftWrapper(
	host host.Host,
	cfg *Config,
	fsm hraft.FSM,
	staging bool,
) (*raftWrapper, error) {

	raftW := &raftWrapper{}
	raftW.config = cfg
	raftW.host = host
	raftW.staging = staging
	// Set correct LocalID
	cfg.RaftConfig.LocalID = hraft.ServerID(peer.Encode(host.ID()))

	df := cfg.GetDataFolder()
	err := makeDataFolder(df)
	if err != nil {
		return nil, err
	}

	raftW.makeServerConfig()

	err = raftW.makeTransport()
	if err != nil {
		return nil, err
	}

	err = raftW.makeStores()
	if err != nil {
		return nil, err
	}

	logger.Debug("creating Raft")
	raftW.raft, err = hraft.NewRaft(
		cfg.RaftConfig,
		fsm,
		raftW.logStore,
		raftW.stableStore,
		raftW.snapshotStore,
		raftW.transport,
	)
	if err != nil {
		logger.Error("initializing raft: ", err)
		return nil, err
	}

	raftW.ctx, raftW.cancel = context.WithCancel(context.Background())
	go raftW.observePeers()

	return raftW, nil
}

// makeDataFolder creates the folder that is meant to store Raft data. Ensures
// we always set 0700 mode.
func makeDataFolder(folder string) error {
	return os.MkdirAll(folder, 0700)
}

func (rw *raftWrapper) makeTransport() (err error) {
	logger.Debug("creating libp2p Raft transport")
	rw.transport, err = p2praft.NewLibp2pTransport(
		rw.host,
		rw.config.NetworkTimeout,
	)
	return err
}

func (rw *raftWrapper) makeStores() error {
	logger.Debug("creating BoltDB store")
	df := rw.config.GetDataFolder()
	store, err := raftboltdb.NewBoltStore(filepath.Join(df, "raft.db"))
	if err != nil {
		return err
	}

	// wraps the store in a LogCache to improve performance.
	// See consul/agent/consul/server.go
	cacheStore, err := hraft.NewLogCache(RaftLogCacheSize, store)
	if err != nil {
		return err
	}

	logger.Debug("creating raft snapshot store")
	snapstore, err := hraft.NewFileSnapshotStoreWithLogger(
		df,
		RaftMaxSnapshots,
		raftStdLogger,
	)
	if err != nil {
		return err
	}

	rw.logStore = cacheStore
	rw.stableStore = store
	rw.snapshotStore = snapstore
	rw.boltdb = store
	return nil
}

// Bootstrap calls BootstrapCluster on the Raft instance with a valid
// Configuration (generated from InitPeerset) when Raft has no state
// and we are not setting up a staging peer. It returns if Raft
// was boostrapped (true) and an error.
func (rw *raftWrapper) Bootstrap() (bool, error) {
	logger.Debug("checking for existing raft states")
	hasState, err := hraft.HasExistingState(
		rw.logStore,
		rw.stableStore,
		rw.snapshotStore,
	)
	if err != nil {
		return false, err
	}

	if hasState {
		logger.Debug("raft cluster is already initialized")

		// Inform the user that we are working with a pre-existing peerset
		logger.Info("existing Raft state found! raft.InitPeerset will be ignored")
		cf := rw.raft.GetConfiguration()
		if err := cf.Error(); err != nil {
			logger.Debug(err)
			return false, err
		}
		currentCfg := cf.Configuration()
		srvs := ""
		for _, s := range currentCfg.Servers {
			srvs += fmt.Sprintf("        %s\n", s.ID)
		}

		logger.Debugf("Current Raft Peerset:\n%s\n", srvs)
		return false, nil
	}

	if rw.staging {
		logger.Debug("staging servers do not need initialization")
		logger.Info("peer is ready to join a cluster")
		return false, nil
	}

	voters := ""
	for _, s := range rw.serverConfig.Servers {
		voters += fmt.Sprintf("        %s\n", s.ID)
	}

	logger.Infof("initializing raft cluster with the following voters:\n%s\n", voters)

	future := rw.raft.BootstrapCluster(rw.serverConfig)
	if err := future.Error(); err != nil {
		logger.Error("bootstrapping cluster: ", err)
		return true, err
	}
	return true, nil
}

// create Raft servers configuration. The result is used
// by Bootstrap() when it proceeds to Bootstrap.
func (rw *raftWrapper) makeServerConfig() {
	rw.serverConfig = makeServerConf(append(rw.config.InitPeerset, rw.host.ID()))
}

// creates a server configuration with all peers as Voters.
func makeServerConf(peers []peer.ID) hraft.Configuration {
	sm := make(map[string]struct{})

	servers := make([]hraft.Server, 0)

	// Servers are peers + self. We avoid duplicate entries below
	for _, pid := range peers {
		p := peer.Encode(pid)
		_, ok := sm[p]
		if !ok { // avoid dups
			sm[p] = struct{}{}
			servers = append(servers, hraft.Server{
				Suffrage: hraft.Voter,
				ID:       hraft.ServerID(p),
				Address:  hraft.ServerAddress(p),
			})
		}
	}
	return hraft.Configuration{Servers: servers}
}

// WaitForLeader holds until Raft says we have a leader.
// Returns if ctx is cancelled.
func (rw *raftWrapper) WaitForLeader(ctx context.Context) (string, error) {
	ctx, span := trace.StartSpan(ctx, "consensus/raft/WaitForLeader")
	defer span.End()

	ticker := time.NewTicker(time.Second / 2)
	for {
		select {
		case <-ticker.C:
			if l := rw.raft.Leader(); l != "" {
				logger.Debug("waitForleaderTimer")
				logger.Infof("Current Raft Leader: %s", l)
				ticker.Stop()
				return string(l), nil
			}
		case <-ctx.Done():
			return "", ctx.Err()
		}
	}
}

func (rw *raftWrapper) WaitForVoter(ctx context.Context) error {
	ctx, span := trace.StartSpan(ctx, "consensus/raft/WaitForVoter")
	defer span.End()

	logger.Debug("waiting until we are promoted to a voter")

	pid := hraft.ServerID(peer.Encode(rw.host.ID()))
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
			logger.Debugf("%s: get configuration", pid)
			configFuture := rw.raft.GetConfiguration()
			if err := configFuture.Error(); err != nil {
				return err
			}

			if isVoter(pid, configFuture.Configuration()) {
				return nil
			}
			logger.Debugf("%s: not voter yet", pid)

			time.Sleep(waitForUpdatesInterval)
		}
	}
}

func isVoter(srvID hraft.ServerID, cfg hraft.Configuration) bool {
	for _, server := range cfg.Servers {
		if server.ID == srvID && server.Suffrage == hraft.Voter {
			return true
		}
	}
	return false
}

// WaitForUpdates holds until Raft has synced to the last index in the log
func (rw *raftWrapper) WaitForUpdates(ctx context.Context) error {
	ctx, span := trace.StartSpan(ctx, "consensus/raft/WaitForUpdates")
	defer span.End()

	logger.Debug("Raft state is catching up to the latest known version. Please wait...")
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
			lai := rw.raft.AppliedIndex()
			li := rw.raft.LastIndex()
			logger.Debugf("current Raft index: %d/%d",
				lai, li)
			if lai == li {
				return nil
			}
			time.Sleep(waitForUpdatesInterval)
		}
	}
}

func (rw *raftWrapper) WaitForPeer(ctx context.Context, pid string, depart bool) error {
	ctx, span := trace.StartSpan(ctx, "consensus/raft/WaitForPeer")
	defer span.End()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
			peers, err := rw.Peers(ctx)
			if err != nil {
				return err
			}

			if len(peers) == 1 && pid == peers[0] && depart {
				return errWaitingForSelf
			}

			found := find(peers, pid)

			// departing
			if depart && !found {
				return nil
			}

			// joining
			if !depart && found {
				return nil
			}

			time.Sleep(50 * time.Millisecond)
		}
	}
}

// Snapshot tells Raft to take a snapshot.
func (rw *raftWrapper) Snapshot() error {
	future := rw.raft.Snapshot()
	err := future.Error()
	if err != nil && err.Error() != hraft.ErrNothingNewToSnapshot.Error() {
		return err
	}
	return nil
}

// snapshotOnShutdown attempts to take a snapshot before a shutdown.
// Snapshotting might fail if the raft applied index is not the last index.
// This waits for the updates and tries to take a snapshot when the
// applied index is up to date.
// It will retry if the snapshot still fails, in case more updates have arrived.
// If waiting for updates times-out, it will not try anymore, since something
// is wrong. This is a best-effort solution as there is no way to tell Raft
// to stop processing entries because we want to take a snapshot before
// shutting down.
func (rw *raftWrapper) snapshotOnShutdown() error {
	var err error
	for i := 0; i < maxShutdownSnapshotRetries; i++ {
		ctx, cancel := context.WithTimeout(context.Background(), waitForUpdatesShutdownTimeout)
		err = rw.WaitForUpdates(ctx)
		cancel()
		if err != nil {
			logger.Warn("timed out waiting for state updates before shutdown. Snapshotting may fail")
			return rw.Snapshot()
		}

		err = rw.Snapshot()
		if err == nil {
			return nil // things worked
		}

		// There was an error
		err = errors.New("could not snapshot raft: " + err.Error())
		logger.Warnf("retrying to snapshot (%d/%d)...", i+1, maxShutdownSnapshotRetries)
	}
	return err
}

// Shutdown shutdown Raft and closes the BoltDB.
func (rw *raftWrapper) Shutdown(ctx context.Context) error {
	_, span := trace.StartSpan(ctx, "consensus/raft/Shutdown")
	defer span.End()

	errMsgs := ""

	rw.cancel()

	err := rw.snapshotOnShutdown()
	if err != nil {
		errMsgs += err.Error() + ".\n"
	}

	future := rw.raft.Shutdown()
	err = future.Error()
	if err != nil {
		errMsgs += "could not shutdown raft: " + err.Error() + ".\n"
	}

	err = rw.boltdb.Close() // important!
	if err != nil {
		errMsgs += "could not close boltdb: " + err.Error()
	}

	if errMsgs != "" {
		return errors.New(errMsgs)
	}

	return nil
}

// AddPeer adds a peer to Raft
func (rw *raftWrapper) AddPeer(ctx context.Context, peer string) error {
	ctx, span := trace.StartSpan(ctx, "consensus/raft/AddPeer")
	defer span.End()

	// Check that we don't have it to not waste
	// log entries if so.
	peers, err := rw.Peers(ctx)
	if err != nil {
		return err
	}
	if find(peers, peer) {
		logger.Infof("%s is already a raft peer", peer)
		return nil
	}

	future := rw.raft.AddVoter(
		hraft.ServerID(peer),
		hraft.ServerAddress(peer),
		0,
		0,
	) // TODO: Extra cfg value?
	err = future.Error()
	if err != nil {
		logger.Error("raft cannot add peer: ", err)
	}
	return err
}

// RemovePeer removes a peer from Raft
func (rw *raftWrapper) RemovePeer(ctx context.Context, peer string) error {
	ctx, span := trace.StartSpan(ctx, "consensus/RemovePeer")
	defer span.End()

	// Check that we have it to not waste
	// log entries if we don't.
	peers, err := rw.Peers(ctx)
	if err != nil {
		return err
	}
	if !find(peers, peer) {
		logger.Infof("%s is not among raft peers", peer)
		return nil
	}

	if len(peers) == 1 && peers[0] == peer {
		return errors.New("cannot remove ourselves from a 1-peer cluster")
	}

	rmFuture := rw.raft.RemoveServer(
		hraft.ServerID(peer),
		0,
		0,
	) // TODO: Extra cfg value?
	err = rmFuture.Error()
	if err != nil {
		logger.Error("raft cannot remove peer: ", err)
		return err
	}

	return nil
}

// Leader returns Raft's leader. It may be an empty string if
// there is no leader or it is unknown.
func (rw *raftWrapper) Leader(ctx context.Context) string {
	_, span := trace.StartSpan(ctx, "consensus/raft/Leader")
	defer span.End()

	return string(rw.raft.Leader())
}

func (rw *raftWrapper) Peers(ctx context.Context) ([]string, error) {
	_, span := trace.StartSpan(ctx, "consensus/raft/Peers")
	defer span.End()

	ids := make([]string, 0)

	configFuture := rw.raft.GetConfiguration()
	if err := configFuture.Error(); err != nil {
		return nil, err
	}

	for _, server := range configFuture.Configuration().Servers {
		ids = append(ids, string(server.ID))
	}

	return ids, nil
}

// latestSnapshot looks for the most recent raft snapshot stored at the
// provided basedir.  It returns the snapshot's metadata, and a reader
// to the snapshot's bytes
func latestSnapshot(raftDataFolder string) (*hraft.SnapshotMeta, io.ReadCloser, error) {
	store, err := hraft.NewFileSnapshotStore(raftDataFolder, RaftMaxSnapshots, nil)
	if err != nil {
		return nil, nil, err
	}
	snapMetas, err := store.List()
	if err != nil {
		return nil, nil, err
	}
	if len(snapMetas) == 0 { // no error if snapshot isn't found
		return nil, nil, nil
	}
	meta, r, err := store.Open(snapMetas[0].ID)
	if err != nil {
		return nil, nil, err
	}
	return meta, r, nil
}

// LastStateRaw returns the bytes of the last snapshot stored, its metadata,
// and a flag indicating whether any snapshot was found.
func LastStateRaw(cfg *Config) (io.Reader, bool, error) {
	// Read most recent snapshot
	dataFolder := cfg.GetDataFolder()
	if _, err := os.Stat(dataFolder); os.IsNotExist(err) {
		// nothing to read
		return nil, false, nil
	}

	meta, r, err := latestSnapshot(dataFolder)
	if err != nil {
		return nil, false, err
	}
	if meta == nil { // no snapshots could be read
		return nil, false, nil
	}
	return r, true, nil
}

// SnapshotSave saves the provided state to a snapshot in the
// raft data path.  Old raft data is backed up and replaced
// by the new snapshot.  pids contains the config-specified
// peer ids to include in the snapshot metadata if no snapshot exists
// from which to copy the raft metadata
func SnapshotSave(cfg *Config, newState state.State, pids []peer.ID) error {
	dataFolder := cfg.GetDataFolder()
	err := makeDataFolder(dataFolder)
	if err != nil {
		return err
	}
	meta, _, err := latestSnapshot(dataFolder)
	if err != nil {
		return err
	}

	// make a new raft snapshot
	var raftSnapVersion hraft.SnapshotVersion = 1 // As of hraft v1.0.0 this is always 1
	configIndex := uint64(1)
	var raftIndex uint64
	var raftTerm uint64
	var srvCfg hraft.Configuration
	if meta != nil {
		raftIndex = meta.Index
		raftTerm = meta.Term
		srvCfg = meta.Configuration
		CleanupRaft(cfg)
	} else {
		// Begin the log after the index of a fresh start so that
		// the snapshot's state propagate's during bootstrap
		raftIndex = uint64(2)
		raftTerm = uint64(1)
		srvCfg = makeServerConf(pids)
	}

	snapshotStore, err := hraft.NewFileSnapshotStoreWithLogger(dataFolder, RaftMaxSnapshots, nil)
	if err != nil {
		return err
	}
	_, dummyTransport := hraft.NewInmemTransport("")

	sink, err := snapshotStore.Create(raftSnapVersion, raftIndex, raftTerm, srvCfg, configIndex, dummyTransport)
	if err != nil {
		return err
	}

	err = p2praft.EncodeSnapshot(newState, sink)
	if err != nil {
		sink.Cancel()
		return err
	}
	err = sink.Close()
	if err != nil {
		return err
	}
	return nil
}

// CleanupRaft moves the current data folder to a backup location
func CleanupRaft(cfg *Config) error {
	dataFolder := cfg.GetDataFolder()
	keep := cfg.BackupsRotate

	meta, _, err := latestSnapshot(dataFolder)
	if meta == nil && err == nil {
		// no snapshots at all. Avoid creating backups
		// from empty state folders.
		logger.Infof("cleaning empty Raft data folder (%s)", dataFolder)
		os.RemoveAll(dataFolder)
		return nil
	}

	logger.Infof("cleaning and backing up Raft data folder (%s)", dataFolder)
	dbh := newDataBackupHelper(dataFolder, keep)
	err = dbh.makeBackup()
	if err != nil {
		logger.Warn(err)
		logger.Warn("the state could not be cleaned properly")
		logger.Warn("manual intervention may be needed before starting cluster again")
	}
	return nil
}

// only call when Raft is shutdown
func (rw *raftWrapper) Clean() error {
	return CleanupRaft(rw.config)
}

func find(s []string, elem string) bool {
	for _, selem := range s {
		if selem == elem {
			return true
		}
	}
	return false
}

func (rw *raftWrapper) observePeers() {
	obsCh := make(chan hraft.Observation, 1)
	defer close(obsCh)

	observer := hraft.NewObserver(obsCh, true, func(o *hraft.Observation) bool {
		po, ok := o.Data.(hraft.PeerObservation)
		return ok && po.Removed
	})

	rw.raft.RegisterObserver(observer)
	defer rw.raft.DeregisterObserver(observer)

	for {
		select {
		case obs := <-obsCh:
			pObs := obs.Data.(hraft.PeerObservation)
			logger.Info("raft peer departed. Removing from peerstore: ", pObs.Peer.ID)
			pID, err := peer.Decode(string(pObs.Peer.ID))
			if err != nil {
				logger.Error(err)
				continue
			}
			rw.host.Peerstore().ClearAddrs(pID)
		case <-rw.ctx.Done():
			logger.Debug("stopped observing raft peers")
			return
		}
	}
}