Tpe dispatchers + tests

apis/tpes_it_test.go

@@ -60,6 +60,7 @@ var (
 		testTPeSetStatQueueProfile,
 		testTPeSetActions,
 		testTPeSetThresholds,
+		testTPeSetDispatcherProfiles,
 		testTPeSExportTariffPlanHalfTariffPlan,
 		testTPeSExportTariffPlanAllTariffPlan,
 		// export again after we will flush the database
@@ -929,22 +930,78 @@ func testTPeSetThresholds(t *testing.T) {
 	}
 }
 
+func testTPeSetDispatcherProfiles(t *testing.T) {
+	dspPrf := &DispatcherWithAPIOpts{
+		DispatcherProfile: &engine.DispatcherProfile{
+			Tenant:    "cgrates.org",
+			ID:        "Dsp1",
+			FilterIDs: []string{"*string:~*req.Account:1001", "*ai:~*req.AnswerTime:2014-07-14T14:25:00Z"},
+			Strategy:  utils.MetaFirst,
+			StrategyParams: map[string]interface{}{
+				utils.MetaDefaultRatio: "false",
+			},
+			Weight: 20,
+			Hosts: engine.DispatcherHostProfiles{
+				{
+					ID:        "C1",
+					FilterIDs: []string{},
+					Weight:    10,
+					Params:    map[string]interface{}{"0": "192.168.54.203"},
+					Blocker:   false,
+				},
+			},
+		},
+	}
+
+	var result string
+	if err := tpeSRPC.Call(context.Background(), utils.AdminSv1SetDispatcherProfile, dspPrf, &result); err != nil {
+		t.Fatal(err)
+	} else if result != utils.OK {
+		t.Error("Unexpected reply returned", result)
+	}
+
+	dspPrf2 := &DispatcherWithAPIOpts{
+		DispatcherProfile: &engine.DispatcherProfile{
+			Tenant:    "cgrates.org",
+			ID:        "Dsp2",
+			FilterIDs: []string{"*string:~*opts.EventType:LoadDispatcher"},
+			Strategy:  utils.MetaWeight,
+			Weight:    10,
+			Hosts: engine.DispatcherHostProfiles{
+				{
+					ID:        "Conn2",
+					FilterIDs: []string{"*suffix:~*opts.*answerTime:45T"},
+					Params:    map[string]interface{}{utils.MetaRatio: 1},
+					Blocker:   false,
+				},
+			},
+		},
+	}
+
+	if err := tpeSRPC.Call(context.Background(), utils.AdminSv1SetDispatcherProfile, dspPrf2, &result); err != nil {
+		t.Fatal(err)
+	} else if result != utils.OK {
+		t.Error("Unexpected reply returned", result)
+	}
+}
+
 func testTPeSExportTariffPlanHalfTariffPlan(t *testing.T) {
 	var replyBts []byte
 	// we will get only the wantes tariff plans in the csv format
 	if err := tpeSRPC.Call(context.Background(), utils.TPeSv1ExportTariffPlan, &tpes.ArgsExportTP{
 		Tenant: "cgrates.org",
 		ExportItems: map[string][]string{
-			utils.MetaAttributes: {"TEST_ATTRIBUTES_IT_TEST"},
-			utils.MetaResources:  {"ResGroup1"},
-			utils.MetaFilters:    {"fltr_for_prf"},
-			utils.MetaRateS:      {"MultipleRates"},
-			utils.MetaChargers:   {"Chargers1"},
-			utils.MetaRoutes:     {"ROUTE_2003"},
-			utils.MetaAccounts:   {"Account_balances"},
-			utils.MetaStats:      {"SQ_basic"},
-			utils.MetaActions:    {"Execute_thd"},
-			utils.MetaThresholds: {"TH_Stats1"},
+			utils.MetaAttributes:  {"TEST_ATTRIBUTES_IT_TEST"},
+			utils.MetaResources:   {"ResGroup1"},
+			utils.MetaFilters:     {"fltr_for_prf"},
+			utils.MetaRateS:       {"MultipleRates"},
+			utils.MetaChargers:    {"Chargers1"},
+			utils.MetaRoutes:      {"ROUTE_2003"},
+			utils.MetaAccounts:    {"Account_balances"},
+			utils.MetaStats:       {"SQ_basic"},
+			utils.MetaActions:     {"Execute_thd"},
+			utils.MetaThresholds:  {"TH_Stats1"},
+			utils.MetaDispatchers: {"Dsp1"},
 		},
 	}, &replyBts); err != nil {
 		t.Error(err)
@@ -961,7 +1018,7 @@ func testTPeSExportTariffPlanHalfTariffPlan(t *testing.T) {
 			t.Fatal(err)
 		}
 		info := csv.NewReader(rc)
-		info.FieldsPerRecord = -1
+		//info.FieldsPerRecord = -1
 		csvFile, err := info.ReadAll()
 		if err != nil {
 			t.Error(err)
@@ -1030,6 +1087,10 @@ func testTPeSExportTariffPlanHalfTariffPlan(t *testing.T) {
 			{"cgrates.org", "TH_Stats1", "*ai:~*req.AnswerTime:2014-07-14T14:35:00Z|2014-07-14T14:36:00Z", "", "0", "0", "", "false", "", "false"},
 			{"cgrates.org", "TH_Stats1", "*string:~*req.Destination:1011", "", "0", "0", "", "false", "", "false"},
 		},
+		utils.DispatcherProfilesCsv: {
+			{"#Tenant", "ID", "FilterIDs", "Weight", "Strategy", "StrategyParameters", "ConnID", "ConnFilterIDs", "ConnWeight", "ConnBlocker", "ConnParameters"},
+			{"cgrates.org", "Dsp1", "*string:~*req.Account:1001;*ai:~*req.AnswerTime:2014-07-14T14:25:00Z", "20", "*first", "false", "C1", "", "10", "false", "*ratio:2;192.168.54.203"},
+		},
 	}
 	expected[utils.RatesCsv] = csvRply[utils.RatesCsv]
 	expected[utils.AccountsCsv] = csvRply[utils.AccountsCsv]
@@ -1045,16 +1106,17 @@ func testTPeSExportTariffPlanAllTariffPlan(t *testing.T) {
 	if err := tpeSRPC.Call(context.Background(), utils.TPeSv1ExportTariffPlan, &tpes.ArgsExportTP{
 		Tenant: "cgrates.org",
 		ExportItems: map[string][]string{
-			utils.MetaAttributes: {"TEST_ATTRIBUTES_IT_TEST", "TEST_ATTRIBUTES_IT_TEST_SECOND"},
-			utils.MetaResources:  {"ResGroup1", "ResGroup2"},
-			utils.MetaFilters:    {"fltr_for_prf", "fltr_changed2"},
-			utils.MetaRateS:      {"MultipleRates", "TEST_RATE_IT_TEST"},
-			utils.MetaChargers:   {"Chargers1", "DifferentCharger"},
-			utils.MetaRoutes:     {"ROUTE_2003", "ROUTE_ACNT_1001"},
-			utils.MetaAccounts:   {"Account_balances", "Account_simple"},
-			utils.MetaStats:      {"SQ_basic", "SQ_2"},
-			utils.MetaActions:    {"Execute_thd", "SET_BAL"},
-			utils.MetaThresholds: {"TH_Stats1", "THD_2"},
+			utils.MetaAttributes:  {"TEST_ATTRIBUTES_IT_TEST", "TEST_ATTRIBUTES_IT_TEST_SECOND"},
+			utils.MetaResources:   {"ResGroup1", "ResGroup2"},
+			utils.MetaFilters:     {"fltr_for_prf", "fltr_changed2"},
+			utils.MetaRateS:       {"MultipleRates", "TEST_RATE_IT_TEST"},
+			utils.MetaChargers:    {"Chargers1", "DifferentCharger"},
+			utils.MetaRoutes:      {"ROUTE_2003", "ROUTE_ACNT_1001"},
+			utils.MetaAccounts:    {"Account_balances", "Account_simple"},
+			utils.MetaStats:       {"SQ_basic", "SQ_2"},
+			utils.MetaActions:     {"Execute_thd", "SET_BAL"},
+			utils.MetaThresholds:  {"TH_Stats1", "THD_2"},
+			utils.MetaDispatchers: {"Dsp1", "Dsp2"},
 		},
 	}, &replyBts); err != nil {
 		t.Error(err)
@@ -1071,7 +1133,7 @@ func testTPeSExportTariffPlanAllTariffPlan(t *testing.T) {
 			t.Fatal(err)
 		}
 		info := csv.NewReader(rc)
-		info.FieldsPerRecord = -1
+		//info.FieldsPerRecord = -1
 		csvFile, err := info.ReadAll()
 		if err != nil {
 			t.Error(err)
@@ -1159,6 +1221,11 @@ func testTPeSExportTariffPlanAllTariffPlan(t *testing.T) {
 			{"cgrates.org", "TH_Stats1", "*string:~*req.Destination:1011", "", "0", "0", "", "false", "", "false"},
 			{"cgrates.org", "THD_2", "*string:~*req.Account:1001", ";20", "7", "0", "", "false", "actPrfID", "true"},
 		},
+		utils.DispatcherProfilesCsv: {
+			{"#Tenant", "ID", "FilterIDs", "Weight", "Strategy", "StrategyParameters", "ConnID", "ConnFilterIDs", "ConnWeight", "ConnBlocker", "ConnParameters"},
+			{"cgrates.org", "Dsp1", "*string:~*req.Account:1001;*ai:~*req.AnswerTime:2014-07-14T14:25:00Z", "20", "*first", "false", "C1", "", "10", "false", "*ratio:2;192.168.54.203"},
+			{"cgrates.org", "Dsp2", "*string:~*opts.EventType:LoadDispatcher", "10", "*weight", "", "Conn2", "*suffix:~*opts.*answerTime:45T", "0", "false", "*ratio:1"},
+		},
 	}
 	expected[utils.RatesCsv] = csvRply[utils.RatesCsv]
 	expected[utils.AccountsCsv] = csvRply[utils.AccountsCsv]
@@ -1221,7 +1288,6 @@ func testTPeSExportAfterFlush(t *testing.T) {
 			t.Fatal(err)
 		}
 		info := csv.NewReader(rc)
-		info.FieldsPerRecord = -1
 		csvFile, err := info.ReadAll()
 		if err != nil {
 			t.Error(err)
@@ -1231,7 +1297,7 @@ func testTPeSExportAfterFlush(t *testing.T) {
 	}
 	// empty exporters, nothing in database to export
 	if len(csvRply) != 0 {
-		t.Errorf("Unexpected length, expected to be 0, no exports were nedeed")
+		t.Errorf("Unexpected length, expected to be 0, no exports were nedeed and got zip containing: %v", utils.ToJSON(csvRply))
 	}
 }
 

tpes/tpe_dispatchers.go

@@ -0,0 +1,80 @@
+/*
+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 tpes
+
+import (
+	"encoding/csv"
+	"fmt"
+	"io"
+
+	"github.com/cgrates/birpc/context"
+
+	"github.com/cgrates/cgrates/engine"
+	"github.com/cgrates/cgrates/utils"
+)
+
+type TPDispatchers struct {
+	dm *engine.DataManager
+}
+
+// newTPDispatchers is the constructor for TPDispatchers
+func newTPDispatchers(dm *engine.DataManager) *TPDispatchers {
+	return &TPDispatchers{
+		dm: dm,
+	}
+}
+
+// exportItems for TPDispatchers will implement the method for tpExporter interface
+func (tpDsp TPDispatchers) exportItems(ctx *context.Context, wrtr io.Writer, tnt string, itmIDs []string) (err error) {
+	csvWriter := csv.NewWriter(wrtr)
+	csvWriter.Comma = utils.CSVSep
+	// before writing the profiles, we must write the headers
+	if err = csvWriter.Write([]string{"#Tenant", "ID", "FilterIDs", "Weight", "Strategy", "StrategyParameters", "ConnID", "ConnFilterIDs", "ConnWeight", "ConnBlocker", "ConnParameters"}); err != nil {
+		return
+	}
+	for _, dspID := range itmIDs {
+		var dspPrf *engine.DispatcherProfile
+		dspPrf, err = tpDsp.dm.GetDispatcherProfile(ctx, tnt, dspID, true, true, utils.NonTransactional)
+		if err != nil {
+			if err.Error() == utils.ErrNotFound.Error() {
+				return fmt.Errorf("<%s> cannot find DispatcherProfile with id: <%v>", err, dspID)
+			}
+			return err
+		}
+		dspMdls := engine.APItoModelTPDispatcherProfile(engine.DispatcherProfileToAPI(dspPrf))
+		if len(dspMdls) == 0 {
+			return
+		}
+		// for every profile, convert it into model to be compatible in csv format
+		for _, tpItem := range dspMdls {
+			// transform every record into a []string
+			var record []string
+			record, err = engine.CsvDump(tpItem)
+			if err != nil {
+				return err
+			}
+			// record is a line of a csv file
+			if err = csvWriter.Write(record); err != nil {
+				return err
+			}
+		}
+	}
+	csvWriter.Flush()
+	return
+}

tpes/tpes.go

@@ -87,6 +87,10 @@ func getTariffPlansKeys(ctx *context.Context, dm *engine.DataManager, tnt, expTy
 		prfx = utils.StatQueueProfilePrefix + tnt + utils.ConcatenatedKeySep
 	case utils.MetaThresholds:
 		prfx = utils.ThresholdProfilePrefix + tnt + utils.ConcatenatedKeySep
+	case utils.MetaDispatchers:
+		prfx = utils.DispatcherProfilePrefix + tnt + utils.ConcatenatedKeySep
+	default:
+		return nil, fmt.Errorf("Unsuported exporter type")
 	}
 	// dbKeys will contain the full name of the key, but we will need just the IDs e.g. "alp_cgrates.org:ATTR_1" -- just ATTR_1
 	var dbKeys []string

tpes/tpexporter.go

@@ -37,8 +37,8 @@ var tpExporterTypes = utils.NewStringSet([]string{
 	utils.MetaStats,
 	utils.MetaActions,
 	utils.MetaThresholds,
+	utils.MetaDispatchers,
 	/*
-		utils.MetaDispatchers, //
 		utils.MetaDispatcherHosts, //
 	*/})
 
@@ -85,6 +85,8 @@ func newTPExporter(expType string, dm *engine.DataManager) (tpE tpExporter, err
 		return newTPActions(dm), nil
 	case utils.MetaThresholds:
 		return newTPThresholds(dm), nil
+	case utils.MetaDispatchers:
+		return newTPDispatchers(dm), nil
 	default:
 		return nil, utils.ErrPrefix(utils.ErrUnsupportedTPExporterType, expType)
 	}