Support custom address broadcasting for ringpop to work in k8s

common/membership/hashring.go

@@ -328,7 +328,14 @@ func (r *ring) emitHashIdentifier() float64 {
 	// in-case it overflowed something. The number itself is meaningless, so additional precision
 	// doesn't really give any advantage, besides reducing the risk of collision
 	trimmedForMetric := float64(hashedView % 1000)
-	r.logger.Debug("Hashring view", tag.Dynamic("hashring-view", sb.String()), tag.Dynamic("trimmed-hash-id", trimmedForMetric), tag.Service(r.service))
+	r.logger.Debug("Hashring view",
+		tag.Dynamic("hashring-view", sb.String()),
+		tag.Dynamic("trimmed-hash-id", trimmedForMetric),
+		tag.Service(r.service),
+		tag.Dynamic("self-addr", self.addr),
+		tag.Dynamic("self-identity", self.identity),
+		tag.Dynamic("self-ip", self.ip),
+	)
 	r.scope.Tagged(
 		metrics.ServiceTag(r.service),
 		metrics.HostTag(self.identity),

common/peerprovider/ringpopprovider/config.go

@@ -65,6 +65,10 @@ const (
 type Config struct {
 	// Name to be used in ringpop advertisement
 	Name string `yaml:"name" validate:"nonzero"`
+	// BroadcastAddress is communicated with peers to connect to this container/host.
+	// This is useful when running cadence in K8s and the containers need to listen on 0.0.0.0 but advertise their pod IP to peers.
+	// If not set, the listen address will be used as broadcast address.
+	BroadcastAddress string `yaml:"broadcastAddress"`
 	// BootstrapMode is a enum that defines the ringpop bootstrap method, currently supports: hosts, files, custom, dns, and dns-srv
 	BootstrapMode BootstrapMode `yaml:"bootstrapMode"`
 	// BootstrapHosts is a list of seed hosts to be used for ringpop bootstrap

common/peerprovider/ringpopprovider/provider.go

@@ -23,11 +23,13 @@
 package ringpopprovider
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"strconv"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/uber/ringpop-go"
 	"github.com/uber/ringpop-go/events"
@@ -37,6 +39,7 @@ import (
 	"go.uber.org/yarpc/transport/tchannel"
 
 	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/backoff"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/common/membership"
@@ -85,14 +88,42 @@ func New(
 		DiscoverProvider: discoveryProvider,
 	}
 
-	rp, err := ringpop.New(config.Name, ringpop.Channel(channel.(*tcg.Channel)))
+	ch := channel.(*tcg.Channel)
+	opts := []ringpop.Option{ringpop.Channel(ch)}
+	if config.BroadcastAddress != "" {
+		broadcastIP := net.ParseIP(config.BroadcastAddress)
+		if broadcastIP == nil {
+			return nil, fmt.Errorf("failed parsing broadcast address %q, err: %w", config.BroadcastAddress, err)
+		}
+		logger.Info("Initializing ringpop with custom broadcast address", tag.Address(broadcastIP.String()))
+		opts = append(opts, ringpop.AddressResolverFunc(broadcastAddrResolver(ch, broadcastIP)))
+	}
+	rp, err := ringpop.New(config.Name, opts...)
 	if err != nil {
 		return nil, fmt.Errorf("ringpop instance creation: %w", err)
 	}
 
 	return NewRingpopProvider(service, rp, portMap, bootstrapOpts, logger), nil
 }
 
+func broadcastAddrResolver(ch *tcg.Channel, broadcastIP net.IP) func() (string, error) {
+	return func() (string, error) {
+		peerInfo := ch.PeerInfo()
+		if peerInfo.IsEphemeralHostPort() {
+			// not listening yet
+			return "", ringpop.ErrEphemeralAddress
+		}
+
+		_, port, err := net.SplitHostPort(peerInfo.HostPort)
+		if err != nil {
+			return "", fmt.Errorf("failed splitting tchannel's hostport %q, err: %w", peerInfo.HostPort, err)
+		}
+
+		// return broadcast_ip:listen_port
+		return net.JoinHostPort(broadcastIP.String(), port), nil
+	}
+}
+
 // NewRingpopProvider sets up ringpop based peer provider
 func NewRingpopProvider(
 	service string,
@@ -122,8 +153,15 @@ func (r *Provider) Start() {
 		return
 	}
 
-	_, err := r.ringpop.Bootstrap(r.bootParams)
-	if err != nil {
+	op := func() error {
+		_, err := r.ringpop.Bootstrap(r.bootParams)
+		if err != nil {
+			r.logger.Warn("unable to bootstrap ringpop, will retry", tag.Error(err))
+		}
+		return err
+	}
+	retry := backoff.NewThrottleRetry(backoff.WithRetryPolicy(bootstrapRetryPolicy()))
+	if err := retry.Do(context.Background(), op); err != nil {
 		r.logger.Fatal("unable to bootstrap ringpop", tag.Error(err))
 	}
 
@@ -148,9 +186,7 @@ func (r *Provider) Start() {
 }
 
 // HandleEvent handles updates from ringpop
-func (r *Provider) HandleEvent(
-	event events.Event,
-) {
+func (r *Provider) HandleEvent(event events.Event) {
 	// We only care about RingChangedEvent
 	e, ok := event.(events.RingChangedEvent)
 	if !ok {
@@ -300,3 +336,10 @@ func labelToPort(label string) (uint16, error) {
 	}
 	return uint16(port), nil
 }
+
+func bootstrapRetryPolicy() backoff.RetryPolicy {
+	policy := backoff.NewExponentialRetryPolicy(5 * time.Second)
+	policy.SetMaximumInterval(5 * time.Second)
+	policy.SetExpirationInterval(45 * time.Second)
+	return policy
+}

docker/config_template.yaml

@@ -97,6 +97,7 @@ persistence:
 
 ringpop:
     name: cadence
+    broadcastAddress: {{ default .Env.BROADCAST_ADDRESS "" }}
     bootstrapMode: {{ default .Env.RINGPOP_BOOTSTRAP_MODE "hosts" }}
     {{- if .Env.RINGPOP_SEEDS }}
     bootstrapHosts: