Add change handler to register callbacks (open-telemetry#1292)

* config: move onChange callback list methods into a thread-safe wrapper

Signed-off-by: Benedikt Bongartz <bongartz@klimlive.de>

* config: keep change handler private

Signed-off-by: Benedikt Bongartz <bongartz@klimlive.de>

* follow naming recommendations

Signed-off-by: Benedikt Bongartz <bongartz@klimlive.de>

Signed-off-by: Benedikt Bongartz <bongartz@klimlive.de>

internal/config/change_handler.go

@@ -0,0 +1,63 @@
+// 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 config contains the operator's runtime configuration.
+package config
+
+import (
+	"sync"
+
+	"github.com/go-logr/logr"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+// changeHandler is implemented by any structure that is able to register callbacks
+// and call them using one single method.
+type changeHandler interface {
+	// Do will call every registered callback.
+	Do() error
+	// Register this function as a callback that will be executed when Do() is called.
+	Register(f func() error)
+}
+
+// newOnChange returns a thread-safe ChangeHandler.
+func newOnChange() changeHandler {
+	return &onChange{
+		logger: logf.Log.WithName("change-handler"),
+	}
+}
+
+type onChange struct {
+	logger logr.Logger
+
+	callbacks   []func() error
+	muCallbacks sync.Mutex
+}
+
+func (o *onChange) Do() error {
+	o.muCallbacks.Lock()
+	defer o.muCallbacks.Unlock()
+	for _, fn := range o.callbacks {
+		if err := fn(); err != nil {
+			o.logger.Error(err, "change callback failed")
+		}
+	}
+	return nil
+}
+
+func (o *onChange) Register(f func() error) {
+	o.muCallbacks.Lock()
+	defer o.muCallbacks.Unlock()
+	o.callbacks = append(o.callbacks, f)
+}

internal/config/change_handler_test.go

@@ -0,0 +1,41 @@
+// 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 config contains the operator's runtime configuration.
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestChangeHandler(t *testing.T) {
+	// prepare
+	internal := 0
+	callback := func() error {
+		internal += 1
+		return nil
+	}
+	h := newOnChange()
+
+	h.Register(callback)
+
+	for i := 0; i < 5; i++ {
+		assert.Equal(t, i, internal)
+		require.NoError(t, h.Do())
+		assert.Equal(t, i+1, internal)
+	}
+}

internal/config/main.go

@@ -44,7 +44,7 @@ type Config struct {
 	targetAllocatorConfigMapEntry  string
 	autoInstrumentationNodeJSImage string
 	autoInstrumentationJavaImage   string
-	onChange                       []func() error
+	onPlatformChange               changeHandler
 	labelsFilter                   []string
 	platform                       platform.Platform
 	autoDetectFrequency            time.Duration
@@ -62,6 +62,7 @@ func New(opts ...Option) Config {
 		platform:                      platform.Unknown,
 		version:                       version.Get(),
 		autoscalingVersion:            autodetect.DefaultAutoscalingVersion,
+		onPlatformChange:              newOnChange(),
 	}
 	for _, opt := range opts {
 		opt(&o)
@@ -75,7 +76,7 @@ func New(opts ...Option) Config {
 		targetAllocatorImage:           o.targetAllocatorImage,
 		targetAllocatorConfigMapEntry:  o.targetAllocatorConfigMapEntry,
 		logger:                         o.logger,
-		onChange:                       o.onChange,
+		onPlatformChange:               o.onPlatformChange,
 		platform:                       o.platform,
 		autoInstrumentationJavaImage:   o.autoInstrumentationJavaImage,
 		autoInstrumentationNodeJSImage: o.autoInstrumentationNodeJSImage,
@@ -125,12 +126,9 @@ func (c *Config) AutoDetect() error {
 	}
 
 	if changed {
-		for _, callback := range c.onChange {
-			if err := callback(); err != nil {
-				// we don't fail if the callback failed, as the auto-detection itself
-				// did work
-				c.logger.Error(err, "configuration change notification failed for callback")
-			}
+		if err := c.onPlatformChange.Do(); err != nil {
+			// Don't fail if the callback failed, as auto-detection itself worked.
+			c.logger.Error(err, "configuration change notification failed for callback")
 		}
 	}
 
@@ -198,3 +196,9 @@ func (c *Config) AutoInstrumentationDotNetImage() string {
 func (c *Config) LabelsFilter() []string {
 	return c.labelsFilter
 }
+
+// RegisterPlatformChangeCallback registers the given function as a callback that
+// is called when the platform detection detects a change.
+func (c *Config) RegisterPlatformChangeCallback(f func() error) {
+	c.onPlatformChange.Register(f)
+}

internal/config/main_test.go

@@ -41,7 +41,7 @@ func TestNewConfig(t *testing.T) {
 	assert.Equal(t, platform.Kubernetes, cfg.Platform())
 }
 
-func TestCallbackOnChanges(t *testing.T) {
+func TestOnPlatformChangeCallback(t *testing.T) {
 	// prepare
 	calledBack := false
 	mock := &mockAutoDetect{
@@ -51,7 +51,7 @@ func TestCallbackOnChanges(t *testing.T) {
 	}
 	cfg := config.New(
 		config.WithAutoDetect(mock),
-		config.WithOnChange(func() error {
+		config.WithOnPlatformChangeCallback(func() error {
 			calledBack = true
 			return nil
 		}),

internal/config/options.go

@@ -41,7 +41,7 @@ type options struct {
 	collectorConfigMapEntry        string
 	targetAllocatorConfigMapEntry  string
 	targetAllocatorImage           string
-	onChange                       []func() error
+	onPlatformChange               changeHandler
 	labelsFilter                   []string
 	platform                       platform.Platform
 	autoDetectFrequency            time.Duration
@@ -84,12 +84,12 @@ func WithLogger(logger logr.Logger) Option {
 		o.logger = logger
 	}
 }
-func WithOnChange(f func() error) Option {
+func WithOnPlatformChangeCallback(f func() error) Option {
 	return func(o *options) {
-		if o.onChange == nil {
-			o.onChange = []func() error{}
+		if o.onPlatformChange == nil {
+			o.onPlatformChange = newOnChange()
 		}
-		o.onChange = append(o.onChange, f)
+		o.onPlatformChange.Register(f)
 	}
 }
 func WithPlatform(plt platform.Platform) Option {