fix: sharder should use peer identity from Peers package

app/app_test.go

@@ -327,19 +327,18 @@ func TestPeerRouting(t *testing.T) {
 	// Parallel integration tests need different ports!
 	t.Parallel()
 
-	peers := &peer.MockPeers{
-		Peers: []string{
-			"http://localhost:11001",
-			"http://localhost:11003",
-		},
-	}
+	peerList := []string{"http://localhost:11001", "http://localhost:11003"}
 
 	var apps [2]*App
 	var senders [2]*transmission.MockSender
 	for i := range apps {
 		var graph inject.Graph
 		basePort := 11000 + (i * 2)
 		senders[i] = &transmission.MockSender{}
+		peers := &peer.MockPeers{
+			Peers: peerList,
+			ID:    peerList[i],
+		}
 		apps[i], graph = newStartedApp(t, senders[i], basePort, peers, false)
 		defer startstop.Stop(graph.Objects(), nil)
 	}
@@ -457,11 +456,9 @@ func TestHostMetadataSpanAdditions(t *testing.T) {
 func TestEventsEndpoint(t *testing.T) {
 	t.Parallel()
 
-	peers := &peer.MockPeers{
-		Peers: []string{
-			"http://localhost:13001",
-			"http://localhost:13003",
-		},
+	peerList := []string{
+		"http://localhost:13001",
+		"http://localhost:13003",
 	}
 
 	var apps [2]*App
@@ -470,6 +467,11 @@ func TestEventsEndpoint(t *testing.T) {
 		var graph inject.Graph
 		basePort := 13000 + (i * 2)
 		senders[i] = &transmission.MockSender{}
+		peers := &peer.MockPeers{
+			Peers: peerList,
+			ID:    peerList[i],
+		}
+
 		apps[i], graph = newStartedApp(t, senders[i], basePort, peers, false)
 		defer startstop.Stop(graph.Objects(), nil)
 	}
@@ -570,18 +572,21 @@ func TestEventsEndpoint(t *testing.T) {
 func TestEventsEndpointWithNonLegacyKey(t *testing.T) {
 	t.Parallel()
 
-	peers := &peer.MockPeers{
-		Peers: []string{
-			"http://localhost:15001",
-			"http://localhost:15003",
-		},
+	peerList := []string{
+		"http://localhost:15001",
+		"http://localhost:15003",
 	}
 
 	var apps [2]*App
 	var senders [2]*transmission.MockSender
 	for i := range apps {
 		basePort := 15000 + (i * 2)
 		senders[i] = &transmission.MockSender{}
+		peers := &peer.MockPeers{
+			Peers: peerList,
+			ID:    peerList[i],
+		}
+
 		app, graph := newStartedApp(t, senders[i], basePort, peers, false)
 		app.IncomingRouter.SetEnvironmentCache(time.Second, func(s string) (string, error) { return "test", nil })
 		app.PeerRouter.SetEnvironmentCache(time.Second, func(s string) (string, error) { return "test", nil })
@@ -830,21 +835,24 @@ func BenchmarkDistributedTraces(b *testing.B) {
 		},
 	}
 
-	peers := &peer.MockPeers{
-		Peers: []string{
-			"http://localhost:12001",
-			"http://localhost:12003",
-			"http://localhost:12005",
-			"http://localhost:12007",
-			"http://localhost:12009",
-		},
+	peerList := []string{
+		"http://localhost:12001",
+		"http://localhost:12003",
+		"http://localhost:12005",
+		"http://localhost:12007",
+		"http://localhost:12009",
 	}
 
 	var apps [5]*App
 	var addrs [5]string
 	for i := range apps {
 		var graph inject.Graph
 		basePort := 12000 + (i * 2)
+		peers := &peer.MockPeers{
+			Peers: peerList,
+			ID:    peerList[i],
+		}
+
 		apps[i], graph = newStartedApp(b, sender, basePort, peers, false)
 		defer startstop.Stop(graph.Objects(), nil)
 

collect/stressRelief.go

@@ -199,15 +199,15 @@ func newStressReliefMessage(level uint, peerID string) *stressReliefMessage {
 }
 
 func (msg *stressReliefMessage) String() string {
-	return msg.peerID + ":" + fmt.Sprint(msg.level)
+	return msg.peerID + "|" + fmt.Sprint(msg.level)
 }
 
 func unmarshalStressReliefMessage(msg string) (*stressReliefMessage, error) {
 	if len(msg) < 2 {
 		return nil, fmt.Errorf("empty message")
 	}
 
-	parts := strings.SplitN(msg, ":", 2)
+	parts := strings.SplitN(msg, "|", 2)
 	level, err := strconv.Atoi(parts[1])
 	if err != nil {
 		return nil, err

internal/peer/file.go

@@ -1,6 +1,7 @@
 package peer
 
 import (
+	"fmt"
 	"net"
 
 	"github.com/honeycombio/refinery/config"
@@ -22,7 +23,12 @@ func (p *FilePeers) GetPeers() ([]string, error) {
 	// logic happy.
 	peers, err := p.Cfg.GetPeers()
 	if len(peers) == 0 {
-		peers = []string{"http://127.0.0.1:8081"}
+		addr, err := p.publicAddr()
+		if err != nil {
+			return nil, err
+
+		}
+		peers = []string{addr}
 	}
 	p.Metrics.Gauge("num_file_peers", float64(len(peers)))
 	return peers, err
@@ -54,14 +60,14 @@ func (p *FilePeers) Stop() error {
 }
 
 func (p *FilePeers) publicAddr() (string, error) {
-	addr, err := p.Cfg.GetListenAddr()
+	addr, err := p.Cfg.GetPeerListenAddr()
 	if err != nil {
 		return "", err
 	}
-	host, _, err := net.SplitHostPort(addr)
+	host, port, err := net.SplitHostPort(addr)
 	if err != nil {
 		return "", err
 	}
 
-	return host, nil
+	return fmt.Sprintf("http://%s:%s", host, port), nil
 }

internal/peer/file_test.go

@@ -10,9 +10,9 @@ func TestFilePeers(t *testing.T) {
 	peers := []string{"peer"}
 
 	c := &config.MockConfig{
-		PeerManagementType: "file",
-		GetPeersVal:        peers,
-		GetListenAddrVal:   "10.244.0.114:8080",
+		PeerManagementType:   "file",
+		GetPeersVal:          peers,
+		GetPeerListenAddrVal: "10.244.0.114:8081",
 	}
 	p, err := newPeers(c)
 	if err != nil {

internal/peer/mock.go

@@ -1,10 +1,5 @@
 package peer
 
-import (
-	"math/rand"
-	"strconv"
-)
-
 var _ Peers = (*MockPeers)(nil)
 
 type MockPeers struct {
@@ -25,8 +20,8 @@ func (p *MockPeers) RegisterUpdatedPeersCallback(callback func()) {
 }
 
 func (p *MockPeers) Start() error {
-	if len(p.ID) == 0 {
-		p.ID = strconv.Itoa(rand.Intn(10000))
+	if len(p.ID) == 0 && len(p.Peers) > 0 {
+		p.ID = p.Peers[0]
 	}
 	return nil
 }

internal/peer/pubsub_redis.go

@@ -197,7 +197,7 @@ func (p *RedisPubsubPeers) GetPeers() ([]string, error) {
 }
 
 func (p *RedisPubsubPeers) GetInstanceID() (string, error) {
-	return p.getIdentifierFromInterface()
+	return p.publicAddr()
 }
 
 func (p *RedisPubsubPeers) RegisterUpdatedPeersCallback(callback func()) {