Track event rules version in RuntimeConfig

Instead of having to determine the rules version with each incoming
request by the listener, we can just make use the already existing
incremental update mechanism to keep the version up-to-date. This way,
the version will only be updated when a rule is added, changed, or deleted,
so there will be no performance impact on the listener side.

Apart from that, the listener responds and verifies the rules version on
a per-source basis now. As required by the internal document, event rules
will be tied to their source because only this source knows how to evaluate
them. Therefore, the database `rule` table has been extended to include a
`source_id` column, and consequently, each source will only receive the
rules that are relevant to it. Thus, initially, when a source submits an
event, it will likely be rejected but at the same time receive the current
rules, so it can retry the event submission with the correct event rules.
This way, no extra HTTP request is needed to fetch the rules, as we will
always respond with the newest ones whenever we detect that they're using
an outdated event rules config.

Author: yhabteab

internal/config/rule.go

@@ -6,8 +6,36 @@ import (
 	"slices"
 )
 
+// SourceRulesInfo holds information about the rules associated with a specific source.
+type SourceRulesInfo struct {
+	// Version is the version of the rules for the source.
+	//
+	// It is a monotonically increasing number that is updated whenever a rule is added, modified, or deleted.
+	// With each state change of the rules referenced by RuleIDs, the Version will always be incremented
+	// by 1, starting from 0. When there are no configured rules for the source, Version will be reset to 0.
+	//
+	// The Version is not unique across different sources, but it is unique for a specific source at a specific time.
+	Version uint64
+
+	// RuleIDs is a list of rule IDs associated with a specific source.
+	//
+	// It is used to quickly access the rules for a specific source without iterating over all rules.
+	RuleIDs []int64
+}
+
+// RuleSet represents the set of event rules currently loaded in the runtime configuration.
+// It contains the rules and their associated information, such as the source they belong to and their version.
+type RuleSet struct {
+	Rules map[int64]*rule.Rule // rules is a map of rule.Rule by their ID.
+
+	RulesBySource map[int64]*SourceRulesInfo // RulesBySource maps source IDs to their rules and version information.
+}
+
 // applyPendingRules synchronizes changed rules.
 func (r *RuntimeConfig) applyPendingRules() {
+	// Keep track of sources the rules were updated for, so we can update their version later.
+	updatedSources := make(map[int64]struct{})
+
 	incrementalApplyPending(
 		r,
 		&r.Rules, &r.configChange.Rules,
@@ -21,6 +49,17 @@ func (r *RuntimeConfig) applyPendingRules() {
 			}
 
 			newElement.Escalations = make(map[int64]*rule.Escalation)
+			updatedSources[newElement.SourceID] = struct{}{}
+			if r.RulesBySource == nil {
+				r.RulesBySource = make(map[int64]*SourceRulesInfo)
+			}
+
+			// Add the new rule to the per-source rules cache.
+			if sourceInfo := r.RulesBySource[newElement.SourceID]; sourceInfo == nil {
+				r.RulesBySource[newElement.SourceID] = &SourceRulesInfo{RuleIDs: []int64{newElement.ID}}
+			} else {
+				sourceInfo.RuleIDs = append(sourceInfo.RuleIDs, newElement.ID)
+			}
 			return nil
 		},
 		func(curElement, update *rule.Rule) error {
@@ -40,10 +79,35 @@ func (r *RuntimeConfig) applyPendingRules() {
 
 			// ObjectFilter{,Expr} are being initialized by config.IncrementalConfigurableInitAndValidatable.
 			curElement.ObjectFilterExpr = update.ObjectFilterExpr
+			updatedSources[curElement.SourceID] = struct{}{}
 
 			return nil
 		},
-		nil)
+		func(delElement *rule.Rule) error {
+			if sourceInfo, ok := r.RulesBySource[delElement.SourceID]; ok {
+				sourceInfo.RuleIDs = slices.DeleteFunc(sourceInfo.RuleIDs, func(id int64) bool {
+					return id == delElement.ID
+				})
+				if len(sourceInfo.RuleIDs) == 0 {
+					delete(r.RulesBySource, delElement.SourceID) // Remove the source if no rules are left.
+				}
+			}
+			return nil
+		},
+	)
+
+	// After applying the rules, we need to update the version of the sources that were modified.
+	// This is done to ensure that the version is incremented whenever a rule is added, modified,
+	// or deleted only once per applyPendingRules call, even if multiple rules from the same source
+	// were changed.
+	for sourceID := range updatedSources {
+		if r.RulesBySource != nil {
+			if sourceInfo, ok := r.RulesBySource[sourceID]; ok {
+				// Invariant: len(sourceInfo.RuleIDs) > 0 if the source exists in RulesBySource (see delete above).
+				sourceInfo.Version++
+			}
+		}
+	}
 
 	incrementalApplyPending(
 		r,

internal/config/runtime.go

@@ -4,8 +4,10 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+	"fmt"
 	"github.com/icinga/icinga-go-library/database"
 	"github.com/icinga/icinga-go-library/logging"
+	"github.com/icinga/icinga-go-library/notifications"
 	"github.com/icinga/icinga-go-library/types"
 	"github.com/icinga/icinga-notifications/internal/channel"
 	"github.com/icinga/icinga-notifications/internal/recipient"
@@ -61,9 +63,10 @@ type ConfigSet struct {
 	Groups           map[int64]*recipient.Group
 	TimePeriods      map[int64]*timeperiod.TimePeriod
 	Schedules        map[int64]*recipient.Schedule
-	Rules            map[int64]*rule.Rule
 	Sources          map[int64]*Source
 
+	RuleSet // RuleSet contains the currently loaded rules and their version.
+
 	// The following fields contain intermediate values, necessary for the incremental config synchronization.
 	// Furthermore, they allow accessing intermediate tables as everything is referred by pointers.
 	groupMembers             map[recipient.GroupMemberKey]*recipient.GroupMember
@@ -162,6 +165,25 @@ func (r *RuntimeConfig) GetRuleEscalation(escalationID int64) *rule.Escalation {
 	return nil
 }
 
+// GetRulesVersionFor retrieves the version of the rules for a specific source.
+//
+// It returns the version as a hexadecimal string, which is a representation of the version number.
+// If the source does not have any rules associated with it, the version will be set to notifications.EmptyRulesVersion.
+//
+// May not be called while holding the write lock on the RuntimeConfig.
+func (r *RuntimeConfig) GetRulesVersionFor(source int64) string {
+	r.RLock()
+	defer r.RUnlock()
+
+	if r.RulesBySource != nil {
+		if sourceInfo, ok := r.RulesBySource[source]; ok && sourceInfo.Version > 0 {
+			return fmt.Sprintf("%x", sourceInfo.Version)
+		}
+	}
+
+	return notifications.EmptyRulesVersion
+}
+
 // GetContact returns *recipient.Contact by the given username (case-insensitive).
 // Returns nil when the given username doesn't exist.
 func (r *RuntimeConfig) GetContact(username string) *recipient.Contact {

internal/listener/listener.go

@@ -45,7 +45,6 @@ func NewListener(db *database.DB, runtimeConfig *config.RuntimeConfig, logs *log
 
 	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
 }
 
@@ -86,28 +85,6 @@ func (l *Listener) Run(ctx context.Context) error {
 	}
 }
 
-// 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) {
@@ -154,12 +131,16 @@ func (l *Listener) ProcessEvent(w http.ResponseWriter, r *http.Request) {
 	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)
+	// If the client uses an outdated rules version, reject the request but send also the current rules version
+	// and rules for this source back to the client, so it can retry the request with the updated rules.
+	if latestRuleVersion := l.runtimeConfig.GetRulesVersionFor(source.ID); ruleVersion != latestRuleVersion {
+		w.WriteHeader(http.StatusFailedDependency)
+		l.writeSourceRulesInfo(w, source)
+
+		l.logger.Debugw("Abort event processing due to outdated rules version",
+			zap.String("current_version", latestRuleVersion),
+			zap.String("provided_version", ruleVersion),
+			zap.String("source", source.Name))
 		return
 	}
 
@@ -207,15 +188,6 @@ func (l *Listener) ProcessEvent(w http.ResponseWriter, r *http.Request) {
 	_, _ = 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 {
@@ -327,17 +299,39 @@ func (l *Listener) DumpRules(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	l.runtimeConfig.RLock()
+	rules := l.runtimeConfig.Rules
+	l.runtimeConfig.RUnlock()
+
+	enc := json.NewEncoder(w)
+	enc.SetIndent("", "  ")
+	_ = enc.Encode(rules)
+}
+
+// writeSourceRulesInfo writes the rules information for a specific source to the response writer.
+//
+// It includes the rules version and a map of rule IDs to their corresponding rule objects.
+func (l *Listener) writeSourceRulesInfo(w http.ResponseWriter, source *config.Source) {
 	type Response struct {
 		Version string
 		Rules   map[int64]*rule.Rule
 	}
 
 	var resp Response
-	resp.Version = l.getRuleVersion()
+	resp.Version = l.runtimeConfig.GetRulesVersionFor(source.ID)
 
-	l.runtimeConfig.RLock()
-	resp.Rules = l.runtimeConfig.Rules
-	l.runtimeConfig.RUnlock()
+	func() { // Use a function to ensure that the RLock and RUnlock are called before writing the response.
+		l.runtimeConfig.RLock()
+		defer l.runtimeConfig.RUnlock()
+
+		sourceInfo := l.runtimeConfig.RulesBySource[source.ID]
+		if sourceInfo != nil {
+			resp.Rules = make(map[int64]*rule.Rule, len(sourceInfo.RuleIDs))
+			for _, rID := range sourceInfo.RuleIDs {
+				resp.Rules[rID] = l.runtimeConfig.Rules[rID]
+			}
+		}
+	}()
 
 	enc := json.NewEncoder(w)
 	enc.SetIndent("", "  ")

internal/rule/rule.go

@@ -15,6 +15,7 @@ type Rule struct {
 	Name             string                 `db:"name"`
 	TimePeriod       *timeperiod.TimePeriod `db:"-"`
 	TimePeriodID     types.Int              `db:"timeperiod_id"`
+	SourceID         int64                  `db:"source_id"`
 	ObjectFilterExpr types.String           `db:"object_filter"`
 	Escalations      map[int64]*Escalation  `db:"-"`
 }
@@ -37,6 +38,7 @@ func (r *Rule) IncrementalInitAndValidate() error {
 func (r *Rule) MarshalLogObject(encoder zapcore.ObjectEncoder) error {
 	encoder.AddInt64("id", r.ID)
 	encoder.AddString("name", r.Name)
+	encoder.AddInt64("source_id", r.SourceID)
 
 	if r.TimePeriodID.Valid && r.TimePeriodID.Int64 != 0 {
 		encoder.AddInt64("timeperiod_id", r.TimePeriodID.Int64)

schema/mysql/schema.sql

@@ -273,13 +273,15 @@ CREATE TABLE rule (
     id bigint NOT NULL AUTO_INCREMENT,
     name text NOT NULL COLLATE utf8mb4_unicode_ci,
     timeperiod_id bigint,
+    source_id bigint NOT NULL, -- the source this rule belongs to
     object_filter text,
 
     changed_at bigint NOT NULL,
     deleted enum('n', 'y') NOT NULL DEFAULT 'n',
 
     CONSTRAINT pk_rule PRIMARY KEY (id),
-    CONSTRAINT fk_rule_timeperiod FOREIGN KEY (timeperiod_id) REFERENCES timeperiod(id)
+    CONSTRAINT fk_rule_timeperiod FOREIGN KEY (timeperiod_id) REFERENCES timeperiod(id),
+    CONSTRAINT fk_rule_source FOREIGN KEY (source_id) REFERENCES source(id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
 
 CREATE INDEX idx_rule_changed_at ON rule(changed_at);

schema/mysql/upgrades/001.sql

@@ -11,3 +11,10 @@ ALTER TABLE source
   Drop COLUMN icinga2_common_name,
   Drop COLUMN icinga2_insecure_tls,
   MODIFY COLUMN listener_password_hash text NOT NULL;
+
+ALTER TABLE rule
+  ADD COLUMN source_id bigint DEFAULT NULL AFTER timeperiod_id,
+  ADD CONSTRAINT fk_rule_source FOREIGN KEY (source_id) REFERENCES source(id);
+
+UPDATE rule SET source_id = (SELECT id FROM source WHERE type = 'icinga2');
+ALTER TABLE rule MODIFY COLUMN source_id bigint NOT NULL;

schema/pgsql/schema.sql

@@ -320,13 +320,15 @@ CREATE TABLE rule (
     id bigserial,
     name citext NOT NULL,
     timeperiod_id bigint,
+    source_id bigint NOT NULL, -- the source this rule belongs to
     object_filter text,
 
     changed_at bigint NOT NULL,
     deleted boolenum NOT NULL DEFAULT 'n',
 
     CONSTRAINT pk_rule PRIMARY KEY (id),
-    CONSTRAINT fk_rule_timeperiod FOREIGN KEY (timeperiod_id) REFERENCES timeperiod(id)
+    CONSTRAINT fk_rule_timeperiod FOREIGN KEY (timeperiod_id) REFERENCES timeperiod(id),
+    CONSTRAINT fk_rule_source FOREIGN KEY (source_id) REFERENCES source(id)
 );
 
 CREATE INDEX idx_rule_changed_at ON rule(changed_at);

schema/pgsql/upgrades/001.sql

@@ -11,3 +11,10 @@ ALTER TABLE source
   Drop COLUMN icinga2_common_name,
   Drop COLUMN icinga2_insecure_tls,
   ALTER COLUMN listener_password_hash SET NOT NULL;
+
+ALTER TABLE rule
+  ADD COLUMN source_id bigint DEFAULT NULL,
+  ADD CONSTRAINT fk_rule_source FOREIGN KEY (source_id) REFERENCES source(id);
+
+UPDATE rule SET source_id = (SELECT id FROM source WHERE type = 'icinga2');
+ALTER TABLE rule ALTER COLUMN source_id SET NOT NULL;