uppal/cgrates
1
0
Fork 1
mirror of https://github.com/cgrates/cgrates.git synced 2026-02-24 08:38:45 +05:00
Code Issues Packages Projects Releases Wiki Activity

Move cdrs to its own package

Browse Source 
Moved engine/cdr.go to utils/cdr.go.
Moved engine/cdrs.go and engine/cdrs_test.go to cdrs package.
Moved api functions from cdrs/cdrs.go to cdrs/apis.go. Same for the
tests.
Deleted CDRsV1 type and its methods.
This commit is contained in:
ionutboangiu
2023-12-11 07:17:40 -05:00
committed by Dan Christian Bogos
parent 1cec3751e4
commit d29f84169c
19 changed files with 1290 additions and 1408 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
engine/cdr.go
View File

@@ -1,126 +0,0 @@
/*
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 engine
import (
"database/sql/driver"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/cgrates/cgrates/utils"
)
type CDR struct {
Tenant string
Opts map[string]interface{}
Event map[string]interface{}
CreatedAt time.Time `json:",omitempty"`
UpdatedAt time.Time `json:",omitempty"`
DeletedAt *time.Time `json:",omitempty"`
}
type CDRSQLTable struct {
ID int64 // this is used for incrementing while seting
Tenant string
Opts JSON `gorm:"type:jsonb"` //string
Event JSON `gorm:"type:jsonb"` //string
CreatedAt time.Time `json:",omitempty"`
UpdatedAt time.Time `json:",omitempty"`
DeletedAt *time.Time `json:",omitempty"`
}
func (CDRSQLTable) TableName() string {
return utils.CDRsTBL
}
// JSON type for storing maps of events and opts into gorm columns as jsob type
type JSON map[string]interface{}
func (j JSON) GormDataType() string {
return "JSONB"
}
// Scan scan value into Jsonb, implements sql.Scanner interface
func (j *JSON) Scan(value interface{}) (err error) {
switch v := value.(type) {
case []byte:
return json.Unmarshal(v, &j)
case string:
return json.Unmarshal([]byte(v), &j)
default:
return errors.New(fmt.Sprint("Failed to unmarshal JSONB value:", value))
}
}
// Value return json value, implement driver.Valuer interface
func (j JSON) Value() (driver.Value, error) {
return json.Marshal(j)
}
func GetUniqueCDRID(cgrEv *utils.CGREvent) string {
if chargeId, ok := cgrEv.APIOpts[utils.MetaChargeID]; ok {
return utils.IfaceAsString(chargeId)
}
if originID, ok := cgrEv.APIOpts[utils.MetaOriginID]; ok {
return utils.IfaceAsString(originID)
}
return utils.UUIDSha1Prefix()
}
func (cdr *CDR) CGREvent() *utils.CGREvent {
return &utils.CGREvent{
Tenant: cdr.Tenant,
ID: utils.Sha1(),
Event: cdr.Event,
APIOpts: cdr.Opts,
}
}
// CDRsToCGREvents converts a slice of *CDR to a slice of *utils.CGREvent.
func CDRsToCGREvents(cdrs []*CDR) []*utils.CGREvent {
cgrEvs := make([]*utils.CGREvent, 0, len(cdrs))
for _, cdr := range cdrs {
cgrEvs = append(cgrEvs, cdr.CGREvent())
}
return cgrEvs
}
// checkNestedFields checks if there are elements or values nested (e.g *opts.*rateSCost.Cost)
func checkNestedFields(elem string, values []string) bool {
if len(strings.Split(elem, utils.NestingSep)) > 2 {
return true
}
for _, val := range values {
if len(strings.Split(val, utils.NestingSep)) > 2 {
return true
}
}
return false
}
type CDRFilters struct {
Tenant string
ID string
FilterIDs []string
APIOpts map[string]interface{}
}

545
engine/cdrs.go
View File

@@ -1,545 +0,0 @@
/*
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 engine
import (
"encoding/json"
"errors"
"fmt"
"net/http"
"github.com/cgrates/birpc/context"
"github.com/cgrates/cgrates/config"
"github.com/cgrates/cgrates/guardian"
"github.com/cgrates/cgrates/utils"
)
func newMapEventFromReqForm(r *http.Request) (mp MapEvent, err error) {
if r.Form == nil {
if err = r.ParseForm(); err != nil {
return
}
}
mp = MapEvent{utils.Source: r.RemoteAddr}
for k, vals := range r.Form {
mp[k] = vals[0] // We only support the first value for now, if more are provided it is considered remote's fault
}
return
}
// NewCDRServer is a constructor for CDRServer
func NewCDRServer(cfg *config.CGRConfig, dm *DataManager, filterS *FilterS, connMgr *ConnManager,
storDBChan chan StorDB) *CDRServer {
cdrDB := <-storDBChan
return &CDRServer{
cfg: cfg,
dm: dm,
db: cdrDB,
guard: guardian.Guardian,
fltrS: filterS,
connMgr: connMgr,
dbChan: storDBChan,
}
}
// CDRServer stores and rates CDRs
type CDRServer struct {
cfg *config.CGRConfig
dm *DataManager
db StorDB
guard *guardian.GuardianLocker
fltrS *FilterS
connMgr *ConnManager
dbChan chan StorDB
}
// ListenAndServe listen for storbd reload
func (cdrS *CDRServer) ListenAndServe(stopChan chan struct{}) {
for {
select {
case <-stopChan:
return
case _, ok := <-cdrS.dbChan:
if !ok { // the channel was closed by the shutdown of the StorDB Service
return
}
}
}
}
// chrgrSProcessEvent forks CGREventWithOpts into multiples based on matching ChargerS profiles
func (cdrS *CDRServer) chrgrSProcessEvent(ctx *context.Context, cgrEv *utils.CGREvent) (cgrEvs []*utils.CGREvent, err error) {
var chrgrs []*ChrgSProcessEventReply
if err = cdrS.connMgr.Call(ctx, cdrS.cfg.CdrsCfg().ChargerSConns,
utils.ChargerSv1ProcessEvent,
cgrEv, &chrgrs); err != nil {
return
}
if len(chrgrs) == 0 {
return
}
cgrEvs = make([]*utils.CGREvent, len(chrgrs))
for i, cgrPrfl := range chrgrs {
cgrEvs[i] = cgrPrfl.CGREvent
}
return
}
// attrSProcessEvent will send the event to StatS if the connection is configured
func (cdrS *CDRServer) attrSProcessEvent(ctx *context.Context, cgrEv *utils.CGREvent) (err error) {
var rplyEv AttrSProcessEventReply
if cgrEv.APIOpts == nil {
cgrEv.APIOpts = make(map[string]any)
}
cgrEv.APIOpts[utils.MetaSubsys] = utils.MetaCDRs
cgrEv.APIOpts[utils.OptsContext] = utils.FirstNonEmpty(
utils.IfaceAsString(cgrEv.APIOpts[utils.OptsContext]),
utils.MetaCDRs)
if err = cdrS.connMgr.Call(ctx, cdrS.cfg.CdrsCfg().AttributeSConns,
utils.AttributeSv1ProcessEvent,
cgrEv, &rplyEv); err == nil && len(rplyEv.AlteredFields) != 0 {
*cgrEv = *rplyEv.CGREvent
} else if err != nil &&
err.Error() == utils.ErrNotFound.Error() {
err = nil // cancel ErrNotFound
}
return
}
// rateSProcessEvent will send the event to rateS and attach the cost received back to event
func (cdrS *CDRServer) rateSCostForEvent(ctx *context.Context, cgrEv *utils.CGREvent) (err error) {
var rpCost utils.RateProfileCost
if err = cdrS.connMgr.Call(ctx, cdrS.cfg.CdrsCfg().RateSConns,
utils.RateSv1CostForEvent,
cgrEv, &rpCost); err != nil {
return
}
cgrEv.APIOpts[utils.MetaRateSCost] = rpCost
return
}
// accountSDebitEvent will send the event to accountS and attach the cost received back to event
func (cdrS *CDRServer) accountSDebitEvent(ctx *context.Context, cgrEv *utils.CGREvent) (err error) {
acntCost := new(utils.EventCharges)
if err = cdrS.connMgr.Call(ctx, cdrS.cfg.CdrsCfg().AccountSConns,
utils.AccountSv1DebitAbstracts, cgrEv, acntCost); err != nil {
return
}
cgrEv.APIOpts[utils.MetaAccountSCost] = acntCost
return
}
// accountSRefundCharges will refund the charges into Account
func (cdrS *CDRServer) accountSRefundCharges(ctx *context.Context, tnt string, eChrgs *utils.EventCharges,
apiOpts map[string]any) (err error) {
var rply string
argsRefund := &utils.APIEventCharges{
Tenant: tnt,
APIOpts: apiOpts,
EventCharges: eChrgs,
}
if err = cdrS.connMgr.Call(ctx, cdrS.cfg.CdrsCfg().AccountSConns,
utils.AccountSv1RefundCharges, argsRefund, &rply); err != nil {
return
}
return
}
// thdSProcessEvent will send the event to ThresholdS
func (cdrS *CDRServer) thdSProcessEvent(ctx *context.Context, cgrEv *utils.CGREvent) (err error) {
var tIDs []string
// we clone the CGREvent so we can add EventType without being propagated
cgrEv = cgrEv.Clone()
cgrEv.APIOpts[utils.MetaEventType] = utils.CDR
if err = cdrS.connMgr.Call(ctx, cdrS.cfg.CdrsCfg().ThresholdSConns,
utils.ThresholdSv1ProcessEvent,
cgrEv, &tIDs); err != nil &&
err.Error() == utils.ErrNotFound.Error() {
err = nil // NotFound is not considered error
}
return
}
// statSProcessEvent will send the event to StatS
func (cdrS *CDRServer) statSProcessEvent(ctx *context.Context, cgrEv *utils.CGREvent) (err error) {
var reply []string
if err = cdrS.connMgr.Call(ctx, cdrS.cfg.CdrsCfg().StatSConns,
utils.StatSv1ProcessEvent,
cgrEv.Clone(), &reply); err != nil &&
err.Error() == utils.ErrNotFound.Error() {
err = nil // NotFound is not considered error
}
return
}
// eeSProcessEvent will process the event with the EEs component
func (cdrS *CDRServer) eeSProcessEvent(ctx *context.Context, cgrEv *utils.CGREventWithEeIDs) (err error) {
var reply map[string]map[string]any
if err = cdrS.connMgr.Call(ctx, cdrS.cfg.CdrsCfg().EEsConns,
utils.EeSv1ProcessEvent,
cgrEv, &reply); err != nil &&
err.Error() == utils.ErrNotFound.Error() {
err = nil // NotFound is not considered error
}
return
}
// processEvents processes CGREvents based on arguments.
// In case of partially executed, both the error and the events will be returned.
func (cdrS *CDRServer) processEvents(ctx *context.Context, evs []*utils.CGREvent) ([]*utils.EventsWithOpts, error) {
for _, ev := range evs {
attrS, err := GetBoolOpts(ctx, ev.Tenant, ev.AsDataProvider(), cdrS.fltrS, cdrS.cfg.CdrsCfg().Opts.Attributes,
config.CDRsAttributesDftOpt, utils.MetaAttributes)
if err != nil {
return nil, fmt.Errorf("retrieving %s option failed: %w", utils.MetaAttributes, err)
}
if !attrS {
continue
}
if err = cdrS.attrSProcessEvent(ctx, ev); err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error: <%v> processing event %s with %s",
utils.CDRs, err, utils.ToJSON(ev), utils.AttributeS))
return nil, utils.NewErrAttributeS(err)
}
}
// Allocate capacity equal to the number of events for slight optimization.
cgrEvs := make([]*utils.CGREvent, 0, len(evs))
for _, ev := range evs {
chrgS, err := GetBoolOpts(ctx, ev.Tenant, ev.AsDataProvider(), cdrS.fltrS, cdrS.cfg.CdrsCfg().Opts.Chargers,
config.CDRsChargersDftOpt, utils.MetaChargers)
if err != nil {
return nil, fmt.Errorf("retrieving %s option failed: %w", utils.MetaChargers, err)
}
if !chrgS {
cgrEvs = append(cgrEvs, ev)
continue
}
chrgEvs, err := cdrS.chrgrSProcessEvent(ctx, ev)
if err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error: <%v> processing event %s with %s",
utils.CDRs, err, utils.ToJSON(ev), utils.ChargerS))
return nil, utils.NewErrChargerS(err)
}
cgrEvs = append(cgrEvs, chrgEvs...)
}
var partiallyExecuted bool // from here actions are optional and a general error is returned
for _, cgrEv := range cgrEvs {
rateS, err := GetBoolOpts(ctx, cgrEv.Tenant, cgrEv.AsDataProvider(), cdrS.fltrS, cdrS.cfg.CdrsCfg().Opts.Rates,
config.CDRsRatesDftOpt, utils.MetaRates)
if err != nil {
return nil, fmt.Errorf("retrieving %s option failed: %w", utils.MetaRates, err)
}
if !rateS {
continue
}
if err := cdrS.rateSCostForEvent(ctx, cgrEv); err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error: <%v> processing event %s with %s",
utils.CDRs, err, utils.ToJSON(cgrEv), utils.RateS))
partiallyExecuted = true
}
}
for _, cgrEv := range cgrEvs {
acntS, err := GetBoolOpts(ctx, cgrEv.Tenant, cgrEv.AsDataProvider(), cdrS.fltrS, cdrS.cfg.CdrsCfg().Opts.Accounts,
config.CDRsAccountsDftOpt, utils.MetaAccounts)
if err != nil {
return nil, fmt.Errorf("retrieving %s option failed: %w", utils.MetaAccounts, err)
}
if !acntS {
continue
}
if ecCostIface, wasCharged := cgrEv.APIOpts[utils.MetaAccountSCost]; wasCharged {
ecCostMap, ok := ecCostIface.(map[string]any)
if !ok {
return nil, fmt.Errorf("expected %s to be a map[string]any, got %T", utils.MetaAccountSCost, ecCostIface)
}
// before converting into EventChargers, we must get the JSON encoding and Unmarshal it into an EventChargers
btsEvCh, err := json.Marshal(ecCostMap)
if err != nil {
return nil, err
}
var ecCost utils.EventCharges
if err = json.Unmarshal(btsEvCh, &ecCost); err != nil {
return nil, err
}
if err := cdrS.accountSRefundCharges(ctx, cgrEv.Tenant, &ecCost, cgrEv.APIOpts); err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error: <%v> processing event %s with %s",
utils.CDRs, err, utils.ToJSON(cgrEv), utils.AccountS))
partiallyExecuted = true
}
}
if err := cdrS.accountSDebitEvent(ctx, cgrEv); err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error: <%v> processing event %s with %s",
utils.CDRs, err, utils.ToJSON(cgrEv), utils.AccountS))
partiallyExecuted = true
}
}
// populate cost from accounts or rates for every event
for _, cgrEv := range cgrEvs {
if cost := populateCost(cgrEv.APIOpts); cost != nil {
cgrEv.APIOpts[utils.MetaCost] = cost
}
}
for _, cgrEv := range cgrEvs {
store, err := GetBoolOpts(ctx, cgrEv.Tenant, cgrEv.AsDataProvider(), cdrS.fltrS, cdrS.cfg.CdrsCfg().Opts.Store,
config.CDRsStoreDftOpt, utils.MetaStore)
if err != nil {
return nil, fmt.Errorf("retrieving %s option failed: %w", utils.MetaStore, err)
}
if !store {
continue
}
rerate, err := GetBoolOpts(ctx, cgrEv.Tenant, cgrEv.AsDataProvider(), cdrS.fltrS, cdrS.cfg.CdrsCfg().Opts.Rerate,
config.CDRsRerateDftOpt, utils.MetaRerate)
if err != nil {
return nil, fmt.Errorf("retrieving %s option failed: %w", utils.MetaRerate, err)
}
cdrID := GetUniqueCDRID(cgrEv)
if cgrEv.APIOpts == nil {
cgrEv.APIOpts = make(map[string]any)
}
cgrEv.APIOpts[utils.MetaCDRID] = cdrID
if err := cdrS.db.SetCDR(ctx, cdrID, cgrEv, false); err != nil {
if err != utils.ErrExists || !rerate {
// ToDo: add refund logic
return nil, fmt.Errorf("storing CDR %s failed: %w", utils.ToJSON(cgrEv), err)
}
if err = cdrS.db.SetCDR(ctx, cdrID, cgrEv, true); err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error: <%s> updating CDR %+v",
utils.CDRs, err.Error(), utils.ToJSON(cgrEv)))
return nil, utils.ErrPartiallyExecuted
}
}
}
for _, cgrEv := range cgrEvs {
export, err := GetBoolOpts(ctx, cgrEv.Tenant, cgrEv.AsDataProvider(), cdrS.fltrS, cdrS.cfg.CdrsCfg().Opts.Export,
config.CDRsExportDftOpt, utils.OptsCDRsExport)
if err != nil {
return nil, fmt.Errorf("retrieving %s option failed: %w", utils.OptsCDRsExport, err)
}
if !export {
continue
}
evWithOpts := &utils.CGREventWithEeIDs{
CGREvent: cgrEv,
EeIDs: cdrS.cfg.CdrsCfg().OnlineCDRExports,
}
if err := cdrS.eeSProcessEvent(ctx, evWithOpts); err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error: <%v> exporting cdr %s",
utils.CDRs, err, utils.ToJSON(evWithOpts)))
partiallyExecuted = true
}
}
for _, cgrEv := range cgrEvs {
thdS, err := GetBoolOpts(ctx, cgrEv.Tenant, cgrEv.AsDataProvider(), cdrS.fltrS, cdrS.cfg.CdrsCfg().Opts.Thresholds,
config.CDRsThresholdsDftOpt, utils.MetaThresholds)
if err != nil {
return nil, fmt.Errorf("retrieving %s option failed: %w", utils.MetaThresholds, err)
}
if !thdS {
continue
}
if err := cdrS.thdSProcessEvent(ctx, cgrEv); err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error: <%v> processing event %s with %s",
utils.CDRs, err, utils.ToJSON(cgrEv), utils.ThresholdS))
partiallyExecuted = true
}
}
for _, cgrEv := range cgrEvs {
stS, err := GetBoolOpts(ctx, cgrEv.Tenant, cgrEv.AsDataProvider(), cdrS.fltrS, cdrS.cfg.CdrsCfg().Opts.Stats,
config.CDRsStatsDftOpt, utils.MetaStats)
if err != nil {
return nil, fmt.Errorf("retrieving %s option failed: %w", utils.MetaStats, err)
}
if !stS {
continue
}
if err := cdrS.statSProcessEvent(ctx, cgrEv); err != nil {
utils.Logger.Warning(
fmt.Sprintf("<%s> error: <%v> processing event %s with %s",
utils.CDRs, err, utils.ToJSON(cgrEv), utils.StatS))
partiallyExecuted = true
}
}
// now that we did all the requested processed events, we have to build our EventsWithOpts
outEvs := make([]*utils.EventsWithOpts, len(cgrEvs))
for i, cgrEv := range cgrEvs {
outEvs[i] = &utils.EventsWithOpts{
Event: cgrEv.Event,
Opts: cgrEv.APIOpts,
}
}
if partiallyExecuted {
return outEvs, utils.ErrPartiallyExecuted
}
return outEvs, nil
}
// V1ProcessEvent will process the CGREvent
func (cdrS *CDRServer) V1ProcessEvent(ctx *context.Context, args *utils.CGREvent, reply *string) (err error) {
if args.ID == utils.EmptyString {
args.ID = utils.GenUUID()
}
if args.Tenant == utils.EmptyString {
args.Tenant = cdrS.cfg.GeneralCfg().DefaultTenant
}
// RPC caching
if config.CgrConfig().CacheCfg().Partitions[utils.CacheRPCResponses].Limit != 0 {
cacheKey := utils.ConcatenatedKey(utils.CDRsV1ProcessEvent, args.ID)
refID := guardian.Guardian.GuardIDs("",
config.CgrConfig().GeneralCfg().LockingTimeout, cacheKey) // RPC caching needs to be atomic
defer guardian.Guardian.UnguardIDs(refID)
if itm, has := Cache.Get(utils.CacheRPCResponses, cacheKey); has {
cachedResp := itm.(*utils.CachedRPCResponse)
if cachedResp.Error == nil {
*reply = *cachedResp.Result.(*string)
}
return cachedResp.Error
}
defer Cache.Set(ctx, utils.CacheRPCResponses, cacheKey,
&utils.CachedRPCResponse{Result: reply, Error: err},
nil, true, utils.NonTransactional)
}
// end of RPC caching
if _, err = cdrS.processEvents(ctx, []*utils.CGREvent{args}); err != nil {
return
}
*reply = utils.OK
return nil
}
// V1ProcessEventWithGet has the same logic with V1ProcessEvent except it adds the proccessed events to the reply
func (cdrS *CDRServer) V1ProcessEventWithGet(ctx *context.Context, args *utils.CGREvent, evs *[]*utils.EventsWithOpts) (err error) {
if args.ID == utils.EmptyString {
args.ID = utils.GenUUID()
}
if args.Tenant == utils.EmptyString {
args.Tenant = cdrS.cfg.GeneralCfg().DefaultTenant
}
// RPC caching
if config.CgrConfig().CacheCfg().Partitions[utils.CacheRPCResponses].Limit != 0 {
cacheKey := utils.ConcatenatedKey(utils.CDRsV1ProcessEventWithGet, args.ID)
refID := guardian.Guardian.GuardIDs("",
config.CgrConfig().GeneralCfg().LockingTimeout, cacheKey) // RPC caching needs to be atomic
defer guardian.Guardian.UnguardIDs(refID)
if itm, has := Cache.Get(utils.CacheRPCResponses, cacheKey); has {
cachedResp := itm.(*utils.CachedRPCResponse)
if cachedResp.Error == nil {
*evs = *cachedResp.Result.(*[]*utils.EventsWithOpts)
}
return cachedResp.Error
}
defer Cache.Set(ctx, utils.CacheRPCResponses, cacheKey,
&utils.CachedRPCResponse{Result: evs, Error: err},
nil, true, utils.NonTransactional)
}
// end of RPC caching
var procEvs []*utils.EventsWithOpts
if procEvs, err = cdrS.processEvents(ctx, []*utils.CGREvent{args}); err != nil {
return
}
*evs = procEvs
return nil
}
func populateCost(cgrOpts map[string]any) *utils.Decimal {
// if the cost is already present, get out
if _, has := cgrOpts[utils.MetaCost]; has {
return nil
}
// check firstly in accounts
if accCost, has := cgrOpts[utils.MetaAccountSCost]; has {
return accCost.(*utils.EventCharges).Concretes
}
// after check in rates
if rtCost, has := cgrOpts[utils.MetaRateSCost]; has {
return rtCost.(utils.RateProfileCost).Cost
}
return nil
}
// V1ProcessStoredEvents processes stored events based on provided filters.
func (cdrS *CDRServer) V1ProcessStoredEvents(ctx *context.Context, args *CDRFilters, reply *string) (err error) {
if args.ID == utils.EmptyString {
args.ID = utils.GenUUID()
}
if args.Tenant == utils.EmptyString {
args.Tenant = cdrS.cfg.GeneralCfg().DefaultTenant
}
// RPC caching
if config.CgrConfig().CacheCfg().Partitions[utils.CacheRPCResponses].Limit != 0 {
cacheKey := utils.ConcatenatedKey(utils.CDRsV1ProcessStoredEvents, args.ID)
refID := guardian.Guardian.GuardIDs("",
config.CgrConfig().GeneralCfg().LockingTimeout, cacheKey)
defer guardian.Guardian.UnguardIDs(refID)
if itm, has := Cache.Get(utils.CacheRPCResponses, cacheKey); has {
cachedResp := itm.(*utils.CachedRPCResponse)
if cachedResp.Error == nil {
*reply = *cachedResp.Result.(*string)
}
return cachedResp.Error
}
defer Cache.Set(ctx, utils.CacheRPCResponses, cacheKey,
&utils.CachedRPCResponse{Result: reply, Error: err},
nil, true, utils.NonTransactional)
}
fltrs, err := GetFilters(ctx, args.FilterIDs, args.Tenant, cdrS.dm)
if err != nil {
return fmt.Errorf("preparing filters failed: %w", err)
}
cdrs, err := cdrS.db.GetCDRs(ctx, fltrs, args.APIOpts)
if err != nil {
return fmt.Errorf("retrieving CDRs failed: %w", err)
}
_, err = cdrS.processEvents(ctx, CDRsToCGREvents(cdrs))
if err != nil && !errors.Is(err, utils.ErrPartiallyExecuted) {
return fmt.Errorf("processing events failed: %w", err)
}
*reply = utils.OK
return err
}

3191
engine/cdrs_test.go
View File

File diff suppressed because it is too large Load Diff

204
engine/connmanager_test.go
View File

@@ -590,41 +590,6 @@ func TestCMDeadLock(t *testing.T) {
}
*/
func TestCMEnableDispatcher(t *testing.T) {
tmp := Cache
defer func() {
Cache = tmp
}()
Cache.Clear(nil)
cfg := config.NewDefaultCGRConfig()
cM := NewConnManager(cfg)
data := NewInternalDB(nil, nil, cfg.DataDbCfg().Items)
dm := NewDataManager(data, cfg.CacheCfg(), nil)
fltrs := NewFilterS(cfg, nil, dm)
Cache = NewCacheS(cfg, dm, nil, nil)
var storDB StorDB
storDBChan := make(chan StorDB, 1)
storDBChan <- storDB
newCDRSrv := NewCDRServer(cfg, dm, fltrs, nil, storDBChan)
srvcNames := []string{utils.AccountS, utils.ActionS, utils.AttributeS,
utils.CacheS, utils.ChargerS, utils.ConfigS, utils.DispatcherS,
utils.GuardianS, utils.RateS, utils.ResourceS, utils.RouteS,
utils.SessionS, utils.StatS, utils.ThresholdS, utils.CDRs,
utils.ReplicatorS, utils.EeS, utils.CoreS, utils.AnalyzerS,
utils.AdminS, utils.LoaderS, utils.ServiceManager}
for _, name := range srvcNames {
newSrvcWName, err := NewServiceWithName(newCDRSrv, name, true)
if err != nil {
t.Error(err)
}
cM.EnableDispatcher(newSrvcWName)
}
}
func TestCMGetInternalChan(t *testing.T) {
cfg := config.NewDefaultCGRConfig()
@@ -657,80 +622,115 @@ func TestCMGetDispInternalChan(t *testing.T) {
}
func TestCMDisableDispatcher(t *testing.T) {
tmp := Cache
defer func() {
Cache = tmp
}()
Cache.Clear(nil)
cfg := config.NewDefaultCGRConfig()
// func TestCMEnableDispatcher(t *testing.T) {
// tmp := Cache
// defer func() {
// Cache = tmp
// }()
// Cache.Clear(nil)
// cfg := config.NewDefaultCGRConfig()
// cM := NewConnManager(cfg)
// data := NewInternalDB(nil, nil, cfg.DataDbCfg().Items)
// dm := NewDataManager(data, cfg.CacheCfg(), nil)
// fltrs := NewFilterS(cfg, nil, dm)
// Cache = NewCacheS(cfg, dm, nil, nil)
// var storDB StorDB
// storDBChan := make(chan StorDB, 1)
// storDBChan <- storDB
// newCDRSrv := NewCDRServer(cfg, dm, fltrs, nil, storDBChan)
cM := &ConnManager{
cfg: cfg,
connCache: ltcache.NewCache(-1, 0, true, nil),
}
cM.connCache.Set("itmID1", "value of first item", nil)
data := NewInternalDB(nil, nil, cfg.DataDbCfg().Items)
dm := NewDataManager(data, cfg.CacheCfg(), nil)
fltrs := NewFilterS(cfg, nil, dm)
var storDB StorDB
storDBChan := make(chan StorDB, 1)
storDBChan <- storDB
newCDRSrv := NewCDRServer(cfg, dm, fltrs, nil, storDBChan)
newSrvcWName, err := NewServiceWithName(newCDRSrv, utils.AccountS, true)
if err != nil {
t.Error(err)
}
cM.EnableDispatcher(newSrvcWName)
// srvcNames := []string{utils.AccountS, utils.ActionS, utils.AttributeS,
// utils.CacheS, utils.ChargerS, utils.ConfigS, utils.DispatcherS,
// utils.GuardianS, utils.RateS, utils.ResourceS, utils.RouteS,
// utils.SessionS, utils.StatS, utils.ThresholdS, utils.CDRs,
// utils.ReplicatorS, utils.EeS, utils.CoreS, utils.AnalyzerS,
// utils.AdminS, utils.LoaderS, utils.ServiceManager}
Cache = NewCacheS(cfg, dm, cM, nil)
Cache.SetWithoutReplicate(utils.CacheRPCConnections, "itmID2",
"value of 2nd item", nil, true, utils.NonTransactional)
// for _, name := range srvcNames {
var exp []string
// newSrvcWName, err := NewServiceWithName(newCDRSrv, name, true)
// if err != nil {
// t.Error(err)
// }
// cM.EnableDispatcher(newSrvcWName)
cM.DisableDispatcher()
rcv1 := cM.connCache.GetItemIDs("itmID1")
rcv2 := Cache.GetItemIDs(utils.CacheRPCConnections, utils.EmptyString)
// }
// }
if !reflect.DeepEqual(rcv1, exp) {
t.Errorf("\nexpected: <%+v>, \nreceived: <%+v>", exp, rcv1)
} else if !reflect.DeepEqual(rcv2, exp) {
t.Errorf("\nexpected: <%+v>, \nreceived: <%+v>", exp, rcv2)
} else if cM.disp != nil || cM.dispIntCh != nil {
t.Errorf("\nexpected nil cM.disp and cM.dispIntCh, \nreceived cM.disp: <%+v>, \n cM.dispIntCh: <%+v>", cM.disp, cM.dispIntCh)
}
// func TestCMDisableDispatcher(t *testing.T) {
// tmp := Cache
// defer func() {
// Cache = tmp
// }()
// Cache.Clear(nil)
// cfg := config.NewDefaultCGRConfig()
}
// cM := &ConnManager{
// cfg: cfg,
// connCache: ltcache.NewCache(-1, 0, true, nil),
// }
// cM.connCache.Set("itmID1", "value of first item", nil)
// data := NewInternalDB(nil, nil, cfg.DataDbCfg().Items)
// dm := NewDataManager(data, cfg.CacheCfg(), nil)
// fltrs := NewFilterS(cfg, nil, dm)
// var storDB StorDB
// storDBChan := make(chan StorDB, 1)
// storDBChan <- storDB
// newCDRSrv := NewCDRServer(cfg, dm, fltrs, nil, storDBChan)
// newSrvcWName, err := NewServiceWithName(newCDRSrv, utils.AccountS, true)
// if err != nil {
// t.Error(err)
// }
// cM.EnableDispatcher(newSrvcWName)
func TestCMgetInternalConnChanFromDisp(t *testing.T) {
tmp := Cache
defer func() {
Cache = tmp
}()
Cache.Clear(nil)
cfg := config.NewDefaultCGRConfig()
// Cache = NewCacheS(cfg, dm, cM, nil)
// Cache.SetWithoutReplicate(utils.CacheRPCConnections, "itmID2",
// "value of 2nd item", nil, true, utils.NonTransactional)
cM := &ConnManager{
cfg: cfg,
connCache: ltcache.NewCache(-1, 0, true, nil),
}
cM.connCache.Set("itmID1", "value of first item", nil)
data := NewInternalDB(nil, nil, cfg.DataDbCfg().Items)
dm := NewDataManager(data, cfg.CacheCfg(), nil)
fltrs := NewFilterS(cfg, nil, dm)
var storDB StorDB
storDBChan := make(chan StorDB, 1)
storDBChan <- storDB
newCDRSrv := NewCDRServer(cfg, dm, fltrs, nil, storDBChan)
newSrvcWName, err := NewServiceWithName(newCDRSrv, utils.AccountS, true)
if err != nil {
t.Error(err)
}
cM.EnableDispatcher(newSrvcWName)
// var exp []string
if rcv, ok := cM.getInternalConnChan(utils.ConcatenatedKey(utils.MetaInternal, utils.MetaAccounts)); !ok {
t.Errorf("Unexpected error getting internalConnChan, Received <%+v>", rcv)
}
// cM.DisableDispatcher()
// rcv1 := cM.connCache.GetItemIDs("itmID1")
// rcv2 := Cache.GetItemIDs(utils.CacheRPCConnections, utils.EmptyString)
}
// if !reflect.DeepEqual(rcv1, exp) {
// t.Errorf("\nexpected: <%+v>, \nreceived: <%+v>", exp, rcv1)
// } else if !reflect.DeepEqual(rcv2, exp) {
// t.Errorf("\nexpected: <%+v>, \nreceived: <%+v>", exp, rcv2)
// } else if cM.disp != nil || cM.dispIntCh != nil {
// t.Errorf("\nexpected nil cM.disp and cM.dispIntCh, \nreceived cM.disp: <%+v>, \n cM.dispIntCh: <%+v>", cM.disp, cM.dispIntCh)
// }
// }
// func TestCMgetInternalConnChanFromDisp(t *testing.T) {
// tmp := Cache
// defer func() {
// Cache = tmp
// }()
// Cache.Clear(nil)
// cfg := config.NewDefaultCGRConfig()
// cM := &ConnManager{
// cfg: cfg,
// connCache: ltcache.NewCache(-1, 0, true, nil),
// }
// cM.connCache.Set("itmID1", "value of first item", nil)
// data := NewInternalDB(nil, nil, cfg.DataDbCfg().Items)
// dm := NewDataManager(data, cfg.CacheCfg(), nil)
// fltrs := NewFilterS(cfg, nil, dm)
// var storDB StorDB
// storDBChan := make(chan StorDB, 1)
// storDBChan <- storDB
// newCDRSrv := NewCDRServer(cfg, dm, fltrs, nil, storDBChan)
// newSrvcWName, err := NewServiceWithName(newCDRSrv, utils.AccountS, true)
// if err != nil {
// t.Error(err)
// }
// cM.EnableDispatcher(newSrvcWName)
// if rcv, ok := cM.getInternalConnChan(utils.ConcatenatedKey(utils.MetaInternal, utils.MetaAccounts)); !ok {
// t.Errorf("Unexpected error getting internalConnChan, Received <%+v>", rcv)
// }
// }

2
engine/storage_interface.go
View File

@@ -109,7 +109,7 @@ type DataDBDriver interface {
type StorDB interface {
Storage
SetCDR(*context.Context, string, *utils.CGREvent, bool) error
GetCDRs(*context.Context, []*Filter, map[string]interface{}) ([]*CDR, error)
GetCDRs(*context.Context, []*Filter, map[string]interface{}) ([]*utils.CDR, error)
RemoveCDRs(*context.Context, []*Filter) error
}

6
engine/storage_internal_stordb.go
View File

@@ -66,7 +66,7 @@ func (iDB *InternalDB) SetCDR(_ *context.Context, cdrID string, cgrEv *utils.CGR
return nil
}
func (iDB *InternalDB) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts map[string]interface{}) (cdrs []*CDR, err error) {
func (iDB *InternalDB) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts map[string]interface{}) (cdrs []*utils.CDR, err error) {
pairFltrs := make(map[string][]string)
notPairFltrs := make(map[string][]string)
notIndexed := []*FilterRule{}
@@ -154,9 +154,9 @@ func (iDB *InternalDB) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts map
return nil, utils.ErrNotFound
}
// convert from event into CDRs
cdrs = make([]*CDR, len(events))
cdrs = make([]*utils.CDR, len(events))
for i, event := range events {
cdrs[i] = &CDR{
cdrs[i] = &utils.CDR{
Tenant: event.Tenant,
Opts: event.APIOpts,
Event: event.Event,

8
engine/storage_mongo_stordb.go
View File

@@ -121,7 +121,7 @@ func (ms *MongoStorage) SetCDR(ctx *context.Context, cdrID string, cdr *utils.CG
_, err := ms.getCol(ColCDRs).InsertOne(
sctx,
&CDR{
&utils.CDR{
Tenant: cdr.Tenant,
Opts: cdr.APIOpts,
Event: cdr.Event,
@@ -169,7 +169,7 @@ func isMongoDuplicateError(err error) bool {
return false
}
func (ms *MongoStorage) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts map[string]interface{}) (cdrs []*CDR, err error) {
func (ms *MongoStorage) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts map[string]interface{}) (cdrs []*utils.CDR, err error) {
fltrs := make(bson.M)
for _, fltr := range qryFltr {
for _, rule := range fltr.Rules {
@@ -209,7 +209,7 @@ func (ms *MongoStorage) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts ma
return err
}
for cur.Next(sctx) {
cdr := CDR{}
cdr := utils.CDR{}
err := cur.Decode(&cdr)
if err != nil {
return err
@@ -372,7 +372,7 @@ func (ms *MongoStorage) RemoveCDRs(ctx *context.Context, qryFltr []*Filter) (err
}
defer cur.Close(sctx)
for cur.Next(sctx) {
cdr := CDR{}
cdr := utils.CDR{}
if err := cur.Decode(&cdr); err != nil {
return err
}

20
engine/storage_sql.go
View File

@@ -146,7 +146,7 @@ func (sqls *SQLStorage) SetCDR(_ *context.Context, cdrID string, cdr *utils.CGRE
if tx.Error != nil {
return tx.Error
}
cdrTable := &CDRSQLTable{
cdrTable := &utils.CDRSQLTable{
Tenant: cdr.Tenant,
Opts: cdr.APIOpts,
Event: cdr.Event,
@@ -166,9 +166,9 @@ func (sqls *SQLStorage) SetCDR(_ *context.Context, cdrID string, cdr *utils.CGRE
return tx.Error
}
updated := tx.Model(&CDRSQLTable{}).Where(
updated := tx.Model(&utils.CDRSQLTable{}).Where(
sqls.cdrIDQuery(cdrID)).Updates(
CDRSQLTable{Opts: cdr.APIOpts, Event: cdr.Event, UpdatedAt: time.Now()})
utils.CDRSQLTable{Opts: cdr.APIOpts, Event: cdr.Event, UpdatedAt: time.Now()})
if updated.Error != nil {
tx.Rollback()
return updated.Error
@@ -180,7 +180,7 @@ func (sqls *SQLStorage) SetCDR(_ *context.Context, cdrID string, cdr *utils.CGRE
// GetCDRs has ability to get the filtered CDRs, count them or simply return them
// qryFltr.Unscoped will ignore soft deletes or delete records permanently
func (sqls *SQLStorage) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts map[string]interface{}) (cdrs []*CDR, err error) {
func (sqls *SQLStorage) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts map[string]interface{}) (cdrs []*utils.CDR, err error) {
q := sqls.db.Table(utils.CDRsTBL)
var excludedCdrQueryFilterTypes []*FilterRule
for _, fltr := range qryFltr {
@@ -221,7 +221,7 @@ func (sqls *SQLStorage) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts ma
q = q.Offset(offset)
// Execute query
results := make([]*CDRSQLTable, 0)
results := make([]*utils.CDRSQLTable, 0)
if err = q.Find(&results).Error; err != nil {
return
}
@@ -229,11 +229,11 @@ func (sqls *SQLStorage) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts ma
return nil, utils.ErrNotFound
}
//convert into CDR
resultCdr := make([]*CDR, 0, len(results))
resultCdr := make([]*utils.CDR, 0, len(results))
for _, val := range results {
// here we wil do our filtration, meaning that we will filter those cdrs who cannot be filtered in the databes eg: *ai, *rsr..
if len(excludedCdrQueryFilterTypes) != 0 {
newCdr := &CDR{
newCdr := &utils.CDR{
Tenant: val.Tenant,
Opts: val.Opts,
Event: val.Event,
@@ -252,7 +252,7 @@ func (sqls *SQLStorage) GetCDRs(ctx *context.Context, qryFltr []*Filter, opts ma
continue
}
}
resultCdr = append(resultCdr, &CDR{
resultCdr = append(resultCdr, &utils.CDR{
Tenant: val.Tenant,
Opts: val.Opts,
Event: val.Event,
@@ -309,7 +309,7 @@ func (sqls *SQLStorage) RemoveCDRs(ctx *context.Context, qryFltr []*Filter) (err
return
}
// in the other case, if we have such filters, check the results based on those filters
results := make([]*CDRSQLTable, 0)
results := make([]*utils.CDRSQLTable, 0)
if err = q.Find(&results).Error; err != nil {
return
}
@@ -322,7 +322,7 @@ func (sqls *SQLStorage) RemoveCDRs(ctx *context.Context, qryFltr []*Filter) (err
remCdr := make([]string, 0, len(results)) // we will keep the *cdrID of every CDR taht matched the those filters
for _, cdr := range results {
if len(excludedCdrQueryFilterTypes) != 0 {
newCdr := &CDR{
newCdr := &utils.CDR{
Tenant: cdr.Tenant,
Opts: cdr.Opts,
Event: cdr.Event,

13
engine/storage_utils.go
View File

@@ -105,3 +105,16 @@ func composeMongoURI(scheme, host, port, db, user, pass string) string {
}
return uri
}
// checkNestedFields checks if there are elements or values nested (e.g *opts.*rateSCost.Cost)
func checkNestedFields(elem string, values []string) bool {
if len(strings.Split(elem, utils.NestingSep)) > 2 {
return true
}
for _, val := range values {
if len(strings.Split(val, utils.NestingSep)) > 2 {
return true
}
}
return false
}