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/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",
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/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..681262943 100644
--- a/data/conf/samples/ers_internal/cgrates.json
+++ b/data/conf/samples/ers_internal/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_mongo/cgrates.json b/data/conf/samples/ers_mongo/cgrates.json
index 35db1a718..02bdda690 100644
--- a/data/conf/samples/ers_mongo/cgrates.json
+++ b/data/conf/samples/ers_mongo/cgrates.json
@@ -327,6 +327,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_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 3062ec1fd..54e85831e 100644
--- a/data/conf/samples/ers_postgres/cgrates.json
+++ b/data/conf/samples/ers_postgres/cgrates.json
@@ -321,6 +321,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/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..19781dd81
--- /dev/null
+++ b/ers/flatstore.go
@@ -0,0 +1,300 @@
+/*
+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]
+ }
+ flatER := &FlatstoreER{
+ cgrCfg: cfg,
+ cfgIdx: cfgIdx,
+ fltrS: fltrS,
+ rdrDir: srcPath,
+ rdrEvents: rdrEvents,
+ rdrError: rdrErr,
+ 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
+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(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 {
+ pr, err := NewUnpairedRecord(record, rdr.Config().Timezone, fName)
+ 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)
+ }
+ }
+
+ // 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(
+ 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
+ }
+ 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, 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, FileName: fileName}
+ 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
+ FileName string
+}
+
+// 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
+}
+
+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
new file mode 100644
index 000000000..00a24530c
--- /dev/null
+++ b/ers/flatstore_it_test.go
@@ -0,0 +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/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
}
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