uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-17 14:19:54 +05:00
Code Issues Packages Projects Releases Wiki Activity

CDRExporter redesign, modified configuration parameters, CDRS replication -> online_cdr_exports

Browse Source
This commit is contained in:
DanB
2017-02-14 17:41:30 +01:00
parent 393c1acbd0
commit d27dc06e7d
15 changed files with 401 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
apier/v1/apier.go
View File

@@ -52,6 +52,7 @@ type ApierV1 struct {
Users rpcclient.RpcClientConnection
CDRs rpcclient.RpcClientConnection // FixMe: populate it from cgr-engine
ServManager *servmanager.ServiceManager // Need to have them capitalize so we can export in V2
HTTPPoster *utils.HTTPPoster
}
func (self *ApierV1) GetDestination(dstId string, reply *engine.Destination) error {

52
apier/v1/cdre.go
View File

@@ -32,8 +32,8 @@ import (
"time"
"unicode/utf8"
"github.com/cgrates/cgrates/cdre"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
)
@@ -90,11 +90,7 @@ func (self *ApierV1) ExportCdrsToZipString(attr utils.AttrExpFileCdrs, reply *st
}
// Export Cdrs to file
func (self *ApierV1) ExportCdrsToFile(attr utils.AttrExpFileCdrs, reply *utils.ExportedFileCdrs) error {
var err error
//cdreReloadStruct := <-self.Config.ConfigReloads[utils.CDRE] // Read the content of the channel, locking it
//defer func() { self.Config.ConfigReloads[utils.CDRE] <- cdreReloadStruct }() // Unlock reloads at exit
func (self *ApierV1) ExportCdrsToFile(attr utils.AttrExpFileCdrs, reply *utils.ExportedFileCdrs) (err error) {
exportTemplate := self.Config.CdreProfiles[utils.META_DEFAULT]
if attr.ExportTemplate != nil && len(*attr.ExportTemplate) != 0 { // Export template prefered, use it
var hasIt bool
@@ -105,11 +101,11 @@ func (self *ApierV1) ExportCdrsToFile(attr utils.AttrExpFileCdrs, reply *utils.E
if exportTemplate == nil {
return fmt.Errorf("%s:ExportTemplate", utils.ErrMandatoryIeMissing.Error())
}
cdrFormat := exportTemplate.CdrFormat
exportFormat := exportTemplate.ExportFormat
if attr.CdrFormat != nil && len(*attr.CdrFormat) != 0 {
cdrFormat = strings.ToLower(*attr.CdrFormat)
exportFormat = strings.ToLower(*attr.CdrFormat)
}
if !utils.IsSliceMember(utils.CdreCdrFormats, cdrFormat) {
if !utils.IsSliceMember(utils.CDRExportFormats, exportFormat) {
return fmt.Errorf("%s:%s", utils.ErrMandatoryIeMissing.Error(), "CdrFormat")
}
fieldSep := exportTemplate.FieldSeparator
@@ -119,37 +115,34 @@ func (self *ApierV1) ExportCdrsToFile(attr utils.AttrExpFileCdrs, reply *utils.E
return fmt.Errorf("%s:FieldSeparator:%s", utils.ErrServerError.Error(), "Invalid")
}
}
exportDir := exportTemplate.ExportDirectory
exportPath := exportTemplate.ExportPath
if attr.ExportDir != nil && len(*attr.ExportDir) != 0 {
exportDir = *attr.ExportDir
exportPath = *attr.ExportDir
}
exportId := strconv.FormatInt(time.Now().Unix(), 10)
exportID := strconv.FormatInt(time.Now().Unix(), 10)
if attr.ExportId != nil && len(*attr.ExportId) != 0 {
exportId = *attr.ExportId
exportID = *attr.ExportId
}
fileName := fmt.Sprintf("cdre_%s.%s", exportId, cdrFormat)
fileName := fmt.Sprintf("cdre_%s.%s", exportID, exportFormat)
if attr.ExportFileName != nil && len(*attr.ExportFileName) != 0 {
fileName = *attr.ExportFileName
}
filePath := path.Join(exportDir, fileName)
if cdrFormat == utils.DRYRUN {
filePath := path.Join(exportPath, fileName)
if exportFormat == utils.DRYRUN {
filePath = utils.DRYRUN
}
dataUsageMultiplyFactor := exportTemplate.DataUsageMultiplyFactor
usageMultiplyFactor := exportTemplate.UsageMultiplyFactor
if attr.DataUsageMultiplyFactor != nil && *attr.DataUsageMultiplyFactor != 0.0 {
dataUsageMultiplyFactor = *attr.DataUsageMultiplyFactor
usageMultiplyFactor[utils.DATA] = *attr.DataUsageMultiplyFactor
}
smsUsageMultiplyFactor := exportTemplate.SMSUsageMultiplyFactor
if attr.SmsUsageMultiplyFactor != nil && *attr.SmsUsageMultiplyFactor != 0.0 {
smsUsageMultiplyFactor = *attr.SmsUsageMultiplyFactor
usageMultiplyFactor[utils.SMS] = *attr.SmsUsageMultiplyFactor
}
mmsUsageMultiplyFactor := exportTemplate.MMSUsageMultiplyFactor
if attr.MmsUsageMultiplyFactor != nil && *attr.MmsUsageMultiplyFactor != 0.0 {
mmsUsageMultiplyFactor = *attr.MmsUsageMultiplyFactor
usageMultiplyFactor[utils.MMS] = *attr.MmsUsageMultiplyFactor
}
genericUsageMultiplyFactor := exportTemplate.GenericUsageMultiplyFactor
if attr.GenericUsageMultiplyFactor != nil && *attr.GenericUsageMultiplyFactor != 0.0 {
genericUsageMultiplyFactor = *attr.GenericUsageMultiplyFactor
usageMultiplyFactor[utils.GENERIC] = *attr.GenericUsageMultiplyFactor
}
costMultiplyFactor := exportTemplate.CostMultiplyFactor
if attr.CostMultiplyFactor != nil && *attr.CostMultiplyFactor != 0.0 {
@@ -166,18 +159,19 @@ func (self *ApierV1) ExportCdrsToFile(attr utils.AttrExpFileCdrs, reply *utils.E
*reply = utils.ExportedFileCdrs{ExportedFilePath: ""}
return nil
}
cdrexp, err := cdre.NewCdrExporter(cdrs, self.CdrDb, exportTemplate, cdrFormat, fieldSep, exportId, dataUsageMultiplyFactor, smsUsageMultiplyFactor,
mmsUsageMultiplyFactor, genericUsageMultiplyFactor, costMultiplyFactor, self.Config.RoundingDecimals, self.Config.HttpSkipTlsVerify)
cdrexp, err := engine.NewCDRExporter(cdrs, exportTemplate, exportFormat, filePath, utils.META_NONE, exportID,
exportTemplate.Synchronous, exportTemplate.Attempts, fieldSep, usageMultiplyFactor,
costMultiplyFactor, self.Config.RoundingDecimals, self.Config.HttpSkipTlsVerify, self.HTTPPoster)
if err != nil {
return utils.NewErrServerError(err)
}
if err := cdrexp.ExportCDRs(); err != nil {
return utils.NewErrServerError(err)
}
if cdrexp.TotalExportedCdrs() == 0 {
*reply = utils.ExportedFileCdrs{ExportedFilePath: ""}
return nil
}
if err := cdrexp.WriteToFile(filePath); err != nil {
return utils.NewErrServerError(err)
}
*reply = utils.ExportedFileCdrs{ExportedFilePath: filePath, TotalRecords: len(cdrs), TotalCost: cdrexp.TotalCost(), FirstOrderId: cdrexp.FirstOrderId(), LastOrderId: cdrexp.LastOrderId()}
if !attr.SuppressCgrIds {
reply.ExportedCgrIds = cdrexp.PositiveExports()

42
apier/v2/cdre.go
View File

@@ -25,7 +25,7 @@ import (
"time"
"unicode/utf8"
"github.com/cgrates/cgrates/cdre"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
)
@@ -41,11 +41,11 @@ func (self *ApierV2) ExportCdrsToFile(attr utils.AttrExportCdrsToFile, reply *ut
return fmt.Errorf("%s:ExportTemplate", utils.ErrNotFound)
}
}
cdrFormat := exportTemplate.CdrFormat
exportFormat := exportTemplate.ExportFormat
if attr.CdrFormat != nil && len(*attr.CdrFormat) != 0 {
cdrFormat = strings.ToLower(*attr.CdrFormat)
exportFormat = strings.ToLower(*attr.CdrFormat)
}
if !utils.IsSliceMember(utils.CdreCdrFormats, cdrFormat) {
if !utils.IsSliceMember(utils.CDRExportFormats, exportFormat) {
return utils.NewErrMandatoryIeMissing("CdrFormat")
}
fieldSep := exportTemplate.FieldSeparator
@@ -55,37 +55,34 @@ func (self *ApierV2) ExportCdrsToFile(attr utils.AttrExportCdrsToFile, reply *ut
return fmt.Errorf("%s:FieldSeparator:%s", utils.ErrServerError, "Invalid")
}
}
eDir := exportTemplate.ExportDirectory
eDir := exportTemplate.ExportPath
if attr.ExportDirectory != nil && len(*attr.ExportDirectory) != 0 {
eDir = *attr.ExportDirectory
}
ExportID := strconv.FormatInt(time.Now().Unix(), 10)
exportID := strconv.FormatInt(time.Now().Unix(), 10)
if attr.ExportID != nil && len(*attr.ExportID) != 0 {
ExportID = *attr.ExportID
exportID = *attr.ExportID
}
fileName := fmt.Sprintf("cdre_%s.%s", ExportID, cdrFormat)
fileName := fmt.Sprintf("cdre_%s.%s", exportID, exportFormat)
if attr.ExportFileName != nil && len(*attr.ExportFileName) != 0 {
fileName = *attr.ExportFileName
}
filePath := path.Join(eDir, fileName)
if cdrFormat == utils.DRYRUN {
if exportFormat == utils.DRYRUN {
filePath = utils.DRYRUN
}
dataUsageMultiplyFactor := exportTemplate.DataUsageMultiplyFactor
usageMultiplyFactor := exportTemplate.UsageMultiplyFactor
if attr.DataUsageMultiplyFactor != nil && *attr.DataUsageMultiplyFactor != 0.0 {
dataUsageMultiplyFactor = *attr.DataUsageMultiplyFactor
usageMultiplyFactor[utils.DATA] = *attr.DataUsageMultiplyFactor
}
SMSUsageMultiplyFactor := exportTemplate.SMSUsageMultiplyFactor
if attr.SMSUsageMultiplyFactor != nil && *attr.SMSUsageMultiplyFactor != 0.0 {
SMSUsageMultiplyFactor = *attr.SMSUsageMultiplyFactor
usageMultiplyFactor[utils.SMS] = *attr.SMSUsageMultiplyFactor
}
MMSUsageMultiplyFactor := exportTemplate.MMSUsageMultiplyFactor
if attr.MMSUsageMultiplyFactor != nil && *attr.MMSUsageMultiplyFactor != 0.0 {
MMSUsageMultiplyFactor = *attr.MMSUsageMultiplyFactor
usageMultiplyFactor[utils.MMS] = *attr.MMSUsageMultiplyFactor
}
genericUsageMultiplyFactor := exportTemplate.GenericUsageMultiplyFactor
if attr.GenericUsageMultiplyFactor != nil && *attr.GenericUsageMultiplyFactor != 0.0 {
genericUsageMultiplyFactor = *attr.GenericUsageMultiplyFactor
usageMultiplyFactor[utils.GENERIC] = *attr.GenericUsageMultiplyFactor
}
costMultiplyFactor := exportTemplate.CostMultiplyFactor
if attr.CostMultiplyFactor != nil && *attr.CostMultiplyFactor != 0.0 {
@@ -106,18 +103,19 @@ func (self *ApierV2) ExportCdrsToFile(attr utils.AttrExportCdrsToFile, reply *ut
if attr.RoundingDecimals != nil {
roundingDecimals = *attr.RoundingDecimals
}
cdrexp, err := cdre.NewCdrExporter(cdrs, self.CdrDb, exportTemplate, cdrFormat, fieldSep, ExportID, dataUsageMultiplyFactor, SMSUsageMultiplyFactor,
MMSUsageMultiplyFactor, genericUsageMultiplyFactor, costMultiplyFactor, roundingDecimals, self.Config.HttpSkipTlsVerify)
cdrexp, err := engine.NewCDRExporter(cdrs, exportTemplate, exportFormat, filePath, utils.META_NONE, exportID,
exportTemplate.Synchronous, exportTemplate.Attempts, fieldSep, usageMultiplyFactor,
costMultiplyFactor, roundingDecimals, self.Config.HttpSkipTlsVerify, self.HTTPPoster)
if err != nil {
return utils.NewErrServerError(err)
}
if err := cdrexp.ExportCDRs(); err != nil {
return utils.NewErrServerError(err)
}
if cdrexp.TotalExportedCdrs() == 0 {
*reply = utils.ExportedFileCdrs{ExportedFilePath: ""}
return nil
}
if err := cdrexp.WriteToFile(filePath); err != nil {
return utils.NewErrServerError(err)
}
*reply = utils.ExportedFileCdrs{ExportedFilePath: filePath, TotalRecords: len(cdrs), TotalCost: cdrexp.TotalCost(), FirstOrderId: cdrexp.FirstOrderId(), LastOrderId: cdrexp.LastOrderId()}
if !attr.Verbose {
reply.ExportedCgrIds = cdrexp.PositiveExports()

2
cmd/cgr-engine/rater.go
View File

@@ -231,7 +231,7 @@ func startRater(internalRaterChan chan rpcclient.RpcClientConnection, cacheDoneC
responder := &engine.Responder{Bal: bal, ExitChan: exitChan}
responder.SetTimeToLive(cfg.ResponseCacheTTL, nil)
apierRpcV1 := &v1.ApierV1{StorDb: loadDb, RatingDb: ratingDb, AccountDb: accountDb, CdrDb: cdrDb,
Config: cfg, Responder: responder, ServManager: serviceManager}
Config: cfg, Responder: responder, ServManager: serviceManager, HTTPPoster: utils.NewHTTPPoster(cfg.HttpSkipTlsVerify, cfg.ReplyTimeout)}
if cdrStats != nil { // ToDo: Fix here properly the init of stats
responder.Stats = cdrStats
apierRpcV1.CdrStatsSrv = cdrStats

7
config/cdreconfig.go
View File

@@ -25,6 +25,8 @@ import (
type CdreConfig struct {
ExportFormat string
ExportPath string
FallbackPath string
CDRFilter utils.RSRFields
Synchronous bool
Attempts int
FieldSeparator rune
@@ -46,6 +48,11 @@ func (self *CdreConfig) loadFromJsonCfg(jsnCfg *CdreJsonCfg) error {
if jsnCfg.Export_path != nil {
self.ExportPath = *jsnCfg.Export_path
}
if jsnCfg.Cdr_filter != nil {
if self.CDRFilter, err = utils.ParseRSRFields(*jsnCfg.Cdr_filter, utils.INFIELD_SEP); err != nil {
return err
}
}
if jsnCfg.Synchronous != nil {
self.Synchronous = *jsnCfg.Synchronous
}

1
config/config_json_test.go
View File

@@ -284,6 +284,7 @@ func TestDfCdreJsonCfgs(t *testing.T) {
utils.META_DEFAULT: &CdreJsonCfg{
Export_format: utils.StringPointer(utils.MetaFileCSV),
Export_path: utils.StringPointer("/var/spool/cgrates/cdre"),
Cdr_filter: utils.StringPointer(""),
Synchronous: utils.BoolPointer(false),
Attempts: utils.IntPointer(1),
Field_separator: utils.StringPointer(","),

1
config/libconfig_json.go
View File

@@ -147,6 +147,7 @@ type CdrFieldJsonCfg struct {
type CdreJsonCfg struct {
Export_format *string
Export_path *string
Cdr_filter *string
Synchronous *bool
Attempts *int
Field_separator *string

18
engine/cdr.go
View File

@@ -792,6 +792,11 @@ func (cdr *CDR) formatField(cfgFld *config.CfgCdrField, httpSkipTlsCheck bool, g
}
// Part of event interface
func (cdr *CDR) AsMapStringIface() (map[string]interface{}, error) {
return nil, utils.ErrNotImplemented
}
// Used in place where we need to export the CDR based on an export template
// ExportRecord is a []string to keep it compatible with encoding/csv Writer
func (cdr *CDR) AsExportRecord(exportFields []*config.CfgCdrField, httpSkipTlsCheck bool, groupedCDRs []*CDR, roundingDecs int) (expRecord []string, err error) {
@@ -812,16 +817,17 @@ func (cdr *CDR) AsExportRecord(exportFields []*config.CfgCdrField, httpSkipTlsCh
return expRecord, nil
}
// Part of event interface
func (cdr *CDR) AsMapStringIface() (map[string]interface{}, error) {
return nil, utils.ErrNotImplemented
}
// AsExportMap converts the CDR into a map[string]string based on export template
// Used in real-time replication as well as remote exports
func (cdr *CDR) AsExportMap(exportFields []*config.CfgCdrField, httpSkipTlsCheck bool, groupedCDRs []*CDR) (expMap map[string]string, err error) {
func (cdr *CDR) AsExportMap(exportFields []*config.CfgCdrField, httpSkipTlsCheck bool, groupedCDRs []*CDR, roundingDecs int) (expMap map[string]string, err error) {
expMap = make(map[string]string)
for _, cfgFld := range exportFields {
if roundingDecs != 0 {
clnFld := new(config.CfgCdrField) // Clone so we can modify the rounding decimals without affecting the template
*clnFld = *cfgFld
clnFld.RoundingDecimals = roundingDecs
cfgFld = clnFld
}
if fmtOut, err := cdr.formatField(cfgFld, httpSkipTlsCheck, groupedCDRs); err != nil {
return nil, err
} else {

2
engine/cdr_test.go
View File

@@ -579,7 +579,7 @@ func TestCDRAsExportMap(t *testing.T) {
&config.CfgCdrField{FieldId: utils.DESTINATION, Type: utils.META_COMPOSED, Value: utils.ParseRSRFieldsMustCompile("~Destination:s/^\\+(\\d+)$/00${1}/", utils.INFIELD_SEP)},
&config.CfgCdrField{FieldId: "FieldExtra1", Type: utils.META_COMPOSED, Value: utils.ParseRSRFieldsMustCompile("field_extr1", utils.INFIELD_SEP)},
}
if cdrMp, err := cdr.AsExportMap(expFlds, false, nil); err != nil {
if cdrMp, err := cdr.AsExportMap(expFlds, false, nil, 0); err != nil {
t.Error(err)
} else if !reflect.DeepEqual(eCDRMp, cdrMp) {
t.Errorf("Expecting: %+v, received: %+v", eCDRMp, cdrMp)

299
engine/cdrexporter.go → engine/cdre.go
View File

@@ -15,19 +15,23 @@ 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 cdre
package engine
import (
"encoding/csv"
"encoding/json"
"fmt"
"io"
"net/url"
"os"
"path"
"strconv"
"sync"
"time"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
"github.com/streadway/amqp"
)
const (
@@ -45,68 +49,68 @@ const (
META_FORMATCOST = "*format_cost"
)
var err error
func NewCdrExporter(cdrs []*engine.CDR, cdrDb engine.CdrStorage, exportTpl *config.CdreConfig, cdrFormat string, fieldSeparator rune, exportId string,
dataUsageMultiplyFactor, smsUsageMultiplyFactor, mmsUsageMultiplyFactor, genericUsageMultiplyFactor, costMultiplyFactor float64,
roundingDecimals int, httpSkipTlsCheck bool) (*CdrExporter, error) {
func NewCDRExporter(cdrs []*CDR, exportTemplate *config.CdreConfig, exportFormat, exportPath, fallbackPath, exportID string,
synchronous bool, attempts int, fieldSeparator rune, usageMultiplyFactor utils.FieldMultiplyFactor,
costMultiplyFactor float64, roundingDecimals int, httpSkipTlsCheck bool, httpPoster *utils.HTTPPoster) (*CDRExporter, error) {
if len(cdrs) == 0 { // Nothing to export
return nil, nil
}
cdre := &CdrExporter{
cdrs: cdrs,
cdrDb: cdrDb,
exportTemplate: exportTpl,
cdrFormat: cdrFormat,
fieldSeparator: fieldSeparator,
exportId: exportId,
dataUsageMultiplyFactor: dataUsageMultiplyFactor,
smsUsageMultiplyFactor: smsUsageMultiplyFactor,
mmsUsageMultiplyFactor: mmsUsageMultiplyFactor,
genericUsageMultiplyFactor: genericUsageMultiplyFactor,
costMultiplyFactor: costMultiplyFactor,
roundingDecimals: roundingDecimals,
httpSkipTlsCheck: httpSkipTlsCheck,
negativeExports: make(map[string]string),
}
if err := cdre.processCdrs(); err != nil {
return nil, err
cdre := &CDRExporter{
cdrs: cdrs,
exportTemplate: exportTemplate,
exportFormat: exportFormat,
exportPath: exportPath,
fallbackPath: fallbackPath,
exportID: exportID,
synchronous: synchronous,
attempts: attempts,
fieldSeparator: fieldSeparator,
usageMultiplyFactor: usageMultiplyFactor,
costMultiplyFactor: costMultiplyFactor,
roundingDecimals: roundingDecimals,
httpSkipTlsCheck: httpSkipTlsCheck,
httpPoster: httpPoster,
negativeExports: make(map[string]string),
}
return cdre, nil
}
type CdrExporter struct {
cdrs []*engine.CDR
cdrDb engine.CdrStorage // Used to extract cost_details if these are requested
exportTemplate *config.CdreConfig
cdrFormat string // csv, fwv
fieldSeparator rune
exportId string // Unique identifier or this export
dataUsageMultiplyFactor,
smsUsageMultiplyFactor, // Multiply the SMS usage (eg: some billing systems billing them as minutes)
mmsUsageMultiplyFactor,
genericUsageMultiplyFactor,
costMultiplyFactor float64
roundingDecimals int
httpSkipTlsCheck bool
header, trailer []string // Header and Trailer fields
content [][]string // Rows of cdr fields
type CDRExporter struct {
cdrs []*CDR
exportTemplate *config.CdreConfig
exportFormat string
exportPath string
fallbackPath string // folder where we save failed CDRs
exportID string // Unique identifier or this export
synchronous bool
attempts int
fieldSeparator rune
usageMultiplyFactor utils.FieldMultiplyFactor
costMultiplyFactor float64
roundingDecimals int
httpSkipTlsCheck bool
httpPoster *utils.HTTPPoster
header, trailer []string // Header and Trailer fields
content [][]string // Rows of cdr fields
firstCdrATime, lastCdrATime time.Time
numberOfRecords int
totalDuration, totalDataUsage, totalSmsUsage,
totalMmsUsage, totalGenericUsage time.Duration
totalCost float64
firstExpOrderId, lastExpOrderId int64
positiveExports []string // CGRIDs of successfully exported CDRs
pEMux sync.RWMutex // protect positiveExports
negativeExports map[string]string // CGRIDs of failed exports
nEMux sync.RWMutex // protect negativeExports
}
// Handle various meta functions used in header/trailer
func (cdre *CdrExporter) metaHandler(tag, arg string) (string, error) {
func (cdre *CDRExporter) metaHandler(tag, arg string) (string, error) {
switch tag {
case META_EXPORTID:
return cdre.exportId, nil
return cdre.exportID, nil
case META_TIMENOW:
return time.Now().Format(arg), nil
case META_FIRSTCDRATIME:
@@ -116,19 +120,19 @@ func (cdre *CdrExporter) metaHandler(tag, arg string) (string, error) {
case META_NRCDRS:
return strconv.Itoa(cdre.numberOfRecords), nil
case META_DURCDRS:
emulatedCdr := &engine.CDR{ToR: utils.VOICE, Usage: cdre.totalDuration}
emulatedCdr := &CDR{ToR: utils.VOICE, Usage: cdre.totalDuration}
return emulatedCdr.FormatUsage(arg), nil
case META_SMSUSAGE:
emulatedCdr := &engine.CDR{ToR: utils.SMS, Usage: cdre.totalSmsUsage}
emulatedCdr := &CDR{ToR: utils.SMS, Usage: cdre.totalSmsUsage}
return emulatedCdr.FormatUsage(arg), nil
case META_MMSUSAGE:
emulatedCdr := &engine.CDR{ToR: utils.MMS, Usage: cdre.totalMmsUsage}
emulatedCdr := &CDR{ToR: utils.MMS, Usage: cdre.totalMmsUsage}
return emulatedCdr.FormatUsage(arg), nil
case META_GENERICUSAGE:
emulatedCdr := &engine.CDR{ToR: utils.GENERIC, Usage: cdre.totalGenericUsage}
emulatedCdr := &CDR{ToR: utils.GENERIC, Usage: cdre.totalGenericUsage}
return emulatedCdr.FormatUsage(arg), nil
case META_DATAUSAGE:
emulatedCdr := &engine.CDR{ToR: utils.DATA, Usage: cdre.totalDataUsage}
emulatedCdr := &CDR{ToR: utils.DATA, Usage: cdre.totalDataUsage}
return emulatedCdr.FormatUsage(arg), nil
case META_COSTCDRS:
return strconv.FormatFloat(utils.Round(cdre.totalCost, cdre.roundingDecimals, utils.ROUNDING_MIDDLE), 'f', -1, 64), nil
@@ -138,7 +142,7 @@ func (cdre *CdrExporter) metaHandler(tag, arg string) (string, error) {
}
// Compose and cache the header
func (cdre *CdrExporter) composeHeader() error {
func (cdre *CDRExporter) composeHeader() (err error) {
for _, cfgFld := range cdre.exportTemplate.HeaderFields {
var outVal string
switch cfgFld.Type {
@@ -167,7 +171,7 @@ func (cdre *CdrExporter) composeHeader() error {
}
// Compose and cache the trailer
func (cdre *CdrExporter) composeTrailer() error {
func (cdre *CDRExporter) composeTrailer() (err error) {
for _, cfgFld := range cdre.exportTemplate.TrailerFields {
var outVal string
switch cfgFld.Type {
@@ -195,35 +199,96 @@ func (cdre *CdrExporter) composeTrailer() error {
return nil
}
func (cdre *CDRExporter) postCdr(cdr *CDR) (err error) {
var body interface{}
switch cdre.exportFormat {
case utils.MetaHTTPjsonCDR, utils.MetaAMQPjsonCDR:
jsn, err := json.Marshal(cdr)
if err != nil {
return err
}
body = jsn
case utils.MetaHTTPjsonMap, utils.MetaAMQPjsonMap:
expMp, err := cdr.AsExportMap(cdre.exportTemplate.ContentFields, cdre.httpSkipTlsCheck, nil, cdre.roundingDecimals)
if err != nil {
return err
}
jsn, err := json.Marshal(expMp)
if err != nil {
return err
}
body = jsn
case utils.META_HTTP_POST:
expMp, err := cdr.AsExportMap(cdre.exportTemplate.ContentFields, cdre.httpSkipTlsCheck, nil, cdre.roundingDecimals)
if err != nil {
return err
}
vals := url.Values{}
for fld, val := range expMp {
vals.Set(fld, val)
}
body = vals
default:
err = fmt.Errorf("unsupported exportFormat: <%s>", cdre.exportFormat)
}
if err != nil {
return
}
// compute fallbackPath
fallbackPath := utils.META_NONE
ffn := &utils.FallbackFileName{Module: utils.CDRPoster, Transport: cdre.exportFormat, Address: cdre.exportPath, RequestID: utils.GenUUID()}
fallbackFileName := ffn.AsString()
if cdre.fallbackPath != utils.META_NONE { // not none, need fallback
fallbackPath = path.Join(cdre.fallbackPath, fallbackFileName)
}
switch cdre.exportFormat {
case utils.MetaHTTPjsonCDR, utils.MetaHTTPjsonMap, utils.MetaHTTPjson, utils.META_HTTP_POST:
_, err = cdre.httpPoster.Post(cdre.exportPath, utils.PosterTransportContentTypes[cdre.exportFormat], body, cdre.attempts, fallbackPath)
case utils.MetaAMQPjsonCDR, utils.MetaAMQPjsonMap:
var amqpPoster *utils.AMQPPoster
amqpPoster, err = utils.AMQPPostersCache.GetAMQPPoster(cdre.exportPath, cdre.attempts, cdre.fallbackPath)
if err == nil { // error will be checked bellow
var chn *amqp.Channel
chn, err = amqpPoster.Post(
nil, utils.PosterTransportContentTypes[cdre.exportFormat], body.([]byte), fallbackFileName)
if chn != nil {
chn.Close()
}
}
}
return
}
// Write individual cdr into content buffer, build stats
func (cdre *CdrExporter) processCdr(cdr *engine.CDR) error {
if cdr == nil || len(cdr.CGRID) == 0 { // We do not export empty CDRs
return nil
} else if cdr.ExtraFields == nil { // Avoid assignment in nil map if not initialized
func (cdre *CDRExporter) processCdr(cdr *CDR) (err error) {
if cdr.ExtraFields == nil { // Avoid assignment in nil map if not initialized
cdr.ExtraFields = make(map[string]string)
}
// Cost multiply
if cdre.dataUsageMultiplyFactor != 0.0 && cdr.ToR == utils.DATA {
cdr.UsageMultiply(cdre.dataUsageMultiplyFactor, cdre.roundingDecimals)
} else if cdre.smsUsageMultiplyFactor != 0 && cdr.ToR == utils.SMS {
cdr.UsageMultiply(cdre.smsUsageMultiplyFactor, cdre.roundingDecimals)
} else if cdre.mmsUsageMultiplyFactor != 0 && cdr.ToR == utils.MMS {
cdr.UsageMultiply(cdre.mmsUsageMultiplyFactor, cdre.roundingDecimals)
} else if cdre.genericUsageMultiplyFactor != 0 && cdr.ToR == utils.GENERIC {
cdr.UsageMultiply(cdre.genericUsageMultiplyFactor, cdre.roundingDecimals)
// Usage multiply, find config based on ToR field or *any
for _, key := range []string{cdr.ToR, utils.ANY} {
if uM, hasIt := cdre.usageMultiplyFactor[key]; hasIt && uM != 1.0 {
cdr.UsageMultiply(uM, cdre.roundingDecimals)
break
}
}
if cdre.costMultiplyFactor != 0.0 {
cdr.CostMultiply(cdre.costMultiplyFactor, cdre.roundingDecimals)
}
cdrRow, err := cdr.AsExportRecord(cdre.exportTemplate.ContentFields, cdre.httpSkipTlsCheck, cdre.cdrs, cdre.roundingDecimals)
if err != nil {
utils.Logger.Err(fmt.Sprintf("<CdreFw> Cannot export CDR with CGRID: %s and runid: %s, error: %s", cdr.CGRID, cdr.RunID, err.Error()))
return err
switch cdre.exportFormat {
case utils.MetaFileFWV, utils.MetaFileCSV:
var cdrRow []string
cdrRow, err = cdr.AsExportRecord(cdre.exportTemplate.ContentFields, cdre.httpSkipTlsCheck, cdre.cdrs, cdre.roundingDecimals)
if len(cdrRow) == 0 { // No CDR data, most likely no configuration fields defined
return
} else {
cdre.content = append(cdre.content, cdrRow)
}
default: // attempt posting CDR
err = cdre.postCdr(cdr)
}
if len(cdrRow) == 0 { // No CDR data, most likely no configuration fields defined
return nil
} else {
cdre.content = append(cdre.content, cdrRow)
if err != nil {
utils.Logger.Err(fmt.Sprintf("<CDRE> Cannot export CDR with CGRID: %s and runid: %s, error: %s", cdr.CGRID, cdr.RunID, err.Error()))
return
}
// Done with writing content, compute stats here
if cdre.firstCdrATime.IsZero() || cdr.AnswerTime.Before(cdre.firstCdrATime) {
@@ -262,14 +327,41 @@ func (cdre *CdrExporter) processCdr(cdr *engine.CDR) error {
}
// Builds header, content and trailers
func (cdre *CdrExporter) processCdrs() error {
func (cdre *CDRExporter) processCdrs() error {
var wg sync.WaitGroup
for _, cdr := range cdre.cdrs {
if err := cdre.processCdr(cdr); err != nil {
cdre.negativeExports[cdr.CGRID] = err.Error()
} else {
cdre.positiveExports = append(cdre.positiveExports, cdr.CGRID)
if cdr == nil || len(cdr.CGRID) == 0 { // CDR needs to exist and it's CGRID needs to be populated
continue
}
passesFilters := true
for _, cdfFltr := range cdre.exportTemplate.CDRFilter {
if !cdfFltr.FilterPasses(cdr.FieldAsString(cdfFltr)) {
passesFilters = false
break
}
}
if !passesFilters { // Not passes filters, ignore this CDR
continue
}
if cdre.synchronous {
wg.Add(1)
}
go func(cdr *CDR) {
if err := cdre.processCdr(cdr); err != nil {
cdre.nEMux.Lock()
cdre.negativeExports[cdr.CGRID] = err.Error()
cdre.nEMux.Unlock()
} else {
cdre.pEMux.Lock()
cdre.positiveExports = append(cdre.positiveExports, cdr.CGRID)
cdre.pEMux.Unlock()
}
if cdre.synchronous {
wg.Done()
}
}(cdr)
}
wg.Wait()
// Process header and trailer after processing cdrs since the metatag functions can access stats out of built cdrs
if cdre.exportTemplate.HeaderFields != nil {
if err := cdre.composeHeader(); err != nil {
@@ -285,7 +377,7 @@ func (cdre *CdrExporter) processCdrs() error {
}
// Simple write method
func (cdre *CdrExporter) writeOut(ioWriter io.Writer) error {
func (cdre *CDRExporter) writeOut(ioWriter io.Writer) error {
if len(cdre.header) != 0 {
for _, fld := range append(cdre.header, "\n") {
if _, err := io.WriteString(ioWriter, fld); err != nil {
@@ -311,7 +403,7 @@ func (cdre *CdrExporter) writeOut(ioWriter io.Writer) error {
}
// csvWriter specific method
func (cdre *CdrExporter) writeCsv(csvWriter *csv.Writer) error {
func (cdre *CDRExporter) writeCsv(csvWriter *csv.Writer) error {
csvWriter.Comma = cdre.fieldSeparator
if len(cdre.header) != 0 {
if err := csvWriter.Write(cdre.header); err != nil {
@@ -332,54 +424,57 @@ func (cdre *CdrExporter) writeCsv(csvWriter *csv.Writer) error {
return nil
}
// General method to write the content out to a file
func (cdre *CdrExporter) WriteToFile(filePath string) error {
fileOut, err := os.Create(filePath)
if err != nil {
return err
func (cdre *CDRExporter) ExportCDRs() (err error) {
if err = cdre.processCdrs(); err != nil {
return
}
defer fileOut.Close()
switch cdre.cdrFormat {
case utils.DRYRUN:
return nil
case utils.CDRE_FIXED_WIDTH:
if err := cdre.writeOut(fileOut); err != nil {
return utils.NewErrServerError(err)
switch cdre.exportFormat {
case utils.MetaFileFWV, utils.MetaFileCSV:
if len(cdre.content) == 0 {
return
}
case utils.CSV:
csvWriter := csv.NewWriter(fileOut)
if err := cdre.writeCsv(csvWriter); err != nil {
return utils.NewErrServerError(err)
fileOut, err := os.Create(cdre.exportPath)
if err != nil {
return err
}
defer fileOut.Close()
if cdre.exportFormat == utils.MetaFileCSV {
return cdre.writeCsv(csv.NewWriter(fileOut))
}
return cdre.writeOut(fileOut)
}
return nil
return
}
// Return the first exported Cdr OrderId
func (cdre *CdrExporter) FirstOrderId() int64 {
func (cdre *CDRExporter) FirstOrderId() int64 {
return cdre.firstExpOrderId
}
// Return the last exported Cdr OrderId
func (cdre *CdrExporter) LastOrderId() int64 {
func (cdre *CDRExporter) LastOrderId() int64 {
return cdre.lastExpOrderId
}
// Return total cost in the exported cdrs
func (cdre *CdrExporter) TotalCost() float64 {
func (cdre *CDRExporter) TotalCost() float64 {
return cdre.totalCost
}
func (cdre *CdrExporter) TotalExportedCdrs() int {
func (cdre *CDRExporter) TotalExportedCdrs() int {
return cdre.numberOfRecords
}
// Return successfully exported CGRIDs
func (cdre *CdrExporter) PositiveExports() []string {
func (cdre *CDRExporter) PositiveExports() []string {
cdre.pEMux.RLock()
defer cdre.pEMux.RUnlock()
return cdre.positiveExports
}
// Return failed exported CGRIDs together with the reason
func (cdre *CdrExporter) NegativeExports() map[string]string {
func (cdre *CDRExporter) NegativeExports() map[string]string {
cdre.nEMux.RLock()
defer cdre.nEMux.RUnlock()
return cdre.negativeExports
}

20
engine/cdrecsv_test.go
View File

@@ -15,7 +15,7 @@ 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 cdre
package engine
import (
"bytes"
@@ -25,24 +25,27 @@ import (
"time"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
)
func TestCsvCdrWriter(t *testing.T) {
writer := &bytes.Buffer{}
cfg, _ := config.NewDefaultCGRConfig()
storedCdr1 := &engine.CDR{
storedCdr1 := &CDR{
CGRID: utils.Sha1("dsafdsaf", time.Unix(1383813745, 0).UTC().String()), ToR: utils.VOICE, OriginID: "dsafdsaf", OriginHost: "192.168.1.1",
RequestType: utils.META_RATED, Direction: "*out", Tenant: "cgrates.org",
Category: "call", Account: "1001", Subject: "1001", Destination: "1002", SetupTime: time.Unix(1383813745, 0).UTC(), AnswerTime: time.Unix(1383813746, 0).UTC(),
Usage: time.Duration(10) * time.Second, RunID: utils.DEFAULT_RUNID,
ExtraFields: map[string]string{"extra1": "val_extra1", "extra2": "val_extra2", "extra3": "val_extra3"}, Cost: 1.01,
}
cdre, err := NewCdrExporter([]*engine.CDR{storedCdr1}, nil, cfg.CdreProfiles["*default"], utils.CSV, ',', "firstexport", 0.0, 0.0, 0.0, 0.0, 0.0, cfg.RoundingDecimals, cfg.HttpSkipTlsVerify)
cdre, err := NewCDRExporter([]*CDR{storedCdr1}, cfg.CdreProfiles["*default"], utils.MetaFileCSV, "", "", "firstexport",
true, 1, ',', map[string]float64{}, 0.0, cfg.RoundingDecimals, cfg.HttpSkipTlsVerify, nil)
if err != nil {
t.Error("Unexpected error received: ", err)
}
if err = cdre.processCdrs(); err != nil {
t.Error(err)
}
csvWriter := csv.NewWriter(writer)
if err := cdre.writeCsv(csvWriter); err != nil {
t.Error("Unexpected error: ", err)
@@ -60,17 +63,20 @@ func TestCsvCdrWriter(t *testing.T) {
func TestAlternativeFieldSeparator(t *testing.T) {
writer := &bytes.Buffer{}
cfg, _ := config.NewDefaultCGRConfig()
storedCdr1 := &engine.CDR{CGRID: utils.Sha1("dsafdsaf", time.Unix(1383813745, 0).UTC().String()), ToR: utils.VOICE, OriginID: "dsafdsaf", OriginHost: "192.168.1.1",
storedCdr1 := &CDR{CGRID: utils.Sha1("dsafdsaf", time.Unix(1383813745, 0).UTC().String()), ToR: utils.VOICE, OriginID: "dsafdsaf", OriginHost: "192.168.1.1",
RequestType: utils.META_RATED, Direction: "*out", Tenant: "cgrates.org",
Category: "call", Account: "1001", Subject: "1001", Destination: "1002", SetupTime: time.Unix(1383813745, 0).UTC(), AnswerTime: time.Unix(1383813746, 0).UTC(),
Usage: time.Duration(10) * time.Second, RunID: utils.DEFAULT_RUNID,
ExtraFields: map[string]string{"extra1": "val_extra1", "extra2": "val_extra2", "extra3": "val_extra3"}, Cost: 1.01,
}
cdre, err := NewCdrExporter([]*engine.CDR{storedCdr1}, nil, cfg.CdreProfiles["*default"], utils.CSV, '|',
"firstexport", 0.0, 0.0, 0.0, 0.0, 0.0, cfg.RoundingDecimals, cfg.HttpSkipTlsVerify)
cdre, err := NewCDRExporter([]*CDR{storedCdr1}, cfg.CdreProfiles["*default"], utils.MetaFileCSV, "", "", "firstexport",
true, 1, '|', map[string]float64{}, 0.0, cfg.RoundingDecimals, cfg.HttpSkipTlsVerify, nil)
if err != nil {
t.Error("Unexpected error received: ", err)
}
if err = cdre.processCdrs(); err != nil {
t.Error(err)
}
csvWriter := csv.NewWriter(writer)
if err := cdre.writeCsv(csvWriter); err != nil {
t.Error("Unexpected error: ", err)

32
engine/cdrefwv_test.go
View File

@@ -15,7 +15,7 @@ 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 cdre
package engine
import (
"bytes"
@@ -24,7 +24,6 @@ import (
"time"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/engine"
"github.com/cgrates/cgrates/utils"
)
@@ -111,12 +110,12 @@ func TestWriteCdr(t *testing.T) {
t.Error(err)
}
cdreCfg := &config.CdreConfig{
CdrFormat: utils.CDRE_FIXED_WIDTH,
ExportFormat: utils.MetaFileFWV,
HeaderFields: hdrCfgFlds,
ContentFields: contentCfgFlds,
TrailerFields: trailerCfgFlds,
}
cdr := &engine.CDR{CGRID: utils.Sha1("dsafdsaf", time.Date(2013, 11, 7, 8, 42, 20, 0, time.UTC).String()),
cdr := &CDR{CGRID: utils.Sha1("dsafdsaf", time.Date(2013, 11, 7, 8, 42, 20, 0, time.UTC).String()),
ToR: utils.VOICE, OrderID: 1, OriginID: "dsafdsaf", OriginHost: "192.168.1.1",
RequestType: utils.META_RATED, Direction: "*out", Tenant: "cgrates.org",
Category: "call", Account: "1001", Subject: "1001", Destination: "1002",
@@ -125,11 +124,15 @@ func TestWriteCdr(t *testing.T) {
Usage: time.Duration(10) * time.Second, RunID: utils.DEFAULT_RUNID, Cost: 2.34567,
ExtraFields: map[string]string{"field_extr1": "val_extr1", "fieldextr2": "valextr2"},
}
cdre, err := NewCdrExporter([]*engine.CDR{cdr}, nil, cdreCfg, utils.CDRE_FIXED_WIDTH, ',', "fwv_1", 0.0, 0.0, 0.0, 0.0, 0.0,
cfg.RoundingDecimals, cfg.HttpSkipTlsVerify)
cdre, err := NewCDRExporter([]*CDR{cdr}, cdreCfg, utils.MetaFileFWV, "", "", "fwv_1",
true, 1, '|', map[string]float64{}, 0.0, cfg.RoundingDecimals, cfg.HttpSkipTlsVerify, nil)
if err != nil {
t.Error(err)
}
if err = cdre.processCdrs(); err != nil {
t.Error(err)
}
eHeader := "10 VOIfwv_107111308420018011511340001 \n"
eContentOut := "201001 1001 1002 0211 07111308420010 1 3dsafdsaf 0002.34570\n"
eTrailer := "90 VOIfwv_100000100000010071113084200071113084200 \n"
@@ -169,12 +172,12 @@ func TestWriteCdr(t *testing.T) {
func TestWriteCdrs(t *testing.T) {
wrBuf := &bytes.Buffer{}
cdreCfg := &config.CdreConfig{
CdrFormat: utils.CDRE_FIXED_WIDTH,
ExportFormat: utils.MetaFileFWV,
HeaderFields: hdrCfgFlds,
ContentFields: contentCfgFlds,
TrailerFields: trailerCfgFlds,
}
cdr1 := &engine.CDR{CGRID: utils.Sha1("aaa1", time.Date(2013, 11, 7, 8, 42, 20, 0, time.UTC).String()),
cdr1 := &CDR{CGRID: utils.Sha1("aaa1", time.Date(2013, 11, 7, 8, 42, 20, 0, time.UTC).String()),
ToR: utils.VOICE, OrderID: 2, OriginID: "aaa1", OriginHost: "192.168.1.1", RequestType: utils.META_RATED, Direction: "*out", Tenant: "cgrates.org",
Category: "call", Account: "1001", Subject: "1001", Destination: "1010",
SetupTime: time.Date(2013, 11, 7, 8, 42, 20, 0, time.UTC),
@@ -182,7 +185,7 @@ func TestWriteCdrs(t *testing.T) {
Usage: time.Duration(10) * time.Second, RunID: utils.DEFAULT_RUNID, Cost: 2.25,
ExtraFields: map[string]string{"productnumber": "12341", "fieldextr2": "valextr2"},
}
cdr2 := &engine.CDR{CGRID: utils.Sha1("aaa2", time.Date(2013, 11, 7, 7, 42, 20, 0, time.UTC).String()),
cdr2 := &CDR{CGRID: utils.Sha1("aaa2", time.Date(2013, 11, 7, 7, 42, 20, 0, time.UTC).String()),
ToR: utils.VOICE, OrderID: 4, OriginID: "aaa2", OriginHost: "192.168.1.2", RequestType: utils.META_PREPAID, Direction: "*out", Tenant: "cgrates.org",
Category: "call", Account: "1002", Subject: "1002", Destination: "1011",
SetupTime: time.Date(2013, 11, 7, 7, 42, 20, 0, time.UTC),
@@ -190,8 +193,8 @@ func TestWriteCdrs(t *testing.T) {
Usage: time.Duration(5) * time.Minute, RunID: utils.DEFAULT_RUNID, Cost: 1.40001,
ExtraFields: map[string]string{"productnumber": "12342", "fieldextr2": "valextr2"},
}
cdr3 := &engine.CDR{}
cdr4 := &engine.CDR{CGRID: utils.Sha1("aaa3", time.Date(2013, 11, 7, 9, 42, 18, 0, time.UTC).String()),
cdr3 := &CDR{}
cdr4 := &CDR{CGRID: utils.Sha1("aaa3", time.Date(2013, 11, 7, 9, 42, 18, 0, time.UTC).String()),
ToR: utils.VOICE, OrderID: 3, OriginID: "aaa4", OriginHost: "192.168.1.4", RequestType: utils.META_POSTPAID, Direction: "*out", Tenant: "cgrates.org",
Category: "call", Account: "1004", Subject: "1004", Destination: "1013",
SetupTime: time.Date(2013, 11, 7, 9, 42, 18, 0, time.UTC),
@@ -200,11 +203,14 @@ func TestWriteCdrs(t *testing.T) {
ExtraFields: map[string]string{"productnumber": "12344", "fieldextr2": "valextr2"},
}
cfg, _ := config.NewDefaultCGRConfig()
cdre, err := NewCdrExporter([]*engine.CDR{cdr1, cdr2, cdr3, cdr4}, nil, cdreCfg, utils.CDRE_FIXED_WIDTH, ',',
"fwv_1", 0.0, 0.0, 0.0, 0.0, 0.0, cfg.RoundingDecimals, cfg.HttpSkipTlsVerify)
cdre, err := NewCDRExporter([]*CDR{cdr1, cdr2, cdr3, cdr4}, cdreCfg, utils.MetaFileFWV, "", "", "fwv_1",
true, 1, ',', map[string]float64{}, 0.0, cfg.RoundingDecimals, cfg.HttpSkipTlsVerify, nil)
if err != nil {
t.Error(err)
}
if err = cdre.processCdrs(); err != nil {
t.Error(err)
}
if err := cdre.writeOut(wrBuf); err != nil {
t.Error(err)
}

178
engine/cdrs.go
View File

@@ -18,12 +18,9 @@ along with this program. If not, see <http://www.gnu.org/licenses/>
package engine
import (
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"net/url"
"path"
"reflect"
"strings"
"time"
@@ -33,7 +30,6 @@ import (
"github.com/cgrates/cgrates/guardian"
"github.com/cgrates/cgrates/utils"
"github.com/cgrates/rpcclient"
"github.com/streadway/amqp"
)
var cdrServer *CdrServer // Share the server so we can use it in http handlers
@@ -195,12 +191,12 @@ func (self *CdrServer) processCdr(cdr *CDR) (err error) {
var out int
go self.stats.Call("CDRStatsV1.AppendCDR", cdr, &out)
}
if len(self.cgrCfg.CDRSCdrReplication) != 0 { // Replicate raw CDR
go self.replicateCdr(cdr)
if len(self.cgrCfg.CDRSOnlineCDRExports) != 0 { // Replicate raw CDR
self.replicateCdr(cdr)
}
if self.rals != nil && !cdr.Rated { // CDRs not rated will be processed by Rating
go self.deriveRateStoreStatsReplicate(cdr, self.cgrCfg.CDRSStoreCdrs, self.stats != nil, len(self.cgrCfg.CDRSCdrReplication) != 0)
self.deriveRateStoreStatsReplicate(cdr, self.cgrCfg.CDRSStoreCdrs, self.stats != nil, len(self.cgrCfg.CDRSOnlineCDRExports) != 0)
}
return nil
}
@@ -452,89 +448,91 @@ func (self *CdrServer) getCostFromRater(cdr *CDR) (*CallCost, error) {
return cc, nil
}
// ToDo: Add websocket support
func (self *CdrServer) replicateCdr(cdr *CDR) error {
for _, rplCfg := range self.cgrCfg.CDRSCdrReplication {
passesFilters := true
for _, cdfFltr := range rplCfg.CdrFilter {
if !cdfFltr.FilterPasses(cdr.FieldAsString(cdfFltr)) {
passesFilters = false
break
}
}
if !passesFilters { // Not passes filters, ignore this replication
continue
}
var body interface{}
var content = ""
switch rplCfg.Transport {
case utils.MetaHTTPjsonCDR, utils.MetaAMQPjsonCDR:
jsn, err := json.Marshal(cdr)
if err != nil {
return err
}
body = jsn
case utils.MetaHTTPjsonMap, utils.MetaAMQPjsonMap:
expMp, err := cdr.AsExportMap(rplCfg.ContentFields, self.cgrCfg.HttpSkipTlsVerify, nil)
if err != nil {
return err
}
jsn, err := json.Marshal(expMp)
if err != nil {
return err
}
body = jsn
case utils.META_HTTP_POST:
expMp, err := cdr.AsExportMap(rplCfg.ContentFields, self.cgrCfg.HttpSkipTlsVerify, nil)
if err != nil {
return err
}
vals := url.Values{}
for fld, val := range expMp {
vals.Set(fld, val)
}
body = vals
}
var errChan chan error
if rplCfg.Synchronous {
errChan = make(chan error)
}
go func(body interface{}, rplCfg *config.CDRReplicationCfg, content string, errChan chan error) {
var err error
fallbackPath := utils.META_NONE
if rplCfg.FallbackFileName() != utils.META_NONE {
fallbackPath = path.Join(self.cgrCfg.FailedPostsDir, rplCfg.FallbackFileName())
}
switch rplCfg.Transport {
case utils.MetaHTTPjsonCDR, utils.MetaHTTPjsonMap, utils.MetaHTTPjson, utils.META_HTTP_POST:
_, err = self.httpPoster.Post(rplCfg.Address, utils.PosterTransportContentTypes[rplCfg.Transport], body, rplCfg.Attempts, fallbackPath)
case utils.MetaAMQPjsonCDR, utils.MetaAMQPjsonMap:
var amqpPoster *utils.AMQPPoster
amqpPoster, err = utils.AMQPPostersCache.GetAMQPPoster(rplCfg.Address, rplCfg.Attempts, self.cgrCfg.FailedPostsDir)
if err == nil { // error will be checked bellow
var chn *amqp.Channel
chn, err = amqpPoster.Post(
nil, utils.PosterTransportContentTypes[rplCfg.Transport], body.([]byte), rplCfg.FallbackFileName())
if chn != nil {
chn.Close()
}
}
default:
err = fmt.Errorf("unsupported replication transport: %s", rplCfg.Transport)
}
if err != nil {
utils.Logger.Err(fmt.Sprintf(
"<CDRReplicator> Replicating CDR: %+v, transport: %s, got error: %s", cdr, rplCfg.Transport, err.Error()))
}
if rplCfg.Synchronous {
errChan <- err
}
}(body, rplCfg, content, errChan)
if rplCfg.Synchronous { // Synchronize here
<-errChan
}
}
return nil
/*
for _, rplCfg := range self.cgrCfg.CDRSOnlineCDRExports {
passesFilters := true
for _, cdfFltr := range rplCfg.CdrFilter {
if !cdfFltr.FilterPasses(cdr.FieldAsString(cdfFltr)) {
passesFilters = false
break
}
}
if !passesFilters { // Not passes filters, ignore this replication
continue
}
var body interface{}
var content = ""
switch rplCfg.Transport {
case utils.MetaHTTPjsonCDR, utils.MetaAMQPjsonCDR:
jsn, err := json.Marshal(cdr)
if err != nil {
return err
}
body = jsn
case utils.MetaHTTPjsonMap, utils.MetaAMQPjsonMap:
expMp, err := cdr.AsExportMap(rplCfg.ContentFields, self.cgrCfg.HttpSkipTlsVerify, nil)
if err != nil {
return err
}
jsn, err := json.Marshal(expMp)
if err != nil {
return err
}
body = jsn
case utils.META_HTTP_POST:
expMp, err := cdr.AsExportMap(rplCfg.ContentFields, self.cgrCfg.HttpSkipTlsVerify, nil)
if err != nil {
return err
}
vals := url.Values{}
for fld, val := range expMp {
vals.Set(fld, val)
}
body = vals
}
var errChan chan error
if rplCfg.Synchronous {
errChan = make(chan error)
}
go func(body interface{}, rplCfg *config.CDRReplicationCfg, content string, errChan chan error) {
var err error
fallbackPath := utils.META_NONE
if rplCfg.FallbackFileName() != utils.META_NONE {
fallbackPath = path.Join(self.cgrCfg.FailedPostsDir, rplCfg.FallbackFileName())
}
switch rplCfg.Transport {
case utils.MetaHTTPjsonCDR, utils.MetaHTTPjsonMap, utils.MetaHTTPjson, utils.META_HTTP_POST:
_, err = self.httpPoster.Post(rplCfg.Address, utils.PosterTransportContentTypes[rplCfg.Transport], body, rplCfg.Attempts, fallbackPath)
case utils.MetaAMQPjsonCDR, utils.MetaAMQPjsonMap:
var amqpPoster *utils.AMQPPoster
amqpPoster, err = utils.AMQPPostersCache.GetAMQPPoster(rplCfg.Address, rplCfg.Attempts, self.cgrCfg.FailedPostsDir)
if err == nil { // error will be checked bellow
var chn *amqp.Channel
chn, err = amqpPoster.Post(
nil, utils.PosterTransportContentTypes[rplCfg.Transport], body.([]byte), rplCfg.FallbackFileName())
if chn != nil {
chn.Close()
}
}
default:
err = fmt.Errorf("unsupported replication transport: %s", rplCfg.Transport)
}
if err != nil {
utils.Logger.Err(fmt.Sprintf(
"<CDRReplicator> Replicating CDR: %+v, transport: %s, got error: %s", cdr, rplCfg.Transport, err.Error()))
}
if rplCfg.Synchronous {
errChan <- err
}
}(body, rplCfg, content, errChan)
if rplCfg.Synchronous { // Synchronize here
<-errChan
}
}
return nil
*/
}
// Called by rate/re-rate API, FixMe: deprecate it once new APIer structure is operational
@@ -544,7 +542,7 @@ func (self *CdrServer) RateCDRs(cdrFltr *utils.CDRsFilter, sendToStats bool) err
return err
}
for _, cdr := range cdrs {
if err := self.deriveRateStoreStatsReplicate(cdr, self.cgrCfg.CDRSStoreCdrs, sendToStats, len(self.cgrCfg.CDRSCdrReplication) != 0); err != nil {
if err := self.deriveRateStoreStatsReplicate(cdr, self.cgrCfg.CDRSStoreCdrs, sendToStats, len(self.cgrCfg.CDRSOnlineCDRExports) != 0); err != nil {
utils.Logger.Err(fmt.Sprintf("<CDRS> Processing CDR %+v, got error: %s", cdr, err.Error()))
}
}
@@ -614,7 +612,7 @@ func (self *CdrServer) V1RateCDRs(attrs utils.AttrRateCDRs, reply *string) error
if attrs.SendToStatS != nil {
sendToStats = *attrs.SendToStatS
}
replicate := len(self.cgrCfg.CDRSCdrReplication) != 0
replicate := len(self.cgrCfg.CDRSOnlineCDRExports) != 0
if attrs.ReplicateCDRs != nil {
replicate = *attrs.ReplicateCDRs
}

15
utils/consts.go
View File

@@ -18,7 +18,7 @@ along with this program. If not, see <http://www.gnu.org/licenses/>
package utils
var (
CdreCdrFormats = []string{CSV, DRYRUN, CDRE_FIXED_WIDTH}
CDRExportFormats = []string{DRYRUN, MetaFileCSV, MetaFileFWV, MetaHTTPjsonCDR, MetaHTTPjsonMap, MetaHTTPjson, META_HTTP_POST, MetaAMQPjsonCDR, MetaAMQPjsonMap}
PrimaryCdrFields = []string{CGRID, CDRSOURCE, CDRHOST, ACCID, TOR, REQTYPE, DIRECTION, TENANT, CATEGORY, ACCOUNT, SUBJECT, DESTINATION, SETUP_TIME, PDD, ANSWER_TIME, USAGE,
SUPPLIER, DISCONNECT_CAUSE, COST, RATED, PartialField, MEDI_RUNID}
GitLastLog string // If set, it will be processed as part of versioning
@@ -30,6 +30,15 @@ var (
MetaAMQPjsonCDR: CONTENT_JSON,
MetaAMQPjsonMap: CONTENT_JSON,
}
CDREFileSuffixes = map[string]string{
MetaHTTPjsonCDR: JSNSuffix,
MetaHTTPjsonMap: JSNSuffix,
MetaAMQPjsonCDR: JSNSuffix,
MetaAMQPjsonMap: JSNSuffix,
META_HTTP_POST: FormSuffix,
MetaFileCSV: CSVSuffix,
MetaFileFWV: FWVSuffix,
}
)
const (
@@ -348,10 +357,14 @@ const (
TxtSuffix = ".txt"
JSNSuffix = ".json"
FormSuffix = ".form"
CSVSuffix = ".csv"
FWVSuffix = ".fwv"
CONTENT_JSON = "json"
CONTENT_FORM = "form"
CONTENT_TEXT = "text"
FileLockPrefix = "file_"
ActionsPoster = "act"
CDRPoster = "cdr"
MetaFileCSV = "*file_csv"
MetaFileFWV = "*file_fwv"
)

3
utils/poster.go
View File

@@ -97,6 +97,9 @@ type FallbackFileName struct {
}
func (ffn *FallbackFileName) AsString() string {
if ffn.FileSuffix == "" { // Autopopulate FileSuffix based on the transport used
ffn.FileSuffix = CDREFileSuffixes[ffn.Transport]
}
return fmt.Sprintf("%s%s%s%s%s%s%s%s", ffn.Module, HandlerArgSep, ffn.Transport, HandlerArgSep, url.QueryEscape(ffn.Address), HandlerArgSep, ffn.RequestID, ffn.FileSuffix)
}