feat: Implement peering with pubsub

app/app_test.go

@@ -85,17 +85,6 @@ func (w *countingWriterSender) waitForCount(t testing.TB, target int) {
 	}
 }
 
-type testPeers struct {
-	peers []string
-}
-
-func (p *testPeers) GetPeers() ([]string, error) {
-	return p.peers, nil
-}
-
-func (p *testPeers) RegisterUpdatedPeersCallback(callback func()) {
-}
-
 func newStartedApp(
 	t testing.TB,
 	libhoneyT transmission.Sender,
@@ -331,8 +320,8 @@ func TestPeerRouting(t *testing.T) {
 	// Parallel integration tests need different ports!
 	t.Parallel()
 
-	peers := &testPeers{
-		peers: []string{
+	peers := &peer.MockPeers{
+		Peers: []string{
 			"http://localhost:11001",
 			"http://localhost:11003",
 		},
@@ -464,8 +453,8 @@ func TestHostMetadataSpanAdditions(t *testing.T) {
 func TestEventsEndpoint(t *testing.T) {
 	t.Parallel()
 
-	peers := &testPeers{
-		peers: []string{
+	peers := &peer.MockPeers{
+		Peers: []string{
 			"http://localhost:13001",
 			"http://localhost:13003",
 		},
@@ -580,8 +569,8 @@ func TestEventsEndpoint(t *testing.T) {
 func TestEventsEndpointWithNonLegacyKey(t *testing.T) {
 	t.Parallel()
 
-	peers := &testPeers{
-		peers: []string{
+	peers := &peer.MockPeers{
+		Peers: []string{
 			"http://localhost:15001",
 			"http://localhost:15003",
 		},
@@ -843,8 +832,8 @@ func BenchmarkDistributedTraces(b *testing.B) {
 		},
 	}
 
-	peers := &testPeers{
-		peers: []string{
+	peers := &peer.MockPeers{
+		Peers: []string{
 			"http://localhost:12001",
 			"http://localhost:12003",
 			"http://localhost:12005",

internal/peer/file.go

@@ -29,3 +29,11 @@ func (p *filePeers) RegisterUpdatedPeersCallback(callback func()) {
 	// otherwise do nothing since they never change
 	callback()
 }
+
+func (p *filePeers) Start() error {
+	return nil
+}
+
+func (p *filePeers) Stop() error {
+	return nil
+}

internal/peer/mock.go

@@ -4,9 +4,21 @@ type MockPeers struct {
 	Peers []string
 }
 
+// ensure that MockPeers implements the Peers interface
+var _ Peers = (*MockPeers)(nil)
+
 func (p *MockPeers) GetPeers() ([]string, error) {
 	return p.Peers, nil
 }
+
 func (p *MockPeers) RegisterUpdatedPeersCallback(callback func()) {
 	callback()
 }
+
+func (p *MockPeers) Start() error {
+	return nil
+}
+
+func (p *MockPeers) Stop() error {
+	return nil
+}

internal/peer/peers.go

@@ -4,14 +4,17 @@ import (
 	"context"
 	"errors"
 
+	"github.com/facebookgo/startstop"
 	"github.com/honeycombio/refinery/config"
 )
 
 // Peers holds the collection of peers for the cluster
 type Peers interface {
 	GetPeers() ([]string, error)
-
 	RegisterUpdatedPeersCallback(callback func())
+	// make it injectable
+	startstop.Starter
+	startstop.Stopper
 }
 
 func NewPeers(ctx context.Context, c config.Config, done chan struct{}) (Peers, error) {
@@ -26,6 +29,8 @@ func NewPeers(ctx context.Context, c config.Config, done chan struct{}) (Peers,
 		return newFilePeers(c), nil
 	case "redis":
 		return newRedisPeers(ctx, c, done)
+	case "pubsub":
+		return newPubsubPeers(c)
 	default:
 		return nil, errors.New("invalid config option 'PeerManagement.Type'")
 	}

internal/peer/peers_test.go

@@ -7,11 +7,12 @@ import (
 	"time"
 
 	"github.com/honeycombio/refinery/config"
+	"github.com/honeycombio/refinery/pubsub"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-func TestNewPeers(t *testing.T) {
+func TestNewPeersFile(t *testing.T) {
 	c := &config.MockConfig{
 		PeerManagementType: "file",
 		PeerTimeout:        5 * time.Second,
@@ -30,21 +31,32 @@ func TestNewPeers(t *testing.T) {
 	default:
 		t.Errorf("received %T expected %T", i, &filePeers{})
 	}
+}
 
-	c = &config.MockConfig{
-		GetPeerListenAddrVal: "0.0.0.0:8081",
-		PeerManagementType:   "redis",
-		PeerTimeout:          5 * time.Second,
+func TestNewPeersPubSub(t *testing.T) {
+	c := &config.MockConfig{
+		GetPeerListenAddrVal:    "0.0.0.0:8081",
+		PeerManagementType:      "pubsub",
+		PeerTimeout:             5 * time.Second,
+		IdentifierInterfaceName: "eth0",
 	}
 
-	p, err = NewPeers(context.Background(), c, done)
+	pubsub := pubsub.LocalPubSub{}
+	pubsub.Start()
+
+	done := make(chan struct{})
+	defer close(done)
+	p, err := NewPeers(context.Background(), c, done)
 	assert.NoError(t, err)
 	require.NotNil(t, p)
+	p.(*pubsubPeers).PubSub = &pubsub
+	p.Start()
+	defer p.Stop()
 
 	switch i := p.(type) {
-	case *redisPeers:
+	case *pubsubPeers:
 	default:
-		t.Errorf("received %T expected %T", i, &redisPeers{})
+		t.Errorf("received %T expected %T", i, &pubsubPeers{})
 	}
 }