Enhancing cdrexporter with dry_run, console export_cdrs command

apier/v1/cdrs.go

@@ -25,28 +25,23 @@ import (
 	"os"
 	"path"
 	"time"
+	"strings"
 )
 
-type AttrExpCsvCdrs struct {
-	TimeStart string // If provided, will represent the starting of the CDRs interval (>=)
-	TimeEnd   string // If provided, will represent the end of the CDRs interval (<)
-}
-
-type ExportedCsvCdrs struct {
-	ExportedFilePath string // Full path to the newly generated export file
-	NumberOfCdrs     int    // Number of CDRs in the export file
-}
-
-func (self *ApierV1) ExportCsvCdrs(attr *AttrExpCsvCdrs, reply *ExportedCsvCdrs) error {
+func (self *ApierV1) ExportCdrsToFile(attr utils.AttrExpFileCdrs, reply *utils.ExportedFileCdrs) error {
 	var tStart, tEnd time.Time
 	var err error
+	cdrFormat := strings.ToLower(attr.CdrFormat)
+	if !utils.IsSliceMember(utils.CdreCdrFormats, cdrFormat) {
+		return fmt.Errorf("%s:%s", utils.ERR_MANDATORY_IE_MISSING, "CdrFormat")
+ 	}
 	if len(attr.TimeStart) != 0 {
-		if tStart, err = utils.ParseDate(attr.TimeStart); err != nil {
+		if tStart, err = utils.ParseTimeDetectLayout(attr.TimeStart); err != nil {
 			return err
 		}
 	}
 	if len(attr.TimeEnd) != 0 {
-		if tEnd, err = utils.ParseDate(attr.TimeEnd); err != nil {
+		if tEnd, err = utils.ParseTimeDetectLayout(attr.TimeEnd); err != nil {
 			return err
 		}
 	}
@@ -54,21 +49,33 @@ func (self *ApierV1) ExportCsvCdrs(attr *AttrExpCsvCdrs, reply *ExportedCsvCdrs)
 	if err != nil {
 		return err
 	}
-	fileName := path.Join(self.Config.CdreDir, "cgr", "csv", fmt.Sprintf("cdrs_%d.csv", time.Now().Unix()))
-	fileOut, err := os.Create(fileName)
-	if err != nil {
-		return err
-	} else {
-		defer fileOut.Close()
-	}
-	csvWriter := cdrexporter.NewCsvCdrWriter(fileOut, self.Config.RoundingDecimals, self.Config.CdreExtraFields)
-	for _, cdr := range cdrs {
-		if err := csvWriter.Write(cdr); err != nil {
-			os.Remove(fileName)
+	var fileName string
+	if cdrFormat == utils.CDRE_CSV && len(cdrs) != 0 {
+		fileName = path.Join(self.Config.CdreDir, fmt.Sprintf("cdrs_%d.csv", time.Now().Unix()))
+		fileOut, err := os.Create(fileName)
+		if err != nil {
 			return err
+		} else {
+			defer fileOut.Close()
+		}
+		csvWriter := cdrexporter.NewCsvCdrWriter(fileOut, self.Config.RoundingDecimals, self.Config.CdreExtraFields)
+		for _, cdr := range cdrs {
+			if err := csvWriter.Write(cdr); err != nil {
+				os.Remove(fileName)
+			return err
+			}
+		}
+		csvWriter.Close()
+		if attr.RemoveFromDb {
+			cgrIds := make([]string, len(cdrs))
+			for idx, cdr := range cdrs {
+				cgrIds[idx] = cdr.CgrId
+			}
+			if err := self.CdrDb.RemRatedCdrs(cgrIds); err != nil {
+				return err
+			}
 		}
 	}
-	csvWriter.Close()
-	*reply = ExportedCsvCdrs{fileName, len(cdrs)}
+	*reply = utils.ExportedFileCdrs{fileName, len(cdrs)}
 	return nil
 }

cdrexporter/csv.go

@@ -37,7 +37,7 @@ func NewCsvCdrWriter(writer io.Writer, roundDecimals int, extraFields []string)
 }
 
 func (dcw *CsvCdrWriter) Write(cdr *utils.RatedCDR) error {
-	primaryFields := []string{cdr.CgrId, cdr.AccId, cdr.CdrHost, cdr.ReqType, cdr.Direction, cdr.Tenant, cdr.TOR, cdr.Account, cdr.Subject,
+	primaryFields := []string{cdr.CgrId, cdr.MediationRunId, cdr.AccId, cdr.CdrHost, cdr.ReqType, cdr.Direction, cdr.Tenant, cdr.TOR, cdr.Account, cdr.Subject,
 		cdr.Destination, cdr.AnswerTime.String(), strconv.Itoa(int(cdr.Duration)), strconv.FormatFloat(cdr.Cost, 'f', dcw.roundDecimals, 64)}
 	if len(dcw.extraFields) == 0 {
 		dcw.extraFields = utils.MapKeys(cdr.ExtraFields)

cdrexporter/csv_test.go

@@ -30,12 +30,12 @@ func TestCsvCdrWriter(t *testing.T) {
 	writer := &bytes.Buffer{}
 	csvCdrWriter := NewCsvCdrWriter(writer, 4, []string{"extra3", "extra1"})
 	ratedCdr := &utils.RatedCDR{CgrId: utils.FSCgrId("dsafdsaf"), AccId: "dsafdsaf", CdrHost: "192.168.1.1", ReqType: "rated", Direction: "*out", Tenant: "cgrates.org",
-		TOR: "call", Account: "1001", Subject: "1001", Destination: "1002", AnswerTime: time.Unix(1383813746, 0).UTC(), Duration: 10,
+		TOR: "call", Account: "1001", Subject: "1001", Destination: "1002", AnswerTime: time.Unix(1383813746, 0).UTC(), Duration: 10, MediationRunId: utils.DEFAULT_RUNID,
 		ExtraFields: map[string]string{"extra1": "val_extra1", "extra2": "val_extra2", "extra3": "val_extra3"}, Cost: 1.01,
 	}
 	csvCdrWriter.Write(ratedCdr)
 	csvCdrWriter.Close()
-	expected := "b18944ef4dc618569f24c27b9872827a242bad0c,dsafdsaf,192.168.1.1,rated,*out,cgrates.org,call,1001,1001,1002,2013-11-07 08:42:26 +0000 UTC,10,1.0100,val_extra3,val_extra1"
+	expected := "b18944ef4dc618569f24c27b9872827a242bad0c,default,dsafdsaf,192.168.1.1,rated,*out,cgrates.org,call,1001,1001,1002,2013-11-07 08:42:26 +0000 UTC,10,1.0100,val_extra3,val_extra1"
 	result := strings.TrimSpace(writer.String())
 	if result != expected {
 		t.Errorf("Expected %s received %s.", expected, result)

console/export_cdrs.go

@@ -0,0 +1,95 @@
+/*
+Rating system designed to be used in VoIP Carriers World
+Copyright (C) 2013 ITsysCOM
+
+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 console
+
+import (
+	"fmt"
+	"github.com/cgrates/cgrates/utils"
+	"time"
+)
+
+func init() {
+	commands["export_cdrs"] = &CmdExportCdrs{}
+}
+
+// Commander implementation
+type CmdExportCdrs struct {
+	rpcMethod string
+	rpcParams *utils.AttrExpFileCdrs
+	rpcResult utils.ExportedFileCdrs
+}
+
+// name should be exec's name
+func (self *CmdExportCdrs) Usage(name string) string {
+	return fmt.Sprintf("\n\tUsage: cgr-console [cfg_opts...{-h}] export_cdrs <csv|dry_run> [<start_time|*one_month> [<stop_time> [remove_from_db]]]")
+}
+
+// set param defaults
+func (self *CmdExportCdrs) defaults() error {
+	self.rpcMethod = "ApierV1.ExportCdrsToFile"
+	self.rpcParams = &utils.AttrExpFileCdrs{CdrFormat:"csv"}
+	return nil
+}
+
+func (self *CmdExportCdrs) FromArgs(args []string) error {
+	self.defaults()
+	var timeStart, timeEnd string
+	if len(args) < 3 {
+		return fmt.Errorf(self.Usage(""))
+	}
+	if !utils.IsSliceMember(utils.CdreCdrFormats, args[2]) {
+		return fmt.Errorf(self.Usage(""))
+	}
+	self.rpcParams.CdrFormat = args[2]
+	switch len(args) {
+	case 4:
+		timeStart = args[3]
+		
+	case 5:
+		timeStart = args[3]
+		timeEnd = args[4]
+	case 6:
+		timeStart = args[3]
+		timeEnd = args[4]
+		if args[5] == "remove_from_db" {
+			self.rpcParams.RemoveFromDb = true
+		}
+	}
+	if timeStart == "*one_month" {
+		now := time.Now()
+		self.rpcParams.TimeStart = now.AddDate(0,-1,0).String()
+		self.rpcParams.TimeEnd = now.String()
+	} else {
+		self.rpcParams.TimeStart = timeStart
+		self.rpcParams.TimeEnd = timeEnd
+	}
+	return nil
+}
+
+func (self *CmdExportCdrs) RpcMethod() string {
+	return self.rpcMethod
+}
+
+func (self *CmdExportCdrs) RpcParams() interface{} {
+	return self.rpcParams
+}
+
+func (self *CmdExportCdrs) RpcResult() interface{} {
+	return &self.rpcResult
+}

engine/storage_interface.go

@@ -96,6 +96,7 @@ type CdrStorage interface {
 	SetCdr(utils.RawCDR) error
 	SetRatedCdr(*utils.RatedCDR, string) error
 	GetRatedCdrs(time.Time, time.Time) ([]*utils.RatedCDR, error)
+	RemRatedCdrs([]string) error
 }
 
 type LogStorage interface {

engine/storage_sql.go

@@ -760,14 +760,14 @@ func (self *SQLStorage) GetRatedCdrs(timeStart, timeEnd time.Time) ([]*utils.Rat
 	for rows.Next() {
 		var cgrid, accid, cdrhost, cdrsrc, reqtype, direction, tenant, tor, account, subject, destination, runid string
 		var extraFields []byte
-		var answerTimestamp, duration int64
+		var answerTime time.Time
+		var duration int64
 		var cost float64
 		var extraFieldsMp map[string]string
-		if err := rows.Scan(&cgrid, &accid, &cdrhost, &cdrsrc, &reqtype, &direction, &tenant, &tor, &account, &subject, &destination, &answerTimestamp, &duration,
+		if err := rows.Scan(&cgrid, &accid, &cdrhost, &cdrsrc, &reqtype, &direction, &tenant, &tor, &account, &subject, &destination, &answerTime, &duration,
 			&extraFields, &runid, &cost); err != nil {
 			return nil, err
 		}
-		answerTime := time.Unix(answerTimestamp, 0)
 		if err := json.Unmarshal(extraFields, &extraFieldsMp); err != nil {
 			return nil, err
 		}
@@ -781,6 +781,32 @@ func (self *SQLStorage) GetRatedCdrs(timeStart, timeEnd time.Time) ([]*utils.Rat
 	return cdrs, nil
 }
 
+// Remove CDR data out of all CDR tables based on their cgrid
+func (self *SQLStorage) RemRatedCdrs(cgrIds []string) error {
+	if len(cgrIds) == 0 {
+		return nil
+	}
+	buffRated := bytes.NewBufferString(fmt.Sprintf("DELETE FROM %s WHERE", utils.TBL_RATED_CDRS))
+	buffCosts := bytes.NewBufferString(fmt.Sprintf("DELETE FROM %s WHERE", utils.TBL_COST_DETAILS))
+	buffCdrExtra := bytes.NewBufferString(fmt.Sprintf("DELETE FROM %s WHERE",utils.TBL_CDRS_EXTRA))
+	buffCdrPrimary := bytes.NewBufferString(fmt.Sprintf("DELETE FROM %s WHERE",utils.TBL_CDRS_PRIMARY))
+	qryBuffers := []*bytes.Buffer{buffRated, buffCosts, buffCdrExtra, buffCdrPrimary}
+	for idx, cgrId := range cgrIds {
+		for _, buffer := range qryBuffers {
+			if idx != 0 {
+				buffer.WriteString(" OR")
+			}
+			buffer.WriteString(fmt.Sprintf(" cgrid='%s'",cgrId))
+		}
+	}
+	for _, buffer := range qryBuffers {
+		if _, err := self.Db.Exec(buffer.String()); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (self *SQLStorage) GetTpDestinations(tpid, tag string) ([]*Destination, error) {
 	var dests []*Destination
 	q := fmt.Sprintf("SELECT * FROM %s WHERE tpid='%s'", utils.TBL_TP_DESTINATIONS, tpid)

utils/apitpdata.go

@@ -304,3 +304,16 @@ type CachedItemAge struct {
 	RatingProfile  time.Duration
 	Action         time.Duration
 }
+
+type AttrExpFileCdrs struct {
+	CdrFormat  string  // Cdr output file format <utils.CdreCdrFormats>
+	TimeStart string // If provided, will represent the starting of the CDRs interval (>=)
+	TimeEnd   string // If provided, will represent the end of the CDRs interval (<)
+	RemoveFromDb  bool  // If true the CDRs will be also deleted after export
+	
+}
+
+type ExportedFileCdrs struct {
+	ExportedFilePath string // Full path to the newly generated export file
+	NumberOfRecords     int    // Number of CDRs in the export file
+}

utils/consts.go

@@ -78,4 +78,10 @@ const (
 	DURATION                 = "duration"
 	DEFAULT_RUNID            = "default"
 	STATIC_VALUE_PREFIX      = "^"
+	CDRE_CSV                 = "csv"
+	CDRE_DRYRUN              = "dry_run"
+)
+
+var (
+	CdreCdrFormats = []string{CDRE_CSV, CDRE_DRYRUN}
 )

utils/coreutils.go

@@ -101,12 +101,15 @@ func Round(x float64, prec int, method string) float64 {
 
 func ParseTimeDetectLayout(tmStr string) (time.Time, error) {
 	var nilTime time.Time
-	rfc3339Rule := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.+`)
-	sqlRule := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}`)
-	unixTimestampRule := regexp.MustCompile(`^\d{10}`)
+	rfc3339Rule := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.+$`)
+	sqlRule := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}$`)
+	gotimeRule := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}\.?\d*\s[+,-]\d+\s\w+$`)
+	unixTimestampRule := regexp.MustCompile(`^\d{10}$`)
 	switch {
 	case rfc3339Rule.MatchString(tmStr):
 		return time.Parse(time.RFC3339, tmStr)
+	case gotimeRule.MatchString(tmStr):
+		return time.Parse("2006-01-02 15:04:05.999999999 -0700 MST", tmStr)
 	case sqlRule.MatchString(tmStr):
 		return time.Parse("2006-01-02 15:04:05", tmStr)
 	case unixTimestampRule.MatchString(tmStr):

utils/utils_test.go

@@ -156,6 +156,28 @@ func TestParseTimeDetectLayout(t *testing.T) {
 	if err == nil {
 		t.Errorf("Expecting error")
 	}
+	goTmStr := "2013-12-30 15:00:01 +0000 UTC"
+	goTm, err := ParseTimeDetectLayout(goTmStr)
+	if err != nil {
+		t.Error(err)
+	} else if !goTm.Equal(expectedTime) {
+		t.Errorf("Unexpected time parsed: %v, expecting: %v", goTm, expectedTime)
+	}
+	_, err = ParseTimeDetectLayout(goTmStr[1:])
+	if err == nil {
+		t.Errorf("Expecting error")
+	}
+	goTmStr = "2013-12-30 15:00:01.000000000 +0000 UTC"
+	goTm, err = ParseTimeDetectLayout(goTmStr)
+	if err != nil {
+		t.Error(err)
+	} else if !goTm.Equal(expectedTime) {
+		t.Errorf("Unexpected time parsed: %v, expecting: %v", goTm, expectedTime)
+	}
+	_, err = ParseTimeDetectLayout(goTmStr[1:])
+	if err == nil {
+		t.Errorf("Expecting error")
+	}
 }
 
 func TestParseDateUnix(t *testing.T) {