Added failover test with multiple events in a single file

data/conf/samples/cdrsv_failover_internal/cgrates.json

@@ -0,0 +1,72 @@
+{
+// CGRateS Configuration file
+//
+// Used in apier_local_tests
+// Starts rater, cdrs and mediator connecting over internal channel
+
+"general": {
+    "log_level": 7,
+	"poster_attempts": 1,									// number of attempts before considering post request failed (eg: *http_post, CDR exports)
+	"failed_posts_ttl": "1s",				// time to wait before writing the failed posts in a single file
+},
+
+"stor_db": {
+	"db_type": "*internal",					// stor database type to use: <mysql|postgres>
+},
+
+
+"rals": {
+	"enabled": true,						// enable Rater service: <true|false>
+	"thresholds_conns": ["*localhost"],
+},
+
+"schedulers": {
+	"enabled": true,
+},
+
+"cdrs": {
+	"enabled": true,
+	"attributes_conns":["*internal"],
+	"chargers_conns":["*localhost"],
+	"rals_conns": ["*localhost"],
+	"stats_conns": ["*localhost"],
+	"thresholds_conns": ["*localhost"],
+	"online_cdr_exports": ["s3_test_file"],
+},
+
+"attributes": {
+	"enabled": true,
+},
+
+"stats": {
+	"enabled": true,
+	"store_interval": "1s",
+	"thresholds_conns": ["*internal"],
+},
+
+"thresholds": {
+	"enabled": true,
+	"store_interval": "1s",
+},
+
+"chargers": {
+	"enabled": true,
+	"attributes_conns": ["*internal"],
+},
+
+"apier": {
+	"scheduler_conns": ["*internal"],
+},
+
+"cdre": {
+	"s3_test_file": {
+		"export_format": "*s3_json_map",
+		// export_path for s3:  "endpoint?aws_region=region&aws_key=IDkey&aws_secret=secret&aws_token=sessionToken&queue_id=cgrates-cdrs"  
+		"export_path": "http://s3.us-east-2.amazonaws.com/?aws_region=eu-west-2&aws_key=testkey&aws_secret=testsecret&queue_id=cgrates-cdrs",
+		"content_fields": [
+			{"tag": "CGRID", "type": "*composed", "value": "~CGRID", "field_id": "CGRID"},
+		],
+	},
+},
+
+}

data/conf/samples/cdrsv_failover_mongo/cgrates.json

@@ -0,0 +1,81 @@
+{
+// CGRateS Configuration file
+//
+// Used in apier_local_tests
+// Starts rater, cdrs and mediator connecting over internal channel
+
+"general": {
+    "log_level": 7,
+	"poster_attempts": 1,									// number of attempts before considering post request failed (eg: *http_post, CDR exports)
+	"failed_posts_ttl": "1s",				// time to wait before writing the failed posts in a single file
+},
+
+"data_db": {
+	"db_type": "mongo",
+	"db_name": "10",
+	"db_port": 27017,
+},
+
+
+"stor_db": {
+	"db_type": "mongo",
+	"db_name": "cgrates",
+	"db_port": 27017,
+},
+
+
+"rals": {
+	"enabled": true,						// enable Rater service: <true|false>
+	"thresholds_conns": ["*localhost"],
+},
+
+"schedulers": {
+	"enabled": true,
+},
+
+"cdrs": {
+	"enabled": true,
+	"attributes_conns":["*internal"],
+	"chargers_conns":["*localhost"],
+	"rals_conns": ["*localhost"],
+	"stats_conns": ["*localhost"],
+	"thresholds_conns": ["*localhost"],
+	"online_cdr_exports": ["s3_test_file"],
+},
+
+"attributes": {
+	"enabled": true,
+},
+
+"stats": {
+	"enabled": true,
+	"store_interval": "1s",
+	"thresholds_conns": ["*internal"],
+},
+
+"thresholds": {
+	"enabled": true,
+	"store_interval": "1s",
+},
+
+"chargers": {
+	"enabled": true,
+	"attributes_conns": ["*internal"],
+},
+
+"apier": {
+	"scheduler_conns": ["*internal"],
+},
+
+"cdre": {
+	"s3_test_file": {
+		"export_format": "*s3_json_map",
+		// export_path for s3:  "endpoint?aws_region=region&aws_key=IDkey&aws_secret=secret&aws_token=sessionToken&queue_id=cgrates-cdrs"  
+		"export_path": "http://s3.us-east-2.amazonaws.com/?aws_region=eu-west-2&aws_key=testkey&aws_secret=testsecret&queue_id=cgrates-cdrs",
+		"content_fields": [
+			{"tag": "CGRID", "type": "*composed", "value": "~CGRID", "field_id": "CGRID"},
+		],
+	},
+},
+
+}

data/conf/samples/cdrsv_failover_mysql/cgrates.json

@@ -0,0 +1,79 @@
+{
+// CGRateS Configuration file
+//
+// Used in apier_local_tests
+// Starts rater, cdrs and mediator connecting over internal channel
+
+"general": {
+    "log_level": 7,
+	"poster_attempts": 1,									// number of attempts before considering post request failed (eg: *http_post, CDR exports)
+	"failed_posts_ttl": "1s",				// time to wait before writing the failed posts in a single file
+},
+
+
+"data_db": {								// database used to store runtime data (eg: accounts, cdr stats)
+	"db_type": "redis",						// data_db type: <redis|mongo>
+	"db_port": 6379, 						// data_db port to reach the database
+	"db_name": "10", 						// data_db database name to connect to
+},
+
+"stor_db": {
+	"db_password": "CGRateS.org",
+},
+
+
+"rals": {
+	"enabled": true,						// enable Rater service: <true|false>
+	"thresholds_conns": ["*localhost"],
+},
+
+"schedulers": {
+	"enabled": true,
+},
+
+"cdrs": {
+	"enabled": true,
+	"attributes_conns":["*internal"],
+	"chargers_conns":["*localhost"],
+	"rals_conns": ["*localhost"],
+	"stats_conns": ["*localhost"],
+	"thresholds_conns": ["*localhost"],
+	"online_cdr_exports": ["s3_test_file"],
+},
+
+"attributes": {
+	"enabled": true,
+},
+
+"stats": {
+	"enabled": true,
+	"store_interval": "1s",
+	"thresholds_conns": ["*internal"],
+},
+
+"thresholds": {
+	"enabled": true,
+	"store_interval": "1s",
+},
+
+"chargers": {
+	"enabled": true,
+	"attributes_conns": ["*internal"],
+},
+
+"apier": {
+	"scheduler_conns": ["*internal"],
+},
+
+"cdre": {
+	"s3_test_file": {
+		"export_format": "*s3_json_map",
+		// export_path for s3:  "endpoint?aws_region=region&aws_key=IDkey&aws_secret=secret&aws_token=sessionToken&queue_id=cgrates-cdrs"  
+		"export_path": "http://s3.us-east-2.amazonaws.com/?aws_region=eu-west-2&aws_key=testkey&aws_secret=testsecret&queue_id=cgrates-cdrs",
+		"content_fields": [
+			{"tag": "CGRID", "type": "*composed", "value": "~CGRID", "field_id": "CGRID"},
+		],
+	},
+},
+
+}

engine/poster.go

@@ -20,14 +20,8 @@ package engine
 
 import (
 	"net/url"
-	"os"
-	"path"
 	"strings"
 	"sync"
-
-	"github.com/cgrates/cgrates/config"
-	"github.com/cgrates/cgrates/guardian"
-	"github.com/cgrates/cgrates/utils"
 )
 
 const (
@@ -68,20 +62,6 @@ type Poster interface {
 	Close()
 }
 
-func writeToFile(fileDir, fileName string, content []byte) (err error) {
-	fallbackFilePath := path.Join(fileDir, fileName)
-	_, err = guardian.Guardian.Guard(func() (interface{}, error) {
-		fileOut, err := os.Create(fallbackFilePath)
-		if err != nil {
-			return nil, err
-		}
-		_, err = fileOut.Write(content)
-		fileOut.Close()
-		return nil, err
-	}, config.CgrConfig().GeneralCfg().LockingTimeout, utils.FileLockPrefix+fallbackFilePath)
-	return
-}
-
 func parseURL(dialURL string) (URL string, qID string, err error) {
 	u, err := url.Parse(dialURL)
 	if err != nil {

general_tests/cdrs_onlexp_it_test.go

@@ -467,8 +467,9 @@ func checkContent(ev *engine.ExportEvents, content []interface{}) error {
 	}
 	return nil
 }
+
 func testCDRsOnExpFileFailover(t *testing.T) {
-	time.Sleep(time.Duration(20 * time.Second))
+	time.Sleep(time.Duration(5 * time.Second))
 	v1 := url.Values{}
 	v2 := url.Values{}
 	v1.Set("OriginID", "httpjsonrpc1")
@@ -518,12 +519,13 @@ func testCDRsOnExpKafkaPosterFileFailover(t *testing.T) {
 	defer reader.Close()
 
 	for i := 0; i < 2; i++ { // no raw CDR
-		ctx, _ := context.WithTimeout(context.Background(), 10*time.Second)
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		if m, err := reader.ReadMessage(ctx); err != nil {
 			t.Fatal(err)
 		} else if !reflect.DeepEqual(failoverContent[0].([]byte), m.Value) && !reflect.DeepEqual(failoverContent[1].([]byte), m.Value) { // Checking just the prefix should do since some content is dynamic
 			t.Errorf("Expecting: %v or %v, received: %v", utils.IfaceAsString(failoverContent[0]), utils.IfaceAsString(failoverContent[1]), string(m.Value))
 		}
+		cancel()
 	}
 }
 

general_tests/cdrs_post_failover_it_test.go

@@ -0,0 +1,210 @@
+// +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 general_tests
+
+import (
+	"io/ioutil"
+	"net/rpc"
+	"path"
+	"testing"
+	"time"
+
+	"github.com/cgrates/cgrates/config"
+	"github.com/cgrates/cgrates/engine"
+	"github.com/cgrates/cgrates/utils"
+)
+
+var (
+	cdrsPostFailCfgPath string
+	cdrsPostFailCfg     *config.CGRConfig
+	cdrsPostFailRpc     *rpc.Client
+	cdrsPostFailConfDIR string // run the tests for specific configuration
+
+	// subtests to be executed for each confDIR
+	sTestsCDRsPostFailIT = []func(t *testing.T){
+		testCDRsPostFailoverInitConfig,
+		testCDRsPostFailoverInitDataDb,
+		testCDRsPostFailoverInitCdrDb,
+		testCDRsPostFailoverStartEngine,
+		testCDRsPostFailoverRpcConn,
+		testCDRsPostFailoverLoadTariffPlanFromFolder,
+		testCDRsPostFailoverProcessCDR,
+		testCDRsPostFailoverToFile,
+		testCDRsPostFailoverKillEngine,
+	}
+)
+
+// Tests starting here
+func TestCDRsPostFailoverIT(t *testing.T) {
+	switch *dbType {
+	case utils.MetaInternal:
+		cdrsPostFailConfDIR = "cdrsv_failover_internal"
+	case utils.MetaSQL:
+		cdrsPostFailConfDIR = "cdrsv_failover_mysql"
+	case utils.MetaMongo:
+		cdrsPostFailConfDIR = "cdrsv_failover_mongo"
+	case utils.MetaPostgres:
+		t.SkipNow()
+	default:
+		t.Fatal("Unknown Database type")
+	}
+
+	for _, stest := range sTestsCDRsPostFailIT {
+		t.Run(cdrsPostFailConfDIR, stest)
+	}
+}
+
+func testCDRsPostFailoverInitConfig(t *testing.T) {
+	var err error
+	cdrsPostFailCfgPath = path.Join(*dataDir, "conf", "samples", cdrsPostFailConfDIR)
+	if cdrsPostFailCfg, err = config.NewCGRConfigFromPath(cdrsPostFailCfgPath); err != nil {
+		t.Fatal("Got config error: ", err.Error())
+	}
+}
+
+func testCDRsPostFailoverInitDataDb(t *testing.T) {
+	if err := engine.InitDataDb(cdrsPostFailCfg); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// InitDb so we can rely on count
+func testCDRsPostFailoverInitCdrDb(t *testing.T) {
+	if err := engine.InitStorDb(cdrsPostFailCfg); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func testCDRsPostFailoverStartEngine(t *testing.T) {
+	if _, err := engine.StopStartEngine(cdrsPostFailCfgPath, *waitRater); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Connect rpc client to rater
+func testCDRsPostFailoverRpcConn(t *testing.T) {
+	cdrsPostFailRpc, err = newRPCClient(cdrsPostFailCfg.ListenCfg()) // We connect over JSON so we can also troubleshoot if needed
+	if err != nil {
+		t.Fatal("Could not connect to rater: ", err.Error())
+	}
+}
+
+func testCDRsPostFailoverLoadTariffPlanFromFolder(t *testing.T) {
+	var loadInst utils.LoadInstance
+	if err := cdrsPostFailRpc.Call(utils.ApierV2LoadTariffPlanFromFolder,
+		&utils.AttrLoadTpFromFolder{FolderPath: path.Join(
+			*dataDir, "tariffplans", "testit")}, &loadInst); err != nil {
+		t.Error(err)
+	}
+	time.Sleep(time.Duration(*waitRater) * time.Millisecond) // Give time for scheduler to execute topups
+	var resp string
+	if err := cdrsPostFailRpc.Call(utils.ApierV1RemoveChargerProfile,
+		&utils.TenantID{Tenant: "cgrates.org", ID: "SupplierCharges"}, &resp); err != nil {
+		t.Error(err)
+	} else if resp != utils.OK {
+		t.Error("Unexpected reply returned", resp)
+	}
+	var reply *engine.ChargerProfile
+	if err := cdrsPostFailRpc.Call(utils.ApierV1GetChargerProfile,
+		&utils.TenantID{Tenant: "cgrates.org", ID: "SupplierCharges"},
+		&reply); err == nil || err.Error() != utils.ErrNotFound.Error() {
+		t.Error(err)
+	}
+}
+
+func testCDRsPostFailoverProcessCDR(t *testing.T) {
+	args := &engine.ArgV1ProcessEvent{
+		Flags: []string{utils.MetaExport, "*attributes:false", "*rals:false", "*chargers:*false",
+			"*store:false", "*thresholds:false", "*stats:false"}, // only export the CDR
+		CGREvent: utils.CGREvent{
+			ID:     "1",
+			Tenant: "cgrates.org",
+			Event: map[string]interface{}{
+				utils.OriginID:    "testCDRsPostFailoverProcessCDR",
+				utils.OriginHost:  "192.168.1.1",
+				utils.Source:      "testCDRsPostFailoverProcessCDR",
+				utils.RequestType: utils.META_RATED,
+				utils.Category:    "call",
+				utils.Account:     "testCDRsPostFailoverProcessCDR",
+				utils.Subject:     "ANY2CNT",
+				utils.Destination: "+4986517174963",
+				utils.AnswerTime:  time.Date(2018, 8, 24, 16, 00, 26, 0, time.UTC),
+				utils.Usage:       time.Duration(1) * time.Minute,
+				"field_extr1":     "val_extr1",
+				"fieldextr2":      "valextr2",
+			},
+		},
+	}
+
+	var reply string
+	if err := cdrsPostFailRpc.Call(utils.CDRsV1ProcessEvent, args, &reply); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply received: ", reply)
+	}
+	args.ID = "2"
+	args.Event[utils.OriginID] = "2"
+	if err := cdrsPostFailRpc.Call(utils.CDRsV1ProcessEvent, args, &reply); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply received: ", reply)
+	}
+	args.ID = "3"
+	args.Event[utils.OriginID] = "3"
+	if err := cdrsPostFailRpc.Call(utils.CDRsV1ProcessEvent, args, &reply); err != nil {
+		t.Error("Unexpected error: ", err.Error())
+	} else if reply != utils.OK {
+		t.Error("Unexpected reply received: ", reply)
+	}
+}
+
+func testCDRsPostFailoverToFile(t *testing.T) {
+	time.Sleep(time.Duration(2 * time.Second))
+	filesInDir, _ := ioutil.ReadDir(cdrsPostFailCfg.GeneralCfg().FailedPostsDir)
+	if len(filesInDir) == 0 {
+		t.Fatalf("No files in directory: %s", cdrsPostFailCfg.GeneralCfg().FailedPostsDir)
+	}
+	for _, file := range filesInDir { // First file in directory is the one we need, harder to find it's name out of config
+		fileName := file.Name()
+		filePath := path.Join(cdrsPostFailCfg.GeneralCfg().FailedPostsDir, fileName)
+
+		ev, err := engine.NewExportEventsFromFile(filePath)
+		if err != nil {
+			t.Errorf("<%s> for file <%s>", err, fileName)
+			continue
+		} else if len(ev.Events) == 0 {
+			t.Error("Expected at least one event")
+			continue
+		}
+		if ev.Format != utils.MetaS3jsonMap {
+			t.Errorf("Expected event to use %q received: %q", utils.MetaS3jsonMap, ev.Format)
+		}
+		if len(ev.Events) != 3 {
+			t.Errorf("Expected all the events to be saved in the same file, ony %v saved in this file.", len(ev.Events))
+		}
+	}
+}
+
+func testCDRsPostFailoverKillEngine(t *testing.T) {
+	if err := engine.KillEngine(*waitRater); err != nil {
+		t.Error(err)
+	}
+}