nmon/internal/alert/query.go

package alert

import (
	"context"
	"encoding/json"
	"fmt"
	"math"
	"strconv"
	"time"

	"github.com/v10networks/nmon/internal/alertpolicy"
)

// query runs a rule and returns one reading per active, trusted entity.
func (ev *Evaluator) query(ctx context.Context, r *compiledRule, now time.Time) ([]reading, error) {
	var rds []Reading
	var err error
	if r.Source != nil {
		rds, err = r.Source.Read(ctx, now)
	} else {
		rds, err = readMetric(ctx, ev.o.Querier, ev.o.StoreName, r, now)
	}
	if err != nil {
		return nil, err
	}
	best := map[string]reading{}
	var order []string
	for _, s := range rds {
		ent, bind, ok := ev.entityOfReading(r, s)
		if !ok {
			continue
		}
		k := ent.Key()
		cur, seen := best[k]
		if !seen {
			order = append(order, k)
		}
		// Duplicate series for one entity: keep the one nearer to alerting.
		if !seen || r.nearer(s.Value, cur.value) {
			best[k] = reading{ent, s.Value, bind}
		}
	}
	out := make([]reading, 0, len(order))
	for _, k := range order {
		out = append(out, best[k])
	}
	return out, nil
}

// readMetric runs the rule's expression and maps each series to the entity its labels name. Series lacking a
// label the rule needs are dropped.
func readMetric(ctx context.Context, q Querier, storeName string, r *compiledRule, now time.Time) ([]Reading, error) {
	qctx, cancel := context.WithTimeout(ctx, queryTimeout)
	defer cancel()
	raw, err := q.Query(qctx, storeName, r.Expr, now)
	if err != nil {
		return nil, err
	}
	series, err := parseVector(raw)
	if err != nil {
		return nil, err
	}
	out := make([]Reading, 0, len(series))
	for _, s := range series {
		rd := Reading{Value: s.value}
		complete := true
		for _, l := range r.Labels {
			v := s.labels[l]
			if v == "" {
				complete = false
				break
			}
			switch l {
			case "device":
				rd.Device = v
			case "iface":
				rd.Iface = v
			case "component":
				rd.Component = v
			}
		}
		if complete {
			out = append(out, rd)
		}
	}
	return out, nil
}

// nearer reports whether a is closer to breaching than b.
func (r *compiledRule) nearer(a, b float64) bool {
	if r.Op == OpLess || r.Op == OpLessEqual {
		return a < b
	}
	return a > b
}

// entityOfReading maps a reading to its entity. It reports false for readings on unreliable interfaces, for
// inactive entities and for entities no alert policy binds.
func (ev *Evaluator) entityOfReading(r *compiledRule, s Reading) (Entity, alertpolicy.Binding, bool) {
	ent := Entity{Rule: r.Name, Device: s.Device, Iface: s.Iface, Component: s.Component}
	if r.SkipUnreliable && ent.Iface != "" && ev.o.Unreliable != nil && ev.o.Unreliable(ent.Iface) {
		return Entity{}, alertpolicy.Binding{}, false
	}
	if ev.o.Active != nil && !ev.o.Active(ent) {
		return Entity{}, alertpolicy.Binding{}, false
	}
	b, ok := ev.o.Policy(ent)
	return ent, b, ok
}

type sample struct {
	labels map[string]string
	value  float64
}

// parseVector reads an instant-query response. NaN samples are dropped (no data).
func parseVector(raw []byte) ([]sample, error) {
	var resp struct {
		Status    string `json:"status"`
		ErrorType string `json:"errorType"`
		Error     string `json:"error"`
		Data      struct {
			ResultType string `json:"resultType"`
			Result     []struct {
				Metric map[string]string  `json:"metric"`
				Value  [2]json.RawMessage `json:"value"`
			} `json:"result"`
		} `json:"data"`
	}
	if err := json.Unmarshal(raw, &resp); err != nil {
		return nil, fmt.Errorf("alert: bad query response: %w", err)
	}
	if resp.Status != "success" {
		return nil, fmt.Errorf("alert: query status %q: %s %s", resp.Status, resp.ErrorType, resp.Error)
	}
	if resp.Data.ResultType != "vector" {
		return nil, fmt.Errorf("alert: query returned %q, want vector", resp.Data.ResultType)
	}
	var out []sample
	for _, r := range resp.Data.Result {
		var str string
		if err := json.Unmarshal(r.Value[1], &str); err != nil {
			return nil, fmt.Errorf("alert: bad sample value %s", r.Value[1])
		}
		v, err := strconv.ParseFloat(str, 64)
		if err != nil {
			return nil, fmt.Errorf("alert: bad sample value %q", str)
		}
		if math.IsNaN(v) {
			continue
		}
		out = append(out, sample{r.Metric, v})
	}
	return out, nil
}