Add support for concatenating multiple embedded uris, or uris with ot…

…her string parts

Signed-off-by: Bogdan Drutu <bogdandrutu@gmail.com>

.chloggen/supportconcat.yaml

@@ -0,0 +1,11 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: confmap
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support to resolve embedded uris inside a string, concatenate results.
+
+# One or more tracking issues or pull requests related to the change
+issues: [6932]

confmap/resolver.go

@@ -26,15 +26,21 @@ import (
 	"go.opentelemetry.io/collector/featuregate"
 )
 
+// schemePattern defines the regexp pattern for scheme names.
+// Scheme name consist of a sequence of characters beginning with a letter and followed by any
+// combination of letters, digits, plus ("+"), period ("."), or hyphen ("-").
+const schemePattern = `[A-Za-z][A-Za-z0-9+.-]+`
+
 var (
 	// follows drive-letter specification:
 	// https://datatracker.ietf.org/doc/html/draft-kerwin-file-scheme-07.html#section-2.2
 	driverLetterRegexp = regexp.MustCompile("^[A-z]:")
 
-	// Scheme name consist of a sequence of characters beginning with a letter and followed by any
-	// combination of letters, digits, plus ("+"), period ("."), or hyphen ("-").
 	// Need to match new line as well in the OpaqueValue, so setting the "s" flag. See https://pkg.go.dev/regexp/syntax.
-	locationRegexp = regexp.MustCompile(`(?s:^(?P<Scheme>[A-Za-z][A-Za-z0-9+.-]+):(?P<OpaqueValue>.*)$)`)
+	uriRegexp = regexp.MustCompile(`(?s:^(?P<Scheme>` + schemePattern + `):(?P<OpaqueValue>.*)$)`)
+
+	// embeddedURI matches "embedded" provider uris into a string value.
+	embeddedURI = regexp.MustCompile(`\${` + schemePattern + `:.*?}`)
 
 	errTooManyRecursiveExpansions = errors.New("too many recursive expansions")
 )
@@ -168,6 +174,7 @@ func (mr *Resolver) Resolve(ctx context.Context) (*Conf, error) {
 		}
 		retMap = NewFromStringMap(cfgMap)
 	}
+
 	// Apply the converters in the given order.
 	for _, confConv := range mr.converters {
 		if err := confConv.Convert(ctx, retMap); err != nil {
@@ -234,28 +241,38 @@ func (mr *Resolver) expandValueRecursively(ctx context.Context, value interface{
 func (mr *Resolver) expandValue(ctx context.Context, value interface{}) (interface{}, bool, error) {
 	switch v := value.(type) {
 	case string:
-		// If it doesn't have the format "${scheme:opaque}" no need to expand.
-		if !strings.HasPrefix(v, "${") || !strings.HasSuffix(v, "}") || !strings.Contains(v, ":") {
-			return value, false, nil
-		}
-		lURI, err := newLocation(v[2 : len(v)-1])
-		if err != nil {
-			// Cannot return error, since a case like "${HOST}:${PORT}" is invalid location,
-			// but is supported in the legacy implementation.
+		// If no embedded "uris" no need to expand. embeddedURI regexp matches uriRegexp as well.
+		if !embeddedURI.MatchString(v) {
 			return value, false, nil
 		}
-		if strings.Contains(lURI.opaqueValue, "$") {
-			return nil, false, fmt.Errorf("the uri %q contains unsupported characters ('$')", lURI.asString())
-		}
-		ret, err := mr.retrieveValue(ctx, lURI)
-		if err != nil {
-			return nil, false, err
+
+		// If the value is a single URI, then the return value can be anything.
+		// This is the case `foo: ${file:some_extra_config.yml}`.
+		if embeddedURI.FindString(v) == v {
+			return mr.expandStringURI(ctx, v)
 		}
-		mr.closers = append(mr.closers, ret.Close)
-		val, err := ret.AsRaw()
-		return val, true, err
-	case []interface{}:
-		nslice := make([]interface{}, 0, len(v))
+
+		// If the URI is embedded into the string, return value must be a string, and we have to concatenate all strings.
+		var nerr error
+		var nchanged bool
+		nv := embeddedURI.ReplaceAllStringFunc(v, func(s string) string {
+			ret, changed, err := mr.expandStringURI(ctx, s)
+			nchanged = nchanged || changed
+			nerr = multierr.Append(nerr, err)
+			if err != nil {
+				return ""
+			}
+			switch val := ret.(type) {
+			case string:
+				return val
+			default:
+				nerr = multierr.Append(nerr, fmt.Errorf("expending %v, expected string value type, got %T", s, val))
+				return v
+			}
+		})
+		return nv, nchanged, nerr
+	case []any:
+		nslice := make([]any, 0, len(v))
 		nchanged := false
 		for _, vint := range v {
 			val, changed, err := mr.expandValue(ctx, vint)
@@ -282,6 +299,23 @@ func (mr *Resolver) expandValue(ctx context.Context, value interface{}) (interfa
 	return value, false, nil
 }
 
+func (mr *Resolver) expandStringURI(ctx context.Context, uri string) (any, bool, error) {
+	lURI, err := newLocation(uri[2 : len(uri)-1])
+	if err != nil {
+		return nil, false, err
+	}
+	if strings.Contains(lURI.opaqueValue, "$") {
+		return nil, false, fmt.Errorf("the uri %q contains unsupported characters ('$')", lURI.asString())
+	}
+	ret, err := mr.retrieveValue(ctx, lURI)
+	if err != nil {
+		return nil, false, err
+	}
+	mr.closers = append(mr.closers, ret.Close)
+	val, err := ret.AsRaw()
+	return val, true, err
+}
+
 type location struct {
 	scheme      string
 	opaqueValue string
@@ -292,7 +326,7 @@ func (c location) asString() string {
 }
 
 func newLocation(uri string) (location, error) {
-	submatches := locationRegexp.FindStringSubmatch(uri)
+	submatches := uriRegexp.FindStringSubmatch(uri)
 	if len(submatches) != 3 {
 		return location{}, fmt.Errorf("invalid uri: %q", uri)
 	}

confmap/resolver_test.go

@@ -423,6 +423,43 @@ func TestResolverExpandMapAndSliceValues(t *testing.T) {
 	assert.Equal(t, expectedMap, cfgMap.ToStringMap())
 }
 
+func TestResolverExpandStringValues(t *testing.T) {
+	provider := newFakeProvider("input", func(context.Context, string, WatcherFunc) (*Retrieved, error) {
+		return NewRetrieved(map[string]any{
+			"test_no_match":             "${HOST}:${PORT}",
+			"test_match_value":          "${env:HOST_PORT}",
+			"test_match_embedded":       "${env:HOST}:3043",
+			"test_match_embedded_multi": "${env:HOST}:${env:PORT}",
+		})
+	})
+
+	testProvider := newFakeProvider("env", func(_ context.Context, uri string, _ WatcherFunc) (*Retrieved, error) {
+		switch uri {
+		case "env:HOST_PORT":
+			return NewRetrieved([]any{"localhost:3042"})
+		case "env:HOST":
+			return NewRetrieved("localhost")
+		case "env:PORT":
+			return NewRetrieved("3044")
+
+		}
+		return nil, errors.New("impossible")
+	})
+
+	resolver, err := NewResolver(ResolverSettings{URIs: []string{"input:"}, Providers: makeMapProvidersMap(provider, testProvider), Converters: nil})
+	require.NoError(t, err)
+
+	cfgMap, err := resolver.Resolve(context.Background())
+	require.NoError(t, err)
+	expectedMap := map[string]any{
+		"test_no_match":             "${HOST}:${PORT}",
+		"test_match_value":          []any{"localhost:3042"},
+		"test_match_embedded":       "localhost:3043",
+		"test_match_embedded_multi": "localhost:3044",
+	}
+	assert.Equal(t, expectedMap, cfgMap.ToStringMap())
+}
+
 func TestResolverInfiniteExpand(t *testing.T) {
 	const receiverValue = "${test:VALUE}"
 	provider := newFakeProvider("input", func(context.Context, string, WatcherFunc) (*Retrieved, error) {
@@ -497,7 +534,7 @@ func TestResolverExpandInvalidOpaqueValue(t *testing.T) {
 	})
 
 	testProvider := newFakeProvider("test", func(context.Context, string, WatcherFunc) (*Retrieved, error) {
-		return NewRetrieved(errors.New("invalid value"))
+		panic("must not be called")
 	})
 
 	resolver, err := NewResolver(ResolverSettings{URIs: []string{"input:"}, Providers: makeMapProvidersMap(provider, testProvider), Converters: nil})
@@ -507,6 +544,55 @@ func TestResolverExpandInvalidOpaqueValue(t *testing.T) {
 	assert.EqualError(t, err, `the uri "test:$VALUE" contains unsupported characters ('$')`)
 }
 
+func TestResolverExpandUnsupportedScheme(t *testing.T) {
+	provider := newFakeProvider("input", func(context.Context, string, WatcherFunc) (*Retrieved, error) {
+		return NewRetrieved(map[string]interface{}{"test": "${unsupported:VALUE}"})
+	})
+
+	testProvider := newFakeProvider("test", func(context.Context, string, WatcherFunc) (*Retrieved, error) {
+		panic("must not be called")
+	})
+
+	resolver, err := NewResolver(ResolverSettings{URIs: []string{"input:"}, Providers: makeMapProvidersMap(provider, testProvider), Converters: nil})
+	require.NoError(t, err)
+
+	_, err = resolver.Resolve(context.Background())
+	assert.EqualError(t, err, `scheme "unsupported" is not supported for uri "unsupported:VALUE"`)
+}
+
+func TestResolverExpandStringValueInvalidReturnValue(t *testing.T) {
+	provider := newFakeProvider("input", func(context.Context, string, WatcherFunc) (*Retrieved, error) {
+		return NewRetrieved(map[string]interface{}{"test": "localhost:${test:PORT}"})
+	})
+
+	testProvider := newFakeProvider("test", func(context.Context, string, WatcherFunc) (*Retrieved, error) {
+		return NewRetrieved(3043)
+	})
+
+	resolver, err := NewResolver(ResolverSettings{URIs: []string{"input:"}, Providers: makeMapProvidersMap(provider, testProvider), Converters: nil})
+	require.NoError(t, err)
+
+	_, err = resolver.Resolve(context.Background())
+	assert.EqualError(t, err, `expending ${test:PORT}, expected string value type, got int`)
+}
+
+func TestResolverExpandStringValueReturnError(t *testing.T) {
+	provider := newFakeProvider("input", func(context.Context, string, WatcherFunc) (*Retrieved, error) {
+		return NewRetrieved(map[string]interface{}{"test": "localhost:${test:PORT}"})
+	})
+
+	myErr := errors.New("my error")
+	testProvider := newFakeProvider("test", func(context.Context, string, WatcherFunc) (*Retrieved, error) {
+		return nil, myErr
+	})
+
+	resolver, err := NewResolver(ResolverSettings{URIs: []string{"input:"}, Providers: makeMapProvidersMap(provider, testProvider), Converters: nil})
+	require.NoError(t, err)
+
+	_, err = resolver.Resolve(context.Background())
+	assert.ErrorIs(t, err, myErr)
+}
+
 func makeMapProvidersMap(providers ...Provider) map[string]Provider {
 	ret := make(map[string]Provider, len(providers))
 	for _, provider := range providers {