ridv/gorealis
1
0
Fork 0
Code Issues Pull requests Projects Releases 23 Packages Wiki Activity Actions

Cleaned up ZK connection code by using the backoff function. Added a test to the end to end to test that we're getting the host correctly from ZK. Changed clusters test to be an outside package.

Browse source
This commit is contained in:
Renan DelValle 2017-11-28 17:30:33 -08:00
parent f60aa3dd88
commit f6a21e0f59
3 changed files with 65 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

123
zk.go
View file

@ -44,81 +44,66 @@ func (NoopLogger) Printf(format string, a ...interface{}) {
// Retrieves current Aurora leader from ZK.
func LeaderFromZK(cluster Cluster) (string, error) {
var err error
var zkurl string
duration := defaultBackoff.Duration
for step := 0; step < defaultBackoff.Steps; step++ {
retryErr := ExponentialBackoff(defaultBackoff, func() (bool, error) {
// Attempt to find leader
zkurl, err = leaderFromZK(cluster)
if err == nil {
return zkurl, err
endpoints := strings.Split(cluster.ZK, ",")
//TODO (rdelvalle): When enabling debugging, change logger here
c, _, err := zk.Connect(endpoints, time.Second*10, func(c *zk.Conn) { c.SetLogger(NoopLogger{}) })
if err != nil {
return false, errors.Wrap(err, "Failed to connect to Zookeeper at "+cluster.ZK)
}
// Backoff if we failed to determine leader
adjusted := duration
if defaultBackoff.Jitter > 0.0 {
adjusted = Jitter(duration, defaultBackoff.Jitter)
defer c.Close()
// Open up descriptor for the ZK path given
children, _, _, err := c.ChildrenW(cluster.SchedZKPath)
if err != nil {
return false, errors.Wrapf(err, "Path %s doesn't exist on Zookeeper ", cluster.SchedZKPath)
}
fmt.Printf("Error determining Aurora leader: %v; retrying in %v\n", err, adjusted)
time.Sleep(adjusted)
duration = time.Duration(float64(duration) * defaultBackoff.Factor)
// Search for the leader through all the children in the given path
serviceInst := new(ServiceInstance)
for _, child := range children {
// Only the leader will start with member_
if strings.HasPrefix(child, "member_") {
data, _, err := c.Get(cluster.SchedZKPath + "/" + child)
if err != nil {
return false, errors.Wrap(err, "Error fetching contents of leader")
}
err = json.Unmarshal([]byte(data), serviceInst)
if err != nil {
return false, errors.Wrap(err, "Unable to unmarshall contents of leader")
}
// Should only be one endpoint
if len(serviceInst.AdditionalEndpoints) > 1 {
fmt.Errorf("Ambiguous end points schemes")
}
var scheme, host, port string
for k, v := range serviceInst.AdditionalEndpoints {
scheme = k
host = v.Host
port = strconv.Itoa(v.Port)
}
zkurl = scheme + "://" + host + ":" + port
return true, nil
}
}
return false, errors.New("No leader found")
})
if retryErr != nil {
return "", errors.Wrapf(retryErr, "Failed to determine leader after %v attempts", defaultBackoff.Steps)
}
return "", errors.Wrapf(err, "Failed to determine leader after %v attempts", defaultBackoff.Steps)
}
func leaderFromZK(cluster Cluster) (string, error) {
endpoints := strings.Split(cluster.ZK, ",")
//TODO (rdelvalle): When enabling debugging, change logger here
c, _, err := zk.Connect(endpoints, time.Second*10, func(c *zk.Conn) { c.SetLogger(NoopLogger{}) })
if err != nil {
return "", errors.Wrap(err, "Failed to connect to Zookeeper at "+cluster.ZK)
}
defer c.Close()
children, _, _, err := c.ChildrenW(cluster.SchedZKPath)
if err != nil {
return "", errors.Wrapf(err, "Path %s doesn't exist on Zookeeper ", cluster.SchedZKPath)
}
serviceInst := new(ServiceInstance)
for _, child := range children {
// Only the leader will start with member_
if strings.HasPrefix(child, "member_") {
data, _, err := c.Get(cluster.SchedZKPath + "/" + child)
if err != nil {
return "", errors.Wrap(err, "Error fetching contents of leader")
}
err = json.Unmarshal([]byte(data), serviceInst)
if err != nil {
return "", errors.Wrap(err, "Unable to unmarshall contents of leader")
}
// Should only be one endpoint
if len(serviceInst.AdditionalEndpoints) > 1 {
fmt.Errorf("Ambiguous end points schemes")
}
var scheme, host, port string
for k, v := range serviceInst.AdditionalEndpoints {
scheme = k
host = v.Host
port = strconv.Itoa(v.Port)
}
return scheme + "://" + host + ":" + port, nil
}
}
return "", errors.New("No leader found")
return zkurl, nil
}