Metrics blockers added in code

engine/libstats.go

@@ -38,12 +38,12 @@ type StatQueueProfile struct {
 	Tenant       string
 	ID           string // QueueID
 	FilterIDs    []string
+	Weights      utils.DynamicWeights
+	Blockers     utils.Blockers // blocker flag to stop processing on filters matched
 	QueueLength  int
 	TTL          time.Duration
 	MinItems     int
 	Stored       bool
-	Weights      utils.DynamicWeights
-	Blockers     utils.Blockers       // blocker flag to stop processing on filters matched
 	ThresholdIDs []string             // list of thresholds to be checked after changes
 	Metrics      []*MetricWithFilters // list of metrics to build
 
@@ -91,8 +91,9 @@ func (sqp *StatQueueProfile) isLocked() bool {
 }
 
 type MetricWithFilters struct {
-	FilterIDs []string
 	MetricID  string
+	FilterIDs []string
+	Blockers  utils.Blockers // blocker flag to stop processing for next metric on filters matched
 }
 
 // NewStoredStatQueue initiates a StoredStatQueue out of StatQueue
@@ -586,9 +587,11 @@ func (sqp *StatQueueProfile) Set(path []string, val interface{}, newBranch bool,
 			sqp.ThresholdIDs = append(sqp.ThresholdIDs, valA...)
 		}
 	case 2:
+		// path =[]string{Metrics, MetricID}
 		if path[0] != utils.Metrics {
 			return utils.ErrWrongPath
 		}
+		// val := *acd;*tcd;*asr
 		if val != utils.EmptyString {
 			if len(sqp.Metrics) == 0 || newBranch {
 				sqp.Metrics = append(sqp.Metrics, new(MetricWithFilters))
@@ -604,7 +607,12 @@ func (sqp *StatQueueProfile) Set(path []string, val interface{}, newBranch bool,
 				for _, mID := range valA[1:] { // add the rest of the metrics
 					sqp.Metrics = append(sqp.Metrics, &MetricWithFilters{MetricID: mID})
 				}
-
+			case utils.BlockersField:
+				var blkrs utils.Blockers
+				if blkrs, err = utils.NewBlockersFromString(utils.IfaceAsString(val), utils.InfieldSep, utils.ANDSep); err != nil {
+					return
+				}
+				sqp.Metrics[len(sqp.Metrics)-1].Blockers = append(sqp.Metrics[len(sqp.Metrics)-1].Blockers, blkrs...)
 			default:
 				return utils.ErrWrongPath
 			}
@@ -734,5 +742,7 @@ func (mf *MetricWithFilters) FieldAsInterface(fldPath []string) (_ interface{},
 		return mf.MetricID, nil
 	case utils.FilterIDs:
 		return mf.FilterIDs, nil
+	case utils.BlockersField:
+		return mf.Blockers, nil
 	}
 }

engine/model_helpers.go

@@ -324,8 +324,7 @@ type StatMdls []*StatMdl
 // CSVHeader return the header for csv fields as a slice of string
 func (tps StatMdls) CSVHeader() (result []string) {
 	return []string{"#" + utils.Tenant, utils.ID, utils.FilterIDs, utils.Weight,
-		utils.QueueLength, utils.TTL, utils.MinItems, utils.MetricIDs, utils.MetricFilterIDs,
-		utils.Stored, utils.BlockersField, utils.ThresholdIDs}
+		utils.QueueLength, utils.TTL, utils.MinItems, utils.MetricIDs, utils.MetricFilterIDs, utils.MetricBlockers, utils.Stored, utils.BlockersField, utils.ThresholdIDs}
 }
 
 func (tps StatMdls) AsTPStats() (result []*utils.TPStatProfile) {
@@ -394,6 +393,9 @@ func (tps StatMdls) AsTPStats() (result []*utils.TPStatProfile) {
 					filterIDs := strings.Split(model.MetricFilterIDs, utils.InfieldSep)
 					stsMetric.FilterIDs = append(stsMetric.FilterIDs, filterIDs...)
 				}
+				if model.MetricBlockers != utils.EmptyString {
+					stsMetric.Blockers = model.MetricBlockers
+				}
 				statMetricsMap[key.TenantID()][metricID] = stsMetric
 			}
 		}
@@ -447,6 +449,7 @@ func APItoModelStats(st *utils.TPStatProfile) (mdls StatMdls) {
 				}
 				mdl.MetricFilterIDs += val
 			}
+			mdl.MetricBlockers = metric.Blockers
 			mdl.MetricIDs = metric.MetricID
 			mdls = append(mdls, mdl)
 		}
@@ -456,14 +459,13 @@ func APItoModelStats(st *utils.TPStatProfile) (mdls StatMdls) {
 
 func APItoStats(tpST *utils.TPStatProfile, timezone string) (st *StatQueueProfile, err error) {
 	st = &StatQueueProfile{
-		Tenant:      tpST.Tenant,
-		ID:          tpST.ID,
-		FilterIDs:   make([]string, len(tpST.FilterIDs)),
-		QueueLength: tpST.QueueLength,
-		MinItems:    tpST.MinItems,
-		Metrics:     make([]*MetricWithFilters, len(tpST.Metrics)),
-		Stored:      tpST.Stored,
-
+		Tenant:       tpST.Tenant,
+		ID:           tpST.ID,
+		FilterIDs:    make([]string, len(tpST.FilterIDs)),
+		QueueLength:  tpST.QueueLength,
+		MinItems:     tpST.MinItems,
+		Metrics:      make([]*MetricWithFilters, len(tpST.Metrics)),
+		Stored:       tpST.Stored,
 		ThresholdIDs: make([]string, len(tpST.ThresholdIDs)),
 	}
 	if tpST.Weights != utils.EmptyString {
@@ -486,6 +488,11 @@ func APItoStats(tpST *utils.TPStatProfile, timezone string) (st *StatQueueProfil
 			MetricID:  metric.MetricID,
 			FilterIDs: metric.FilterIDs,
 		}
+		if metric.Blockers != utils.EmptyString {
+			if st.Metrics[i].Blockers, err = utils.NewBlockersFromString(metric.Blockers, utils.InfieldSep, utils.ANDSep); err != nil {
+				return
+			}
+		}
 	}
 	for i, trh := range tpST.ThresholdIDs {
 		st.ThresholdIDs[i] = trh
@@ -512,6 +519,7 @@ func StatQueueProfileToAPI(st *StatQueueProfile) (tpST *utils.TPStatProfile) {
 	for i, metric := range st.Metrics {
 		tpST.Metrics[i] = &utils.MetricWithFilters{
 			MetricID: metric.MetricID,
+			Blockers: metric.Blockers.String(utils.InfieldSep, utils.ANDSep),
 		}
 		if len(metric.FilterIDs) != 0 {
 			tpST.Metrics[i].FilterIDs = make([]string, len(metric.FilterIDs))
@@ -519,7 +527,6 @@ func StatQueueProfileToAPI(st *StatQueueProfile) (tpST *utils.TPStatProfile) {
 				tpST.Metrics[i].FilterIDs[j] = fltr
 			}
 		}
-
 	}
 	if st.TTL != time.Duration(0) {
 		tpST.TTL = st.TTL.String()

engine/models.go

@@ -59,9 +59,10 @@ type StatMdl struct {
 	MinItems        int    `index:"6" re:""`
 	MetricIDs       string `index:"7" re:""`
 	MetricFilterIDs string `index:"8" re:""`
-	Stored          bool   `index:"9" re:""`
-	Blockers        string `index:"10" re:""`
-	ThresholdIDs    string `index:"11" re:""`
+	MetricBlockers  string `index:"9" re:""`
+	Stored          bool   `index:"10" re:""`
+	Blockers        string `index:"11" re:""`
+	ThresholdIDs    string `index:"12" re:""`
 	CreatedAt       time.Time
 }
 

engine/stats.go

@@ -220,6 +220,18 @@ func (sS *StatS) matchingStatQueuesForEvent(ctx *context.Context, tnt string, st
 		if sqPrfl.TTL > 0 {
 			sq.ttl = utils.DurationPointer(sqPrfl.TTL)
 		}
+		// every metrics has a blocker, verify them
+		for idx, metric := range sqPrfl.Metrics {
+			var blocker bool
+			if blocker, err = BlockerFromDynamics(ctx, metric.Blockers, sS.fltrS, tnt, evNm); err != nil {
+				return
+			}
+			// if we have blocker, ignore the rest of the metrics
+			if blocker {
+				sqPrfl.Metrics = sqPrfl.Metrics[:idx+1]
+				break
+			}
+		}
 		sq.sqPrfl = sqPrfl
 		if sq.weight, err = WeightFromDynamics(ctx, sqPrfl.Weights,
 			sS.fltrS, tnt, evNm); err != nil {