Improved reverse idx code + thresholds implementation and tests

engine/datamanager.go

@@ -788,12 +788,20 @@ func (dm *DataManager) SetThresholdProfile(th *ThresholdProfile, withIndex bool)
 					th.Tenant).RemoveItemFromIndex(th.Tenant, th.ID, oldTh.FilterIDs); err != nil {
 					return
 				}
+				if err = removeReverseFilterIndexForFilter(dm, utils.CacheThresholdFilterIndexes, utils.EmptyString,
+					th.Tenant, th.ID, th.FilterIDs); err != nil {
+					return
+				}
 			}
 		}
 		if err := createAndIndex(utils.ThresholdProfilePrefix, th.Tenant,
 			utils.EmptyString, th.ID, th.FilterIDs, dm); err != nil {
 			return err
 		}
+		if err = addReverseFilterIndexForFilter(dm, utils.CacheThresholdFilterIndexes, utils.EmptyString,
+			th.Tenant, th.ID, th.FilterIDs); err != nil {
+			return
+		}
 	}
 	if itm := config.CgrConfig().DataDbCfg().Items[utils.MetaThresholdProfiles]; itm.Replicate {
 		if err = replicate(dm.connMgr, config.CgrConfig().DataDbCfg().RplConns,
@@ -824,6 +832,10 @@ func (dm *DataManager) RemoveThresholdProfile(tenant, id,
 			tenant).RemoveItemFromIndex(tenant, id, oldTh.FilterIDs); err != nil {
 			return
 		}
+		if err = removeReverseFilterIndexForFilter(dm, utils.CacheThresholdFilterIndexes, utils.EmptyString,
+			oldTh.Tenant, oldTh.ID, oldTh.FilterIDs); err != nil {
+			return
+		}
 	}
 	if itm := config.CgrConfig().DataDbCfg().Items[utils.MetaThresholdProfiles]; itm.Replicate {
 		replicate(dm.connMgr, config.CgrConfig().DataDbCfg().RplConns,

engine/filter_indexes.go

@@ -25,10 +25,87 @@ import (
 	"github.com/cgrates/cgrates/utils"
 )
 
+func ComputeThresholdIndexes(dm *DataManager, tenant string, thIDs *[]string,
+	transactionID string) (filterIndexer *FilterIndexer, err error) {
+	var thresholdIDs []string
+	var thdsIndexers *FilterIndexer
+	if thIDs == nil {
+		ids, err := dm.DataDB().GetKeysForPrefix(utils.ThresholdProfilePrefix)
+		if err != nil {
+			return nil, err
+		}
+		for _, id := range ids {
+			thresholdIDs = append(thresholdIDs, strings.Split(id, utils.CONCATENATED_KEY_SEP)[1])
+		}
+		// this will be on ComputeIndexes that contains empty indexes
+		thdsIndexers = NewFilterIndexer(dm, utils.ThresholdProfilePrefix, tenant)
+	} else {
+		// this will be on ComputeIndexesIDs that contains the old indexes from the next getter
+		var oldIDx map[string]utils.StringMap
+		if oldIDx, err = dm.GetFilterIndexes(utils.PrefixToIndexCache[utils.ThresholdProfilePrefix],
+			tenant, utils.EmptyString, nil); err != nil || oldIDx == nil {
+			thdsIndexers = NewFilterIndexer(dm, utils.ThresholdProfilePrefix, tenant)
+		} else {
+			thdsIndexers = NewFilterIndexerWithIndexes(dm, utils.ThresholdProfilePrefix, tenant, oldIDx)
+		}
+		thresholdIDs = *thIDs
+		transactionID = utils.NonTransactional
+	}
+	for _, id := range thresholdIDs {
+		th, err := dm.GetThresholdProfile(tenant, id, true, false, utils.NonTransactional)
+		if err != nil {
+			return nil, err
+		}
+		fltrIDs := make([]string, len(th.FilterIDs))
+		for i, fltrID := range th.FilterIDs {
+			fltrIDs[i] = fltrID
+		}
+		if len(fltrIDs) == 0 {
+			fltrIDs = []string{utils.META_NONE}
+		}
+		for _, fltrID := range fltrIDs {
+			var fltr *Filter
+			if fltrID == utils.META_NONE {
+				fltr = &Filter{
+					Tenant: th.Tenant,
+					ID:     th.ID,
+					Rules: []*FilterRule{
+						{
+							Type:    utils.META_NONE,
+							Element: utils.META_ANY,
+							Values:  []string{utils.META_ANY},
+						},
+					},
+				}
+			} else if fltr, err = GetFilter(dm, th.Tenant, fltrID,
+				true, false, utils.NonTransactional); err != nil {
+				if err == utils.ErrNotFound {
+					err = fmt.Errorf("broken reference to filter: %+v for threshold: %+v",
+						fltrID, th)
+				}
+				return nil, err
+			}
+			thdsIndexers.IndexTPFilter(FilterToTPFilter(fltr), th.ID)
+		}
+	}
+
+	if transactionID == utils.NonTransactional {
+		if err := thdsIndexers.StoreIndexes(true, transactionID); err != nil {
+			return nil, err
+		}
+		return nil, nil
+	} else {
+		if err := thdsIndexers.StoreIndexes(false, transactionID); err != nil {
+			return nil, err
+		}
+	}
+	return thdsIndexers, nil
+}
+
 func ComputeChargerIndexes(dm *DataManager, tenant string, cppIDs *[]string,
-	transactionID string) (cppIndexes *FilterIndexer, err error) {
+	transactionID string) (filterIndexer *FilterIndexer, err error) {
 	var chargerIDs []string
-	//var cppIndexes *FilterIndexer
+	var cppIndexes *FilterIndexer
 	if cppIDs == nil {
 		ids, err := dm.DataDB().GetKeysForPrefix(utils.ChargerProfilePrefix)
 		if err != nil {

engine/libindex.go

@@ -116,6 +116,7 @@ func UpdateFilterIndexes(dm *DataManager, tnt string, oldFltr *Filter, newFltr *
 		err = nil // if the error is NOT_FOUND, it means that no indexes were found for this filter, so no need to update
 		return
 	}
+
 	removeIndexKeys := removeRules.Slice()
 
 	for idxItmType, index := range rcvIndexes {
@@ -132,6 +133,18 @@ func UpdateFilterIndexes(dm *DataManager, tnt string, oldFltr *Filter, newFltr *
 				utils.NonTransactional); err != nil {
 				return err
 			}
+		case utils.CacheThresholdFilterIndexes:
+			// remove the indexes from this filter for this partition
+			if err = removeFilterIndexesForFilter(dm, idxItmType, utils.CacheThresholdProfiles,
+				tnt, removeIndexKeys, index); err != nil {
+				return
+			}
+			// we removed the old reverse indexes, now we have to compute the new ones
+			thresholdIDs := index.Slice()
+			if _, err = ComputeThresholdIndexes(dm, newFltr.Tenant, &thresholdIDs,
+				utils.NonTransactional); err != nil {
+				return err
+			}
 		}
 	}
 	return nil
@@ -190,6 +203,9 @@ func addReverseFilterIndexForFilter(dm *DataManager, idxItmType, ctx, tnt,
 				idxItmType: make(map[string]bool), // not found in database any reverse, we declare them to add in the next steps
 			}
 		}
+		if indexes[idxItmType] == nil {
+			indexes[idxItmType] = make(utils.StringMap)
+		}
 		indexes[idxItmType].Copy(map[string]bool{
 			itemID: true,
 		})