Make the default reported value and error configurable (#85)

Signed-off-by: Jirka Kremser <jiri.kremser@gmail.com>

Author: jkremser

helmchart/otel-add-on/templates/deployment.yaml

@@ -46,6 +46,10 @@ spec:
               value: {{ .Values.settings.metricStore.lazySeries | quote }}
             - name: METRIC_STORE_LAZY_AGGREGATES
               value: {{ .Values.settings.metricStore.lazyAggregates | quote }}
+            - name: METRIC_STORE_ERROR_IF_NOT_FOUND
+              value: {{ .Values.settings.metricStore.errIfNotFound | quote }}
+            - name: METRIC_STORE_VALUE_IF_NOT_FOUND
+              value: {{ .Values.settings.metricStore.valueIfNotFound | quote }}
             - name: IS_ACTIVE_POLLING_INTERVAL_MS
               value: {{ .Values.settings.isActivePollingIntervalMilliseconds | quote }}
             - name: INTERNAL_METRICS_PORT

helmchart/otel-add-on/values.yaml

@@ -24,6 +24,12 @@ settings:
     #  (by default, we calculate and store all of them - sum, rate, min, max, etc.)
     lazyAggregates: false
 
+    # -- when enabled, the scaler will be returning error to KEDA's `GetMetrics()` call
+    errIfNotFound: false
+
+    # -- default value, that is reported in case of error or if the value is not in the mem store
+    valueIfNotFound: 0.
+
   # -- how often (in milliseconds) should the IsActive method be tried
   isActivePollingIntervalMilliseconds: 500
 

main.go

@@ -48,9 +48,6 @@ func main() {
 	cfg := util.MustParseConfig()
 	otlpReceiverPort := cfg.OTLPReceiverPort
 	restApiPort := cfg.RestApiPort
-	metricStoreRetentionSeconds := cfg.MetricStoreRetentionSeconds
-	lazySeries := cfg.MetricStoreLazySeries
-	lazyAggregates := cfg.MetricStoreLazyAggregates
 
 	lvl := util.SetupLog(cfg.NoColor)
 	isDebug = util.IsDebug(lvl)
@@ -61,7 +58,7 @@ func main() {
 
 	ctx := util.ContextWithLogger(ctrl.SetupSignalHandler(), setupLog)
 	eg, ctx := errgroup.WithContext(ctx)
-	ms := metric.NewMetricStore(metricStoreRetentionSeconds, lazySeries, lazyAggregates)
+	ms := metric.NewMetricStore(cfg)
 	mp := metric.NewParser()
 
 	eg.Go(func() error {

metric/mem_store.go

@@ -21,18 +21,20 @@ type ms struct {
 	metricsExporter    *InternalMetrics
 	lazySeries         bool
 	lazyAggregates     bool
+	valIfNotFound      float64
 	subscriptions      map[t.MetricName][]t.OperationOverTime
 }
 
-func NewMetricStore(stalePeriodSeconds int, lazySeries, lazyAggregates bool) t.MemStore {
+func NewMetricStore(cfg *util.Config) t.MemStore {
 	m := Metrics()
 	m.Init()
 	return ms{
 		store:              &t.Map[string, *t.Map[t.LabelsHash, *t.MetricData]]{},
-		stalePeriodSeconds: stalePeriodSeconds,
+		stalePeriodSeconds: cfg.MetricStoreRetentionSeconds,
 		metricsExporter:    m,
-		lazySeries:         lazySeries,
-		lazyAggregates:     lazyAggregates,
+		lazySeries:         cfg.MetricStoreLazySeries,
+		lazyAggregates:     cfg.MetricStoreLazyAggregates,
+		valIfNotFound:      cfg.MetricStoreValueIfNotFound,
 		subscriptions:      map[t.MetricName][]t.OperationOverTime{},
 	}
 }
@@ -51,20 +53,20 @@ func (m ms) Get(unescapedName t.MetricName, searchLabels t.Labels, timeOp t.Oper
 func (m ms) get(name t.MetricName, searchLabels t.Labels, timeOp t.OperationOverTime, vecOp t.AggregationOverVectors) (float64, t.Found, error) {
 	now := time.Now().Unix()
 	if err := util.CheckTimeOp(timeOp); err != nil {
-		return -1., false, err
+		return m.valIfNotFound, false, err
 	}
 	if err := checkVectorAggregation(vecOp); err != nil {
-		return -1., false, err
+		return m.valIfNotFound, false, err
 	}
 	if m.lazySeries || m.lazyAggregates {
 		if firstTime := subscribe(m.subscriptions, m.lazyAggregates, name, timeOp); firstTime {
-			return -1., false, nil
+			return m.valIfNotFound, false, nil
 		}
 	}
 	storedMetrics, foundMetric := m.store.Load(string(name))
 	if !foundMetric {
 		// not found
-		return -1., false, nil
+		return m.valIfNotFound, false, nil
 	}
 	if md, f := storedMetrics.Load(hashOfMap(searchLabels)); f {
 		// found exact label match
@@ -74,7 +76,7 @@ func (m ms) get(name t.MetricName, searchLabels t.Labels, timeOp t.OperationOver
 		if !m.isStale(md.LastUpdate, now) {
 			ret, f := md.AggregatesOverTime.Load(timeOp)
 			if !f {
-				return -1., false, fmt.Errorf("unknown OperationOverTime: %s", timeOp)
+				return m.valIfNotFound, false, fmt.Errorf("unknown OperationOverTime: %s", timeOp)
 			}
 			return ret, true, nil
 		} else {

metric/mem_store_lazy_test.go

@@ -7,11 +7,15 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 
 	ty "github.com/kedify/otel-add-on/types"
+	"github.com/kedify/otel-add-on/util"
 )
 
 func TestLazyMemStorePutOneAndGetNothing(t *testing.T) {
 	// setup
-	ms := NewMetricStore(5, true, false)
+	ms := NewMetricStore(&util.Config{
+		MetricStoreRetentionSeconds: 5,
+		MetricStoreLazySeries:       true,
+	})
 	ms.Put(ty.NewMetricEntry{
 		Name:             "metric1",
 		MeasurementTime:  pcommon.Timestamp(time.Now().Unix()),
@@ -28,7 +32,10 @@ func TestLazyMemStorePutOneAndGetNothing(t *testing.T) {
 
 func TestLazyMemStoreGetPutAndGet(t *testing.T) {
 	// setup
-	ms := NewMetricStore(70, true, false)
+	ms := NewMetricStore(&util.Config{
+		MetricStoreRetentionSeconds: 70,
+		MetricStoreLazySeries:       true,
+	})
 	labels := map[string]any{
 		"a": "1",
 	}
@@ -48,7 +55,10 @@ func TestLazyMemStoreGetPutAndGet(t *testing.T) {
 
 func TestMemStoreLazyAggregatesFirstCallNotFound(t *testing.T) {
 	// setup
-	ms := NewMetricStore(5, false, true)
+	ms := NewMetricStore(&util.Config{
+		MetricStoreRetentionSeconds: 5,
+		MetricStoreLazyAggregates:   true,
+	})
 	ms.Put(ty.NewMetricEntry{
 		Name:             "metric1",
 		MeasurementTime:  pcommon.Timestamp(time.Now().Unix()),
@@ -65,7 +75,10 @@ func TestMemStoreLazyAggregatesFirstCallNotFound(t *testing.T) {
 
 func TestMemStoreLazyAggregatesOneAgg(t *testing.T) {
 	// setup
-	ms := NewMetricStore(70, false, true)
+	ms := NewMetricStore(&util.Config{
+		MetricStoreRetentionSeconds: 70,
+		MetricStoreLazyAggregates:   true,
+	})
 	labels := map[string]any{
 		"a": "1",
 	}
@@ -82,7 +95,10 @@ func TestMemStoreLazyAggregatesOneAgg(t *testing.T) {
 
 func TestMemStoreLazyAggregatesAddingOnTheFly(t *testing.T) {
 	// setup
-	ms := NewMetricStore(70, false, true)
+	ms := NewMetricStore(&util.Config{
+		MetricStoreRetentionSeconds: 70,
+		MetricStoreLazyAggregates:   true,
+	})
 	labels := map[string]any{
 		"a": "1",
 	}
@@ -103,7 +119,11 @@ func TestMemStoreLazyAggregatesAddingOnTheFly(t *testing.T) {
 
 func TestLazyMemStoreAndLazyAggregatesComplex(t *testing.T) {
 	// setup
-	ms := NewMetricStore(700, true, true)
+	ms := NewMetricStore(&util.Config{
+		MetricStoreRetentionSeconds: 700,
+		MetricStoreLazySeries:       true,
+		MetricStoreLazyAggregates:   true,
+	})
 	labels := map[string]any{
 		"a": "1",
 		"b": "2",