Added *remove_session_costs action. fixes#1423

engine/action.go

@@ -84,6 +84,7 @@ const (
 	SetExpiry                 = "*set_expiry"
 	MetaPublishAccount        = "*publish_account"
 	MetaPublishBalance        = "*publish_balance"
+	MetaRemoveSessionCosts    = "*remove_session_costs"
 )
 
 func (a *Action) Clone() *Action {
@@ -127,6 +128,7 @@ func getActionFunc(typ string) (actionTypeFunc, bool) {
 		utils.MetaAMQPjsonMap:     sendAMQP,
 		utils.MetaAWSjsonMap:      sendAWS,
 		utils.MetaSQSjsonMap:      sendSQS,
+		MetaRemoveSessionCosts:    cleanSessionCosts,
 	}
 	f, exists := actionFuncMap[typ]
 	return f, exists
@@ -960,3 +962,28 @@ func (cdrP *cdrLogProvider) AsNavigableMap([]*config.FCTemplate) (
 func (cdrP *cdrLogProvider) RemoteHost() net.Addr {
 	return utils.LocalAddr()
 }
+
+func cleanSessionCosts(_ *Account, _ *Action, _ Actions, extraData interface{}) error { // FiltersID;inlineFilter
+	fltrs, err := utils.IfaceAsString(extraData)
+	if err != nil {
+		return err
+	}
+	tenant := config.CgrConfig().GeneralCfg().DefaultTenant
+	smcFilter := new(utils.SMCostFilter)
+	smcFilter.Usage = make([]*time.Duration, 2)
+	for _, fltrID := range strings.Split(fltrs, utils.INFIELD_SEP) {
+		fltr, err := dm.GetFilter(tenant, fltrID, true, true, utils.NonTransactional)
+		if err != nil {
+			utils.Logger.Warning(fmt.Sprintf("<%s>  Error: %s for filter: %s in action: <%s>",
+				utils.Actions, err.Error(), fltrID, MetaRemoveSessionCosts))
+			continue
+		}
+		for _, rule := range fltr.Rules {
+			smcFilter, err = utils.AppendToSMCostFilter(smcFilter, rule.Type, rule.FieldName, rule.Values, config.CgrConfig().GeneralCfg().DefaultTimezone)
+			if err != nil {
+				utils.Logger.Warning(fmt.Sprintf("<%s> %s in action: <%s>", utils.Actions, err.Error(), MetaRemoveSessionCosts))
+			}
+		}
+	}
+	return cdrStorage.RemoveSMCosts(smcFilter)
+}

engine/storage_interface.go

@@ -142,6 +142,7 @@ type CdrStorage interface {
 	SetSMCost(smc *SMCost) error
 	GetSMCosts(cgrid, runid, originHost, originIDPrfx string) ([]*SMCost, error)
 	RemoveSMCost(*SMCost) error
+	RemoveSMCosts(qryFltr *utils.SMCostFilter) error
 	GetCDRs(*utils.CDRsFilter, bool) ([]*CDR, int64, error)
 }
 

engine/storage_map_stordb.go

@@ -158,6 +158,9 @@ func (ms *MapStorage) SetCDR(cdr *CDR, allowUpdate bool) (err error) {
 func (ms *MapStorage) RemoveSMCost(smc *SMCost) (err error) {
 	return utils.ErrNotImplemented
 }
+func (ms *MapStorage) RemoveSMCosts(qryFltr *utils.SMCostFilter) error {
+	return utils.ErrNotImplemented
+}
 func (ms *MapStorage) GetCDRs(filter *utils.CDRsFilter, remove bool) (cdrs []*CDR, count int64, err error) {
 	return nil, 0, utils.ErrNotImplemented
 }

engine/storage_mongo_datadb.go

@@ -99,6 +99,7 @@ var (
 	CostDetailsLow     = strings.ToLower(utils.CostDetails)
 	DestinationLow     = strings.ToLower(utils.Destination)
 	CostLow            = strings.ToLower(utils.COST)
+	CostSourceLow      = strings.ToLower(utils.CostSource)
 
 	tTime = reflect.TypeOf(time.Time{})
 )

engine/storage_mongo_stordb.go

@@ -912,6 +912,23 @@ func (ms *MongoStorage) GetSMCosts(cgrid, runid, originHost, originIDPrefix stri
 	return smcs, err
 }
 
+func (ms *MongoStorage) RemoveSMCosts(qryFltr *utils.SMCostFilter) error {
+	filters := bson.M{
+		CGRIDLow:      bson.M{"$in": qryFltr.CgrIDs, "$nin": qryFltr.NotCgrIDs},
+		RunIDLow:      bson.M{"$in": qryFltr.RunIDs, "$nin": qryFltr.NotRunIDs},
+		OriginHostLow: bson.M{"$in": qryFltr.OriginHosts, "$nin": qryFltr.NotOriginHosts},
+		OriginIDLow:   bson.M{"$in": qryFltr.OriginIDs, "$nin": qryFltr.NotOriginIDs},
+		CostSourceLow: bson.M{"$in": qryFltr.CostSources, "$nin": qryFltr.NotCostSources},
+		UsageLow:      bson.M{"$gte": qryFltr.Usage[0], "$lt": qryFltr.Usage[1]},
+		CreatedAtLow:  bson.M{"$gte": qryFltr.CreatedAtStart, "$lt": qryFltr.CreatedAtEnd},
+	}
+	ms.cleanEmptyFilters(filters)
+	return ms.query(func(sctx mongo.SessionContext) (err error) {
+		_, err = ms.getCol(utils.SessionCostsTBL).DeleteMany(sctx, filters)
+		return err
+	})
+}
+
 func (ms *MongoStorage) SetCDR(cdr *CDR, allowUpdate bool) (err error) {
 	if cdr.OrderID == 0 {
 		cdr.OrderID = ms.cnter.Next()

engine/storage_sql.go

@@ -739,6 +739,66 @@ func (self *SQLStorage) RemoveSMCost(smc *SMCost) error {
 	return nil
 }
 
+func (self *SQLStorage) RemoveSMCosts(qryFltr *utils.SMCostFilter) error {
+	q := self.db.Table(utils.SessionCostsTBL).Select("*")
+	// Add filters, use in to replace the high number of ORs
+	if len(qryFltr.CgrIDs) != 0 {
+		q = q.Where("cgrid in (?)", qryFltr.CgrIDs)
+	}
+	if len(qryFltr.NotCgrIDs) != 0 {
+		q = q.Where("cgrid not in (?)", qryFltr.NotCgrIDs)
+	}
+	if len(qryFltr.RunIDs) != 0 {
+		q = q.Where("run_id in (?)", qryFltr.RunIDs)
+	}
+	if len(qryFltr.NotRunIDs) != 0 {
+		q = q.Where("run_id not in (?)", qryFltr.NotRunIDs)
+	}
+	if len(qryFltr.OriginIDs) != 0 {
+		q = q.Where("origin_id in (?)", qryFltr.OriginIDs)
+	}
+	if len(qryFltr.NotOriginIDs) != 0 {
+		q = q.Where("origin_id not in (?)", qryFltr.NotOriginIDs)
+	}
+	if len(qryFltr.OriginHosts) != 0 {
+		q = q.Where("origin_host in (?)", qryFltr.OriginHosts)
+	}
+	if len(qryFltr.NotOriginHosts) != 0 {
+		q = q.Where("origin_host not in (?)", qryFltr.NotOriginHosts)
+	}
+	if len(qryFltr.CostSources) != 0 {
+		q = q.Where("costsource in (?)", qryFltr.CostSources)
+	}
+	if len(qryFltr.NotCostSources) != 0 {
+		q = q.Where("costsource not in (?)", qryFltr.NotCostSources)
+	}
+	if qryFltr.CreatedAtStart != nil {
+		q = q.Where("created_at >= ?", qryFltr.CreatedAtStart)
+	}
+	if qryFltr.CreatedAtEnd != nil {
+		q = q.Where("created_at < ?", qryFltr.CreatedAtEnd)
+	}
+	if qryFltr.Usage[0] != nil {
+		if self.db.Dialect().GetName() == utils.MYSQL { // MySQL needs escaping for usage
+			q = q.Where("`usage` >= ?", qryFltr.Usage[0].Nanoseconds())
+		} else {
+			q = q.Where("usage >= ?", qryFltr.Usage[0].Nanoseconds())
+		}
+	}
+	if qryFltr.Usage[1] != nil {
+		if self.db.Dialect().GetName() == utils.MYSQL { // MySQL needs escaping for usage
+			q = q.Where("`usage` < ?", qryFltr.Usage[1].Nanoseconds())
+		} else {
+			q = q.Where("usage < ?", qryFltr.Usage[1].Nanoseconds())
+		}
+	}
+	if err := q.Delete(nil).Error; err != nil {
+		q.Rollback()
+		return err
+	}
+	return nil
+}
+
 // GetSMCosts is used to retrieve one or multiple SMCosts based on filter
 func (self *SQLStorage) GetSMCosts(cgrid, runid, originHost, originIDPrefix string) ([]*SMCost, error) {
 	var smCosts []*SMCost