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

CHANGES.md

@@ -1,6 +1,24 @@
 ringpop-go changes
 ==================
 
+NEXT
+------
+
+### Release notes
+
+#### Change to identity-option
+
+Prior to ringpop NEXT the address was used as the identity of a member. Starting with version NEXT, it's possible to
+configure a separate identity. As a result, the behaviour of the `Identity` and `IdentityResolverFunc` has been changed.
+The `Identity` option now configures the identity of a member and will return an error when it matches an ip:port; services 
+that were using `Identity` or `IdentityResolverFunc` should now use the `Address` and `AddressResolverFunc` options.
+You could use the following `gofmt` snippets to easily refactor:
+
+```
+gofmt -r 'ringpop.Identity(a) -> ringpop.Address(a)' -w .
+gofmt -r 'ringpop.IdentityResolverFunc(a) -> ringpop.AddressResolverFunc(a)' -w .
+```
+
 v0.7.0
 ------
 

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/

errors.go

@@ -7,10 +7,10 @@ var (
 	// be bootstrapped before they can operate correctly.
 	ErrNotBootstrapped = errors.New("ringpop is not bootstrapped")
 
-	// ErrEphemeralIdentity is returned by the identity resolver if TChannel is
+	// ErrEphemeralAddress is returned by the address resolver if TChannel is
 	// using port 0 and is not listening (and thus has not been assigned a port by
 	// the OS).
-	ErrEphemeralIdentity = errors.New("unable to resolve this node's identity from channel that is not yet listening")
+	ErrEphemeralAddress = errors.New("unable to resolve this node's address from channel that is not yet listening")
 
 	// ErrChannelNotListening is returned on bootstrap if TChannel is not
 	// listening.

events/events.go

@@ -159,14 +159,21 @@ func (a *SyncEventEmitter) EmitEvent(event Event) {
 // A RingChangedEvent is sent when servers are added and/or removed from the ring
 type RingChangedEvent struct {
 	ServersAdded   []string
+	ServersUpdated []string
 	ServersRemoved []string
 }
 
 // RingChecksumEvent is sent when a server is removed or added and a new checksum
 // for the ring is calculated
 type RingChecksumEvent struct {
+	// OldChecksum contains the previous legacy checksum. Note: might be deprecated in the future.
 	OldChecksum uint32
+	// NewChecksum contains the new legacy checksum. Note: might be deprecated in the future.
 	NewChecksum uint32
+	// OldChecksums contains the map of previous checksums
+	OldChecksums map[string]uint32
+	// NewChecksums contains the map with new checksums
+	NewChecksums map[string]uint32
 }
 
 // A LookupEvent is sent when a lookup is performed on the Ringpop's ring

examples/ping-json/main.go

@@ -84,11 +84,11 @@ func (w *worker) PingHandler(ctx json.Context, request *Ping) (*Pong, error) {
 	forwardOptions := &forward.Options{Headers: []byte(marshaledHeaders)}
 	handle, err := w.ringpop.HandleOrForward(request.Key, request.Bytes(), &res, "pingchannel", "/ping", tchannel.JSON, forwardOptions)
 	if handle {
-		identity, err := w.ringpop.WhoAmI()
+		address, err := w.ringpop.WhoAmI()
 		if err != nil {
 			return nil, err
 		}
-		return &Pong{"Hello, world!", identity, headers["p"]}, nil
+		return &Pong{"Hello, world!", address, headers["p"]}, nil
 	}
 
 	if err := json2.Unmarshal(res, &pong); err != nil {
@@ -111,7 +111,7 @@ func main() {
 
 	rp, err := ringpop.New("ping-app",
 		ringpop.Channel(ch),
-		ringpop.Identity(*hostport),
+		ringpop.Address(*hostport),
 		ringpop.Logger(bark.NewLoggerFromLogrus(logger)),
 	)
 	if err != nil {

examples/ping-thrift-gen/main.go

@@ -86,15 +86,15 @@ func (w *worker) RegisterPing() error {
 }
 
 func (w *worker) Ping(ctx thrift.Context, request *gen.Ping) (*gen.Pong, error) {
-	identity, err := w.ringpop.WhoAmI()
+	address, err := w.ringpop.WhoAmI()
 	if err != nil {
 		return nil, err
 	}
 	headers := ctx.Headers()
 	pHeader := headers["p"]
 	return &gen.Pong{
 		Message: "Hello, world!",
-		From_:   identity,
+		From_:   address,
 		Pheader: &pHeader,
 	}, nil
 }
@@ -111,7 +111,7 @@ func main() {
 
 	rp, err := ringpop.New("ping-app",
 		ringpop.Channel(ch),
-		ringpop.Identity(*hostport),
+		ringpop.Address(*hostport),
 		ringpop.Logger(bark.NewLoggerFromLogrus(logger)),
 	)
 	if err != nil {

examples/ping-thrift/main.go

@@ -74,14 +74,14 @@ func (w *worker) Ping(ctx thrift.Context, request *gen.Ping) (*gen.Pong, error)
 	forwardOptions := &forward.Options{Headers: marshaledHeaders.Bytes()}
 	handle, err := w.ringpop.HandleOrForward(request.Key, req, &res, "pingchannel", "PingPongService::Ping", tchannel.Thrift, forwardOptions)
 	if handle {
-		identity, err := w.ringpop.WhoAmI()
+		address, err := w.ringpop.WhoAmI()
 		if err != nil {
 			return nil, err
 		}
 		pHeader := headers["p"]
 		return &gen.Pong{
 			Message: "Hello, world!",
-			From_:   identity,
+			From_:   address,
 			Pheader: &pHeader,
 		}, nil
 	}
@@ -106,7 +106,7 @@ func main() {
 
 	rp, err := ringpop.New("ping-app",
 		ringpop.Channel(ch),
-		ringpop.Identity(*hostport),
+		ringpop.Address(*hostport),
 		ringpop.Logger(bark.NewLoggerFromLogrus(logger)),
 	)
 	if err != nil {

examples/role-labels/main.go

@@ -82,7 +82,7 @@ func main() {
 
 	rp, err := ringpop.New("ping-app",
 		ringpop.Channel(ch),
-		ringpop.Identity(*hostport),
+		ringpop.Address(*hostport),
 		ringpop.Logger(bark.NewLoggerFromLogrus(logger)),
 	)
 	if err != nil {

forward/forwarder.go

@@ -99,8 +99,8 @@ type Forwarder struct {
 func NewForwarder(s Sender, ch shared.SubChannel) *Forwarder {
 
 	logger := logging.Logger("forwarder")
-	if identity, err := s.WhoAmI(); err == nil {
-		logger = logger.WithField("local", identity)
+	if address, err := s.WhoAmI(); err == nil {
+		logger = logger.WithField("local", address)
 	}
 
 	return &Forwarder{

forward/request_sender.go

@@ -71,8 +71,8 @@ func newRequestSender(sender Sender, emitter events.EventEmitter, channel shared
 	destination, service, endpoint string, format tchannel.Format, opts *Options) *requestSender {
 
 	logger := logging.Logger("sender")
-	if identity, err := sender.WhoAmI(); err != nil {
-		logger = logger.WithField("local", identity)
+	if address, err := sender.WhoAmI(); err != nil {
+		logger = logger.WithField("local", address)
 	}
 
 	return &requestSender{
@@ -118,10 +118,10 @@ func (s *requestSender) Send() (res []byte, err error) {
 			return s.ScheduleRetry()
 		}
 
-		identity, _ := s.sender.WhoAmI()
+		address, _ := s.sender.WhoAmI()
 
 		s.logger.WithFields(log.Fields{
-			"local":       identity,
+			"local":       address,
 			"destination": s.destination,
 			"service":     s.service,
 			"endpoint":    s.endpoint,
@@ -132,10 +132,10 @@ func (s *requestSender) Send() (res []byte, err error) {
 		return nil, errors.New("max retries exceeded")
 	case <-ctx.Done(): // request timed out
 
-		identity, _ := s.sender.WhoAmI()
+		address, _ := s.sender.WhoAmI()
 
 		s.logger.WithFields(log.Fields{
-			"local":       identity,
+			"local":       address,
 			"destination": s.destination,
 			"service":     s.service,
 			"endpoint":    s.endpoint,

hashring/checksum.go

@@ -0,0 +1,67 @@
+package hashring
+
+import (
+	"bytes"
+	"sort"
+	"strconv"
+	"strings"
+
+	"github.com/dgryski/go-farm"
+)
+
+// Checksum computes a checksum for an instance of a HashRing. The
+// checksum can be used to compare two rings for equality.
+type Checksum interface {
+	// Compute calculates the checksum for the hashring that is passed in.
+	// Compute will be called while having at least a read-lock on the hashring so
+	// it is safe to read from the ring, but not safe to change the ring. There
+	// might be multiple Checksum Computes initiated at the same time, but every
+	// Checksum will only be called once per hashring at once
+	Compute(ring *HashRing) (checksum uint32)
+}
+
+// addressChecksum calculates checksums for all addresses that are added to the
+// hashring.
+type addressChecksum struct{}
+
+func (i *addressChecksum) Compute(ring *HashRing) uint32 {
+	addresses := ring.copyServersNoLock()
+	sort.Strings(addresses)
+	bytes := []byte(strings.Join(addresses, ";"))
+	return farm.Fingerprint32(bytes)
+}
+
+type identityChecksum struct{}
+
+func (i *identityChecksum) Compute(ring *HashRing) uint32 {
+	identitySet := make(map[string]struct{})
+	ring.tree.root.walk(func(node *redBlackNode) bool {
+		identitySet[node.key.(replicaPoint).identity] = struct{}{}
+		return true
+	})
+
+	identities := make([]string, 0, len(identitySet))
+	for identity := range identitySet {
+		identities = append(identities, identity)
+	}
+
+	sort.Strings(identities)
+	bytes := []byte(strings.Join(identities, ";"))
+	return farm.Fingerprint32(bytes)
+}
+
+type replicaPointChecksum struct{}
+
+func (r *replicaPointChecksum) Compute(ring *HashRing) uint32 {
+	buffer := bytes.Buffer{}
+
+	ring.tree.root.walk(func(node *redBlackNode) bool {
+		buffer.WriteString(strconv.Itoa(node.key.(replicaPoint).hash))
+		buffer.WriteString("-")
+		buffer.WriteString(node.value.(string))
+		buffer.WriteString(";")
+		return true
+	})
+
+	return farm.Fingerprint32(buffer.Bytes())
+}

hashring/checksum_test.go

@@ -0,0 +1,85 @@
+// Copyright (c) 2015 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package hashring
+
+import (
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/dgryski/go-farm"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAddressChecksum_Compute(t *testing.T) {
+	members := genMembers(1, 1, 10, false)
+	ring := New(farm.Fingerprint32, 1)
+	ring.AddMembers(members...)
+	checksum := &addressChecksum{}
+
+	addresses := make([]string, 0, 10)
+	for _, members := range members {
+		addresses = append(addresses, members.GetAddress())
+	}
+
+	sort.Strings(addresses)
+	bytes := []byte(strings.Join(addresses, ";"))
+
+	expected := farm.Fingerprint32(bytes)
+	actual := checksum.Compute(ring)
+
+	assert.Equal(t, expected, actual)
+}
+
+func TestIdentityChecksum_Compute(t *testing.T) {
+	identityChecksummer := &identityChecksum{}
+
+	ringWithoutIdentities := New(farm.Fingerprint32, 1)
+	ringWithoutIdentities.AddMembers(genMembers(1, 1, 10, false)...)
+
+	legacyChecksum := (&addressChecksum{}).Compute(ringWithoutIdentities)
+	identityChecksum := identityChecksummer.Compute(ringWithoutIdentities)
+
+	assert.Equal(t, legacyChecksum, identityChecksum, "Identity checksum should be the same as legacy on ring without identities")
+
+	ringWithIdentities := New(farm.Fingerprint32, 1)
+	ringWithIdentities.AddMembers(genMembers(1, 1, 10, true)...)
+
+	identityChecksum = identityChecksummer.Compute(ringWithIdentities)
+
+	assert.NotEqual(t, legacyChecksum, identityChecksum, "IdentityChecksummer should not match legacy checksummer on ring with identites ")
+}
+
+func TestReplicaPointChecksum_Compute(t *testing.T) {
+	replicaPointChecksummer := &replicaPointChecksum{}
+	members := genMembers(1, 1, 10, false)
+
+	ring1ReplicaPoint := New(farm.Fingerprint32, 1)
+	ring1ReplicaPoint.AddMembers(members...)
+
+	ring2ReplicaPoints := New(farm.Fingerprint32, 2)
+	ring2ReplicaPoints.AddMembers(members...)
+
+	checksum1ReplicaPoint := replicaPointChecksummer.Compute(ring1ReplicaPoint)
+	checksum2ReplicaPoints := replicaPointChecksummer.Compute(ring2ReplicaPoints)
+
+	assert.NotEqual(t, checksum1ReplicaPoint, checksum2ReplicaPoints, "Checksum should not match with different replica point counts")
+}