support switching with only a subset of clients responding
MODIFY
features/redis_auto_failover.feature
features/redis_auto_failover.feature
@@ -42,6 +42,22 @@ Feature: Redis auto failover
And the retry timeout for the redis master check is reached
Then the role of redis server "redis-2" should be "slave"
+ Scenario: Not all redis configuration clients available (successful switch using confidence level 50)
+ Given a redis configuration server using redis servers "redis-1,redis-2" with clients "rc-client-1,rc-client-2" and confidence level "50" exists
+ And a redis configuration client "rc-client-1" using redis servers "redis-1,redis-2" exists
+ And a beetle handler using the redis-master file from "rc-client-1" exists
+ And redis configuration client "rc-client-2" is down
+ And redis server "redis-1" is down
+ And the retry timeout for the redis master check is reached
+ Then a system notification for "redis-1" not being available should be sent
+ And the role of redis server "redis-2" should be "master"
+ And the redis master file of the redis configuration server should contain "redis-2"
+ And the redis master of "rc-client-1" should be "redis-2"
+ And the redis master of the beetle handler should be "redis-2"
+ And a system notification for switching from "redis-1" to "redis-2" should be sent
+ Given a redis server "redis-1" exists as master
+ Then the role of redis server "redis-1" should be "slave"
+
Scenario: No redis slave available to become new master (no switch possible)
Given a redis configuration server using redis servers "redis-1,redis-2" with clients "rc-client-1,rc-client-2" exists
And a redis configuration client "rc-client-1" using redis servers "redis-1,redis-2" exists

MODIFY
features/step_definitions/redis_auto_failover_steps.rb
features/step_definitions/redis_auto_failover_steps.rb
@@ -21,9 +21,9 @@ Given /^redis server "([^\"]*)" is slave of "([^\"]*)"$/ do |redis_name, redis_m
end while !slave.slave_of?(master.host, master.port)
end
-Given /^a redis configuration server using redis servers "([^\"]*)" with clients "([^\"]*)" exists$/ do |redis_names, redis_configuration_client_names|
+Given /^a redis configuration server using redis servers "([^\"]*)" with clients "([^\"]*)" (?:and confidence level "([^\"]*)" )?exists$/ do |redis_names, redis_configuration_client_names, confidence_level|
redis_servers = redis_names.split(",").map { |redis_name| TestDaemons::Redis[redis_name].ip_with_port }.join(",")
- TestDaemons::RedisConfigurationServer.start(redis_servers, redis_configuration_client_names)
+ TestDaemons::RedisConfigurationServer.start(redis_servers, redis_configuration_client_names, (confidence_level || 100).to_i)
end
Given /^a redis configuration client "([^\"]*)" using redis servers "([^\"]*)" exists$/ do |redis_configuration_client_name, redis_names|
@@ -37,6 +37,10 @@ Given /^redis server "([^\"]*)" is down$/ do |redis_name|
TestDaemons::Redis[redis_name].stop
end
+Given /^redis configuration client "([^\"]*)" is down$/ do |redis_configuration_client_name|
+ TestDaemons::RedisConfigurationClient[redis_configuration_client_name].stop
+end
+
Given /^the retry timeout for the redis master check is reached$/ do
basetime = Time.now
i = 0

MODIFY
features/support/test_daemons/redis_configuration_server.rb
features/support/test_daemons/redis_configuration_server.rb
@@ -9,11 +9,13 @@ module TestDaemons
@@redis_servers = ""
@@redis_configuration_clients = ""
+ @@confidence_level = 100
- def self.start(redis_servers, redis_configuration_clients)
+ def self.start(redis_servers, redis_configuration_clients, confidence_level)
stop
@@redis_servers = redis_servers
@@redis_configuration_clients = redis_configuration_clients
+ @@confidence_level = confidence_level
daemon_controller.start
end
@@ -25,7 +27,7 @@ module TestDaemons
clients_parameter_string = @@redis_configuration_clients.blank? ? "" : "--client-ids #{@@redis_configuration_clients}"
DaemonController.new(
:identifier => "Redis configuration test server",
- :start_command => "./beetle configuration_server -v -d --redis-master-file #{redis_master_file} --redis-servers #{@@redis_servers} #{clients_parameter_string} --redis-master-retry-interval 1 --pid-file #{pid_file} --log-file #{log_file}",
+ :start_command => "./beetle configuration_server -v -d --redis-master-file #{redis_master_file} --redis-servers #{@@redis_servers} #{clients_parameter_string} --redis-master-retry-interval 1 --pid-file #{pid_file} --log-file #{log_file} --redis-failover-confidence-level #{@@confidence_level}",
:ping_command => lambda{ answers_text_requests? },
:pid_file => pid_file,
:log_file => log_file,

MODIFY
go/src/github.com/xing/beetle/beetle.go
go/src/github.com/xing/beetle/beetle.go
@@ -38,6 +38,7 @@ var opts struct {
GcDatabases string `long:"redis-gc-databases" description:"Database numbers to collect keys from (e.g. 0,4). Defaults to 4."`
MailTo string `long:"mail-to" description:"Send notifcation mails to this address."`
DialTimeout int `long:"dial-timeout" description:"Number of seconds to wait until a connection attempt to the master times out. Defaults to 5."`
+ ConfidenceLevel string `long:"redis-failover-confidence-level" description:"A number between 0 and 100, defining the percent of clients which have to agree in an election process. Values are clamped to the interval [0,100]. Defaults to 100."`
}
// Verbose stores verbosity or logging purposoes.
@@ -204,6 +205,7 @@ func getProgramParameters() *Config {
ClientIds: opts.ClientIds,
ClientHeartbeat: opts.ClientHeartbeatInterval,
ClientTimeout: opts.ClientTimeout,
+ ConfidenceLevel: opts.ConfidenceLevel,
RedisMasterRetries: opts.RedisMasterRetries,
RedisMasterRetryInterval: opts.RedisMasterRetryInterval,
RedisMasterFile: opts.RedisMasterFile,

MODIFY
go/src/github.com/xing/beetle/config.go
go/src/github.com/xing/beetle/config.go
@@ -23,6 +23,7 @@ type Config struct {
GcDatabases string `yaml:"redis_gc_databases"`
MailTo string `yaml:"mail_to"`
DialTimeout int `yaml:"dial_timeout"`
+ ConfidenceLevel string `yaml:"redis_failover_confidence_level"`
}
// Clone copies a give config.
@@ -53,7 +54,8 @@ func (c *Config) Sanitize() {
}
}
-// SetDefaults sets default values for all config options.
+// SetDefaults sets default values for all config options. It is called as the
+// last step to set up a Config.
func (c *Config) SetDefaults() *Config {
if c.ClientTimeout == 0 {
c.ClientTimeout = 10
@@ -88,6 +90,9 @@ func (c *Config) SetDefaults() *Config {
if c.DialTimeout == 0 {
c.DialTimeout = 5
}
+ if c.ConfidenceLevel == "" {
+ c.ConfidenceLevel = "100"
+ }
c.Sanitize()
return c
}
@@ -136,6 +141,12 @@ func (c *Config) Merge(d *Config) *Config {
if c.DialTimeout == 0 {
c.DialTimeout = d.DialTimeout
}
+ if c.DialTimeout == 0 {
+ c.DialTimeout = d.DialTimeout
+ }
+ if c.ConfidenceLevel == "" {
+ c.ConfidenceLevel = d.ConfidenceLevel
+ }
c.Sanitize()
return c
}
@@ -199,6 +210,9 @@ func configFromConsulEnv(env consul.Env) *Config {
c.DialTimeout = d
}
}
+ if v, ok := env["REDIS_FAILOVER_CONFIDENCE_LEVEL"]; ok {
+ c.ConfidenceLevel = v
+ }
c.Sanitize()
return &c
}

MODIFY
go/src/github.com/xing/beetle/server.go
go/src/github.com/xing/beetle/server.go
@@ -66,16 +66,19 @@ type ServerState struct {
currentMaster *RedisShim // Current redis master.
currentTokenInt int // Token to identify election rounds.
currentToken string // String representation of current token.
+ pinging bool // Whether or not we're waiting for pings
clientPongIdsReceived StringSet // During a pong phase, the set of clients which have answered.
+ invalidating bool // Whether or not we're waiting for invalidations.
clientInvalidatedIdsReceived StringSet // During the invalidation phase, the set of clients which have answered.
timerChannel chan string // Channel used to send an abort message to the dispatcher go routine.
- invalidateTimer *time.Timer // Timer used to abort waiting for answers from all clients (invalidate/invalidated).
- availabilityTimer *time.Timer // Timer used to abort waiting for answers from all clients (ping/pong).
+ invalidateTimer *time.Timer // Timer used to abort waiting for answers from clients (invalidate/invalidated).
+ availabilityTimer *time.Timer // Timer used to abort waiting for answers from clients (ping/pong).
retries int // Count down for checking a master to come back after it has become unreachable.
watching bool // Whether we're currently watching a redis master (false during election process).
watchTick int // One second tick counter which gets reset every RedisMasterRetryInterval seconds.
waitGroup sync.WaitGroup // Used to organize the shutdown process.
configChanges chan consul.Env // Environment chnages from consul arrive on this channel.
+ failoverConfidenceLevel float64 // Failover confidence level, normalized to the interval [0,1.0]
}
// GetConfig returns the server state in a thread safe manner.
@@ -94,6 +97,19 @@ func (s *ServerState) SetConfig(config *Config) *Config {
return oldconfig
}
+func (s *ServerState) determineFailoverConfidenceLevel() {
+ config := s.GetConfig()
+ level, err := strconv.Atoi(config.ConfidenceLevel)
+ if err != nil {
+ level = 100
+ } else if level < 0 {
+ level = 0
+ } else if level > 100 {
+ level = 100
+ }
+ s.failoverConfidenceLevel = float64(level) / 100.0
+}
+
// ServerStatus is used to faciliate JSON conversion of parts of the server state.
type ServerStatus struct {
BeetleVersion string `json:"beetle_version"`
@@ -230,6 +246,17 @@ func (l StringSet) Equals(s StringSet) bool {
return true
}
+// Intersect computes the intersection of two string sets
+func (l StringSet) Intersect(r StringSet) StringSet {
+ intersection := make(StringSet)
+ for s := range l {
+ if r[s] {
+ intersection[s] = true
+ }
+ }
+ return intersection
+}
+
// StringList implements convenience functions on string slices.
type StringList []string
@@ -361,6 +388,7 @@ func (s *ServerState) dispatcher() {
case env := <-s.configChanges:
newconfig := buildConfig(env)
s.SetConfig(newconfig)
+ s.determineFailoverConfidenceLevel()
logInfo("updated server config from consul: %s", s.GetConfig())
}
}
@@ -399,6 +427,7 @@ func (s *ServerState) handleWebSocketMsg(msg *WsMsg) {
func NewServerState(o ServerOptions) *ServerState {
s := &ServerState{clientChannels: make(ChannelMap), notificationChannels: make(ChannelSet)}
s.opts = o
+ s.determineFailoverConfidenceLevel()
s.redis = NewRedisServerInfo(s.GetConfig().RedisServers)
s.upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
@@ -766,8 +795,10 @@ func (s *ServerState) Pong(msg MsgBody) {
return
}
s.clientPongIdsReceived.Add(msg.Id)
- if s.AllClientPongIdsReceived() {
- logInfo("All client pong ids received!")
+ level, enough := s.ReceivedEnoughClientPongIds()
+ if s.pinging && enough {
+ logInfo("Received a sufficient number of pong ids!. Confidence level: %f.", level)
+ s.pinging = false
if s.availabilityTimer != nil {
s.availabilityTimer.Stop()
s.availabilityTimer = nil
@@ -814,8 +845,10 @@ func (s *ServerState) ClientInvalidated(msg MsgBody) {
}
logInfo("Received client_invalidated message from id '%s' with token '%s'", msg.Id, msg.Token)
s.clientInvalidatedIdsReceived.Add(msg.Id)
- if s.AllClientInvalidatedIdsReceived() {
- logInfo("All client invalidated ids received")
+ level, enough := s.ReceivedEnoughClientInvalidatedIds()
+ if s.invalidating && enough {
+ logInfo("Received a sufficient number of client invalidated ids! Confidence level: %f.", level)
+ s.invalidating = false
if s.invalidateTimer != nil {
s.invalidateTimer.Stop()
s.invalidateTimer = nil
@@ -1009,11 +1042,13 @@ func (s *ServerState) GenerateNewToken() {
func (s *ServerState) StartInvalidation() {
s.clientPongIdsReceived = make(StringSet)
s.clientInvalidatedIdsReceived = make(StringSet)
- s.CheckAllClientsAvailable()
+ s.pinging = true
+ s.invalidating = false
+ s.CheckEnoughClientsAvailable()
}
-// CheckAllClientsAvailable sends a PING message to all connected clients.
-func (s *ServerState) CheckAllClientsAvailable() {
+// CheckEnoughClientsAvailable sends a PING message to all connected clients.
+func (s *ServerState) CheckEnoughClientsAvailable() {
s.GenerateNewToken()
logInfo("Sending ping messages with token '%s'", s.currentToken)
msg := &MsgBody{Name: PING, Token: s.currentToken}
@@ -1028,6 +1063,7 @@ func (s *ServerState) CheckAllClientsAvailable() {
// clients.
func (s *ServerState) InvalidateCurrentMaster() {
s.GenerateNewToken()
+ s.invalidating = true
logInfo("Sending invalidate messages with token '%s'", s.currentToken)
msg := &MsgBody{Name: INVALIDATE, Token: s.currentToken}
s.SendToWebSockets(msg)
@@ -1040,20 +1076,26 @@ func (s *ServerState) InvalidateCurrentMaster() {
// CancelInvalidation generates a new token to the next vote and unpauses the
// watcher.
func (s *ServerState) CancelInvalidation() {
+ s.pinging = false
+ s.invalidating = false
s.GenerateNewToken()
s.StartWatcher()
}
-// AllClientPongIdsReceived checks whether all client's have answered the PING
-// message by sending a PONG.
-func (s *ServerState) AllClientPongIdsReceived() bool {
- return s.clientIds.Equals(s.clientPongIdsReceived)
+// ReceivedEnoughClientPongIds checks whether a sufficient number of client's have
+// answered the PING message by sending a PONG.
+func (s *ServerState) ReceivedEnoughClientPongIds() (float64, bool) {
+ received := s.clientPongIdsReceived.Intersect(s.clientIds)
+ level := float64(len(received)) / float64(len(s.clientIds))
+ return level, level >= s.failoverConfidenceLevel
}
-// AllClientInvalidatedIdsReceived checks whether all client's have answered the
+// ReceivedEnoughClientInvalidatedIds checks whether all client's have answered the
// INVALIDATE message by sending a CLIENT_INVALIDATED message back.
-func (s *ServerState) AllClientInvalidatedIdsReceived() bool {
- return s.clientIds.Equals(s.clientInvalidatedIdsReceived)
+func (s *ServerState) ReceivedEnoughClientInvalidatedIds() (float64, bool) {
+ received := s.clientInvalidatedIdsReceived.Intersect(s.clientIds)
+ level := float64(len(received)) / float64(len(s.clientIds))
+ return level, level >= s.failoverConfidenceLevel
}
// SwitchMaster is called after a successfully completed vote and performs a

support switching with only a subset of clients responding