From d2ba7afaaf7fec1046f122fd7e468df45d3cbc11 Mon Sep 17 00:00:00 2001 From: Radu Ioan Fericean Date: Fri, 6 Nov 2015 20:19:49 +0200 Subject: [PATCH 1/9] building version --- devel.yaml | 2 +- engine/calldesc.go | 22 ++-- engine/loader_local_test.go | 8 +- engine/storage_redis.go | 193 +++++++++++++++++++----------------- glide.yaml | 71 ++++++------- 5 files changed, 156 insertions(+), 140 deletions(-) diff --git a/devel.yaml b/devel.yaml index cd66f283d..2dc82d038 100644 --- a/devel.yaml +++ b/devel.yaml @@ -4,7 +4,6 @@ import: - package: github.com/jinzhu/gorm - package: golang.org/x/net - package: github.com/DisposaBoy/JsonConfigReader - - package: github.com/hoisie/redis - package: github.com/go-sql-driver/mysql - package: github.com/gorhill/cronexpr - package: github.com/lib/pq @@ -17,3 +16,4 @@ import: - package: github.com/cgrates/osipsdagram - package: github.com/cgrates/kamevapi - package: github.com/cgrates/fsock + - package: github.com/mediocregopher/radix.v2/... diff --git a/engine/calldesc.go b/engine/calldesc.go index fe9db8c9f..bdfd10f11 100644 --- a/engine/calldesc.go +++ b/engine/calldesc.go @@ -37,7 +37,7 @@ const ( RECURSION_MAX_DEPTH = 3 MIN_PREFIX_MATCH = 1 FALLBACK_SUBJECT = utils.ANY - DEBUG = true + DEBUG = false ) func init() { @@ -46,23 +46,23 @@ func init() { accountingStorage, _ = NewMapStorage() } else { var err error - ratingStorage, err = NewMongoStorage("127.0.0.1", "27017", "cgrates_rating_test", "", "") - if err != nil { - log.Fatal(err) - } - accountingStorage, err = NewMongoStorage("127.0.0.1", "27017", "cgrates_accounting_test", "", "") - if err != nil { - log.Fatal(err) - } + /* ratingStorage, err = NewMongoStorage("127.0.0.1", "27017", "cgrates_rating_test", "", "") + if err != nil { + log.Fatal(err) + } + accountingStorage, err = NewMongoStorage("127.0.0.1", "27017", "cgrates_accounting_test", "", "") + if err != nil { + log.Fatal(err) + }*/ - /*ratingStorage, _ = NewRedisStorage("127.0.0.1:6379", 12, "", utils.MSGPACK) + ratingStorage, _ = NewRedisStorage("127.0.0.1:6379", 12, "", utils.MSGPACK) if err != nil { log.Fatal(err) } accountingStorage, _ = NewRedisStorage("127.0.0.1:6379", 13, "", utils.MSGPACK) if err != nil { log.Fatal(err) - }*/ + } } storageLogger = ratingStorage.(LogStorage) diff --git a/engine/loader_local_test.go b/engine/loader_local_test.go index f502f1740..4c338d503 100644 --- a/engine/loader_local_test.go +++ b/engine/loader_local_test.go @@ -388,14 +388,14 @@ func TestMatchLoadCsvWithStorRating(t *testing.T) { } rsStor := ratingDbStor.(*RedisStorage) rsApier := ratingDbApier.(*RedisStorage) - keysCsv, err := rsCsv.db.Keys("*") + keysCsv, err := rsCsv.db.Cmd("KEYS", "*").List() if err != nil { t.Fatal("Failed querying redis keys for csv data") } for _, key := range keysCsv { var refVal []byte for idx, rs := range []*RedisStorage{rsCsv, rsStor, rsApier} { - qVal, err := rs.db.Get(key) + qVal, err := rs.db.Cmd("GET", key).Bytes() if err != nil { t.Fatalf("Run: %d, could not retrieve key %s, error: %s", idx, key, err.Error()) } @@ -420,7 +420,7 @@ func TestMatchLoadCsvWithStorAccounting(t *testing.T) { } rsStor := accountDbStor.(*RedisStorage) rsApier := accountDbApier.(*RedisStorage) - keysCsv, err := rsCsv.db.Keys("*") + keysCsv, err := rsCsv.db.Cmd("KEYS", "*").List() if err != nil { t.Fatal("Failed querying redis keys for csv data") } @@ -430,7 +430,7 @@ func TestMatchLoadCsvWithStorAccounting(t *testing.T) { continue } for idx, rs := range []*RedisStorage{rsCsv, rsStor, rsApier} { - qVal, err := rs.db.Get(key) + qVal, err := rs.db.Cmd("GET", key).Bytes() if err != nil { t.Fatalf("Run: %d, could not retrieve key %s, error: %s", idx, key, err.Error()) } diff --git a/engine/storage_redis.go b/engine/storage_redis.go index 72ccbb664..db4463908 100644 --- a/engine/storage_redis.go +++ b/engine/storage_redis.go @@ -27,21 +27,30 @@ import ( "github.com/cgrates/cgrates/cache2go" "github.com/cgrates/cgrates/utils" - "github.com/hoisie/redis" + "github.com/mediocregopher/radix.v2/pool" "io/ioutil" "time" ) type RedisStorage struct { - dbNb int - db *redis.Client - ms Marshaler + db *pool.Pool + ms Marshaler } func NewRedisStorage(address string, db int, pass, mrshlerStr string) (*RedisStorage, error) { - ndb := &redis.Client{Addr: address, Db: db, Password: pass} - + p, err := pool.New("tcp", address, 10) + if err != nil { + return nil, err + } + if err := p.Cmd("SELECT", db).Err; err != nil { + return nil, err + } + if pass != "" { + if err := p.Cmd("AUTH", pass).Err; err != nil { + return nil, err + } + } var mrshler Marshaler if mrshlerStr == utils.MSGPACK { mrshler = NewCodecMsgpackMarshaler() @@ -50,7 +59,7 @@ func NewRedisStorage(address string, db int, pass, mrshlerStr string) (*RedisSto } else { return nil, fmt.Errorf("Unsupported marshaler: %v", mrshlerStr) } - return &RedisStorage{db: ndb, dbNb: db, ms: mrshler}, nil + return &RedisStorage{db: p, ms: mrshler}, nil } func (rs *RedisStorage) Close() { @@ -58,13 +67,16 @@ func (rs *RedisStorage) Close() { //rs.db.Quit() } -func (rs *RedisStorage) Flush(ignore string) (err error) { - err = rs.db.Flush(false) - return +func (rs *RedisStorage) Flush(ignore string) error { + return rs.db.Cmd("FLUSH").Err } func (rs *RedisStorage) GetKeysForPrefix(prefix string) ([]string, error) { - return rs.db.Keys(prefix + "*") + r := rs.db.Cmd("KEYS", "*") + if r.Err != nil { + return nil, r.Err + } + return r.List() } func (rs *RedisStorage) CacheRatingAll() error { @@ -114,7 +126,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac if dKeys == nil || (float64(cache2go.CountEntries(utils.DESTINATION_PREFIX))*utils.DESTINATIONS_LOAD_THRESHOLD < float64(len(dKeys))) { // if need to load more than a half of exiting keys load them all utils.Logger.Info("Caching all destinations") - if dKeys, err = rs.db.Keys(utils.DESTINATION_PREFIX + "*"); err != nil { + if dKeys, err = rs.db.Cmd("KEYS", utils.DESTINATION_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -138,7 +150,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if rpKeys == nil { utils.Logger.Info("Caching all rating plans") - if rpKeys, err = rs.db.Keys(utils.RATING_PLAN_PREFIX + "*"); err != nil { + if rpKeys, err = rs.db.Cmd("KEYS", utils.RATING_PLAN_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -158,7 +170,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if rpfKeys == nil { utils.Logger.Info("Caching all rating profiles") - if rpfKeys, err = rs.db.Keys(utils.RATING_PROFILE_PREFIX + "*"); err != nil { + if rpfKeys, err = rs.db.Cmd("KEYS", utils.RATING_PROFILE_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -178,7 +190,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if lcrKeys == nil { utils.Logger.Info("Caching LCR rules.") - if lcrKeys, err = rs.db.Keys(utils.LCR_PREFIX + "*"); err != nil { + if lcrKeys, err = rs.db.Cmd("KEYS", utils.LCR_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -199,7 +211,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac // DerivedChargers caching if dcsKeys == nil { utils.Logger.Info("Caching all derived chargers") - if dcsKeys, err = rs.db.Keys(utils.DERIVEDCHARGERS_PREFIX + "*"); err != nil { + if dcsKeys, err = rs.db.Cmd("KEYS", utils.DERIVEDCHARGERS_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -222,7 +234,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if actKeys == nil { utils.Logger.Info("Caching all actions") - if actKeys, err = rs.db.Keys(utils.ACTION_PREFIX + "*"); err != nil { + if actKeys, err = rs.db.Cmd("KEYS", utils.ACTION_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -246,7 +258,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if shgKeys == nil { utils.Logger.Info("Caching all shared groups") - if shgKeys, err = rs.db.Keys(utils.SHARED_GROUP_PREFIX + "*"); err != nil { + if shgKeys, err = rs.db.Cmd("KEYS", utils.SHARED_GROUP_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -305,7 +317,7 @@ func (rs *RedisStorage) cacheAccounting(alsKeys []string) (err error) { } if alsKeys == nil { utils.Logger.Info("Caching all aliases") - if alsKeys, err = rs.db.Keys(utils.ALIASES_PREFIX + "*"); err != nil { + if alsKeys, err = rs.db.Cmd("KEYS", utils.ALIASES_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -336,7 +348,8 @@ func (rs *RedisStorage) cacheAccounting(alsKeys []string) (err error) { func (rs *RedisStorage) HasData(category, subject string) (bool, error) { switch category { case utils.DESTINATION_PREFIX, utils.RATING_PLAN_PREFIX, utils.RATING_PROFILE_PREFIX, utils.ACTION_PREFIX, utils.ACTION_PLAN_PREFIX, utils.ACCOUNT_PREFIX: - return rs.db.Exists(category + subject) + i, err := rs.db.Cmd("EXISTS", category+subject).Int() + return i == 1, err } return false, errors.New("Unsupported category in HasData") } @@ -351,7 +364,7 @@ func (rs *RedisStorage) GetRatingPlan(key string, skipCache bool) (rp *RatingPla } } var values []byte - if values, err = rs.db.Get(key); err == nil { + if values, err = rs.db.Cmd("GET", key).Bytes(); err == nil { b := bytes.NewBuffer(values) r, err := zlib.NewReader(b) if err != nil { @@ -375,7 +388,7 @@ func (rs *RedisStorage) SetRatingPlan(rp *RatingPlan) (err error) { w := zlib.NewWriter(&b) w.Write(result) w.Close() - err = rs.db.Set(utils.RATING_PLAN_PREFIX+rp.Id, b.Bytes()) + err = rs.db.Cmd("SET", utils.RATING_PLAN_PREFIX+rp.Id, b.Bytes()).Err if err == nil && historyScribe != nil { response := 0 go historyScribe.Record(rp.GetHistoryRecord(), &response) @@ -394,7 +407,7 @@ func (rs *RedisStorage) GetRatingProfile(key string, skipCache bool) (rpf *Ratin } } var values []byte - if values, err = rs.db.Get(key); err == nil { + if values, err = rs.db.Cmd("GET", key).Bytes(); err == nil { rpf = new(RatingProfile) err = rs.ms.Unmarshal(values, rpf) cache2go.Cache(key, rpf) @@ -404,7 +417,7 @@ func (rs *RedisStorage) GetRatingProfile(key string, skipCache bool) (rpf *Ratin func (rs *RedisStorage) SetRatingProfile(rpf *RatingProfile) (err error) { result, err := rs.ms.Marshal(rpf) - err = rs.db.Set(utils.RATING_PROFILE_PREFIX+rpf.Id, result) + err = rs.db.Cmd("SET", utils.RATING_PROFILE_PREFIX+rpf.Id, result).Err if err == nil && historyScribe != nil { response := 0 go historyScribe.Record(rpf.GetHistoryRecord(false), &response) @@ -413,12 +426,12 @@ func (rs *RedisStorage) SetRatingProfile(rpf *RatingProfile) (err error) { } func (rs *RedisStorage) RemoveRatingProfile(key string) error { - keys, err := rs.db.Keys(utils.RATING_PROFILE_PREFIX + key + "*") + keys, err := rs.db.Cmd("KEYS", utils.RATING_PROFILE_PREFIX+key+"*").List() if err != nil { return err } for _, key := range keys { - if _, err = rs.db.Del(key); err != nil { + if err = rs.db.Cmd("DEL", key).Err; err != nil { return err } cache2go.RemKey(key) @@ -441,7 +454,7 @@ func (rs *RedisStorage) GetLCR(key string, skipCache bool) (lcr *LCR, err error) } } var values []byte - if values, err = rs.db.Get(key); err == nil { + if values, err = rs.db.Cmd("GET", key).Bytes(); err == nil { err = rs.ms.Unmarshal(values, &lcr) cache2go.Cache(key, lcr) } @@ -450,7 +463,7 @@ func (rs *RedisStorage) GetLCR(key string, skipCache bool) (lcr *LCR, err error) func (rs *RedisStorage) SetLCR(lcr *LCR) (err error) { result, err := rs.ms.Marshal(lcr) - err = rs.db.Set(utils.LCR_PREFIX+lcr.GetId(), result) + err = rs.db.Cmd("SET", utils.LCR_PREFIX+lcr.GetId(), result).Err cache2go.Cache(utils.LCR_PREFIX+lcr.GetId(), lcr) return } @@ -458,7 +471,7 @@ func (rs *RedisStorage) SetLCR(lcr *LCR) (err error) { func (rs *RedisStorage) GetDestination(key string) (dest *Destination, err error) { key = utils.DESTINATION_PREFIX + key var values []byte - if values, err = rs.db.Get(key); len(values) > 0 && err == nil { + if values, err = rs.db.Cmd("GET", key).Bytes(); len(values) > 0 && err == nil { b := bytes.NewBuffer(values) r, err := zlib.NewReader(b) if err != nil { @@ -490,7 +503,7 @@ func (rs *RedisStorage) SetDestination(dest *Destination) (err error) { w := zlib.NewWriter(&b) w.Write(result) w.Close() - err = rs.db.Set(utils.DESTINATION_PREFIX+dest.Id, b.Bytes()) + err = rs.db.Cmd("SET", utils.DESTINATION_PREFIX+dest.Id, b.Bytes()).Err if err == nil && historyScribe != nil { response := 0 go historyScribe.Record(dest.GetHistoryRecord(), &response) @@ -508,7 +521,7 @@ func (rs *RedisStorage) GetActions(key string, skipCache bool) (as Actions, err } } var values []byte - if values, err = rs.db.Get(key); err == nil { + if values, err = rs.db.Cmd("GET", key).Bytes(); err == nil { err = rs.ms.Unmarshal(values, &as) cache2go.Cache(key, as) } @@ -517,7 +530,7 @@ func (rs *RedisStorage) GetActions(key string, skipCache bool) (as Actions, err func (rs *RedisStorage) SetActions(key string, as Actions) (err error) { result, err := rs.ms.Marshal(&as) - err = rs.db.Set(utils.ACTION_PREFIX+key, result) + err = rs.db.Cmd("SET", utils.ACTION_PREFIX+key, result).Err return } @@ -531,7 +544,7 @@ func (rs *RedisStorage) GetSharedGroup(key string, skipCache bool) (sg *SharedGr } } var values []byte - if values, err = rs.db.Get(key); err == nil { + if values, err = rs.db.Cmd("GET", key).Bytes(); err == nil { err = rs.ms.Unmarshal(values, &sg) cache2go.Cache(key, sg) } @@ -540,13 +553,13 @@ func (rs *RedisStorage) GetSharedGroup(key string, skipCache bool) (sg *SharedGr func (rs *RedisStorage) SetSharedGroup(sg *SharedGroup) (err error) { result, err := rs.ms.Marshal(sg) - err = rs.db.Set(utils.SHARED_GROUP_PREFIX+sg.Id, result) + err = rs.db.Cmd("SET", utils.SHARED_GROUP_PREFIX+sg.Id, result).Err return } func (rs *RedisStorage) GetAccount(key string) (ub *Account, err error) { var values []byte - if values, err = rs.db.Get(utils.ACCOUNT_PREFIX + key); err == nil { + if values, err = rs.db.Cmd("GET", utils.ACCOUNT_PREFIX+key).Bytes(); err == nil { ub = &Account{Id: key} err = rs.ms.Unmarshal(values, ub) } @@ -568,19 +581,18 @@ func (rs *RedisStorage) SetAccount(ub *Account) (err error) { } } result, err := rs.ms.Marshal(ub) - err = rs.db.Set(utils.ACCOUNT_PREFIX+ub.Id, result) + err = rs.db.Cmd("SET", utils.ACCOUNT_PREFIX+ub.Id, result).Err return } func (rs *RedisStorage) RemoveAccount(key string) (err error) { - _, err = rs.db.Del(utils.ACCOUNT_PREFIX + key) - return + return rs.db.Cmd("DEL", utils.ACCOUNT_PREFIX+key).Err } func (rs *RedisStorage) GetCdrStatsQueue(key string) (sq *StatsQueue, err error) { var values []byte - if values, err = rs.db.Get(utils.CDR_STATS_QUEUE_PREFIX + key); err == nil { + if values, err = rs.db.Cmd("GET", utils.CDR_STATS_QUEUE_PREFIX+key).Bytes(); err == nil { sq = &StatsQueue{} err = rs.ms.Unmarshal(values, &sq) } @@ -589,18 +601,18 @@ func (rs *RedisStorage) GetCdrStatsQueue(key string) (sq *StatsQueue, err error) func (rs *RedisStorage) SetCdrStatsQueue(sq *StatsQueue) (err error) { result, err := rs.ms.Marshal(sq) - err = rs.db.Set(utils.CDR_STATS_QUEUE_PREFIX+sq.GetId(), result) + err = rs.db.Cmd("SET", utils.CDR_STATS_QUEUE_PREFIX+sq.GetId(), result).Err return } func (rs *RedisStorage) GetSubscribers() (result map[string]*SubscriberData, err error) { - keys, err := rs.db.Keys(utils.PUBSUB_SUBSCRIBERS_PREFIX + "*") + keys, err := rs.db.Cmd("KEYS", utils.PUBSUB_SUBSCRIBERS_PREFIX+"*").List() if err != nil { return nil, err } result = make(map[string]*SubscriberData) for _, key := range keys { - if values, err := rs.db.Get(key); err == nil { + if values, err := rs.db.Cmd("GET", key).Bytes(); err == nil { sub := &SubscriberData{} err = rs.ms.Unmarshal(values, sub) result[key[len(utils.PUBSUB_SUBSCRIBERS_PREFIX):]] = sub @@ -613,24 +625,28 @@ func (rs *RedisStorage) GetSubscribers() (result map[string]*SubscriberData, err func (rs *RedisStorage) SetSubscriber(key string, sub *SubscriberData) (err error) { result, err := rs.ms.Marshal(sub) - rs.db.Set(utils.PUBSUB_SUBSCRIBERS_PREFIX+key, result) - return + if err != nil { + return err + } + return rs.db.Cmd("SET", utils.PUBSUB_SUBSCRIBERS_PREFIX+key, result).Err } func (rs *RedisStorage) RemoveSubscriber(key string) (err error) { - _, err = rs.db.Del(utils.PUBSUB_SUBSCRIBERS_PREFIX + key) + err = rs.db.Cmd("DEL", utils.PUBSUB_SUBSCRIBERS_PREFIX+key).Err return } func (rs *RedisStorage) SetUser(up *UserProfile) (err error) { result, err := rs.ms.Marshal(up) - rs.db.Set(utils.USERS_PREFIX+up.GetId(), result) - return + if err != nil { + return err + } + return rs.db.Cmd("SET", utils.USERS_PREFIX+up.GetId(), result).Err } func (rs *RedisStorage) GetUser(key string) (up *UserProfile, err error) { var values []byte - if values, err = rs.db.Get(utils.USERS_PREFIX + key); err == nil { + if values, err = rs.db.Cmd("GET", utils.USERS_PREFIX+key).Bytes(); err == nil { up = &UserProfile{} err = rs.ms.Unmarshal(values, &up) } @@ -638,12 +654,12 @@ func (rs *RedisStorage) GetUser(key string) (up *UserProfile, err error) { } func (rs *RedisStorage) GetUsers() (result []*UserProfile, err error) { - keys, err := rs.db.Keys(utils.USERS_PREFIX + "*") + keys, err := rs.db.Cmd("KEYS", utils.USERS_PREFIX+"*").List() if err != nil { return nil, err } for _, key := range keys { - if values, err := rs.db.Get(key); err == nil { + if values, err := rs.db.Cmd("GET", key).Bytes(); err == nil { up := &UserProfile{} err = rs.ms.Unmarshal(values, up) result = append(result, up) @@ -655,14 +671,15 @@ func (rs *RedisStorage) GetUsers() (result []*UserProfile, err error) { } func (rs *RedisStorage) RemoveUser(key string) (err error) { - _, err = rs.db.Del(utils.USERS_PREFIX + key) - return + return rs.db.Cmd("DEL", utils.USERS_PREFIX+key).Err } func (rs *RedisStorage) SetAlias(al *Alias) (err error) { result, err := rs.ms.Marshal(al.Values) - rs.db.Set(utils.ALIASES_PREFIX+al.GetId(), result) - return + if err != nil { + return err + } + return rs.db.Cmd("SET", utils.ALIASES_PREFIX+al.GetId(), result).Err } func (rs *RedisStorage) GetAlias(key string, skipCache bool) (al *Alias, err error) { @@ -678,7 +695,7 @@ func (rs *RedisStorage) GetAlias(key string, skipCache bool) (al *Alias, err err } } var values []byte - if values, err = rs.db.Get(key); err == nil { + if values, err = rs.db.Cmd("GET", key).Bytes(); err == nil { al = &Alias{Values: make(AliasValues, 0)} al.SetId(origKey) err = rs.ms.Unmarshal(values, &al.Values) @@ -711,10 +728,10 @@ func (rs *RedisStorage) RemoveAlias(key string) (err error) { origKey := key key = utils.ALIASES_PREFIX + key aliasValues := make(AliasValues, 0) - if values, err := rs.db.Get(key); err == nil { + if values, err := rs.db.Cmd("GET", key).Bytes(); err == nil { rs.ms.Unmarshal(values, &aliasValues) } - _, err = rs.db.Del(key) + err = rs.db.Cmd("DEL", key).Err if err == nil { for _, value := range aliasValues { for target, pairs := range value.Pairs { @@ -761,7 +778,7 @@ func (rs *RedisStorage) GetLoadHistory(limit int, skipCache bool) ([]*LoadInstan if limit != -1 { limit -= -1 // Decrease limit to match redis approach on lrange } - marshaleds, err := rs.db.Lrange(utils.LOADINST_KEY, 0, limit) + marshaleds, err := rs.db.Cmd("LRANGE", utils.LOADINST_KEY, 0, limit).ListBytes() if err != nil { return nil, err } @@ -789,16 +806,16 @@ func (rs *RedisStorage) AddLoadHistory(ldInst *LoadInstance, loadHistSize int) e return err } _, err = Guardian.Guard(func() (interface{}, error) { // Make sure we do it locked since other instance can modify history while we read it - histLen, err := rs.db.Llen(utils.LOADINST_KEY) + histLen, err := rs.db.Cmd("LLEN", utils.LOADINST_KEY).Int() if err != nil { return nil, err } if histLen >= loadHistSize { // Have hit maximum history allowed, remove oldest element in order to add new one - if _, err := rs.db.Rpop(utils.LOADINST_KEY); err != nil { + if err := rs.db.Cmd("RPOP", utils.LOADINST_KEY).Err; err != nil { return nil, err } } - err = rs.db.Lpush(utils.LOADINST_KEY, marshaled) + err = rs.db.Cmd("LPUSH", utils.LOADINST_KEY, marshaled).Err return nil, err }, 0, utils.LOADINST_KEY) return err @@ -806,7 +823,7 @@ func (rs *RedisStorage) AddLoadHistory(ldInst *LoadInstance, loadHistSize int) e func (rs *RedisStorage) GetActionTriggers(key string) (atrs ActionTriggers, err error) { var values []byte - if values, err = rs.db.Get(utils.ACTION_TRIGGER_PREFIX + key); err == nil { + if values, err = rs.db.Cmd("GET", utils.ACTION_TRIGGER_PREFIX+key).Bytes(); err == nil { err = rs.ms.Unmarshal(values, &atrs) } return @@ -815,20 +832,18 @@ func (rs *RedisStorage) GetActionTriggers(key string) (atrs ActionTriggers, err func (rs *RedisStorage) SetActionTriggers(key string, atrs ActionTriggers) (err error) { if len(atrs) == 0 { // delete the key - _, err = rs.db.Del(utils.ACTION_TRIGGER_PREFIX + key) - return err + return rs.db.Cmd("DEL", utils.ACTION_TRIGGER_PREFIX+key).Err } result, err := rs.ms.Marshal(&atrs) if err != nil { return err } - err = rs.db.Set(utils.ACTION_TRIGGER_PREFIX+key, result) - return + return rs.db.Cmd("SET", utils.ACTION_TRIGGER_PREFIX+key, result).Err } func (rs *RedisStorage) GetActionPlans(key string) (ats ActionPlans, err error) { var values []byte - if values, err = rs.db.Get(utils.ACTION_PLAN_PREFIX + key); err == nil { + if values, err = rs.db.Cmd("GET", utils.ACTION_PLAN_PREFIX+key).Bytes(); err == nil { err = rs.ms.Unmarshal(values, &ats) } return @@ -837,25 +852,23 @@ func (rs *RedisStorage) GetActionPlans(key string) (ats ActionPlans, err error) func (rs *RedisStorage) SetActionPlans(key string, ats ActionPlans) (err error) { if len(ats) == 0 { // delete the key - _, err = rs.db.Del(utils.ACTION_PLAN_PREFIX + key) - return err + return rs.db.Cmd("DEL", utils.ACTION_PLAN_PREFIX+key).Err } result, err := rs.ms.Marshal(&ats) if err != nil { return err } - err = rs.db.Set(utils.ACTION_PLAN_PREFIX+key, result) - return + return rs.db.Cmd("SET", utils.ACTION_PLAN_PREFIX+key, result).Err } func (rs *RedisStorage) GetAllActionPlans() (ats map[string]ActionPlans, err error) { - keys, err := rs.db.Keys(utils.ACTION_PLAN_PREFIX + "*") + keys, err := rs.db.Cmd("KEYS", utils.ACTION_PLAN_PREFIX+"*").List() if err != nil { return nil, err } ats = make(map[string]ActionPlans, len(keys)) for _, key := range keys { - values, err := rs.db.Get(key) + values, err := rs.db.Cmd("GET", key).Bytes() if err != nil { continue } @@ -877,7 +890,7 @@ func (rs *RedisStorage) GetDerivedChargers(key string, skipCache bool) (dcs util } } var values []byte - if values, err = rs.db.Get(key); err == nil { + if values, err = rs.db.Cmd("GET", key).Bytes(); err == nil { err = rs.ms.Unmarshal(values, &dcs) cache2go.Cache(key, dcs) } @@ -886,36 +899,39 @@ func (rs *RedisStorage) GetDerivedChargers(key string, skipCache bool) (dcs util func (rs *RedisStorage) SetDerivedChargers(key string, dcs utils.DerivedChargers) (err error) { if len(dcs) == 0 { - _, err = rs.db.Del(utils.DERIVEDCHARGERS_PREFIX + key) // FIXME: Does cache need cleanup too? - return err + return rs.db.Cmd("DEL", utils.DERIVEDCHARGERS_PREFIX+key).Err } marshaled, err := rs.ms.Marshal(dcs) - err = rs.db.Set(utils.DERIVEDCHARGERS_PREFIX+key, marshaled) - return err + if err != nil { + return err + } + return rs.db.Cmd("SET", utils.DERIVEDCHARGERS_PREFIX+key, marshaled).Err } func (rs *RedisStorage) SetCdrStats(cs *CdrStats) error { marshaled, err := rs.ms.Marshal(cs) - err = rs.db.Set(utils.CDR_STATS_PREFIX+cs.Id, marshaled) - return err + if err != nil { + return err + } + return rs.db.Cmd("SET", utils.CDR_STATS_PREFIX+cs.Id, marshaled).Err } func (rs *RedisStorage) GetCdrStats(key string) (cs *CdrStats, err error) { var values []byte - if values, err = rs.db.Get(utils.CDR_STATS_PREFIX + key); err == nil { + if values, err = rs.db.Cmd("GET", utils.CDR_STATS_PREFIX+key).Bytes(); err == nil { err = rs.ms.Unmarshal(values, &cs) } return } func (rs *RedisStorage) GetAllCdrStats() (css []*CdrStats, err error) { - keys, err := rs.db.Keys(utils.CDR_STATS_PREFIX + "*") + keys, err := rs.db.Cmd("KEYS", utils.CDR_STATS_PREFIX+"*").List() if err != nil { return nil, err } for _, key := range keys { - value, err := rs.db.Get(key) + value, err := rs.db.Cmd("GET", key).Bytes() if err != nil { continue } @@ -932,13 +948,12 @@ func (rs *RedisStorage) LogCallCost(cgrid, source, runid string, cc *CallCost) ( if err != nil { return } - err = rs.db.Set(utils.LOG_CALL_COST_PREFIX+source+runid+"_"+cgrid, result) - return + return rs.db.Cmd("SET", utils.LOG_CALL_COST_PREFIX+source+runid+"_"+cgrid, result).Err } func (rs *RedisStorage) GetCallCostLog(cgrid, source, runid string) (cc *CallCost, err error) { var values []byte - if values, err = rs.db.Get(utils.LOG_CALL_COST_PREFIX + source + runid + "_" + cgrid); err == nil { + if values, err = rs.db.Cmd("GET", utils.LOG_CALL_COST_PREFIX+source+runid+"_"+cgrid).Bytes(); err == nil { err = rs.ms.Unmarshal(values, cc) } return @@ -953,8 +968,7 @@ func (rs *RedisStorage) LogActionTrigger(ubId, source string, at *ActionTrigger, if err != nil { return } - rs.db.Set(utils.LOG_ACTION_TRIGGER_PREFIX+source+"_"+time.Now().Format(time.RFC3339Nano), []byte(fmt.Sprintf("%v*%v*%v", ubId, string(mat), string(mas)))) - return + return rs.db.Cmd("SET", utils.LOG_ACTION_TRIGGER_PREFIX+source+"_"+time.Now().Format(time.RFC3339Nano), []byte(fmt.Sprintf("%v*%v*%v", ubId, string(mat), string(mas)))).Err } func (rs *RedisStorage) LogActionPlan(source string, at *ActionPlan, as Actions) (err error) { @@ -966,6 +980,5 @@ func (rs *RedisStorage) LogActionPlan(source string, at *ActionPlan, as Actions) if err != nil { return } - err = rs.db.Set(utils.LOG_ACTION_TIMMING_PREFIX+source+"_"+time.Now().Format(time.RFC3339Nano), []byte(fmt.Sprintf("%v*%v", string(mat), string(mas)))) - return + return rs.db.Cmd("SET", utils.LOG_ACTION_TIMMING_PREFIX+source+"_"+time.Now().Format(time.RFC3339Nano), []byte(fmt.Sprintf("%v*%v", string(mat), string(mas)))).Err } diff --git a/glide.yaml b/glide.yaml index e2f5b6ed4..4e5cd0e36 100644 --- a/glide.yaml +++ b/glide.yaml @@ -1,36 +1,39 @@ +parent: null package: github.com/cgrates/cgrates import: - - package: github.com/jinzhu/gorm - ref: 20e37a05337026c18c4f74285ae1f47323d0e9eb - - package: github.com/qor/inflection - ref: 3272df6c21d04180007eb3349844c89a3856bc25 - - package: gopkg.in/fsnotify.v1 - ref: 96c060f6a6b7e0d6f75fddd10efeaca3e5d1bcb0 - - package: github.com/cgrates/rpcclient - ref: f0f378951e30943373953b44b910cc65ba832da3 - - package: github.com/cgrates/fsock - ref: 7ec136e2798fbfe44f713f5ee5828ce643dec4c2 - - package: github.com/go-sql-driver/mysql - ref: 527bcd55aab2e53314f1a150922560174b493034 - - package: github.com/lib/pq - ref: ffe986aba3e6cfcded8b06615965941408891ef0 - - package: github.com/peterh/liner - ref: b850cf8c6d0ee52309aad09ac610508c6c75e819 - - package: golang.org/x/net - ref: 6d10a0c3eafbed530eb495a38c41ccdf8ebdcb4e - - package: github.com/cgrates/kamevapi - ref: a376b1f937ba959857929fa3e111c0f3243278c0 - - package: github.com/gorhill/cronexpr - ref: a557574d6c024ed6e36acc8b610f5f211c91568a - - package: gopkg.in/mgo.v2 - ref: f402e3a216db333ae6b3ba68b9152a34a0bc6984 - - package: github.com/kr/pty - ref: f7ee69f31298ecbe5d2b349c711e2547a617d398 - - package: github.com/cgrates/osipsdagram - ref: 3d6beed663452471dec3ca194137a30d379d9e8f - - package: github.com/ugorji/go - ref: 8a2a3a8c488c3ebd98f422a965260278267a0551 - - package: github.com/DisposaBoy/JsonConfigReader - ref: 33a99fdf1d5ee1f79b5077e9c06f955ad356d5f4 - - package: github.com/hoisie/redis - ref: 788f01e396a99c96c8f56338383926f16841ebae +- package: github.com/ugorji/go + version: f1f1a805ed361a0e078bb537e4ea78cd37dcf065 +- package: github.com/jinzhu/gorm + version: 67f88d977b01a8c87e0e942f791ce2cc21503223 +- package: golang.org/x/net + version: b4e17d61b15679caf2335da776c614169a1b4643 +- package: github.com/DisposaBoy/JsonConfigReader + version: 33a99fdf1d5ee1f79b5077e9c06f955ad356d5f4 +- package: github.com/go-sql-driver/mysql + version: 69e3ed7607d7c139386480824801584c947c67cf +- package: github.com/gorhill/cronexpr + version: a557574d6c024ed6e36acc8b610f5f211c91568a +- package: github.com/lib/pq + version: 83c4f410d0aed80a0f44bac6a576a7f2435791f3 +- package: github.com/qor/inflection + version: 3272df6c21d04180007eb3349844c89a3856bc25 +- package: gopkg.in/mgo.v2 + version: 4d04138ffef2791c479c0c8bbffc30b34081b8d9 +- package: github.com/kr/pty + version: f7ee69f31298ecbe5d2b349c711e2547a617d398 +- package: gopkg.in/fsnotify.v1 + version: 7be54206639f256967dd82fa767397ba5f8f48f5 +- package: github.com/peterh/liner + version: b850cf8c6d0ee52309aad09ac610508c6c75e819 +- package: github.com/cgrates/rpcclient + version: f0f378951e30943373953b44b910cc65ba832da3 +- package: github.com/cgrates/osipsdagram + version: 3d6beed663452471dec3ca194137a30d379d9e8f +- package: github.com/cgrates/kamevapi + version: a376b1f937ba959857929fa3e111c0f3243278c0 +- package: github.com/cgrates/fsock + version: 7ec136e2798fbfe44f713f5ee5828ce643dec4c2 +- package: github.com/mediocregopher/radix.v2 + version: 9c9a238f8e52b72881f19c4e9113182689f11de0 + subpackages: + - '...' From e48413d81c89db112bb19d14720a9271ff23c6fc Mon Sep 17 00:00:00 2001 From: Radu Ioan Fericean Date: Sat, 7 Nov 2015 12:22:09 +0200 Subject: [PATCH 2/9] using Get and Put on redis pool --- cmd/cgr-loader/migrator_rc8.go | 44 ++++++++----- engine/storage_redis.go | 113 ++++++++++++++++++++++++--------- 2 files changed, 112 insertions(+), 45 deletions(-) diff --git a/cmd/cgr-loader/migrator_rc8.go b/cmd/cgr-loader/migrator_rc8.go index 51a7ef182..51935c514 100644 --- a/cmd/cgr-loader/migrator_rc8.go +++ b/cmd/cgr-loader/migrator_rc8.go @@ -8,19 +8,29 @@ import ( "github.com/cgrates/cgrates/engine" "github.com/cgrates/cgrates/utils" - "github.com/hoisie/redis" + "github.com/mediocregopher/radix.v2/redis" ) const OLD_ACCOUNT_PREFIX = "ubl_" type MigratorRC8 struct { - dbNb int - db *redis.Client - ms engine.Marshaler + db *redis.Client + ms engine.Marshaler } func NewMigratorRC8(address string, db int, pass, mrshlerStr string) (*MigratorRC8, error) { - ndb := &redis.Client{Addr: address, Db: db, Password: pass} + client, err := redis.Dial("tcp", address) + if err != nil { + return nil, err + } + if err := client.Cmd("SELECT", db).Err; err != nil { + return nil, err + } + if pass != "" { + if err := client.Cmd("AUTH", pass).Err; err != nil { + return nil, err + } + } var mrshler engine.Marshaler if mrshlerStr == utils.MSGPACK { @@ -30,7 +40,7 @@ func NewMigratorRC8(address string, db int, pass, mrshlerStr string) (*MigratorR } else { return nil, fmt.Errorf("Unsupported marshaler: %v", mrshlerStr) } - return &MigratorRC8{db: ndb, dbNb: db, ms: mrshler}, nil + return &MigratorRC8{db: client, ms: mrshler}, nil } type Account struct { @@ -116,7 +126,7 @@ type Action struct { } func (mig MigratorRC8) migrateAccounts() error { - keys, err := mig.db.Keys(OLD_ACCOUNT_PREFIX + "*") + keys, err := mig.db.Cmd("KEYS", OLD_ACCOUNT_PREFIX+"*").List() if err != nil { return err } @@ -124,7 +134,7 @@ func (mig MigratorRC8) migrateAccounts() error { // get existing accounts for keyIndex, key := range keys { log.Printf("Migrating account: %s...", key) - values, err := mig.db.Get(key) + values, err := mig.db.Cmd("GET", key).Bytes() if err != nil { continue } @@ -240,20 +250,22 @@ func (mig MigratorRC8) migrateAccounts() error { if err != nil { return err } - if err := mig.db.Set(utils.ACCOUNT_PREFIX+newAcc.Id, result); err != nil { + if err := mig.db.Cmd("SET", utils.ACCOUNT_PREFIX+newAcc.Id, result).Err; err != nil { return err } } // delete old data log.Printf("Deleting old accounts: %s...", OLD_ACCOUNT_PREFIX+"*") for _, key := range keys { - _, err = mig.db.Del(key) + if err := mig.db.Cmd("DEL", key).Err; err != nil { + return err + } } return err } func (mig MigratorRC8) migrateActionTriggers() error { - keys, err := mig.db.Keys(utils.ACTION_TRIGGER_PREFIX + "*") + keys, err := mig.db.Cmd("KEYS", utils.ACTION_TRIGGER_PREFIX+"*").List() if err != nil { return err } @@ -262,7 +274,7 @@ func (mig MigratorRC8) migrateActionTriggers() error { log.Printf("Migrating action trigger: %s...", key) var oldAtrs ActionTriggers var values []byte - if values, err = mig.db.Get(key); err == nil { + if values, err = mig.db.Cmd("GET", key).Bytes(); err == nil { if err := mig.ms.Unmarshal(values, &oldAtrs); err != nil { return err } @@ -304,7 +316,7 @@ func (mig MigratorRC8) migrateActionTriggers() error { if err != nil { return err } - if err = mig.db.Set(key, result); err != nil { + if err = mig.db.Cmd("SET", key, result).Err; err != nil { return err } } @@ -312,7 +324,7 @@ func (mig MigratorRC8) migrateActionTriggers() error { } func (mig MigratorRC8) migrateActions() error { - keys, err := mig.db.Keys(utils.ACTION_PREFIX + "*") + keys, err := mig.db.Cmd("KEYS", utils.ACTION_PREFIX+"*").List() if err != nil { return err } @@ -321,7 +333,7 @@ func (mig MigratorRC8) migrateActions() error { log.Printf("Migrating action: %s...", key) var oldAcs Actions var values []byte - if values, err = mig.db.Get(key); err == nil { + if values, err = mig.db.Cmd("GET", key).Bytes(); err == nil { if err := mig.ms.Unmarshal(values, &oldAcs); err != nil { return err } @@ -360,7 +372,7 @@ func (mig MigratorRC8) migrateActions() error { if err != nil { return err } - if err = mig.db.Set(key, result); err != nil { + if err = mig.db.Cmd("SET", key, result).Err; err != nil { return err } } diff --git a/engine/storage_redis.go b/engine/storage_redis.go index db4463908..ca959daa9 100644 --- a/engine/storage_redis.go +++ b/engine/storage_redis.go @@ -40,14 +40,20 @@ type RedisStorage struct { func NewRedisStorage(address string, db int, pass, mrshlerStr string) (*RedisStorage, error) { p, err := pool.New("tcp", address, 10) + //p, err := redis.Dial("tcp", address) if err != nil { return nil, err } - if err := p.Cmd("SELECT", db).Err; err != nil { + conn, err := p.Get() + if err != nil { + return nil, err + } + defer p.Put(conn) + if err := conn.Cmd("SELECT", db).Err; err != nil { return nil, err } if pass != "" { - if err := p.Cmd("AUTH", pass).Err; err != nil { + if err := conn.Cmd("AUTH", pass).Err; err != nil { return nil, err } } @@ -63,8 +69,7 @@ func NewRedisStorage(address string, db int, pass, mrshlerStr string) (*RedisSto } func (rs *RedisStorage) Close() { - // no close for me - //rs.db.Quit() + rs.db.Empty() } func (rs *RedisStorage) Flush(ignore string) error { @@ -123,10 +128,15 @@ func (rs *RedisStorage) CacheRatingPrefixValues(prefixes map[string][]string) er func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, actKeys, shgKeys []string) (err error) { cache2go.BeginTransaction() + conn, err := rs.db.Get() + if err != nil { + return err + } + defer rs.db.Put(conn) if dKeys == nil || (float64(cache2go.CountEntries(utils.DESTINATION_PREFIX))*utils.DESTINATIONS_LOAD_THRESHOLD < float64(len(dKeys))) { // if need to load more than a half of exiting keys load them all utils.Logger.Info("Caching all destinations") - if dKeys, err = rs.db.Cmd("KEYS", utils.DESTINATION_PREFIX+"*").List(); err != nil { + if dKeys, err = conn.Cmd("KEYS", utils.DESTINATION_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -150,7 +160,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if rpKeys == nil { utils.Logger.Info("Caching all rating plans") - if rpKeys, err = rs.db.Cmd("KEYS", utils.RATING_PLAN_PREFIX+"*").List(); err != nil { + if rpKeys, err = conn.Cmd("KEYS", utils.RATING_PLAN_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -170,7 +180,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if rpfKeys == nil { utils.Logger.Info("Caching all rating profiles") - if rpfKeys, err = rs.db.Cmd("KEYS", utils.RATING_PROFILE_PREFIX+"*").List(); err != nil { + if rpfKeys, err = conn.Cmd("KEYS", utils.RATING_PROFILE_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -190,7 +200,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if lcrKeys == nil { utils.Logger.Info("Caching LCR rules.") - if lcrKeys, err = rs.db.Cmd("KEYS", utils.LCR_PREFIX+"*").List(); err != nil { + if lcrKeys, err = conn.Cmd("KEYS", utils.LCR_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -211,7 +221,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac // DerivedChargers caching if dcsKeys == nil { utils.Logger.Info("Caching all derived chargers") - if dcsKeys, err = rs.db.Cmd("KEYS", utils.DERIVEDCHARGERS_PREFIX+"*").List(); err != nil { + if dcsKeys, err = conn.Cmd("KEYS", utils.DERIVEDCHARGERS_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -234,7 +244,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if actKeys == nil { utils.Logger.Info("Caching all actions") - if actKeys, err = rs.db.Cmd("KEYS", utils.ACTION_PREFIX+"*").List(); err != nil { + if actKeys, err = conn.Cmd("KEYS", utils.ACTION_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -258,7 +268,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if shgKeys == nil { utils.Logger.Info("Caching all shared groups") - if shgKeys, err = rs.db.Cmd("KEYS", utils.SHARED_GROUP_PREFIX+"*").List(); err != nil { + if shgKeys, err = conn.Cmd("KEYS", utils.SHARED_GROUP_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -312,12 +322,17 @@ func (rs *RedisStorage) CacheAccountingPrefixValues(prefixes map[string][]string func (rs *RedisStorage) cacheAccounting(alsKeys []string) (err error) { cache2go.BeginTransaction() + conn, err := rs.db.Get() + if err != nil { + return err + } + defer rs.db.Put(conn) if alsKeys == nil { cache2go.RemPrefixKey(utils.ALIASES_PREFIX) } if alsKeys == nil { utils.Logger.Info("Caching all aliases") - if alsKeys, err = rs.db.Cmd("KEYS", utils.ALIASES_PREFIX+"*").List(); err != nil { + if alsKeys, err = conn.Cmd("KEYS", utils.ALIASES_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } @@ -426,12 +441,17 @@ func (rs *RedisStorage) SetRatingProfile(rpf *RatingProfile) (err error) { } func (rs *RedisStorage) RemoveRatingProfile(key string) error { - keys, err := rs.db.Cmd("KEYS", utils.RATING_PROFILE_PREFIX+key+"*").List() + conn, err := rs.db.Get() + if err != nil { + return err + } + defer rs.db.Put(conn) + keys, err := conn.Cmd("KEYS", utils.RATING_PROFILE_PREFIX+key+"*").List() if err != nil { return err } for _, key := range keys { - if err = rs.db.Cmd("DEL", key).Err; err != nil { + if err = conn.Cmd("DEL", key).Err; err != nil { return err } cache2go.RemKey(key) @@ -606,13 +626,18 @@ func (rs *RedisStorage) SetCdrStatsQueue(sq *StatsQueue) (err error) { } func (rs *RedisStorage) GetSubscribers() (result map[string]*SubscriberData, err error) { - keys, err := rs.db.Cmd("KEYS", utils.PUBSUB_SUBSCRIBERS_PREFIX+"*").List() + conn, err := rs.db.Get() + if err != nil { + return nil, err + } + defer rs.db.Put(conn) + keys, err := conn.Cmd("KEYS", utils.PUBSUB_SUBSCRIBERS_PREFIX+"*").List() if err != nil { return nil, err } result = make(map[string]*SubscriberData) for _, key := range keys { - if values, err := rs.db.Cmd("GET", key).Bytes(); err == nil { + if values, err := conn.Cmd("GET", key).Bytes(); err == nil { sub := &SubscriberData{} err = rs.ms.Unmarshal(values, sub) result[key[len(utils.PUBSUB_SUBSCRIBERS_PREFIX):]] = sub @@ -654,12 +679,17 @@ func (rs *RedisStorage) GetUser(key string) (up *UserProfile, err error) { } func (rs *RedisStorage) GetUsers() (result []*UserProfile, err error) { - keys, err := rs.db.Cmd("KEYS", utils.USERS_PREFIX+"*").List() + conn, err := rs.db.Get() + if err != nil { + return nil, err + } + defer rs.db.Put(conn) + keys, err := conn.Cmd("KEYS", utils.USERS_PREFIX+"*").List() if err != nil { return nil, err } for _, key := range keys { - if values, err := rs.db.Cmd("GET", key).Bytes(); err == nil { + if values, err := conn.Cmd("GET", key).Bytes(); err == nil { up := &UserProfile{} err = rs.ms.Unmarshal(values, up) result = append(result, up) @@ -723,15 +753,20 @@ func (rs *RedisStorage) GetAlias(key string, skipCache bool) (al *Alias, err err } func (rs *RedisStorage) RemoveAlias(key string) (err error) { + conn, err := rs.db.Get() + if err != nil { + return err + } + defer rs.db.Put(conn) al := &Alias{} al.SetId(key) origKey := key key = utils.ALIASES_PREFIX + key aliasValues := make(AliasValues, 0) - if values, err := rs.db.Cmd("GET", key).Bytes(); err == nil { + if values, err := conn.Cmd("GET", key).Bytes(); err == nil { rs.ms.Unmarshal(values, &aliasValues) } - err = rs.db.Cmd("DEL", key).Err + err = conn.Cmd("DEL", key).Err if err == nil { for _, value := range aliasValues { for target, pairs := range value.Pairs { @@ -798,6 +833,11 @@ func (rs *RedisStorage) GetLoadHistory(limit int, skipCache bool) ([]*LoadInstan // Adds a single load instance to load history func (rs *RedisStorage) AddLoadHistory(ldInst *LoadInstance, loadHistSize int) error { + conn, err := rs.db.Get() + if err != nil { + return err + } + defer rs.db.Put(conn) if loadHistSize == 0 { // Load history disabled return nil } @@ -806,16 +846,16 @@ func (rs *RedisStorage) AddLoadHistory(ldInst *LoadInstance, loadHistSize int) e return err } _, err = Guardian.Guard(func() (interface{}, error) { // Make sure we do it locked since other instance can modify history while we read it - histLen, err := rs.db.Cmd("LLEN", utils.LOADINST_KEY).Int() + histLen, err := conn.Cmd("LLEN", utils.LOADINST_KEY).Int() if err != nil { return nil, err } if histLen >= loadHistSize { // Have hit maximum history allowed, remove oldest element in order to add new one - if err := rs.db.Cmd("RPOP", utils.LOADINST_KEY).Err; err != nil { + if err := conn.Cmd("RPOP", utils.LOADINST_KEY).Err; err != nil { return nil, err } } - err = rs.db.Cmd("LPUSH", utils.LOADINST_KEY, marshaled).Err + err = conn.Cmd("LPUSH", utils.LOADINST_KEY, marshaled).Err return nil, err }, 0, utils.LOADINST_KEY) return err @@ -830,15 +870,20 @@ func (rs *RedisStorage) GetActionTriggers(key string) (atrs ActionTriggers, err } func (rs *RedisStorage) SetActionTriggers(key string, atrs ActionTriggers) (err error) { + conn, err := rs.db.Get() + if err != nil { + return err + } + defer rs.db.Put(conn) if len(atrs) == 0 { // delete the key - return rs.db.Cmd("DEL", utils.ACTION_TRIGGER_PREFIX+key).Err + return conn.Cmd("DEL", utils.ACTION_TRIGGER_PREFIX+key).Err } result, err := rs.ms.Marshal(&atrs) if err != nil { return err } - return rs.db.Cmd("SET", utils.ACTION_TRIGGER_PREFIX+key, result).Err + return conn.Cmd("SET", utils.ACTION_TRIGGER_PREFIX+key, result).Err } func (rs *RedisStorage) GetActionPlans(key string) (ats ActionPlans, err error) { @@ -862,13 +907,18 @@ func (rs *RedisStorage) SetActionPlans(key string, ats ActionPlans) (err error) } func (rs *RedisStorage) GetAllActionPlans() (ats map[string]ActionPlans, err error) { - keys, err := rs.db.Cmd("KEYS", utils.ACTION_PLAN_PREFIX+"*").List() + conn, err := rs.db.Get() + if err != nil { + return nil, err + } + defer rs.db.Put(conn) + keys, err := conn.Cmd("KEYS", utils.ACTION_PLAN_PREFIX+"*").List() if err != nil { return nil, err } ats = make(map[string]ActionPlans, len(keys)) for _, key := range keys { - values, err := rs.db.Cmd("GET", key).Bytes() + values, err := conn.Cmd("GET", key).Bytes() if err != nil { continue } @@ -926,12 +976,17 @@ func (rs *RedisStorage) GetCdrStats(key string) (cs *CdrStats, err error) { } func (rs *RedisStorage) GetAllCdrStats() (css []*CdrStats, err error) { - keys, err := rs.db.Cmd("KEYS", utils.CDR_STATS_PREFIX+"*").List() + conn, err := rs.db.Get() + if err != nil { + return nil, err + } + defer rs.db.Put(conn) + keys, err := conn.Cmd("KEYS", utils.CDR_STATS_PREFIX+"*").List() if err != nil { return nil, err } for _, key := range keys { - value, err := rs.db.Cmd("GET", key).Bytes() + value, err := conn.Cmd("GET", key).Bytes() if err != nil { continue } From 5df030b1e7a34534a129d2b54a2b37cbfc9e4a62 Mon Sep 17 00:00:00 2001 From: Radu Ioan Fericean Date: Sat, 7 Nov 2015 14:23:13 +0200 Subject: [PATCH 3/9] removed all balances expired --- engine/account.go | 11 ----------- engine/account_test.go | 2 +- engine/storage_map.go | 12 ------------ engine/storage_mongo.go | 12 ------------ engine/storage_redis.go | 14 +------------- 5 files changed, 2 insertions(+), 49 deletions(-) diff --git a/engine/account.go b/engine/account.go index a83a99dcc..3ce64f10c 100644 --- a/engine/account.go +++ b/engine/account.go @@ -602,17 +602,6 @@ func (acc *Account) CleanExpiredBalances() { } } -func (acc *Account) allBalancesExpired() bool { - for _, bm := range acc.BalanceMap { - for i := 0; i < len(bm); i++ { - if !bm[i].IsExpired() { - return false - } - } - } - return true -} - // returns the shared groups that this user balance belnongs to func (acc *Account) GetSharedGroups() (groups []string) { for _, balanceChain := range acc.BalanceMap { diff --git a/engine/account_test.go b/engine/account_test.go index e579232c8..84c6afae9 100644 --- a/engine/account_test.go +++ b/engine/account_test.go @@ -90,7 +90,7 @@ func TestAccountStorageStoreRestore(t *testing.T) { ub1, err := accountingStorage.GetAccount("other") if err != nil || !ub1.BalanceMap[utils.MONETARY].Equal(rifsBalance.BalanceMap[utils.MONETARY]) { t.Log("UB: ", ub1) - t.Errorf("Expected %v was %v", rifsBalance.BalanceMap[utils.MONETARY], ub1.BalanceMap[utils.MONETARY]) + t.Errorf("Expected %v was %v", rifsBalance, ub1) } } diff --git a/engine/storage_map.go b/engine/storage_map.go index da13aec99..76b695484 100644 --- a/engine/storage_map.go +++ b/engine/storage_map.go @@ -456,18 +456,6 @@ func (ms *MapStorage) GetAccount(key string) (ub *Account, err error) { } func (ms *MapStorage) SetAccount(ub *Account) (err error) { - // never override existing account with an empty one - // UPDATE: if all balances expired and were clean it makes - // sense to write empty balance map - if len(ub.BalanceMap) == 0 { - if ac, err := ms.GetAccount(ub.Id); err == nil && !ac.allBalancesExpired() { - ac.ActionTriggers = ub.ActionTriggers - ac.UnitCounters = ub.UnitCounters - ac.AllowNegative = ub.AllowNegative - ac.Disabled = ub.Disabled - ub = ac - } - } result, err := ms.ms.Marshal(ub) ms.dict[utils.ACCOUNT_PREFIX+ub.Id] = result return diff --git a/engine/storage_mongo.go b/engine/storage_mongo.go index 3edad9cc9..5450d063d 100644 --- a/engine/storage_mongo.go +++ b/engine/storage_mongo.go @@ -732,18 +732,6 @@ func (ms *MongoStorage) GetAccount(key string) (result *Account, err error) { } func (ms *MongoStorage) SetAccount(acc *Account) error { - // never override existing account with an empty one - // UPDATE: if all balances expired and were cleaned it makes - // sense to write empty balance map - if len(acc.BalanceMap) == 0 { - if ac, err := ms.GetAccount(acc.Id); err == nil && !ac.allBalancesExpired() { - ac.ActionTriggers = acc.ActionTriggers - ac.UnitCounters = acc.UnitCounters - ac.AllowNegative = acc.AllowNegative - ac.Disabled = acc.Disabled - acc = ac - } - } _, err := ms.db.C(colAcc).Upsert(bson.M{"id": acc.Id}, acc) return err } diff --git a/engine/storage_redis.go b/engine/storage_redis.go index ca959daa9..7c434df3b 100644 --- a/engine/storage_redis.go +++ b/engine/storage_redis.go @@ -366,7 +366,7 @@ func (rs *RedisStorage) HasData(category, subject string) (bool, error) { i, err := rs.db.Cmd("EXISTS", category+subject).Int() return i == 1, err } - return false, errors.New("Unsupported category in HasData") + return false, errors.New("unsupported HasData category") } func (rs *RedisStorage) GetRatingPlan(key string, skipCache bool) (rp *RatingPlan, err error) { @@ -588,18 +588,6 @@ func (rs *RedisStorage) GetAccount(key string) (ub *Account, err error) { } func (rs *RedisStorage) SetAccount(ub *Account) (err error) { - // never override existing account with an empty one - // UPDATE: if all balances expired and were cleaned it makes - // sense to write empty balance map - if len(ub.BalanceMap) == 0 { - if ac, err := rs.GetAccount(ub.Id); err == nil && !ac.allBalancesExpired() { - ac.ActionTriggers = ub.ActionTriggers - ac.UnitCounters = ub.UnitCounters - ac.AllowNegative = ub.AllowNegative - ac.Disabled = ub.Disabled - ub = ac - } - } result, err := rs.ms.Marshal(ub) err = rs.db.Cmd("SET", utils.ACCOUNT_PREFIX+ub.Id, result).Err return From 073bd682c84b97b110d8d1dbd194a86bf6559b72 Mon Sep 17 00:00:00 2001 From: Radu Ioan Fericean Date: Mon, 16 Nov 2015 11:00:10 +0200 Subject: [PATCH 4/9] small fix for radix --- engine/storage_redis.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/engine/storage_redis.go b/engine/storage_redis.go index 5793c1afb..2395ae7d9 100644 --- a/engine/storage_redis.go +++ b/engine/storage_redis.go @@ -270,7 +270,7 @@ func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, ac } if aplKeys == nil { utils.Logger.Info("Caching all action plans") - if aplKeys, err = rs.db.Keys(utils.ACTION_PLAN_PREFIX + "*"); err != nil { + if aplKeys, err = rs.db.Cmd("KEYS", utils.ACTION_PLAN_PREFIX+"*").List(); err != nil { cache2go.RollbackTransaction() return err } From ea0fae3ad032b89de10dbcf2a6d6aa47bced1af3 Mon Sep 17 00:00:00 2001 From: Radu Ioan Fericean Date: Mon, 16 Nov 2015 12:44:33 +0200 Subject: [PATCH 5/9] fixes for radix branch --- engine/account.go | 11 +++++++++++ engine/storage_map.go | 12 ++++++++++++ engine/storage_mongo.go | 12 ++++++++++++ engine/storage_redis.go | 12 ++++++++++++ 4 files changed, 47 insertions(+) diff --git a/engine/account.go b/engine/account.go index cef47a190..65c4f5ad3 100644 --- a/engine/account.go +++ b/engine/account.go @@ -587,6 +587,17 @@ func (acc *Account) CleanExpiredBalances() { } } +func (acc *Account) allBalancesExpired() bool { + for _, bm := range acc.BalanceMap { + for i := 0; i < len(bm); i++ { + if !bm[i].IsExpired() { + return false + } + } + } + return true +} + // returns the shared groups that this user balance belnongs to func (acc *Account) GetSharedGroups() (groups []string) { for _, balanceChain := range acc.BalanceMap { diff --git a/engine/storage_map.go b/engine/storage_map.go index 47468d21b..3cc637969 100644 --- a/engine/storage_map.go +++ b/engine/storage_map.go @@ -468,6 +468,18 @@ func (ms *MapStorage) GetAccount(key string) (ub *Account, err error) { } func (ms *MapStorage) SetAccount(ub *Account) (err error) { + // never override existing account with an empty one + // UPDATE: if all balances expired and were cleaned it makes + // sense to write empty balance map + if len(ub.BalanceMap) == 0 { + if ac, err := ms.GetAccount(ub.Id); err == nil && !ac.allBalancesExpired() { + ac.ActionTriggers = ub.ActionTriggers + ac.UnitCounters = ub.UnitCounters + ac.AllowNegative = ub.AllowNegative + ac.Disabled = ub.Disabled + ub = ac + } + } result, err := ms.ms.Marshal(ub) ms.dict[utils.ACCOUNT_PREFIX+ub.Id] = result return diff --git a/engine/storage_mongo.go b/engine/storage_mongo.go index b08a43ab8..599799bfc 100644 --- a/engine/storage_mongo.go +++ b/engine/storage_mongo.go @@ -763,6 +763,18 @@ func (ms *MongoStorage) GetAccount(key string) (result *Account, err error) { } func (ms *MongoStorage) SetAccount(acc *Account) error { + // never override existing account with an empty one + // UPDATE: if all balances expired and were cleaned it makes + // sense to write empty balance map + if len(acc.BalanceMap) == 0 { + if ac, err := ms.GetAccount(acc.Id); err == nil && !ac.allBalancesExpired() { + ac.ActionTriggers = acc.ActionTriggers + ac.UnitCounters = acc.UnitCounters + ac.AllowNegative = acc.AllowNegative + ac.Disabled = acc.Disabled + acc = ac + } + } _, err := ms.db.C(colAcc).Upsert(bson.M{"id": acc.Id}, acc) return err } diff --git a/engine/storage_redis.go b/engine/storage_redis.go index 2395ae7d9..50b2b3711 100644 --- a/engine/storage_redis.go +++ b/engine/storage_redis.go @@ -614,6 +614,18 @@ func (rs *RedisStorage) GetAccount(key string) (ub *Account, err error) { } func (rs *RedisStorage) SetAccount(ub *Account) (err error) { + // never override existing account with an empty one + // UPDATE: if all balances expired and were cleaned it makes + // sense to write empty balance map + if len(ub.BalanceMap) == 0 { + if ac, err := rs.GetAccount(ub.Id); err == nil && !ac.allBalancesExpired() { + ac.ActionTriggers = ub.ActionTriggers + ac.UnitCounters = ub.UnitCounters + ac.AllowNegative = ub.AllowNegative + ac.Disabled = ub.Disabled + ub = ac + } + } result, err := rs.ms.Marshal(ub) err = rs.db.Cmd("SET", utils.ACCOUNT_PREFIX+ub.Id, result).Err return From c6516c3aee125ebc21a3a42c22842993d78f7899 Mon Sep 17 00:00:00 2001 From: Radu Ioan Fericean Date: Mon, 16 Nov 2015 14:31:38 +0200 Subject: [PATCH 6/9] change migrator to use radix --- cmd/cgr-loader/migrator_rc8.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cmd/cgr-loader/migrator_rc8.go b/cmd/cgr-loader/migrator_rc8.go index c54527aaf..caca132c3 100644 --- a/cmd/cgr-loader/migrator_rc8.go +++ b/cmd/cgr-loader/migrator_rc8.go @@ -380,7 +380,7 @@ func (mig MigratorRC8) migrateActions() error { } func (mig MigratorRC8) migrateDerivedChargers() error { - keys, err := mig.db.Keys(utils.DERIVEDCHARGERS_PREFIX + "*") + keys, err := mig.db.Cmd("KEYS", utils.DERIVEDCHARGERS_PREFIX+"*").List() if err != nil { return err } @@ -389,7 +389,7 @@ func (mig MigratorRC8) migrateDerivedChargers() error { log.Printf("Migrating derived charger: %s...", key) var oldDcs []*utils.DerivedCharger var values []byte - if values, err = mig.db.Get(key); err == nil { + if values, err = mig.db.Cmd("GET", key).Bytes(); err == nil { if err := mig.ms.Unmarshal(values, &oldDcs); err != nil { return err } @@ -406,7 +406,7 @@ func (mig MigratorRC8) migrateDerivedChargers() error { if err != nil { return err } - if err = mig.db.Set(key, result); err != nil { + if err = mig.db.Cmd("SET", key, result).Err; err != nil { return err } } From 2643523787c947a7a4a25aa7ff1d202e2d2fcb27 Mon Sep 17 00:00:00 2001 From: DanB Date: Mon, 16 Nov 2015 13:36:41 +0100 Subject: [PATCH 7/9] Modified redis to handle auth and select per connection in the pool, added maxConns parameter in constructor --- engine/calldesc.go | 4 ++-- engine/storage_redis.go | 38 +++++++++++++++++------------- engine/storage_redis_local_test.go | 2 +- engine/storage_utils.go | 4 ++-- utils/consts.go | 1 + 5 files changed, 28 insertions(+), 21 deletions(-) diff --git a/engine/calldesc.go b/engine/calldesc.go index c37da1bf1..31290a67f 100644 --- a/engine/calldesc.go +++ b/engine/calldesc.go @@ -56,11 +56,11 @@ func init() { log.Fatal(err) } case "redis": - ratingStorage, _ = NewRedisStorage("127.0.0.1:6379", 12, "", utils.MSGPACK) + ratingStorage, _ = NewRedisStorage("127.0.0.1:6379", 12, "", utils.MSGPACK, utils.REDIS_MAX_CONNS) if err != nil { log.Fatal(err) } - accountingStorage, _ = NewRedisStorage("127.0.0.1:6379", 13, "", utils.MSGPACK) + accountingStorage, _ = NewRedisStorage("127.0.0.1:6379", 13, "", utils.MSGPACK, utils.REDIS_MAX_CONNS) if err != nil { log.Fatal(err) } diff --git a/engine/storage_redis.go b/engine/storage_redis.go index 2395ae7d9..9c85221d5 100644 --- a/engine/storage_redis.go +++ b/engine/storage_redis.go @@ -28,6 +28,7 @@ import ( "github.com/cgrates/cgrates/cache2go" "github.com/cgrates/cgrates/utils" "github.com/mediocregopher/radix.v2/pool" + "github.com/mediocregopher/radix.v2/redis" "io/ioutil" "time" @@ -38,24 +39,29 @@ type RedisStorage struct { ms Marshaler } -func NewRedisStorage(address string, db int, pass, mrshlerStr string) (*RedisStorage, error) { - p, err := pool.New("tcp", address, 10) - //p, err := redis.Dial("tcp", address) - if err != nil { - return nil, err - } - conn, err := p.Get() - if err != nil { - return nil, err - } - defer p.Put(conn) - if err := conn.Cmd("SELECT", db).Err; err != nil { - return nil, err - } - if pass != "" { - if err := conn.Cmd("AUTH", pass).Err; err != nil { +func NewRedisStorage(address string, db int, pass, mrshlerStr string, maxConns int) (*RedisStorage, error) { + df := func(network, addr string) (*redis.Client, error) { + client, err := redis.Dial(network, addr) + if err != nil { return nil, err } + if len(pass) != 0 { + if err = client.Cmd("AUTH", pass).Err; err != nil { + client.Close() + return nil, err + } + } + if db != 0 { + if err = client.Cmd("SELECT", db).Err; err != nil { + client.Close() + return nil, err + } + } + return client, nil + } + p, err := pool.NewCustom("tcp", address, maxConns, df) + if err != nil { + return nil, err } var mrshler Marshaler if mrshlerStr == utils.MSGPACK { diff --git a/engine/storage_redis_local_test.go b/engine/storage_redis_local_test.go index 5a977bf6d..b4c43b05e 100644 --- a/engine/storage_redis_local_test.go +++ b/engine/storage_redis_local_test.go @@ -35,7 +35,7 @@ func TestConnectRedis(t *testing.T) { return } cfg, _ := config.NewDefaultCGRConfig() - rds, err = NewRedisStorage(fmt.Sprintf("%s:%s", cfg.TpDbHost, cfg.TpDbPort), 4, cfg.TpDbPass, cfg.DBDataEncoding) + rds, err = NewRedisStorage(fmt.Sprintf("%s:%s", cfg.TpDbHost, cfg.TpDbPort), 4, cfg.TpDbPass, cfg.DBDataEncoding, utils.REDIS_MAX_CONNS) if err != nil { t.Fatal("Could not connect to Redis", err.Error()) } diff --git a/engine/storage_utils.go b/engine/storage_utils.go index ee2293d7a..da1292c3e 100644 --- a/engine/storage_utils.go +++ b/engine/storage_utils.go @@ -40,7 +40,7 @@ func ConfigureRatingStorage(db_type, host, port, name, user, pass, marshaler str if port != "" { host += ":" + port } - d, err = NewRedisStorage(host, db_nb, pass, marshaler) + d, err = NewRedisStorage(host, db_nb, pass, marshaler, utils.REDIS_MAX_CONNS) default: err = errors.New("unknown db") } @@ -63,7 +63,7 @@ func ConfigureAccountingStorage(db_type, host, port, name, user, pass, marshaler if port != "" { host += ":" + port } - d, err = NewRedisStorage(host, db_nb, pass, marshaler) + d, err = NewRedisStorage(host, db_nb, pass, marshaler, utils.REDIS_MAX_CONNS) case utils.MONGO: d, err = NewMongoStorage(host, port, name, user, pass) db = d.(AccountingStorage) diff --git a/utils/consts.go b/utils/consts.go index 204589e43..53f1446a8 100644 --- a/utils/consts.go +++ b/utils/consts.go @@ -31,6 +31,7 @@ var ( const ( VERSION = "0.9.1~rc8" DIAMETER_FIRMWARE_REVISION = 918 + REDIS_MAX_CONNS = 10 POSTGRES = "postgres" MYSQL = "mysql" MONGO = "mongo" From ebe9dc4f3504702aca231cf50ac852249042daa0 Mon Sep 17 00:00:00 2001 From: DanB Date: Mon, 16 Nov 2015 14:00:18 +0100 Subject: [PATCH 8/9] Fix flushdb command --- engine/storage_redis.go | 14 +------------- 1 file changed, 1 insertion(+), 13 deletions(-) diff --git a/engine/storage_redis.go b/engine/storage_redis.go index eda468131..2cddfa257 100644 --- a/engine/storage_redis.go +++ b/engine/storage_redis.go @@ -79,7 +79,7 @@ func (rs *RedisStorage) Close() { } func (rs *RedisStorage) Flush(ignore string) error { - return rs.db.Cmd("FLUSH").Err + return rs.db.Cmd("FLUSHDB").Err } func (rs *RedisStorage) GetKeysForPrefix(prefix string) ([]string, error) { @@ -620,18 +620,6 @@ func (rs *RedisStorage) GetAccount(key string) (ub *Account, err error) { } func (rs *RedisStorage) SetAccount(ub *Account) (err error) { - // never override existing account with an empty one - // UPDATE: if all balances expired and were cleaned it makes - // sense to write empty balance map - if len(ub.BalanceMap) == 0 { - if ac, err := rs.GetAccount(ub.Id); err == nil && !ac.allBalancesExpired() { - ac.ActionTriggers = ub.ActionTriggers - ac.UnitCounters = ub.UnitCounters - ac.AllowNegative = ub.AllowNegative - ac.Disabled = ub.Disabled - ub = ac - } - } result, err := rs.ms.Marshal(ub) err = rs.db.Cmd("SET", utils.ACCOUNT_PREFIX+ub.Id, result).Err return From 576e16259d4f8eebe4b65baf2b86f50c425dfb7b Mon Sep 17 00:00:00 2001 From: Radu Ioan Fericean Date: Mon, 16 Nov 2015 15:17:48 +0200 Subject: [PATCH 9/9] fix radix set account --- engine/storage_redis.go | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/engine/storage_redis.go b/engine/storage_redis.go index 2cddfa257..28f836ea4 100644 --- a/engine/storage_redis.go +++ b/engine/storage_redis.go @@ -620,6 +620,18 @@ func (rs *RedisStorage) GetAccount(key string) (ub *Account, err error) { } func (rs *RedisStorage) SetAccount(ub *Account) (err error) { + // never override existing account with an empty one + // UPDATE: if all balances expired and were cleaned it makes + // sense to write empty balance map + if len(ub.BalanceMap) == 0 { + if ac, err := rs.GetAccount(ub.Id); err == nil && !ac.allBalancesExpired() { + ac.ActionTriggers = ub.ActionTriggers + ac.UnitCounters = ub.UnitCounters + ac.AllowNegative = ub.AllowNegative + ac.Disabled = ub.Disabled + ub = ac + } + } result, err := rs.ms.Marshal(ub) err = rs.db.Cmd("SET", utils.ACCOUNT_PREFIX+ub.Id, result).Err return