diff --git a/data/conf/samples/cdrsv_failover_internal/cgrates.json b/data/conf/samples/cdrsv_failover_internal/cgrates.json new file mode 100644 index 000000000..cfe8d9e30 --- /dev/null +++ b/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: +}, + + +"rals": { + "enabled": true, // enable Rater service: + "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"}, + ], + }, +}, + +} diff --git a/data/conf/samples/cdrsv_failover_mongo/cgrates.json b/data/conf/samples/cdrsv_failover_mongo/cgrates.json new file mode 100644 index 000000000..2dd9551b0 --- /dev/null +++ b/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: + "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"}, + ], + }, +}, + +} diff --git a/data/conf/samples/cdrsv_failover_mysql/cgrates.json b/data/conf/samples/cdrsv_failover_mysql/cgrates.json new file mode 100644 index 000000000..1a7b6576f --- /dev/null +++ b/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: + "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: + "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"}, + ], + }, +}, + +} diff --git a/engine/poster.go b/engine/poster.go index d7a5d1f3a..887bd1a77 100644 --- a/engine/poster.go +++ b/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 { diff --git a/general_tests/cdrs_onlexp_it_test.go b/general_tests/cdrs_onlexp_it_test.go index ccf253a65..a1bf711b2 100644 --- a/general_tests/cdrs_onlexp_it_test.go +++ b/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() } } diff --git a/general_tests/cdrs_post_failover_it_test.go b/general_tests/cdrs_post_failover_it_test.go new file mode 100644 index 000000000..98991cabe --- /dev/null +++ b/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 +*/ + +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) + } +}