Don't apply unnecessary otel config updates (#9985) (#10098)

* Don't apply unnecessary otel config updates

* Add a comment explaining the confmap hashing logic

* Add debug log when we skip config application

* Use real otel config in hashing test

* Remove outdated comment

* Switch the hash function to FNV128

* Use bytes.Equal instead of bytes.Compare

(cherry picked from commit d19d1b5)

Co-authored-by: Mikołaj Świątek <[email protected]>

Author: mergify[bot]

internal/pkg/otel/manager/manager.go

@@ -5,9 +5,12 @@
 package manager
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"hash/fnv"
 	"os"
 	"sync"
 	"sync/atomic"
@@ -70,7 +73,8 @@ type OTelManager struct {
 
 	// The current configuration that the OTel collector is using. In the case that
 	// the mergedCollectorCfg is nil then the collector is not running.
-	mergedCollectorCfg *confmap.Conf
+	mergedCollectorCfg     *confmap.Conf
+	mergedCollectorCfgHash []byte
 
 	currentCollectorStatus *status.AggregateStatus
 	currentComponentStates map[string]runtime.ComponentComponentState
@@ -282,14 +286,30 @@ func (m *OTelManager) Run(ctx context.Context) error {
 
 			// this is the only place where we mutate the internal config attributes, take a write lock for the duration
 			m.mx.Lock()
-			m.mergedCollectorCfg = mergedCfg
+			previousConfigHash := m.mergedCollectorCfgHash
+			configChanged, configUpdateErr := m.maybeUpdateMergedConfig(mergedCfg)
 			m.collectorCfg = cfgUpdate.collectorCfg
 			m.components = cfgUpdate.components
 			m.mx.Unlock()
 
-			err = m.applyMergedConfig(ctx, collectorStatusCh, m.collectorRunErr)
-			// report the error unconditionally to indicate that the config was applied
-			reportErr(ctx, m.errCh, err)
+			if configUpdateErr != nil {
+				m.logger.Warn("failed to calculate hash of merged config, proceeding with update", zap.Error(configUpdateErr))
+			}
+
+			if configChanged {
+				m.logger.Debugf(
+					"new config hash (%d) is different than the old config hash (%d), applying update",
+					m.mergedCollectorCfgHash, previousConfigHash)
+				applyErr := m.applyMergedConfig(ctx, collectorStatusCh, m.collectorRunErr)
+				// only report the error if we actually apply the update
+				// otherwise, we could override an actual error with a nil in the channel when the collector
+				// state doesn't actually change
+				reportErr(ctx, m.errCh, applyErr)
+			} else {
+				m.logger.Debugf(
+					"new config hash (%d) is identical to the old config hash (%d), skipping update",
+					m.mergedCollectorCfgHash, previousConfigHash)
+			}
 
 		case otelStatus := <-collectorStatusCh:
 			err = m.reportOtelStatusUpdate(ctx, otelStatus)
@@ -513,6 +533,19 @@ func (m *OTelManager) processComponentStates(componentStates []runtime.Component
 	return componentStates
 }
 
+// maybeUpdateMergedConfig updates the merged config if it's different from the current value. It checks this by
+// calculating a hash and comparing. It returns a value indicating if the configuration was updated.
+// If an error is encountered when calculating the hash, this will always be true.
+func (m *OTelManager) maybeUpdateMergedConfig(mergedCfg *confmap.Conf) (updated bool, err error) {
+	// if we get an error here, we just proceed with the update, worst that can happen is that we reload unnecessarily
+	mergedCfgHash, err := calculateConfmapHash(mergedCfg)
+	previousConfigHash := m.mergedCollectorCfgHash
+
+	m.mergedCollectorCfg = mergedCfg
+	m.mergedCollectorCfgHash = mergedCfgHash
+	return !bytes.Equal(mergedCfgHash, previousConfigHash) || err != nil, err
+}
+
 // reportComponentStateUpdates sends component state updates to the component watch channel. It first drains
 // the channel to ensure that only the most recent status is kept, as intermediate statuses can be safely discarded.
 // This ensures the receiver always observes the latest reported status.
@@ -532,3 +565,28 @@ func (m *OTelManager) reportComponentStateUpdates(ctx context.Context, component
 		return
 	}
 }
+
+// calculateConfmapHash calculates a hash of a given configuration. It's optimized for speed, which is why it
+// json encodes the values directly into a xxhash instance, instead of converting to a map[string]any first.
+func calculateConfmapHash(conf *confmap.Conf) ([]byte, error) {
+	if conf == nil {
+		return nil, nil
+	}
+
+	h := fnv.New128()
+	// We encode the configuration to json instead of yaml, because it's simpler and more performant.
+	// In general otel configuration can be marshalled to any format supported by koanf, but the confmap
+	// API doesn't expose this. This is why the small workaround below to avoid converting to a Go map is necessary.
+	encoder := json.NewEncoder(h)
+
+	for _, key := range conf.AllKeys() { // this is a sorted list, so the output is consistent
+		if err := encoder.Encode(key); err != nil {
+			return nil, err
+		}
+		if err := encoder.Encode(conf.Get(key)); err != nil {
+			return nil, err
+		}
+	}
+
+	return h.Sum(nil), nil
+}

internal/pkg/otel/manager/manager_test.go

@@ -53,6 +53,9 @@ var (
 					"level":   "none",
 					"readers": []any{},
 				},
+				"logs": map[string]interface{}{
+					"level": "info",
+				},
 			},
 			"pipelines": map[string]interface{}{
 				"traces": map[string]interface{}{
@@ -311,8 +314,10 @@ func TestOTelManager_Run(t *testing.T) {
 				m.Update(cfg, nil)
 				e.EnsureHealthy(t, updateTime)
 
-				// trigger update (no config compare is due externally to otel collector)
+				// trigger update
 				updateTime = time.Now()
+				ok := cfg.Delete("service::telemetry::logs::level") // modify the config
+				require.True(t, ok)
 				m.Update(cfg, nil)
 				e.EnsureHealthy(t, updateTime)
 
@@ -336,8 +341,10 @@ func TestOTelManager_Run(t *testing.T) {
 				m.Update(cfg, nil)
 				e.EnsureHealthy(t, updateTime)
 
-				// trigger update (no config compare is due externally to otel collector)
+				// trigger update
 				updateTime = time.Now()
+				ok := cfg.Delete("service::telemetry::logs::level") // modify the config
+				require.True(t, ok)
 				m.Update(cfg, nil)
 				e.EnsureHealthy(t, updateTime)
 
@@ -1361,3 +1368,160 @@ func assertOtelStatusesEqualIgnoringTimestamps(t require.TestingT, a, b *status.
 
 	return true
 }
+
+func TestCalculateConfmapHash(t *testing.T) {
+	t.Run("nil config returns zero", func(t *testing.T) {
+		hash, err := calculateConfmapHash(nil)
+		require.NoError(t, err)
+		assert.Equal(t, []byte(nil), hash)
+	})
+
+	t.Run("same value gives same result", func(t *testing.T) {
+		conf := confmap.NewFromStringMap(map[string]any{
+			"key1": "value1",
+			"key2": 123,
+		})
+		hash1, err := calculateConfmapHash(conf)
+		require.NoError(t, err)
+		hash2, err := calculateConfmapHash(conf)
+		require.NoError(t, err)
+		assert.Equal(t, hash1, hash2)
+	})
+
+	t.Run("different values give different results", func(t *testing.T) {
+		conf1 := confmap.NewFromStringMap(map[string]any{
+			"key1": "value1",
+		})
+		hash1, err := calculateConfmapHash(conf1)
+		require.NoError(t, err)
+
+		conf2 := confmap.NewFromStringMap(map[string]any{
+			"key1": "value2",
+		})
+		hash2, err := calculateConfmapHash(conf2)
+		require.NoError(t, err)
+
+		assert.NotEqual(t, hash1, hash2)
+	})
+
+	t.Run("list of maps is processed correctly", func(t *testing.T) {
+		conf1 := confmap.NewFromStringMap(map[string]any{
+			"items": []any{
+				map[string]any{"name": "A", "value": 1},
+				map[string]any{"name": "B", "value": 2},
+			},
+		})
+		hash1, err := calculateConfmapHash(conf1)
+		require.NoError(t, err)
+
+		t.Run("same list of maps gives same hash", func(t *testing.T) {
+			conf2 := confmap.NewFromStringMap(map[string]any{
+				"items": []any{
+					map[string]any{"name": "A", "value": 1},
+					map[string]any{"name": "B", "value": 2},
+				},
+			})
+			hash2, err := calculateConfmapHash(conf2)
+			require.NoError(t, err)
+			assert.Equal(t, hash1, hash2)
+		})
+
+		t.Run("different order in list gives different hash", func(t *testing.T) {
+			conf3 := confmap.NewFromStringMap(map[string]any{
+				"items": []any{
+					map[string]any{"name": "B", "value": 2},
+					map[string]any{"name": "A", "value": 1},
+				},
+			})
+			hash3, err := calculateConfmapHash(conf3)
+			require.NoError(t, err)
+			assert.NotEqual(t, hash1, hash3)
+		})
+	})
+}
+
+func TestOTelManager_maybeUpdateMergedConfig(t *testing.T) {
+	t.Run("initial config", func(t *testing.T) {
+		m := &OTelManager{}
+		conf := confmap.NewFromStringMap(testConfig)
+
+		updated, err := m.maybeUpdateMergedConfig(conf)
+
+		require.NoError(t, err)
+		assert.True(t, updated)
+		assert.Equal(t, conf, m.mergedCollectorCfg)
+		assert.NotEqual(t, uint64(0), m.mergedCollectorCfgHash)
+	})
+
+	t.Run("same config", func(t *testing.T) {
+		conf := confmap.NewFromStringMap(testConfig)
+		hash, err := calculateConfmapHash(conf)
+		require.NoError(t, err)
+
+		m := &OTelManager{
+			mergedCollectorCfg:     conf,
+			mergedCollectorCfgHash: hash,
+		}
+
+		updated, err := m.maybeUpdateMergedConfig(conf)
+
+		require.NoError(t, err)
+		assert.False(t, updated)
+		assert.Equal(t, conf, m.mergedCollectorCfg)
+		assert.Equal(t, hash, m.mergedCollectorCfgHash)
+	})
+
+	t.Run("different config", func(t *testing.T) {
+		conf1 := confmap.NewFromStringMap(map[string]any{"key": "value1"})
+		hash1, err := calculateConfmapHash(conf1)
+		require.NoError(t, err)
+
+		m := &OTelManager{
+			mergedCollectorCfg:     conf1,
+			mergedCollectorCfgHash: hash1,
+		}
+
+		conf2 := confmap.NewFromStringMap(map[string]any{"key": "value2"})
+		hash2, err := calculateConfmapHash(conf2)
+		require.NoError(t, err)
+
+		updated, err := m.maybeUpdateMergedConfig(conf2)
+
+		require.NoError(t, err)
+		assert.True(t, updated)
+		assert.Equal(t, conf2, m.mergedCollectorCfg)
+		assert.Equal(t, hash2, m.mergedCollectorCfgHash)
+		assert.NotEqual(t, hash1, m.mergedCollectorCfgHash)
+	})
+
+	t.Run("hashing error with previous config", func(t *testing.T) {
+		conf1 := confmap.NewFromStringMap(map[string]any{"key": "value1"})
+		hash1, err := calculateConfmapHash(conf1)
+		require.NoError(t, err)
+
+		m := &OTelManager{
+			mergedCollectorCfg:     conf1,
+			mergedCollectorCfgHash: hash1,
+		}
+
+		badConf := confmap.NewFromStringMap(map[string]any{"bad": make(chan int)})
+		updated, err := m.maybeUpdateMergedConfig(badConf)
+
+		require.Error(t, err)
+		assert.True(t, updated, "update should proceed on hashing error")
+		assert.Equal(t, badConf, m.mergedCollectorCfg)
+		assert.Equal(t, []byte(nil), m.mergedCollectorCfgHash)
+	})
+
+	t.Run("hashing error with no previous config", func(t *testing.T) {
+		m := &OTelManager{}
+
+		badConf := confmap.NewFromStringMap(map[string]any{"bad": make(chan int)})
+		updated, err := m.maybeUpdateMergedConfig(badConf)
+
+		require.Error(t, err)
+		assert.True(t, updated, "update should proceed on hashing error, even with no previous config")
+		assert.Equal(t, badConf, m.mergedCollectorCfg)
+		assert.Equal(t, []byte(nil), m.mergedCollectorCfgHash)
+	})
+}