Fix redis connections and remote pidfilelocation usage internal

TykTechnologies · Jan 8, 2025 · a7ca7df · a7ca7df
1 parent f4e4130
commit a7ca7df
Show file tree

Hide file tree

Showing 13 changed files with 101 additions and 27 deletions.
diff --git a/gateway/api.go b/gateway/api.go
@@ -2236,13 +2236,16 @@ func (gw *Gateway) createOauthClient(w http.ResponseWriter, r *http.Request) {
 					storageManager := gw.getGlobalMDCBStorageHandler(prefix, false)
 					storageManager.Connect()
 
+					storageDriver := &storage.RedisCluster{KeyPrefix: prefix, HashKeys: false, ConnectionHandler: gw.StorageConnectionHandler}
+					storageDriver.Connect()
+
 					apiSpec.OAuthManager = &OAuthManager{
 						OsinServer: gw.TykOsinNewServer(
 							&osin.ServerConfig{},
 							&RedisOsinStorageInterface{
 								storageManager,
 								gw.GlobalSessionManager,
-								&storage.RedisCluster{KeyPrefix: prefix, HashKeys: false, ConnectionHandler: gw.StorageConnectionHandler},
+								storageDriver,
 								apiSpec.OrgID,
 								gw,
 							}),
@@ -2623,12 +2626,16 @@ func (gw *Gateway) getOauthClientDetails(keyName, apiID string) (interface{}, in
 		prefix := generateOAuthPrefix(apiSpec.APIID)
 		storageManager := gw.getGlobalMDCBStorageHandler(prefix, false)
 		storageManager.Connect()
+
+		storageDriver := &storage.RedisCluster{KeyPrefix: prefix, HashKeys: false, ConnectionHandler: gw.StorageConnectionHandler}
+		storageDriver.Connect()
+
 		apiSpec.OAuthManager = &OAuthManager{
 			OsinServer: gw.TykOsinNewServer(&osin.ServerConfig{},
 				&RedisOsinStorageInterface{
 					storageManager,
 					gw.GlobalSessionManager,
-					&storage.RedisCluster{KeyPrefix: prefix, HashKeys: false, ConnectionHandler: gw.StorageConnectionHandler},
+					storageDriver,
 					apiSpec.OrgID,
 					gw,
 				}),

diff --git a/gateway/delete_api_cache.go b/gateway/delete_api_cache.go
@@ -8,5 +8,7 @@ import (
 
 func (gw *Gateway) invalidateAPICache(apiID string) bool {
 	store := storage.RedisCluster{IsCache: true, ConnectionHandler: gw.StorageConnectionHandler}
+	store.Connect()
+
 	return store.DeleteScanMatch(fmt.Sprintf("cache-%s*", apiID))
 }
diff --git a/gateway/gateway_test.go b/gateway/gateway_test.go
@@ -945,7 +945,10 @@ func TestCacheAllSafeRequests(t *testing.T) {
 	t.Skip() // DeleteScanMatch interferes with other tests.
 
 	ts := StartTest(nil)
+
 	cache := storage.RedisCluster{KeyPrefix: "cache-", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	cache.Connect()
+
 	t.Cleanup(func() {
 		ts.Close()
 		cache.DeleteScanMatch("*")
@@ -975,7 +978,10 @@ func TestCacheAllSafeRequestsWithCachedHeaders(t *testing.T) {
 	t.Skip() // DeleteScanMatch interferes with other tests.
 
 	ts := StartTest(nil)
+
 	cache := storage.RedisCluster{KeyPrefix: "cache-", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	cache.Connect()
+
 	t.Cleanup(func() {
 		ts.Close()
 		cache.DeleteScanMatch("*")
@@ -1021,7 +1027,10 @@ func TestCacheWithAdvanceUrlRewrite(t *testing.T) {
 	t.Skip() // DeleteScanMatch interferes with other tests.
 
 	ts := StartTest(nil)
+
 	cache := storage.RedisCluster{KeyPrefix: "cache-", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	cache.Connect()
+
 	t.Cleanup(func() {
 		ts.Close()
 		cache.DeleteScanMatch("*")
@@ -1080,7 +1089,10 @@ func TestCachePostRequest(t *testing.T) {
 	t.Skip() // DeleteScanMatch interferes with other tests.
 
 	ts := StartTest(nil)
+
 	cache := storage.RedisCluster{KeyPrefix: "cache-", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	cache.Connect()
+
 	t.Cleanup(func() {
 		ts.Close()
 		cache.DeleteScanMatch("*")
@@ -1127,7 +1139,10 @@ func TestAdvanceCachePutRequest(t *testing.T) {
 	t.Skip() // DeleteScanMatch interferes with other tests.
 
 	ts := StartTest(nil)
+
 	cache := storage.RedisCluster{KeyPrefix: "cache-", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	cache.Connect()
+
 	t.Cleanup(func() {
 		ts.Close()
 		cache.DeleteScanMatch("*")
@@ -1220,7 +1235,10 @@ func TestCacheAllSafeRequestsWithAdvancedCacheEndpoint(t *testing.T) {
 	t.Skip() // DeleteScanMatch interferes with other tests.
 
 	ts := StartTest(nil)
+
 	cache := storage.RedisCluster{KeyPrefix: "cache-", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	cache.Connect()
+
 	t.Cleanup(func() {
 		ts.Close()
 		cache.DeleteScanMatch("*")
@@ -1259,7 +1277,9 @@ func TestCacheEtag(t *testing.T) {
 	t.Skip() // DeleteScanMatch interferes with other tests.
 
 	ts := StartTest(nil)
+
 	cache := storage.RedisCluster{KeyPrefix: "cache-", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	cache.Connect()
 
 	t.Cleanup(func() {
 		ts.Close()
@@ -1314,7 +1334,10 @@ func TestOldCachePlugin(t *testing.T) {
 		t.Helper()
 
 		ts := StartTest(nil)
+
 		cache := storage.RedisCluster{KeyPrefix: "cache-", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+		cache.Connect()
+
 		t.Cleanup(func() {
 			ts.Close()
 			cache.DeleteScanMatch("*")
@@ -1343,7 +1366,10 @@ func TestAdvanceCacheTimeoutPerEndpoint(t *testing.T) {
 	t.Skip() // DeleteScanMatch interferes with other tests.
 
 	ts := StartTest(nil)
+
 	cache := storage.RedisCluster{KeyPrefix: "cache-", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	cache.Connect()
+
 	t.Cleanup(func() {
 		ts.Close()
 		cache.DeleteScanMatch("*")

diff --git a/gateway/health_check.go b/gateway/health_check.go
@@ -79,6 +79,7 @@ func (gw *Gateway) gatherHealthChecks() {
 	allInfos := SafeHealthCheck{info: make(map[string]HealthCheckItem, 3)}
 
 	redisStore := storage.RedisCluster{KeyPrefix: "livenesscheck-", ConnectionHandler: gw.StorageConnectionHandler}
+	redisStore.Connect()
 
 	key := "tyk-liveness-probe"
 

diff --git a/gateway/host_checker_manager_test.go b/gateway/host_checker_manager_test.go
@@ -14,7 +14,10 @@ func TestHostCheckerManagerInit(t *testing.T) {
 	defer ts.Close()
 
 	hc := HostCheckerManager{Gw: ts.Gw}
+
 	redisStorage := &storage.RedisCluster{KeyPrefix: "host-checker-test:", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	redisStorage.Connect()
+
 	hc.Init(redisStorage)
 
 	if hc.Id == "" {
@@ -46,6 +49,8 @@ func TestAmIPolling(t *testing.T) {
 	ts.Gw.SetConfig(globalConf)
 
 	redisStorage := &storage.RedisCluster{KeyPrefix: "host-checker-test:", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	redisStorage.Connect()
+
 	hc.Init(redisStorage)
 	hc2 := HostCheckerManager{Gw: ts.Gw}
 	hc2.Init(redisStorage)
@@ -74,7 +79,10 @@ func TestAmIPolling(t *testing.T) {
 
 	//Testing if the PollerCacheKey doesn't contains the poller_group by default
 	hc = HostCheckerManager{Gw: ts.Gw}
+
 	redisStorage = &storage.RedisCluster{KeyPrefix: "host-checker-test:", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	redisStorage.Connect()
+
 	hc.Init(redisStorage)
 	hc.AmIPolling()
 
@@ -106,7 +114,10 @@ func TestCheckActivePollerLoop(t *testing.T) {
 	defer ts.Close()
 
 	hc := &HostCheckerManager{Gw: ts.Gw}
+
 	redisStorage := &storage.RedisCluster{KeyPrefix: "host-checker-test-1:", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	redisStorage.Connect()
+
 	hc.Init(redisStorage)
 
 	go hc.CheckActivePollerLoop(ts.Gw.ctx)
@@ -122,7 +133,10 @@ func TestStartPoller(t *testing.T) {
 	defer ts.Close()
 
 	hc := HostCheckerManager{Gw: ts.Gw}
+
 	redisStorage := &storage.RedisCluster{KeyPrefix: "host-checker-TestStartPoller:", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	redisStorage.Connect()
+
 	hc.Init(redisStorage)
 
 	hc.StartPoller(ts.Gw.ctx)
@@ -135,9 +149,12 @@ func TestStartPoller(t *testing.T) {
 func TestRecordUptimeAnalytics(t *testing.T) {
 	ts := StartTest(nil)
 	defer ts.Close()
+
 	hc := &HostCheckerManager{Gw: ts.Gw}
 
 	redisStorage := &storage.RedisCluster{KeyPrefix: "host-checker-test-analytics:", ConnectionHandler: ts.Gw.StorageConnectionHandler}
+	redisStorage.Connect()
+
 	hc.Init(redisStorage)
 
 	spec := &APISpec{}

diff --git a/gateway/mw_external_oauth.go b/gateway/mw_external_oauth.go
@@ -296,7 +296,10 @@ func isExpired(claims jwt.MapClaims) bool {
 }
 
 func newIntrospectionCache(gw *Gateway) *introspectionCache {
-	return &introspectionCache{RedisCluster: storage.RedisCluster{KeyPrefix: "introspection-", ConnectionHandler: gw.StorageConnectionHandler}}
+	conn := storage.RedisCluster{KeyPrefix: "introspection-", ConnectionHandler: gw.StorageConnectionHandler}
+	conn.Connect()
+
+	return &introspectionCache{RedisCluster: conn}
 }
 
 type introspectionCache struct {

diff --git a/gateway/mw_jwt.go b/gateway/mw_jwt.go
@@ -579,12 +579,16 @@ func (k *JWTMiddleware) processCentralisedJWT(r *http.Request, token *jwt.Token)
 			prefix := generateOAuthPrefix(k.Spec.APIID)
 			storageManager := k.Gw.getGlobalMDCBStorageHandler(prefix, false)
 			storageManager.Connect()
+
+			storageDriver := &storage.RedisCluster{KeyPrefix: prefix, HashKeys: false, ConnectionHandler: k.Gw.StorageConnectionHandler}
+			storageDriver.Connect()
+
 			k.Spec.OAuthManager = &OAuthManager{
 				OsinServer: k.Gw.TykOsinNewServer(&osin.ServerConfig{},
 					&RedisOsinStorageInterface{
 						storageManager,
 						k.Gw.GlobalSessionManager,
-						&storage.RedisCluster{KeyPrefix: prefix, HashKeys: false, ConnectionHandler: k.Gw.StorageConnectionHandler},
+						storageDriver,
 						k.Spec.OrgID,
 						k.Gw,
 					}),

diff --git a/gateway/mw_oauth2_auth_ee.go b/gateway/mw_oauth2_auth_ee.go
@@ -22,9 +22,13 @@ func getUpstreamOAuthMw(base *BaseMiddleware) TykMiddleware {
 }
 
 func getClientCredentialsStorageHandler(base *BaseMiddleware) *storage.RedisCluster {
-	return &storage.RedisCluster{KeyPrefix: "upstreamOAuthCC-", ConnectionHandler: base.Gw.StorageConnectionHandler}
+	handler := &storage.RedisCluster{KeyPrefix: "upstreamOAuthCC-", ConnectionHandler: base.Gw.StorageConnectionHandler}
+	handler.Connect()
+	return handler
 }
 
 func getPasswordStorageHandler(base *BaseMiddleware) *storage.RedisCluster {
-	return &storage.RedisCluster{KeyPrefix: "upstreamOAuthPW-", ConnectionHandler: base.Gw.StorageConnectionHandler}
+	handler := &storage.RedisCluster{KeyPrefix: "upstreamOAuthPW-", ConnectionHandler: base.Gw.StorageConnectionHandler}
+	handler.Connect()
+	return handler
 }
diff --git a/gateway/oauth_manager.go b/gateway/oauth_manager.go
@@ -1195,6 +1195,7 @@ func (gw *Gateway) purgeLapsedOAuthTokens() error {
 	}
 
 	redisCluster := &storage.RedisCluster{KeyPrefix: "", HashKeys: false, ConnectionHandler: gw.StorageConnectionHandler}
+	redisCluster.Connect()
 
 	ok, err := redisCluster.Lock("oauth-purge-lock", time.Minute)
 	if err != nil {

diff --git a/gateway/redis_logrus_hook.go b/gateway/redis_logrus_hook.go
@@ -16,13 +16,13 @@ type redisChannelHook struct {
 func (gw *Gateway) newRedisHook() *redisChannelHook {
 	hook := &redisChannelHook{}
 	hook.formatter = new(logrus.JSONFormatter)
-	hook.notifier.store = &storage.RedisCluster{KeyPrefix: "gateway-notifications:", ConnectionHandler: gw.StorageConnectionHandler}
 	hook.notifier.channel = "dashboard.ui.messages"
+	hook.notifier.store = &storage.RedisCluster{KeyPrefix: "gateway-notifications:", ConnectionHandler: gw.StorageConnectionHandler}
+	hook.notifier.store.Connect()
 	return hook
 }
 
 func (hook *redisChannelHook) Fire(entry *logrus.Entry) error {
-
 	orgId, found := entry.Data["org_id"]
 	if !found {
 		return nil
@@ -44,6 +44,7 @@ func (hook *redisChannelHook) Fire(entry *logrus.Entry) error {
 	}
 
 	go hook.notifier.Notify(n)
+
 	return nil
 }
 

diff --git a/gateway/rpc_backup_handlers.go b/gateway/rpc_backup_handlers.go
@@ -30,8 +30,9 @@ func (gw *Gateway) LoadDefinitionsFromRPCBackup() ([]*APISpec, error) {
 	tagList := getTagListAsString(gw.GetConfig().DBAppConfOptions.Tags)
 	checkKey := BackupApiKeyBase + tagList
 
-	store := storage.RedisCluster{KeyPrefix: RPCKeyPrefix, ConnectionHandler: gw.StorageConnectionHandler}
+	store := &storage.RedisCluster{KeyPrefix: RPCKeyPrefix, ConnectionHandler: gw.StorageConnectionHandler}
 	connected := store.Connect()
+
 	log.Info("[RPC] --> Loading API definitions from backup")
 
 	if !connected {
@@ -60,7 +61,7 @@ func (gw *Gateway) saveRPCDefinitionsBackup(list string) error {
 
 	log.Info("--> Connecting to DB")
 
-	store := storage.RedisCluster{KeyPrefix: RPCKeyPrefix, ConnectionHandler: gw.StorageConnectionHandler}
+	store := &storage.RedisCluster{KeyPrefix: RPCKeyPrefix, ConnectionHandler: gw.StorageConnectionHandler}
 	connected := store.Connect()
 
 	log.Info("--> Connected to DB")
@@ -83,9 +84,9 @@ func (gw *Gateway) LoadPoliciesFromRPCBackup() (map[string]user.Policy, error) {
 	tagList := getTagListAsString(gw.GetConfig().DBAppConfOptions.Tags)
 	checkKey := BackupPolicyKeyBase + tagList
 
-	store := storage.RedisCluster{KeyPrefix: RPCKeyPrefix, ConnectionHandler: gw.StorageConnectionHandler}
-
+	store := &storage.RedisCluster{KeyPrefix: RPCKeyPrefix, ConnectionHandler: gw.StorageConnectionHandler}
 	connected := store.Connect()
+
 	log.Info("[RPC] Loading Policies from backup")
 
 	if !connected {

diff --git a/gateway/rpc_storage_handler.go b/gateway/rpc_storage_handler.go
@@ -162,7 +162,7 @@ func (r *RPCStorageHandler) buildNodeInfo() []byte {
 		intCheckDuration = int64(checkDuration / time.Second)
 	}
 
-	r.Gw.getHostDetails(r.Gw.GetConfig().PIDFileLocation)
+	r.Gw.getHostDetails()
 	node := model.NodeData{
 		NodeID:          r.Gw.GetNodeID(),
 		GroupID:         config.SlaveOptions.GroupID,