Fixed RPC loader so that it uses a singleton (more efficient with connections) and identified a bug where key checkes would always hist the hybrid master

Author: lonelycode

api_definition_manager.go

@@ -522,7 +522,7 @@ func (a *APIDefinitionLoader) LoadDefinitionsFromRPC(orgId string) *[]*APISpec {
 
 	apiCollection := store.GetApiDefinitions(orgId, tags)
 
-	store.Disconnect()
+	//store.Disconnect()
 
 	if RPC_LoadCount > 0 {
 		SaveRPCDefinitionsBackup(apiCollection)

api_loader.go

@@ -586,11 +586,6 @@ func loadApps(APISpecs *[]*APISpec, Muxer *mux.Router) {
 	// Only create this once, add other types here as needed, seems wasteful but we can let the GC handle it
 	redisStore, redisOrgStore, healthStore, rpcAuthStore, rpcOrgStore := prepareStorage()
 
-	if config.SlaveOptions.UseRPC {
-		StartRPCKeepaliveWatcher(rpcAuthStore)
-		StartRPCKeepaliveWatcher(rpcOrgStore)
-	}
-
 	prepareSortOrder(APISpecs)
 
 	chainChannel := make(chan *ChainObject)
@@ -673,4 +668,10 @@ func loadApps(APISpecs *[]*APISpec, Muxer *mux.Router) {
 		"prefix": "main",
 	}).Info("Initialised API Definitions")
 
+	if config.SlaveOptions.UseRPC {
+		//log.Warning("TODO: PUT THE KEEPALIVE WATCHER BACK")
+		StartRPCKeepaliveWatcher(rpcAuthStore)
+		StartRPCKeepaliveWatcher(rpcOrgStore)
+	}
+
 }

auth_manager.go

@@ -3,10 +3,11 @@ package main
 import (
 	"encoding/base64"
 	"encoding/json"
-	"github.com/TykTechnologies/logrus"
-	"github.com/nu7hatch/gouuid"
 	"strings"
 	"time"
+
+	"github.com/TykTechnologies/logrus"
+	"github.com/nu7hatch/gouuid"
 )
 
 // AuthorisationHandler is used to validate a session key,

handler_success.go

@@ -3,14 +3,15 @@ package main
 import (
 	"bytes"
 	b64 "encoding/base64"
-	"github.com/gorilla/context"
-	"github.com/pmylund/go-cache"
 	"io"
 	"net/http"
 	"runtime/pprof"
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/gorilla/context"
+	"github.com/pmylund/go-cache"
 )
 
 // ContextKey is a key type to avoid collisions
@@ -181,11 +182,12 @@ func (t TykMiddleware) CheckSessionAndIdentityForValidKey(key string) (SessionSt
 	var thisSession SessionState
 	var found bool
 
+	log.Debug("Querying local cache")
 	// Check in-memory cache
 	if !config.LocalSessionCache.DisableCacheSessionState {
 		cachedVal, found := SessionCache.Get(key)
 		if found {
-			log.Debug("Key found in local cache")
+			log.Debug("--> Key found in local cache")
 			thisSession = cachedVal.(SessionState)
 			t.ApplyPolicyIfExists(key, &thisSession)
 			return thisSession, true
@@ -202,10 +204,11 @@ func (t TykMiddleware) CheckSessionAndIdentityForValidKey(key string) (SessionSt
 
 		// Check for a policy, if there is a policy, pull it and overwrite the session values
 		t.ApplyPolicyIfExists(key, &thisSession)
-		log.Debug("Got key")
+		log.Debug("--> Got key")
 		return thisSession, true
 	}
 
+	log.Debug("Querying authstore")
 	// 2. If not there, get it from the AuthorizationHandler
 	thisSession, found = t.Spec.AuthManager.IsKeyAuthorised(key)
 	if found {
@@ -217,6 +220,10 @@ func (t TykMiddleware) CheckSessionAndIdentityForValidKey(key string) (SessionSt
 
 		// Check for a policy, if there is a policy, pull it and overwrite the session values
 		t.ApplyPolicyIfExists(key, &thisSession)
+
+		log.Debug("Lifetime is: ", GetLifetime(t.Spec, &thisSession))
+		// Need to set this in order for the write to work!
+		thisSession.LastUpdated = time.Now().String()
 		t.Spec.SessionManager.UpdateSession(key, thisSession, GetLifetime(t.Spec, &thisSession))
 	}
 

main.go

@@ -983,23 +983,32 @@ func getCmdArguments() map[string]interface{} {
 	return arguments
 }
 
+var KeepaliveRunning bool
+
 func StartRPCKeepaliveWatcher(engine *RPCStorageHandler) {
+	if KeepaliveRunning {
+		return
+	}
+
 	go func() {
 		log.WithFields(logrus.Fields{
 			"prefix": "RPC Conn Mgr",
 		}).Info("[RPC Conn Mgr] Starting keepalive watcher...")
 		for {
+			KeepaliveRunning = true
 			RPCKeepAliveCheck(engine)
 			if engine == nil {
 				log.WithFields(logrus.Fields{
 					"prefix": "RPC Conn Mgr",
 				}).Info("No engine, break")
+				KeepaliveRunning = false
 				break
 			}
 			if engine.Killed == true {
 				log.WithFields(logrus.Fields{
 					"prefix": "RPC Conn Mgr",
 				}).Debug("[RPC Conn Mgr] this connection killed")
+				KeepaliveRunning = false
 				break
 			}
 		}

policy.go

@@ -196,7 +196,7 @@ func LoadPoliciesFromRPC(orgId string) map[string]Policy {
 
 	rpcPolicies := store.GetPolicies(orgId)
 
-	store.Disconnect()
+	//store.Disconnect()
 
 	jErr1 := json.Unmarshal([]byte(rpcPolicies), &dbPolicyList)
 

rpc_analytics_purger.go

@@ -2,10 +2,11 @@ package main
 
 import (
 	"encoding/json"
+	"time"
+
 	"github.com/TykTechnologies/logrus"
 	"github.com/lonelycode/gorpc"
 	"gopkg.in/vmihailenco/msgpack.v2"
-	"time"
 )
 
 // Purger is an interface that will define how the in-memory store will be purged
@@ -31,6 +32,17 @@ func (r *RPCPurger) ReConnect() {
 
 // Connect Connects to RPC
 func (r *RPCPurger) Connect() {
+	if RPCClientIsConnected {
+		if RPCCLientSingleton != nil {
+			if RPCFuncClientSingleton != nil {
+				log.Info("RPC Analytics client using singleton")
+				r.RPCClient = RPCCLientSingleton
+				r.Client = RPCFuncClientSingleton
+				return
+			}
+		}
+	}
+
 	log.Info("Connecting to RPC Analytics service")
 	r.RPCClient = gorpc.NewTCPClient(r.Address)
 

rpc_storage_handler.go

@@ -58,6 +58,7 @@ var RPCCLientRWMutex sync.RWMutex = sync.RWMutex{}
 var RPCClients = map[string]chan int{}
 
 func ClearRPCClients() {
+	return
 	log.Info("Found: ", len(RPCClients), " RPC connections, terminating")
 	for _, c := range RPCClients {
 
@@ -74,12 +75,14 @@ func ClearRPCClients() {
 
 func RPCKeepAliveCheck(r *RPCStorageHandler) {
 	// Only run when connected
-	if r.Connected {
+	if RPCClientIsConnected && r.cache != nil {
 		// Make sure the auth back end is still alive
 		c1 := make(chan string, 1)
 
 		go func() {
+			log.Debug("Getting keyspace check test key")
 			r.GetKey("0000")
+			log.Debug("--> done")
 			c1 <- "1"
 			close(c1)
 		}()
@@ -145,31 +148,47 @@ func (r *RPCStorageHandler) checkDisconnect() {
 
 func (r *RPCStorageHandler) ReConnect() {
 	// Should only be used by reload checker
-	r.Disconnect()
+	// r.Disconnect()
 	r.Connect()
 	log.Info("Reconnected.")
 }
 
+var RPCCLientSingleton *gorpc.Client
+var RPCFuncClientSingleton *gorpc.DispatcherClient
+var RPCGlobalCache = cache.New(30*time.Second, 15*time.Second)
+var RPCClientIsConnected bool
+
 // Connect will establish a connection to the DB
 func (r *RPCStorageHandler) Connect() bool {
-	// We don't want to constantly connect
-	if r.Connected {
+
+	if RPCClientIsConnected {
+		log.Debug("Using RPC singleton for connection")
+		r.RPCClient = RPCCLientSingleton
+		r.Client = RPCFuncClientSingleton
+		r.cache = RPCGlobalCache
 		return true
 	}
 
+	// RPC Client is unset
 	// Set up the cache
-	r.cache = cache.New(30*time.Second, 15*time.Second)
-	r.RPCClient = gorpc.NewTCPClient(r.Address)
+	log.Info("Setting new RPC connection!")
+	r.cache = RPCGlobalCache
+	RPCCLientSingleton = gorpc.NewTCPClient(r.Address)
+	r.RPCClient = RPCCLientSingleton
 
 	if log.Level != logrus.DebugLevel {
 		gorpc.SetErrorLogger(gorpc.NilErrorLogger)
 	}
 
 	r.RPCClient.OnConnect = r.OnConnectFunc
-	r.RPCClient.Conns = 10
+	r.RPCClient.Conns = 50
 	r.RPCClient.Start()
 	d := GetDispatcher()
-	r.Client = d.NewFuncClient(r.RPCClient)
+
+	if RPCFuncClientSingleton == nil {
+		RPCFuncClientSingleton = d.NewFuncClient(r.RPCClient)
+	}
+	r.Client = RPCFuncClientSingleton
 	r.Login()
 
 	if !r.SuppressRegister {
@@ -181,14 +200,14 @@ func (r *RPCStorageHandler) Connect() bool {
 }
 
 func (r *RPCStorageHandler) OnConnectFunc(remoteAddr string, rwc io.ReadWriteCloser) (io.ReadWriteCloser, error) {
-	r.Connected = true
+	RPCClientIsConnected = true
 	return rwc, nil
 }
 
 func (r *RPCStorageHandler) Disconnect() bool {
-	if r.Connected {
+	if RPCClientIsConnected {
 		go r.RPCClient.Stop()
-		r.Connected = false
+		RPCClientIsConnected = false
 		RPCCLientRWMutex.Lock()
 		delete(RPCClients, r.ID)
 		RPCCLientRWMutex.Unlock()
@@ -301,7 +320,9 @@ func (r *RPCStorageHandler) GetKey(keyName string) (string, error) {
 
 	// Check the cache first
 	if config.SlaveOptions.EnableRPCCache {
+		log.Debug("Using cache for: ", keyName)
 		cachedVal, found := r.cache.Get(r.fixKey(keyName))
+		log.Debug("--> Found? ", found)
 		if found {
 			elapsed := time.Since(start)
 			log.Debug("GetKey took ", elapsed)
@@ -718,7 +739,7 @@ func (r *RPCStorageHandler) StartRPCLoopCheck(orgId string) {
 		return
 	}
 
-	log.Info("Starting keyspace poller")
+	log.Info("[RPC] Starting keyspace poller")
 
 	for {
 		r.CheckForKeyspaceChanges(orgId)