Refactor configmapprovider.Provider and Retrieved interfaces

This uses a callback instead of a blocking WaitForUpdate function to notify about config changes. This simplifies the usage of the Provider, especially when watching for changes is required. A follow up PR will add watching capabilities to existing Provider implementations.
open-telemetry · Nov 11, 2021 · 3d76ad1 · 3d76ad1
1 parent 93c9cab
commit 3d76ad1
Show file tree

Hide file tree

Showing 15 changed files with 186 additions and 121 deletions.
diff --git a/config/configmapprovider/default_test.go b/config/configmapprovider/default_test.go
@@ -27,7 +27,7 @@ import (
 
 func TestDefaultMapProvider(t *testing.T) {
 	mp := NewDefault("testdata/default-config.yaml", nil)
-	retr, err := mp.Retrieve(context.Background())
+	retr, err := mp.Retrieve(context.Background(), nil)
 	require.NoError(t, err)
 
 	expectedMap, err := config.NewMapFromBuffer(strings.NewReader(`
@@ -37,14 +37,16 @@ exporters:
   otlp:
     endpoint: "localhost:4317"`))
 	require.NoError(t, err)
-	assert.Equal(t, expectedMap, retr.Get())
+	m, err := retr.Get(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, expectedMap, m)
 
-	assert.NoError(t, mp.Close(context.Background()))
+	assert.NoError(t, mp.Shutdown(context.Background()))
 }
 
 func TestDefaultMapProvider_AddNewConfig(t *testing.T) {
 	mp := NewDefault("testdata/default-config.yaml", []string{"processors.batch.timeout=2s"})
-	cp, err := mp.Retrieve(context.Background())
+	cp, err := mp.Retrieve(context.Background(), nil)
 	require.NoError(t, err)
 
 	expectedMap, err := config.NewMapFromBuffer(strings.NewReader(`
@@ -55,16 +57,18 @@ exporters:
   otlp:
     endpoint: "localhost:4317"`))
 	require.NoError(t, err)
-	assert.Equal(t, expectedMap, cp.Get())
+	m, err := cp.Get(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, expectedMap, m)
 
-	assert.NoError(t, mp.Close(context.Background()))
+	assert.NoError(t, mp.Shutdown(context.Background()))
 }
 
 func TestDefaultMapProvider_OverwriteConfig(t *testing.T) {
 	mp := NewDefault(
 		"testdata/default-config.yaml",
 		[]string{"processors.batch.timeout=2s", "exporters.otlp.endpoint=localhost:1234"})
-	cp, err := mp.Retrieve(context.Background())
+	cp, err := mp.Retrieve(context.Background(), nil)
 	require.NoError(t, err)
 
 	expectedMap, err := config.NewMapFromBuffer(strings.NewReader(`
@@ -75,24 +79,26 @@ exporters:
   otlp:
     endpoint: "localhost:1234"`))
 	require.NoError(t, err)
-	assert.Equal(t, expectedMap, cp.Get())
+	m, err := cp.Get(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, expectedMap, m)
 
-	assert.NoError(t, mp.Close(context.Background()))
+	assert.NoError(t, mp.Shutdown(context.Background()))
 }
 
 func TestDefaultMapProvider_InexistentFile(t *testing.T) {
 	mp := NewDefault("testdata/otelcol-config.yaml", nil)
 	require.NotNil(t, mp)
-	_, err := mp.Retrieve(context.Background())
+	_, err := mp.Retrieve(context.Background(), nil)
 	require.Error(t, err)
 
-	assert.NoError(t, mp.Close(context.Background()))
+	assert.NoError(t, mp.Shutdown(context.Background()))
 }
 
 func TestDefaultMapProvider_EmptyFileName(t *testing.T) {
 	mp := NewDefault("", nil)
-	_, err := mp.Retrieve(context.Background())
+	_, err := mp.Retrieve(context.Background(), nil)
 	require.Error(t, err)
 
-	assert.NoError(t, mp.Close(context.Background()))
+	assert.NoError(t, mp.Shutdown(context.Background()))
 }
diff --git a/config/configmapprovider/expand.go b/config/configmapprovider/expand.go
@@ -31,20 +31,23 @@ func NewExpand(base Provider) Provider {
 	}
 }
 
-func (emp *expandMapProvider) Retrieve(ctx context.Context) (Retrieved, error) {
-	retr, err := emp.base.Retrieve(ctx)
+func (emp *expandMapProvider) Retrieve(ctx context.Context, onChange func(*ChangeEvent)) (Retrieved, error) {
+	retr, err := emp.base.Retrieve(ctx, onChange)
+	if err != nil {
+		return nil, err
+	}
+	cfgMap, err := retr.Get(ctx)
 	if err != nil {
 		return nil, err
 	}
-	cfgMap := retr.Get()
 	for _, k := range cfgMap.AllKeys() {
 		cfgMap.Set(k, expandStringValues(cfgMap.Get(k)))
 	}
 	return &simpleRetrieved{confMap: cfgMap}, nil
 }
 
-func (emp *expandMapProvider) Close(ctx context.Context) error {
-	return emp.base.Close(ctx)
+func (emp *expandMapProvider) Shutdown(ctx context.Context) error {
+	return emp.base.Shutdown(ctx)
 }
 
 func expandStringValues(value interface{}) interface{} {

diff --git a/config/configmapprovider/expand_test.go b/config/configmapprovider/expand_test.go
@@ -57,11 +57,13 @@ func TestExpand(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			// Retrieve the config
 			emp := NewExpand(NewFile(path.Join("testdata", test.name)))
-			cp, err := emp.Retrieve(context.Background())
+			cp, err := emp.Retrieve(context.Background(), nil)
 			require.NoError(t, err, "Unable to get config")
 
 			// Test that expanded configs are the same with the simple config with no env vars.
-			assert.Equal(t, expectedCfgMap.ToStringMap(), cp.Get().ToStringMap())
+			m, err := cp.Get(context.Background())
+			require.NoError(t, err)
+			assert.Equal(t, expectedCfgMap.ToStringMap(), m.ToStringMap())
 		})
 	}
 }
@@ -75,7 +77,7 @@ func TestExpand_EscapedEnvVars(t *testing.T) {
 
 	// Retrieve the config
 	emp := NewExpand(NewFile(path.Join("testdata", "expand-escaped-env.yaml")))
-	cp, err := emp.Retrieve(context.Background())
+	cp, err := emp.Retrieve(context.Background(), nil)
 	require.NoError(t, err, "Unable to get config")
 
 	expectedMap := map[string]interface{}{
@@ -95,5 +97,7 @@ func TestExpand_EscapedEnvVars(t *testing.T) {
 			// escaped $ alone
 			"recv.7": "$",
 		}}
-	assert.Equal(t, expectedMap, cp.Get().ToStringMap())
+	m, err := cp.Get(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, expectedMap, m.ToStringMap())
 }
diff --git a/config/configmapprovider/file.go b/config/configmapprovider/file.go
@@ -33,7 +33,7 @@ func NewFile(fileName string) Provider {
 	}
 }
 
-func (fmp *fileMapProvider) Retrieve(context.Context) (Retrieved, error) {
+func (fmp *fileMapProvider) Retrieve(_ context.Context, _ func(*ChangeEvent)) (Retrieved, error) {
 	if fmp.fileName == "" {
 		return nil, errors.New("config file not specified")
 	}
@@ -46,6 +46,6 @@ func (fmp *fileMapProvider) Retrieve(context.Context) (Retrieved, error) {
 	return &simpleRetrieved{confMap: cp}, nil
 }
 
-func (*fileMapProvider) Close(context.Context) error {
+func (*fileMapProvider) Shutdown(context.Context) error {
 	return nil
 }
diff --git a/config/configmapprovider/inmemory.go b/config/configmapprovider/inmemory.go
@@ -30,14 +30,14 @@ func NewInMemory(buf io.Reader) Provider {
 	return &inMemoryMapProvider{buf: buf}
 }
 
-func (inp *inMemoryMapProvider) Retrieve(context.Context) (Retrieved, error) {
+func (inp *inMemoryMapProvider) Retrieve(_ context.Context, onChange func(*ChangeEvent)) (Retrieved, error) {
 	cfg, err := config.NewMapFromBuffer(inp.buf)
 	if err != nil {
 		return nil, err
 	}
 	return &simpleRetrieved{confMap: cfg}, nil
 }
 
-func (inp *inMemoryMapProvider) Close(context.Context) error {
+func (inp *inMemoryMapProvider) Shutdown(context.Context) error {
 	return nil
 }
diff --git a/config/configmapprovider/merge.go b/config/configmapprovider/merge.go
@@ -35,24 +35,28 @@ func NewMerge(ps ...Provider) Provider {
 	return &mergeMapProvider{providers: ps}
 }
 
-func (mp *mergeMapProvider) Retrieve(ctx context.Context) (Retrieved, error) {
+func (mp *mergeMapProvider) Retrieve(ctx context.Context, onChange func(*ChangeEvent)) (Retrieved, error) {
 	retCfgMap := config.NewMap()
 	for _, p := range mp.providers {
-		retr, err := p.Retrieve(ctx)
+		retr, err := p.Retrieve(ctx, onChange)
 		if err != nil {
 			return nil, err
 		}
-		if err = retCfgMap.Merge(retr.Get()); err != nil {
+		cfgMap, err := retr.Get(ctx)
+		if err != nil {
+			return nil, err
+		}
+		if err = retCfgMap.Merge(cfgMap); err != nil {
 			return nil, err
 		}
 	}
 	return &simpleRetrieved{confMap: retCfgMap}, nil
 }
 
-func (mp *mergeMapProvider) Close(ctx context.Context) error {
+func (mp *mergeMapProvider) Shutdown(ctx context.Context) error {
 	var errs error
 	for _, p := range mp.providers {
-		errs = multierr.Append(errs, p.Close(ctx))
+		errs = multierr.Append(errs, p.Shutdown(ctx))
 	}
 
 	return errs

diff --git a/config/configmapprovider/merge_test.go b/config/configmapprovider/merge_test.go
@@ -28,28 +28,28 @@ import (
 func TestMerge_GetError(t *testing.T) {
 	pl := NewMerge(&errProvider{err: nil}, &errProvider{errors.New("my error")})
 	require.NotNil(t, pl)
-	cp, err := pl.Retrieve(context.Background())
+	cp, err := pl.Retrieve(context.Background(), nil)
 	assert.Error(t, err)
 	assert.Nil(t, cp)
 }
 
 func TestMerge_CloseError(t *testing.T) {
 	pl := NewMerge(&errProvider{err: nil}, &errProvider{errors.New("my error")})
 	require.NotNil(t, pl)
-	assert.Error(t, pl.Close(context.Background()))
+	assert.Error(t, pl.Shutdown(context.Background()))
 }
 
 type errProvider struct {
 	err error
 }
 
-func (epl *errProvider) Retrieve(context.Context) (Retrieved, error) {
+func (epl *errProvider) Retrieve(context.Context, func(*ChangeEvent)) (Retrieved, error) {
 	if epl.err == nil {
 		return &simpleRetrieved{confMap: config.NewMap()}, nil
 	}
 	return nil, epl.err
 }
 
-func (epl *errProvider) Close(context.Context) error {
+func (epl *errProvider) Shutdown(context.Context) error {
 	return epl.err
 }
diff --git a/config/configmapprovider/properties.go b/config/configmapprovider/properties.go
@@ -40,7 +40,7 @@ func NewProperties(properties []string) Provider {
 	}
 }
 
-func (pmp *propertiesMapProvider) Retrieve(context.Context) (Retrieved, error) {
+func (pmp *propertiesMapProvider) Retrieve(_ context.Context, onChange func(*ChangeEvent)) (Retrieved, error) {
 	if len(pmp.properties) == 0 {
 		return &simpleRetrieved{confMap: config.NewMap()}, nil
 	}
@@ -70,6 +70,6 @@ func (pmp *propertiesMapProvider) Retrieve(context.Context) (Retrieved, error) {
 	return &simpleRetrieved{confMap: config.NewMapFromStringMap(prop)}, nil
 }
 
-func (*propertiesMapProvider) Close(context.Context) error {
+func (*propertiesMapProvider) Shutdown(context.Context) error {
 	return nil
 }
diff --git a/config/configmapprovider/properties_test.go b/config/configmapprovider/properties_test.go
@@ -31,23 +31,25 @@ func TestPropertiesProvider(t *testing.T) {
 	}
 
 	pmp := NewProperties(setFlagStr)
-	retr, err := pmp.Retrieve(context.Background())
+	retr, err := pmp.Retrieve(context.Background(), nil)
+	require.NoError(t, err)
+	cfgMap, err := retr.Get(context.Background())
 	require.NoError(t, err)
-	cfgMap := retr.Get()
 	keys := cfgMap.AllKeys()
 	assert.Len(t, keys, 4)
 	assert.Equal(t, "2s", cfgMap.Get("processors::batch::timeout"))
 	assert.Equal(t, "3s", cfgMap.Get("processors::batch/foo::timeout"))
 	assert.Equal(t, "foo:9200,foo2:9200", cfgMap.Get("exporters::kafka::brokers"))
 	assert.Equal(t, "localhost:1818", cfgMap.Get("receivers::otlp::protocols::grpc::endpoint"))
-	require.NoError(t, pmp.Close(context.Background()))
+	require.NoError(t, pmp.Shutdown(context.Background()))
 }
 
 func TestPropertiesProvider_empty(t *testing.T) {
 	pmp := NewProperties(nil)
-	retr, err := pmp.Retrieve(context.Background())
+	retr, err := pmp.Retrieve(context.Background(), nil)
+	require.NoError(t, err)
+	cfgMap, err := retr.Get(context.Background())
 	require.NoError(t, err)
-	cfgMap := retr.Get()
 	assert.Equal(t, 0, len(cfgMap.AllKeys()))
-	require.NoError(t, pmp.Close(context.Background()))
+	require.NoError(t, pmp.Shutdown(context.Background()))
 }