fix: batch-server redis-counter-deleter fails because of the temporary keys (#2165)

* fix: batch-server redis-counter-deleter fails because of the temporary keys

* refactor: improve temporary key detection in redisCounterDeleter

Author: nnnkkk7

pkg/batch/jobs/rediscounter/redis_counter_deleter.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"regexp"
 	"strconv"
+	"strings"
 	"time"
 
 	"go.uber.org/zap"
@@ -204,12 +205,26 @@ func (r *redisCounterDeleter) scan(environmentId, kind, key string) ([]string, e
 	return keys, nil
 }
 
+func (r *redisCounterDeleter) isTemporaryKey(key string) bool {
+	// Check if the key contains patterns for temporary keys
+	// pfmerge-key: temporary key used for HyperLogLog PFMERGE operations
+	return strings.Contains(key, "pfmerge-key")
+}
+
 func (r *redisCounterDeleter) filterKeysOlderThanThirtyOneDays(
 	environmentId, kind string,
 	keys []string,
 ) ([]string, error) {
 	filteredKeys := make([]string, 0, len(keys))
+	skippedKeys := make([]string, 0)
 	for _, key := range keys {
+		// Skip temporary keys like pfmerge-key
+		// These keys have TTL (10 minutes) and will be automatically deleted
+		if r.isTemporaryKey(key) {
+			skippedKeys = append(skippedKeys, key)
+			continue
+		}
+
 		// E.g. environment_id:uc:1689835532:feature_id:variation_id
 		var regex string
 		if environmentId == "" {
@@ -243,6 +258,21 @@ func (r *redisCounterDeleter) filterKeysOlderThanThirtyOneDays(
 		}
 		filteredKeys = append(filteredKeys, key)
 	}
+
+	if len(skippedKeys) > 0 {
+		// Log only a sample of skipped keys to avoid log flooding
+		sampleSize := len(skippedKeys)
+		if sampleSize > 5 {
+			sampleSize = 5
+		}
+		r.logger.Warn("Skipped temporary keys",
+			zap.String("environmentId", environmentId),
+			zap.String("kind", kind),
+			zap.Int("skippedCount", len(skippedKeys)),
+			zap.Strings("examples", skippedKeys[:sampleSize]),
+		)
+	}
+
 	return filteredKeys, nil
 }
 

pkg/batch/jobs/rediscounter/redis_counter_deleter_test.go

@@ -307,6 +307,34 @@ func TestFilterKeysOlderThanThirtyOneDays(t *testing.T) {
 		expected          []string
 		expectedError     error
 	}{
+		{
+			desc:              "success: skip pfmerge keys",
+			inputEnvNamespace: "",
+			inputKind:         "uc",
+			inputKeys: []string{
+				"uc:pfmerge-key:ios-premium-tab-enabled",
+				"uc:pfmerge-key:android-feature",
+				fmt.Sprintf("uc:%d:feature_id:variation_id", now.Unix()-31*day),
+			},
+			expected: []string{
+				fmt.Sprintf("uc:%d:feature_id:variation_id", now.Unix()-31*day),
+			},
+			expectedError: nil,
+		},
+		{
+			desc:              "success: skip pfmerge keys with environment id",
+			inputEnvNamespace: "dev",
+			inputKind:         "uc",
+			inputKeys: []string{
+				"dev:pfmerge:uc:pfmerge-key:feature_id:variation_id",
+				fmt.Sprintf("dev:uc:%d:feature_id_1:variation_id", now.Unix()-3*day),
+				fmt.Sprintf("dev:uc:%d:feature_id_2:variation_id", now.Unix()-31*day),
+			},
+			expected: []string{
+				fmt.Sprintf("dev:uc:%d:feature_id_2:variation_id", now.Unix()-31*day),
+			},
+			expectedError: nil,
+		},
 		{
 			desc:              "errSubmatchStringNotFound",
 			inputEnvNamespace: "dev",