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

Issue:
Support interpolation in mongo processor
Hi, I was wondering if it is possible to support [interpolations](https://www.benthos.dev/docs/configuration/interpolation/) in the mongo processor. For example, I wanted to be able to do something like this:

Depending on the country, the name of the collection changes.

```
- mongodb:
         url: ${DB_MONGO_URL}
         database: ${DB_MONGO_NAME}
         collection: '${! meta("country") }-property'
         username: ${DB_USER}
         password: ${DB_PASS}
         operation: find-one
         filter_map: |-
            root.id = this.id
            hint_map: ""
         write_concern:
            w: ""
            j: false
            w_timeout: "60s"
```

Error: 

`{"@timestamp":"2021-11-30T17:13:47-03:00","@service":"benthos","component":"benthos.pipeline.processor.2.0","json":"{\"error\":\"(InvalidNamespace) Invalid collection name specified 'ads.${! meta(\\\"country\\\")}-property\",\"is_error\":true}","meta":"{\"action\":\"delete\",\"country\":\"co\",\"entity\":\"property\",\"gcp_pubsub_publish_time_unix\":\"1638303226\",\"key\":\"1458513\"}","level":"DEBUG","message":" Avro data parsed"}`

Benthos version

```
Version: 3.59.0
Date: 2021-11-26T04:48:24Z
```

Thank you for everything. It's a great project ❤️ 


Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4917cad6e3..c78bb598d3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,7 @@ All notable changes to this project will be documented in this file.
 - Field `sync` added to the `gcp_pubsub` input.
 - New input, processor, and output config field types added to the plugin APIs.
 - Added new experimental `parquet` processor.
+- Field `collection` in `mongodb` processor and output now supports interpolation functions.
 - Field `output_raw` added to the `jq` processor.
 
 ### Fixed
diff --git a/internal/impl/mongodb/cache.go b/internal/impl/mongodb/cache.go
index 58f609941d..1fbe1b48ac 100644
--- a/internal/impl/mongodb/cache.go
+++ b/internal/impl/mongodb/cache.go
@@ -34,6 +34,7 @@ func init() {
 		Description: icache.Description(false, ""),
 		Config: docs.FieldComponent().WithChildren(
 			client.ConfigDocs().Add(
+				docs.FieldCommon("collection", "The name of the target collection in the MongoDB DB.").IsInterpolated(),
 				docs.FieldCommon("key_field", "The field in the document that is used as the key."),
 				docs.FieldCommon("value_field", "The field in the document that is used as the value."),
 			)...,
diff --git a/internal/impl/mongodb/client/config.go b/internal/impl/mongodb/client/config.go
index 23f0315ac0..9f38481fe9 100644
--- a/internal/impl/mongodb/client/config.go
+++ b/internal/impl/mongodb/client/config.go
@@ -119,7 +119,6 @@ func ConfigDocs() docs.FieldSpecs {
 			"mongodb://localhost:27017",
 		),
 		docs.FieldCommon("database", "The name of the target MongoDB DB."),
-		docs.FieldCommon("collection", "The name of the target collection in the MongoDB DB."),
 		docs.FieldCommon("username", "The username to connect to the database."),
 		docs.FieldCommon("password", "The password to connect to the database."),
 	}
diff --git a/internal/impl/mongodb/output.go b/internal/impl/mongodb/output.go
index eb77403e28..a5924d92f0 100644
--- a/internal/impl/mongodb/output.go
+++ b/internal/impl/mongodb/output.go
@@ -9,11 +9,13 @@ import (
 	"time"
 
 	ibatch "github.com/Jeffail/benthos/v3/internal/batch"
+	"github.com/Jeffail/benthos/v3/internal/bloblang/field"
 	"github.com/Jeffail/benthos/v3/internal/bloblang/mapping"
 	"github.com/Jeffail/benthos/v3/internal/bundle"
 	ioutput "github.com/Jeffail/benthos/v3/internal/component/output"
 	"github.com/Jeffail/benthos/v3/internal/docs"
 	"github.com/Jeffail/benthos/v3/internal/impl/mongodb/client"
+	"github.com/Jeffail/benthos/v3/internal/interop"
 	"github.com/Jeffail/benthos/v3/internal/shutdown"
 	"github.com/Jeffail/benthos/v3/lib/log"
 	"github.com/Jeffail/benthos/v3/lib/message/batch"
@@ -43,6 +45,7 @@ func init() {
 		Config: docs.FieldComponent().WithChildren(
 			client.ConfigDocs().Add(
 				outputOperationDocs(client.OperationUpdateOne),
+				docs.FieldCommon("collection", "The name of the target collection in the MongoDB DB.").IsInterpolated(),
 				docs.FieldCommon(
 					"write_concern",
 					"The write concern settings for the mongo connection.",
@@ -169,6 +172,10 @@ func NewWriter(
 	if db.wcTimeout, err = time.ParseDuration(conf.WriteConcern.WTimeout); err != nil {
 		return nil, fmt.Errorf("failed to parse write concern wtimeout string: %v", err)
 	}
+	if db.collection, err = interop.NewBloblangField(mgr, conf.MongoConfig.Collection); err != nil {
+		return nil, fmt.Errorf("failed to parse collection expression: %v", err)
+	}
+
 	return db, nil
 }
 
@@ -186,9 +193,11 @@ type Writer struct {
 	hintMap     *mapping.Executor
 	operation   client.Operation
 
-	mu         sync.Mutex
-	client     *mongo.Client
-	collection *mongo.Collection
+	mu                           sync.Mutex
+	client                       *mongo.Client
+	collection                   *field.Expression
+	database                     *mongo.Database
+	writeConcernCollectionOption *options.CollectionOptions
 
 	shutSig *shutdown.Signaller
 }
@@ -235,9 +244,9 @@ func (m *Writer) ConnectWithContext(ctx context.Context) error {
 		return fmt.Errorf("write_concern validation error: %w", err)
 	}
 
-	m.collection = client.
-		Database(m.conf.MongoConfig.Database).
-		Collection(m.conf.MongoConfig.Collection, options.Collection().SetWriteConcern(writeConcern))
+	m.database = client.Database(m.conf.MongoConfig.Database)
+	m.writeConcernCollectionOption = options.Collection().SetWriteConcern(writeConcern)
+
 	m.client = client
 	return nil
 }
@@ -252,7 +261,7 @@ func (m *Writer) WriteWithContext(ctx context.Context, msg types.Message) error
 		return types.ErrNotConnected
 	}
 
-	var writeModels []mongo.WriteModel
+	writeModelsMap := map[*mongo.Collection][]mongo.WriteModel{}
 	err := writer.IterateBatchedSend(msg, func(i int, _ types.Part) error {
 		var err error
 		var filterVal, documentVal types.Part
@@ -307,6 +316,7 @@ func (m *Writer) WriteWithContext(ctx context.Context, msg types.Message) error
 		}
 
 		var writeModel mongo.WriteModel
+		collection := m.database.Collection(collection.String(i, msg), m.writeConcernCollectionOption)
 
 		switch m.operation {
 		case client.OperationInsertOne:
@@ -340,11 +350,12 @@ func (m *Writer) WriteWithContext(ctx context.Context, msg types.Message) error
 		}
 
 		if writeModel != nil {
-			writeModels = append(writeModels, writeModel)
+			writeModelsMap[collection] = append(writeModelsMap[collection], writeModel)
 		}
 		return nil
 	})
 
+	// Check for fatal errors and exit immediately if we encounter one
 	var batchErr *ibatch.Error
 	if err != nil {
 		if !errors.As(err, &batchErr) {
@@ -352,12 +363,17 @@ func (m *Writer) WriteWithContext(ctx context.Context, msg types.Message) error
 		}
 	}
 
-	if len(writeModels) > 0 {
-		if _, err = collection.BulkWrite(ctx, writeModels); err != nil {
-			return err
+	// Dispatch any documents which IterateBatchedSend managed to process successfully
+	if len(writeModelsMap) > 0 {
+		for collection, writeModels := range writeModelsMap {
+			// We should have at least one write model in the slice
+			if _, err := collection.BulkWrite(context.Background(), writeModels); err != nil {
+				return err
+			}
 		}
 	}
 
+	// Return any errors produced by invalid messages from the batch
 	if batchErr != nil {
 		return batchErr
 	}
diff --git a/internal/impl/mongodb/processor.go b/internal/impl/mongodb/processor.go
index da727742fe..b0aa1f1dc0 100644
--- a/internal/impl/mongodb/processor.go
+++ b/internal/impl/mongodb/processor.go
@@ -7,10 +7,12 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/Jeffail/benthos/v3/internal/bloblang/field"
 	"github.com/Jeffail/benthos/v3/internal/bloblang/mapping"
 	"github.com/Jeffail/benthos/v3/internal/bundle"
 	"github.com/Jeffail/benthos/v3/internal/docs"
 	"github.com/Jeffail/benthos/v3/internal/impl/mongodb/client"
+	"github.com/Jeffail/benthos/v3/internal/interop"
 	"github.com/Jeffail/benthos/v3/internal/shutdown"
 	"github.com/Jeffail/benthos/v3/internal/tracing"
 	"github.com/Jeffail/benthos/v3/lib/log"
@@ -41,6 +43,7 @@ func init() {
 		Config: docs.FieldComponent().WithChildren(
 			client.ConfigDocs().Add(
 				processorOperationDocs(client.OperationInsertOne),
+				docs.FieldCommon("collection", "The name of the target collection in the MongoDB DB.").IsInterpolated(),
 				docs.FieldCommon(
 					"write_concern",
 					"The write_concern settings for the mongo connection.",
@@ -94,8 +97,10 @@ type Processor struct {
 	log   log.Modular
 	stats metrics.Type
 
-	client     *mongo.Client
-	collection *mongo.Collection
+	client                       *mongo.Client
+	collection                   *field.Expression
+	database                     *mongo.Database
+	writeConcernCollectionOption *options.CollectionOptions
 
 	parts       []int
 	filterMap   *mapping.Executor
@@ -220,9 +225,12 @@ func NewProcessor(
 		return nil, fmt.Errorf("write_concern validation error: %w", err)
 	}
 
-	m.collection = m.client.
-		Database(conf.MongoDB.MongoDB.Database).
-		Collection(conf.MongoDB.MongoDB.Collection, options.Collection().SetWriteConcern(writeConcern))
+	if m.collection, err = interop.NewBloblangField(mgr, m.conf.MongoDB.Collection); err != nil {
+		return nil, fmt.Errorf("failed to parse collection expression: %v", err)
+	}
+
+	m.database = m.client.Database(conf.MongoDB.MongoDB.Database)
+	m.writeConcernCollectionOption = options.Collection().SetWriteConcern(writeConcern)
 
 	return m, nil
 }
@@ -233,7 +241,7 @@ func (m *Processor) ProcessMessage(msg types.Message) ([]types.Message, types.Re
 	m.mCount.Incr(1)
 	newMsg := msg.Copy()
 
-	var writeModels []mongo.WriteModel
+	writeModelsMap := map[*mongo.Collection][]mongo.WriteModel{}
 	processor.IteratePartsWithSpanV2("mongodb", m.parts, newMsg, func(i int, s *tracing.Span, p types.Part) error {
 		var err error
 		var filterVal, documentVal types.Part
@@ -290,6 +298,7 @@ func (m *Processor) ProcessMessage(msg types.Message) ([]types.Message, types.Re
 		}
 
 		var writeModel mongo.WriteModel
+		collection := m.database.Collection(m.collection.String(i, msg), m.writeConcernCollectionOption)
 
 		switch m.operation {
 		case client.OperationInsertOne:
@@ -322,7 +331,7 @@ func (m *Processor) ProcessMessage(msg types.Message) ([]types.Message, types.Re
 			}
 		case client.OperationFindOne:
 			var decoded interface{}
-			err := m.collection.FindOne(context.Background(), filterJSON, findOptions).Decode(&decoded)
+			err := collection.FindOne(context.Background(), filterJSON, findOptions).Decode(&decoded)
 			if err != nil {
 				if err == mongo.ErrNoDocuments {
 					return err
@@ -341,16 +350,19 @@ func (m *Processor) ProcessMessage(msg types.Message) ([]types.Message, types.Re
 		}
 
 		if writeModel != nil {
-			writeModels = append(writeModels, writeModel)
+			writeModelsMap[collection] = append(writeModelsMap[collection], writeModel)
 		}
 		return nil
 	})
 
-	if len(writeModels) > 0 {
-		if _, err := m.collection.BulkWrite(context.Background(), writeModels); err != nil {
-			m.log.Errorf("Bulk write failed in mongodb processor: %v", err)
-			for _, n := range m.parts {
-				processor.FlagErr(newMsg.Get(n), err)
+	if len(writeModelsMap) > 0 {
+		for collection, writeModels := range writeModelsMap {
+			// We should have at least one write model in the slice
+			if _, err := collection.BulkWrite(context.Background(), writeModels); err != nil {
+				m.log.Errorf("Bulk write failed in mongodb processor: %v", err)
+				for _, n := range m.parts {
+					processor.FlagErr(newMsg.Get(n), err)
+				}
 			}
 		}
 	}
diff --git a/website/docs/components/caches/mongodb.md b/website/docs/components/caches/mongodb.md
index 94b50c59ae..23b834ec6d 100644
--- a/website/docs/components/caches/mongodb.md
+++ b/website/docs/components/caches/mongodb.md
@@ -27,9 +27,9 @@ label: ""
 mongodb:
   url: ""
   database: ""
-  collection: ""
   username: ""
   password: ""
+  collection: ""
   key_field: ""
   value_field: ""
 ```
@@ -58,25 +58,26 @@ The name of the target MongoDB DB.
 Type: `string`  
 Default: `""`  
 
-### `collection`
+### `username`
 
-The name of the target collection in the MongoDB DB.
+The username to connect to the database.
 
 
 Type: `string`  
 Default: `""`  
 
-### `username`
+### `password`
 
-The username to connect to the database.
+The password to connect to the database.
 
 
 Type: `string`  
 Default: `""`  
 
-### `password`
+### `collection`
 
-The password to connect to the database.
+The name of the target collection in the MongoDB DB.
+This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries).
 
 
 Type: `string`  
diff --git a/website/docs/components/outputs/mongodb.md b/website/docs/components/outputs/mongodb.md
index 9ee513a814..b15841547a 100644
--- a/website/docs/components/outputs/mongodb.md
+++ b/website/docs/components/outputs/mongodb.md
@@ -37,10 +37,10 @@ output:
   mongodb:
     url: ""
     database: ""
-    collection: ""
     username: ""
     password: ""
     operation: update-one
+    collection: ""
     write_concern:
       w: ""
       j: false
@@ -67,10 +67,10 @@ output:
   mongodb:
     url: ""
     database: ""
-    collection: ""
     username: ""
     password: ""
     operation: update-one
+    collection: ""
     write_concern:
       w: ""
       j: false
@@ -128,14 +128,6 @@ url: mongodb://localhost:27017
 The name of the target MongoDB DB.
 
 
-Type: `string`  
-Default: `""`  
-
-### `collection`
-
-The name of the target collection in the MongoDB DB.
-
-
 Type: `string`  
 Default: `""`  
 
@@ -164,6 +156,15 @@ Type: `string`
 Default: `"update-one"`  
 Options: `insert-one`, `delete-one`, `delete-many`, `replace-one`, `update-one`.
 
+### `collection`
+
+The name of the target collection in the MongoDB DB.
+This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries).
+
+
+Type: `string`  
+Default: `""`  
+
 ### `write_concern`
 
 The write concern settings for the mongo connection.
diff --git a/website/docs/components/processors/mongodb.md b/website/docs/components/processors/mongodb.md
index 1268c7ddb7..d5055f703e 100644
--- a/website/docs/components/processors/mongodb.md
+++ b/website/docs/components/processors/mongodb.md
@@ -36,10 +36,10 @@ label: ""
 mongodb:
   url: ""
   database: ""
-  collection: ""
   username: ""
   password: ""
   operation: insert-one
+  collection: ""
   write_concern:
     w: ""
     j: false
@@ -59,10 +59,10 @@ label: ""
 mongodb:
   url: ""
   database: ""
-  collection: ""
   username: ""
   password: ""
   operation: insert-one
+  collection: ""
   write_concern:
     w: ""
     j: false
@@ -104,14 +104,6 @@ url: mongodb://localhost:27017
 The name of the target MongoDB DB.
 
 
-Type: `string`  
-Default: `""`  
-
-### `collection`
-
-The name of the target collection in the MongoDB DB.
-
-
 Type: `string`  
 Default: `""`  
 
@@ -140,6 +132,15 @@ Type: `string`
 Default: `"insert-one"`  
 Options: `insert-one`, `delete-one`, `delete-many`, `replace-one`, `update-one`, `find-one`.
 
+### `collection`
+
+The name of the target collection in the MongoDB DB.
+This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries).
+
+
+Type: `string`  
+Default: `""`  
+
 ### `write_concern`
 
 The write_concern settings for the mongo connection.


Test patch:
diff --git a/internal/impl/mongodb/integration_test.go b/internal/impl/mongodb/integration_test.go
index 1ebae62f86..0f46877ec8 100644
--- a/internal/impl/mongodb/integration_test.go
+++ b/internal/impl/mongodb/integration_test.go
@@ -52,8 +52,6 @@ func TestIntegrationMongoDB(t *testing.T) {
 		url := "mongodb://localhost:" + resource.GetPort("27017/tcp")
 		conf := client.NewConfig()
 		conf.URL = url
-		conf.Database = "TestDB"
-		conf.Collection = "TestCollection"
 		conf.Username = "mongoadmin"
 		conf.Password = "secret"
 
@@ -113,7 +111,7 @@ output:
 			integration.StreamTestOptPreTest(func(t testing.TB, ctx context.Context, testID string, vars *integration.StreamTestConfigVars) {
 				cName := generateCollectionName(testID)
 				vars.Var1 = cName
-				require.NoError(t, createCollection(resource, cName, "mongoadmin", "secret"))
+				require.NoError(t, mongoClient.Database("TestDB").CreateCollection(ctx, cName))
 			}),
 		)
 	})
@@ -144,7 +142,7 @@ cache_resources:
 			integration.CacheTestOptPreTest(func(t testing.TB, ctx context.Context, testID string, vars *integration.CacheTestConfigVars) {
 				cName := generateCollectionName(testID)
 				vars.Var1 = cName
-				require.NoError(t, createCollection(resource, cName, "mongoadmin", "secret"))
+				require.NoError(t, mongoClient.Database("TestDB").CreateCollection(ctx, cName))
 			}),
 		)
 	})
diff --git a/internal/impl/mongodb/processor_test.go b/internal/impl/mongodb/processor_test.go
index 5a4de6cb2b..302e5c3190 100644
--- a/internal/impl/mongodb/processor_test.go
+++ b/internal/impl/mongodb/processor_test.go
@@ -6,7 +6,6 @@ import (
 	"regexp"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/Jeffail/benthos/v3/internal/impl/mongodb"
 	"github.com/Jeffail/benthos/v3/internal/impl/mongodb/client"
@@ -58,8 +57,6 @@ func TestProcessorIntegration(t *testing.T) {
 		url := "mongodb://localhost:" + resource.GetPort("27017/tcp")
 		conf := client.NewConfig()
 		conf.URL = url
-		conf.Database = "TestDB"
-		conf.Collection = "TestCollection"
 		conf.Username = "mongoadmin"
 		conf.Password = "secret"
 
@@ -74,7 +71,7 @@ func TestProcessorIntegration(t *testing.T) {
 			return err
 		}
 
-		return createCollection(resource, "TestCollection", "mongoadmin", "secret")
+		return mongoClient.Database("TestDB").CreateCollection(context.Background(), "TestCollection")
 	}))
 
 	port := resource.GetPort("27017/tcp")
@@ -459,6 +456,7 @@ func testMongoDBProcessorFindOne(port string, t *testing.T) {
 		name        string
 		message     string
 		marshalMode client.JSONMarshalMode
+		collection  string
 		expected    string
 		expectedErr error
 	}{
@@ -479,7 +477,18 @@ func testMongoDBProcessorFindOne(port string, t *testing.T) {
 			message:     `{"a":"notfound"}`,
 			expectedErr: mongo.ErrNoDocuments,
 		},
+		{
+			name:        "collection interpolation",
+			marshalMode: client.JSONMarshalModeCanonical,
+			collection:  `${!json("col")}`,
+			message:     `{"col":"TestCollection","a":"foo"}`,
+			expected:    `{"a":"foo","b":"bar","c":"baz","answer_to_everything":{"$numberInt":"42"}}`,
+		},
 	} {
+		if tt.collection != "" {
+			conf.MongoDB.MongoDB.Collection = tt.collection
+		}
+
 		conf.MongoDB.JSONMarshalMode = tt.marshalMode
 
 		m, err := mongodb.NewProcessor(conf, mgr, log.Noop(), metrics.Noop())
@@ -497,20 +506,3 @@ func testMongoDBProcessorFindOne(port string, t *testing.T) {
 		assert.Equalf(t, jsondiff.SupersetMatch.String(), diff.String(), "%s: %s", tt.name, explanation)
 	}
 }
-
-func createCollection(resource *dockertest.Resource, collectionName, username, password string) error {
-	_, err := resource.Exec([]string{
-		"mongo",
-		"-u", username,
-		"-p", password,
-		"--authenticationDatabase", "admin",
-		"--eval", "db.createCollection(\"" + collectionName + "\")",
-		"TestDB",
-	}, dockertest.ExecOptions{})
-	if err != nil {
-		return err
-	}
-
-	time.Sleep(time.Second * 1)
-	return nil
-}