Influxdata
/
influxdb
mirror of https://github.com/influxdata/influxdb.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

merge InfluxJoinCommand with AddPotentialServerCommand

pull/482/head
John Shahid 2014-04-28 18:26:32 -04:00
parent 876b05956f
commit 4866197788
3 changed files with 87 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
src/cluster/cluster_configuration.go
View File

@ -55,34 +55,35 @@ const (
the servers in the cluster and their state, databases, users, and which continuous queries are running. the servers in the cluster and their state, databases, users, and which continuous queries are running.
*/ */
type ClusterConfiguration struct { type ClusterConfiguration struct {
createDatabaseLock sync.RWMutex createDatabaseLock sync.RWMutex
DatabaseReplicationFactors map[string]uint8 DatabaseReplicationFactors map[string]uint8
usersLock sync.RWMutex usersLock sync.RWMutex
clusterAdmins map[string]*ClusterAdmin clusterAdmins map[string]*ClusterAdmin
dbUsers map[string]map[string]*DbUser dbUsers map[string]map[string]*DbUser
servers []*ClusterServer servers []*ClusterServer
serversLock sync.RWMutex serversLock sync.RWMutex
continuousQueries map[string][]*ContinuousQuery continuousQueries map[string][]*ContinuousQuery
continuousQueriesLock sync.RWMutex continuousQueriesLock sync.RWMutex
ParsedContinuousQueries map[string]map[uint32]*parser.SelectQuery ParsedContinuousQueries map[string]map[uint32]*parser.SelectQuery
continuousQueryTimestamp time.Time continuousQueryTimestamp time.Time
LocalServerId uint32 LocalServerId uint32
config *configuration.Configuration LocalProtobufConnectionString string
addedLocalServerWait chan bool config *configuration.Configuration
addedLocalServer bool addedLocalServerWait chan bool
connectionCreator func(string) ServerConnection addedLocalServer bool
shardStore LocalShardStore connectionCreator func(string) ServerConnection
wal WAL shardStore LocalShardStore
longTermShards []*ShardData wal WAL
shortTermShards []*ShardData longTermShards []*ShardData
random *rand.Rand shortTermShards []*ShardData
lastServerToGetShard *ClusterServer random *rand.Rand
shardCreator ShardCreator lastServerToGetShard *ClusterServer
shardLock sync.Mutex shardCreator ShardCreator
shardsById map[uint32]*ShardData shardLock sync.Mutex
shardsByIdLock sync.RWMutex shardsById map[uint32]*ShardData
LocalRaftName string shardsByIdLock sync.RWMutex
writeBuffers []*WriteBuffer LocalRaftName string
writeBuffers []*WriteBuffer
} }
type ContinuousQuery struct { type ContinuousQuery struct {
@ -217,22 +218,33 @@ func (self *ClusterConfiguration) AddPotentialServer(server *ClusterServer) {
server.Id = uint32(len(self.servers)) server.Id = uint32(len(self.servers))
log.Info("Added server to cluster config: %d, %s, %s", server.Id, server.RaftConnectionString, server.ProtobufConnectionString) log.Info("Added server to cluster config: %d, %s, %s", server.Id, server.RaftConnectionString, server.ProtobufConnectionString)
log.Info("Checking whether this is the local server new: %s, local: %s\n", self.config.ProtobufConnectionString(), server.ProtobufConnectionString) log.Info("Checking whether this is the local server new: %s, local: %s\n", self.config.ProtobufConnectionString(), server.ProtobufConnectionString)
if server.RaftName != self.LocalRaftName {
log.Info("Connecting to ProtobufServer: %s", server.ProtobufConnectionString, self.config.ProtobufConnectionString()) if server.RaftName == self.LocalRaftName && self.addedLocalServer {
if server.connection == nil { panic("how did we add the same server twice ?")
server.connection = self.connectionCreator(server.ProtobufConnectionString) }
server.Connect()
} // if this is the local server unblock WaitForLocalServerLoaded()
writeBuffer := NewWriteBuffer(fmt.Sprintf("%d", server.GetId()), server, self.wal, server.Id, self.config.PerServerWriteBufferSize) // and set the local connection string and id
self.writeBuffers = append(self.writeBuffers, writeBuffer) if server.RaftName == self.LocalRaftName {
server.SetWriteBuffer(writeBuffer)
server.StartHeartbeat()
} else if !self.addedLocalServer {
log.Info("Added the local server") log.Info("Added the local server")
self.LocalServerId = server.Id self.LocalServerId = server.Id
self.LocalProtobufConnectionString = server.ProtobufConnectionString
self.addedLocalServerWait <- true self.addedLocalServerWait <- true
self.addedLocalServer = true self.addedLocalServer = true
return
} }
// if this isn't the local server, connect to it
log.Info("Connecting to ProtobufServer: %s", server.ProtobufConnectionString, self.config.ProtobufConnectionString())
if server.connection == nil {
server.connection = self.connectionCreator(server.ProtobufConnectionString)
server.Connect()
}
writeBuffer := NewWriteBuffer(fmt.Sprintf("%d", server.GetId()), server, self.wal, server.Id, self.config.PerServerWriteBufferSize)
self.writeBuffers = append(self.writeBuffers, writeBuffer)
server.SetWriteBuffer(writeBuffer)
server.StartHeartbeat()
return
} }
func (self *ClusterConfiguration) GetDatabases() []*Database { func (self *ClusterConfiguration) GetDatabases() []*Database {
@ -344,6 +356,10 @@ func (self *ClusterConfiguration) GetContinuousQueries(db string) []*ContinuousQ
return self.continuousQueries[db] return self.continuousQueries[db]
} }
func (self *ClusterConfiguration) GetLocalConfiguration() *configuration.Configuration {
return self.config
}
func (self *ClusterConfiguration) GetDbUsers(db string) []common.User { func (self *ClusterConfiguration) GetDbUsers(db string) []common.User {
self.usersLock.RLock() self.usersLock.RLock()
defer self.usersLock.RUnlock() defer self.usersLock.RUnlock()

43
src/coordinator/command.go
View File

@ -3,6 +3,7 @@ package coordinator
import ( import (
"cluster" "cluster"
"encoding/json" "encoding/json"
"fmt"
"io" "io"
"time" "time"
@ -15,7 +16,7 @@ var internalRaftCommands map[string]raft.Command
func init() { func init() {
internalRaftCommands = map[string]raft.Command{} internalRaftCommands = map[string]raft.Command{}
for _, command := range []raft.Command{ for _, command := range []raft.Command{
&AddPotentialServerCommand{}, &InfluxJoinCommand{},
&CreateDatabaseCommand{}, &CreateDatabaseCommand{},
&DropDatabaseCommand{}, &DropDatabaseCommand{},
&SaveDbUserCommand{}, &SaveDbUserCommand{},
@ -216,24 +217,6 @@ func (c *SaveClusterAdminCommand) Apply(server raft.Server) (interface{}, error)
return nil, nil return nil, nil
} }
type AddPotentialServerCommand struct {
Server *cluster.ClusterServer
}
func NewAddPotentialServerCommand(s *cluster.ClusterServer) *AddPotentialServerCommand {
return &AddPotentialServerCommand{Server: s}
}
func (c *AddPotentialServerCommand) CommandName() string {
return "add_server"
}
func (c *AddPotentialServerCommand) Apply(server raft.Server) (interface{}, error) {
config := server.Context().(*cluster.ClusterConfiguration)
config.AddPotentialServer(c.Server)
return nil, nil
}
type InfluxJoinCommand struct { type InfluxJoinCommand struct {
Name string `json:"name"` Name string `json:"name"`
ConnectionString string `json:"connectionString"` ConnectionString string `json:"connectionString"`
@ -242,13 +225,31 @@ type InfluxJoinCommand struct {
// The name of the Join command in the log // The name of the Join command in the log
func (c *InfluxJoinCommand) CommandName() string { func (c *InfluxJoinCommand) CommandName() string {
return "raft:join" return "join"
} }
func (c *InfluxJoinCommand) Apply(server raft.Server) (interface{}, error) { func (c *InfluxJoinCommand) Apply(server raft.Server) (interface{}, error) {
err := server.AddPeer(c.Name, c.ConnectionString) err := server.AddPeer(c.Name, c.ConnectionString)
if err != nil {
return nil, err
}
return []byte("join"), err clusterConfig := server.Context().(*cluster.ClusterConfiguration)
newServer := clusterConfig.GetServerByRaftName(c.Name)
// it's a new server the cluster has never seen, make it a potential
if newServer != nil {
return nil, fmt.Errorf("Server %s already exist", c.Name)
}
log.Info("Adding new server to the cluster config %s", c.Name)
clusterServer := cluster.NewClusterServer(c.Name,
c.ConnectionString,
c.ProtobufConnectionString,
nil,
clusterConfig.GetLocalConfiguration())
clusterConfig.AddPotentialServer(clusterServer)
return nil, nil
} }
func (c *InfluxJoinCommand) NodeName() string { func (c *InfluxJoinCommand) NodeName() string {

48
src/coordinator/raft_server.go
View File

@ -380,18 +380,6 @@ func (s *RaftServer) startRaft() error {
if err != nil { if err != nil {
log.Error(err) log.Error(err)
} }
protobufConnectString := s.config.ProtobufConnectionString()
clusterServer := cluster.NewClusterServer(name,
connectionString,
protobufConnectString,
nil,
s.config)
command := NewAddPotentialServerCommand(clusterServer)
_, err = s.doOrProxyCommand(command)
if err != nil {
return err
}
err = s.CreateRootUser() err = s.CreateRootUser()
return err return err
} }
@ -608,6 +596,7 @@ func (s *RaftServer) retryCommand(command raft.Command, retries int) (ret interf
} }
func (s *RaftServer) joinHandler(w http.ResponseWriter, req *http.Request) { func (s *RaftServer) joinHandler(w http.ResponseWriter, req *http.Request) {
// if this is the leader, process the command
if s.raftServer.State() == raft.Leader { if s.raftServer.State() == raft.Leader {
command := &InfluxJoinCommand{} command := &InfluxJoinCommand{}
if err := json.NewDecoder(req.Body).Decode(&command); err != nil { if err := json.NewDecoder(req.Body).Decode(&command); err != nil {
@ -619,34 +608,17 @@ func (s *RaftServer) joinHandler(w http.ResponseWriter, req *http.Request) {
if _, err := s.raftServer.Do(command); err != nil { if _, err := s.raftServer.Do(command); err != nil {
log.Error("Can't process %v: %s", command, err) log.Error("Can't process %v: %s", command, err)
http.Error(w, err.Error(), http.StatusInternalServerError) http.Error(w, err.Error(), http.StatusInternalServerError)
return
} }
server := s.clusterConfig.GetServerByRaftName(command.Name) return
// it's a new server the cluster has never seen, make it a potential }
if server == nil {
log.Info("Adding new server to the cluster config %s", command.Name) leader, ok := s.leaderConnectString()
clusterServer := cluster.NewClusterServer(command.Name, log.Debug("Non-leader redirecting to: (%v, %v)", leader, ok)
command.ConnectionString, if ok {
command.ProtobufConnectionString, log.Debug("redirecting to leader to join...")
nil, http.Redirect(w, req, leader+"/join", http.StatusTemporaryRedirect)
s.config)
addServer := NewAddPotentialServerCommand(clusterServer)
if _, err := s.raftServer.Do(addServer); err != nil {
log.Error("Error joining raft server: ", err, command)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
}
log.Info("Server %s already exist in the cluster config", command.Name)
} else { } else {
leader, ok := s.leaderConnectString() http.Error(w, errors.New("Couldn't find leader of the cluster to join").Error(), http.StatusInternalServerError)
log.Debug("Non-leader redirecting to: (%v, %v)", leader, ok)
if ok {
log.Debug("redirecting to leader to join...")
http.Redirect(w, req, leader+"/join", http.StatusTemporaryRedirect)
} else {
http.Error(w, errors.New("Couldn't find leader of the cluster to join").Error(), http.StatusInternalServerError)
}
} }
} }