Add Flastore Support in ERs

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()

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

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"}
+			],
 		}
 	],
 },

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"}
+			],
 		}
 	],
 },

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"}
+			],
 		}
 	],
 },

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)
 		}

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)
 		}

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)
 		}

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 <http://www.gnu.org/licenses/>
+*/
+
+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
+}

ers/flatstore_it_test.go

@@ -0,0 +1 @@
+package ers

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)
 		}

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
 }