From e7f95a98f3b7cf8f9535ff61759adfeca51531a0 Mon Sep 17 00:00:00 2001 From: TeoV Date: Tue, 21 Jan 2020 18:03:26 +0200 Subject: [PATCH 1/3] Add Flastore Support in ERs --- config/erscfg.go | 5 + config/libconfig_json.go | 1 + data/conf/samples/ers_internal/cgrates.json | 25 ++ data/conf/samples/ers_mongo/cgrates.json | 25 ++ data/conf/samples/ers_postgres/cgrates.json | 25 ++ ers/filecsv_it_test.go | 5 +- ers/filefwv_it_test.go | 5 +- ers/filexml_it_test.go | 5 +- ers/flatstore.go | 273 ++++++++++++++++++++ ers/flatstore_it_test.go | 1 + ers/partial_csv_it_test.go | 5 +- ers/reader.go | 2 + 12 files changed, 369 insertions(+), 8 deletions(-) create mode 100644 ers/flatstore.go create mode 100644 ers/flatstore_it_test.go diff --git a/config/erscfg.go b/config/erscfg.go index f46fa5bba..db801358c 100644 --- a/config/erscfg.go +++ b/config/erscfg.go @@ -111,6 +111,7 @@ type EventReaderCfg struct { Timezone string Filters []string Flags utils.FlagsWithParams + FailedCallsPrefix string // Used in case of flatstore CDRs to avoid searching for BYE records PartialRecordCache time.Duration // Duration to cache partial records when not pairing PartialCacheExpiryAction string HeaderFields []*FCTemplate @@ -166,6 +167,9 @@ func (er *EventReaderCfg) loadFromJsonCfg(jsnCfg *EventReaderJsonCfg, sep string return } } + if jsnCfg.Failed_calls_prefix != nil { + er.FailedCallsPrefix = *jsnCfg.Failed_calls_prefix + } if jsnCfg.Partial_record_cache != nil { if er.PartialRecordCache, err = utils.ParseDurationWithNanosecs(*jsnCfg.Partial_record_cache); err != nil { return err @@ -223,6 +227,7 @@ func (er *EventReaderCfg) Clone() (cln *EventReaderCfg) { } } cln.Flags = er.Flags + cln.FailedCallsPrefix = er.FailedCallsPrefix cln.HeaderFields = make([]*FCTemplate, len(er.HeaderFields)) for idx, fld := range er.HeaderFields { cln.HeaderFields[idx] = fld.Clone() diff --git a/config/libconfig_json.go b/config/libconfig_json.go index b792b43e9..e8d057bf8 100755 --- a/config/libconfig_json.go +++ b/config/libconfig_json.go @@ -205,6 +205,7 @@ type EventReaderJsonCfg struct { Timezone *string Filters *[]string Flags *[]string + Failed_calls_prefix *string Partial_record_cache *string Partial_cache_expiry_action *string Header_fields *[]*FcTemplateJsonCfg diff --git a/data/conf/samples/ers_internal/cgrates.json b/data/conf/samples/ers_internal/cgrates.json index 6be160b68..8de6afea6 100644 --- a/data/conf/samples/ers_internal/cgrates.json +++ b/data/conf/samples/ers_internal/cgrates.json @@ -324,6 +324,31 @@ {"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", + "source_path": "/tmp/flatstoreErs/in", + "processed_path": "/tmp/flatstoreErs/out", + "flags": ["*cdrs"], + "partial_record_cache": "1s", + "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": "*none", "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 35db1a718..1f7a59fb7 100644 --- a/data/conf/samples/ers_mongo/cgrates.json +++ b/data/conf/samples/ers_mongo/cgrates.json @@ -327,6 +327,31 @@ {"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", + "source_path": "/tmp/flatstoreErs/in", + "processed_path": "/tmp/flatstoreErs/out", + "flags": ["*cdrs"], + "partial_record_cache": "1s", + "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": "*none", "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 3062ec1fd..c39bf3d69 100644 --- a/data/conf/samples/ers_postgres/cgrates.json +++ b/data/conf/samples/ers_postgres/cgrates.json @@ -321,6 +321,31 @@ {"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", + "source_path": "/tmp/flatstoreErs/in", + "processed_path": "/tmp/flatstoreErs/out", + "flags": ["*cdrs"], + "partial_record_cache": "1s", + "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": "*none", "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/filecsv_it_test.go b/ers/filecsv_it_test.go index af3d99185..a14ca9f19 100644 --- a/ers/filecsv_it_test.go +++ b/ers/filecsv_it_test.go @@ -126,7 +126,8 @@ func testCsvITCreateCdrDirs(t *testing.T) { "/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/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) } @@ -350,7 +351,7 @@ func testCsvITCleanupFiles(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/partErs1", "/tmp/partErs2", "tmp/flatstoreErs"} { if err := os.RemoveAll(dir); err != nil { t.Fatal("Error removing folder: ", dir, err) } diff --git a/ers/filefwv_it_test.go b/ers/filefwv_it_test.go index 130f24eac..07b84b9a9 100644 --- a/ers/filefwv_it_test.go +++ b/ers/filefwv_it_test.go @@ -80,7 +80,8 @@ func testFWVITCreateCdrDirs(t *testing.T) { "/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/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) } @@ -214,7 +215,7 @@ func testFWVITCleanupFiles(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/partErs1", "/tmp/partErs2", "tmp/flatstoreErs"} { if err := os.RemoveAll(dir); err != nil { t.Fatal("Error removing folder: ", dir, err) } diff --git a/ers/filexml_it_test.go b/ers/filexml_it_test.go index e66ac10f1..123066b37 100644 --- a/ers/filexml_it_test.go +++ b/ers/filexml_it_test.go @@ -78,7 +78,8 @@ func testXMLITCreateCdrDirs(t *testing.T) { "/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/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) } @@ -300,7 +301,7 @@ func testXMLITCleanupFiles(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/partErs1", "/tmp/partErs2", "tmp/flatstoreErs"} { if err := os.RemoveAll(dir); err != nil { t.Fatal("Error removing folder: ", dir, err) } diff --git a/ers/flatstore.go b/ers/flatstore.go new file mode 100644 index 000000000..41827bae6 --- /dev/null +++ b/ers/flatstore.go @@ -0,0 +1,273 @@ +/* +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 ( + "bufio" + "encoding/csv" + "errors" + "fmt" + "io" + "io/ioutil" + "os" + "path" + "strconv" + "strings" + "sync" + "time" + + "github.com/cgrates/ltcache" + + "github.com/cgrates/cgrates/agents" + "github.com/cgrates/cgrates/config" + "github.com/cgrates/cgrates/engine" + "github.com/cgrates/cgrates/utils" +) + +func NewFlatstoreER(cfg *config.CGRConfig, cfgIdx int, + rdrEvents chan *erEvent, rdrErr chan error, + fltrS *engine.FilterS, rdrExit chan struct{}) (er EventReader, err error) { + srcPath := cfg.ERsCfg().Readers[cfgIdx].SourcePath + if strings.HasSuffix(srcPath, utils.Slash) { + srcPath = srcPath[:len(srcPath)-1] + } + return &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 +} + +// FlatstoreER implements EventReader interface for Flatstore CDR +type FlatstoreER struct { + sync.RWMutex + cgrCfg *config.CGRConfig + cfgIdx int // index of config instance within ERsCfg.Readers + fltrS *engine.FilterS + cache *ltcache.Cache + rdrDir string + rdrEvents chan *erEvent // channel to dispatch the events created to + rdrError chan error + rdrExit chan struct{} + conReqs chan struct{} // limit number of opened files +} + +func (rdr *FlatstoreER) Config() *config.EventReaderCfg { + return rdr.cgrCfg.ERsCfg().Readers[rdr.cfgIdx] +} + +func (rdr *FlatstoreER) Serve() (err error) { + switch rdr.Config().RunDelay { + case time.Duration(0): // 0 disables the automatic read, maybe done per API + return + case time.Duration(-1): + return watchDir(rdr.rdrDir, rdr.processFile, + utils.ERs, rdr.rdrExit) + default: + go func() { + for { + // Not automated, process and sleep approach + select { + case <-rdr.rdrExit: + utils.Logger.Info( + fmt.Sprintf("<%s> stop monitoring path <%s>", + utils.ERs, rdr.rdrDir)) + return + default: + } + filesInDir, _ := ioutil.ReadDir(rdr.rdrDir) + for _, file := range filesInDir { + if !strings.HasSuffix(file.Name(), utils.CSVSuffix) { // hardcoded file extension for csv event reader + continue // used in order to filter the files from directory + } + go func(fileName string) { + if err := rdr.processFile(rdr.rdrDir, fileName); err != nil { + utils.Logger.Warning( + fmt.Sprintf("<%s> processing file %s, error: %s", + utils.ERs, fileName, err.Error())) + } + }(file.Name()) + } + time.Sleep(rdr.Config().RunDelay) + } + }() + } + return +} + +// processFile is called for each file in a directory and dispatches erEvents from it +func (rdr *FlatstoreER) processFile(fPath, fName string) (err error) { + if cap(rdr.conReqs) != 0 { // 0 goes for no limit + processFile := <-rdr.conReqs // Queue here for maxOpenFiles + defer func() { rdr.conReqs <- processFile }() + } + absPath := path.Join(fPath, fName) + utils.Logger.Info( + fmt.Sprintf("<%s> parsing <%s>", utils.ERs, absPath)) + var file *os.File + if file, err = os.Open(absPath); err != nil { + return + } + defer file.Close() + csvReader := csv.NewReader(bufio.NewReader(file)) + csvReader.Comma = ',' + if len(rdr.Config().FieldSep) > 0 { + csvReader.Comma = rune(rdr.Config().FieldSep[0]) + } + csvReader.Comment = '#' + rowNr := 0 // This counts the rows in the file, not really number of CDRs + evsPosted := 0 + timeStart := time.Now() + reqVars := make(map[string]interface{}) + for { + var record []string + if record, err = csvReader.Read(); err != nil { + if err == io.EOF { + break + } + 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 + } 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) + } + rowNr++ // increment the rowNr after checking if it's not the end of file + agReq := agents.NewAgentRequest( + config.NewSliceDP(record, utils.EmptyString), reqVars, + nil, nil, rdr.Config().Tenant, + rdr.cgrCfg.GeneralCfg().DefaultTenant, + utils.FirstNonEmpty(rdr.Config().Timezone, + rdr.cgrCfg.GeneralCfg().DefaultTimezone), + rdr.fltrS) // create an AgentRequest + if pass, err := rdr.fltrS.Pass(agReq.Tenant, rdr.Config().Filters, + 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( + fmt.Sprintf("<%s> reading file: <%s> row <%d>, ignoring due to error: <%s>", + utils.ERs, absPath, rowNr, err.Error())) + continue + } + + rdr.rdrEvents <- &erEvent{cgrEvent: navMp.AsCGREvent( + agReq.Tenant, utils.NestingSep), + rdrCfg: rdr.Config()} + evsPosted++ + } + if rdr.Config().ProcessedPath != "" { + // Finished with file, move it to processed folder + outPath := path.Join(rdr.Config().ProcessedPath, fName) + if err = os.Rename(absPath, outPath); err != nil { + return + } + } + + utils.Logger.Info( + fmt.Sprintf("%s finished processing file <%s>. Total records processed: %d, events posted: %d, run duration: %s", + utils.ERs, absPath, rowNr, evsPosted, time.Now().Sub(timeStart))) + return +} + +func NewUnpairedRecord(record []string, timezone 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} + var err error + if pr.Timestamp, err = utils.ParseTimeDetectLayout(record[6], timezone); err != nil { + return nil, err + } + return pr, nil +} + +// This is a partial record received from Flatstore, can be INVITE or BYE and it needs to be paired in order to produce duration +type UnpairedRecord struct { + Method string // INVITE or BYE + 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 +} + +// Pairs INVITE and BYE into final record containing as last element the duration +func pairToRecord(part1, part2 *UnpairedRecord) ([]string, error) { + var invite, bye *UnpairedRecord + if part1.Method == "INVITE" { + invite = part1 + } else if part2.Method == "INVITE" { + invite = part2 + } else { + return nil, errors.New("MISSING_INVITE") + } + if part1.Method == "BYE" { + bye = part1 + } else if part2.Method == "BYE" { + bye = part2 + } else { + return nil, errors.New("MISSING_BYE") + } + if len(invite.Values) != len(bye.Values) { + return nil, errors.New("INCONSISTENT_VALUES_LENGTH") + } + record := invite.Values + for idx := range record { + switch idx { + case 0, 1, 2, 3, 6: // Leave these values as they are + case 4, 5: + record[idx] = bye.Values[idx] // Update record with status from bye + default: + if bye.Values[idx] != "" { // Any value higher than 6 is dynamically inserted, overwrite if non empty + record[idx] = bye.Values[idx] + } + + } + } + callDur := bye.Timestamp.Sub(invite.Timestamp) + record = append(record, strconv.FormatFloat(callDur.Seconds(), 'f', -1, 64)) + return record, nil +} diff --git a/ers/flatstore_it_test.go b/ers/flatstore_it_test.go new file mode 100644 index 000000000..15008088f --- /dev/null +++ b/ers/flatstore_it_test.go @@ -0,0 +1 @@ +package ers diff --git a/ers/partial_csv_it_test.go b/ers/partial_csv_it_test.go index 884e9b2a1..85dbf0837 100644 --- a/ers/partial_csv_it_test.go +++ b/ers/partial_csv_it_test.go @@ -96,7 +96,8 @@ func testPartITCreateCdrDirs(t *testing.T) { "/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/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) } @@ -242,7 +243,7 @@ func testPartITCleanupFiles(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/partErs1", "/tmp/partErs2", "tmp/flatstoreErs"} { if err := os.RemoveAll(dir); err != nil { t.Fatal("Error removing folder: ", dir, err) } diff --git a/ers/reader.go b/ers/reader.go index 4aceaba3a..19b096c72 100644 --- a/ers/reader.go +++ b/ers/reader.go @@ -50,6 +50,8 @@ func NewEventReader(cfg *config.CGRConfig, cfgIdx int, return NewKafkaER(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit) case utils.MetaSQL: return NewSQLEventReader(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit) + case utils.MetaKamFlatstore, utils.MetaOsipsFlatstore: + return NewFlatstoreER(cfg, cfgIdx, rdrEvents, rdrErr, fltrS, rdrExit) } return } From 0d4e9898322af5a7bdbc12fff954c99657e5dac4 Mon Sep 17 00:00:00 2001 From: TeoV Date: Wed, 22 Jan 2020 11:06:29 +0200 Subject: [PATCH 2/3] Add test for Flatstore in ERs --- config/config.go | 2 +- config/configsanity.go | 2 +- data/conf/samples/ers_internal/cgrates.json | 6 +- data/conf/samples/ers_mongo/cgrates.json | 6 +- data/conf/samples/ers_mysql/cgrates.json | 27 +++ data/conf/samples/ers_postgres/cgrates.json | 6 +- ers/flatstore.go | 83 ++++--- ers/flatstore_it_test.go | 234 ++++++++++++++++++++ ers/partial_csv.go | 6 +- utils/consts.go | 1 + 10 files changed, 334 insertions(+), 39 deletions(-) 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 From 39f8502f7c74c9a5a2f17d321c083290e90d3df6 Mon Sep 17 00:00:00 2001 From: TeoV Date: Wed, 22 Jan 2020 12:23:00 +0200 Subject: [PATCH 3/3] Update integration test for config --- config/config_it_test.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/config/config_it_test.go b/config/config_it_test.go index 4f10290cd..008926040 100644 --- a/config/config_it_test.go +++ b/config/config_it_test.go @@ -853,6 +853,7 @@ func testCgrCfgV1ReloadConfigSection(t *testing.T) { "Filters": []interface{}{}, "Flags": map[string]interface{}{}, "HeaderFields": []interface{}{}, + "FailedCallsPrefix": "", "ID": "*default", "ProcessedPath": "/var/spool/cgrates/cdrc/out", "RunDelay": 0, @@ -872,6 +873,7 @@ func testCgrCfgV1ReloadConfigSection(t *testing.T) { "*dryrun": []interface{}{}, }, "HeaderFields": []interface{}{}, + "FailedCallsPrefix": "", "PartialCacheExpiryAction": "", "PartialRecordCache": 0, "ID": "file_reader1",