Add NATS publisher support to reminder (#4829)

* Minor cleanup work Signed-off-by: Vyom Yadav <[email protected]> * Add NATS publisher support to reminder Signed-off-by: Vyom Yadav <[email protected]> --------- Signed-off-by: Vyom Yadav <[email protected]>
mindersec · Jan 3, 2025 · a6d9b22 · a6d9b22
1 parent 04ac00a
commit a6d9b22
Show file tree

Hide file tree

Showing 8 changed files with 109 additions and 98 deletions.
diff --git a/config/reminder-config.yaml.example b/config/reminder-config.yaml.example
@@ -18,10 +18,18 @@ logging:
   level: "debug"
 
 events:
-  sql_connection:
-    dbhost: "watermill-postgres"
-    dbport: 5432
-    dbuser: postgres
-    dbpass: postgres
-    dbname: watermill
-    sslmode: disable
+  driver: "sql"
+# only sql and cloudevents-nats drivers are supported
+#  driver: "cloudevents-nats"
+  sql:
+    connection:
+      dbhost: "watermill-postgres"
+      dbport: 5432
+      dbuser: postgres
+      dbpass: postgres
+      dbname: watermill
+      sslmode: disable
+#  nats:
+#    url: "nats://nats:4222"
+#    prefix: "minder"
+#    queue: "minder"
diff --git a/internal/events/nats/natschannel.go b/internal/events/nats/natschannel.go
@@ -1,7 +1,7 @@
 // SPDX-FileCopyrightText: Copyright 2024 The Minder Authors
 // SPDX-License-Identifier: Apache-2.0
 
-// Package nats provides a nants+cloudevents implementation of the eventer interface
+// Package nats provides a nats+cloudevents implementation of the eventer interface
 package nats
 
 import (
@@ -17,6 +17,7 @@ import (
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/client"
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/rs/zerolog"
 
 	"github.com/mindersec/minder/internal/events/common"
@@ -114,25 +115,21 @@ func (c *cloudEventsNatsAdapter) ensureTopic(ctx context.Context, topic string,
 	return &state, nil
 }
 
-func (c *cloudEventsNatsAdapter) ensureStream(_ context.Context) error {
+func (c *cloudEventsNatsAdapter) ensureStream(ctx context.Context) error {
 	conn, err := nats.Connect(c.cfg.URL)
 	if err != nil {
 		return err
 	}
 	defer conn.Close()
-	js, err := conn.JetStream()
+	js, err := jetstream.New(conn)
 	if err != nil {
 		return err
 	}
-	si, err := js.StreamInfo(c.cfg.Prefix)
-	if si == nil || err != nil && err.Error() == "stream not found" {
-		_, err = js.AddStream(&nats.StreamConfig{
-			Name:     c.cfg.Prefix,
-			Subjects: []string{c.cfg.Prefix + ".>"},
-		})
-		return err
-	}
-	return nil
+	_, err = js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
+		Name:     c.cfg.Prefix,
+		Subjects: []string{c.cfg.Prefix + ".>"},
+	})
+	return err
 }
 
 // Subscribe implements message.Subscriber.
@@ -191,13 +188,13 @@ func (c *cloudEventsNatsAdapter) Publish(topic string, messages ...*message.Mess
 
 	state, err := c.ensureTopic(ctx, subject, "sender")
 	if err != nil {
-		return fmt.Errorf("Error creating topic %q: %w", subject, err)
+		return fmt.Errorf("error creating topic %q: %w", subject, err)
 	}
 
 	for _, msg := range messages {
 		err := sendEvent(ctx, subject, state.ceClient, msg)
 		if err != nil {
-			return fmt.Errorf("Error sending event to %q: %w", subject, err)
+			return fmt.Errorf("error sending event to %q: %w", subject, err)
 		}
 	}
 	return nil
@@ -214,7 +211,7 @@ func sendEvent(
 	// All our current payloads are encoded JSON; we need to unmarshal
 	payload := map[string]any{}
 	if err := json.Unmarshal(msg.Payload, &payload); err != nil {
-		return fmt.Errorf("Error unmarshalling payload: %w", err)
+		return fmt.Errorf("error unmarshalling payload: %w", err)
 	}
 
 	err := event.SetData("application/json", payload)

diff --git a/internal/reminder/publisher.go b/internal/reminder/publisher.go
@@ -0,0 +1,22 @@
+// SPDX-FileCopyrightText: Copyright 2024 The Minder Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package reminder
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/ThreeDotsLabs/watermill/message"
+
+	"github.com/mindersec/minder/pkg/eventer"
+)
+
+func (r *reminder) getMessagePublisher(ctx context.Context) (message.Publisher, error) {
+	pub, err := eventer.New(ctx, nil, &r.cfg.EventConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create publisher: %w", err)
+	}
+
+	return pub, nil
+}
diff --git a/internal/reminder/reminder.go b/internal/reminder/reminder.go
@@ -16,7 +16,6 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/mindersec/minder/internal/db"
-	"github.com/mindersec/minder/internal/events/common"
 	remindermessages "github.com/mindersec/minder/internal/reminder/messages"
 	reminderconfig "github.com/mindersec/minder/pkg/config/reminder"
 	"github.com/mindersec/minder/pkg/eventer/constants"
@@ -43,7 +42,6 @@ type reminder struct {
 	ticker *time.Ticker
 
 	eventPublisher message.Publisher
-	eventDBCloser  common.DriverCloser
 }
 
 // NewReminder creates a new reminder instance
@@ -59,13 +57,12 @@ func NewReminder(ctx context.Context, store db.Store, config *reminderconfig.Con
 	logger := zerolog.Ctx(ctx)
 	logger.Info().Msgf("initial repository cursor: %s", r.repositoryCursor)
 
-	pub, cl, err := r.setupSQLPublisher(ctx)
+	pub, err := r.getMessagePublisher(ctx)
 	if err != nil {
 		return nil, err
 	}
 
 	r.eventPublisher = pub
-	r.eventDBCloser = cl
 	return r, nil
 }
 
@@ -118,7 +115,10 @@ func (r *reminder) Stop() {
 	}
 	r.stopOnce.Do(func() {
 		close(r.stop)
-		r.eventDBCloser()
+		err := r.eventPublisher.Close()
+		if err != nil {
+			zerolog.Ctx(context.Background()).Error().Err(err).Msg("error closing event publisher")
+		}
 	})
 }
 

diff --git a/internal/reminder/sql_publisher.go b/internal/reminder/sql_publisher.go
diff --git a/pkg/config/reminder/config.go b/pkg/config/reminder/config.go
@@ -5,20 +5,23 @@
 package reminder
 
 import (
+	"fmt"
 	"strings"
 
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
 
 	"github.com/mindersec/minder/pkg/config"
+	serverconfig "github.com/mindersec/minder/pkg/config/server"
+	"github.com/mindersec/minder/pkg/eventer/constants"
 )
 
 // Config contains the configuration for the reminder service
 type Config struct {
-	Database         config.DatabaseConfig `mapstructure:"database"`
-	RecurrenceConfig RecurrenceConfig      `mapstructure:"recurrence"`
-	EventConfig      EventConfig           `mapstructure:"events"`
-	LoggingConfig    LoggingConfig         `mapstructure:"logging"`
+	Database         config.DatabaseConfig    `mapstructure:"database"`
+	RecurrenceConfig RecurrenceConfig         `mapstructure:"recurrence"`
+	EventConfig      serverconfig.EventConfig `mapstructure:"events"`
+	LoggingConfig    LoggingConfig            `mapstructure:"logging"`
 }
 
 // Validate validates the configuration
@@ -28,6 +31,11 @@ func (c Config) Validate() error {
 		return err
 	}
 
+	err = validateEventConfig(c.EventConfig)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -52,3 +60,14 @@ func RegisterReminderFlags(v *viper.Viper, flags *pflag.FlagSet) error {
 
 	return registerRecurrenceFlags(v, flags)
 }
+
+func validateEventConfig(cfg serverconfig.EventConfig) error {
+	switch cfg.Driver {
+	case constants.NATSDriver:
+	case constants.SQLDriver:
+	default:
+		return fmt.Errorf("events.driver %s is not supported", cfg.Driver)
+	}
+
+	return nil
+}
diff --git a/pkg/config/reminder/config_test.go b/pkg/config/reminder/config_test.go
@@ -5,6 +5,7 @@ package reminder_test
 
 import (
 	"bytes"
+	"fmt"
 	"os"
 	"testing"
 	"time"
@@ -16,6 +17,8 @@ import (
 
 	"github.com/mindersec/minder/pkg/config"
 	"github.com/mindersec/minder/pkg/config/reminder"
+	serverconfig "github.com/mindersec/minder/pkg/config/server"
+	"github.com/mindersec/minder/pkg/eventer/constants"
 )
 
 func TestValidateConfig(t *testing.T) {
@@ -34,9 +37,12 @@ func TestValidateConfig(t *testing.T) {
 					BatchSize:  100,
 					MinElapsed: parseTimeDuration(t, "1h"),
 				},
-				EventConfig: reminder.EventConfig{
-					Connection: config.DatabaseConfig{
-						Port: 8080,
+				EventConfig: serverconfig.EventConfig{
+					Driver: constants.SQLDriver,
+					SQLPubSub: serverconfig.SQLEventConfig{
+						Connection: config.DatabaseConfig{
+							Port: 8080,
+						},
 					},
 				},
 			},
@@ -49,6 +55,9 @@ func TestValidateConfig(t *testing.T) {
 					BatchSize:  100,
 					MinElapsed: parseTimeDuration(t, "1h"),
 				},
+				EventConfig: serverconfig.EventConfig{
+					Driver: constants.SQLDriver,
+				},
 			},
 			errMsg: "cannot be negative",
 		},
@@ -60,9 +69,26 @@ func TestValidateConfig(t *testing.T) {
 					BatchSize:  100,
 					MinElapsed: parseTimeDuration(t, "-1h"),
 				},
+				EventConfig: serverconfig.EventConfig{
+					Driver: constants.SQLDriver,
+				},
 			},
 			errMsg: "cannot be negative",
 		},
+		{
+			name: "UnsupportedDriver",
+			config: reminder.Config{
+				RecurrenceConfig: reminder.RecurrenceConfig{
+					Interval:   parseTimeDuration(t, "1h"),
+					BatchSize:  100,
+					MinElapsed: parseTimeDuration(t, "1h"),
+				},
+				EventConfig: serverconfig.EventConfig{
+					Driver: constants.GoChannelDriver,
+				},
+			},
+			errMsg: fmt.Sprintf("%s is not supported", constants.GoChannelDriver),
+		},
 	}
 
 	for _, tt := range tests {
@@ -153,10 +179,9 @@ func TestSetViperDefaults(t *testing.T) {
 	require.Equal(t, parseTimeDuration(t, "1h"), parseTimeDuration(t, v.GetString("recurrence.interval")))
 	require.Equal(t, 100, v.GetInt("recurrence.batch_size"))
 	require.Equal(t, parseTimeDuration(t, "1h"), parseTimeDuration(t, v.GetString("recurrence.min_elapsed")))
-	require.Equal(t, "reminder", v.GetString("events.sql_connection.dbname"))
-	require.Equal(t, "reminder-event-postgres", v.GetString("events.sql_connection.dbhost"))
-	require.Equal(t, "reminder-event-postgres", v.GetString("events.sql_connection.dbhost"))
-	require.Equal(t, "postgres", v.GetString("events.sql_connection.dbuser"))
+	require.Equal(t, "watermill", v.GetString("events.sql.connection.dbname"))
+	require.Equal(t, "localhost", v.GetString("events.sql.connection.dbhost"))
+	require.Equal(t, "postgres", v.GetString("events.sql.connection.dbuser"))
 }
 
 // TestOverrideConfigByEnvVar tests that the configuration can be overridden by environment variables

diff --git a/pkg/config/reminder/events.go b/pkg/config/reminder/events.go