Added tests for dispatcher with load strategy

apier/v1/api_interfaces.go

@@ -174,6 +174,7 @@ type ConfigSv1Interface interface {
 type CoreSv1Interface interface {
 	Status(arg *utils.TenantWithOpts, reply *map[string]interface{}) error
 	Ping(ign *utils.CGREventWithOpts, reply *string) error
+	Sleep(arg *utils.DurationArgs, reply *string) error
 }
 
 type RateSv1Interface interface {

apier/v1/concreqs_it_test.go

@@ -93,7 +93,7 @@ func testConcReqsStartEngine(t *testing.T) {
 	}
 }
 
-func handlePing(clnt *rpc2.Client, arg *DurationArgs, reply *string) error {
+func handlePing(clnt *rpc2.Client, arg *utils.DurationArgs, reply *string) error {
 	time.Sleep(arg.DurationTime)
 	*reply = utils.OK
 	return nil
@@ -124,7 +124,7 @@ func testConcReqsBusyAPIs(t *testing.T) {
 		go func() {
 			var resp string
 			if err := concReqsRPC.Call(utils.CoreSv1Sleep,
-				&DurationArgs{DurationTime: time.Duration(10 * time.Millisecond)},
+				&utils.DurationArgs{DurationTime: time.Duration(10 * time.Millisecond)},
 				&resp); err != nil {
 				lock.Lock()
 				failedAPIs++
@@ -151,7 +151,7 @@ func testConcReqsQueueAPIs(t *testing.T) {
 		go func() {
 			var resp string
 			if err := concReqsRPC.Call(utils.CoreSv1Sleep,
-				&DurationArgs{DurationTime: time.Duration(10 * time.Millisecond)},
+				&utils.DurationArgs{DurationTime: time.Duration(10 * time.Millisecond)},
 				&resp); err != nil {
 				wg.Done()
 				t.Error(err)
@@ -234,7 +234,7 @@ func testConcReqsOnBiJSONBusy(t *testing.T) {
 		go func() {
 			var resp string
 			if err := concReqsBiRPC.Call(utils.SessionSv1Sleep,
-				&DurationArgs{DurationTime: time.Duration(10 * time.Millisecond)},
+				&utils.DurationArgs{DurationTime: time.Duration(10 * time.Millisecond)},
 				&resp); err != nil {
 				lock.Lock()
 				failedAPIs++
@@ -261,7 +261,7 @@ func testConcReqsOnBiJSONQueue(t *testing.T) {
 		go func() {
 			var resp string
 			if err := concReqsBiRPC.Call(utils.SessionSv1Sleep,
-				&DurationArgs{DurationTime: time.Duration(10 * time.Millisecond)},
+				&utils.DurationArgs{DurationTime: time.Duration(10 * time.Millisecond)},
 				&resp); err != nil {
 				wg.Done()
 				t.Error(err)

apier/v1/core.go

@@ -50,12 +50,8 @@ func (cS *CoreSv1) Ping(ign *utils.CGREventWithOpts, reply *string) error {
 	return nil
 }
 
-type DurationArgs struct {
-	DurationTime time.Duration
-}
-
 // Sleep is used to test the concurrent requests mechanism
-func (cS *CoreSv1) Sleep(arg *DurationArgs, reply *string) error {
+func (cS *CoreSv1) Sleep(arg *utils.DurationArgs, reply *string) error {
 	time.Sleep(arg.DurationTime)
 	*reply = utils.OK
 	return nil

apier/v1/dispatcher.go

@@ -34,11 +34,11 @@ func (apierSv1 *APIerSv1) GetDispatcherProfile(arg *utils.TenantID, reply *engin
 	if missing := utils.MissingStructFields(arg, []string{"Tenant", "ID"}); len(missing) != 0 { //Params missing
 		return utils.NewErrMandatoryIeMissing(missing...)
 	}
-	if dpp, err := apierSv1.DataManager.GetDispatcherProfile(arg.Tenant, arg.ID, true, true, utils.NonTransactional); err != nil {
+	dpp, err := apierSv1.DataManager.GetDispatcherProfile(arg.Tenant, arg.ID, true, true, utils.NonTransactional)
+	if err != nil {
 		return utils.APIErrorHandler(err)
-	} else {
-		*reply = *dpp
 	}
+	*reply = *dpp
 	return nil
 }
 
@@ -124,11 +124,11 @@ func (apierSv1 *APIerSv1) GetDispatcherHost(arg *utils.TenantID, reply *engine.D
 	if missing := utils.MissingStructFields(arg, []string{"Tenant", "ID"}); len(missing) != 0 { //Params missing
 		return utils.NewErrMandatoryIeMissing(missing...)
 	}
-	if dpp, err := apierSv1.DataManager.GetDispatcherHost(arg.Tenant, arg.ID, true, false, utils.NonTransactional); err != nil {
+	dpp, err := apierSv1.DataManager.GetDispatcherHost(arg.Tenant, arg.ID, true, false, utils.NonTransactional)
+	if err != nil {
 		return utils.APIErrorHandler(err)
-	} else {
-		*reply = *dpp
 	}
+	*reply = *dpp
 	return nil
 }
 
@@ -841,6 +841,10 @@ func (dS *DispatcherCoreSv1) Ping(args *utils.CGREventWithOpts, reply *string) e
 	return dS.dS.CoreSv1Ping(args, reply)
 }
 
+func (dS *DispatcherCoreSv1) Sleep(arg *utils.DurationArgs, reply *string) error {
+	return dS.dS.CoreSv1Sleep(arg, reply)
+}
+
 func NewDispatcherRALsV1(dps *dispatchers.DispatcherService) *DispatcherRALsV1 {
 	return &DispatcherRALsV1{dS: dps}
 }

apier/v1/sessionsbirpc.go

@@ -303,7 +303,7 @@ func (ssv1 *SessionSv1) BiRPCV1STIRIdentity(clnt *rpc2.Client,
 	return ssv1.Ss.BiRPCv1STIRIdentity(nil, args, reply)
 }
 
-func (ssv1 *SessionSv1) BiRPCV1Sleep(clnt *rpc2.Client, arg *DurationArgs,
+func (ssv1 *SessionSv1) BiRPCV1Sleep(clnt *rpc2.Client, arg *utils.DurationArgs,
 	reply *string) (err error) {
 	if err = utils.ConReqs.Allocate(); err != nil {
 		return

data/tariffplans/dispatchers/Attributes.csv

@@ -23,3 +23,4 @@ cgrates.org,ATTR_API_RALS_AUTH,*auth,*string:~*req.ApiKey:rals12345,,,*req.APIMe
 cgrates.org,ATTR_API_REPLICATOR_AUTH,*auth,*string:~*req.ApiKey:repl12345,,,*req.APIMethods,*constant,ReplicatorSv1.Ping&ReplicatorSv1.GetAccount&ReplicatorSv1.SetAccount&ReplicatorSv1.RemoveAccount&ReplicatorSv1.GetRouteProfile&ReplicatorSv1.SetRouteProfile&ReplicatorSv1.RemoveRouteProfile&ReplicatorSv1.GetAttributeProfile&ReplicatorSv1.SetAttributeProfile&ReplicatorSv1.RemoveAttributeProfile&ReplicatorSv1.SetChargerProfile&ReplicatorSv1.GetChargerProfile&ReplicatorSv1.RemoveChargerProfile&ReplicatorSv1.GetDispatcherProfile&ReplicatorSv1.SetDispatcherProfile&ReplicatorSv1.RemoveDispatcherProfile&ReplicatorSv1.GetDispatcherHost&ReplicatorSv1.SetDispatcherHost&ReplicatorSv1.RemoveDispatcherHost&ReplicatorSv1.GetFilter&ReplicatorSv1.SetFilter&ReplicatorSv1.RemoveFilter&ReplicatorSv1.GetThreshold&ReplicatorSv1.SetThreshold&ReplicatorSv1.RemoveThreshold&ReplicatorSv1.GetStatQueue&ReplicatorSv1.SetStatQueue&ReplicatorSv1.RemoveStatQueue&ReplicatorSv1.GetResource&ReplicatorSv1.SetResource&ReplicatorSv1.RemoveResource&ReplicatorSv1.GetResourceProfile&ReplicatorSv1.SetResourceProfile&ReplicatorSv1.RemoveResourceProfile&ReplicatorSv1.GetStatQueueProfile&ReplicatorSv1.SetStatQueueProfile&ReplicatorSv1.RemoveStatQueueProfile&ReplicatorSv1.GetThresholdProfile&ReplicatorSv1.SetThresholdProfile&ReplicatorSv1.RemoveThresholdProfile&ReplicatorSv1.GetTiming&ReplicatorSv1.SetTiming&ReplicatorSv1.RemoveTiming&ReplicatorSv1.GetActionTriggers&ReplicatorSv1.SetActionTriggers&ReplicatorSv1.RemoveActionTriggers&ReplicatorSv1.SetSharedGroup&ReplicatorSv1.GetSharedGroup&ReplicatorSv1.RemoveSharedGroup&ReplicatorSv1.SetActions&ReplicatorSv1.GetActions&ReplicatorSv1.RemoveActions&ReplicatorSv1.SetActionPlan&ReplicatorSv1.GetActionPlan&ReplicatorSv1.RemoveActionPlan&ReplicatorSv1.SetAccountActionPlans&ReplicatorSv1.GetAccountActionPlans&ReplicatorSv1.RemAccountActionPlans&ReplicatorSv1.SetRatingPlan&ReplicatorSv1.GetRatingPlan&ReplicatorSv1.RemoveRatingPlan&ReplicatorSv1.SetRatingProfile&ReplicatorSv1.GetRatingProfile&ReplicatorSv1.RemoveRatingProfile&ReplicatorSv1.SetDestination&ReplicatorSv1.GetDestination&ReplicatorSv1.RemoveDestination&ReplicatorSv1.SetLoadIDs&ReplicatorSv1.GetItemLoadIDs&ReplicatorSv1.SetRateProfile&ReplicatorSv1.GetRateProfile&ReplicatorSv1.RemoveRateProfile,false,20
 cgrates.org,ATTR_API_CDRSV2,*auth,*string:~*req.ApiKey:cdrsv212345,,,*req.APIMethods,*constant,CDRsV2.ProcessEvent&CDRsV2.StoreSessionCost,false,20
 cgrates.org,ATTR_API_RATES_AUTH,*auth,*string:~*req.ApiKey:rPrf12345,,,*req.APIMethods,*constant,RateSv1.Ping,false,20
+cgrates.org,ATTR_API_CORE_AUTH,*auth,*string:~*req.ApiKey:core12345,,,*req.APIMethods,*constant,CoreSv1.Status&CoreSv1.Ping&CoreSv1.Sleep,false,20

data/tariffplans/dispatchers/DispatcherProfiles.csv

@@ -12,4 +12,6 @@ cgrates.org,EVENT3,,,,,,ALL,,10,,,
 cgrates.org,EVENT4,*any,*string:~*req.EventName:Broadcast,,*broadcast,,ALL2,,20,false,,20
 cgrates.org,EVENT4,,,,,,ALL,,10,,,
 cgrates.org,EVENT5,*any,*string:~*req.EventName:Internal,,*weight,,SELF,,20,false,,20
-cgrates.org,EVENT6,*any,*string:~*opts.*method:DispatcherSv1.GetProfileForEvent,,*weight,,SELF,,20,false,,20
+cgrates.org,EVENT6,*any,*string:~*opts.*method:DispatcherSv1.GetProfileForEvent,,*weight,,SELF,,20,false,,20
+cgrates.org,EVENT7,*any,*string:~*opts.EventType:LoadDispatcher,,*weight,,ALL,,20,false,,207
+cgrates.org,EVENT7,*any,,,*weight,,ALL2,,20,false,*ratio:1,207

data/tariffplans/dispatchers_gob/Attributes.csv

@@ -23,3 +23,4 @@ cgrates.org,ATTR_API_RALS_AUTH,*auth,*string:~*req.ApiKey:rals12345,,,*req.APIMe
 cgrates.org,ATTR_API_REPLICATOR_AUTH,*auth,*string:~*req.ApiKey:repl12345,,,*req.APIMethods,*constant,ReplicatorSv1.Ping&ReplicatorSv1.GetAccount&ReplicatorSv1.SetAccount&ReplicatorSv1.RemoveAccount&ReplicatorSv1.GetRouteProfile&ReplicatorSv1.SetRouteProfile&ReplicatorSv1.RemoveRouteProfile&ReplicatorSv1.GetAttributeProfile&ReplicatorSv1.SetAttributeProfile&ReplicatorSv1.RemoveAttributeProfile&ReplicatorSv1.SetChargerProfile&ReplicatorSv1.GetChargerProfile&ReplicatorSv1.RemoveChargerProfile&ReplicatorSv1.GetDispatcherProfile&ReplicatorSv1.SetDispatcherProfile&ReplicatorSv1.RemoveDispatcherProfile&ReplicatorSv1.GetDispatcherHost&ReplicatorSv1.SetDispatcherHost&ReplicatorSv1.RemoveDispatcherHost&ReplicatorSv1.GetFilter&ReplicatorSv1.SetFilter&ReplicatorSv1.RemoveFilter&ReplicatorSv1.GetThreshold&ReplicatorSv1.SetThreshold&ReplicatorSv1.RemoveThreshold&ReplicatorSv1.GetStatQueue&ReplicatorSv1.SetStatQueue&ReplicatorSv1.RemoveStatQueue&ReplicatorSv1.GetResource&ReplicatorSv1.SetResource&ReplicatorSv1.RemoveResource&ReplicatorSv1.GetResourceProfile&ReplicatorSv1.SetResourceProfile&ReplicatorSv1.RemoveResourceProfile&ReplicatorSv1.GetStatQueueProfile&ReplicatorSv1.SetStatQueueProfile&ReplicatorSv1.RemoveStatQueueProfile&ReplicatorSv1.GetThresholdProfile&ReplicatorSv1.SetThresholdProfile&ReplicatorSv1.RemoveThresholdProfile&ReplicatorSv1.GetTiming&ReplicatorSv1.SetTiming&ReplicatorSv1.RemoveTiming&ReplicatorSv1.GetActionTriggers&ReplicatorSv1.SetActionTriggers&ReplicatorSv1.RemoveActionTriggers&ReplicatorSv1.SetSharedGroup&ReplicatorSv1.GetSharedGroup&ReplicatorSv1.RemoveSharedGroup&ReplicatorSv1.SetActions&ReplicatorSv1.GetActions&ReplicatorSv1.RemoveActions&ReplicatorSv1.SetActionPlan&ReplicatorSv1.GetActionPlan&ReplicatorSv1.RemoveActionPlan&ReplicatorSv1.SetAccountActionPlans&ReplicatorSv1.GetAccountActionPlans&ReplicatorSv1.RemAccountActionPlans&ReplicatorSv1.SetRatingPlan&ReplicatorSv1.GetRatingPlan&ReplicatorSv1.RemoveRatingPlan&ReplicatorSv1.SetRatingProfile&ReplicatorSv1.GetRatingProfile&ReplicatorSv1.RemoveRatingProfile&ReplicatorSv1.SetDestination&ReplicatorSv1.GetDestination&ReplicatorSv1.RemoveDestination&ReplicatorSv1.SetLoadIDs&ReplicatorSv1.GetItemLoadIDs&ReplicatorSv1.SetRateProfile&ReplicatorSv1.GetRateProfile&ReplicatorSv1.RemoveRateProfile,false,20
 cgrates.org,ATTR_API_CDRSV2,*auth,*string:~*req.ApiKey:cdrsv212345,,,*req.APIMethods,*constant,CDRsV2.ProcessEvent&CDRsV2.StoreSessionCost,false,20
 cgrates.org,ATTR_API_RATES_AUTH,*auth,*string:~*req.ApiKey:rPrf12345,,,*req.APIMethods,*constant,RateSv1.Ping,false,20
+cgrates.org,ATTR_API_CORE_AUTH,*auth,*string:~*req.ApiKey:core12345,,,*req.APIMethods,*constant,CoreSv1.Status&CoreSv1.Ping&CoreSv1.Sleep,false,20

data/tariffplans/dispatchers_gob/DispatcherProfiles.csv

@@ -12,3 +12,6 @@ cgrates.org,EVENT3,,,,,,ALL,,10,,,
 cgrates.org,EVENT4,*any,*string:~*req.EventName:Broadcast,,*broadcast,,ALL2,,20,false,,20
 cgrates.org,EVENT4,,,,,,ALL,,10,,,
 cgrates.org,EVENT5,*any,*string:~*req.EventName:Internal,,*weight,,SELF,,20,false,,20
+cgrates.org,EVENT6,*any,*string:~*opts.*method:DispatcherSv1.GetProfileForEvent,,*weight,,SELF,,20,false,,20
+cgrates.org,EVENT7,*any,*string:~*opts.EventType:LoadDispatcher,,*weight,,ALL,,20,false,,207
+cgrates.org,EVENT7,*any,,,*weight,,ALL2,,20,false,*ratio:1,207

dispatchers/core.go

@@ -57,3 +57,23 @@ func (dS *DispatcherService) CoreSv1Ping(args *utils.CGREventWithOpts, reply *st
 	}
 	return dS.Dispatch(args, utils.MetaCore, utils.CoreSv1Ping, args, reply)
 }
+
+func (dS *DispatcherService) CoreSv1Sleep(args *utils.DurationArgs,
+	reply *string) (err error) {
+	tnt := dS.cfg.GeneralCfg().DefaultTenant
+	if args.TenantArg != nil && args.TenantArg.Tenant != utils.EmptyString {
+		tnt = args.TenantArg.Tenant
+	}
+	if len(dS.cfg.DispatcherSCfg().AttributeSConns) != 0 {
+		if err = dS.authorize(utils.CoreSv1Sleep, tnt,
+			utils.IfaceAsString(args.Opts[utils.OptsAPIKey]), utils.TimePointer(time.Now())); err != nil {
+			return
+		}
+	}
+	return dS.Dispatch(&utils.CGREventWithOpts{
+		CGREvent: &utils.CGREvent{
+			Tenant: tnt,
+		},
+		Opts: args.Opts,
+	}, utils.MetaCore, utils.CoreSv1Sleep, args, reply)
+}

dispatchers/core_it_test.go

@@ -0,0 +1,113 @@
+// +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 dispatchers
+
+import (
+	"net/rpc"
+	"testing"
+	"time"
+
+	"github.com/cgrates/cgrates/utils"
+)
+
+var sTestsDspCore = []func(t *testing.T){
+	testDspCoreLoad,
+}
+
+//Test start here
+func TestDspCoreIT(t *testing.T) {
+	var config1, config2, config3 string
+	switch *dbType {
+	case utils.MetaInternal:
+		t.SkipNow()
+	case utils.MetaMySQL:
+		config1 = "all_mysql"
+		config2 = "all2_mysql"
+		config3 = "dispatchers_mysql"
+	case utils.MetaMongo:
+		config1 = "all_mongo"
+		config2 = "all2_mongo"
+		config3 = "dispatchers_mongo"
+	case utils.MetaPostgres:
+		t.SkipNow()
+	default:
+		t.Fatal("Unknown Database type")
+	}
+
+	dispDIR := "dispatchers"
+	if *encoding == utils.MetaGOB {
+		dispDIR += "_gob"
+	}
+	testDsp(t, sTestsDspCore, "TestDspCoreIT", config1, config2, config3, "tutorial", "oldtutorial", dispDIR)
+}
+
+func testDspCoreLoad(t *testing.T) {
+	var status map[string]interface{}
+	statusTnt := utils.TenantWithOpts{
+		TenantArg: &utils.TenantArg{
+			Tenant: "cgrates.org",
+		},
+		Opts: map[string]interface{}{
+			utils.OptsAPIKey:  "core12345",
+			utils.OptsRouteID: "core1",
+			"EventType":       "LoadDispatcher",
+		},
+	}
+	expNodeID := "ALL"
+	if err := dispEngine.RPC.Call(utils.CoreSv1Status, statusTnt, &status); err != nil {
+		t.Error(err)
+	} else if status[utils.NodeID] == "ALL2" {
+		expNodeID = "ALL2"
+	}
+	dur := &utils.DurationArgs{
+		DurationTime: 500 * time.Millisecond,
+		TenantArg: &utils.TenantArg{
+			Tenant: "cgrates.org",
+		},
+		Opts: map[string]interface{}{
+			utils.OptsAPIKey:  "core12345",
+			utils.OptsRouteID: "core1",
+			"EventType":       "LoadDispatcher",
+		},
+	}
+	var rply string
+	statusTnt2 := utils.TenantWithOpts{
+		TenantArg: &utils.TenantArg{
+			Tenant: "cgrates.org",
+		},
+		Opts: map[string]interface{}{
+			utils.OptsAPIKey: "core12345",
+			"EventType":      "LoadDispatcher",
+		},
+	}
+	call := dispEngine.RPC.Go(utils.CoreSv1Sleep, dur, &rply, make(chan *rpc.Call, 1))
+	if err := dispEngine.RPC.Call(utils.CoreSv1Status, statusTnt2, &status); err != nil {
+		t.Error(err)
+	} else if status[utils.NodeID] != expNodeID {
+		t.Errorf("Expected status to be called on node <%s> but it was called on <%s>", expNodeID, status[utils.NodeID])
+	}
+	if ans := <-call.Done; ans.Error != nil {
+		t.Fatal(ans.Error)
+	} else if rply != utils.OK {
+		t.Errorf("Expected: %q ,received: %q", utils.OK, rply)
+	}
+
+}

dispatchers/libdispatcher_test.go

@@ -19,6 +19,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>
 package dispatchers
 
 import (
+	"reflect"
 	"testing"
 
 	"github.com/cgrates/cgrates/engine"
@@ -59,3 +60,33 @@ func TestLoadMetricsGetHosts(t *testing.T) {
 		t.Errorf("Expected: %q ,received: %q", "DSP_2", rply[0])
 	}
 }
+
+func TestNewSingleStrategyDispatcher(t *testing.T) {
+	dhp := engine.DispatcherHostProfiles{
+		{ID: "DSP_1"},
+		{ID: "DSP_2"},
+		{ID: "DSP_3"},
+		{ID: "DSP_4"},
+		{ID: "DSP_5"},
+	}
+	var exp strategyDispatcher = new(singleResultstrategyDispatcher)
+	if rply := newSingleStrategyDispatcher(dhp, utils.EmptyString); !reflect.DeepEqual(exp, rply) {
+		t.Errorf("Expected:  singleResultstrategyDispatcher structure,received: %s", utils.ToJSON(rply))
+	}
+
+	dhp = engine.DispatcherHostProfiles{
+		{ID: "DSP_1"},
+		{ID: "DSP_2"},
+		{ID: "DSP_3"},
+		{ID: "DSP_4"},
+		{ID: "DSP_5", Params: map[string]interface{}{utils.MetaRatio: 1}},
+	}
+	exp = &loadStrategyDispatcher{
+		hosts: dhp,
+		tntID: "cgrates.org",
+	}
+	if rply := newSingleStrategyDispatcher(dhp, "cgrates.org"); !reflect.DeepEqual(exp, rply) {
+		t.Errorf("Expected:  loadStrategyDispatcher structure,received: %s", utils.ToJSON(rply))
+
+	}
+}

utils/coreutils.go

@@ -940,3 +940,9 @@ type SetIndexesArg struct {
 	TenantArg
 	Opts map[string]interface{}
 }
+
+type DurationArgs struct {
+	DurationTime time.Duration
+	Opts         map[string]interface{}
+	*TenantArg
+}