[events] Support ephemeral consumers (#184)

[loadbalancer-manager-haproxy](https://github.com/infratographer/loadbalancer-manager-haproxy) does not have a need for durable message queues, as we want each instance to consume all messages received related to loadbalancers. Currently, we generate a unique durable queue for each manager instance, which eventually puts strain on nats with the number of consumers that can be generated over time. This PR aims to put support for ephemeral consumers back into event subscriptions, if requested, by leaving `QueueGroup` empty in the nats config. --------- Signed-off-by: Matt Siwiec <[email protected]>
infratographer · Jan 22, 2024 · a5142b6 · a5142b6
1 parent 4fc0e32
commit a5142b6
Show file tree

Hide file tree

Showing 4 changed files with 72 additions and 39 deletions.
diff --git a/events/nats_config.go b/events/nats_config.go
@@ -55,7 +55,7 @@ type NATSConfig struct {
 
 // Configured checks whether the provider has been configured.
 func (c NATSConfig) Configured() bool {
-	return c.URL != "" || c.QueueGroup != ""
+	return c.URL != ""
 }
 
 // Validate ensures the configuration is valid.

diff --git a/events/nats_connection.go b/events/nats_connection.go
@@ -121,6 +121,10 @@ func NewNATSConnection(config NATSConfig, options ...NATSOption) (*NATSConnectio
 		}
 	}
 
+	if config.QueueGroup == "" {
+		nc.logger.Warn("NATS QueueGroup is not set. Subscriptions will not be durable.")
+	}
+
 	conn, err := nats.Connect(config.URL, nc.connectOptions...)
 	if err != nil {
 		return nil, err
@@ -143,7 +147,12 @@ func NewNATSConnection(config NATSConfig, options ...NATSOption) (*NATSConnectio
 }
 
 // NATSConsumerDurableName is the generator function to create a new durable consumer name.
+// If queueGroup is empty, an empty durable name is returned to support ephemeral consumers.
 func NATSConsumerDurableName(queueGroup, subject string) string {
+	if queueGroup == "" {
+		return ""
+	}
+
 	hash := md5.Sum([]byte(subject))
 
 	return queueGroup + hex.EncodeToString(hash[:])

diff --git a/events/nats_test.go b/events/nats_test.go
@@ -20,55 +20,76 @@ var errTimeout = errors.New("timeout waiting for event")
 
 func TestNATSPublishAndSubscribe(t *testing.T) {
 	ctx := context.Background()
-	nats, err := eventtools.NewNatsServer()
-	require.NoError(t, err)
 
-	defer nats.Close()
+	testCases := []struct {
+		name       string
+		queueGroup string
+	}{
+		{
+			name:       "with ephemeral consumer",
+			queueGroup: "",
+		},
+		{
+			name:       "with durable consumer",
+			queueGroup: "testing-durable",
+		},
+	}
 
-	conn, err := events.NewNATSConnection(nats.Config.NATS)
-	require.NoError(t, err)
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			nats, err := eventtools.NewNatsServer()
+			require.NoError(t, err)
 
-	defer conn.Shutdown(ctx) //nolint:errcheck // within test
+			defer nats.Close()
 
-	change := testCreateChange()
+			natsCfg := nats.Config.NATS
+			natsCfg.QueueGroup = tc.queueGroup
+			conn, err := events.NewNATSConnection(natsCfg)
+			require.NoError(t, err)
 
-	msg, err := conn.PublishChange(ctx, "test", change)
-	require.NoError(t, err)
-	require.Equal(t, change, msg.Message())
+			defer conn.Shutdown(ctx) //nolint:errcheck // within test
 
-	change2 := testCreateChange()
+			change := testCreateChange()
 
-	msg, err = conn.PublishChange(ctx, "test", change2)
-	require.NoError(t, err)
-	require.Equal(t, change2, msg.Message())
+			msg, err := conn.PublishChange(ctx, "test", change)
+			require.NoError(t, err)
+			require.Equal(t, change, msg.Message())
 
-	change3 := testCreateChange()
-	change3.ActorID = ""
+			change2 := testCreateChange()
 
-	msg, err = conn.PublishChange(ctx, "test", change3)
-	require.NoError(t, err)
-	require.NotEqual(t, change3, msg.Message())
+			msg, err = conn.PublishChange(ctx, "test", change2)
+			require.NoError(t, err)
+			require.Equal(t, change2, msg.Message())
 
-	messages, err := conn.SubscribeChanges(ctx, ">")
-	require.NoError(t, err)
+			change3 := testCreateChange()
+			change3.ActorID = ""
 
-	receivedMsg, err := getSingleMessage(messages, time.Second*1)
-	require.NoError(t, err)
-	require.NoError(t, receivedMsg.Error())
-	assert.EqualValues(t, change, receivedMsg.Message())
+			msg, err = conn.PublishChange(ctx, "test", change3)
+			require.NoError(t, err)
+			require.NotEqual(t, change3, msg.Message())
 
-	receivedMsg, err = getSingleMessage(messages, time.Second*1)
-	require.NoError(t, err)
-	require.NoError(t, receivedMsg.Error())
-	assert.EqualValues(t, change2, receivedMsg.Message())
-	assert.NoError(t, receivedMsg.Ack())
+			messages, err := conn.SubscribeChanges(ctx, ">")
+			require.NoError(t, err)
 
-	receivedMsg, err = getSingleMessage(messages, time.Second*1)
-	require.NoError(t, err)
-	require.NoError(t, receivedMsg.Error())
-	assert.NotEqualValues(t, change3, receivedMsg.Message())
-	assert.Equal(t, "unknown-actor", receivedMsg.Message().ActorID.String())
-	assert.NoError(t, receivedMsg.Ack())
+			receivedMsg, err := getSingleMessage(messages, time.Second*1)
+			require.NoError(t, err)
+			require.NoError(t, receivedMsg.Error())
+			assert.EqualValues(t, change, receivedMsg.Message())
+
+			receivedMsg, err = getSingleMessage(messages, time.Second*1)
+			require.NoError(t, err)
+			require.NoError(t, receivedMsg.Error())
+			assert.EqualValues(t, change2, receivedMsg.Message())
+			assert.NoError(t, receivedMsg.Ack())
+
+			receivedMsg, err = getSingleMessage(messages, time.Second*1)
+			require.NoError(t, err)
+			require.NoError(t, receivedMsg.Error())
+			assert.NotEqualValues(t, change3, receivedMsg.Message())
+			assert.Equal(t, "unknown-actor", receivedMsg.Message().ActorID.String())
+			assert.NoError(t, receivedMsg.Ack())
+		})
+	}
 }
 
 func TestNATSRequestReply(t *testing.T) {

diff --git a/testing/eventtools/nats_test.go b/testing/eventtools/nats_test.go
@@ -20,14 +20,17 @@ var errTimeout = errors.New("timeout waiting for event")
 func TestNats(t *testing.T) {
 	ctx := context.Background()
 
-	consumerName := events.NATSConsumerDurableName("", eventtools.Prefix+".changes.>")
+	consumerName := events.NATSConsumerDurableName("nats-testing", eventtools.Prefix+".changes.>")
 
 	nats, err := eventtools.NewNatsServer()
 	require.NoError(t, err)
 
 	defer nats.Close()
 
-	conn, err := events.NewNATSConnection(nats.Config.NATS)
+	natsCfg := nats.Config.NATS
+	natsCfg.QueueGroup = "nats-testing"
+
+	conn, err := events.NewNATSConnection(natsCfg)
 	require.NoError(t, err)
 
 	change1 := testCreateChange()