From 3e2ff64ee1f6b52ad664a7cf5be526c3415186c3 Mon Sep 17 00:00:00 2001
From: Trial97 <alexandru.tripon@itsyscom.com>
Date: Tue, 12 Oct 2021 15:19:59 +0300
Subject: [PATCH] Added protection for Redis limit when setting indexes

---
 engine/storage_redis.go                | 15 +++++++-
 general_tests/indexes_redis_it_test.go | 51 ++++++++++++++++++++++++++
 2 files changed, 65 insertions(+), 1 deletion(-)
 create mode 100644 general_tests/indexes_redis_it_test.go

diff --git a/engine/storage_redis.go b/engine/storage_redis.go
index f3ca6e5c0..0b1e7d526 100644
--- a/engine/storage_redis.go
+++ b/engine/storage_redis.go
@@ -68,10 +68,11 @@ const (
 	redis_HMSET    = "HMSET"
 
 	redisLoadError = "Redis is loading the dataset in memory"
+	RedisLimit     = 524287 // https://github.com/StackExchange/StackExchange.Redis/issues/201#issuecomment-98639005
 )
 
 func NewRedisStorage(address string, db int, user, pass, mrshlerStr string,
-	maxConns,attempts int, sentinelName string, isCluster bool, clusterSync,
+	maxConns, attempts int, sentinelName string, isCluster bool, clusterSync,
 	clusterOnDownDelay time.Duration, tlsConn bool,
 	tlsClientCert, tlsClientKey, tlsCACert string) (_ *RedisStorage, err error) {
 	var ms Marshaler
@@ -1203,6 +1204,12 @@ func (rs *RedisStorage) SetIndexesDrv(idxItmType, tntCtx string,
 	for key, strMp := range indexes {
 		if len(strMp) == 0 { // remove with no more elements inside
 			deleteArgs = append(deleteArgs, key)
+			if len(deleteArgs) == RedisLimit+1 { // minus dbkey
+				if err = rs.Cmd(nil, redis_HDEL, deleteArgs...); err != nil {
+					return
+				}
+				deleteArgs = []string{dbKey} // the dbkey is necesary for the HDEL command
+			}
 			continue
 		}
 		var encodedMp []byte
@@ -1210,6 +1217,12 @@ func (rs *RedisStorage) SetIndexesDrv(idxItmType, tntCtx string,
 			return
 		}
 		mp[key] = string(encodedMp)
+		if len(mp) == RedisLimit {
+			if err = rs.FlatCmd(nil, redis_HMSET, dbKey, mp); err != nil {
+				return
+			}
+			mp = make(map[string]string)
+		}
 	}
 	if len(deleteArgs) != 1 {
 		if err = rs.Cmd(nil, redis_HDEL, deleteArgs...); err != nil {
diff --git a/general_tests/indexes_redis_it_test.go b/general_tests/indexes_redis_it_test.go
new file mode 100644
index 000000000..b521b96f5
--- /dev/null
+++ b/general_tests/indexes_redis_it_test.go
@@ -0,0 +1,51 @@
+//go:build integration
+// +build integration
+
+/*
+Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
+Copyright (C) ITsysCOM GmbH
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>
+*/
+
+package general_tests
+
+import (
+	"strconv"
+	"testing"
+
+	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/engine"
+	"github.com/cgrates/cgrates/utils"
+)
+
+func TestIndexesRedis(t *testing.T) {
+	cfg := config.NewDefaultCGRConfig()
+	db, err := engine.NewRedisStorage(cfg.DataDbCfg().Host+":"+cfg.DataDbCfg().Port, 10, cfg.DataDbCfg().User,
+		cfg.DataDbCfg().Password, cfg.GeneralCfg().DBDataEncoding, utils.RedisMaxConns, utils.RedisMaxAttempts,
+		utils.EmptyString, false, 0, 0, false, utils.EmptyString, utils.EmptyString, utils.EmptyString)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	limit := engine.RedisLimit + 1
+	indx := make(map[string]utils.StringSet)
+	for i := 0; i < limit; i++ {
+		indx["*string:*req.Destination:"+strconv.Itoa(i)] = utils.StringSet{"ATTR_New": {}}
+	}
+	if err = db.SetIndexesDrv(utils.CacheAttributeFilterIndexes, "cgrates.org:*any", indx,
+		false, utils.NonTransactional); err != nil {
+		t.Fatal(err)
+	}
+}