uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-14 12:49:54 +05:00
Code Issues Packages Projects Releases Wiki Activity

Solved comments from last pull request and added integration test for computeStatIndexes

Browse Source
This commit is contained in:
edwardro22
2018-01-03 18:52:10 +02:00
committed by Dan Christian Bogos
parent 9ad0e2cefe
commit f3bbcc5dbb
8 changed files with 625 additions and 280 deletions
Download Patch File Download Diff File Expand all files Collapse all files

476
apier/v1/filter_indexes.go
View File

@@ -37,269 +37,143 @@ type ArgsComputeFilterIndexes struct {
*/
func (self *ApierV1) ComputeFilterIndexes(args utils.ArgsComputeFilterIndexes, reply *string) error {
ifnil := []string{}
//ThresholdProfile Indexes
var thresholdIDs []string
thdsIndexers := engine.NewReqFilterIndexer(self.DataManager, utils.ThresholdProfilePrefix, args.Tenant)
if args.ThresholdIDs == nil {
if args.ThresholdIDs == nil || len(*args.ThresholdIDs) != 0 {
if args.ThresholdIDs == nil {
args.ThresholdIDs = &ifnil
}
if err := self.computeThresholdIndexes(args.Tenant, *args.ThresholdIDs); err != nil {
return utils.APIErrorHandler(err)
}
}
//StatQueueProfile Indexes
if args.StatIDs == nil || len(*args.StatIDs) != 0 {
if args.StatIDs == nil {
args.StatIDs = &ifnil
}
if err := self.computeStatIndexes(args.Tenant, *args.StatIDs); err != nil {
return utils.APIErrorHandler(err)
}
}
//ResourceProfile Indexes
if args.ResourceIDs == nil || len(*args.ResourceIDs) != 0 {
if args.ResourceIDs == nil {
args.ResourceIDs = &ifnil
}
if err := self.computeResourceIndexes(args.Tenant, *args.ResourceIDs); err != nil {
return utils.APIErrorHandler(err)
}
}
//SupplierProfile Indexes
if args.SupplierIDs == nil || len(*args.SupplierIDs) != 0 {
if args.SupplierIDs == nil {
args.SupplierIDs = &ifnil
}
if err := self.computeSupplierIndexes(args.Tenant, *args.SupplierIDs); err != nil {
return utils.APIErrorHandler(err)
}
}
//AttributeProfile Indexes
if args.AttributeIDs == nil || len(*args.AttributeIDs) != 0 {
if args.AttributeIDs == nil {
args.AttributeIDs = &ifnil
}
if err := self.computeAttributeIndexes(args.Tenant, *args.AttributeIDs); err != nil {
return utils.APIErrorHandler(err)
}
}
*reply = utils.OK
return nil
}
func (self *ApierV1) computeThresholdIndexes(tenant string, thresholdIDs []string) error {
var zeroIDS bool
thdsIndexers := engine.NewReqFilterIndexer(self.DataManager, utils.ThresholdProfilePrefix, tenant)
if len(thresholdIDs) == 0 {
zeroIDS = true
ids, err := self.DataManager.DataDB().GetKeysForPrefix(utils.ThresholdProfilePrefix)
if err != nil {
return utils.APIErrorHandler(err)
return err
}
for _, id := range ids {
thresholdIDs = append(thresholdIDs, strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
}
} else {
thresholdIDs = *args.ThresholdIDs
}
for _, id := range thresholdIDs {
th, err := self.DataManager.GetThresholdProfile(args.Tenant, id, false, utils.NonTransactional)
th, err := self.DataManager.GetThresholdProfile(tenant, id, false, utils.NonTransactional)
if err != nil {
return utils.APIErrorHandler(err)
return err
}
for _, fltrID := range th.FilterIDs {
fltr, err := self.DataManager.GetFilter(args.Tenant, fltrID, false, utils.NonTransactional)
fltr, err := self.DataManager.GetFilter(tenant, fltrID, false, utils.NonTransactional)
if err != nil {
if err == utils.ErrNotFound {
err = fmt.Errorf("broken reference to filter: %+v for threshold: %+v", fltrID, th)
}
return utils.APIErrorHandler(err)
return err
} else {
tpFltr := engine.FilterToTPFilter(fltr)
thdsIndexers.IndexTPFilter(tpFltr, th.ID)
thdsIndexers.IndexTPFilter(engine.FilterToTPFilter(fltr), th.ID)
}
}
}
if args.ThresholdIDs == nil {
if zeroIDS {
if err := self.DataManager.RemoveFilterIndexes(engine.GetDBIndexKey(utils.ThresholdProfilePrefix,
args.Tenant, false)); err != nil {
tenant, false)); err != nil {
if err != utils.ErrNotFound {
return utils.APIErrorHandler(err)
return err
}
}
if err := self.DataManager.RemoveFilterReverseIndexes(engine.GetDBIndexKey(utils.ThresholdProfilePrefix,
args.Tenant, true), ""); err != nil {
tenant, true), ""); err != nil {
if err != utils.ErrNotFound {
return utils.APIErrorHandler(err)
return err
}
}
} else {
indexRemover := engine.NewReqFilterIndexer(self.DataManager, utils.ThresholdProfilePrefix, args.Tenant)
indexRemover := engine.NewReqFilterIndexer(self.DataManager, utils.ThresholdProfilePrefix, tenant)
for _, id := range thresholdIDs {
if err := indexRemover.RemoveItemFromIndex(id); err != nil &&
err.Error() != utils.ErrNotFound.Error() {
return utils.APIErrorHandler(err)
return err
}
}
}
if err := thdsIndexers.StoreIndexes(); err != nil {
return utils.APIErrorHandler(err)
}
//StatQueueProfile Indexes
var statIDs []string
sqpIndexers := engine.NewReqFilterIndexer(self.DataManager, utils.StatQueueProfilePrefix, args.Tenant)
if args.StatIDs == nil {
ids, err := self.DataManager.DataDB().GetKeysForPrefix(utils.StatQueueProfilePrefix)
if err != nil {
return utils.APIErrorHandler(err)
}
for _, id := range ids {
err = sqpIndexers.RemoveItemFromIndex(strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
if err != nil && err.Error() != utils.ErrNotFound.Error() {
return utils.APIErrorHandler(err)
}
statIDs = append(statIDs, strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
}
} else {
statIDs = *args.StatIDs
}
for _, id := range statIDs {
sqp, err := self.DataManager.GetStatQueueProfile(args.Tenant, id, false, utils.NonTransactional)
if err != nil {
return utils.APIErrorHandler(err)
}
for _, fltrID := range sqp.FilterIDs {
fltr, err := self.DataManager.GetFilter(args.Tenant, fltrID, false, utils.NonTransactional)
if err != nil {
if err == utils.ErrNotFound {
err = fmt.Errorf("broken reference to filter: %+v for stats queue: %+v", fltrID, sqp)
}
return utils.APIErrorHandler(err)
} else {
tpFltr := engine.FilterToTPFilter(fltr)
sqpIndexers.IndexTPFilter(tpFltr, sqp.ID)
}
}
}
if args.StatIDs == nil {
if err := self.DataManager.RemoveFilterIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix,
args.Tenant, false)); err != nil {
return utils.APIErrorHandler(err)
}
if err := self.DataManager.RemoveFilterReverseIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix,
args.Tenant, true), ""); err != nil {
return utils.APIErrorHandler(err)
}
} else {
indexRemover := engine.NewReqFilterIndexer(self.DataManager, utils.StatQueueProfilePrefix, args.Tenant)
for _, id := range statIDs {
if err := indexRemover.RemoveItemFromIndex(id); err != nil &&
err.Error() != utils.ErrNotFound.Error() {
return utils.APIErrorHandler(err)
}
}
}
if err := sqpIndexers.StoreIndexes(); err != nil {
return err
}
//ResourceProfile Indexes
var resourceIDs []string
rpIndexers := engine.NewReqFilterIndexer(self.DataManager, utils.ResourceProfilesPrefix, args.Tenant)
if args.ResourceIDs == nil {
ids, err := self.DataManager.DataDB().GetKeysForPrefix(utils.ResourceProfilesPrefix)
if err != nil {
return utils.APIErrorHandler(err)
}
for _, id := range ids {
err = rpIndexers.RemoveItemFromIndex(strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
if err != nil && err.Error() != utils.ErrNotFound.Error() {
return utils.APIErrorHandler(err)
}
resourceIDs = append(resourceIDs, strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
}
} else {
resourceIDs = *args.ResourceIDs
}
for _, id := range resourceIDs {
rp, err := self.DataManager.GetResourceProfile(args.Tenant, id, false, utils.NonTransactional)
if err != nil {
return utils.APIErrorHandler(err)
}
for _, fltrID := range rp.FilterIDs {
fltr, err := self.DataManager.GetFilter(args.Tenant, fltrID, false, utils.NonTransactional)
if err != nil {
if err == utils.ErrNotFound {
err = fmt.Errorf("broken reference to filter: %+v for stats queue: %+v", fltrID, rp)
}
return utils.APIErrorHandler(err)
} else {
tpFltr := engine.FilterToTPFilter(fltr)
rpIndexers.IndexTPFilter(tpFltr, rp.ID)
return nil
}
}
}
}
if args.ResourceIDs == nil {
if err := self.DataManager.RemoveFilterIndexes(engine.GetDBIndexKey(utils.ResourceProfilesPrefix,
args.Tenant, false)); err != nil {
return utils.APIErrorHandler(err)
}
if err := self.DataManager.RemoveFilterReverseIndexes(engine.GetDBIndexKey(utils.ResourceProfilesPrefix,
args.Tenant, true), ""); err != nil {
return utils.APIErrorHandler(err)
}
} else {
indexRemover := engine.NewReqFilterIndexer(self.DataManager, utils.ResourceProfilesPrefix, args.Tenant)
for _, id := range resourceIDs {
if err := indexRemover.RemoveItemFromIndex(id); err != nil &&
err.Error() != utils.ErrNotFound.Error() {
return utils.APIErrorHandler(err)
}
}
}
if err := rpIndexers.StoreIndexes(); err != nil {
return err
}
//SupplierProfile Indexes
var supplierIDs []string
sppIndexers := engine.NewReqFilterIndexer(self.DataManager, utils.SupplierProfilePrefix, args.Tenant)
if args.SupplierIDs == nil {
ids, err := self.DataManager.DataDB().GetKeysForPrefix(utils.SupplierProfilePrefix)
if err != nil {
return utils.APIErrorHandler(err)
}
for _, id := range ids {
err = sppIndexers.RemoveItemFromIndex(strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
if err != nil && err.Error() != utils.ErrNotFound.Error() {
return utils.APIErrorHandler(err)
}
supplierIDs = append(supplierIDs, strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
}
} else {
supplierIDs = *args.SupplierIDs
}
for _, id := range supplierIDs {
spp, err := self.DataManager.GetSupplierProfile(args.Tenant, id, false, utils.NonTransactional)
if err != nil {
return utils.APIErrorHandler(err)
}
for _, fltrID := range spp.FilterIDs {
fltr, err := self.DataManager.GetFilter(args.Tenant, fltrID, false, utils.NonTransactional)
if err != nil {
if err == utils.ErrNotFound {
err = fmt.Errorf("broken reference to filter: %+v for stats queue: %+v", fltrID, spp)
}
return utils.APIErrorHandler(err)
} else {
tpFltr := engine.FilterToTPFilter(fltr)
sppIndexers.IndexTPFilter(tpFltr, spp.ID)
}
}
}
if args.SupplierIDs == nil {
if err := self.DataManager.RemoveFilterIndexes(engine.GetDBIndexKey(utils.SupplierProfilePrefix,
args.Tenant, false)); err != nil {
return utils.APIErrorHandler(err)
}
if err := self.DataManager.RemoveFilterReverseIndexes(engine.GetDBIndexKey(utils.SupplierProfilePrefix,
args.Tenant, true), ""); err != nil {
return utils.APIErrorHandler(err)
}
} else {
indexRemover := engine.NewReqFilterIndexer(self.DataManager, utils.SupplierProfilePrefix, args.Tenant)
for _, id := range resourceIDs {
if err := indexRemover.RemoveItemFromIndex(id); err != nil &&
err.Error() != utils.ErrNotFound.Error() {
return utils.APIErrorHandler(err)
}
}
}
if err := sppIndexers.StoreIndexes(); err != nil {
return err
}
//AttributeProfile Indexes
var attributeIDs []string
attrIndexers := engine.NewReqFilterIndexer(self.DataManager, utils.AttributeProfilePrefix, args.Tenant)
if args.AttributeIDs == nil {
func (self *ApierV1) computeAttributeIndexes(tenant string, attributeIDs []string) error {
var zeroIDS bool
attrIndexers := engine.NewReqFilterIndexer(self.DataManager, utils.AttributeProfilePrefix, tenant)
if len(attributeIDs) == 0 {
zeroIDS = true
ids, err := self.DataManager.DataDB().GetKeysForPrefix(utils.AttributeProfilePrefix)
if err != nil {
return utils.APIErrorHandler(err)
return err
}
for _, id := range ids {
err = attrIndexers.RemoveItemFromIndex(strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
if err != nil && err.Error() != utils.ErrNotFound.Error() {
return utils.APIErrorHandler(err)
return err
}
attributeIDs = append(attributeIDs, strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
}
} else {
attributeIDs = *args.AttributeIDs
}
for _, id := range attributeIDs {
ap, err := self.DataManager.GetAttributeProfile(args.Tenant, id, false, utils.NonTransactional)
ap, err := self.DataManager.GetAttributeProfile(tenant, id, false, utils.NonTransactional)
if err != nil {
return utils.APIErrorHandler(err)
return err
}
for _, fltrID := range ap.FilterIDs {
fltr, err := self.DataManager.GetFilter(args.Tenant, fltrID, false, utils.NonTransactional)
fltr, err := self.DataManager.GetFilter(tenant, fltrID, false, utils.NonTransactional)
if err != nil {
if err == utils.ErrNotFound {
err = fmt.Errorf("broken reference to filter: %+v for stats queue: %+v", fltrID, ap)
}
return utils.APIErrorHandler(err)
return err
} else {
tpFltr := engine.FilterToTPFilter(fltr)
attrIndexers.IndexTPFilter(tpFltr, ap.ID)
@@ -307,28 +181,202 @@ func (self *ApierV1) ComputeFilterIndexes(args utils.ArgsComputeFilterIndexes, r
}
}
}
if args.AttributeIDs == nil {
if zeroIDS {
if err := self.DataManager.RemoveFilterIndexes(engine.GetDBIndexKey(utils.AttributeProfilePrefix,
args.Tenant, false)); err != nil {
return utils.APIErrorHandler(err)
tenant, false)); err != nil {
return err
}
if err := self.DataManager.RemoveFilterReverseIndexes(engine.GetDBIndexKey(utils.AttributeProfilePrefix,
args.Tenant, true), ""); err != nil {
return utils.APIErrorHandler(err)
tenant, true), ""); err != nil {
return err
}
} else {
indexRemover := engine.NewReqFilterIndexer(self.DataManager, utils.AttributeProfilePrefix, args.Tenant)
indexRemover := engine.NewReqFilterIndexer(self.DataManager, utils.AttributeProfilePrefix, tenant)
for _, id := range attributeIDs {
if err := indexRemover.RemoveItemFromIndex(id); err != nil &&
err.Error() != utils.ErrNotFound.Error() {
return utils.APIErrorHandler(err)
return err
}
}
}
if err := attrIndexers.StoreIndexes(); err != nil {
return err
}
*reply = utils.OK
return nil
}
func (self *ApierV1) computeResourceIndexes(tenant string, resourceIDs []string) error {
var zeroIDS bool
rpIndexers := engine.NewReqFilterIndexer(self.DataManager, utils.ResourceProfilesPrefix, tenant)
if len(resourceIDs) == 0 {
zeroIDS = true
ids, err := self.DataManager.DataDB().GetKeysForPrefix(utils.ResourceProfilesPrefix)
if err != nil {
return err
}
for _, id := range ids {
err = rpIndexers.RemoveItemFromIndex(strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
if err != nil && err.Error() != utils.ErrNotFound.Error() {
return err
}
resourceIDs = append(resourceIDs, strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
}
}
for _, id := range resourceIDs {
rp, err := self.DataManager.GetResourceProfile(tenant, id, false, utils.NonTransactional)
if err != nil {
return err
}
for _, fltrID := range rp.FilterIDs {
fltr, err := self.DataManager.GetFilter(tenant, fltrID, false, utils.NonTransactional)
if err != nil {
if err == utils.ErrNotFound {
err = fmt.Errorf("broken reference to filter: %+v for stats queue: %+v", fltrID, rp)
}
return err
} else {
rpIndexers.IndexTPFilter(engine.FilterToTPFilter(fltr), rp.ID)
}
}
}
if zeroIDS {
if err := self.DataManager.RemoveFilterIndexes(engine.GetDBIndexKey(utils.ResourceProfilesPrefix,
tenant, false)); err != nil {
return err
}
if err := self.DataManager.RemoveFilterReverseIndexes(engine.GetDBIndexKey(utils.ResourceProfilesPrefix,
tenant, true), ""); err != nil {
return err
}
} else {
indexRemover := engine.NewReqFilterIndexer(self.DataManager, utils.ResourceProfilesPrefix, tenant)
for _, id := range resourceIDs {
if err := indexRemover.RemoveItemFromIndex(id); err != nil &&
err.Error() != utils.ErrNotFound.Error() {
return err
}
}
}
if err := rpIndexers.StoreIndexes(); err != nil {
return err
}
return nil
}
func (self *ApierV1) computeStatIndexes(tenant string, statIDs []string) error {
var zeroIDS bool
sqpIndexers := engine.NewReqFilterIndexer(self.DataManager, utils.StatQueueProfilePrefix, tenant)
if len(statIDs) == 0 {
zeroIDS = true
ids, err := self.DataManager.DataDB().GetKeysForPrefix(utils.StatQueueProfilePrefix)
if err != nil {
return err
}
for _, id := range ids {
err = sqpIndexers.RemoveItemFromIndex(strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
if err != nil && err.Error() != utils.ErrNotFound.Error() {
return err
}
statIDs = append(statIDs, strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
}
}
for _, id := range statIDs {
sqp, err := self.DataManager.GetStatQueueProfile(tenant, id, false, utils.NonTransactional)
if err != nil {
return err
}
for _, fltrID := range sqp.FilterIDs {
fltr, err := self.DataManager.GetFilter(tenant, fltrID, false, utils.NonTransactional)
if err != nil {
if err == utils.ErrNotFound {
err = fmt.Errorf("broken reference to filter: %+v for stats queue: %+v", fltrID, sqp)
}
return err
} else {
sqpIndexers.IndexTPFilter(engine.FilterToTPFilter(fltr), sqp.ID)
}
}
}
if zeroIDS {
if err := self.DataManager.RemoveFilterIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix,
tenant, false)); err != nil {
return err
}
if err := self.DataManager.RemoveFilterReverseIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix,
tenant, true), ""); err != nil {
return err
}
} else {
indexRemover := engine.NewReqFilterIndexer(self.DataManager, utils.StatQueueProfilePrefix, tenant)
for _, id := range statIDs {
if err := indexRemover.RemoveItemFromIndex(id); err != nil &&
err.Error() != utils.ErrNotFound.Error() {
return err
}
}
}
if err := sqpIndexers.StoreIndexes(); err != nil {
return err
}
return nil
}
func (self *ApierV1) computeSupplierIndexes(tenant string, supplierIDs []string) error {
var zeroIDS bool
sppIndexers := engine.NewReqFilterIndexer(self.DataManager, utils.SupplierProfilePrefix, tenant)
if len(supplierIDs) == 0 {
zeroIDS = true
ids, err := self.DataManager.DataDB().GetKeysForPrefix(utils.SupplierProfilePrefix)
if err != nil {
return err
}
for _, id := range ids {
err = sppIndexers.RemoveItemFromIndex(strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
if err != nil && err.Error() != utils.ErrNotFound.Error() {
return err
}
supplierIDs = append(supplierIDs, strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
}
}
for _, id := range supplierIDs {
spp, err := self.DataManager.GetSupplierProfile(tenant, id, false, utils.NonTransactional)
if err != nil {
return err
}
for _, fltrID := range spp.FilterIDs {
fltr, err := self.DataManager.GetFilter(tenant, fltrID, false, utils.NonTransactional)
if err != nil {
if err == utils.ErrNotFound {
err = fmt.Errorf("broken reference to filter: %+v for stats queue: %+v", fltrID, spp)
}
return err
} else {
tpFltr := engine.FilterToTPFilter(fltr)
sppIndexers.IndexTPFilter(tpFltr, spp.ID)
}
}
}
if zeroIDS {
if err := self.DataManager.RemoveFilterIndexes(engine.GetDBIndexKey(utils.SupplierProfilePrefix,
tenant, false)); err != nil {
return err
}
if err := self.DataManager.RemoveFilterReverseIndexes(engine.GetDBIndexKey(utils.SupplierProfilePrefix,
tenant, true), ""); err != nil {
return err
}
} else {
indexRemover := engine.NewReqFilterIndexer(self.DataManager, utils.SupplierProfilePrefix, tenant)
for _, id := range supplierIDs {
if err := indexRemover.RemoveItemFromIndex(id); err != nil &&
err.Error() != utils.ErrNotFound.Error() {
return err
}
}
}
if err := sppIndexers.StoreIndexes(); err != nil {
return err
}
return nil
}

295
apier/v1/filter_indexes_it_test.go
View File

@@ -20,8 +20,8 @@ along with this program. If not, see <http://www.gnu.org/licenses/>
package v1
import (
// "net/rpc"
"fmt"
"net/rpc"
"net/rpc/jsonrpc"
"path"
"reflect"
@@ -35,11 +35,12 @@ import (
)
var (
rdsITdb *engine.RedisStorage
mgoITdb *engine.MongoStorage
onStor *engine.DataManager
err error
indexes map[string]utils.StringMap
tFIdxRpc *rpc.Client
rdsITdb *engine.RedisStorage
mgoITdb *engine.MongoStorage
onStor *engine.DataManager
err error
indexes map[string]utils.StringMap
)
var sTestsFilterIndexesSV1 = []func(t *testing.T){
@@ -53,6 +54,10 @@ var sTestsFilterIndexesSV1 = []func(t *testing.T){
testV1FIdxComputeThresholdsIndexes,
testV1FIdxSetSecondThresholdProfile,
testV1FIdxSecondComputeThresholdsIndexes,
testV1FIdxSetStatQueueProfileIndexes,
testV1FIdxComputeStatQueueProfileIndexes,
testV1FIdxSetSecondStatQueueProfileIndexes,
testV1FIdxSecondComputeStatQueueProfileIndexes,
//to add testV1TSGetThresholdsAfterRestart,
// testV1FIdxSetThresholdProfile,
// testV1FIdxUpdateThresholdProfile,
@@ -143,12 +148,13 @@ func testV1FIdxStartEngine(t *testing.T) {
func testV1FIdxRpcConn(t *testing.T) {
var err error
tSv1Rpc, err = jsonrpc.Dial("tcp", tSv1Cfg.RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
tFIdxRpc, err = jsonrpc.Dial("tcp", tSv1Cfg.RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
if err != nil {
t.Fatal("Could not connect to rater: ", err.Error())
}
}
//ThresholdProfile
func testV1FIdxSetThresholdProfile(t *testing.T) {
tenant := "cgrates.org"
var reply *engine.ThresholdProfile
@@ -169,12 +175,12 @@ func testV1FIdxSetThresholdProfile(t *testing.T) {
}
var result string
if err := tSv1Rpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
if err := tFIdxRpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := tSv1Rpc.Call("ApierV1.GetThresholdProfile",
if err := tFIdxRpc.Call("ApierV1.GetThresholdProfile",
&utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE1"}, &reply); err == nil ||
err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
@@ -194,12 +200,12 @@ func testV1FIdxSetThresholdProfile(t *testing.T) {
ActionIDs: []string{"ACT_1", "ACT_2"},
Async: true,
}
if err := tSv1Rpc.Call("ApierV1.SetThresholdProfile", tPrfl, &result); err != nil {
if err := tFIdxRpc.Call("ApierV1.SetThresholdProfile", tPrfl, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := tSv1Rpc.Call("ApierV1.GetThresholdProfile",
if err := tFIdxRpc.Call("ApierV1.GetThresholdProfile",
&utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE1"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(tPrfl, reply) {
@@ -224,7 +230,7 @@ func testV1FIdxComputeThresholdsIndexes(t *testing.T) {
tenant := "cgrates.org"
emptySlice := []string{}
var reply2 string
if err := tSv1Rpc.Call(utils.ComputeFilterIndexes, utils.ArgsComputeFilterIndexes{
if err := tFIdxRpc.Call(utils.ApierV1ComputeFilterIndexes, utils.ArgsComputeFilterIndexes{
Tenant: "cgrates.org",
ThresholdIDs: nil,
AttributeIDs: &emptySlice,
@@ -275,12 +281,12 @@ func testV1FIdxSetSecondThresholdProfile(t *testing.T) {
}
var result string
if err := tSv1Rpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
if err := tFIdxRpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := tSv1Rpc.Call("ApierV1.GetThresholdProfile",
if err := tFIdxRpc.Call("ApierV1.GetThresholdProfile",
&utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE2"}, &reply); err == nil ||
err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
@@ -300,12 +306,12 @@ func testV1FIdxSetSecondThresholdProfile(t *testing.T) {
ActionIDs: []string{"ACT_1", "ACT_2"},
Async: true,
}
if err := tSv1Rpc.Call("ApierV1.SetThresholdProfile", tPrfl, &result); err != nil {
if err := tFIdxRpc.Call("ApierV1.SetThresholdProfile", tPrfl, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := tSv1Rpc.Call("ApierV1.GetThresholdProfile",
if err := tFIdxRpc.Call("ApierV1.GetThresholdProfile",
&utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE2"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(tPrfl, reply) {
@@ -330,7 +336,7 @@ func testV1FIdxSecondComputeThresholdsIndexes(t *testing.T) {
thid := []string{"TEST_PROFILE2"}
emptySlice := []string{}
var reply2 string
if err := tSv1Rpc.Call(utils.ComputeFilterIndexes, utils.ArgsComputeFilterIndexes{
if err := tFIdxRpc.Call(utils.ApierV1ComputeFilterIndexes, utils.ArgsComputeFilterIndexes{
Tenant: "cgrates.org",
ThresholdIDs: &thid,
AttributeIDs: &emptySlice,
@@ -361,6 +367,239 @@ func testV1FIdxSecondComputeThresholdsIndexes(t *testing.T) {
}
}
//StatQueueProfile
func testV1FIdxSetStatQueueProfileIndexes(t *testing.T) {
tenant := "cgrates.org"
var reply *engine.StatQueueProfile
filter = &engine.Filter{
Tenant: tenant,
ID: "FLTR_1",
RequestFilters: []*engine.RequestFilter{
&engine.RequestFilter{
FieldName: "Account",
Type: "*string",
Values: []string{"1001"},
},
},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
ExpiryTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
},
}
var result string
if err := tFIdxRpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := tFIdxRpc.Call("ApierV1.GetStatQueueProfile",
&utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE1"}, &reply); err == nil ||
err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
statConfig = &engine.StatQueueProfile{
Tenant: "cgrates.org",
ID: "TEST_PROFILE1",
FilterIDs: []string{"FLTR_1"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
ExpiryTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
},
QueueLength: 10,
TTL: time.Duration(10) * time.Second,
Metrics: []*utils.MetricWithParams{
&utils.MetricWithParams{
MetricID: "MetricValue",
Parameters: "",
},
&utils.MetricWithParams{
MetricID: "MetricValueTwo",
Parameters: "",
},
},
Thresholds: []string{"Val1", "Val2"},
Blocker: true,
Stored: true,
Weight: 20,
MinItems: 1,
}
if err := tFIdxRpc.Call("ApierV1.SetStatQueueProfile", statConfig, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := tFIdxRpc.Call("ApierV1.GetStatQueueProfile",
&utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE1"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(statConfig, reply) {
t.Errorf("Expecting: %+v, received: %+v", statConfig, reply)
}
if err = onStor.RemoveFilterIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix,
tenant, false)); err != nil {
t.Error(err)
}
if err := onStor.RemoveFilterReverseIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix,
tenant, true), ""); err != nil {
t.Error(err)
}
if indexes, err = onStor.GetFilterIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix, tenant, false),
nil); err != utils.ErrNotFound {
t.Error(err)
}
}
func testV1FIdxComputeStatQueueProfileIndexes(t *testing.T) {
tenant := "cgrates.org"
emptySlice := []string{}
var reply2 string
if err := tFIdxRpc.Call(utils.ApierV1ComputeFilterIndexes, utils.ArgsComputeFilterIndexes{
Tenant: "cgrates.org",
ThresholdIDs: &emptySlice,
AttributeIDs: &emptySlice,
ResourceIDs: &emptySlice,
StatIDs: nil,
SupplierIDs: &emptySlice,
}, &reply2); err != nil {
t.Error(err)
}
if reply2 != utils.OK {
t.Errorf("Error: %+v", reply2)
}
expectedIDX := map[string]utils.StringMap{"Account:1001": {"TEST_PROFILE1": true}}
indexes, err := onStor.GetFilterIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix, tenant, false), nil)
if err != nil {
t.Error(err)
}
if !reflect.DeepEqual(expectedIDX, indexes) {
t.Errorf("Expecting: %+v, received: %+v", expectedIDX, utils.ToJSON(indexes))
}
expectedRevIDX := map[string]utils.StringMap{"TEST_PROFILE1": {"Account:1001": true}}
indexes, err = onStor.GetFilterReverseIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix, tenant, true), nil)
if err != nil {
t.Error(err)
}
if !reflect.DeepEqual(expectedRevIDX, indexes) {
t.Errorf("Expecting: %+v, received: %+v", expectedRevIDX, utils.ToJSON(indexes))
}
}
func testV1FIdxSetSecondStatQueueProfileIndexes(t *testing.T) {
tenant := "cgrates.org"
var reply *engine.StatQueueProfile
filter = &engine.Filter{
Tenant: tenant,
ID: "FLTR_2",
RequestFilters: []*engine.RequestFilter{
&engine.RequestFilter{
FieldName: "Account",
Type: "*string",
Values: []string{"1001"},
},
},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
ExpiryTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
},
}
var result string
if err := tFIdxRpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := tFIdxRpc.Call("ApierV1.GetStatQueueProfile",
&utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE2"}, &reply); err == nil ||
err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
}
statConfig = &engine.StatQueueProfile{
Tenant: "cgrates.org",
ID: "TEST_PROFILE2",
FilterIDs: []string{"FLTR_2"},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
ExpiryTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
},
QueueLength: 10,
TTL: time.Duration(10) * time.Second,
Metrics: []*utils.MetricWithParams{
&utils.MetricWithParams{
MetricID: "MetricValue",
Parameters: "",
},
&utils.MetricWithParams{
MetricID: "MetricValueTwo",
Parameters: "",
},
},
Thresholds: []string{"Val1", "Val2"},
Blocker: true,
Stored: true,
Weight: 20,
MinItems: 1,
}
if err := tFIdxRpc.Call("ApierV1.SetStatQueueProfile", statConfig, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := tFIdxRpc.Call("ApierV1.GetStatQueueProfile",
&utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE2"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(statConfig, reply) {
t.Errorf("Expecting: %+v, received: %+v", statConfig, reply)
}
if err = onStor.RemoveFilterIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix,
tenant, false)); err != nil {
t.Error(err)
}
if err := onStor.RemoveFilterReverseIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix,
tenant, true), ""); err != nil {
t.Error(err)
}
if indexes, err = onStor.GetFilterIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix, tenant, false),
nil); err != utils.ErrNotFound {
t.Error(err)
}
}
func testV1FIdxSecondComputeStatQueueProfileIndexes(t *testing.T) {
tenant := "cgrates.org"
stid := []string{"TEST_PROFILE2"}
emptySlice := []string{}
var reply2 string
if err := tFIdxRpc.Call(utils.ApierV1ComputeFilterIndexes, utils.ArgsComputeFilterIndexes{
Tenant: "cgrates.org",
ThresholdIDs: &emptySlice,
AttributeIDs: &emptySlice,
ResourceIDs: &emptySlice,
StatIDs: &stid,
SupplierIDs: &emptySlice,
}, &reply2); err != nil {
t.Error(err)
}
if reply2 != utils.OK {
t.Errorf("Error: %+v", reply2)
}
expectedIDX := map[string]utils.StringMap{"Account:1001": {"TEST_PROFILE2": true}}
indexes, err := onStor.GetFilterIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix, tenant, false), nil)
if err != nil {
t.Error(err)
}
if !reflect.DeepEqual(expectedIDX, indexes) {
t.Errorf("Expecting: %+v, received: %+v", expectedIDX, utils.ToJSON(indexes))
}
expectedRevIDX := map[string]utils.StringMap{"TEST_PROFILE2": {"Account:1001": true}}
indexes, err = onStor.GetFilterReverseIndexes(engine.GetDBIndexKey(utils.StatQueueProfilePrefix, tenant, true), nil)
if err != nil {
t.Error(err)
}
if !reflect.DeepEqual(expectedRevIDX, indexes) {
t.Errorf("Expecting: %+v, received: %+v", expectedRevIDX, utils.ToJSON(indexes))
}
}
// 1.set threshold in datadb fara sa faca indexuri
// 2.fac querri la index sa fiu sigur ca is 0
// 3.compile indexes all
@@ -375,12 +614,12 @@ func testV1FIdxSecondComputeThresholdsIndexes(t *testing.T) {
// t.Fatal(err)
// }
// var err error
// tSv1Rpc, err = jsonrpc.Dial("tcp", tSv1Cfg.RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
// tFIdxRpc, err = jsonrpc.Dial("tcp", tSv1Cfg.RPCJSONListen) // We connect over JSON so we can also troubleshoot if needed
// if err != nil {
// t.Fatal("Could not connect to rater: ", err.Error())
// }
// var td engine.Threshold
// if err := tSv1Rpc.Call(utils.ThresholdSv1GetThreshold,
// if err := tFIdxRpc.Call(utils.ThresholdSv1GetThreshold,
// &utils.TenantID{Tenant: "cgrates.org", ID: "THD_ACNT_BALANCE_1"}, &td); err != nil {
// t.Error(err)
// } else if td.Snooze.IsZero() { // make sure Snooze time was reset during execution
@@ -409,12 +648,12 @@ func testV1FIdxSecondComputeThresholdsIndexes(t *testing.T) {
}
var result string
if err := tSv1Rpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
if err := tFIdxRpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := tSv1Rpc.Call("ApierV1.GetThresholdProfile",
if err := tFIdxRpc.Call("ApierV1.GetThresholdProfile",
&utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE1"}, &reply); err == nil ||
err.Error() != utils.ErrNotFound.Error() {
t.Error(err)
@@ -434,12 +673,12 @@ func testV1FIdxSecondComputeThresholdsIndexes(t *testing.T) {
ActionIDs: []string{"ACT_1", "ACT_2"},
Async: true,
}
if err := tSv1Rpc.Call("ApierV1.SetThresholdProfile", tPrfl, &result); err != nil {
if err := tFIdxRpc.Call("ApierV1.SetThresholdProfile", tPrfl, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := tSv1Rpc.Call("ApierV1.GetThresholdProfile",
if err := tFIdxRpc.Call("ApierV1.GetThresholdProfile",
&utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE1"}, &reply); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(tPrfl, reply) {
@@ -466,20 +705,20 @@ func testV1FIdxSecondComputeThresholdsIndexes(t *testing.T) {
// },
// }
// if err := tSv1Rpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
// if err := tFIdxRpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
// t.Error(err)
// } else if result != utils.OK {
// t.Error("Unexpected reply returned", result)
// }
// tPrfl.FilterIDs = []string{"TestFilter", "TestFilter2"}
// if err := tSv1Rpc.Call("ApierV1.SetThresholdProfile", tPrfl, &result); err != nil {
// if err := tFIdxRpc.Call("ApierV1.SetThresholdProfile", tPrfl, &result); err != nil {
// t.Error(err)
// } else if result != utils.OK {
// t.Error("Unexpected reply returned", result)
// }
// time.Sleep(time.Duration(100 * time.Millisecond)) // mongo is async
// var reply *engine.ThresholdProfile
// if err := tSv1Rpc.Call("ApierV1.GetThresholdProfile",
// if err := tFIdxRpc.Call("ApierV1.GetThresholdProfile",
// &utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE1"}, &reply); err != nil {
// t.Error(err)
// } else if !reflect.DeepEqual(tPrfl, reply) {
@@ -489,14 +728,14 @@ func testV1FIdxSecondComputeThresholdsIndexes(t *testing.T) {
// func testV1FIdxRemoveThresholdProfile(t *testing.T) {
// var resp string
// if err := tSv1Rpc.Call("ApierV1.RemThresholdProfile",
// if err := tFIdxRpc.Call("ApierV1.RemThresholdProfile",
// &utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE1"}, &resp); err != nil {
// t.Error(err)
// } else if resp != utils.OK {
// t.Error("Unexpected reply returned", resp)
// }
// var sqp *engine.ThresholdProfile
// if err := tSv1Rpc.Call("ApierV1.GetThresholdProfile",
// if err := tFIdxRpc.Call("ApierV1.GetThresholdProfile",
// &utils.TenantID{Tenant: "cgrates.org", ID: "TEST_PROFILE1"}, &sqp); err == nil ||
// err.Error() != utils.ErrNotFound.Error() {
// t.Error(err)

44
apier/v1/resourcesv1_it_test.go
View File

@@ -45,7 +45,7 @@ var sTestsRLSV1 = []func(t *testing.T){
testV1RsLoadConfig,
testV1RsInitDataDb,
testV1RsResetStorDb,
testV1RsStartEngine,
// testV1RsStartEngine,
testV1RsRpcConn,
testV1RsFromFolder,
testV1RsGetResourcesForEvent,
@@ -61,7 +61,7 @@ var sTestsRLSV1 = []func(t *testing.T){
testV1RsGetResourceProfileAfterUpdate,
testV1RsRemResourceProfile,
testV1RsGetResourceProfileAfterDelete,
testV1RsStopEngine,
// testV1RsStopEngine,
}
//Test start here
@@ -560,7 +560,27 @@ func testV1RsSetResourceProfile(t *testing.T) {
Weight: 20,
Thresholds: []string{"Val1", "Val2"},
}
filter = &engine.Filter{
Tenant: "cgrates.org",
ID: "FLTR_RES_RCFG1",
RequestFilters: []*engine.RequestFilter{
&engine.RequestFilter{
FieldName: "*string",
Type: "Account",
Values: []string{"1001", "1002"},
},
},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
ExpiryTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
},
}
var result string
if err := rlsV1Rpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := rlsV1Rpc.Call("ApierV1.SetResourceProfile", rlsConfig, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
@@ -582,6 +602,26 @@ func testV1RsGetResourceProfileAfterSet(t *testing.T) {
func testV1RsUpdateResourceProfile(t *testing.T) {
var result string
rlsConfig.FilterIDs = []string{"FLTR_RES_RCFG1", "FLTR_RES_RCFG2"}
filter = &engine.Filter{
Tenant: "cgrates.org",
ID: "FLTR_RES_RCFG2",
RequestFilters: []*engine.RequestFilter{
&engine.RequestFilter{
FieldName: "*string",
Type: "Account",
Values: []string{"1001", "1002"},
},
},
ActivationInterval: &utils.ActivationInterval{
ActivationTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
ExpiryTime: time.Date(2014, 7, 14, 14, 25, 0, 0, time.UTC).Local(),
},
}
if err := rlsV1Rpc.Call("ApierV1.SetFilter", filter, &result); err != nil {
t.Error(err)
} else if result != utils.OK {
t.Error("Unexpected reply returned", result)
}
if err := rlsV1Rpc.Call("ApierV1.SetResourceProfile", rlsConfig, &result); err != nil {
t.Error(err)
} else if result != utils.OK {

2
apier/v1/stats.go
View File

@@ -43,7 +43,7 @@ func (apierV1 *ApierV1) SetStatQueueProfile(sqp *engine.StatQueueProfile, reply
if missing := utils.MissingStructFields(sqp, []string{"Tenant", "ID"}); len(missing) != 0 {
return utils.NewErrMandatoryIeMissing(missing...)
}
if err := apierV1.DataManager.SetStatQueueProfile(sqp, false); err != nil {
if err := apierV1.DataManager.SetStatQueueProfile(sqp, true); err != nil {
return utils.APIErrorHandler(err)
}
cache.RemKey(utils.StatQueueProfilePrefix+utils.ConcatenatedKey(sqp.Tenant, sqp.ID),

15
cmd/cgr-migrator/cgr-migrator.go
View File

@@ -40,28 +40,28 @@ var (
"\n <*set_versions|*cost_details|*accounts|*actions|*action_triggers|*action_plans|*shared_groups|*stordb|*datadb> ")
version = flag.Bool("version", false, "Prints the application version.")
outDataDBType = flag.String("out_datadb_type", "", "The type of the DataDb Database <redis|mongo>")
outDataDBType = flag.String("out_datadb_type", "", "The type of the DataDb Database <*redis|*redis>")
outDataDBHost = flag.String("out_datadb_host", config.CgrConfig().DataDbHost, "The DataDb host to connect to.")
outDataDBPort = flag.String("out_datadb_port", config.CgrConfig().DataDbPort, "The DataDb port to bind to.")
outDataDBName = flag.String("out_datadb_name", config.CgrConfig().DataDbName, "The name/number of the DataDb to connect to.")
outDataDBUser = flag.String("out_datadb_user", config.CgrConfig().DataDbUser, "The DataDb user to sign in as.")
outDataDBPass = flag.String("out_datadb_passwd", config.CgrConfig().DataDbPass, "The DataDb user's password.")
outStorDBType = flag.String("out_stordb_type", "", "The type of the StorDB Database <mysql|postgres|mongo>")
outStorDBType = flag.String("out_stordb_type", "", "The type of the StorDB Database <*mysql|*postgres|*mongo>")
outStorDBHost = flag.String("out_stordb_host", config.CgrConfig().StorDBHost, "The StorDB host to connect to.")
outStorDBPort = flag.String("out_stordb_port", config.CgrConfig().StorDBPort, "The StorDB port to bind to.")
outStorDBName = flag.String("out_stordb_name", config.CgrConfig().StorDBName, "The name/number of the StorDB to connect to.")
outStorDBUser = flag.String("out_stordb_user", config.CgrConfig().StorDBUser, "The StorDB user to sign in as.")
outStorDBPass = flag.String("out_stordb_passwd", config.CgrConfig().StorDBPass, "The StorDB user's password.")
inDataDBType = flag.String("datadb_type", config.CgrConfig().DataDbType, "The type of the DataDb Database <redis>")
inDataDBType = flag.String("datadb_type", config.CgrConfig().DataDbType, "The type of the DataDb Database <*redis>")
inDataDBHost = flag.String("datadb_host", config.CgrConfig().DataDbHost, "The DataDb host to connect to.")
inDataDBPort = flag.String("datadb_port", config.CgrConfig().DataDbPort, "The DataDb port to bind to.")
inDataDBName = flag.String("datadb_name", config.CgrConfig().DataDbName, "The name/number of the DataDb to connect to.")
inDataDBUser = flag.String("datadb_user", config.CgrConfig().DataDbUser, "The DataDb user to sign in as.")
inDataDBPass = flag.String("datadb_passwd", config.CgrConfig().DataDbPass, "The DataDb user's password.")
inStorDBType = flag.String("stordb_type", config.CgrConfig().StorDBType, "The type of the StorDB Database <mysql|postgres>")
inStorDBType = flag.String("stordb_type", config.CgrConfig().StorDBType, "The type of the StorDB Database <*mysql|*postgres>")
inStorDBHost = flag.String("stordb_host", config.CgrConfig().StorDBHost, "The StorDB host to connect to.")
inStorDBPort = flag.String("stordb_port", config.CgrConfig().StorDBPort, "The StorDB port to bind to.")
inStorDBName = flag.String("stordb_name", config.CgrConfig().StorDBName, "The name/number of the StorDB to connect to.")
@@ -85,10 +85,11 @@ func main() {
fmt.Println(utils.GetCGRVersion())
return
}
*inDataDBType = strings.TrimPrefix(*inDataDBType, "*")
var dmIN *engine.DataManager
dmIN, _ = engine.ConfigureDataStorage(*inDataDBType, *inDataDBHost, *inDataDBPort,
*inDataDBName, *inDataDBUser, *inDataDBPass, *dbDataEncoding, config.CgrConfig().CacheCfg(), *loadHistorySize)
*inStorDBType = strings.TrimPrefix(*inStorDBType, "*")
instorDB, err := engine.ConfigureStorStorage(*inStorDBType, *inStorDBHost, *inStorDBPort, *inStorDBName, *inStorDBUser, *inStorDBPass, *inDBDataEncoding,
config.CgrConfig().StorDBMaxOpenConns, config.CgrConfig().StorDBMaxIdleConns, config.CgrConfig().StorDBConnMaxLifetime, config.CgrConfig().StorDBCDRSIndexes)
if err != nil {
@@ -101,6 +102,8 @@ func main() {
*outDataDBName = *inDataDBName
*outDataDBUser = *inDataDBUser
*outDataDBPass = *inDataDBPass
} else {
*outDataDBType = strings.TrimPrefix(*outDataDBType, "*")
}
var dmOUT *engine.DataManager
dmOUT, _ = engine.ConfigureDataStorage(*outDataDBType, *outDataDBHost, *outDataDBPort,
@@ -110,8 +113,8 @@ func main() {
log.Fatal(err)
}
storDB = instorDB
if *outStorDBType != "" {
*outStorDBType = strings.TrimPrefix(*outStorDBType, "*")
storDB, err = engine.ConfigureStorStorage(*outStorDBType, *outStorDBHost, *outStorDBPort, *outStorDBName, *outStorDBUser, *outStorDBPass, *dbDataEncoding,
config.CgrConfig().StorDBMaxOpenConns, config.CgrConfig().StorDBMaxIdleConns, config.CgrConfig().StorDBConnMaxLifetime, config.CgrConfig().StorDBCDRSIndexes)
if err != nil {

69
engine/datamanager.go
View File

@@ -380,6 +380,9 @@ func (dm *DataManager) GetThresholdProfile(tenant, id string, skipCache bool, tr
}
func (dm *DataManager) SetThresholdProfile(th *ThresholdProfile, withIndex bool) (err error) {
if err = dm.DataDB().SetThresholdProfileDrv(th); err != nil {
return err
}
if withIndex {
indexer := NewReqFilterIndexer(dm, utils.ThresholdProfilePrefix, th.Tenant)
//remove old ThresholdProfile indexes
@@ -403,7 +406,7 @@ func (dm *DataManager) SetThresholdProfile(th *ThresholdProfile, withIndex bool)
return
}
}
return dm.DataDB().SetThresholdProfileDrv(th)
return
}
func (dm *DataManager) RemoveThresholdProfile(tenant, id, transactionID string) (err error) {
@@ -438,6 +441,9 @@ func (dm *DataManager) GetStatQueueProfile(tenant, id string, skipCache bool, tr
}
func (dm *DataManager) SetStatQueueProfile(sqp *StatQueueProfile, withIndex bool) (err error) {
if err = dm.DataDB().SetStatQueueProfileDrv(sqp); err != nil {
return err
}
// if withIndex {
// indexer := NewReqFilterIndexer(dm, utils.ThresholdProfilePrefix, sqp.Tenant)
// //remove old StatQueueProfile indexes
@@ -460,7 +466,7 @@ func (dm *DataManager) SetStatQueueProfile(sqp *StatQueueProfile, withIndex bool
// return
// }
// }
return dm.DataDB().SetStatQueueProfileDrv(sqp)
return
}
func (dm *DataManager) RemoveStatQueueProfile(tenant, id, transactionID string) (err error) {
@@ -561,30 +567,33 @@ func (dm *DataManager) GetResourceProfile(tenant, id string, skipCache bool, tra
}
func (dm *DataManager) SetResourceProfile(rp *ResourceProfile, withIndex bool) (err error) {
if err = dm.DataDB().SetResourceProfileDrv(rp); err != nil {
return err
}
//to be implemented in tests
// if withIndex {
// indexer := NewReqFilterIndexer(dm, utils.ResourceProfilesPrefix, rp.Tenant)
// //remove old ResourceProfiles indexes
// if err = indexer.RemoveItemFromIndex(rp.ID); err != nil &&
// err.Error() != utils.ErrNotFound.Error() {
// return
// }
// //Verify matching Filters for every FilterID from ResourceProfiles
// for _, fltrID := range rp.FilterIDs {
// var fltr *Filter
// if fltr, err = dm.GetFilter(rp.Tenant, fltrID, false, utils.NonTransactional); err != nil {
// if err == utils.ErrNotFound {
// err = fmt.Errorf("broken reference to filter: %+v for threshold: %+v", fltrID, rp)
// }
// return
// }
// indexer.IndexTPFilter(FilterToTPFilter(fltr), rp.ID)
// }
// if err = indexer.StoreIndexes(); err != nil {
// return
// }
// }
return dm.DataDB().SetResourceProfileDrv(rp)
if withIndex {
indexer := NewReqFilterIndexer(dm, utils.ResourceProfilesPrefix, rp.Tenant)
//remove old ResourceProfiles indexes
if err = indexer.RemoveItemFromIndex(rp.ID); err != nil &&
err.Error() != utils.ErrNotFound.Error() {
return
}
//Verify matching Filters for every FilterID from ResourceProfiles
for _, fltrID := range rp.FilterIDs {
var fltr *Filter
if fltr, err = dm.GetFilter(rp.Tenant, fltrID, false, utils.NonTransactional); err != nil {
if err == utils.ErrNotFound {
err = fmt.Errorf("broken reference to filter: %+v for threshold: %+v", fltrID, rp)
}
return
}
indexer.IndexTPFilter(FilterToTPFilter(fltr), rp.ID)
}
if err = indexer.StoreIndexes(); err != nil {
return
}
}
return
}
func (dm *DataManager) RemoveResourceProfile(tenant, id, transactionID string) (err error) {
@@ -1003,6 +1012,9 @@ func (dm *DataManager) GetSupplierProfile(tenant, id string, skipCache bool, tra
}
func (dm *DataManager) SetSupplierProfile(supp *SupplierProfile, withIndex bool) (err error) {
if err = dm.DataDB().SetSupplierProfileDrv(supp); err != nil {
return err
}
//to be implemented in tests
// if withIndex {
// indexer := NewReqFilterIndexer(dm, utils.SupplierProfilePrefix, supp.Tenant)
@@ -1026,7 +1038,7 @@ func (dm *DataManager) SetSupplierProfile(supp *SupplierProfile, withIndex bool)
// return
// }
// }
return dm.DataDB().SetSupplierProfileDrv(supp)
return
}
func (dm *DataManager) RemoveSupplierProfile(tenant, id, transactionID string) (err error) {
@@ -1060,6 +1072,9 @@ func (dm *DataManager) GetAttributeProfile(tenant, id string, skipCache bool, tr
}
func (dm *DataManager) SetAttributeProfile(ap *AttributeProfile, withIndex bool) (err error) {
if err = dm.DataDB().SetAttributeProfileDrv(ap); err != nil {
return err
}
//to be implemented in tests
// if withIndex {
// indexer := NewReqFilterIndexer(dm, utils.AttributeProfilePrefix, ap.Tenant)
@@ -1083,7 +1098,7 @@ func (dm *DataManager) SetAttributeProfile(ap *AttributeProfile, withIndex bool)
// return
// }
// }
return dm.DataDB().SetAttributeProfileDrv(ap)
return
}
func (dm *DataManager) RemoveAttributeProfile(tenant, id, transactionID string) (err error) {

2
engine/onstor_it_test.go
View File

@@ -967,7 +967,7 @@ func testOnStorITCacheStatQueueProfile(t *testing.T) {
TTL: time.Duration(10) * time.Second,
Metrics: []*utils.MetricWithParams{
&utils.MetricWithParams{
MetricID: "ASR",
MetricID: "*asr",
Parameters: "",
},
},

2
utils/consts.go
View File

@@ -545,7 +545,7 @@ const (
// MetaFilterIndexesAPIs
const (
ComputeFilterIndexes = "ApierV1.ComputeFilterIndexes"
ApierV1ComputeFilterIndexes = "ApierV1.ComputeFilterIndexes"
)
// MetaSupplierAPIs