gobeetle: update failover sets when config changes
MODIFY
go/config.go
go/config.go
@@ -60,11 +60,21 @@ type FailoverSet struct {
spec string
}
+type FailoverSets []FailoverSet
+
+func (fs *FailoverSets) SystemNames() StringList {
+ res := make(StringList, 0, len(*fs))
+ for _, s := range *fs {
+ res = append(res, s.name)
+ }
+ return res
+}
+
// FailoverSets parses the redis server spec and returns a list of pairs of system
// names and comma separated strings of redis server specs (host:port pairs). Examples:
// "a1:5,a2:5" ==> {"system": "a1:5,a2:5" } "primary/a1:5,a2:5\nsecondary/b1:3,b2:3" ==>
// {"primary": "a1:5,a2:5", "secondary": "b1:3,b2:3"}
-func (c *Config) FailoverSets() []FailoverSet {
+func (c *Config) FailoverSets() FailoverSets {
fs := []FailoverSet{}
for _, line := range strings.Split(c.RedisServers, "\n") {
if line == "" {

MODIFY
go/config_test.go
go/config_test.go
@@ -8,12 +8,12 @@ import (
func TestCanGetFailoverSetsFromConfig(t *testing.T) {
fmt.Println("=== CanGetFailoverSetsFromConfig =================================")
c := Config{RedisServers: "a.xxx.com:6379,b.xxx.com:6379"}
- expected := []FailoverSet{{name: "system", spec: "a.xxx.com:6379,b.xxx.com:6379"}}
+ expected := FailoverSets{{name: "system", spec: "a.xxx.com:6379,b.xxx.com:6379"}}
actual := c.FailoverSets()
checkEqual(t, actual, expected)
c = Config{RedisServers: "s1/a.xxx.com:6379,b.xxx.com:6379\ns2/a.yyy.com:6379,b.yyy.com:6379"}
- expected = []FailoverSet{{name: "s1", spec: "a.xxx.com:6379,b.xxx.com:6379"}, {name: "s2", spec: "a.yyy.com:6379,b.yyy.com:6379"}}
+ expected = FailoverSets{{name: "s1", spec: "a.xxx.com:6379,b.xxx.com:6379"}, {name: "s2", spec: "a.yyy.com:6379,b.yyy.com:6379"}}
actual = c.FailoverSets()
checkEqual(t, actual, expected)
}

MODIFY
go/redis_server_info.go
go/redis_server_info.go
@@ -33,6 +33,7 @@ func (shims RedisShims) Servers() []string {
// RedisServerInfo contans a list slice of RedisShim objects and a lookup index
// on server strings (host:port format).
type RedisServerInfo struct {
+ servers string
instances RedisShims
serverInfo map[string]RedisShims
}
@@ -40,7 +41,7 @@ type RedisServerInfo struct {
// NewRedisServerInfo creates a new RedisServerInfo from a comma separated list
// of servers (host:port format).
func NewRedisServerInfo(servers string) *RedisServerInfo {
- si := &RedisServerInfo{}
+ si := &RedisServerInfo{servers: servers}
si.instances = make(RedisShims, 0)
if servers != "" {
serverList := regexp.MustCompile(" *, *").Split(servers, -1)
@@ -67,7 +68,7 @@ func (si *RedisServerInfo) Reset() {
si.serverInfo = m
}
-// Refresh contacts all redis servers and dtermines their current role.
+// Refresh contacts all redis servers and determines their current role.
func (si *RedisServerInfo) Refresh() {
logDebug("refreshing server info")
si.Reset()
@@ -104,7 +105,7 @@ func (si *RedisServerInfo) Unknowns() RedisShims {
return si.serverInfo[UNKNOWN]
}
-// SlavesOf returns all shims for servers which are salves of the given server.
+// SlavesOf returns all shims for servers which are slaves of the given server.
func (si *RedisServerInfo) SlavesOf(master *RedisShim) RedisShims {
slaves := make(RedisShims, 0)
for _, s := range si.Slaves() {

MODIFY
go/server_state.go
go/server_state.go
@@ -41,7 +41,7 @@ type ServerState struct {
waitGroup sync.WaitGroup // Used to organize the shutdown process.
configChanges chan consul.Env // Environment changes from consul arrive on this channel.
failoverConfidenceLevel float64 // Failover confidence level, normalized to the interval [0,1.0]
- systemNames []string // All system names. Firste on is used for saving server state.
+ systemNames StringList // All system names. Firste on is used for saving server state.
failovers map[string]*FailoverState // Maps system name to failover state.
cmdChannel chan command // Channel for messages to perform state access/changing in the dispatcher thread, passed as closures.
}
@@ -326,9 +326,6 @@ func (s *ServerState) SendNotification(text string) (err error) {
func (s *ServerState) dispatcher() {
ticker := time.NewTicker(1 * time.Second)
- for _, fs := range s.failovers {
- fs.StartWatcher()
- }
for !interrupted {
select {
case cmd := <-s.cmdChannel:
@@ -353,6 +350,7 @@ func (s *ServerState) dispatcher() {
s.SetConfig(newconfig)
s.determineFailoverConfidenceLevel()
s.updateClientIds()
+ s.updateFailoverSets()
logInfo("updated server config from consul: %s", s.GetConfig())
}
}
@@ -412,21 +410,45 @@ func NewServerState(o ServerOptions) *ServerState {
s.unknownClientIds = make(StringList, 0)
s.clientsLastSeen = make(TimeSet)
s.failovers = make(map[string]*FailoverState)
- s.systemNames = make([]string, 0)
- for _, fs := range s.GetConfig().FailoverSets() {
- s.systemNames = append(s.systemNames, fs.name)
- initalToken := int(time.Now().UnixNano() / 1000000) // millisecond resolution
- s.failovers[fs.name] = &FailoverState{
+ s.systemNames = make(StringList, 0)
+ s.updateFailoverSets()
+ return s
+}
+
+func (s *ServerState) updateFailoverSets() {
+ failovers := s.GetConfig().FailoverSets()
+ // delete obsolete failover sets
+ newSystemNames := failovers.SystemNames()
+ knownSystemNames := s.systemNames
+ for _, known := range knownSystemNames {
+ if !newSystemNames.Include(known) {
+ delete(s.failovers, known)
+ }
+ }
+ s.systemNames = newSystemNames
+ // add new failoversets and update already existing ones
+ for _, fs := range failovers {
+ existing := s.failovers[fs.name]
+ if existing != nil {
+ if existing.redis.servers != fs.spec {
+ existing.redis = NewRedisServerInfo(fs.spec)
+ existing.redis.Refresh()
+ }
+ continue
+ }
+ initalTokenInt := int(time.Now().UnixNano() / 1000000) // millisecond resolution
+ newFailoverState := &FailoverState{
server: s,
system: fs.name,
- currentTokenInt: initalToken,
- currentToken: strconv.Itoa(initalToken),
+ currentTokenInt: initalTokenInt,
+ currentToken: strconv.Itoa(initalTokenInt),
redis: NewRedisServerInfo(fs.spec),
clientPongIdsReceived: make(StringSet),
clientInvalidatedIdsReceived: make(StringSet),
}
+ s.failovers[fs.name] = newFailoverState
+ newFailoverState.StartWatcher()
}
- return s
}
// SaveState stores some aspects of the server state to the current redis master

gobeetle: update failover sets when config changes