Implement RCM code

Signed-off-by: Niranjani Vivek <niranjaniv@google.com>

Author: niranjanivivek

translib/db/db.go

@@ -168,6 +168,11 @@ type Options struct {
 	KeySeparator       string //Overriden by the DB config file's separator.
 	IsWriteDisabled    bool   //Is write/set mode disabled ?
 	IsCacheEnabled     bool   //Is cache (Per Connection) allowed?
+	// Redis connection pools do not support transactional Redis operations.
+	// If the DB object will be used to perform transactions, the Transacation
+	// flag must be set to request a unique Redis client. Transactions include
+	// SCAN and MULTI.
+	TransactionsRequired bool
 
 	// OnChange caching for the DBs passed from Translib's Subscribe Infra
 	// to the Apps. SDB is the SubscribeDB() returned handle on which
@@ -402,7 +407,15 @@ func NewDB(opt Options) (*DB, error) {
 	var dur time.Duration
 	now = time.Now()
 
-	d := DB{client: redis.NewClient(adjustRedisOpts(&opt)),
+	var rc *redis.Client
+	if opt.TransactionsRequired {
+		rc = TransactionalRedisClient(opt.DBNo)
+	} else {
+		rc = RedisClient(opt.DBNo)
+	}
+
+	d := DB{
+		client:            rc,
 		Opts:              &opt,
 		txState:           txStateNone,
 		txCmds:            make([]_txCmd, 0, InitialTxPipelineSize),
@@ -455,7 +468,7 @@ func NewDB(opt Options) (*DB, error) {
 	if opt.IsSession && opt.IsOnChangeEnabled {
 		glog.Error("NewDB: Subscription on Config Session not supported : ",
 			d.Name())
-		d.client.Close()
+		CloseRedisClient(d.client)
 		e = tlerr.TranslibDBNotSupported{
 			Description: "Subscription on Config Session not supported"}
 		goto NewDBExit
@@ -464,7 +477,7 @@ func NewDB(opt Options) (*DB, error) {
 	if opt.IsSession && opt.DBNo != ConfigDB {
 		glog.Error("NewDB: Non-Config DB on Config Session not supported : ",
 			d.Name())
-		d.client.Close()
+		CloseRedisClient(d.client)
 		e = tlerr.TranslibDBNotSupported{
 			Description: "Non-Config DB on Config Session not supported"}
 		goto NewDBExit
@@ -498,7 +511,7 @@ func NewDB(opt Options) (*DB, error) {
 				glog.Error("NewDB: Database not inited: ", d.Name(), ": GET ",
 					d.Opts.InitIndicator, " returns err: ", err)
 			}
-			d.client.Close()
+			CloseRedisClient(d.client)
 			e = tlerr.TranslibDBNotInit{}
 			goto NewDBExit
 		}
@@ -512,7 +525,7 @@ func NewDB(opt Options) (*DB, error) {
 
 		if e = ConfigDBTryLock(noSessionToken); e != nil {
 			glog.Errorf("NewDB: ConfigDB possibly locked: %s", e)
-			d.client.Close()
+			CloseRedisClient(d.client)
 			goto NewDBExit
 		}
 		d.configDBLocked = true
@@ -570,7 +583,7 @@ func (d *DB) DeleteDB() error {
 		d.unRegisterSessionDB()
 	}
 
-	err := d.client.Close()
+	err := CloseRedisClient(d.client)
 	d.client = nil
 	return err
 }

translib/db/db_lock.go

@@ -76,7 +76,7 @@ func (lt *LockStruct) tryLock() error {
 	if client, err = getStateDB(); err != nil {
 		return err
 	}
-	defer client.Close()
+	defer CloseRedisClient(client)
 
 	// HSETNX: Set Hash Field if Not Exist
 	args := []interface{}{"HSETNX", lockTableKey, lt.Name,
@@ -116,7 +116,7 @@ func (lt *LockStruct) unlock() error {
 	if client, err = getStateDB(); err != nil {
 		return err
 	}
-	defer client.Close()
+	defer CloseRedisClient(client)
 
 	// Run the LUA Script to HDEL if we set the key
 	if reply, err = luaScriptUnlock.Run(client,
@@ -265,8 +265,7 @@ func dumpStack(begin, end int) {
 func getStateDB() (*redis.Client, error) {
 	var client *redis.Client
 	var err error
-	if client = redis.NewClient(adjustRedisOpts(&Options{
-		DBNo: StateDB})); client == nil {
+	if client = TransactionalRedisClient(StateDB); client == nil {
 
 		glog.Error("getStateDB: Could not create redis client: STATE_DB")
 		err = tlerr.TranslibDBCannotOpen{}

translib/db/db_redis_opts.go

@@ -144,7 +144,7 @@ func adjustRedisOpts(dbOpt *Options) *redis.Options {
 	return &redisOpts
 }
 
-func init() {
+func initializeRedisOpts() {
 	flag.StringVar(&goRedisOpts, "go_redis_opts", "", "Options for go-redis")
 }
 
@@ -197,7 +197,10 @@ func (config *_DBRedisOptsConfig) handleReconfigureSignal() error {
 ////////////////////////////////////////////////////////////////////////////////
 
 func (config *_DBRedisOptsConfig) readFromDB() error {
-	fields, e := readRedis("TRANSLIB_DB|default")
+	rc := TransactionalRedisClient(ConfigDB)
+	defer CloseRedisClient(rc)
+	fields, e := rc.HGetAll("TRANSLIB_DB|default").Result()
+
 	if e == nil {
 		if optsString, ok := fields["go_redis_opts"]; ok {
 			// Parse optsString into config.opts

translib/db/db_stats.go

@@ -23,8 +23,6 @@ import (
 	"reflect"
 	"sync"
 	"time"
-
-	"github.com/go-redis/redis/v7"
 )
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -524,29 +522,9 @@ func (config *DBStatsConfig) readFromDB() error {
 
 func readRedis(key string) (map[string]string, error) {
 
-	ipAddr := DefaultRedisLocalTCPEP
-	dbId := int(ConfigDB)
-	dbPassword := ""
-	if dbInstName := getDBInstName(ConfigDB); dbInstName != "" {
-		if isDbInstPresent(dbInstName) {
-			ipAddr = getDbTcpAddr(dbInstName)
-			dbId = getDbId(dbInstName)
-			dbPassword = getDbPassword(dbInstName)
-		}
-	}
-
-	client := redis.NewClient(&redis.Options{
-		Network:     "tcp",
-		Addr:        ipAddr,
-		Password:    dbPassword,
-		DB:          dbId,
-		DialTimeout: 0,
-		PoolSize:    1,
-	})
-
+	client := RedisClient(ConfigDB)
+	defer CloseRedisClient(client)
 	fields, e := client.HGetAll(key).Result()
 
-	client.Close()
-
 	return fields, e
 }