From e7f95a98f3b7cf8f9535ff61759adfeca51531a0 Mon Sep 17 00:00:00 2001 From: TeoV Date: Tue, 21 Jan 2020 18:03:26 +0200 Subject: [PATCH] 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 }