Start cluster before broker.

CHANGELOG.md

@@ -3,6 +3,7 @@
 ### Bugfixes
 - [#2282](https://github.com/influxdb/influxdb/pull/2282): Use "value" as field name for OpenTSDB input.
 - [#2283](https://github.com/influxdb/influxdb/pull/2283): Fix bug when restarting an entire existing cluster.
+- [#2293](https://github.com/influxdb/influxdb/pull/2293): Open cluster listener before starting broker.
 - [#2287](https://github.com/influxdb/influxdb/pull/2287): Fix data race during SHOW RETENTION POLICIES.
 
 ## Features

cmd/influxd/run.go

@@ -254,27 +254,21 @@ func (cmd *RunCommand) Open(config *Config, join string) *Node {
 	// Parse join urls from the --join flag.
 	joinURLs := parseURLs(join)
 
+	// Start the broker handler.
+	h := &Handler{Config: config}
+	if err := cmd.node.openClusterListener(cmd.config.ClusterAddr(), h); err != nil {
+		log.Fatalf("Cluster server failed to listen on %s. %s ", cmd.config.ClusterAddr(), err)
+	}
+	log.Printf("Cluster server listening on %s", cmd.config.ClusterAddr())
+
 	// Open broker & raft log, initialize or join as necessary.
 	if cmd.config.Broker.Enabled {
-		cmd.openBroker(joinURLs)
+		cmd.openBroker(joinURLs, h)
 		// If were running as a broker locally, always connect to it since it must
 		// be ready before we can start the data node.
 		joinURLs = []url.URL{cmd.node.Broker.URL()}
 	}
 
-	// Start the broker handler.
-	h := &Handler{
-		Config: config,
-		Broker: cmd.node.Broker,
-		Log:    cmd.node.raftLog,
-	}
-
-	err := cmd.node.openClusterListener(cmd.config.ClusterAddr(), h)
-	if err != nil {
-		log.Fatalf("Cluster server failed to listen on %s. %s ", cmd.config.ClusterAddr(), err)
-	}
-	log.Printf("Cluster server listening on %s", cmd.config.ClusterAddr())
-
 	var s *influxdb.Server
 	// Open server, initialize or join as necessary.
 	if cmd.config.Data.Enabled {
@@ -469,7 +463,7 @@ func writePIDFile(path string) {
 }
 
 // creates and initializes a broker.
-func (cmd *RunCommand) openBroker(brokerURLs []url.URL) {
+func (cmd *RunCommand) openBroker(brokerURLs []url.URL, h *Handler) {
 	path := cmd.config.BrokerDir()
 	u := cmd.config.ClusterURL()
 	raftTracing := cmd.config.Logging.RaftTracing
@@ -502,6 +496,10 @@ func (cmd *RunCommand) openBroker(brokerURLs []url.URL) {
 		log.Fatalf("raft: %s", err)
 	}
 
+	// Attach broker and log to handler.
+	h.Broker = b
+	h.Log = l
+
 	// Checks to see if the raft index is 0.  If it's 0, it might be the first
 	// node in the cluster and must initialize or join
 	index, _ := l.LastLogIndexTerm()

raft/internal_test.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"io/ioutil"
 	"os"
+	"sync"
 	"testing"
 )
 
@@ -44,25 +45,36 @@ func CloseLog(l *Log) {
 
 // IndexFSM represents a state machine that only records the last applied index.
 type IndexFSM struct {
+	mu    sync.Mutex
 	index uint64
 }
 
 // MustApply updates the index.
 func (fsm *IndexFSM) Apply(entry *LogEntry) error {
+	fsm.mu.Lock()
 	fsm.index = entry.Index
+	fsm.mu.Unlock()
 	return nil
 }
 
 // Index returns the highest applied index.
-func (fsm *IndexFSM) Index() uint64 { return fsm.index }
+func (fsm *IndexFSM) Index() uint64 {
+	fsm.mu.Lock()
+	defer fsm.mu.Unlock()
+	return fsm.index
+}
 
 // WriteTo writes a snapshot of the FSM to w.
 func (fsm *IndexFSM) WriteTo(w io.Writer) (n int64, err error) {
+	fsm.mu.Lock()
+	defer fsm.mu.Unlock()
 	return 0, binary.Write(w, binary.BigEndian, fsm.index)
 }
 
 // ReadFrom reads an FSM snapshot from r.
 func (fsm *IndexFSM) ReadFrom(r io.Reader) (n int64, err error) {
+	fsm.mu.Lock()
+	defer fsm.mu.Unlock()
 	return 0, binary.Read(r, binary.BigEndian, &fsm.index)
 }
 

raft/log.go

@@ -886,8 +886,11 @@ func (l *Log) followerLoop(closing <-chan struct{}) State {
 			close(ch)
 
 			// Ignore timeout if we are snapshotting.
+			// Or if we haven't received confirmation of join.
 			if l.isSnapshotting() {
 				continue
+			} else if l.FSM.Index() == 0 {
+				continue
 			}
 
 			// TODO: Prevote before becoming candidate.
@@ -1349,6 +1352,7 @@ func (l *Log) appendToWriters(buf []byte) {
 
 		// If an error occurs then remove the writer and close it.
 		if _, err := w.Write(buf); err != nil {
+			l.Logger.Printf("append to writers error: %s", err)
 			l.removeWriter(w)
 			i--
 			continue
@@ -1797,6 +1801,7 @@ func (l *Log) removeWriter(writer *logWriter) {
 			l.writers[len(l.writers)-1] = nil
 			l.writers = l.writers[:len(l.writers)-1]
 			_ = w.Close()
+			l.Logger.Printf("writer removed: %#v", w)
 			break
 		}
 	}