diff --git a/config/config.go b/config/config.go index 1fae0c530..e6c8fb83d 100755 --- a/config/config.go +++ b/config/config.go @@ -305,7 +305,7 @@ var posibleLoaderTypes = utils.NewStringSet([]string{utils.MetaAttributes, var possibleReaderTypes = utils.NewStringSet([]string{utils.MetaFileCSV, utils.MetaKafkajsonMap, utils.MetaFileXML, utils.MetaSQL, utils.MetaFileFWV, - utils.MetaPartialCSV}) + utils.MetaPartialCSV, utils.MetaOsipsFlatstore, utils.MetaKamFlatstore}) func (cfg *CGRConfig) LazySanityCheck() { for _, cdrePrfl := range cfg.cdrsCfg.OnlineCDRExports { diff --git a/config/configsanity.go b/config/configsanity.go index 7bbe2e6d0..b59dd7eea 100644 --- a/config/configsanity.go +++ b/config/configsanity.go @@ -421,7 +421,7 @@ func (cfg *CGRConfig) checkConfigSanity() error { } switch rdr.Type { - case utils.MetaFileCSV, utils.MetaPartialCSV: + case utils.MetaFileCSV, utils.MetaPartialCSV, utils.MetaOsipsFlatstore, utils.MetaKamFlatstore: for _, dir := range []string{rdr.ProcessedPath, rdr.SourcePath} { if _, err := os.Stat(dir); err != nil && os.IsNotExist(err) { return fmt.Errorf("<%s> Nonexistent folder: %s for reader with ID: %s", utils.ERs, dir, rdr.ID) diff --git a/data/conf/samples/ers_internal/cgrates.json b/data/conf/samples/ers_internal/cgrates.json index 8de6afea6..681262943 100644 --- a/data/conf/samples/ers_internal/cgrates.json +++ b/data/conf/samples/ers_internal/cgrates.json @@ -330,10 +330,12 @@ "enabled": true, "run_delay": -1, "type": "*opensips_flatstore", + "field_separator": "|", "source_path": "/tmp/flatstoreErs/in", "processed_path": "/tmp/flatstoreErs/out", + "failed_calls_prefix": "missed_calls", "flags": ["*cdrs"], - "partial_record_cache": "1s", + "partial_record_cache": "2s", "content_fields":[ {"tag": "Tor", "field_id": "ToR", "type": "*constant", "value": "*voice", "mandatory": true}, {"tag": "OriginID", "field_id": "OriginID", "type": "*composed","value":"~*req.3;~*req.1;~*req.2", "mandatory": true}, @@ -345,7 +347,7 @@ {"tag": "Destination", "field_id": "Destination", "type": "*variable", "value": "~*req.9", "mandatory": true}, {"tag": "SetupTime", "field_id": "SetupTime", "type": "*variable", "value": "~*req.6", "mandatory": true}, {"tag": "AnswerTime", "field_id": "AnswerTime", "type": "*variable", "value": "~*req.6", "mandatory": true}, - {"tag": "Usage", "field_id": "Usage", "type": "*none", "mandatory": true}, // Value for Usage is composed based on record + {"tag": "Usage", "field_id": "Usage", "type": "*variable", "mandatory": true}, // Value for Usage is composed based on record {"tag": "DisconnectCause", "field_id": "DisconnectCause", "type": "*composed", "value": "~*req.4; ;~*req.5", "mandatory": true}, {"tag": "DialogId", "field_id": "DialogId", "type": "*variable", "value": "~*req.11"} ], diff --git a/data/conf/samples/ers_mongo/cgrates.json b/data/conf/samples/ers_mongo/cgrates.json index 1f7a59fb7..02bdda690 100644 --- a/data/conf/samples/ers_mongo/cgrates.json +++ b/data/conf/samples/ers_mongo/cgrates.json @@ -333,10 +333,12 @@ "enabled": true, "run_delay": -1, "type": "*opensips_flatstore", + "field_separator": "|", "source_path": "/tmp/flatstoreErs/in", "processed_path": "/tmp/flatstoreErs/out", + "failed_calls_prefix": "missed_calls", "flags": ["*cdrs"], - "partial_record_cache": "1s", + "partial_record_cache": "2s", "content_fields":[ {"tag": "Tor", "field_id": "ToR", "type": "*constant", "value": "*voice", "mandatory": true}, {"tag": "OriginID", "field_id": "OriginID", "type": "*composed","value":"~*req.3;~*req.1;~*req.2", "mandatory": true}, @@ -348,7 +350,7 @@ {"tag": "Destination", "field_id": "Destination", "type": "*variable", "value": "~*req.9", "mandatory": true}, {"tag": "SetupTime", "field_id": "SetupTime", "type": "*variable", "value": "~*req.6", "mandatory": true}, {"tag": "AnswerTime", "field_id": "AnswerTime", "type": "*variable", "value": "~*req.6", "mandatory": true}, - {"tag": "Usage", "field_id": "Usage", "type": "*none", "mandatory": true}, // Value for Usage is composed based on record + {"tag": "Usage", "field_id": "Usage", "type": "*variable", "mandatory": true}, // Value for Usage is composed based on record {"tag": "DisconnectCause", "field_id": "DisconnectCause", "type": "*composed", "value": "~*req.4; ;~*req.5", "mandatory": true}, {"tag": "DialogId", "field_id": "DialogId", "type": "*variable", "value": "~*req.11"} ], diff --git a/data/conf/samples/ers_mysql/cgrates.json b/data/conf/samples/ers_mysql/cgrates.json index c956a341f..06c103360 100644 --- a/data/conf/samples/ers_mysql/cgrates.json +++ b/data/conf/samples/ers_mysql/cgrates.json @@ -324,6 +324,33 @@ {"tag": "Usage", "field_id": "Usage", "type": "*variable", "value": "~*req.6:s/^(\\d+)$/${1}s/", "mandatory": true}, {"tag": "Partial", "field_id": "Partial", "type": "*constant", "value": "true", "filters":["*string:~*req.10:partial"]} ], + }, + { + "id": "FlatstoreOsips", + "enabled": true, + "run_delay": -1, + "type": "*opensips_flatstore", + "field_separator": "|", + "source_path": "/tmp/flatstoreErs/in", + "processed_path": "/tmp/flatstoreErs/out", + "failed_calls_prefix": "missed_calls", + "flags": ["*cdrs"], + "partial_record_cache": "2s", + "content_fields":[ + {"tag": "Tor", "field_id": "ToR", "type": "*constant", "value": "*voice", "mandatory": true}, + {"tag": "OriginID", "field_id": "OriginID", "type": "*composed","value":"~*req.3;~*req.1;~*req.2", "mandatory": true}, + {"tag": "RequestType", "field_id": "RequestType", "type": "*variable", "value": "~*req.7", "mandatory": true}, + {"tag": "Tenant", "field_id": "Tenant", "type": "*constant", "value": "cgrates.org", "mandatory": true}, + {"tag": "Category", "field_id": "Category", "type": "*constant", "value": "call", "mandatory": true}, + {"tag": "Account", "field_id": "Account", "type": "*variable", "value": "~*req.8", "mandatory": true}, + {"tag": "Subject", "field_id": "Subject", "type": "*variable", "value": "~*req.8", "mandatory": true}, + {"tag": "Destination", "field_id": "Destination", "type": "*variable", "value": "~*req.9", "mandatory": true}, + {"tag": "SetupTime", "field_id": "SetupTime", "type": "*variable", "value": "~*req.6", "mandatory": true}, + {"tag": "AnswerTime", "field_id": "AnswerTime", "type": "*variable", "value": "~*req.6", "mandatory": true}, + {"tag": "Usage", "field_id": "Usage", "type": "*variable", "mandatory": true}, // Value for Usage is composed based on record + {"tag": "DisconnectCause", "field_id": "DisconnectCause", "type": "*composed", "value": "~*req.4; ;~*req.5", "mandatory": true}, + {"tag": "DialogId", "field_id": "DialogId", "type": "*variable", "value": "~*req.11"} + ], } ], }, diff --git a/data/conf/samples/ers_postgres/cgrates.json b/data/conf/samples/ers_postgres/cgrates.json index c39bf3d69..54e85831e 100644 --- a/data/conf/samples/ers_postgres/cgrates.json +++ b/data/conf/samples/ers_postgres/cgrates.json @@ -327,10 +327,12 @@ "enabled": true, "run_delay": -1, "type": "*opensips_flatstore", + "field_separator": "|", "source_path": "/tmp/flatstoreErs/in", "processed_path": "/tmp/flatstoreErs/out", + "failed_calls_prefix": "missed_calls", "flags": ["*cdrs"], - "partial_record_cache": "1s", + "partial_record_cache": "2s", "content_fields":[ {"tag": "Tor", "field_id": "ToR", "type": "*constant", "value": "*voice", "mandatory": true}, {"tag": "OriginID", "field_id": "OriginID", "type": "*composed","value":"~*req.3;~*req.1;~*req.2", "mandatory": true}, @@ -342,7 +344,7 @@ {"tag": "Destination", "field_id": "Destination", "type": "*variable", "value": "~*req.9", "mandatory": true}, {"tag": "SetupTime", "field_id": "SetupTime", "type": "*variable", "value": "~*req.6", "mandatory": true}, {"tag": "AnswerTime", "field_id": "AnswerTime", "type": "*variable", "value": "~*req.6", "mandatory": true}, - {"tag": "Usage", "field_id": "Usage", "type": "*none", "mandatory": true}, // Value for Usage is composed based on record + {"tag": "Usage", "field_id": "Usage", "type": "*variable", "mandatory": true}, // Value for Usage is composed based on record {"tag": "DisconnectCause", "field_id": "DisconnectCause", "type": "*composed", "value": "~*req.4; ;~*req.5", "mandatory": true}, {"tag": "DialogId", "field_id": "DialogId", "type": "*variable", "value": "~*req.11"} ], diff --git a/ers/flatstore.go b/ers/flatstore.go index 41827bae6..19781dd81 100644 --- a/ers/flatstore.go +++ b/ers/flatstore.go @@ -47,15 +47,18 @@ func NewFlatstoreER(cfg *config.CGRConfig, cfgIdx int, if strings.HasSuffix(srcPath, utils.Slash) { srcPath = srcPath[:len(srcPath)-1] } - return &FlatstoreER{ + flatER := &FlatstoreER{ cgrCfg: cfg, cfgIdx: cfgIdx, fltrS: fltrS, - cache: ltcache.NewCache(ltcache.UnlimitedCaching, cfg.ERsCfg().Readers[cfgIdx].PartialRecordCache, false, nil), rdrDir: srcPath, rdrEvents: rdrEvents, rdrError: rdrErr, - rdrExit: rdrExit}, nil + rdrExit: rdrExit, + } + flatER.cache = ltcache.NewCache(ltcache.UnlimitedCaching, cfg.ERsCfg().Readers[cfgIdx].PartialRecordCache, false, flatER.dumpToFile) + return flatER, err + } // FlatstoreER implements EventReader interface for Flatstore CDR @@ -147,27 +150,35 @@ func (rdr *FlatstoreER) processFile(fPath, fName string) (err error) { } return } - if strings.HasPrefix(file.Name(), rdr.Config().FailedCallsPrefix) { // Use the first index since they should be the same in all configs - record = append(record, "0") // Append duration 0 for failed calls flatstore CDR and do not process it further - } - pr, err := NewUnpairedRecord(record, rdr.Config().Timezone) - if err != nil { - fmt.Sprintf("<%s> Converting row : <%s> to unpairedRecord , ignoring due to error: <%s>", - utils.ERs, record, err.Error()) - continue - } - if val, has := rdr.cache.Get(pr.OriginID); !has { - rdr.cache.Set(pr.OriginID, pr, nil) - continue + if strings.HasPrefix(fName, rdr.Config().FailedCallsPrefix) { // Use the first index since they should be the same in all configs + record = append(record, "0") // Append duration 0 for failed calls flatstore CDR } else { - pair := val.(*UnpairedRecord) - record, err = pairToRecord(pair, pr) + pr, err := NewUnpairedRecord(record, rdr.Config().Timezone, fName) if err != nil { - fmt.Sprintf("<%s> Merging unpairedRecords : <%s> and <%s> to record , ignoring due to error: <%s>", - utils.ERs, utils.ToJSON(pair), utils.ToJSON(pr), err.Error()) + fmt.Sprintf("<%s> Converting row : <%s> to unpairedRecord , ignoring due to error: <%s>", + utils.ERs, record, err.Error()) continue } - rdr.cache.Remove(pr.OriginID) + if val, has := rdr.cache.Get(pr.OriginID); !has { + rdr.cache.Set(pr.OriginID, pr, nil) + continue + } else { + pair := val.(*UnpairedRecord) + record, err = pairToRecord(pair, pr) + if err != nil { + fmt.Sprintf("<%s> Merging unpairedRecords : <%s> and <%s> to record , ignoring due to error: <%s>", + utils.ERs, utils.ToJSON(pair), utils.ToJSON(pr), err.Error()) + continue + } + rdr.cache.Remove(pr.OriginID) + } + } + + // build Usage from contentFields based on record lenght + for i, cntFld := range rdr.Config().ContentFields { + if cntFld.FieldId == utils.Usage { + rdr.Config().ContentFields[i].Value = config.NewRSRParsersMustCompile("~*req."+strconv.Itoa(len(record)-1), true, utils.INFIELD_SEP) // in case of flatstore, last element will be the duration computed by us + } } rowNr++ // increment the rowNr after checking if it's not the end of file agReq := agents.NewAgentRequest( @@ -181,12 +192,6 @@ func (rdr *FlatstoreER) processFile(fPath, fName string) (err error) { agReq); err != nil || !pass { continue } - // build Usage from contentFields based on record lenght - for i, cntFld := range rdr.Config().ContentFields { - if cntFld.FieldId == utils.Usage { - rdr.Config().ContentFields[i].Value = config.NewRSRParsersMustCompile("~*req."+strconv.Itoa(len(record)-1), true, utils.INFIELD_SEP) // in case of flatstore, last element will be the duration computed by us - } - } navMp, err := agReq.AsNavigableMap(rdr.Config().ContentFields) if err != nil { utils.Logger.Warning( @@ -214,11 +219,11 @@ func (rdr *FlatstoreER) processFile(fPath, fName string) (err error) { return } -func NewUnpairedRecord(record []string, timezone string) (*UnpairedRecord, error) { +func NewUnpairedRecord(record []string, timezone string, fileName string) (*UnpairedRecord, error) { if len(record) < 7 { return nil, errors.New("MISSING_IE") } - pr := &UnpairedRecord{Method: record[0], OriginID: record[3] + record[1] + record[2], Values: record} + pr := &UnpairedRecord{Method: record[0], OriginID: record[3] + record[1] + record[2], Values: record, FileName: fileName} var err error if pr.Timestamp, err = utils.ParseTimeDetectLayout(record[6], timezone); err != nil { return nil, err @@ -232,6 +237,7 @@ type UnpairedRecord struct { OriginID string // Copute here the OriginID Timestamp time.Time // Timestamp of the event, as written by db_flastore module Values []string // Can contain original values or updated via UpdateValues + FileName string } // Pairs INVITE and BYE into final record containing as last element the duration @@ -271,3 +277,24 @@ func pairToRecord(part1, part2 *UnpairedRecord) ([]string, error) { record = append(record, strconv.FormatFloat(callDur.Seconds(), 'f', -1, 64)) return record, nil } + +func (rdr *FlatstoreER) dumpToFile(itmID string, value interface{}) { + unpRcd := value.(*UnpairedRecord) + + dumpFilePath := path.Join(rdr.Config().ProcessedPath, unpRcd.FileName+utils.TmpSuffix) + fileOut, err := os.Create(dumpFilePath) + if err != nil { + utils.Logger.Err(fmt.Sprintf("<%s> Failed creating %s, error: %s", + utils.ERs, dumpFilePath, err.Error())) + return + } + csvWriter := csv.NewWriter(fileOut) + csvWriter.Comma = rune(rdr.Config().FieldSep[0]) + if err = csvWriter.Write(unpRcd.Values); err != nil { + utils.Logger.Err(fmt.Sprintf("<%s> Failed writing partial record %v to file: %s, error: %s", + utils.ERs, unpRcd.Values, dumpFilePath, err.Error())) + return + } + + csvWriter.Flush() +} diff --git a/ers/flatstore_it_test.go b/ers/flatstore_it_test.go index 15008088f..00a24530c 100644 --- a/ers/flatstore_it_test.go +++ b/ers/flatstore_it_test.go @@ -1 +1,235 @@ +// +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 ers + +import ( + "io/ioutil" + "net/rpc" + "os" + "path" + "testing" + "time" + + v1 "github.com/cgrates/cgrates/apier/v1" + + "github.com/cgrates/cgrates/config" + "github.com/cgrates/cgrates/engine" + "github.com/cgrates/cgrates/utils" +) + +var ( + flatstoreCfgPath string + flatstoreCfgDIR string + flatstoreCfg *config.CGRConfig + flatstoreRPC *rpc.Client + + fullSuccessfull = `INVITE|2daec40c|548625ac|dd0c4c617a9919d29a6175cdff223a9e@0:0:0:0:0:0:0:0|200|OK|1436454408|*prepaid|1001|1002||3401:2069362475 +BYE|2daec40c|548625ac|dd0c4c617a9919d29a6175cdff223a9e@0:0:0:0:0:0:0:0|200|OK|1436454410|||||3401:2069362475 +INVITE|f9d3d5c3|c863a6e3|214d8f52b566e33a9349b184e72a4cca@0:0:0:0:0:0:0:0|200|OK|1436454647|*postpaid|1002|1001||1877:893549741 +BYE|f9d3d5c3|c863a6e3|214d8f52b566e33a9349b184e72a4cca@0:0:0:0:0:0:0:0|200|OK|1436454651|||||1877:893549741 +INVITE|36e39a5|42d996f9|3a63321dd3b325eec688dc2aefb6ac2d@0:0:0:0:0:0:0:0|200|OK|1436454657|*prepaid|1001|1002||2407:1884881533 +BYE|36e39a5|42d996f9|3a63321dd3b325eec688dc2aefb6ac2d@0:0:0:0:0:0:0:0|200|OK|1436454661|||||2407:1884881533 +INVITE|3111f3c9|49ca4c42|a58ebaae40d08d6757d8424fb09c4c54@0:0:0:0:0:0:0:0|200|OK|1436454690|*prepaid|1001|1002||3099:1909036290 +BYE|3111f3c9|49ca4c42|a58ebaae40d08d6757d8424fb09c4c54@0:0:0:0:0:0:0:0|200|OK|1436454692|||||3099:1909036290` + + fullMissed = `INVITE|ef6c6256|da501581|0bfdd176d1b93e7df3de5c6f4873ee04@0:0:0:0:0:0:0:0|487|Request Terminated|1436454643|*prepaid|1001|1002||1224:339382783 +INVITE|7905e511||81880da80a94bda81b425b09009e055c@0:0:0:0:0:0:0:0|404|Not Found|1436454668|*prepaid|1001|1002||1980:1216490844 +INVITE|324cb497|d4af7023|8deaadf2ae9a17809a391f05af31afb0@0:0:0:0:0:0:0:0|486|Busy here|1436454687|*postpaid|1002|1001||474:130115066` + + part1 = `BYE|f9d3d5c3|c863a6e3|214d8f52b566e33a9349b184e72a4ccb@0:0:0:0:0:0:0:0|200|OK|1436454651|||||1877:893549742` + + part2 = `INVITE|f9d3d5c3|c863a6e3|214d8f52b566e33a9349b184e72a4ccb@0:0:0:0:0:0:0:0|200|OK|1436454647|*postpaid|1002|1003||1877:893549742 +INVITE|2daec40c|548625ac|dd0c4c617a9919d29a6175cdff223a9p@0:0:0:0:0:0:0:0|200|OK|1436454408|*prepaid|1001|1002||3401:2069362475` + + flatstoreTests = []func(t *testing.T){ + testFlatstoreITCreateCdrDirs, + testFlatstoreITInitConfig, + testFlatstoreITInitCdrDb, + testFlatstoreITResetDataDb, + testFlatstoreITStartEngine, + testFlatstoreITRpcConn, + testFlatstoreITLoadTPFromFolder, + testFlatstoreITHandleCdr1File, + testFlatstoreITAnalyseCDRs, + testFlatstoreITCleanupFiles, + testFlatstoreITKillEngine, + } +) + +func TestFlatstoreFile(t *testing.T) { + switch *dbType { + case utils.MetaInternal: + flatstoreCfgDIR = "ers_internal" + case utils.MetaSQL: + flatstoreCfgDIR = "ers_mysql" + case utils.MetaMongo: + flatstoreCfgDIR = "ers_mongo" + case utils.MetaPostgres: + flatstoreCfgDIR = "ers_postgres" + default: + t.Fatal("Unknown Database type") + } + for _, test := range flatstoreTests { + t.Run(flatstoreCfgDIR, test) + } +} + +func testFlatstoreITCreateCdrDirs(t *testing.T) { + for _, dir := range []string{"/tmp/ers/in", "/tmp/ers/out", + "/tmp/ers2/in", "/tmp/ers2/out", "/tmp/init_session/in", "/tmp/init_session/out", + "/tmp/terminate_session/in", "/tmp/terminate_session/out", "/tmp/cdrs/in", + "/tmp/cdrs/out", "/tmp/ers_with_filters/in", "/tmp/ers_with_filters/out", + "/tmp/xmlErs/in", "/tmp/xmlErs/out", "/tmp/fwvErs/in", "/tmp/fwvErs/out", + "/tmp/partErs1/in", "/tmp/partErs1/out", "/tmp/partErs2/in", "/tmp/partErs2/out", + "/tmp/flatstoreErs/in", "/tmp/flatstoreErs/out"} { + if err := os.RemoveAll(dir); err != nil { + t.Fatal("Error removing folder: ", dir, err) + } + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatal("Error creating folder: ", dir, err) + } + } +} + +func testFlatstoreITInitConfig(t *testing.T) { + var err error + flatstoreCfgPath = path.Join(*dataDir, "conf", "samples", flatstoreCfgDIR) + if flatstoreCfg, err = config.NewCGRConfigFromPath(flatstoreCfgPath); err != nil { + t.Fatal("Got config error: ", err.Error()) + } +} + +// InitDb so we can rely on count +func testFlatstoreITInitCdrDb(t *testing.T) { + if err := engine.InitStorDb(flatstoreCfg); err != nil { + t.Fatal(err) + } +} + +// Remove data in both rating and accounting db +func testFlatstoreITResetDataDb(t *testing.T) { + if err := engine.InitDataDb(flatstoreCfg); err != nil { + t.Fatal(err) + } +} + +func testFlatstoreITStartEngine(t *testing.T) { + if _, err := engine.StopStartEngine(flatstoreCfgPath, *waitRater); err != nil { + t.Fatal(err) + } +} + +// Connect rpc client to rater +func testFlatstoreITRpcConn(t *testing.T) { + var err error + flatstoreRPC, err = newRPCClient(flatstoreCfg.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 testFlatstoreITLoadTPFromFolder(t *testing.T) { + //add a default charger + chargerProfile := &v1.ChargerWithCache{ + ChargerProfile: &engine.ChargerProfile{ + Tenant: "cgrates.org", + ID: "Default", + RunID: utils.MetaDefault, + AttributeIDs: []string{"*none"}, + Weight: 20, + }, + } + var result string + if err := flatstoreRPC.Call(utils.ApierV1SetChargerProfile, chargerProfile, &result); err != nil { + t.Error(err) + } else if result != utils.OK { + t.Error("Unexpected reply returned", result) + } +} + +// The default scenario, out of cdrc defined in .cfg file +func testFlatstoreITHandleCdr1File(t *testing.T) { + if err := ioutil.WriteFile(path.Join("/tmp", "acc_1.log"), []byte(fullSuccessfull), 0644); err != nil { + t.Fatal(err.Error()) + } + if err := ioutil.WriteFile(path.Join("/tmp", "missed_calls_1.log"), []byte(fullMissed), 0644); err != nil { + t.Fatal(err.Error()) + } + if err := ioutil.WriteFile(path.Join("/tmp", "acc_2.log"), []byte(part1), 0644); err != nil { + t.Fatal(err.Error()) + } + if err := ioutil.WriteFile(path.Join("/tmp", "acc_3.log"), []byte(part2), 0644); err != nil { + t.Fatal(err.Error()) + } + //Rename(oldpath, newpath string) + for _, fileName := range []string{"acc_1.log", "missed_calls_1.log", "acc_2.log", "acc_3.log"} { + if err := os.Rename(path.Join("/tmp", fileName), path.Join("/tmp/flatstoreErs/in", fileName)); err != nil { + t.Fatal(err) + } + } + time.Sleep(3 * time.Second) + // check the files to be processed + filesInDir, _ := ioutil.ReadDir("/tmp/flatstoreErs/in") + if len(filesInDir) != 0 { + t.Errorf("Files in cdrcInDir: %+v", filesInDir) + } + filesOutDir, _ := ioutil.ReadDir("/tmp/flatstoreErs/out") + if len(filesOutDir) != 6 { + t.Errorf("Unexpected number of files in output directory: %+v", len(filesOutDir)) + } + ePartContent := "INVITE|2daec40c|548625ac|dd0c4c617a9919d29a6175cdff223a9p@0:0:0:0:0:0:0:0|200|OK|1436454408|*prepaid|1001|1002||3401:2069362475\n" + if partContent, err := ioutil.ReadFile(path.Join("/tmp/flatstoreErs/out", "acc_3.log.tmp")); err != nil { + t.Error(err) + } else if (ePartContent) != (string(partContent)) { + t.Errorf("Expecting:\n%s\nReceived:\n%s", ePartContent, string(partContent)) + } +} + +func testFlatstoreITAnalyseCDRs(t *testing.T) { + var reply []*engine.ExternalCDR + if err := flatstoreRPC.Call(utils.ApierV2GetCDRs, utils.RPCCDRsFilter{}, &reply); err != nil { + t.Error("Unexpected error: ", err.Error()) + } else if len(reply) != 8 { + t.Error("Unexpected number of CDRs returned: ", len(reply)) + } + if err := flatstoreRPC.Call(utils.ApierV2GetCDRs, utils.RPCCDRsFilter{MinUsage: "1"}, &reply); err != nil { + t.Error("Unexpected error: ", err.Error()) + } else if len(reply) != 5 { + t.Error("Unexpected number of CDRs returned: ", len(reply)) + } +} + +func testFlatstoreITCleanupFiles(t *testing.T) { + for _, dir := range []string{"/tmp/ers", + "/tmp/ers2", "/tmp/init_session", "/tmp/terminate_session", + "/tmp/cdrs", "/tmp/ers_with_filters", "/tmp/xmlErs", "/tmp/fwvErs", + "/tmp/partErs1", "/tmp/partErs2", "tmp/flatstoreErs"} { + if err := os.RemoveAll(dir); err != nil { + t.Fatal("Error removing folder: ", dir, err) + } + } +} + +func testFlatstoreITKillEngine(t *testing.T) { + if err := engine.KillEngine(*waitRater); err != nil { + t.Error(err) + } +} diff --git a/ers/partial_csv.go b/ers/partial_csv.go index 3da12f7d9..992fdc4cd 100644 --- a/ers/partial_csv.go +++ b/ers/partial_csv.go @@ -284,8 +284,8 @@ func (rdr *PartialCSVFileER) dumpToFile(itmID string, value interface{}) { utils.ERs, cdr.CGRID, err.Error())) return } - dumpFilePath := path.Join(rdr.Config().ProcessedPath, fmt.Sprintf("%s.%s.%d", - cdr.OriginID, PartialRecordsSuffix, time.Now().Unix())) + dumpFilePath := path.Join(rdr.Config().ProcessedPath, fmt.Sprintf("%s%s.%d", + cdr.OriginID, utils.TmpSuffix, time.Now().Unix())) fileOut, err := os.Create(dumpFilePath) if err != nil { utils.Logger.Err(fmt.Sprintf("<%s> Failed creating %s, error: %s", @@ -293,7 +293,7 @@ func (rdr *PartialCSVFileER) dumpToFile(itmID string, value interface{}) { return } csvWriter := csv.NewWriter(fileOut) - csvWriter.Comma = utils.CSV_SEP + csvWriter.Comma = rune(rdr.Config().FieldSep[0]) if err = csvWriter.Write(record); err != nil { utils.Logger.Err(fmt.Sprintf("<%s> Failed writing partial record %v to file: %s, error: %s", utils.ERs, record, dumpFilePath, err.Error())) diff --git a/utils/consts.go b/utils/consts.go index a93fb8ec8..ced040552 100755 --- a/utils/consts.go +++ b/utils/consts.go @@ -619,6 +619,7 @@ const ( MinCost = "MinCost" MaxCost = "MaxCost" MetaLoaders = "*loaders" + TmpSuffix = ".tmp" ) // Migrator Action