<commit_before>package raftwrapper

import (
	"encoding/json"
	"errors"
	"fmt"
	"golang.org/x/net/context"
	"log"
	"strconv"
	"sync"
	"time"

	"github.com/cenk/backoff"

	"github.com/coreos/etcd/etcdserver/stats"
	"github.com/coreos/etcd/pkg/types"
	"github.com/coreos/etcd/raft"
	"github.com/coreos/etcd/raft/raftpb"
	"github.com/coreos/etcd/rafthttp"
	"github.com/coreos/etcd/snap"
	"github.com/coreos/etcd/wal"
)

type LogData []byte

// because WAL and Snap look to see if ANY files exist in the dir
// for confirmation. Meaning that if one or the other is enabled
// but not the other, then checks will fail
var walDirExtension = "/wal"
var snapDirExtension = "/snap"

type Node struct {
	node           raft.Node
	raftStorage    *raft.MemoryStorage
	transport      *rafthttp.Transport
	bootstrapPeers []string
	bootstrapNode  bool
	peerMap        map[uint64]confChangeNodeContext
	id             uint64
	cid            uint64
	raftPort       int

	apiPort int

	raftConfig *raft.Config

	started     bool
	initialized bool
	running     bool

	proposeC chan string
	fsm      FSM

	observers     map[uint64]*Observer
	observersLock sync.RWMutex

	initBackoffArgs *InitializationBackoffArgs
	snapshotConfig  *SnapshotConfig

	dataDir string
	ss      *snap.Snapshotter
	wal     *wal.WAL

	lastConfState *raftpb.ConfState

	stopc chan struct{}
}

type NodeConfig struct {
	// If not specified or 0, will autogenerate a new UUID
	ID uint64

	// If not specified 0x100 will be used
	ClusterID uint64

	FSM            FSM
	RaftPort       int
	APIPort        int
	BootstrapPeers []string
	BootstrapNode  bool

	// DataDir is where your data will be persisted to disk
	// for use when either you need to restart a node, or
	// it goes offline and needs to be restarted
	DataDir string

	InitBackoff *InitializationBackoffArgs
	// if nil, then default to no snapshotting
	SnapshotConfig *SnapshotConfig
}

type SnapshotConfig struct {

	// How often do you want to Snapshot and compact logs?
	Interval time.Duration

	// If the interval ticks but not enough logs have been commited then ignore
	// the snapshot this interval
	MinCommittedLogs uint64

	// If the interval hasn't ticked but we've gone over a commited log threshold then snapshot
	// Note: Use this with care. Snapshotting is a fairly expenseive process.
	// Interval is suggested best method for triggering snapshots
	MaxCommittedLogs uint64
}

var DefaultSnapshotConfig = &SnapshotConfig{
	Interval:         -1 * time.Second,
	MinCommittedLogs: 0,
	MaxCommittedLogs: 0,
}

type InitializationBackoffArgs struct {
	InitialInterval     time.Duration
	Multiplier          float64
	MaxInterval         time.Duration
	MaxElapsedTime      time.Duration
	RandomizationFactor float64
}

var DefaultInitializationBackoffArgs = &InitializationBackoffArgs{
	InitialInterval:     500 * time.Millisecond,
	RandomizationFactor: .5,
	Multiplier:          2,
	MaxInterval:         5 * time.Second,
	MaxElapsedTime:      2 * time.Minute,
}

// note: peers is only for asking to join the cluster.
// It will not be able to connect if the peers don't respond to cluster node add request
// This is because each node defines it's own uuid at startup. We must be told this UUID
// by another node.
// TODO: Look into which config options we want others to specify. For now hardcoded
// TODO: Allow user to specify KV pairs of known nodes, and bypass the http discovery
// NOTE: Peers are used EXCLUSIVELY to round-robin to other nodes and attempt to add
//		ourselves to an existing cluster or bootstrap node
func NewNode(args *NodeConfig) (*Node, error) {
	rn, err := nonInitNode(args)
	if err != nil {
		return nil, err
	}

	return rn, nil
}

func (rn *Node) shouldRejoinCluster() bool {
	return wal.Exist(rn.walDir()) && rn.walDir() != ""
}

func (rn *Node) advanceTicksForElection() error {
	for i := 0; i < rn.raftConfig.ElectionTick-1; i++ {
		rn.node.Tick()
	}
	return nil
}

// TODO: Don't panic on these. Return an err if they happen
func (rn *Node) Start() error {
	walEnabled := rn.walDir() != ""
	rejoinCluster := rn.shouldRejoinCluster()
	if rn.started {
		return nil
	}
	rn.stopc = make(chan struct{})

	if walEnabled {
		if err := rn.restoreRaft(); err != nil {
			return err
		}
	}

	if rejoinCluster {
		rn.node = raft.RestartNode(rn.raftConfig)
	} else {
		if rn.bootstrapNode {
			rn.node = raft.StartNode(rn.raftConfig, []raft.Peer{raft.Peer{ID: rn.id}})
		} else {
			rn.node = raft.StartNode(rn.raftConfig, nil)
		}
	}
	if err := rn.advanceTicksForElection(); err != nil {
		return err
	}

	if err := rn.attachTransport(); err != nil {
		return err
	}

	if err := rn.transport.Start(); err != nil {
		return err
	}

	go rn.scanReady()

	// Start config http service
	go func(rn *Node) {
		if err := rn.serveHTTP(); err != nil {
			panic(err)
		}
	}(rn)

	// start raft
	go func(rn *Node) {
		if err := rn.serveRaft(); err != nil {
			panic(err)
		}
	}(rn)
	rn.started = true

	if rejoinCluster {
		if err := rn.selfRejoinCluster(); err != nil {
			return err
		}
	} else if !rn.bootstrapNode {
		if err := rn.addSelfToCluster(); err != nil {
			return err
		}
	}
	// final step to mark node as initialized
	rn.running = true
	return nil
}<commit_msg>Fix issue with rejoining a cluster<commit_after>package raftwrapper

import (
	"encoding/json"
	"errors"
	"fmt"
	"golang.org/x/net/context"
	"log"
	"strconv"
	"sync"
	"time"

	"github.com/cenk/backoff"

	"github.com/coreos/etcd/etcdserver/stats"
	"github.com/coreos/etcd/pkg/types"
	"github.com/coreos/etcd/raft"
	"github.com/coreos/etcd/raft/raftpb"
	"github.com/coreos/etcd/rafthttp"
	"github.com/coreos/etcd/snap"
	"github.com/coreos/etcd/wal"
)

type LogData []byte

// because WAL and Snap look to see if ANY files exist in the dir
// for confirmation. Meaning that if one or the other is enabled
// but not the other, then checks will fail
var walDirExtension = "/wal"
var snapDirExtension = "/snap"

type Node struct {
	node           raft.Node
	raftStorage    *raft.MemoryStorage
	transport      *rafthttp.Transport
	bootstrapPeers []string
	bootstrapNode  bool
	peerMap        map[uint64]confChangeNodeContext
	id             uint64
	cid            uint64
	raftPort       int

	apiPort int

	raftConfig *raft.Config

	started     bool
	initialized bool
	running     bool

	proposeC chan string
	fsm      FSM

	observers     map[uint64]*Observer
	observersLock sync.RWMutex

	initBackoffArgs *InitializationBackoffArgs
	snapshotConfig  *SnapshotConfig

	dataDir string
	ss      *snap.Snapshotter
	wal     *wal.WAL

	lastConfState *raftpb.ConfState

	stopc chan struct{}
}

type NodeConfig struct {
	// If not specified or 0, will autogenerate a new UUID
	ID uint64

	// If not specified 0x100 will be used
	ClusterID uint64

	FSM            FSM
	RaftPort       int
	APIPort        int
	BootstrapPeers []string
	BootstrapNode  bool

	// DataDir is where your data will be persisted to disk
	// for use when either you need to restart a node, or
	// it goes offline and needs to be restarted
	DataDir string

	InitBackoff *InitializationBackoffArgs
	// if nil, then default to no snapshotting
	SnapshotConfig *SnapshotConfig
}

type SnapshotConfig struct {

	// How often do you want to Snapshot and compact logs?
	Interval time.Duration

	// If the interval ticks but not enough logs have been commited then ignore
	// the snapshot this interval
	MinCommittedLogs uint64

	// If the interval hasn't ticked but we've gone over a commited log threshold then snapshot
	// Note: Use this with care. Snapshotting is a fairly expenseive process.
	// Interval is suggested best method for triggering snapshots
	MaxCommittedLogs uint64
}

var DefaultSnapshotConfig = &SnapshotConfig{
	Interval:         -1 * time.Second,
	MinCommittedLogs: 0,
	MaxCommittedLogs: 0,
}

type InitializationBackoffArgs struct {
	InitialInterval     time.Duration
	Multiplier          float64
	MaxInterval         time.Duration
	MaxElapsedTime      time.Duration
	RandomizationFactor float64
}

var DefaultInitializationBackoffArgs = &InitializationBackoffArgs{
	InitialInterval:     500 * time.Millisecond,
	RandomizationFactor: .5,
	Multiplier:          2,
	MaxInterval:         5 * time.Second,
	MaxElapsedTime:      2 * time.Minute,
}

// note: peers is only for asking to join the cluster.
// It will not be able to connect if the peers don't respond to cluster node add request
// This is because each node defines it's own uuid at startup. We must be told this UUID
// by another node.
// TODO: Look into which config options we want others to specify. For now hardcoded
// TODO: Allow user to specify KV pairs of known nodes, and bypass the http discovery
// NOTE: Peers are used EXCLUSIVELY to round-robin to other nodes and attempt to add
//		ourselves to an existing cluster or bootstrap node
func NewNode(args *NodeConfig) (*Node, error) {
	rn, err := nonInitNode(args)
	if err != nil {
		return nil, err
	}

	return rn, nil
}

func (rn *Node) shouldRejoinCluster() bool {
	return wal.Exist(rn.walDir()) && rn.walDir() != ""
}

func (rn *Node) advanceTicksForElection() error {
	for i := 0; i < rn.raftConfig.ElectionTick-1; i++ {
		rn.node.Tick()
	}
	return nil
}

func (rn *Node) Start() error {
	walEnabled := rn.walDir() != ""
	rejoinCluster := rn.shouldRejoinCluster()
	if rn.started {
		return nil
	}
	rn.stopc = make(chan struct{})

	if walEnabled {
		if err := rn.restoreRaft(); err != nil {
			return err
		}
	}

	if rejoinCluster {
		rn.node = raft.RestartNode(rn.raftConfig)
	} else {
		if rn.bootstrapNode {
			rn.node = raft.StartNode(rn.raftConfig, []raft.Peer{raft.Peer{ID: rn.id}})
		} else {
			rn.node = raft.StartNode(rn.raftConfig, nil)
		}
	}
	if err := rn.advanceTicksForElection(); err != nil {
		return err
	}

	if err := rn.attachTransport(); err != nil {
		return err
	}

	if err := rn.transport.Start(); err != nil {
		return err
	}
	rn.initialized = true

	go func(rn *Node) {
		if err := rn.scanReady(); err != nil {
			panic(err)
		}
	}(rn)

	// Start config http service
	go func(rn *Node) {
		if err := rn.serveHTTP(); err != nil {
			panic(err)
		}
	}(rn)

	// start raft
	go func(rn *Node) {
		if err := rn.serveRaft(); err != nil {
			panic(err)
		}
	}(rn)
	rn.started = true

	if rejoinCluster {
		if err := rn.selfRejoinCluster(); err != nil {
			return err
		}
	} else if !rn.bootstrapNode {
		if err := rn.addSelfToCluster(); err != nil {
			return err
		}
	}
	// final step to mark node as initialized
	rn.running = true
	return nil
}
{'fraction_non_alphanumeric': 0.09459717449408171, 'fraction_numerical': 0.007063764795723558, 'mean_word_length': 3.1924769907963184, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '542413', 'n_tokens_mistral': 3658, 'n_tokens_neox': 3447, 'n_words': 1144}