fix #407. Start processing continuous queries after initializing the WAL

CHANGELOG.md

@@ -17,6 +17,7 @@
 - [Issue #405](https://github.com/influxdb/influxdb/issues/405). Percentile shouldn't crash for small number of values
 - [Issue #408](https://github.com/influxdb/influxdb/issues/408). Make InfluxDB recover from internal bugs and panics
 - [Issue #390](https://github.com/influxdb/influxdb/issues/390). Multiple response.WriteHeader when querying as admin
+- [Issue #407](https://github.com/influxdb/influxdb/issues/407). Start processing continuous queries only after the WAL is initialized
 - Close leveldb databases properly if we couldn't create a new Shard. See leveldb\_shard\_datastore\_test:131
 
 ## v0.5.4 [2014-04-02]

src/coordinator/raft_server.go

@@ -33,21 +33,22 @@ const (
 // The raftd server is a combination of the Raft server and an HTTP
 // server which acts as the transport.
 type RaftServer struct {
-	name          string
-	host          string
-	port          int
-	path          string
-	bind_address  string
-	router        *mux.Router
-	raftServer    raft.Server
-	httpServer    *http.Server
-	clusterConfig *cluster.ClusterConfiguration
-	mutex         sync.RWMutex
-	listener      net.Listener
-	closing       bool
-	config        *configuration.Configuration
-	notLeader     chan bool
-	coordinator   *CoordinatorImpl
+	name                     string
+	host                     string
+	port                     int
+	path                     string
+	bind_address             string
+	router                   *mux.Router
+	raftServer               raft.Server
+	httpServer               *http.Server
+	clusterConfig            *cluster.ClusterConfiguration
+	mutex                    sync.RWMutex
+	listener                 net.Listener
+	closing                  bool
+	config                   *configuration.Configuration
+	notLeader                chan bool
+	coordinator              *CoordinatorImpl
+	processContinuousQueries bool
 }
 
 var registeredCommands bool
@@ -417,7 +418,15 @@ func (s *RaftServer) raftLeaderLoop(loopTimer *time.Ticker) {
 	}
 }
 
+func (s *RaftServer) StartProcessingContinuousQueries() {
+	s.processContinuousQueries = true
+}
+
 func (s *RaftServer) checkContinuousQueries() {
+	if !s.processContinuousQueries {
+		return
+	}
+
 	if !s.clusterConfig.HasContinuousQueries() {
 		return
 	}

src/server/server.go

@@ -111,6 +111,10 @@ func (self *Server) ListenAndServe() error {
 	}
 	log.Info("Starting Http Api server on port %d", self.Config.ApiHttpPort)
 	self.HttpApi.ListenAndServe()
+
+	// start processing continuous queries
+	self.RaftServer.StartProcessingContinuousQueries()
+
 	return nil
 }