Extend Expand converter to support expanding values from any provider

There are still not 100% finalize designed choices:
1. This PR requires a change in the current Provider interface to support returning a config.Map (maybe a map[string]interface{}) or a string (maybe accept any interface{} and set that)
2. It is a bit unclear how to support watching for config updates, there are 2 options: a) extend the Converter to also accept a WatcherFunc; b) remove expand converter and embed this capability into our implementation of the ConfigProvider (allow both constructors NewConfigProvider and NewDefaultConfigProvider to allow this to be configured somehow).
3. The current "os.Expand" does not allow to return an error or to handle an error. It may require us to copy that or look for an alternative API.

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

config/configmapprovider/retrieved.go

@@ -0,0 +1,93 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package configmapprovider // import "go.opentelemetry.io/collector/config/configmapprovider"
+
+import (
+	"context"
+	"errors"
+
+	"go.opentelemetry.io/collector/config"
+)
+
+// Retrieved holds the result of a call to the Retrieve method of a Provider object.
+// This interface cannot be directly implemented. Implementations must use the NewRetrieved helper.
+type Retrieved interface {
+	// Get returns the config Map. Should never be called after Close.
+	// Should never be called concurrently with itself or Close.
+	Get(ctx context.Context) (interface{}, error)
+
+	// Close signals that the configuration for which it was used to retrieve values is
+	// no longer in use and should close and release any watchers that it may have created.
+	//
+	// Should block until all resources are closed, and guarantee that `onChange` is not
+	// going to be called after it returns except when `ctx` is cancelled.
+	//
+	// Should never be called concurrently with itself or Get.
+	Close(ctx context.Context) error
+
+	// privateRetrieved is an unexported func to disallow direct implementation.
+	privateRetrieved()
+}
+
+// GetFunc specifies the function invoked when the Retrieved.Get is being called.
+type GetFunc func(context.Context) (*config.Map, error)
+
+// Get implements the Retrieved.Get.
+func (f GetFunc) Get(ctx context.Context) (*config.Map, error) {
+	return f(ctx)
+}
+
+// CloseFunc specifies the function invoked when the Retrieved.Close is being called.
+type CloseFunc func(context.Context) error
+
+// Close implements the Retrieved.Close.
+func (f CloseFunc) Close(ctx context.Context) error {
+	if f == nil {
+		return nil
+	}
+	return f(ctx)
+}
+
+// RetrievedOption represents the possible options for NewRetrieved.
+type RetrievedOption func(*retrieved)
+
+// WithClose overrides the default `Close` function for a Retrieved.
+// The default always returns nil.
+func WithClose(closeFunc CloseFunc) RetrievedOption {
+	return func(o *retrieved) {
+		o.CloseFunc = closeFunc
+	}
+}
+
+type retrieved struct {
+	GetFunc
+	CloseFunc
+}
+
+func (retrieved) privateRetrieved() {}
+
+// NewRetrieved returns a Retrieved configured with the provided options.
+func NewRetrieved(getFunc GetFunc, options ...RetrievedOption) (Retrieved, error) {
+	if getFunc == nil {
+		return nil, errors.New("nil getFunc")
+	}
+	ret := &retrieved{
+		GetFunc: getFunc,
+	}
+	for _, op := range options {
+		op(ret)
+	}
+	return ret, nil
+}

config/mapconverter/expandmapconverter/expand.go

@@ -17,52 +17,67 @@ package expandmapconverter // import "go.opentelemetry.io/collector/config/mapco
 import (
 	"context"
 	"os"
+	"strings"
 
 	"go.opentelemetry.io/collector/config"
 )
 
 // New returns a config.MapConverterFunc, that expands all environment variables for a given config.Map.
 //
 // Notice: This API is experimental.
-func New() config.MapConverterFunc {
-	return func(_ context.Context, cfgMap *config.Map) error {
+func New(configMapProviders map[string]config.MapProvider) config.MapConverterFunc {
+	return func(ctx context.Context, cfgMap *config.Map) error {
 		for _, k := range cfgMap.AllKeys() {
-			cfgMap.Set(k, expandStringValues(cfgMap.Get(k)))
+			cfgMap.Set(k, expandStringValues(cfgMap.Get(k), func(location string) string {
+				// This allows escaping environment variable substitution via $$, e.g.
+				// - $FOO will be substituted with env var FOO
+				// - $$FOO will be replaced with $FOO
+				// - $$$FOO will be replaced with $ + substituted env var FOO
+				if location == "$" {
+					return "$"
+				}
+				scheme := "env"
+				if idx := strings.Index(location, ":"); idx != -1 {
+					scheme = location[:idx]
+				} else {
+					// MapProviders require the location to always have the scheme.
+					location = scheme + ":" + location
+				}
+				p, ok := configMapProviders[scheme]
+				// TODO: Figure out how to return an error when not registered.
+				if !ok {
+					return ""
+				}
+				// TODO: Figure out how to return an error when retrieve returns an error.
+				ret, _ := p.Retrieve(ctx, location, nil)
+				return cfg.(string)
+			}))
 		}
 		return nil
 	}
 }
 
-func expandStringValues(value interface{}) interface{} {
+func expandStringValues(value interface{}, mapping func(string) string) interface{} {
 	switch v := value.(type) {
 	case string:
-		return expandEnv(v)
+		return expandEnv(v, mapping)
 	case []interface{}:
 		nslice := make([]interface{}, 0, len(v))
 		for _, vint := range v {
-			nslice = append(nslice, expandStringValues(vint))
+			nslice = append(nslice, expandStringValues(vint, mapping))
 		}
 		return nslice
 	case map[string]interface{}:
 		nmap := map[string]interface{}{}
 		for mk, mv := range v {
-			nmap[mk] = expandStringValues(mv)
+			nmap[mk] = expandStringValues(mv, mapping)
 		}
 		return nmap
 	default:
 		return v
 	}
 }
 
-func expandEnv(s string) string {
-	return os.Expand(s, func(str string) string {
-		// This allows escaping environment variable substitution via $$, e.g.
-		// - $FOO will be substituted with env var FOO
-		// - $$FOO will be replaced with $FOO
-		// - $$$FOO will be replaced with $ + substituted env var FOO
-		if str == "$" {
-			return "$"
-		}
-		return os.Getenv(str)
-	})
+func expandEnv(s string, mapping func(string) string) string {
+	return os.Expand(s, mapping)
 }