uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-11 18:16:24 +05:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'radix'

Browse Source
This commit is contained in:
Radu Ioan Fericean
2015-11-16 15:21:02 +02:00
parent e21db25892 576e16259d
commit d18c6c7a7c
11 changed files with 213 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
cmd/cgr-loader/migrator_rc8.go
View File

@@ -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
}
@@ -125,7 +135,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
}
@@ -249,14 +259,16 @@ 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 migrated accounts...")
for _, key := range migratedKeys {
_, err = mig.db.Del(key)
if err := mig.db.Cmd("DEL", key).Err; err != nil {
return err
}
}
notMigrated := len(keys) - len(migratedKeys)
if notMigrated > 0 {
@@ -266,7 +278,7 @@ func (mig MigratorRC8) migrateAccounts() error {
}
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
}
@@ -275,7 +287,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
}
@@ -317,7 +329,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
}
}
@@ -325,7 +337,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
}
@@ -334,7 +346,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
}
@@ -373,7 +385,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
}
}
@@ -381,7 +393,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
}
@@ -390,7 +402,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
}
@@ -407,7 +419,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
}
}

2
devel.yaml
View File

@@ -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,6 +16,7 @@ import:
- package: github.com/cgrates/osipsdagram
- package: github.com/cgrates/kamevapi
- package: github.com/cgrates/fsock
- package: github.com/mediocregopher/radix.v2/...
- package: github.com/cenkalti/rpc2
- package: github.com/cenkalti/hub
- package: github.com/fiorix/go-diameter

2
engine/account_test.go
View File

@@ -91,7 +91,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)
}
}

6
engine/calldesc.go
View File

@@ -37,7 +37,7 @@ const (
RECURSION_MAX_DEPTH = 3
MIN_PREFIX_MATCH = 1
FALLBACK_SUBJECT = utils.ANY
DB = "map"
DB = "redis"
)
func init() {
@@ -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)
}

8
engine/loader_local_test.go
View File

@@ -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())
}

2
engine/storage_map.go
View File

@@ -469,7 +469,7 @@ 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
// 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() {

254
engine/storage_redis.go
View File

@@ -27,21 +27,42 @@ import (
"github.com/cgrates/cgrates/cache2go"
"github.com/cgrates/cgrates/utils"
"github.com/hoisie/redis"
"github.com/mediocregopher/radix.v2/pool"
"github.com/mediocregopher/radix.v2/redis"
"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}
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 {
mrshler = NewCodecMsgpackMarshaler()
@@ -50,21 +71,23 @@ 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() {
// no close for me
//rs.db.Quit()
rs.db.Empty()
}
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("FLUSHDB").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 {
@@ -113,10 +136,15 @@ func (rs *RedisStorage) CacheRatingPrefixValues(prefixes map[string][]string) er
func (rs *RedisStorage) cacheRating(dKeys, rpKeys, rpfKeys, lcrKeys, dcsKeys, actKeys, aplKeys, 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.Keys(utils.DESTINATION_PREFIX + "*"); err != nil {
if dKeys, err = conn.Cmd("KEYS", utils.DESTINATION_PREFIX+"*").List(); err != nil {
cache2go.RollbackTransaction()
return err
}
@@ -140,7 +168,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 = conn.Cmd("KEYS", utils.RATING_PLAN_PREFIX+"*").List(); err != nil {
cache2go.RollbackTransaction()
return err
}
@@ -160,7 +188,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 = conn.Cmd("KEYS", utils.RATING_PROFILE_PREFIX+"*").List(); err != nil {
cache2go.RollbackTransaction()
return err
}
@@ -180,7 +208,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 = conn.Cmd("KEYS", utils.LCR_PREFIX+"*").List(); err != nil {
cache2go.RollbackTransaction()
return err
}
@@ -201,7 +229,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 = conn.Cmd("KEYS", utils.DERIVEDCHARGERS_PREFIX+"*").List(); err != nil {
cache2go.RollbackTransaction()
return err
}
@@ -224,7 +252,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 = conn.Cmd("KEYS", utils.ACTION_PREFIX+"*").List(); err != nil {
cache2go.RollbackTransaction()
return err
}
@@ -248,7 +276,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
}
@@ -272,7 +300,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 = conn.Cmd("KEYS", utils.SHARED_GROUP_PREFIX+"*").List(); err != nil {
cache2go.RollbackTransaction()
return err
}
@@ -326,12 +354,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.Keys(utils.ALIASES_PREFIX + "*"); err != nil {
if alsKeys, err = conn.Cmd("KEYS", utils.ALIASES_PREFIX+"*").List(); err != nil {
cache2go.RollbackTransaction()
return err
}
@@ -362,9 +395,10 @@ 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")
return false, errors.New("unsupported HasData category")
}
func (rs *RedisStorage) GetRatingPlan(key string, skipCache bool) (rp *RatingPlan, err error) {
@@ -377,7 +411,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 {
@@ -401,7 +435,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)
@@ -420,7 +454,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)
@@ -430,7 +464,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)
@@ -439,12 +473,17 @@ 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 + "*")
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.Del(key); err != nil {
if err = conn.Cmd("DEL", key).Err; err != nil {
return err
}
cache2go.RemKey(key)
@@ -467,7 +506,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)
}
@@ -476,7 +515,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
}
@@ -484,7 +523,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 {
@@ -516,7 +555,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)
@@ -534,7 +573,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)
}
@@ -543,7 +582,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
}
@@ -557,7 +596,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)
}
@@ -566,13 +605,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)
}
@@ -594,19 +633,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)
}
@@ -615,18 +653,23 @@ 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 + "*")
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.Get(key); 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
@@ -639,24 +682,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)
}
@@ -664,12 +711,17 @@ 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 + "*")
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.Get(key); err == nil {
if values, err := conn.Cmd("GET", key).Bytes(); err == nil {
up := &UserProfile{}
err = rs.ms.Unmarshal(values, up)
result = append(result, up)
@@ -681,14 +733,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) {
@@ -704,7 +757,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)
@@ -732,15 +785,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.Get(key); err == nil {
if values, err := conn.Cmd("GET", key).Bytes(); err == nil {
rs.ms.Unmarshal(values, &aliasValues)
}
_, err = rs.db.Del(key)
err = conn.Cmd("DEL", key).Err
if err == nil {
for _, value := range aliasValues {
for target, pairs := range value.Pairs {
@@ -787,7 +845,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
}
@@ -807,6 +865,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
}
@@ -815,16 +878,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 := 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.Rpop(utils.LOADINST_KEY); err != nil {
if err := conn.Cmd("RPOP", utils.LOADINST_KEY).Err; err != nil {
return nil, err
}
}
err = rs.db.Lpush(utils.LOADINST_KEY, marshaled)
err = conn.Cmd("LPUSH", utils.LOADINST_KEY, marshaled).Err
return nil, err
}, 0, utils.LOADINST_KEY)
return err
@@ -832,24 +895,27 @@ 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
}
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
_, err = rs.db.Del(utils.ACTION_TRIGGER_PREFIX + key)
return err
return conn.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 conn.Cmd("SET", utils.ACTION_TRIGGER_PREFIX+key, result).Err
}
func (rs *RedisStorage) GetActionPlans(key string, skipCache bool) (ats ActionPlans, err error) {
@@ -862,7 +928,8 @@ func (rs *RedisStorage) GetActionPlans(key string, skipCache bool) (ats ActionPl
}
}
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, &ats)
cache2go.Cache(key, ats)
}
@@ -872,7 +939,7 @@ func (rs *RedisStorage) GetActionPlans(key string, skipCache bool) (ats ActionPl
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)
err = rs.db.Cmd("DEL", utils.ACTION_PLAN_PREFIX+key).Err
cache2go.RemKey(utils.ACTION_PLAN_PREFIX + key)
return err
}
@@ -880,8 +947,7 @@ func (rs *RedisStorage) SetActionPlans(key string, ats ActionPlans) (err error)
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) {
@@ -909,7 +975,7 @@ func (rs *RedisStorage) GetDerivedChargers(key string, skipCache bool) (dcs *uti
}
}
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)
}
@@ -918,36 +984,45 @@ func (rs *RedisStorage) GetDerivedChargers(key string, skipCache bool) (dcs *uti
func (rs *RedisStorage) SetDerivedChargers(key string, dcs *utils.DerivedChargers) (err error) {
if dcs == nil || len(dcs.Chargers) == 0 {
_, err = rs.db.Del(utils.DERIVEDCHARGERS_PREFIX + key)
err = rs.db.Cmd("DEL", utils.DERIVEDCHARGERS_PREFIX+key).Err
cache2go.RemKey(utils.DERIVEDCHARGERS_PREFIX + key)
return 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 + "*")
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.Get(key)
value, err := conn.Cmd("GET", key).Bytes()
if err != nil {
continue
}
@@ -964,13 +1039,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
@@ -985,8 +1059,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) {
@@ -998,6 +1071,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
}

2
engine/storage_redis_local_test.go
View File

@@ -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())
}

4
engine/storage_utils.go
View File

@@ -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)

5
glide.yaml
View File

@@ -1,3 +1,4 @@
parent: null
package: github.com/cgrates/cgrates
import:
- package: github.com/ugorji/go
@@ -40,3 +41,7 @@ import:
ref: 57d753b5f4856e77b3cf8ecce78c97215a7d324d
- package: github.com/fiorix/go-diameter
ref: b4c1bac20b8e8e1ac7e17fb54dc83b155aacba21
- package: github.com/mediocregopher/radix.v2
version: 9c9a238f8e52b72881f19c4e9113182689f11de0
subpackages:
- '...'

1
utils/consts.go
View File

@@ -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"