fix: alert eval for "="/"!=" combination with "at least once"/"all th…

[srikanthccv]

…e times"

Fixes #3578

Assuming that prepared queries are correct, the tests mock the required ClickHouse conn in runChQuery and rows.Next returns the rows as if this was an actual database connection and verifies the number of alerts from evaluation.

signoz/pkg/query-service/rules/thresholdRule.go

Lines 420 to 627 in 043e5ca

	func (r *ThresholdRule) runChQuery(ctx context.Context, db clickhouse.Conn, query string) (Vector, error) {
	rows, err := db.Query(ctx, query)
	if err != nil {
	zap.S().Errorf("rule:", r.Name(), "\t failed to get alert query result")
	return nil, err
	}
	columnTypes := rows.ColumnTypes()
	if err != nil {
	return nil, err
	}
	columnNames := rows.Columns()
	if err != nil {
	return nil, err
	}
	vars := make([]interface{}, len(columnTypes))
	for i := range columnTypes {
	vars[i] = reflect.New(columnTypes[i].ScanType()).Interface()
	}
	// []sample list
	var result Vector
	// map[fingerprint]sample
	resultMap := make(map[uint64]Sample, 0)
	// for rates we want to skip the first record
	// but we dont know when the rates are being used
	// so we always pick timeframe - 30 seconds interval
	// and skip the first record for a given label combo
	// NOTE: this is not applicable for raw queries
	skipFirstRecord := make(map[uint64]bool, 0)
	defer rows.Close()
	for rows.Next() {
	if err := rows.Scan(vars...); err != nil {
	return nil, err
	}
	sample := Sample{}
	lbls := labels.NewBuilder(labels.Labels{})
	for i, v := range vars {
	colName := columnNames[i]
	switch v := v.(type) {
	case *string:
	lbls.Set(colName, *v)
	case *time.Time:
	timval := *v
	if colName == "ts" || colName == "interval" {
	sample.Point.T = timval.Unix()
	} else {
	lbls.Set(colName, timval.Format("2006-01-02 15:04:05"))
	}
	case *float64:
	if _, ok := constants.ReservedColumnTargetAliases[colName]; ok {
	sample.Point.V = *v
	} else {
	lbls.Set(colName, fmt.Sprintf("%f", *v))
	}
	case **float64:
	// ch seems to return this type when column is derived from
	// SELECT count(*)/ SELECT count(*)
	floatVal := *v
	if floatVal != nil {
	if _, ok := constants.ReservedColumnTargetAliases[colName]; ok {
	sample.Point.V = *floatVal
	} else {
	lbls.Set(colName, fmt.Sprintf("%f", *floatVal))
	}
	}
	case *float32:
	float32Val := float32(*v)
	if _, ok := constants.ReservedColumnTargetAliases[colName]; ok {
	sample.Point.V = float64(float32Val)
	} else {
	lbls.Set(colName, fmt.Sprintf("%f", float32Val))
	}
	case *uint8, *uint64, *uint16, *uint32:
	if _, ok := constants.ReservedColumnTargetAliases[colName]; ok {
	sample.Point.V = float64(reflect.ValueOf(v).Elem().Uint())
	} else {
	lbls.Set(colName, fmt.Sprintf("%v", reflect.ValueOf(v).Elem().Uint()))
	}
	case *int8, *int16, *int32, *int64:
	if _, ok := constants.ReservedColumnTargetAliases[colName]; ok {
	sample.Point.V = float64(reflect.ValueOf(v).Elem().Int())
	} else {
	lbls.Set(colName, fmt.Sprintf("%v", reflect.ValueOf(v).Elem().Int()))
	}
	default:
	zap.S().Errorf("ruleId:", r.ID(), "\t error: invalid var found in query result", v, columnNames[i])
	}
	}
	if math.IsNaN(sample.Point.V) {
	continue
	}
	// capture lables in result
	sample.Metric = lbls.Labels()
	labelHash := lbls.Labels().Hash()
	// here we walk through values of time series
	// and calculate the final value used to compare
	// with rule target
	if existing, ok := resultMap[labelHash]; ok {
	switch r.matchType() {
	case AllTheTimes:
	if r.compareOp() == ValueIsAbove {
	sample.Point.V = math.Min(existing.Point.V, sample.Point.V)
	resultMap[labelHash] = sample
	} else if r.compareOp() == ValueIsBelow {
	sample.Point.V = math.Max(existing.Point.V, sample.Point.V)
	resultMap[labelHash] = sample
	}
	case AtleastOnce:
	if r.compareOp() == ValueIsAbove {
	sample.Point.V = math.Max(existing.Point.V, sample.Point.V)
	resultMap[labelHash] = sample
	} else if r.compareOp() == ValueIsBelow {
	sample.Point.V = math.Min(existing.Point.V, sample.Point.V)
	resultMap[labelHash] = sample
	}
	case OnAverage:
	sample.Point.V = (existing.Point.V + sample.Point.V) / 2
	resultMap[labelHash] = sample
	case InTotal:
	sample.Point.V = (existing.Point.V + sample.Point.V)
	resultMap[labelHash] = sample
	}
	} else {
	if r.Condition().QueryType() == v3.QueryTypeBuilder {
	// for query builder, time series data
	// we skip the first record to support rate cases correctly
	// improvement(amol): explore approaches to limit this only for
	// rate uses cases
	if exists := skipFirstRecord[labelHash]; exists || !r.shouldSkipFirstRecord() {
	resultMap[labelHash] = sample
	} else {
	// looks like the first record for this label combo, skip it
	skipFirstRecord[labelHash] = true
	}
	} else {
	// for clickhouse raw queries, all records are considered
	// improvement(amol): think about supporting rate queries
	// written by user. may have to skip a record, similar to qb case(above)
	resultMap[labelHash] = sample
	}
	}
	if s, ok := resultMap[labelHash]; ok {
	s.Point.Vs = append(s.Point.Vs, s.Point.V)
	}
	}
	for _, s := range resultMap {
	if r.matchType() == AllTheTimes && r.compareOp() == ValueIsEq {
	for _, v := range s.Point.Vs {
	if v != r.targetVal() { // if any of the values is not equal to target, alert shouldn't be sent
	s.Point.V = v
	}
	}
	} else if r.matchType() == AllTheTimes && r.compareOp() == ValueIsNotEq {
	for _, v := range s.Point.Vs {
	if v == r.targetVal() { // if any of the values is equal to target, alert shouldn't be sent
	s.Point.V = v
	}
	}
	} else if r.matchType() == AtleastOnce && r.compareOp() == ValueIsEq {
	for _, v := range s.Point.Vs {
	if v == r.targetVal() { // if any of the values is equal to target, alert should be sent
	s.Point.V = v
	}
	}
	} else if r.matchType() == AtleastOnce && r.compareOp() == ValueIsNotEq {
	for _, v := range s.Point.Vs {
	if v != r.targetVal() { // if any of the values is not equal to target, alert should be sent
	s.Point.V = v
	}
	}
	}
	}
	zap.S().Debugf("ruleid:", r.ID(), "\t resultmap(potential alerts):", len(resultMap))
	for _, sample := range resultMap {
	// check alert rule condition before dumping results, if sendUnmatchedResults
	// is set then add results irrespective of condition
	if r.opts.SendUnmatched || r.CheckCondition(sample.Point.V) {
	result = append(result, sample)
	}
	}
	if len(result) != 0 {
	zap.S().Infof("For rule %s, with ClickHouseQuery %s, found %d alerts", r.ID(), query, len(result))
	}
	return result, nil
	}

[srikanthccv]

Please review. I would like to get this merged and make it part of release.