Add bloblang scalar type to template fields

Signed-off-by: Mihai Todor <[email protected]>

CHANGELOG.md

@@ -9,6 +9,7 @@ All notable changes to this project will be documented in this file.
 
 - Field `label` added to the template tests definitions. (@mihaitodor)
 - Metadata field `label` can now be utilized within a template's `mapping` field to access the label that is associated with the template instantiation in a config. (@mihaitodor)
+- `bloblang` scalar type added to template fields. (@mihaitodor)
 
 ### Changed
 

internal/template/config.go

@@ -72,6 +72,9 @@ func (c FieldConfig) FieldSpec() (docs.FieldSpec, error) {
 			return f, fmt.Errorf("unrecognised scalar type: %v", *c.Kind)
 		}
 	}
+	if f.Type == "bloblang" {
+		f = f.HasType(docs.FieldTypeString).IsBloblang()
+	}
 	return f, nil
 }
 
@@ -222,6 +225,7 @@ func FieldConfigSpec() docs.FieldSpecs {
 			"int", "standard integer type",
 			"float", "standard float type",
 			"bool", "a boolean true/false",
+			"bloblang", "a bloblang mapping",
 			"unknown", "allows for nesting arbitrary configuration inside of a field",
 		),
 		docs.FieldString("kind", "The kind of the field.").HasOptions(

internal/template/template_test.go

@@ -2,6 +2,7 @@ package template_test
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
@@ -13,6 +14,7 @@ import (
 	"github.com/redpanda-data/benthos/v4/internal/component/output"
 	"github.com/redpanda-data/benthos/v4/internal/component/processor"
 	"github.com/redpanda-data/benthos/v4/internal/component/ratelimit"
+	"github.com/redpanda-data/benthos/v4/internal/docs"
 	"github.com/redpanda-data/benthos/v4/internal/manager"
 	"github.com/redpanda-data/benthos/v4/internal/message"
 	"github.com/redpanda-data/benthos/v4/internal/template"
@@ -360,3 +362,153 @@ mapping: |
 		})
 	}
 }
+
+func TestProcessorTemplateFieldLinting(t *testing.T) {
+	tests := []struct {
+		name         string
+		fieldType    string
+		fieldValue   any
+		message      string
+		expected     string
+		errContains  string
+		lintContains string
+	}{
+		{
+			name:       "valid string field",
+			fieldType:  "string",
+			fieldValue: `"foobar"`,
+			message:    "test",
+			expected:   "foobar",
+		},
+		{
+			name:       "valid int field",
+			fieldType:  "int",
+			fieldValue: 42,
+			message:    "test",
+			expected:   "42",
+		},
+		{
+			name:       "invalid int field",
+			fieldType:  "int",
+			fieldValue: "foobar",
+			message:    "test",
+			// TODO: Should this also trigger a lint error?
+			errContains: `expected number value, got string ("foobar")`,
+		},
+		{
+			name:       "valid float field",
+			fieldType:  "float",
+			fieldValue: 3.14,
+			message:    "test",
+			expected:   "3.14",
+		},
+		{
+			name:       "invalid float field",
+			fieldType:  "float",
+			fieldValue: "foobar",
+			message:    "test",
+			// TODO: Should this also trigger a lint error?
+			errContains: `expected number value, got string ("foobar")`,
+		},
+		{
+			name:       "valid bool field",
+			fieldType:  "bool",
+			fieldValue: true,
+			message:    "test",
+			expected:   "true",
+		},
+		{
+			name:       "invalid bool field",
+			fieldType:  "bool",
+			fieldValue: "foobar",
+			message:    "test",
+			// TODO: Should this also trigger a lint error?
+			errContains: `expected bool value, got string ("foobar")`,
+		},
+		{
+			name:       "valid unknown field",
+			fieldType:  "unknown",
+			fieldValue: `"foobar"`,
+			message:    "test",
+			expected:   "foobar",
+		},
+		{
+			name:       "valid bloblang mapping",
+			fieldType:  "bloblang",
+			fieldValue: `root = content().uppercase()`,
+			message:    "kaboom!",
+			expected:   "KABOOM!",
+		},
+		{
+			name:         "invalid bloblang mapping",
+			fieldType:    "bloblang",
+			fieldValue:   `root = # invalid`,
+			message:      "kaboom!",
+			errContains:  "failed to parse bloblang mapping '': expected query, got: # inv",
+			lintContains: "expected whitespace, but reached end of input",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			mgr, err := manager.New(manager.NewResourceConfig())
+			require.NoError(t, err)
+
+			var mapping string
+			if test.fieldType != "bloblang" {
+				mapping = fmt.Sprintf(`"""root = %v """`, test.fieldValue)
+			} else {
+				mapping = fmt.Sprintf(`"%s"`, test.fieldValue)
+			}
+			tmpl := fmt.Sprintf(`
+name: foobar
+type: processor
+
+fields:
+  - name: test
+    type: %s
+
+mapping: |
+  root.mapping = %s
+`, test.fieldType, mapping)
+			require.NoError(t, template.RegisterTemplateYAML(mgr.Environment(), mgr.BloblEnvironment(), []byte(tmpl)))
+
+			spec, ok := mgr.Environment().GetDocs("foobar", docs.TypeProcessor)
+			require.True(t, ok)
+
+			node, err := docs.UnmarshalYAML([]byte(fmt.Sprintf(`test: %v`, test.fieldValue)))
+			require.NoError(t, err)
+
+			lints := spec.Config.LintYAML(docs.NewLintContext(docs.NewLintConfig(mgr.Environment())), node)
+			if test.lintContains != "" {
+				require.Len(t, lints, 1)
+				assert.Contains(t, lints[0].What, test.lintContains)
+			} else {
+				require.Empty(t, lints)
+			}
+
+			conf, err := processor.FromAny(mgr, map[string]any{
+				"foobar": map[string]any{
+					"test": test.fieldValue,
+				},
+			})
+			require.NoError(t, err)
+
+			p, err := mgr.NewProcessor(conf)
+			if test.errContains != "" {
+				require.ErrorContains(t, err, test.errContains)
+				return
+			}
+			require.NoError(t, err)
+
+			res, err := p.ProcessBatch(context.Background(), message.Batch{
+				message.NewPart([]byte(test.message)),
+			})
+			require.NoError(t, err)
+			require.Len(t, res, 1)
+			require.Len(t, res[0], 1)
+			require.NoError(t, res[0][0].ErrorGet())
+			assert.Equal(t, test.expected, string(res[0][0].AsBytes()))
+		})
+	}
+}