Merge pull request #99 from nirmata/bucket-1.10

initial changes for bucket boundaries

pkg/config/metricsconfig.go

@@ -4,6 +4,8 @@ import (
 	"sync"
 	"time"
 
+	"go.opentelemetry.io/otel/attribute"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"golang.org/x/exp/slices"
 	corev1 "k8s.io/api/core/v1"
 )
@@ -18,6 +20,10 @@ type MetricsConfiguration interface {
 	GetMetricsRefreshInterval() time.Duration
 	// CheckNamespace returns `true` if the namespace has to be considered
 	CheckNamespace(string) bool
+	// GetBucketBoundaries returns the bucket boundaries for Histogram metrics
+	GetBucketBoundaries() []float64
+	// BuildMeterProviderViews returns OTL view removing attributes which were disabled in the config
+	BuildMeterProviderViews() []sdkmetric.View
 	// Load loads configuration from a configmap
 	Load(*corev1.ConfigMap)
 	// OnChanged adds a callback to be invoked when the configuration is reloaded
@@ -28,19 +34,17 @@ type MetricsConfiguration interface {
 type metricsConfig struct {
 	namespaces             namespacesConfig
 	metricsRefreshInterval time.Duration
+	bucketBoundaries       []float64
+	metricsExposure        map[string]metricExposureConfig
 	mux                    sync.RWMutex
 	callbacks              []func()
 }
 
 // NewDefaultMetricsConfiguration ...
 func NewDefaultMetricsConfiguration() *metricsConfig {
-	return &metricsConfig{
-		metricsRefreshInterval: 0,
-		namespaces: namespacesConfig{
-			IncludeNamespaces: []string{},
-			ExcludeNamespaces: []string{},
-		},
-	}
+	config := metricsConfig{}
+	config.reset()
+	return &config
 }
 
 func (cd *metricsConfig) OnChanged(callback func()) {
@@ -70,6 +74,42 @@ func (mcd *metricsConfig) GetMetricsRefreshInterval() time.Duration {
 	return mcd.metricsRefreshInterval
 }
 
+func (mcd *metricsConfig) GetBucketBoundaries() []float64 {
+	mcd.mux.RLock()
+	defer mcd.mux.RUnlock()
+	return mcd.bucketBoundaries
+}
+
+func (mcd *metricsConfig) BuildMeterProviderViews() []sdkmetric.View {
+	mcd.mux.RLock()
+	defer mcd.mux.RUnlock()
+	var views []sdkmetric.View
+	for key, value := range mcd.metricsExposure {
+		if *value.Enabled {
+			views = append(views, sdkmetric.NewView(
+				sdkmetric.Instrument{Name: key},
+				sdkmetric.Stream{
+					AttributeFilter: func(kv attribute.KeyValue) bool {
+						return !slices.Contains(value.DisabledLabelDimensions, string(kv.Key))
+					},
+					Aggregation: sdkmetric.AggregationExplicitBucketHistogram{
+						Boundaries: value.BucketBoundaries,
+						NoMinMax:   false,
+					},
+				},
+			))
+		} else if !*value.Enabled {
+			views = append(views, sdkmetric.NewView(
+				sdkmetric.Instrument{Name: key},
+				sdkmetric.Stream{
+					Aggregation: sdkmetric.AggregationDrop{},
+				},
+			))
+		}
+	}
+	return views
+}
+
 // CheckNamespace returns `true` if the namespace has to be considered
 func (mcd *metricsConfig) CheckNamespace(namespace string) bool {
 	mcd.mux.RLock()
@@ -105,11 +145,7 @@ func (cd *metricsConfig) load(cm *corev1.ConfigMap) {
 		data = map[string]string{}
 	}
 	// reset
-	cd.metricsRefreshInterval = 0
-	cd.namespaces = namespacesConfig{
-		IncludeNamespaces: []string{},
-		ExcludeNamespaces: []string{},
-	}
+	cd.reset()
 	// load metricsRefreshInterval
 	metricsRefreshInterval, ok := data["metricsRefreshInterval"]
 	if !ok {
@@ -138,17 +174,67 @@ func (cd *metricsConfig) load(cm *corev1.ConfigMap) {
 			logger.Info("namespaces configured")
 		}
 	}
+	// load bucket boundaries
+	bucketBoundariesString, ok := data["bucketBoundaries"]
+	if !ok {
+		logger.Info("bucketBoundaries not set")
+	} else {
+		logger := logger.WithValues("bucketBoundaries", bucketBoundariesString)
+		bucketBoundaries, err := parseBucketBoundariesConfig(bucketBoundariesString)
+		if err != nil {
+			logger.Error(err, "failed to parse bucketBoundariesString")
+		} else {
+			cd.bucketBoundaries = bucketBoundaries
+			logger.Info("bucketBoundaries configured")
+		}
+	}
+	// load include resource details
+	metricsExposureString, ok := data["metricsExposure"]
+	if !ok {
+		logger.Info("metricsExposure not set")
+	} else {
+		logger := logger.WithValues("metricsExposure", metricsExposureString)
+		metricsExposure, err := parseMetricExposureConfig(metricsExposureString, cd.bucketBoundaries)
+		if err != nil {
+			logger.Error(err, "failed to parse metricsExposure")
+		} else {
+			cd.metricsExposure = metricsExposure
+			logger.Info("metricsExposure configured")
+		}
+	}
 }
 
 func (mcd *metricsConfig) unload() {
 	mcd.mux.Lock()
 	defer mcd.mux.Unlock()
 	defer mcd.notify()
+	mcd.reset()
+}
+
+func (mcd *metricsConfig) reset() {
 	mcd.metricsRefreshInterval = 0
 	mcd.namespaces = namespacesConfig{
 		IncludeNamespaces: []string{},
 		ExcludeNamespaces: []string{},
 	}
+	mcd.bucketBoundaries = []float64{
+		0.005,
+		0.01,
+		0.025,
+		0.05,
+		0.1,
+		0.25,
+		0.5,
+		1,
+		2.5,
+		5,
+		10,
+		15,
+		20,
+		25,
+		30,
+	}
+	mcd.metricsExposure = map[string]metricExposureConfig{}
 }
 
 func (mcd *metricsConfig) notify() {

pkg/config/metricsconfig_test.go

@@ -0,0 +1,184 @@
+package config
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func boolPtr(b bool) *bool {
+	return &b
+}
+
+func Test_metricsConfig_load(t *testing.T) {
+	tests := []struct {
+		name          string
+		configMap     *corev1.ConfigMap
+		expectedValue *metricsConfig
+	}{
+		{
+			name: "Case 1: Test defaults",
+			configMap: &corev1.ConfigMap{
+				Data: map[string]string{},
+			},
+			expectedValue: &metricsConfig{
+				metricsRefreshInterval: 0,
+				namespaces:             namespacesConfig{IncludeNamespaces: []string{}, ExcludeNamespaces: []string{}},
+				bucketBoundaries:       []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 20, 25, 30},
+				metricsExposure:        map[string]metricExposureConfig{},
+			},
+		},
+		{
+			name: "Case 2: All fields provided",
+			configMap: &corev1.ConfigMap{
+				Data: map[string]string{
+					"metricsRefreshInterval": "10s",
+					"namespaces":             `{"include": ["namespace1"], "exclude": ["namespace2"]}`,
+					"bucketBoundaries":       "0.005, 0.01, 0.025, 0.05",
+					"metricsExposure":        `{"metric1": {"enabled": true, "disabledLabelDimensions": ["dim1"]}, "metric2": {"enabled": true, "disabledLabelDimensions": ["dim1","dim2"], "bucketBoundaries": [0.025, 0.05]}}`,
+				},
+			},
+			expectedValue: &metricsConfig{
+				metricsRefreshInterval: 10 * time.Second,
+				namespaces:             namespacesConfig{IncludeNamespaces: []string{"namespace1"}, ExcludeNamespaces: []string{"namespace2"}},
+				bucketBoundaries:       []float64{0.005, 0.01, 0.025, 0.05},
+				metricsExposure: map[string]metricExposureConfig{
+					"metric1": {Enabled: boolPtr(true), DisabledLabelDimensions: []string{"dim1"}, BucketBoundaries: []float64{0.005, 0.01, 0.025, 0.05}},
+					"metric2": {Enabled: boolPtr(true), DisabledLabelDimensions: []string{"dim1", "dim2"}, BucketBoundaries: []float64{0.025, 0.05}},
+				},
+			},
+		},
+		{
+			name: "Case 3: Some of the fields provided",
+			configMap: &corev1.ConfigMap{
+				Data: map[string]string{
+					"namespaces":      `{"include": ["namespace1"], "exclude": ["namespace2"]}`,
+					"metricsExposure": `{"metric1": {"enabled": true, "disabledLabelDimensions": ["dim1"]}, "metric2": {"enabled": true, "disabledLabelDimensions": ["dim1","dim2"], "bucketBoundaries": [0.025, 0.05]}}`,
+				},
+			},
+			expectedValue: &metricsConfig{
+				metricsRefreshInterval: 0,
+				namespaces:             namespacesConfig{IncludeNamespaces: []string{"namespace1"}, ExcludeNamespaces: []string{"namespace2"}},
+				bucketBoundaries:       []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 20, 25, 30},
+				metricsExposure: map[string]metricExposureConfig{
+					"metric1": {Enabled: boolPtr(true), DisabledLabelDimensions: []string{"dim1"}, BucketBoundaries: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 20, 25, 30}},
+					"metric2": {Enabled: boolPtr(true), DisabledLabelDimensions: []string{"dim1", "dim2"}, BucketBoundaries: []float64{0.025, 0.05}},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cd := NewDefaultMetricsConfiguration()
+			cd.load(tt.configMap)
+
+			if !reflect.DeepEqual(cd.metricsRefreshInterval, tt.expectedValue.metricsRefreshInterval) {
+				t.Errorf("Expected %+v, but got %+v", tt.expectedValue.metricsRefreshInterval, cd.metricsRefreshInterval)
+			}
+			if !reflect.DeepEqual(cd.namespaces, tt.expectedValue.namespaces) {
+				t.Errorf("Expected %+v, but got %+v", tt.expectedValue.namespaces, cd.namespaces)
+			}
+			if !reflect.DeepEqual(cd.bucketBoundaries, tt.expectedValue.bucketBoundaries) {
+				t.Errorf("Expected %+v, but got %+v", tt.expectedValue.bucketBoundaries, cd.bucketBoundaries)
+			}
+			if !reflect.DeepEqual(cd.metricsExposure, tt.expectedValue.metricsExposure) {
+				t.Errorf("Expected %+v, but got %+v", tt.expectedValue.metricsExposure, cd.metricsRefreshInterval)
+			}
+		})
+	}
+}
+
+func Test_metricsConfig_BuildMeterProviderViews(t *testing.T) {
+	tests := []struct {
+		name            string
+		metricsExposure map[string]metricExposureConfig
+		expectedSize    int
+		validateFunc    func([]sdkmetric.View) bool
+	}{
+		{
+			name:            "Case 1: defaults",
+			metricsExposure: map[string]metricExposureConfig{},
+			expectedSize:    0,
+		},
+		{
+			name: "Case 2: metrics enabled",
+			metricsExposure: map[string]metricExposureConfig{
+				"metric1": {Enabled: boolPtr(true), DisabledLabelDimensions: []string{"dim1"}, BucketBoundaries: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 20, 25, 30}},
+			},
+			expectedSize: 1,
+			validateFunc: func(views []sdkmetric.View) bool {
+				stream, _ := views[0](sdkmetric.Instrument{Name: "metric1"})
+				assert := stream.AttributeFilter(attribute.String("policy_validation_mode", ""))
+				assert = assert && !stream.AttributeFilter(attribute.String("dim1", ""))
+				assert = assert && reflect.DeepEqual(stream.Aggregation, sdkmetric.AggregationExplicitBucketHistogram{
+					Boundaries: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 20, 25, 30},
+					NoMinMax:   false,
+				})
+				return assert
+			},
+		},
+		{
+			name: "Case 3: metrics disabled",
+			metricsExposure: map[string]metricExposureConfig{
+				"metric1": {Enabled: boolPtr(false)},
+			},
+			expectedSize: 1,
+			validateFunc: func(views []sdkmetric.View) bool {
+				stream, _ := views[0](sdkmetric.Instrument{Name: "metric1"})
+				return reflect.DeepEqual(stream.Aggregation, sdkmetric.AggregationDrop{})
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mcd := NewDefaultMetricsConfiguration()
+			mcd.metricsExposure = tt.metricsExposure
+			got := mcd.BuildMeterProviderViews()
+			if len(got) != tt.expectedSize {
+				t.Errorf("Expected result size to be %v, but got %v", tt.expectedSize, len(got))
+			}
+			if tt.validateFunc != nil {
+				if !tt.validateFunc(got) {
+					t.Errorf("The validation function did not return true!")
+				}
+			}
+		})
+	}
+}
+
+func Test_metricsConfig_GetBucketBoundaries(t *testing.T) {
+	tests := []struct {
+		name     string
+		provided []float64
+		want     []float64
+	}{
+		{
+			name:     "Case 1: Test defaults",
+			provided: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 20, 25, 30},
+			want:     []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 20, 25, 30},
+		},
+		{
+			name:     "Case 2: Custom",
+			provided: []float64{0.005, 0.01, 0.025, 0.05},
+			want:     []float64{0.005, 0.01, 0.025, 0.05},
+		},
+		{
+			name:     "Case 3: Empty",
+			provided: []float64{},
+			want:     []float64{},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mcd := NewDefaultMetricsConfiguration()
+			mcd.bucketBoundaries = tt.provided
+			if got := mcd.GetBucketBoundaries(); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("GetBucketBoundaries() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}