fix: metric tasks_queue_compaction_tasks_by_hook

Author: timmilesdw

pkg/metrics/metrics.go

@@ -43,9 +43,9 @@ var (
 	// TasksQueueLength shows the current length of the task queue
 	TasksQueueLength = "{PREFIX}tasks_queue_length"
 	// TasksQueueCompactionOperationsTotal counts compaction operations per hook
-	TasksQueueCompactionOperationsTotal = "{PREFIX}tasks_queue_compaction_operations_total"
-	// TasksQueueCompactionTasksByHook shows count of tasks per hook in queue (only when count > 20)
-	TasksQueueCompactionTasksByHook = "{PREFIX}tasks_queue_compaction_tasks_by_hook"
+	TasksQueueCompactionOperationsTotal = "d8_telemetry_{PREFIX}tasks_queue_compaction_operations_total"
+	// TasksQueueCompactionTasksByHook shows the number of tasks in queue for each hook when count exceeds 20
+	TasksQueueCompactionTasksByHook = "d8_telemetry_{PREFIX}tasks_queue_compaction_tasks_by_hook"
 
 	// ============================================================================
 	// Hook Execution Metrics
@@ -103,8 +103,8 @@ func InitMetrics(prefix string) {
 	// ============================================================================
 	TasksQueueActionDurationSeconds = ReplacePrefix(TasksQueueActionDurationSeconds, prefix)
 	TasksQueueLength = ReplacePrefix(TasksQueueLength, prefix)
-	TasksQueueCompactionInQueueTasks = ReplacePrefix(TasksQueueCompactionInQueueTasks, prefix)
-	TasksQueueCompactionReached = ReplacePrefix(TasksQueueCompactionReached, prefix)
+	TasksQueueCompactionOperationsTotal = ReplacePrefix(TasksQueueCompactionOperationsTotal, prefix)
+	TasksQueueCompactionTasksByHook = ReplacePrefix(TasksQueueCompactionTasksByHook, prefix)
 
 	// ============================================================================
 	// Hook Execution Metrics
@@ -351,10 +351,10 @@ func RegisterTaskQueueMetrics(metricStorage metricsstorage.Storage) error {
 		return fmt.Errorf("failed to register %s: %w", TasksQueueCompactionOperationsTotal, err)
 	}
 
-	// Register tasks by hook gauge (only when count > 20)
+	// Register compaction tasks by hook gauge
 	_, err = metricStorage.RegisterGauge(
 		TasksQueueCompactionTasksByHook, compactionLabels,
-		options.WithHelp("Gauge showing count of tasks per hook in queue (only when count > 20)"),
+		options.WithHelp("Gauge showing the number of tasks in queue for each hook when the count exceeds 20. Updated in real-time as tasks are added/removed and during compaction operations."),
 	)
 	if err != nil {
 		return fmt.Errorf("failed to register %s: %w", TasksQueueCompactionTasksByHook, err)

pkg/task/queue/task_counter.go

@@ -5,17 +5,22 @@ import (
 
 	metricsstorage "github.com/deckhouse/deckhouse/pkg/metrics-storage"
 
+	"github.com/flant/shell-operator/pkg/metrics"
 	"github.com/flant/shell-operator/pkg/task"
 )
 
-const taskCap = 100
+const (
+	taskCap                    = 100
+	compactionMetricsThreshold = 20
+)
 
 type TaskCounter struct {
 	mu sync.RWMutex
 
 	queueName      string
 	counter        map[string]uint
 	reachedCap     map[string]struct{}
+	hookCounter    map[string]uint // tracks tasks by hook name
 	metricStorage  metricsstorage.Storage
 	countableTypes map[task.TaskType]struct{}
 }
@@ -29,6 +34,7 @@ func NewTaskCounter(name string, countableTypes map[task.TaskType]struct{}, metr
 		queueName:      name,
 		counter:        make(map[string]uint, 32),
 		reachedCap:     make(map[string]struct{}, 32),
+		hookCounter:    make(map[string]uint, 32),
 		metricStorage:  metricStorage,
 		countableTypes: countableTypes,
 	}
@@ -52,7 +58,6 @@ func (tc *TaskCounter) Add(task task.Task) {
 	}
 
 	counter++
-
 	tc.counter[id] = counter
 
 	if counter == taskCap {
@@ -123,3 +128,36 @@ func (tc *TaskCounter) ResetReachedCap() {
 
 	tc.reachedCap = make(map[string]struct{}, 32)
 }
+
+// UpdateHookMetricsFromSnapshot updates metrics for all hooks based on a snapshot of hook counts.
+// Only hooks with task count above the threshold are published to avoid metric cardinality explosion.
+func (tc *TaskCounter) UpdateHookMetricsFromSnapshot(hookCounts map[string]uint) {
+	if tc.metricStorage == nil {
+		return
+	}
+
+	tc.mu.Lock()
+	defer tc.mu.Unlock()
+
+	// Clear tracking for hooks that are no longer above threshold
+	for hookName := range tc.hookCounter {
+		if count, exists := hookCounts[hookName]; !exists || count <= compactionMetricsThreshold {
+			// Hook dropped below threshold or disappeared - stop tracking it
+			delete(tc.hookCounter, hookName)
+		}
+	}
+
+	// Update metrics only for hooks above threshold
+	for hookName, count := range hookCounts {
+		if count > compactionMetricsThreshold {
+			// Track and publish metric
+			tc.hookCounter[hookName] = count
+
+			labels := map[string]string{
+				"queue_name": tc.queueName,
+				"hook":       hookName,
+			}
+			tc.metricStorage.GaugeSet(metrics.TasksQueueCompactionTasksByHook, float64(count), labels)
+		}
+	}
+}

pkg/task/queue/task_counter_test.go

@@ -1,98 +1,99 @@
 package queue
 
-// import (
-// 	"sync"
-// 	"testing"
-
-// 	"github.com/stretchr/testify/require"
-
-// 	"github.com/flant/shell-operator/pkg/hook/task_metadata"
-// 	"github.com/flant/shell-operator/pkg/metric"
-// 	"github.com/flant/shell-operator/pkg/metrics"
-// 	"github.com/flant/shell-operator/pkg/task"
-// )
-
-// func TestTaskCounterRemoveUsesQueueName(t *testing.T) {
-// 	metricStorage := metric.NewStorageMock(t)
-
-// 	type gaugeCall struct {
-// 		metric string
-// 		value  float64
-// 		labels map[string]string
-// 	}
-
-// 	var (
-// 		mu    sync.Mutex
-// 		calls []gaugeCall
-// 	)
-
-// 	metricStorage.GaugeSetMock.Set(func(metric string, value float64, labels map[string]string) {
-// 		cloned := make(map[string]string, len(labels))
-// 		for k, v := range labels {
-// 			cloned[k] = v
-// 		}
-
-// 		mu.Lock()
-// 		calls = append(calls, gaugeCall{
-// 			metric: metric,
-// 			value:  value,
-// 			labels: cloned,
-// 		})
-// 		mu.Unlock()
-// 	})
-
-// 	tc := NewTaskCounter("main", nil, metricStorage)
-
-// 	testTask := task.NewTask(task_metadata.HookRun).
-// 		WithCompactionID("test-hook")
-
-// 	tc.Add(testTask)
-// 	tc.Remove(testTask)
-
-// 	mu.Lock()
-// 	require.NotEmpty(t, calls)
-// 	lastCall := calls[len(calls)-1]
-// 	mu.Unlock()
-
-// 	require.Equal(t, metrics.TasksQueueCompactionInQueueTasks, lastCall.metric)
-// 	require.Equal(t, float64(0), lastCall.value)
-// 	require.Equal(t, "main", lastCall.labels["queue_name"])
-// 	require.Equal(t, "test-hook", lastCall.labels["task_id"])
-// }
-
-// func TestTaskCounterRemoveClearsReachedCap(t *testing.T) {
-// 	metricStorage := metric.NewStorageMock(t)
-
-// 	var (
-// 		mu            sync.Mutex
-// 		reachedValues []float64
-// 	)
-
-// 	metricStorage.GaugeSetMock.Set(func(metric string, value float64, labels map[string]string) {
-// 		if metric == metrics.TasksQueueCompactionReached {
-// 			mu.Lock()
-// 			reachedValues = append(reachedValues, value)
-// 			mu.Unlock()
-// 		}
-// 	})
-
-// 	tc := NewTaskCounter("main", nil, metricStorage)
-
-// 	testTask := task.NewTask(task_metadata.HookRun).
-// 		WithCompactionID("test-hook")
-
-// 	for i := 0; i < taskCap; i++ {
-// 		tc.Add(testTask)
-// 	}
-
-// 	require.True(t, tc.IsAnyCapReached())
-
-// 	tc.Remove(testTask)
-
-// 	require.False(t, tc.IsAnyCapReached())
-
-// 	mu.Lock()
-// 	require.NotEmpty(t, reachedValues)
-// 	require.Equal(t, float64(0), reachedValues[len(reachedValues)-1])
-// 	mu.Unlock()
-// }
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/flant/shell-operator/pkg/metric"
+	"github.com/flant/shell-operator/pkg/metrics"
+)
+
+func TestTaskCounterUpdateHookMetricsFromSnapshot(t *testing.T) {
+	metricStorage := metric.NewStorageMock(t)
+
+	type gaugeCall struct {
+		metric string
+		value  float64
+		labels map[string]string
+	}
+
+	var (
+		mu    sync.Mutex
+		calls []gaugeCall
+	)
+
+	metricStorage.GaugeSetMock.Set(func(metric string, value float64, labels map[string]string) {
+		cloned := make(map[string]string, len(labels))
+		for k, v := range labels {
+			cloned[k] = v
+		}
+
+		mu.Lock()
+		calls = append(calls, gaugeCall{
+			metric: metric,
+			value:  value,
+			labels: cloned,
+		})
+		mu.Unlock()
+	})
+
+	tc := NewTaskCounter("main", nil, metricStorage)
+
+	// Simulate initial state with hooks above threshold by setting up a snapshot
+	initialSnapshot := map[string]uint{
+		"hook1": 26,
+		"hook2": 31,
+		"hook3": 51,
+	}
+	tc.UpdateHookMetricsFromSnapshot(initialSnapshot)
+
+	mu.Lock()
+	calls = nil // Clear previous calls
+	mu.Unlock()
+
+	// Update with new snapshot where:
+	// - hook1 still has high count (25 tasks)
+	// - hook2 dropped below threshold (15 tasks) - should not be published
+	// - hook3 is completely gone (0 tasks in new snapshot) - should not be published
+	// - hook4 is new (30 tasks)
+	newSnapshot := map[string]uint{
+		"hook1": 25,
+		"hook2": 15,
+		"hook4": 30,
+	}
+
+	tc.UpdateHookMetricsFromSnapshot(newSnapshot)
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	// Verify that metrics were set correctly
+	require.NotEmpty(t, calls)
+
+	// Build a map of last call for each hook
+	lastCallByHook := make(map[string]gaugeCall)
+	for _, call := range calls {
+		if call.metric == metrics.TasksQueueCompactionTasksByHook {
+			hook := call.labels["hook"]
+			lastCallByHook[hook] = call
+		}
+	}
+
+	// hook1: should have value 25
+	require.Contains(t, lastCallByHook, "hook1")
+	require.Equal(t, float64(25), lastCallByHook["hook1"].value)
+	require.Equal(t, "main", lastCallByHook["hook1"].labels["queue_name"])
+
+	// hook2: should NOT be published (below threshold)
+	require.NotContains(t, lastCallByHook, "hook2")
+
+	// hook3: should NOT be published (removed from snapshot and was above threshold before)
+	require.NotContains(t, lastCallByHook, "hook3")
+
+	// hook4: should have value 30
+	require.Contains(t, lastCallByHook, "hook4")
+	require.Equal(t, float64(30), lastCallByHook["hook4"].value)
+	require.Equal(t, "main", lastCallByHook["hook4"].labels["queue_name"])
+}