[Feature] Identity carry over part 6 / Configure Identity from constuctor

Makefile

@@ -1,4 +1,4 @@
-.PHONY: clean clean-mocks coveralls testpop lint mocks out setup test test-integration test-unit test-race
+.PHONY: clean clean-common clean-mocks coveralls testpop lint mocks out setup test test-integration test-unit test-race
 
 SHELL = /bin/bash
 
@@ -24,6 +24,9 @@ out:	test
 clean:
 	rm -f testpop
 
+clean-common:
+	rm -rf test/ringpop-common
+
 clean-mocks:
 	rm -f test/mocks/*.go forward/mock_*.go
 	rm -rf test/thrift/pingpong/

options.go

@@ -28,8 +28,10 @@ import (
 	log "github.com/uber-common/bark"
 	"github.com/uber/ringpop-go/hashring"
 	"github.com/uber/ringpop-go/logging"
+	"github.com/uber/ringpop-go/membership"
 	"github.com/uber/ringpop-go/shared"
 	"github.com/uber/ringpop-go/swim"
+	"github.com/uber/ringpop-go/util"
 )
 
 type configuration struct {
@@ -52,6 +54,9 @@ type configuration struct {
 	// number of labels and the size of key and value can be configured.
 	LabelLimits swim.LabelOptions
 
+	// InitialLabels configures the initial labels.
+	InitialLabels swim.LabelMap
+
 	// SelfEvict holds the settings with regards to self eviction
 	SelfEvict swim.SelfEvictOptions
 }
@@ -173,6 +178,19 @@ func Statter(s log.StatsReporter) Option {
 	}
 }
 
+// Identity is used to specify an identity as this Ringpop instance's identity.
+// Since a ringpop instance is by default identified by its hostport, it's not allowed to manually specify a
+// hostport as its identity.
+func Identity(identity string) Option {
+	return func(r *Ringpop) error {
+		if util.HostportPattern.MatchString(identity) {
+			return errors.New("a hostport is not valid as an identity")
+		}
+		r.config.InitialLabels[membership.IdentityLabelKey] = identity
+		return nil
+	}
+}
+
 // Address is used to specify a static hostport string as this Ringpop
 // instance's address.
 //

options_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/stretchr/testify/suite"
 	"github.com/uber/ringpop-go/hashring"
 	"github.com/uber/ringpop-go/logging"
+	"github.com/uber/ringpop-go/membership"
 	"github.com/uber/ringpop-go/swim"
 	"github.com/uber/ringpop-go/test/mocks"
 	"github.com/uber/tchannel-go"
@@ -183,6 +184,31 @@ func (s *RingpopOptionsTestSuite) TestMissingAddressResolver() {
 	s.Error(err)
 }
 
+func (s *RingpopOptionsTestSuite) TestIdentity() {
+	rp, err := New("test", Channel(s.channel), Identity("identity"))
+	s.Require().NotNil(rp)
+	s.Require().NoError(err)
+
+	identity, has := rp.config.InitialLabels[membership.IdentityLabelKey]
+	s.Equal(true, has, "Identity label set")
+	s.Equal("identity", identity)
+}
+
+func (s *RingpopOptionsTestSuite) TestDefaultIdentity() {
+	rp, err := New("test", Channel(s.channel))
+	s.Require().NotNil(rp)
+	s.Require().NoError(err)
+
+	_, has := rp.config.InitialLabels[membership.IdentityLabelKey]
+	s.Equal(false, has, "Identity label not set")
+}
+
+func (s *RingpopOptionsTestSuite) TestHostPortIdentity() {
+	rp, err := New("test", Channel(s.channel), Identity("127.0.0.1:1234"))
+	s.Nil(rp)
+	s.Error(err)
+}
+
 // TestClockNil confirms that nil clock option returns an error.
 func (s *RingpopOptionsTestSuite) TestClockNil() {
 	rp, err := New("test", Clock(nil))

ringpop.go

@@ -144,7 +144,8 @@ func New(app string, opts ...Option) (*Ringpop, error) {
 
 	ringpop := &Ringpop{
 		config: &configuration{
-			App: app,
+			App:           app,
+			InitialLabels: make(swim.LabelMap),
 		},
 		logger: logging.Logger("ringpop"),
 	}
@@ -193,6 +194,7 @@ func (rp *Ringpop) init() error {
 		StateTimeouts: rp.config.StateTimeouts,
 		Clock:         rp.clock,
 		LabelLimits:   rp.config.LabelLimits,
+		InitialLabels: rp.config.InitialLabels,
 		SelfEvict:     rp.config.SelfEvict,
 	})
 	rp.node.AddListener(rp)
@@ -797,18 +799,6 @@ func (rp *Ringpop) Labels() (*swim.NodeLabels, error) {
 	return rp.node.Labels(), nil
 }
 
-// SetIdentity changes the identity for this process. The identity is used by
-// all members of the membership to calculate the position of this node in the
-// hashring.
-func (rp *Ringpop) SetIdentity(identity string) error {
-	if !rp.Ready() {
-		return ErrNotBootstrapped
-	}
-
-	rp.node.SetIdentity(identity)
-	return nil
-}
-
 // SerializeThrift takes a thrift struct and returns the serialized bytes
 // of that struct using the thrift binary protocol. This is a temporary
 // measure before frames can forwarded directly past the endpoint to the proper

ringpop_test.go

@@ -754,6 +754,14 @@ func (s *RingpopTestSuite) TestStartTimersIdempotance() {
 	s.ringpop.stopTimers()
 }
 
+func (s *RingpopTestSuite) TestInitialLabels() {
+	s.ringpop.config.InitialLabels["key"] = "value"
+	s.ringpop.init()
+	value, has := s.ringpop.node.Labels().Get("key")
+	s.Equal("value", value, "Label correctly set on local node")
+	s.True(has, "Label correctly set on local node")
+}
+
 func (s *RingpopTestSuite) TestReadyEvent() {
 	called := make(chan bool, 1)
 

scripts/testpop/testpop.go

@@ -48,6 +48,8 @@ var (
 	faultyPeriod    = flag.Int("faulty-period", 24*60*60*1000, "The lifetime of a faulty member in ms. After that the member becomes a tombstone.")
 	tombstonePeriod = flag.Int("tombstone-period", 5000, "The lifetime of a tombstone member in ms. After that the member is removed from the membership.")
 
+	identity = flag.String("identity", "", "specify an identity")
+
 	hostportPattern = regexp.MustCompile(`^(\d+.\d+.\d+.\d+):\d+$`)
 )
 
@@ -78,6 +80,10 @@ func main() {
 		ringpop.TombstonePeriod(time.Duration(*tombstonePeriod) * time.Millisecond),
 	}
 
+	if *identity != "" {
+		options = append(options, ringpop.Identity(*identity))
+	}
+
 	if *statsUDP != "" && *statsFile != "" {
 		log.Fatalf("-stats-udp and stats-file are mutually exclusive.")
 	}

swim/join_sender.go

@@ -56,10 +56,11 @@ var errJoinTimeout = errors.New("join timed out")
 
 // A joinRequest is used to request a join to a remote node
 type joinRequest struct {
-	App         string        `json:"app"`
-	Source      string        `json:"source"`
-	Incarnation int64         `json:"incarnationNumber"`
-	Timeout     time.Duration `json:"timeout"`
+	App         string            `json:"app"`
+	Source      string            `json:"source"`
+	Incarnation int64             `json:"incarnationNumber"`
+	Timeout     time.Duration     `json:"timeout"`
+	Labels      map[string]string `json:"labels,omitempty"`
 }
 
 // joinOpts are opts to perform a join with
@@ -446,6 +447,7 @@ func sendJoinRequest(node *Node, target string, timeout time.Duration) (*joinRes
 		Source:      node.address,
 		Incarnation: node.Incarnation(),
 		Timeout:     timeout,
+		Labels:      node.Labels().AsMap(),
 	}
 	res := &joinResponse{}
 

swim/memberlist.go

@@ -61,7 +61,7 @@ type memberlist struct {
 }
 
 // newMemberlist returns a new member list
-func newMemberlist(n *Node) *memberlist {
+func newMemberlist(n *Node, initialLabels LabelMap) *memberlist {
 	m := &memberlist{
 		node:   n,
 		logger: logging.Logger("membership").WithField("local", n.address),
@@ -71,6 +71,7 @@ func newMemberlist(n *Node) *memberlist {
 			Address:     n.Address(),
 			Incarnation: nowInMillis(n.clock),
 			Status:      Alive,
+			Labels:      initialLabels,
 		},
 	}
 

swim/node.go

@@ -68,7 +68,8 @@ type Options struct {
 	PartitionHealPeriod           time.Duration
 	PartitionHealBaseProbabillity float64
 
-	LabelLimits LabelOptions
+	LabelLimits   LabelOptions
+	InitialLabels LabelMap
 
 	Clock clock.Clock
 
@@ -250,7 +251,7 @@ func NewNode(app, address string, channel shared.SubChannel, opts *Options) *Nod
 
 	node.labelLimits = opts.LabelLimits
 
-	node.memberlist = newMemberlist(node)
+	node.memberlist = newMemberlist(node, opts.InitialLabels)
 	node.memberiter = newMemberlistIter(node.memberlist)
 	node.stateTransitions = newStateTransitions(node, opts.StateTimeouts)
 

test/lib.sh

@@ -2,7 +2,7 @@
 
 declare project_root="${0%/*}/.."
 declare ringpop_common_dir="${0%/*}/ringpop-common"
-declare ringpop_common_branch="menno/allow-multi-checksum"
+declare ringpop_common_branch="feature/identity"
 
 #
 # Clones or updates the ringpop-common repository.