Repository: redpanda-data/connect (language: go)

Issue:
Allow redundant URLs for AMQP_1
We'd like the same option as amqp_0_9 (see #912 ) for amqp_1 regarding specifying multiple urls for our amqp-clusters
(our prio is the input)


Discussion / hints:

Fix patch:
diff --git a/internal/impl/amqp1/config.go b/internal/impl/amqp1/config.go
index 160fdc61f9..fa2c6d2c1f 100644
--- a/internal/impl/amqp1/config.go
+++ b/internal/impl/amqp1/config.go
@@ -11,6 +11,7 @@ import (
 const (
 	// Shared
 	urlField      = "url"
+	urlsField     = "urls"
 	tlsField      = "tls"
 	saslField     = "sasl"
 	saslMechField = "mechanism"
diff --git a/internal/impl/amqp1/input.go b/internal/impl/amqp1/input.go
index aacf1cd291..4eb4ea9c0e 100644
--- a/internal/impl/amqp1/input.go
+++ b/internal/impl/amqp1/input.go
@@ -38,7 +38,15 @@ You can access these metadata fields using [function interpolation](/docs/config
 			service.NewURLField(urlField).
 				Description("A URL to connect to.").
 				Example("amqp://localhost:5672/").
-				Example("amqps://guest:guest@localhost:5672/"),
+				Example("amqps://guest:guest@localhost:5672/").
+				Optional(),
+			service.NewURLListField(urlsField).
+				Description("A list of URLs to connect to. The first URL to successfully establish a connection will be used until the connection is closed. If an item of the list contains commas it will be expanded into multiple URLs.").
+				Example([]string{"amqp://guest:guest@127.0.0.1:5672/"}).
+				Example([]string{"amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/"}).
+				Example([]string{"amqp://127.0.0.1:5672/", "amqp://127.0.0.2:5672/"}).
+				Optional().
+				Version("T.B.D"),
 			service.NewStringField(sourceAddrField).
 				Description("The source address to consume from.").
 				Example("/foo").
@@ -67,7 +75,7 @@ func init() {
 //------------------------------------------------------------------------------
 
 type amqp1Reader struct {
-	url        string
+	urls       []string
 	sourceAddr string
 	renewLock  bool
 	connOpts   *amqp.ConnOptions
@@ -83,11 +91,29 @@ func amqp1ReaderFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (
 		connOpts: &amqp.ConnOptions{},
 	}
 
-	var err error
-	if a.url, err = conf.FieldString(urlField); err != nil {
+	urlStrs, err := conf.FieldStringList(urlsField)
+	if err != nil {
 		return nil, err
 	}
 
+	for _, u := range urlStrs {
+		for _, splitURL := range strings.Split(u, ",") {
+			if trimmed := strings.TrimSpace(splitURL); len(trimmed) > 0 {
+				a.urls = append(a.urls, trimmed)
+			}
+		}
+	}
+
+	if len(a.urls) == 0 {
+		singleURL, err := conf.FieldString(urlField)
+		if err != nil {
+			return nil, err
+		}
+
+		a.urls = append(a.urls, singleURL)
+
+	}
+
 	if a.sourceAddr, err = conf.FieldString(sourceAddrField); err != nil {
 		return nil, err
 	}
@@ -125,8 +151,8 @@ func (a *amqp1Reader) Connect(ctx context.Context) (err error) {
 	}
 
 	// Create client
-	if conn.client, err = amqp.Dial(ctx, a.url, a.connOpts); err != nil {
-		return
+	if conn.client, err = a.reDial(ctx, a.urls); err != nil {
+		return err
 	}
 
 	// Open a session
@@ -249,6 +275,26 @@ func (a *amqp1Reader) Close(ctx context.Context) error {
 	return a.disconnect(ctx)
 }
 
+// reDial connection to amqp with one or more fallback URLs.
+func (a *amqp1Reader) reDial(ctx context.Context, urls []string) (conn *amqp.Conn, err error) {
+	for i, url := range urls {
+		conn, err = amqp.Dial(ctx, url, a.connOpts)
+		if err != nil {
+			a.log.With("error", err).Warnf("unable to connect to url %q #%d, trying next", url, i)
+
+			continue
+		}
+
+		a.log.Tracef("successful connection to use %q #%d", url, i)
+
+		return conn, nil
+	}
+
+	a.log.With("error", err).Tracef("unable to connect to any of %d urls, return error", len(a.urls))
+
+	return nil, err
+}
+
 //------------------------------------------------------------------------------
 
 type amqp1Conn struct {
diff --git a/internal/impl/amqp1/output.go b/internal/impl/amqp1/output.go
index b41fc9b1be..a7a4e4aa2f 100644
--- a/internal/impl/amqp1/output.go
+++ b/internal/impl/amqp1/output.go
@@ -3,6 +3,7 @@ package amqp1
 import (
 	"context"
 	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/Azure/go-amqp"
@@ -29,7 +30,15 @@ This output benefits from sending multiple messages in flight in parallel for im
 			service.NewURLField(urlField).
 				Description("A URL to connect to.").
 				Example("amqp://localhost:5672/").
-				Example("amqps://guest:guest@localhost:5672/"),
+				Example("amqps://guest:guest@localhost:5672/").
+				Optional(),
+			service.NewURLListField(urlsField).
+				Description("A list of URLs to connect to. The first URL to successfully establish a connection will be used until the connection is closed. If an item of the list contains commas it will be expanded into multiple URLs.").
+				Example([]string{"amqp://guest:guest@127.0.0.1:5672/"}).
+				Example([]string{"amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/"}).
+				Example([]string{"amqp://127.0.0.1:5672/", "amqp://127.0.0.2:5672/"}).
+				Optional().
+				Version("T.B.D"),
 			service.NewStringField(targetAddrField).
 				Description("The target address to write to.").
 				Example("/foo").
@@ -72,7 +81,7 @@ type amqp1Writer struct {
 	session *amqp.Session
 	sender  *amqp.Sender
 
-	url                      string
+	urls                     []string
 	targetAddr               string
 	metaFilter               *service.MetadataExcludeFilter
 	applicationPropertiesMap *bloblang.Executor
@@ -88,11 +97,28 @@ func amqp1WriterFromParsed(conf *service.ParsedConfig, mgr *service.Resources) (
 		connOpts: &amqp.ConnOptions{},
 	}
 
-	var err error
-	if a.url, err = conf.FieldString(urlField); err != nil {
+	urlStrs, err := conf.FieldStringList(urlsField)
+	if err != nil {
 		return nil, err
 	}
 
+	for _, u := range urlStrs {
+		for _, splitURL := range strings.Split(u, ",") {
+			if trimmed := strings.TrimSpace(splitURL); len(trimmed) > 0 {
+				a.urls = append(a.urls, trimmed)
+			}
+		}
+	}
+
+	if len(a.urls) == 0 {
+		singleURL, err := conf.FieldString(urlField)
+		if err != nil {
+			return nil, err
+		}
+
+		a.urls = []string{singleURL}
+	}
+
 	if a.targetAddr, err = conf.FieldString(targetAddrField); err != nil {
 		return nil, err
 	}
@@ -136,8 +162,8 @@ func (a *amqp1Writer) Connect(ctx context.Context) (err error) {
 	)
 
 	// Create client
-	if client, err = amqp.Dial(ctx, a.url, a.connOpts); err != nil {
-		return
+	if client, err = a.reDial(ctx, a.urls); err != nil {
+		return err
 	}
 
 	// Open a session
@@ -251,3 +277,23 @@ func (a *amqp1Writer) Write(ctx context.Context, msg *service.Message) error {
 func (a *amqp1Writer) Close(ctx context.Context) error {
 	return a.disconnect(ctx)
 }
+
+// reDial connection to amqp with one or more fallback URLs.
+func (a *amqp1Writer) reDial(ctx context.Context, urls []string) (conn *amqp.Conn, err error) {
+	for i, url := range urls {
+		conn, err = amqp.Dial(ctx, url, a.connOpts)
+		if err != nil {
+			a.log.With("error", err).Warnf("unable to connect to url %q #%d, trying next", url, i)
+
+			continue
+		}
+
+		a.log.Tracef("successful connection to use %q #%d", url, i)
+
+		return conn, nil
+	}
+
+	a.log.With("error", err).Tracef("unable to connect to any of %d urls, return error", len(a.urls))
+
+	return nil, err
+}
diff --git a/website/docs/components/inputs/amqp_1.md b/website/docs/components/inputs/amqp_1.md
index 4b8ba890d1..c6f99b1a6e 100644
--- a/website/docs/components/inputs/amqp_1.md
+++ b/website/docs/components/inputs/amqp_1.md
@@ -29,7 +29,8 @@ Reads messages from an AMQP (1.0) server.
 input:
   label: ""
   amqp_1:
-    url: amqp://localhost:5672/ # No default (required)
+    url: amqp://localhost:5672/ # No default (optional)
+    urls: [] # No default (optional)
     source_address: /foo # No default (required)
 ```
 
@@ -41,7 +42,8 @@ input:
 input:
   label: ""
   amqp_1:
-    url: amqp://localhost:5672/ # No default (required)
+    url: amqp://localhost:5672/ # No default (optional)
+    urls: [] # No default (optional)
     source_address: /foo # No default (required)
     azure_renew_lock: false
     tls:
@@ -90,6 +92,28 @@ url: amqp://localhost:5672/
 url: amqps://guest:guest@localhost:5672/
 ```
 
+### `urls`
+
+A list of URLs to connect to. The first URL to successfully establish a connection will be used until the connection is closed. If an item of the list contains commas it will be expanded into multiple URLs.
+
+
+Type: `array`  
+Requires version T.B.D or newer  
+
+```yml
+# Examples
+
+urls:
+  - amqp://guest:guest@127.0.0.1:5672/
+
+urls:
+  - amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/
+
+urls:
+  - amqp://127.0.0.1:5672/
+  - amqp://127.0.0.2:5672/
+```
+
 ### `source_address`
 
 The source address to consume from.
diff --git a/website/docs/components/outputs/amqp_1.md b/website/docs/components/outputs/amqp_1.md
index b85b328eae..322083112d 100644
--- a/website/docs/components/outputs/amqp_1.md
+++ b/website/docs/components/outputs/amqp_1.md
@@ -29,7 +29,8 @@ Sends messages to an AMQP (1.0) server.
 output:
   label: ""
   amqp_1:
-    url: amqp://localhost:5672/ # No default (required)
+    url: amqp://localhost:5672/ # No default (optional)
+    urls: [] # No default (optional)
     target_address: /foo # No default (required)
     max_in_flight: 64
     metadata:
@@ -44,7 +45,8 @@ output:
 output:
   label: ""
   amqp_1:
-    url: amqp://localhost:5672/ # No default (required)
+    url: amqp://localhost:5672/ # No default (optional)
+    urls: [] # No default (optional)
     target_address: /foo # No default (required)
     max_in_flight: 64
     tls:
@@ -91,6 +93,28 @@ url: amqp://localhost:5672/
 url: amqps://guest:guest@localhost:5672/
 ```
 
+### `urls`
+
+A list of URLs to connect to. The first URL to successfully establish a connection will be used until the connection is closed. If an item of the list contains commas it will be expanded into multiple URLs.
+
+
+Type: `array`  
+Requires version T.B.D or newer  
+
+```yml
+# Examples
+
+urls:
+  - amqp://guest:guest@127.0.0.1:5672/
+
+urls:
+  - amqp://127.0.0.1:5672/,amqp://127.0.0.2:5672/
+
+urls:
+  - amqp://127.0.0.1:5672/
+  - amqp://127.0.0.2:5672/
+```
+
 ### `target_address`
 
 The target address to write to.


Test patch:
diff --git a/internal/impl/amqp1/integration_test.go b/internal/impl/amqp1/integration_test.go
index 537ac05abb..8d141ea381 100644
--- a/internal/impl/amqp1/integration_test.go
+++ b/internal/impl/amqp1/integration_test.go
@@ -45,7 +45,7 @@ func TestIntegrationAMQP1(t *testing.T) {
 		return err
 	}))
 
-	template := `
+	templateWithFieldURL := `
 output:
   amqp_1:
     url: amqp://guest:guest@localhost:$PORT/
@@ -59,28 +59,70 @@ input:
     url: amqp://guest:guest@localhost:$PORT/
     source_address: "queue:/$ID"
 `
-	suite := integration.StreamTests(
-		integration.StreamTestOpenClose(),
-		integration.StreamTestSendBatch(10),
-		integration.StreamTestStreamSequential(1000),
-		integration.StreamTestStreamParallel(1000),
-		integration.StreamTestMetadata(),
-		integration.StreamTestMetadataFilter(),
-	)
-	suite.Run(
-		t, template,
-		integration.StreamTestOptSleepAfterInput(100*time.Millisecond),
-		integration.StreamTestOptSleepAfterOutput(100*time.Millisecond),
-		integration.StreamTestOptPort(resource.GetPort("5672/tcp")),
-	)
-	t.Run("with max in flight", func(t *testing.T) {
-		t.Parallel()
-		suite.Run(
-			t, template,
-			integration.StreamTestOptSleepAfterInput(100*time.Millisecond),
-			integration.StreamTestOptSleepAfterOutput(100*time.Millisecond),
-			integration.StreamTestOptPort(resource.GetPort("5672/tcp")),
-			integration.StreamTestOptMaxInFlight(10),
-		)
-	})
+
+	templateWithFieldURLS := `
+output:
+  amqp_1:
+    urls:
+      - amqp://guest:guest@localhost:1234/
+      - amqp://guest:guest@localhost:$PORT/ # fallback URL
+      - amqp://guest:guest@localhost:4567/
+    target_address: "queue:/$ID"
+    max_in_flight: $MAX_IN_FLIGHT
+    metadata:
+      exclude_prefixes: [ $OUTPUT_META_EXCLUDE_PREFIX ]
+
+input:
+  amqp_1:
+    urls:
+      - amqp://guest:guest@localhost:1234/
+      - amqp://guest:guest@localhost:$PORT/ # fallback URL
+      - amqp://guest:guest@localhost:4567/
+    source_address: "queue:/$ID"
+`
+
+	testcases := []struct {
+		label    string
+		template string
+	}{
+		{
+			label:    "should handle old field url",
+			template: templateWithFieldURL,
+		},
+		{
+			label:    "should handle new field urls",
+			template: templateWithFieldURLS,
+		},
+	}
+
+	for _, tc := range testcases {
+		tc := tc
+		t.Run(tc.label, func(t *testing.T) {
+			suite := integration.StreamTests(
+				integration.StreamTestOpenClose(),
+				integration.StreamTestSendBatch(10),
+				integration.StreamTestStreamSequential(1000),
+				integration.StreamTestStreamParallel(1000),
+				integration.StreamTestMetadata(),
+				integration.StreamTestMetadataFilter(),
+			)
+			suite.Run(
+				t, tc.template,
+				integration.StreamTestOptSleepAfterInput(100*time.Millisecond),
+				integration.StreamTestOptSleepAfterOutput(100*time.Millisecond),
+				integration.StreamTestOptPort(resource.GetPort("5672/tcp")),
+			)
+
+			t.Run("with max in flight", func(t *testing.T) {
+				t.Parallel()
+				suite.Run(
+					t, tc.template,
+					integration.StreamTestOptSleepAfterInput(100*time.Millisecond),
+					integration.StreamTestOptSleepAfterOutput(100*time.Millisecond),
+					integration.StreamTestOptPort(resource.GetPort("5672/tcp")),
+					integration.StreamTestOptMaxInFlight(10),
+				)
+			})
+		})
+	}
 }