[TT-10520] Migrating from go-redis to storage library

config.go

@@ -175,7 +175,7 @@ type TykPumpConfiguration struct {
 	// Sets the analytics storage type. Where the pump will be fetching data from. Currently, only
 	// the `redis` option is supported.
 	AnalyticsStorageType string `json:"analytics_storage_type"`
-	// Example Redis storage configuration:
+	// Example Temporal storage configuration:
 	// ```{.json}
 	//   "analytics_storage_config": {
 	//     "type": "redis",
@@ -188,11 +188,11 @@ type TykPumpConfiguration struct {
 	//     "optimisation_max_idle": 100,
 	//     "optimisation_max_active": 0,
 	//     "enable_cluster": false,
-	//     "redis_use_ssl": false,
-	//     "redis_ssl_insecure_skip_verify": false
+	//     "use_ssl": false,
+	//     "ssl_insecure_skip_verify": false
 	//   },
 	// ```
-	AnalyticsStorageConfig storage.RedisStorageConfig `json:"analytics_storage_config"`
+	AnalyticsStorageConfig storage.TemporalStorageConfig `json:"analytics_storage_config"`
 	// Connection string for StatsD monitoring for information please see the
 	// [Instrumentation docs](https://tyk.io/docs/basic-config-and-security/report-monitor-trigger-events/instrumentation/).
 	StatsdConnectionString string `json:"statsd_connection_string"`

go.mod

@@ -6,15 +6,14 @@ require (
 	github.com/DataDog/datadog-go v4.7.0+incompatible
 	github.com/TykTechnologies/gorpc v0.0.0-20210624160652-fe65bda0ccb9
 	github.com/TykTechnologies/murmur3 v0.0.0-20230310161213-aad17efd5632
-	github.com/TykTechnologies/storage v1.0.8
+	github.com/TykTechnologies/storage v1.1.0
 	github.com/aws/aws-sdk-go-v2 v1.22.1
 	github.com/aws/aws-sdk-go-v2/config v1.9.0
 	github.com/aws/aws-sdk-go-v2/credentials v1.5.0
 	github.com/aws/aws-sdk-go-v2/service/sqs v1.26.0
 	github.com/aws/aws-sdk-go-v2/service/timestreamwrite v1.9.0
 	github.com/cenkalti/backoff/v4 v4.0.2
 	github.com/fatih/structs v1.1.0
-	github.com/go-redis/redis/v8 v8.3.1
 	github.com/gocraft/health v0.0.0-20170925182251-8675af27fef0
 	github.com/gofrs/uuid v4.0.0+incompatible
 	github.com/golang/protobuf v1.5.3
@@ -100,13 +99,13 @@ require (
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe // indirect
 	github.com/olivere/elastic v6.2.31+incompatible // indirect
-	github.com/onsi/ginkgo v1.16.4 // indirect
 	github.com/onsi/gomega v1.20.0 // indirect
 	github.com/pierrec/lz4 v2.6.0+incompatible // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.42.0 // indirect
 	github.com/prometheus/procfs v0.10.1 // indirect
+	github.com/redis/go-redis/v9 v9.3.1 // indirect
 	github.com/rogpeppe/go-internal v1.11.0 // indirect
 	github.com/segmentio/backo-go v0.0.0-20160424052352-204274ad699c // indirect
 	github.com/shirou/gopsutil v3.20.11+incompatible // indirect
@@ -120,7 +119,6 @@ require (
 	github.com/xtgo/uuid v0.0.0-20140804021211-a0b114877d4c // indirect
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
 	go.mongodb.org/mongo-driver v1.11.2 // indirect
-	go.opentelemetry.io/otel v0.13.0 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
 	golang.org/x/crypto v0.14.0 // indirect
 	golang.org/x/sync v0.2.0 // indirect

main.go

@@ -23,12 +23,14 @@ import (
 	kingpin "gopkg.in/alecthomas/kingpin.v2"
 )
 
-var SystemConfig TykPumpConfiguration
-var AnalyticsStore storage.AnalyticsStorage
-var UptimeStorage storage.AnalyticsStorage
-var Pumps []pumps.Pump
-var UptimePump pumps.UptimePump
-var AnalyticsSerializers []serializer.AnalyticsSerializer
+var (
+	SystemConfig         TykPumpConfiguration
+	AnalyticsStore       storage.AnalyticsStorage
+	UptimeStorage        storage.AnalyticsStorage
+	Pumps                []pumps.Pump
+	UptimePump           pumps.UptimePump
+	AnalyticsSerializers []serializer.AnalyticsSerializer
+)
 
 var log = logger.GetLogger()
 
@@ -66,7 +68,7 @@ func Init() {
 		demoMode = &envDemo
 	}
 
-	//Serializer init
+	// Serializer init
 	AnalyticsSerializers = []serializer.AnalyticsSerializer{serializer.NewAnalyticsSerializer(serializer.MSGP_SERIALIZER), serializer.NewAnalyticsSerializer(serializer.PROTOBUF_SERIALIZER)}
 
 	log.WithFields(logrus.Fields{
@@ -96,17 +98,32 @@ func Init() {
 	if *debugMode {
 		log.Level = logrus.DebugLevel
 	}
-
 }
 
 func setupAnalyticsStore() {
 	switch SystemConfig.AnalyticsStorageType {
 	case "redis":
-		AnalyticsStore = &storage.RedisClusterStorageManager{}
-		UptimeStorage = &storage.RedisClusterStorageManager{}
+		AnalyticsStore = &storage.TemporalStorageHandler{
+			Config: storage.TemporalStorageConfig{
+				Type: "redis",
+			},
+		}
+		UptimeStorage = &storage.TemporalStorageHandler{
+			Config: storage.TemporalStorageConfig{
+				Type: "redis",
+			},
+		}
 	default:
-		AnalyticsStore = &storage.RedisClusterStorageManager{}
-		UptimeStorage = &storage.RedisClusterStorageManager{}
+		AnalyticsStore = &storage.TemporalStorageHandler{
+			Config: storage.TemporalStorageConfig{
+				Type: "redis",
+			},
+		}
+		UptimeStorage = &storage.TemporalStorageHandler{
+			Config: storage.TemporalStorageConfig{
+				Type: "redis",
+			},
+		}
 	}
 
 	AnalyticsStore.Init(SystemConfig.AnalyticsStorageConfig)
@@ -115,12 +132,12 @@ func setupAnalyticsStore() {
 	uptimeConf := SystemConfig.AnalyticsStorageConfig
 
 	// Swap key prefixes for uptime purger
-	uptimeConf.RedisKeyPrefix = "host-checker:"
+	uptimeConf.KeyPrefix = "host-checker:"
 	UptimeStorage.Init(uptimeConf)
 }
 
 func storeVersion() {
-	var versionStore = &storage.RedisClusterStorageManager{}
+	versionStore := &storage.TemporalStorageHandler{}
 	versionConf := SystemConfig.AnalyticsStorageConfig
 	versionStore.KeyPrefix = "version-check-"
 	versionStore.Config = versionConf
@@ -177,7 +194,6 @@ func initialisePumps() {
 	if !SystemConfig.DontPurgeUptimeData {
 		initialiseUptimePump()
 	}
-
 }
 
 func initialiseUptimePump() {
@@ -210,15 +226,20 @@ func StartPurgeLoop(wg *sync.WaitGroup, ctx context.Context, secInterval int, ch
 		for i := -1; i < 10; i++ {
 			var analyticsKeyName string
 			if i == -1 {
-				//if it's the first iteration, we look for tyk-system-analytics to maintain backwards compatibility or if analytics_config.enable_multiple_analytics_keys is disabled in the gateway
+				// if it's the first iteration, we look for tyk-system-analytics to maintain backwards compatibility or if analytics_config.enable_multiple_analytics_keys is disabled in the gateway
 				analyticsKeyName = storage.ANALYTICS_KEYNAME
 			} else {
 				analyticsKeyName = fmt.Sprintf("%v_%v", storage.ANALYTICS_KEYNAME, i)
 			}
 
 			for _, serializerMethod := range AnalyticsSerializers {
 				analyticsKeyName += serializerMethod.GetSuffix()
-				AnalyticsValues := AnalyticsStore.GetAndDeleteSet(analyticsKeyName, chunkSize, expire)
+				AnalyticsValues, err := AnalyticsStore.GetAndDeleteSet(analyticsKeyName, chunkSize, expire)
+				if err != nil {
+					log.WithFields(logrus.Fields{
+						"prefix": mainPrefix,
+					}).Error("Error on Purge Loop. Is Temporal Storage down?: " + err.Error())
+				}
 				if len(AnalyticsValues) > 0 {
 					PreprocessAnalyticsValues(AnalyticsValues, serializerMethod, analyticsKeyName, omitDetails, job, startTime, secInterval)
 				}
@@ -229,7 +250,12 @@ func StartPurgeLoop(wg *sync.WaitGroup, ctx context.Context, secInterval int, ch
 		job.Timing("purge_time_all", time.Since(startTime).Nanoseconds())
 
 		if !SystemConfig.DontPurgeUptimeData {
-			UptimeValues := UptimeStorage.GetAndDeleteSet(storage.UptimeAnalytics_KEYNAME, chunkSize, expire)
+			UptimeValues, err := UptimeStorage.GetAndDeleteSet(storage.UptimeAnalytics_KEYNAME, chunkSize, expire)
+			if err != nil {
+				log.WithFields(logrus.Fields{
+					"prefix": mainPrefix,
+				}).Error("Error on Purge Loop. Is Temporal Storage down?: " + err.Error())
+			}
 			UptimePump.WriteUptimeData(UptimeValues)
 		}
 
@@ -305,7 +331,6 @@ func writeToPumps(keys []interface{}, job *health.Job, startTime time.Time, purg
 }
 
 func filterData(pump pumps.Pump, keys []interface{}) []interface{} {
-
 	shouldTrim := SystemConfig.MaxRecordSize != 0 || pump.GetMaxRecordSize() != 0
 	filters := pump.GetFilters()
 	ignoreFields := pump.GetIgnoreFields()
@@ -381,11 +406,11 @@ func execPumpWriting(wg *sync.WaitGroup, pmp pumps.Pump, keys *[]interface{}, pu
 	}).Debug("Writing to: ", pmp.GetName())
 
 	ch := make(chan error, 1)
-	//Load pump timeout
+	// Load pump timeout
 	timeout := pmp.GetTimeout()
 	var ctx context.Context
 	var cancel context.CancelFunc
-	//Initialize context depending if the pump has a configured timeout
+	// Initialize context depending if the pump has a configured timeout
 	if timeout > 0 {
 		ctx, cancel = context.WithTimeout(context.Background(), time.Duration(timeout)*time.Second)
 	} else {

storage/init.go

@@ -11,5 +11,7 @@ func init() {
 	AvailableStores = make(map[string]AnalyticsStorage)
 
 	// Register all the storage handlers here
-	AvailableStores["redis"] = &RedisClusterStorageManager{}
+	AvailableStores["redis"] = &TemporalStorageHandler{Config: TemporalStorageConfig{
+		Type: "redis",
+	}}
 }