uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-20 06:38:45 +05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #707 from TeoV/master

Browse Source 
Add TPStats and test for and add loader_it_test.go in apier/v1/
This commit is contained in:
Dan Christian Bogos
2017-08-24 18:51:08 +02:00
committed by GitHub
parent edae4299a3 d862a435d2
commit 2d0146d496
20 changed files with 524 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
apier/v1/loader_it_test.go Normal file
View File

@@ -0,0 +1,99 @@
// +build integration
/*
Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
Copyright (C) ITsysCOM GmbH
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>
*/
package v1
import (
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
"net/rpc"
"net/rpc/jsonrpc"
"path"
"testing"
)
var loaderCfgPath string
var loaderCfg *config.CGRConfig
var loaderRPC *rpc.Client
var loaderDataDir = "/usr/share/cgrates"
func TestLoaderInitCfg(t *testing.T) {
var err error
loaderCfgPath = path.Join(loaderDataDir, "conf", "samples", "tutmysql")
loaderCfg, err = config.NewCGRConfigFromFolder(loaderCfgPath)
if err != nil {
t.Error(err)
}
loaderCfg.DataFolderPath = loaderDataDir // Share DataFolderPath through config towards StoreDb for Flush()
config.SetCgrConfig(loaderCfg)
}
func TestLoaderInitDataDb(t *testing.T) {
if err := engine.InitDataDb(loaderCfg); err != nil {
t.Fatal(err)
}
}
// Wipe out the cdr database
func TestLoaderStorDb(t *testing.T) {
if err := engine.InitStorDb(loaderCfg); err != nil {
t.Fatal(err)
}
}
// Start CGR Engine
func TestLoaderStartEngine(t *testing.T) {
if _, err := engine.StopStartEngine(loaderCfgPath, 1000); err != nil {
t.Fatal(err)
}
}
// Connect rpc client to rater
func TestLoaderRpcConn(t *testing.T) {
var err error
loaderRPC, err = jsonrpc.Dial("tcp", loaderCfg.RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
if err != nil {
t.Fatal(err)
}
}
func TestLoaderImportTPFromFolderPath(t *testing.T) {
var reply string
if err := loaderRPC.Call("ApierV1.ImportTariffPlanFromFolder", utils.AttrImportTPFromFolder{TPid: "TEST_LOADER", FolderPath: path.Join(loaderDataDir, "tariffplans", "tutorial")}, &reply); err != nil {
t.Error("Got error on ApierV1.ImportTarrifPlanFromFolder: ", err.Error())
} else if reply != utils.OK {
t.Error("Calling ApierV1.ImportTarrifPlanFromFolder got reply: ", reply)
}
}
func TestLoaderLoadTariffPlanFromStorDbDryRun(t *testing.T) {
var reply string
if err := loaderRPC.Call("ApierV1.LoadTariffPlanFromStorDb", AttrLoadTpFromStorDb{TPid: "TEST_LOADER", DryRun: true}, &reply); err != nil {
t.Error("Got error on ApierV1.LoadTariffPlanFromStorDb: ", err.Error())
} else if reply != utils.OK {
t.Error("Calling ApierV1.LoadTariffPlanFromStorDb got reply: ", reply)
}
}
func TestLoaderKillEngine(t *testing.T) {
if err := engine.KillEngine(100); err != nil {
t.Error(err)
}
}

13
apier/v1/resourcesv1.go
View File

@@ -79,3 +79,16 @@ func (rsv1 *ResourceSV1) AllocateResource(args utils.AttrRLsResourceUsage, reply
func (rsv1 *ResourceSV1) ReleaseResource(args utils.AttrRLsResourceUsage, reply *string) error {
return rsv1.rls.V1ReleaseResource(args, reply)
}
//after implement test for it
func (apierV1 *ApierV1) GetResourceConfig() {
}
func (apierV1 *ApierV1) SetResourceConfig() {
}
func (apierV1 *ApierV1) RemResourceConfig() {
}

13
apier/v1/stats.go
View File

@@ -87,3 +87,16 @@ func (stsv1 *StatSV1) GetFloatMetrics(queueID string, reply *map[string]float64)
func (stsv1 *StatSV1) LoadQueues(args stats.ArgsLoadQueues, reply *string) (err error) {
return stsv1.sts.V1LoadQueues(args, reply)
}
//after implement test for it
func (apierV1 *ApierV1) GetStatConfig() {
}
func (apierV1 *ApierV1) SetStatConfig() {
}
func (apierV1 *ApierV1) RemStatConfig() {
}

85
apier/v1/tpstats.go Normal file
View File

@@ -0,0 +1,85 @@
/*
Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
Copyright (C) ITsysCOM GmbH
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>
*/
package v1
import (
"github.com/cgrates/cgrates/utils"
)
func (self *ApierV1) SetTPStat(attr utils.TPStats, reply *string) error {
if missing := utils.MissingStructFields(&attr, []string{"TPid", "ID"}); len(missing) != 0 {
return utils.NewErrMandatoryIeMissing(missing...)
}
if err := self.StorDb.SetTPStats([]*utils.TPStats{&attr}); err != nil {
return utils.APIErrorHandler(err)
}
*reply = utils.OK
return nil
}
type AttrGetTPStat struct {
TPid string // Tariff plan id
ID string
}
func (self *ApierV1) GetTPStat(attr AttrGetTPStat, reply *utils.TPStats) error {
if missing := utils.MissingStructFields(&attr, []string{"TPid", "ID"}); len(missing) != 0 { //Params missing
return utils.NewErrMandatoryIeMissing(missing...)
}
if rls, err := self.StorDb.GetTPStats(attr.TPid, attr.ID); err != nil {
if err.Error() != utils.ErrNotFound.Error() {
err = utils.NewErrServerError(err)
}
return err
} else {
*reply = *rls[0]
}
return nil
}
type AttrGetTPStatIds struct {
TPid string // Tariff plan id
utils.Paginator
}
func (self *ApierV1) GetTPStatIDs(attrs AttrGetTPStatIds, reply *[]string) error {
if missing := utils.MissingStructFields(&attrs, []string{"TPid"}); len(missing) != 0 { //Params missing
return utils.NewErrMandatoryIeMissing(missing...)
}
if ids, err := self.StorDb.GetTpTableIds(attrs.TPid, utils.TBLTPStats, utils.TPDistinctIds{"tag"}, nil, &attrs.Paginator); err != nil {
return utils.NewErrServerError(err)
} else if ids == nil {
return utils.ErrNotFound
} else {
*reply = ids
}
return nil
}
func (self *ApierV1) RemTPStat(attrs AttrGetTPStat, reply *string) error {
if missing := utils.MissingStructFields(&attrs, []string{"TPid", "ID"}); len(missing) != 0 { //Params missing
return utils.NewErrMandatoryIeMissing(missing...)
}
if err := self.StorDb.RemTpData(utils.TBLTPStats, attrs.TPid, map[string]string{"tag": attrs.ID}); err != nil {
return utils.NewErrServerError(err)
} else {
*reply = utils.OK
}
return nil
}

178
apier/v1/tpstats_it_test.go Normal file
View File

@@ -0,0 +1,178 @@
// +build integration
/*
Real-time Online/Offline Charging System (OCS) for Telecom & ISP environments
Copyright (C) ITsysCOM GmbH
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>
*/
package v1
import (
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
"net/rpc"
"net/rpc/jsonrpc"
"path"
"reflect"
"testing"
)
var tpCfgPath string
var tpCfg *config.CGRConfig
var tpRPC *rpc.Client
var tpDataDir = "/usr/share/cgrates"
func TestTPStatInitCfg(t *testing.T) {
var err error
tpCfgPath = path.Join(tpDataDir, "conf", "samples", "tutmysql")
tpCfg, err = config.NewCGRConfigFromFolder(tpCfgPath)
if err != nil {
t.Error(err)
}
tpCfg.DataFolderPath = tpDataDir // Share DataFolderPath through config towards StoreDb for Flush()
config.SetCgrConfig(tpCfg)
}
// Wipe out the cdr database
func TestTPStatResetStorDb(t *testing.T) {
if err := engine.InitStorDb(tpCfg); err != nil {
t.Fatal(err)
}
}
// Start CGR Engine
func TestTPStatStartEngine(t *testing.T) {
if _, err := engine.StopStartEngine(tpCfgPath, 1000); err != nil {
t.Fatal(err)
}
}
// Connect rpc client to rater
func TestTPStatRpcConn(t *testing.T) {
var err error
tpRPC, err = jsonrpc.Dial("tcp", tpCfg.RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
if err != nil {
t.Fatal(err)
}
}
var tpStat = &utils.TPStats{
TPid: "TPS1",
ID: "Stat1",
Filters: []*utils.TPRequestFilter{
&utils.TPRequestFilter{
Type: "*string",
FieldName: "Account",
Values: []string{"1001", "1002"},
},
&utils.TPRequestFilter{
Type: "*string_prefix",
FieldName: "Destination",
Values: []string{"10", "20"},
},
},
ActivationInterval: &utils.TPActivationInterval{
ActivationTime: "2014-07-29T15:00:00Z",
ExpiryTime: "",
},
TTL: "1",
Metrics: []string{"MetricValue", "MetricValueTwo"},
Blocker: true,
Stored: true,
Weight: 20,
Thresholds: nil,
}
func TestTPStatGetTPStatIDs(t *testing.T) {
var reply []string
if err := tpRPC.Call("ApierV1.GetTPStatIDs", AttrGetTPStatIds{TPid: "TPS1"}, &reply); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
}
func TestTPStatSetTPStat(t *testing.T) {
var result string
if err := tpRPC.Call("ApierV1.SetTPStat", tpStat, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
}
func TestTPStatGetTPStat(t *testing.T) {
var respond *utils.TPStats
if err := tpRPC.Call("ApierV1.GetTPStat", &AttrGetTPStat{TPid: tpStat.TPid, ID: tpStat.ID}, &respond); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(tpStat, respond) {
t.Errorf("Expecting: %+v, received: %+v", tpStat.TPid, respond.TPid)
}
}
func TestTPStatUpdateTPStat(t *testing.T) {
var result string
tpStat.Weight = 21
tpStat.Filters = []*utils.TPRequestFilter{
&utils.TPRequestFilter{
Type: "*string",
FieldName: "Account",
Values: []string{"1001", "1002"},
},
&utils.TPRequestFilter{
Type: "*string_prefix",
FieldName: "Destination",
Values: []string{"10", "20"},
},
&utils.TPRequestFilter{
Type: "*rsr_fields",
FieldName: "",
Values: []string{"Subject(~^1.*1$)", "Destination(1002)"},
},
}
if err := tpRPC.Call("ApierV1.SetTPStat", tpStat, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
var expectedTPS *utils.TPStats
if err := tpRPC.Call("ApierV1.GetTPStat", &AttrGetTPStat{TPid: tpStat.TPid, ID: tpStat.ID}, &expectedTPS); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(tpStat, expectedTPS) {
t.Errorf("Expecting: %+v, received: %+v", tpStat, expectedTPS)
}
}
func TestTPStatRemTPStat(t *testing.T) {
var resp string
if err := tpRPC.Call("ApierV1.RemTPStat", &AttrGetTPStat{TPid: tpStat.TPid, ID: tpStat.ID}, &resp); err != nil {
t.Error(err)
} else if resp != utils.OK {
t.Error("Unexpected reply returned", resp)
}
}
func TestTPStatCheckDelete(t *testing.T) {
var respond *utils.TPStats
if err := tpRPC.Call("ApierV1.GetTPStat", &AttrGetTPStat{TPid: "TPS1", ID: "Stat1"}, &respond); err == nil || err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
}
func TestTPStatKillEngine(t *testing.T) {
if err := engine.KillEngine(100); err != nil {
t.Error(err)
}
}

2
engine/statsqueue.go → engine/libstats.go
View File

@@ -39,7 +39,7 @@ type SQStoredMetrics struct {
}
// StatsQueue represents the configuration of a StatsInstance in StatS
type StatsQueue struct {
type StatsConfig struct {
ID string // QueueID
Filters []*RequestFilter
ActivationInterval *utils.ActivationInterval // Activation interval

11
engine/model_helpers.go
View File

@@ -2024,8 +2024,11 @@ func APItoModelStats(st *utils.TPStats) (mdls TpStatsS) {
mdl.Stored = st.Stored
mdl.Weight = st.Weight
mdl.QueueLength = st.QueueLength
for _, val := range st.Metrics {
mdl.Metrics = mdl.Metrics + utils.INFIELD_SEP + val
for i, val := range st.Metrics {
if i != 0 {
mdl.Metrics += utils.INFIELD_SEP
}
mdl.Metrics += val
}
for _, val := range st.Thresholds {
mdl.Thresholds = mdl.Thresholds + utils.INFIELD_SEP + val
@@ -2053,8 +2056,8 @@ func APItoModelStats(st *utils.TPStats) (mdls TpStatsS) {
return
}
func APItoStats(tpST *utils.TPStats, timezone string) (st *StatsQueue, err error) {
st = &StatsQueue{
func APItoStats(tpST *utils.TPStats, timezone string) (st *StatsConfig, err error) {
st = &StatsConfig{
ID: tpST.ID,
QueueLength: tpST.QueueLength,
Weight: tpST.Weight,

44
engine/model_helpers_test.go
View File

@@ -15,6 +15,7 @@ GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>
*/
package engine
import (
@@ -188,6 +189,47 @@ func TestApierTPTimingAsExportSlice(t *testing.T) {
}
}
/*
func TestAPItoModelStats(t *testing.T) {
tpS := &utils.TPStats{
TPid: "TPS1",
ID: "Stat1",
Filters: []*utils.TPRequestFilter{
&utils.TPRequestFilter{
Type: "*string",
FieldName: "Account",
Values: []string{"1002"},
},
},
ActivationInterval: &utils.TPActivationInterval{
ActivationTime: "2014-07-29T15:00:00Z",
ExpiryTime: "",
},
TTL: "1",
Metrics: []string{"MetricValue"},
Blocker: true,
Stored: true,
Weight: 20,
Thresholds: nil,
}
expectedSlc := [][]string{
[]string{,"TPS1", "*Stat1", "*string", "*Account", "1002", "2014-07-29T15:00:00Z","","1","MetricValue",},
}
expectedtpS := APItoModelStats(tpS)
var slc [][]string
lc, err := csvDump(expectedtpS)
if err != nil {
t.Error("Error dumping to csv: ", err)
}
slc = append(slc, lc)
if !reflect.DeepEqual(expectedtpS, tpS) {
t.Errorf("Expecting: %+v, received: %+v", expectedtpS, slc)
}
}
*/
func TestTPRatingPlanAsExportSlice(t *testing.T) {
tpRpln := &utils.TPRatingPlan{
TPid: "TEST_TPID",
@@ -859,7 +901,7 @@ func TestAPItoTPStats(t *testing.T) {
Weight: 20.0,
}
eTPs := &StatsQueue{ID: tps.ID,
eTPs := &StatsConfig{ID: tps.ID,
QueueLength: tps.QueueLength,
Metrics: []string{"*asr", "*acd", "*acc"},
Thresholds: []string{"THRESH1", "THRESH2"},

7
engine/storage_interface.go
View File

@@ -111,9 +111,10 @@ type DataDB interface {
GetReqFilterIndexes(dbKey string) (indexes map[string]map[string]utils.StringMap, err error)
SetReqFilterIndexes(dbKey string, indexes map[string]map[string]utils.StringMap) (err error)
MatchReqFilterIndex(dbKey, fieldName, fieldVal string) (itemIDs utils.StringMap, err error)
GetStatsQueue(sqID string) (sq *StatsQueue, err error)
SetStatsQueue(sq *StatsQueue) (err error)
RemStatsQueue(sqID string) (err error)
//modicari
GetStatsConfig(sqID string) (sq *StatsConfig, err error)
SetStatsConfig(sq *StatsConfig) (err error)
RemStatsConfig(sqID string) (err error)
GetSQStoredMetrics(sqID string) (sqSM *SQStoredMetrics, err error)
SetSQStoredMetrics(sqSM *SQStoredMetrics) (err error)
RemSQStoredMetrics(sqID string) (err error)

18
engine/storage_map.go
View File

@@ -1475,19 +1475,19 @@ func (ms *MapStorage) RemoveVersions(vrs Versions) (err error) {
}
// GetStatsQueue retrieves a StatsQueue from dataDB
func (ms *MapStorage) GetStatsQueue(sqID string) (sq *StatsQueue, err error) {
func (ms *MapStorage) GetStatsConfig(sqID string) (scf *StatsConfig, err error) {
ms.mu.RLock()
defer ms.mu.RUnlock()
key := utils.StatsQueuePrefix + sqID
key := utils.StatsConfigPrefix + sqID
values, ok := ms.dict[key]
if !ok {
return nil, utils.ErrNotFound
}
err = ms.ms.Unmarshal(values, &sq)
err = ms.ms.Unmarshal(values, &scf)
if err != nil {
return nil, err
}
for _, fltr := range sq.Filters {
for _, fltr := range scf.Filters {
if err := fltr.CompileValues(); err != nil {
return nil, err
}
@@ -1496,22 +1496,22 @@ func (ms *MapStorage) GetStatsQueue(sqID string) (sq *StatsQueue, err error) {
}
// SetStatsQueue stores a StatsQueue into DataDB
func (ms *MapStorage) SetStatsQueue(sq *StatsQueue) (err error) {
func (ms *MapStorage) SetStatsConfig(scf *StatsConfig) (err error) {
ms.mu.Lock()
defer ms.mu.Unlock()
result, err := ms.ms.Marshal(sq)
result, err := ms.ms.Marshal(scf)
if err != nil {
return err
}
ms.dict[utils.StatsQueuePrefix+sq.ID] = result
ms.dict[utils.StatsConfigPrefix+scf.ID] = result
return
}
// RemStatsQueue removes a StatsQueue from dataDB
func (ms *MapStorage) RemStatsQueue(sqID string) (err error) {
func (ms *MapStorage) RemStatsConfig(scfID string) (err error) {
ms.mu.Lock()
defer ms.mu.Unlock()
key := utils.StatsQueuePrefix + sqID
key := utils.StatsConfigPrefix + scfID
delete(ms.dict, key)
return
}

16
engine/storage_mongo_datadb.go
View File

@@ -180,7 +180,7 @@ func (ms *MongoStorage) EnsureIndexes() (err error) {
Sparse: false,
}
for _, col := range []string{utils.TBLTPTimings, utils.TBLTPDestinations, utils.TBLTPDestinationRates, utils.TBLTPRatingPlans,
utils.TBLTPSharedGroups, utils.TBLTPCdrStats, utils.TBLTPActions, utils.TBLTPActionPlans, utils.TBLTPActionTriggers} {
utils.TBLTPSharedGroups, utils.TBLTPCdrStats, utils.TBLTPActions, utils.TBLTPActionPlans, utils.TBLTPActionTriggers, utils.TBLTPStats} {
if err = db.C(col).EnsureIndex(idx); err != nil {
return
}
@@ -1985,10 +1985,10 @@ func (ms *MongoStorage) MatchReqFilterIndex(dbKey, fldName, fldVal string) (item
}
// GetStatsQueue retrieves a StatsQueue from dataDB
func (ms *MongoStorage) GetStatsQueue(sqID string) (sq *StatsQueue, err error) {
session, col := ms.conn(utils.StatsQueuePrefix)
func (ms *MongoStorage) GetStatsConfig(sqID string) (sq *StatsConfig, err error) {
session, col := ms.conn(utils.StatsConfigPrefix)
defer session.Close()
sq = new(StatsQueue)
sq = new(StatsConfig)
if err = col.Find(bson.M{"id": sqID}).One(&sq); err != nil {
if err == mgo.ErrNotFound {
err = utils.ErrNotFound
@@ -2004,16 +2004,16 @@ func (ms *MongoStorage) GetStatsQueue(sqID string) (sq *StatsQueue, err error) {
}
// SetStatsQueue stores a StatsQueue into DataDB
func (ms *MongoStorage) SetStatsQueue(sq *StatsQueue) (err error) {
session, col := ms.conn(utils.StatsQueuePrefix)
func (ms *MongoStorage) SetStatsConfig(sq *StatsConfig) (err error) {
session, col := ms.conn(utils.StatsConfigPrefix)
defer session.Close()
_, err = col.UpsertId(bson.M{"id": sq.ID}, sq)
return
}
// RemStatsQueue removes a StatsQueue from dataDB
func (ms *MongoStorage) RemStatsQueue(sqID string) (err error) {
session, col := ms.conn(utils.StatsQueuePrefix)
func (ms *MongoStorage) RemStatsConfig(sqID string) (err error) {
session, col := ms.conn(utils.StatsConfigPrefix)
err = col.Remove(bson.M{"id": sqID})
if err != nil {
return err

33
engine/storage_mongo_stordb.go
View File

@@ -379,6 +379,23 @@ func (ms *MongoStorage) GetTPResourceLimits(tpid, id string) ([]*utils.TPResourc
return results, err
}
func (ms *MongoStorage) GetTPStats(tpid, id string) ([]*utils.TPStats, error) {
filter := bson.M{
"tpid": tpid,
}
if id != "" {
filter["id"] = id
}
var results []*utils.TPStats
session, col := ms.conn(utils.TBLTPStats)
defer session.Close()
err := col.Find(filter).All(&results)
if len(results) == 0 {
return results, utils.ErrNotFound
}
return results, err
}
func (ms *MongoStorage) GetTPDerivedChargers(tp *utils.TPDerivedChargers) ([]*utils.TPDerivedChargers, error) {
filter := bson.M{"tpid": tp.TPid}
if tp.Direction != "" {
@@ -841,6 +858,20 @@ func (ms *MongoStorage) SetTPResourceLimits(tpRLs []*utils.TPResourceLimit) (err
return
}
func (ms *MongoStorage) SetTPRStats(tpS []*utils.TPStats) (err error) {
if len(tpS) == 0 {
return
}
session, col := ms.conn(utils.TBLTPStats)
defer session.Close()
tx := col.Bulk()
for _, tp := range tpS {
tx.Upsert(bson.M{"tpid": tp.TPid, "id": tp.ID}, tp)
}
_, err = tx.Run()
return
}
func (ms *MongoStorage) SetSMCost(smc *SMCost) error {
if smc.CostDetails == nil {
return nil
@@ -1105,7 +1136,7 @@ func (ms *MongoStorage) GetCDRs(qryFltr *utils.CDRsFilter, remove bool) ([]*CDR,
return cdrs, 0, nil
}
func (ms *MongoStorage) GetTPStats(tpid, id string) ([]*utils.TPStats, error) {
func (ms *MongoStorage) GetTPStat(tpid, id string) ([]*utils.TPStats, error) {
filter := bson.M{
"tpid": tpid,
}

12
engine/storage_redis.go
View File

@@ -1542,8 +1542,8 @@ func (rs *RedisStorage) RemoveVersions(vrs Versions) (err error) {
}
// GetStatsQueue retrieves a StatsQueue from dataDB
func (rs *RedisStorage) GetStatsQueue(sqID string) (sq *StatsQueue, err error) {
key := utils.StatsQueuePrefix + sqID
func (rs *RedisStorage) GetStatsConfig(sqID string) (sq *StatsConfig, err error) {
key := utils.StatsConfigPrefix + sqID
var values []byte
if values, err = rs.Cmd("GET", key).Bytes(); err != nil {
if err == redis.ErrRespNil {
@@ -1563,18 +1563,18 @@ func (rs *RedisStorage) GetStatsQueue(sqID string) (sq *StatsQueue, err error) {
}
// SetStatsQueue stores a StatsQueue into DataDB
func (rs *RedisStorage) SetStatsQueue(sq *StatsQueue) (err error) {
func (rs *RedisStorage) SetStatsConfig(sq *StatsConfig) (err error) {
var result []byte
result, err = rs.ms.Marshal(sq)
if err != nil {
return
}
return rs.Cmd("SET", utils.StatsQueuePrefix+sq.ID, result).Err
return rs.Cmd("SET", utils.StatsConfigPrefix+sq.ID, result).Err
}
// RemStatsQueue removes a StatsQueue from dataDB
func (rs *RedisStorage) RemStatsQueue(sqID string) (err error) {
key := utils.StatsQueuePrefix + sqID
func (rs *RedisStorage) RemStatsConfig(sqID string) (err error) {
key := utils.StatsConfigPrefix + sqID
err = rs.Cmd("DEL", key).Err
return
}

10
engine/storage_sql.go
View File

@@ -190,9 +190,7 @@ func (self *SQLStorage) RemTpData(table, tpid string, args map[string]string) er
if len(table) == 0 { // Remove tpid out of all tables
for _, tblName := range []string{utils.TBLTPTimings, utils.TBLTPDestinations, utils.TBLTPRates, utils.TBLTPDestinationRates, utils.TBLTPRatingPlans, utils.TBLTPRateProfiles,
utils.TBLTPSharedGroups, utils.TBLTPCdrStats, utils.TBLTPLcrs, utils.TBLTPActions, utils.TBLTPActionPlans, utils.TBLTPActionTriggers, utils.TBLTPAccountActions,
utils.TBLTPDerivedChargers, utils.TBLTPAliases, utils.TBLTPUsers, utils.TBLTPResourceLimits,
// utils.TBLTPStats
} {
utils.TBLTPDerivedChargers, utils.TBLTPAliases, utils.TBLTPUsers, utils.TBLTPResourceLimits, utils.TBLTPStats} {
if err := tx.Table(tblName).Where("tpid = ?", tpid).Delete(nil).Error; err != nil {
tx.Rollback()
return err
@@ -201,6 +199,7 @@ func (self *SQLStorage) RemTpData(table, tpid string, args map[string]string) er
tx.Commit()
return nil
}
utils.Logger.Debug(fmt.Sprintf("#Rem sterge %s", tpid))
// Remove from a single table
tx = tx.Table(table).Where("tpid = ?", tpid)
// Compose filters
@@ -219,8 +218,10 @@ func (self *SQLStorage) SetTPTimings(timings []*utils.ApierTPTiming) error {
if len(timings) == 0 {
return nil
}
tx := self.db.Begin()
for _, timing := range timings {
utils.Logger.Debug(fmt.Sprintf("#1(set) Id care trimite %s", timing.ID))
if err := tx.Where(&TpTiming{Tpid: timing.TPid, Tag: timing.ID}).Delete(TpTiming{}).Error; err != nil {
tx.Rollback()
return err
@@ -1170,6 +1171,8 @@ func (self *SQLStorage) GetTPDestinationRates(tpid, id string, pagination *utils
func (self *SQLStorage) GetTPTimings(tpid, id string) ([]*utils.ApierTPTiming, error) {
var tpTimings TpTimings
q := self.db.Where("tpid = ?", tpid)
utils.Logger.Debug(fmt.Sprintf("#1 Id care trimite %s", id))
utils.Logger.Debug(fmt.Sprintf("#1 TPId care trimite %s", tpid))
if len(id) != 0 {
q = q.Where("tag = ?", id)
}
@@ -1177,6 +1180,7 @@ func (self *SQLStorage) GetTPTimings(tpid, id string) ([]*utils.ApierTPTiming, e
return nil, err
}
ts := tpTimings.AsTPTimings()
utils.Logger.Debug(fmt.Sprintf("#2 ce gaseste : %s", ts))
if len(ts) == 0 {
return ts, utils.ErrNotFound
}

2
engine/tp_reader.go
View File

@@ -1955,7 +1955,7 @@ func (tpr *TpReader) WriteToDatabase(flush, verbose, disable_reverse bool) (err
if err != nil {
return err
}
if err = tpr.dataStorage.SetStatsQueue(st); err != nil {
if err = tpr.dataStorage.SetStatsConfig(st); err != nil {
return err
}
if verbose {

4
stats/queue.go
View File

@@ -49,7 +49,7 @@ func (sis StatsInstances) remWithID(qID string) {
// NewStatsInstance instantiates a StatsInstance
func NewStatsInstance(sec *StatsEventCache, ms engine.Marshaler,
sqCfg *engine.StatsQueue, sqSM *engine.SQStoredMetrics) (si *StatsInstance, err error) {
sqCfg *engine.StatsConfig, sqSM *engine.SQStoredMetrics) (si *StatsInstance, err error) {
si = &StatsInstance{sec: sec, ms: ms, cfg: sqCfg, sqMetrics: make(map[string]StatsMetric)}
for _, metricID := range sqCfg.Metrics {
if si.sqMetrics[metricID], err = NewStatsMetric(metricID); err != nil {
@@ -85,7 +85,7 @@ type StatsInstance struct {
sqItems []*engine.SQItem
sqMetrics map[string]StatsMetric
ms engine.Marshaler // used to get/set Metrics
cfg *engine.StatsQueue
cfg *engine.StatsConfig
}
// GetSQStoredMetrics retrieves the data used for store to DB

16
stats/queue_test.go
View File

@@ -26,17 +26,17 @@ import (
func TestStatsInstancesSort(t *testing.T) {
sInsts := StatsInstances{
&StatsInstance{cfg: &engine.StatsQueue{ID: "FIRST", Weight: 30.0}},
&StatsInstance{cfg: &engine.StatsQueue{ID: "SECOND", Weight: 40.0}},
&StatsInstance{cfg: &engine.StatsQueue{ID: "THIRD", Weight: 30.0}},
&StatsInstance{cfg: &engine.StatsQueue{ID: "FOURTH", Weight: 35.0}},
&StatsInstance{cfg: &engine.StatsConfig{ID: "FIRST", Weight: 30.0}},
&StatsInstance{cfg: &engine.StatsConfig{ID: "SECOND", Weight: 40.0}},
&StatsInstance{cfg: &engine.StatsConfig{ID: "THIRD", Weight: 30.0}},
&StatsInstance{cfg: &engine.StatsConfig{ID: "FOURTH", Weight: 35.0}},
}
sInsts.Sort()
eSInst := StatsInstances{
&StatsInstance{cfg: &engine.StatsQueue{ID: "SECOND", Weight: 40.0}},
&StatsInstance{cfg: &engine.StatsQueue{ID: "FOURTH", Weight: 35.0}},
&StatsInstance{cfg: &engine.StatsQueue{ID: "FIRST", Weight: 30.0}},
&StatsInstance{cfg: &engine.StatsQueue{ID: "THIRD", Weight: 30.0}},
&StatsInstance{cfg: &engine.StatsConfig{ID: "SECOND", Weight: 40.0}},
&StatsInstance{cfg: &engine.StatsConfig{ID: "FOURTH", Weight: 35.0}},
&StatsInstance{cfg: &engine.StatsConfig{ID: "FIRST", Weight: 30.0}},
&StatsInstance{cfg: &engine.StatsConfig{ID: "THIRD", Weight: 30.0}},
}
if !reflect.DeepEqual(eSInst, sInsts) {
t.Errorf("expecting: %+v, received: %+v", eSInst, sInsts)

10
stats/service.go
View File

@@ -39,14 +39,14 @@ func init() {
func NewStatService(dataDB engine.DataDB, ms engine.Marshaler, storeInterval time.Duration) (ss *StatService, err error) {
ss = &StatService{dataDB: dataDB, ms: ms, storeInterval: storeInterval,
stopStoring: make(chan struct{}), evCache: NewStatsEventCache()}
sqPrfxs, err := dataDB.GetKeysForPrefix(utils.StatsQueuePrefix)
sqPrfxs, err := dataDB.GetKeysForPrefix(utils.StatsConfigPrefix)
if err != nil {
return nil, err
}
ss.queuesCache = make(map[string]*StatsInstance)
ss.queues = make(StatsInstances, 0)
for _, prfx := range sqPrfxs {
if q, err := ss.loadQueue(prfx[len(utils.StatsQueuePrefix):]); err != nil {
if q, err := ss.loadQueue(prfx[len(utils.StatsConfigPrefix):]); err != nil {
utils.Logger.Err(fmt.Sprintf("<StatS> failed loading quueue with id: <%s>, err: <%s>",
q.cfg.ID, err.Error()))
continue
@@ -92,7 +92,7 @@ func (ss *StatService) Shutdown() error {
// setQueue adds or modifies a queue into cache
// sort will reorder the ss.queues
func (ss *StatService) loadQueue(qID string) (q *StatsInstance, err error) {
sq, err := ss.dataDB.GetStatsQueue(qID)
sq, err := ss.dataDB.GetStatsConfig(qID)
if err != nil {
return nil, err
}
@@ -225,13 +225,13 @@ type ArgsLoadQueues struct {
func (ss *StatService) V1LoadQueues(args ArgsLoadQueues, reply *string) (err error) {
qIDs := args.QueueIDs
if qIDs == nil {
sqPrfxs, err := ss.dataDB.GetKeysForPrefix(utils.StatsQueuePrefix)
sqPrfxs, err := ss.dataDB.GetKeysForPrefix(utils.StatsConfigPrefix)
if err != nil {
return err
}
queueIDs := make([]string, len(sqPrfxs))
for i, prfx := range sqPrfxs {
queueIDs[i] = prfx[len(utils.StatsQueuePrefix):]
queueIDs[i] = prfx[len(utils.StatsConfigPrefix):]
}
if len(queueIDs) != 0 {
qIDs = &queueIDs

4
stats/service_test.go
View File

@@ -32,8 +32,8 @@ func TestReqFilterPassStatS(t *testing.T) {
config.SetCgrConfig(cgrCfg)
}
dataStorage, _ := engine.NewMapStorage()
dataStorage.SetStatsQueue(
&engine.StatsQueue{ID: "CDRST1",
dataStorage.SetStatsConfig(
&engine.StatsConfig{ID: "CDRST1",
Filters: []*engine.RequestFilter{
&engine.RequestFilter{Type: engine.MetaString, FieldName: "Tenant",
Values: []string{"cgrates.org"}}},

2
utils/consts.go
View File

@@ -251,7 +251,7 @@ const (
LOG_CDR = "cdr_"
LOG_MEDIATED_CDR = "mcd_"
SQStoredMetricsPrefix = "ssm_"
StatsQueuePrefix = "stq_"
StatsConfigPrefix = "scf_"
ThresholdCfgPrefix = "thc_"
LOADINST_KEY = "load_history"
SESSION_MANAGER_SOURCE = "SMR"