store master file contents in consul
MODIFY
README.rdoc
README.rdoc
@@ -96,6 +96,7 @@ For development, you'll need
* {mocha}[http://github.com/floehopper/mocha]
* {cucumber}[http://github.com/aslakhellesoy/cucumber]
* {daemon_controller}[http://github.com/FooBarWidget/daemon_controller]
+* {consul}[https://www.consul.io/downloads.html]
For tests, you'll need
* {activerecord}[https://github.com/rails/rails/tree/master/activerecord]
@@ -120,9 +121,11 @@ distribution.
1. Fork it
2. Create your feature branch (`git checkout -b my-new-feature`)
-3. Commit your changes (`git commit -am 'Add some feature'`)
-4. Push to the branch (`git push origin my-new-feature`)
-5. Create new Pull Request
+3. Hack along and test your code using rake test and rake
+ cucumber. You must start consul in order to use cucumber.
+4. Commit your changes (`git commit -am 'Add some feature'`)
+5. Push to the branch (`git push origin my-new-feature`)
+6. Create new Pull Request
Don't increase the gem version in your pull requests. It will be done after merging the request,
to allow merging of pull requests in a flexible order.

MODIFY
Rakefile
Rakefile
@@ -82,6 +82,13 @@ namespace :redis do
end
end
+namespace :consul do
+ desc "start consul agent in development mode"
+ task :start do
+ system "consul agent -dev -node machine"
+ end
+end
+
Cucumber::Rake::Task.new(:cucumber) do |t|
t.cucumber_opts = "features --format progress"
end

MODIFY
features/redis_auto_failover.feature
features/redis_auto_failover.feature
@@ -5,6 +5,7 @@ Feature: Redis auto failover
Background:
Given a redis server "redis-1" exists as master
And a redis server "redis-2" exists as slave of "redis-1"
+ And consul state has been cleared
Scenario: Successful redis master switch
Given a redis configuration server using redis servers "redis-1,redis-2" with clients "rc-client-1,rc-client-2" exists

MODIFY
features/step_definitions/redis_auto_failover_steps.rb
features/step_definitions/redis_auto_failover_steps.rb
+Given /^consul state has been cleared$/ do
+ system "killall beetle beetle_handler >/dev/null 2>/dev/null"
+ system "curl --silent --request PUT http://localhost:8500/v1/kv/apps/beetle/config/ >/dev/null"
+ system "curl --silent --request PUT http://localhost:8500/v1/kv/shared/config/ >/dev/null"
+ system "curl --silent --request DELETE http://localhost:8500/v1/kv/apps/beetle/state/redis_master_file_content >/dev/null"
+end
+
Given /^a redis server "([^\"]*)" exists as master$/ do |redis_name|
TestDaemons::Redis[redis_name].restart
TestDaemons::Redis[redis_name].master

MODIFY
features/support/test_daemons/redis_configuration_server.rb
features/support/test_daemons/redis_configuration_server.rb
@@ -27,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} --redis-failover-confidence-level #{@@confidence_level}",
+ :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} --consul http://loclahost:8500",
:ping_command => lambda{ answers_text_requests? },
:pid_file => pid_file,
:log_file => log_file,

MODIFY
go/consul/consul.go
go/consul/consul.go
package consul
import (
+ "bytes"
"encoding/base64"
"encoding/json"
+ "fmt"
"io/ioutil"
"log"
"net"
@@ -11,6 +13,7 @@ import (
"sort"
"strconv"
"strings"
+ "time"
"github.com/pkg/errors"
)
@@ -40,11 +43,23 @@ func (es Entries) Swap(i, j int) {
es[i], es[j] = es[j], es[i]
}
+// Update inserts or updates a key value pair
+func (es *Entries) Update(key string, value string) {
+ for i, e := range *es {
+ if e.Key == key {
+ (*es)[i].Value = value
+ return
+ }
+ }
+ *es = append(*es, Entry{Key: key, Value: value})
+}
+
// Space represents a space in consul
type Space struct {
prefix string
modifyIndex int
entries Entries
+ upcase bool
}
// Client is used to access consul
@@ -53,6 +68,7 @@ type Client struct {
appName string
appConfig Space
sharedConfig Space
+ state Space
dataCenter string
dataCenters []string
}
@@ -62,8 +78,9 @@ func NewClient(consulUrl string, appName string) *Client {
client := Client{
consulUrl: consulUrl,
appName: appName,
- appConfig: Space{prefix: "apps/" + appName + "/config/"},
- sharedConfig: Space{prefix: "shared/config/"},
+ appConfig: Space{prefix: "apps/" + appName + "/config/", upcase: true},
+ sharedConfig: Space{prefix: "shared/config/", upcase: true},
+ state: Space{prefix: "apps/" + appName + "/state/", upcase: false},
}
n := len(client.consulUrl)
if n > 0 && client.consulUrl[n-1] != '/' {
@@ -168,6 +185,51 @@ func (c *Client) GetData(space *Space, useIndex bool) error {
return nil
}
+// GetState loads the state space from consul
+func (c *Client) GetState() (env Env, err error) {
+ if Verbose {
+ log.Printf("Retrieving state for %s from consul %s\n", c.appName, c.consulUrl)
+ }
+ if err = c.GetData(&c.state, false); err != nil {
+ return
+ }
+ env = make(Env)
+ c.addEntriesToEnv(&c.state, env)
+ return
+}
+
+// UpdateState stores a single key value pair in the state
+func (c *Client) UpdateState(key string, value string) error {
+ client := &http.Client{Timeout: time.Second * 5}
+ uri := c.kvUrl(c.state.prefix + key)
+ if Verbose {
+ log.Printf("PUT %s\n", uri)
+ log.Printf("VALUE %s\n", value)
+ }
+ body := bytes.NewBufferString(value)
+ req, err := http.NewRequest(http.MethodPut, uri, body)
+ if err != nil {
+ return errors.Wrapf(err, "PUT %q failed", uri)
+ }
+ resp, err := client.Do(req)
+ if err != nil {
+ return errors.Wrapf(err, "PUT %q failed", uri)
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ return fmt.Errorf("PUT %q failed with status: %s", uri, resp.Status)
+ }
+ d, err := ioutil.ReadAll(resp.Body)
+ if err != nil {
+ return errors.Wrapf(err, "PUT %q failed", uri)
+ }
+ if Verbose {
+ log.Printf("PUT response: %s", string(d))
+ }
+ c.state.entries.Update(key, value)
+ return nil
+}
+
// GetEnv loads shared and app specific config from consul and returns it as a
// string map
func (c *Client) GetEnv() (env Env, err error) {
@@ -244,7 +306,9 @@ func (c *Client) addEntriesToEnv(space *Space, env Env) (err error) {
}
x.Key = c.transformKeyAccordingToDC(x.Key)
x.Key = strings.Replace(x.Key, "/", "_", -1)
+ if space.upcase {
x.Key = strings.ToUpper(x.Key)
+ }
if x.Key != "" && x.Key != "RESTART" && !strings.HasSuffix(x.Key, "_") {
if Verbose {
log.Printf("adding %s=%v\n", x.Key, x.Value)

MODIFY
go/consul/consul_test.go
go/consul/consul_test.go
@@ -4,14 +4,16 @@ import (
"fmt"
"io/ioutil"
"log"
+ "math/rand"
"os"
"reflect"
+ "strconv"
"testing"
"time"
)
var testUrl = ""
-var testApp = "activities"
+var testApp = "beetle"
func init() {
log.SetFlags(0)
@@ -19,11 +21,7 @@ func init() {
if !Verbose {
log.SetOutput(ioutil.Discard)
}
- testUrl = os.Getenv("BEETLE_CONSUL_TEST_URL")
- if testUrl == "" {
- fmt.Println("BEETLE_CONSUL_TEST_URL needs to be set to the consul url used for testing (e.g. your preview server)")
- os.Exit(1)
- }
+ testUrl = "http://localhost:8500"
}
func TestConnect(t *testing.T) {
@@ -36,6 +34,21 @@ func TestConnect(t *testing.T) {
t.Errorf("could not retrieve environment: %s", err)
}
}
+func TestState(t *testing.T) {
+ client := NewClient(testUrl, testApp)
+ client.Initialize()
+ value := strconv.Itoa(rand.Int())
+ if err := client.UpdateState("test", value); err != nil {
+ t.Errorf("could not set test key: %s", err)
+ }
+ kv, err := client.GetState()
+ if err != nil {
+ t.Errorf("could not get state keys: %s", err)
+ }
+ if kv["test"] != value {
+ t.Errorf("retrieved test keys have wrong value: %+v", kv)
+ }
+}
func TestWatching(t *testing.T) {
client := NewClient(testUrl, testApp)

MODIFY
go/go.sum
go/go.sum
@@ -362,6 +362,7 @@ golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8/go.mod h1:STP8DvDyc/dI5b8T5h
golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20181206074257-70b957f3b65e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190102155601-82a175fd1598/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190116161447-11f53e031339 h1:g/Jesu8+QLnA0CPzF3E1pURg0Byr7i6jLoX5sqjcAh0=
golang.org/x/sys v0.0.0-20190116161447-11f53e031339/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

MODIFY
go/server.go
go/server.go
@@ -659,7 +659,7 @@ func (s *ServerState) serveNotifications(w http.ResponseWriter, r *http.Request)
ws, err := s.upgrader.Upgrade(w, r, nil)
if err != nil {
if _, ok := err.(websocket.HandshakeError); !ok {
- logError(err.Error())
+ logError("serveNotifications: %s", err)
}
return
}
@@ -706,7 +706,7 @@ func (s *ServerState) serveWs(w http.ResponseWriter, r *http.Request) {
ws, err := s.upgrader.Upgrade(w, r, nil)
if err != nil {
if _, ok := err.(websocket.HandshakeError); !ok {
- logError(err.Error())
+ logError("serveWS: %s", err)
}
return
}
@@ -837,7 +837,16 @@ func (s *ServerState) wsWriter(clientID string, ws *websocket.Conn, inputFromDis
func (s *ServerState) Initialize() {
path := s.GetConfig().RedisMasterFile
VerifyMasterFileString(path)
- masters := RedisMastersFromMasterFile(path)
+ var masters map[string]string
+ if MasterFileExists(path) {
+ masters = RedisMastersFromMasterFile(path)
+ } else if s.opts.ConsulClient != nil {
+ kv, err := s.opts.ConsulClient.GetState()
+ if err != nil {
+ logError("Could not load state from consul: %s", err)
+ }
+ masters = UnmarshalMasterFileContent(kv["redis_master_file_content"])
+ }
for system, fs := range s.failovers {
fs.CheckRedisConfiguration()
fs.redis.Refresh()
@@ -1062,6 +1071,12 @@ func (s *ServerState) UpdateMasterFile() {
}
content := MarshalMasterFileContent(systems)
WriteRedisMasterFile(path, content)
+ if s.opts.ConsulClient != nil {
+ err := s.opts.ConsulClient.UpdateState("redis_master_file_content", content)
+ if err != nil {
+ logError("could not update consul state")
+ }
+ }
}
// DetermineInitialMaste either uses information from the master file on disk

store master file contents in consul