Move Rule evaluation to Source

This is a first version to move the rule evaluation from Icinga
Notifications into the source, starting with Icinga DB.

For an end user or another computer, the /process-event API endpoint was
slightly modified. Two new HTTP request headers were introduced to tell
which rules match and to share the state.

Another new /event-rules API endpoint allows querying all rules.

Author: oxzi

internal/config/rule.go

@@ -39,7 +39,6 @@ func (r *RuntimeConfig) applyPendingRules() {
 			}
 
 			// ObjectFilter{,Expr} are being initialized by config.IncrementalConfigurableInitAndValidatable.
-			curElement.ObjectFilter = update.ObjectFilter
 			curElement.ObjectFilterExpr = update.ObjectFilterExpr
 
 			return nil

internal/config/verify.go

@@ -248,10 +248,6 @@ func (r *RuntimeConfig) debugVerifyRule(id int64, rule *rule.Rule) error {
 		}
 	}
 
-	if rule.ObjectFilterExpr.Valid && rule.ObjectFilter == nil {
-		return fmt.Errorf("rule has a ObjectFilterExpr but ObjectFilter is nil")
-	}
-
 	for escalationID, escalation := range rule.Escalations {
 		if escalation == nil {
 			return fmt.Errorf("rule.Escalations[%d] is nil", escalationID)

internal/incident/incident.go

@@ -4,9 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sync"
-	"time"
-
 	"github.com/icinga/icinga-go-library/database"
 	"github.com/icinga/icinga-go-library/types"
 	"github.com/icinga/icinga-notifications/internal/config"
@@ -18,6 +15,10 @@ import (
 	"github.com/icinga/icinga-notifications/internal/rule"
 	"github.com/jmoiron/sqlx"
 	"go.uber.org/zap"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
 )
 
 type ruleID = int64
@@ -178,7 +179,7 @@ func (i *Incident) ProcessEvent(ctx context.Context, ev *event.Event) error {
 
 		// Check if any (additional) rules match this object. Filters of rules that already have a state don't have
 		// to be checked again, these rules already matched and stay effective for the ongoing incident.
-		err = i.evaluateRules(ctx, tx, ev.ID)
+		err = i.evaluateRules(ctx, tx, ev)
 		if err != nil {
 			return err
 		}
@@ -398,26 +399,36 @@ func (i *Incident) handleMuteUnmute(ctx context.Context, tx *sqlx.Tx, ev *event.
 // evaluateRules evaluates all the configured rules for this *incident.Object and
 // generates history entries for each matched rule.
 // Returns error on database failure.
-func (i *Incident) evaluateRules(ctx context.Context, tx *sqlx.Tx, eventID int64) error {
+func (i *Incident) evaluateRules(ctx context.Context, tx *sqlx.Tx, ev *event.Event) error {
 	if i.Rules == nil {
 		i.Rules = make(map[int64]struct{})
 	}
 
-	for _, r := range i.runtimeConfig.Rules {
-		if _, ok := i.Rules[r.ID]; !ok {
-			matched, err := r.Eval(i.Object)
-			if err != nil {
-				i.logger.Warnw("Failed to evaluate object filter", zap.Object("rule", r), zap.Error(err))
-			}
+	ruleIdsStr, ok := ev.ExtraTags["rules"]
+	if !ok {
+		return errors.New("event has no rules extra tag marker")
+	}
+	ruleIdsArr := strings.Split(ruleIdsStr, ",")
 
-			if err != nil || !matched {
-				continue
-			}
+	ruleIds := make(map[ruleID]struct{})
+	for _, idStr := range ruleIdsArr {
+		idInt, err := strconv.ParseInt(idStr, 10, 64)
+		if err != nil {
+			return fmt.Errorf("cannot parse %q from rules as int64", idStr)
+		}
+		ruleIds[idInt] = struct{}{}
+	}
 
+	for _, r := range i.runtimeConfig.Rules {
+		if _, ok := ruleIds[r.ID]; !ok {
+			continue
+		}
+
+		if _, ok := i.Rules[r.ID]; !ok {
 			i.Rules[r.ID] = struct{}{}
 			i.logger.Infow("Rule matches", zap.Object("rule", r))
 
-			err = i.AddRuleMatched(ctx, tx, r)
+			err := i.AddRuleMatched(ctx, tx, r)
 			if err != nil {
 				i.logger.Errorw("Failed to upsert incident rule", zap.Object("rule", r), zap.Error(err))
 				return err
@@ -426,7 +437,7 @@ func (i *Incident) evaluateRules(ctx context.Context, tx *sqlx.Tx, eventID int64
 			hr := &HistoryRow{
 				IncidentID: i.Id,
 				Time:       types.UnixMilli(time.Now()),
-				EventID:    types.MakeInt(eventID, types.TransformZeroIntToNull),
+				EventID:    types.MakeInt(ev.ID, types.TransformZeroIntToNull),
 				RuleID:     types.MakeInt(r.ID, types.TransformZeroIntToNull),
 				Type:       RuleMatched,
 			}

internal/listener/listener.go

@@ -13,6 +13,7 @@ import (
 	"github.com/icinga/icinga-notifications/internal/daemon"
 	"github.com/icinga/icinga-notifications/internal/event"
 	"github.com/icinga/icinga-notifications/internal/incident"
+	"github.com/icinga/icinga-notifications/internal/rule"
 	"go.uber.org/zap"
 	"net/http"
 	"time"
@@ -39,9 +40,11 @@ func NewListener(db *database.DB, runtimeConfig *config.RuntimeConfig, logs *log
 	debugMux.HandleFunc("/dump-config", l.DumpConfig)
 	debugMux.HandleFunc("/dump-incidents", l.DumpIncidents)
 	debugMux.HandleFunc("/dump-schedules", l.DumpSchedules)
+	debugMux.HandleFunc("/dump-rules", l.DumpRules)
 
 	l.mux.Handle("/debug/", http.StripPrefix("/debug", l.requireDebugAuth(debugMux)))
 	l.mux.HandleFunc("/process-event", l.ProcessEvent)
+	l.mux.HandleFunc("/event-rules", l.RulesForFilters)
 	return l
 }
 
@@ -82,7 +85,45 @@ func (l *Listener) Run(ctx context.Context) error {
 	}
 }
 
-func (l *Listener) ProcessEvent(w http.ResponseWriter, req *http.Request) {
+// getRuleVersion returns the latest rule version.
+//
+// Technically, the rule version is an encoded string representation of the latest changed_at value from each rule.
+// Being an implementation detail, it might change over time. For the moment, a simple equality check is enough.
+func (l *Listener) getRuleVersion() string {
+	l.runtimeConfig.RLock()
+	defer l.runtimeConfig.RUnlock()
+
+	var latest time.Time
+	for _, r := range l.runtimeConfig.Rules {
+		if t := r.ChangedAt.Time(); t.After(latest) {
+			latest = t
+		}
+	}
+
+	if latest.IsZero() {
+		return "NA"
+	}
+
+	return fmt.Sprintf("%x", latest.UnixNano())
+}
+
+// sourceFromAuthOrAbort extracts a *config.Source from the HTTP Basic Auth. If the credentials are wrong, (nil, false) is
+// returned and 401 was written back to the response writer.
+func (l *Listener) sourceFromAuthOrAbort(w http.ResponseWriter, r *http.Request) (*config.Source, bool) {
+	if authUser, authPass, authOk := r.BasicAuth(); authOk {
+		src := l.runtimeConfig.GetSourceFromCredentials(authUser, authPass, l.logger)
+		if src != nil {
+			return src, true
+		}
+	}
+
+	w.Header().Set("WWW-Authenticate", `Basic realm="icinga-notifications"`)
+	w.WriteHeader(http.StatusUnauthorized)
+	_, _ = fmt.Fprintln(w, "please provide the debug-password as basic auth credentials (user is ignored)")
+	return nil, false
+}
+
+func (l *Listener) ProcessEvent(w http.ResponseWriter, r *http.Request) {
 	// abort the current connection by sending the status code and an error both to the log and back to the client.
 	abort := func(statusCode int, ev *event.Event, format string, a ...any) {
 		msg := format
@@ -99,24 +140,37 @@ func (l *Listener) ProcessEvent(w http.ResponseWriter, req *http.Request) {
 		logger.Debugw("Abort listener submitted event processing")
 	}
 
-	if req.Method != http.MethodPost {
+	if r.Method != http.MethodPost {
 		abort(http.StatusMethodNotAllowed, nil, "POST required")
 		return
 	}
 
-	var source *config.Source
-	if authUser, authPass, authOk := req.BasicAuth(); authOk {
-		source = l.runtimeConfig.GetSourceFromCredentials(authUser, authPass, l.logger)
+	source, validAuth := l.sourceFromAuthOrAbort(w, r)
+	if !validAuth {
+		return
 	}
-	if source == nil {
-		w.Header().Set("WWW-Authenticate", `Basic realm="icinga-notifications"`)
-		abort(http.StatusUnauthorized, nil, "HTTP authorization required")
+
+	ruleIdsStr := r.Header.Get("X-Rule-Ids")
+	ruleVersion := r.Header.Get("X-Rule-Version")
+
+	if latestRuleVersion := l.getRuleVersion(); ruleVersion != latestRuleVersion {
+		abort(http.StatusFailedDependency,
+			nil,
+			"X-Rule-Version %q does not match %q, refetch rules",
+			ruleVersion,
+			latestRuleVersion)
+		return
+	}
+
+	// TODO: parse and verify ruleIdsStr
+	if ruleIdsStr == "" {
+		// Case for empty X-Rule-Ids where the event was just sent to check if new rules are available (previous if).
+		abort(http.StatusNoContent, nil, "dismissed event due to empty X-Rule-Ids")
 		return
 	}
 
 	var ev event.Event
-	err := json.NewDecoder(req.Body).Decode(&ev)
-	if err != nil {
+	if err := json.NewDecoder(r.Body).Decode(&ev); err != nil {
 		abort(http.StatusBadRequest, nil, "cannot parse JSON body: %v", err)
 		return
 	}
@@ -136,8 +190,11 @@ func (l *Listener) ProcessEvent(w http.ResponseWriter, req *http.Request) {
 		return
 	}
 
+	ev.ExtraTags = make(map[string]string)
+	ev.ExtraTags["rules"] = ruleIdsStr
+
 	l.logger.Infow("Processing event", zap.String("event", ev.String()))
-	err = incident.ProcessEvent(context.Background(), l.db, l.logs, l.runtimeConfig, &ev)
+	err := incident.ProcessEvent(context.Background(), l.db, l.logs, l.runtimeConfig, &ev)
 	if errors.Is(err, event.ErrSuperfluousStateChange) || errors.Is(err, event.ErrSuperfluousMuteUnmuteEvent) {
 		abort(http.StatusNotAcceptable, &ev, "%v", err)
 		return
@@ -149,11 +206,20 @@ func (l *Listener) ProcessEvent(w http.ResponseWriter, req *http.Request) {
 
 	l.logger.Infow("Successfully processed event", zap.String("event", ev.String()))
 
-	w.WriteHeader(http.StatusOK)
+	w.WriteHeader(http.StatusAccepted)
 	_, _ = fmt.Fprintln(w, "event processed successfully")
 	_, _ = fmt.Fprintln(w)
 }
 
+func (l *Listener) RulesForFilters(w http.ResponseWriter, r *http.Request) {
+	_, validAuth := l.sourceFromAuthOrAbort(w, r)
+	if !validAuth {
+		return
+	}
+
+	l.DumpRules(w, r)
+}
+
 // requireDebugAuth is a middleware that checks if the valid debug password was provided. If there is no password
 // configured or the supplied password is incorrect, it sends an error code and does not redirect the request.
 func (l *Listener) requireDebugAuth(next http.Handler) http.Handler {
@@ -256,3 +322,28 @@ func (l *Listener) DumpSchedules(w http.ResponseWriter, r *http.Request) {
 		_, _ = fmt.Fprintln(w)
 	}
 }
+
+// DumpRules is used as /debug prefixed endpoint to dump all rules. The authorization has to be done beforehand.
+func (l *Listener) DumpRules(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		_, _ = fmt.Fprintln(w, "GET required")
+		return
+	}
+
+	type Response struct {
+		Version string
+		Rules   map[int64]*rule.Rule
+	}
+
+	var resp Response
+	resp.Version = l.getRuleVersion()
+
+	l.runtimeConfig.RLock()
+	resp.Rules = l.runtimeConfig.Rules
+	l.runtimeConfig.RUnlock()
+
+	enc := json.NewEncoder(w)
+	enc.SetIndent("", "  ")
+	_ = enc.Encode(resp)
+}

internal/rule/rule.go

@@ -3,7 +3,6 @@ package rule
 import (
 	"github.com/icinga/icinga-go-library/types"
 	"github.com/icinga/icinga-notifications/internal/config/baseconf"
-	"github.com/icinga/icinga-notifications/internal/filter"
 	"github.com/icinga/icinga-notifications/internal/recipient"
 	"github.com/icinga/icinga-notifications/internal/timeperiod"
 	"go.uber.org/zap/zapcore"
@@ -16,21 +15,20 @@ type Rule struct {
 	Name             string                 `db:"name"`
 	TimePeriod       *timeperiod.TimePeriod `db:"-"`
 	TimePeriodID     types.Int              `db:"timeperiod_id"`
-	ObjectFilter     filter.Filter          `db:"-"`
 	ObjectFilterExpr types.String           `db:"object_filter"`
 	Escalations      map[int64]*Escalation  `db:"-"`
 }
 
 // IncrementalInitAndValidate implements the config.IncrementalConfigurableInitAndValidatable interface.
 func (r *Rule) IncrementalInitAndValidate() error {
-	if r.ObjectFilterExpr.Valid {
-		f, err := filter.Parse(r.ObjectFilterExpr.String)
-		if err != nil {
-			return err
-		}
+	// if r.ObjectFilterExpr.Valid {
+	// 	f, err := filter.Parse(r.ObjectFilterExpr.String)
+	// 	if err != nil {
+	// 		return err
+	// 	}
 
-		r.ObjectFilter = f
-	}
+	// 	r.ObjectFilter = f
+	// }
 
 	return nil
 }
@@ -50,16 +48,6 @@ func (r *Rule) MarshalLogObject(encoder zapcore.ObjectEncoder) error {
 	return nil
 }
 
-// Eval evaluates the configured object filter for the provided filterable.
-// Returns always true if the current rule doesn't have a configured object filter.
-func (r *Rule) Eval(filterable filter.Filterable) (bool, error) {
-	if r.ObjectFilter == nil {
-		return true, nil
-	}
-
-	return r.ObjectFilter.Eval(filterable)
-}
-
 // ContactChannels stores a set of channel IDs for each set of individual contacts.
 type ContactChannels map[*recipient.Contact]map[int64]bool