|
@@ -1,6 +1,8 @@
|
|
package daemon
|
|
package daemon
|
|
|
|
|
|
import (
|
|
import (
|
|
|
|
+ "fmt"
|
|
|
|
+ "strconv"
|
|
"time"
|
|
"time"
|
|
|
|
|
|
log "github.com/Sirupsen/logrus"
|
|
log "github.com/Sirupsen/logrus"
|
|
@@ -13,22 +15,63 @@ import (
|
|
const (
|
|
const (
|
|
// defaultDiscoveryHeartbeat is the default value for discovery heartbeat interval.
|
|
// defaultDiscoveryHeartbeat is the default value for discovery heartbeat interval.
|
|
defaultDiscoveryHeartbeat = 20 * time.Second
|
|
defaultDiscoveryHeartbeat = 20 * time.Second
|
|
-
|
|
|
|
- // defaultDiscoveryTTL is the default TTL interface for discovery.
|
|
|
|
- defaultDiscoveryTTL = 60 * time.Second
|
|
|
|
|
|
+ // defaultDiscoveryTTLFactor is the default TTL factor for discovery
|
|
|
|
+ defaultDiscoveryTTLFactor = 3
|
|
)
|
|
)
|
|
|
|
|
|
|
|
+func discoveryOpts(clusterOpts map[string]string) (time.Duration, time.Duration, error) {
|
|
|
|
+ var (
|
|
|
|
+ heartbeat = defaultDiscoveryHeartbeat
|
|
|
|
+ ttl = defaultDiscoveryTTLFactor * defaultDiscoveryHeartbeat
|
|
|
|
+ )
|
|
|
|
+
|
|
|
|
+ if hb, ok := clusterOpts["discovery.heartbeat"]; ok {
|
|
|
|
+ h, err := strconv.Atoi(hb)
|
|
|
|
+ if err != nil {
|
|
|
|
+ return time.Duration(0), time.Duration(0), err
|
|
|
|
+ }
|
|
|
|
+ heartbeat = time.Duration(h) * time.Second
|
|
|
|
+ ttl = defaultDiscoveryTTLFactor * heartbeat
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ if tstr, ok := clusterOpts["discovery.ttl"]; ok {
|
|
|
|
+ t, err := strconv.Atoi(tstr)
|
|
|
|
+ if err != nil {
|
|
|
|
+ return time.Duration(0), time.Duration(0), err
|
|
|
|
+ }
|
|
|
|
+ ttl = time.Duration(t) * time.Second
|
|
|
|
+
|
|
|
|
+ if _, ok := clusterOpts["discovery.heartbeat"]; !ok {
|
|
|
|
+ h := int(t / defaultDiscoveryTTLFactor)
|
|
|
|
+ heartbeat = time.Duration(h) * time.Second
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ if ttl <= heartbeat {
|
|
|
|
+ return time.Duration(0), time.Duration(0),
|
|
|
|
+ fmt.Errorf("discovery.ttl timer must be greater than discovery.heartbeat")
|
|
|
|
+ }
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ return heartbeat, ttl, nil
|
|
|
|
+}
|
|
|
|
+
|
|
// initDiscovery initialized the nodes discovery subsystem by connecting to the specified backend
|
|
// initDiscovery initialized the nodes discovery subsystem by connecting to the specified backend
|
|
// and start a registration loop to advertise the current node under the specified address.
|
|
// and start a registration loop to advertise the current node under the specified address.
|
|
func initDiscovery(backend, address string, clusterOpts map[string]string) (discovery.Backend, error) {
|
|
func initDiscovery(backend, address string, clusterOpts map[string]string) (discovery.Backend, error) {
|
|
- discoveryBackend, err := discovery.New(backend, defaultDiscoveryHeartbeat, defaultDiscoveryTTL, clusterOpts)
|
|
|
|
|
|
+
|
|
|
|
+ heartbeat, ttl, err := discoveryOpts(clusterOpts)
|
|
|
|
+ if err != nil {
|
|
|
|
+ return nil, err
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ discoveryBackend, err := discovery.New(backend, heartbeat, ttl, clusterOpts)
|
|
if err != nil {
|
|
if err != nil {
|
|
return nil, err
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
|
|
// We call Register() on the discovery backend in a loop for the whole lifetime of the daemon,
|
|
// We call Register() on the discovery backend in a loop for the whole lifetime of the daemon,
|
|
// but we never actually Watch() for nodes appearing and disappearing for the moment.
|
|
// but we never actually Watch() for nodes appearing and disappearing for the moment.
|
|
- go registrationLoop(discoveryBackend, address)
|
|
|
|
|
|
+ go registrationLoop(discoveryBackend, address, heartbeat)
|
|
return discoveryBackend, nil
|
|
return discoveryBackend, nil
|
|
}
|
|
}
|
|
|
|
|
|
@@ -41,9 +84,9 @@ func registerAddr(backend discovery.Backend, addr string) {
|
|
// registrationLoop registers the current node against the discovery backend using the specified
|
|
// registrationLoop registers the current node against the discovery backend using the specified
|
|
// address. The function never returns, as registration against the backend comes with a TTL and
|
|
// address. The function never returns, as registration against the backend comes with a TTL and
|
|
// requires regular heartbeats.
|
|
// requires regular heartbeats.
|
|
-func registrationLoop(discoveryBackend discovery.Backend, address string) {
|
|
|
|
|
|
+func registrationLoop(discoveryBackend discovery.Backend, address string, heartbeat time.Duration) {
|
|
registerAddr(discoveryBackend, address)
|
|
registerAddr(discoveryBackend, address)
|
|
- for range time.Tick(defaultDiscoveryHeartbeat) {
|
|
|
|
|
|
+ for range time.Tick(heartbeat) {
|
|
registerAddr(discoveryBackend, address)
|
|
registerAddr(discoveryBackend, address)
|
|
}
|
|
}
|
|
}
|
|
}
|