Icinga DB Source: Address Review

- Update IGL to include rules and rule versions in the event. Thus,
  being available in JSON and no longer in a HTTP header.
- Simplify schema rule version incrementation logic by ignoring the edge
  case of sources w/o rules. In this case, the rule will not be reset,
  but further incremented.
- Fix import names.
- Abort processing incoming events for unknown rules. This could only
  occur if there is a race between changing rules and a processed event,
  as rules are initially checked.
- Erase race condition between mutex locks and unlocks for HTTP rule
  endpoint and ensure debug dump rules has a mutex all the time.
- Change the SQL CHECK for the listener_password_hash to allow different
  versions of bcrypt. The "$2y$" thingy is PHP specific to highlight
  "old" bcrypt passwords being affected by a PHP implementation bug, not
  a bcrypt specification bug or change.

Author: oxzi

go.mod

@@ -2,12 +2,14 @@ module github.com/icinga/icinga-notifications
 
 go 1.24.0
 
+toolchain go1.24.6
+
 require (
 	github.com/creasty/defaults v1.8.0
 	github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6
 	github.com/emersion/go-smtp v0.24.0
 	github.com/google/uuid v1.6.0
-	github.com/icinga/icinga-go-library v0.7.3-0.20250904130608-5032573a325e
+	github.com/icinga/icinga-go-library v0.7.3-0.20250909100113-20db23663e45
 	github.com/jhillyerd/enmime v1.3.0
 	github.com/jmoiron/sqlx v1.4.0
 	github.com/okzk/sdnotify v0.0.0-20180710141335-d9becc38acbd

go.sum

@@ -35,8 +35,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/icinga/icinga-go-library v0.7.3-0.20250904130608-5032573a325e h1:yZPWPPCHKozWRm9VedDHd8igJh9uI4U2CJdgl1On+/4=
-github.com/icinga/icinga-go-library v0.7.3-0.20250904130608-5032573a325e/go.mod h1:exEJdfik2GPYrvZM6Gn4BXIBLIGg6OrCCMnILT+mTUs=
+github.com/icinga/icinga-go-library v0.7.3-0.20250909100113-20db23663e45 h1:Wz6ttTYgYB7y8FH7snBSnnllLuzhE0QSp6m3P9b/QfM=
+github.com/icinga/icinga-go-library v0.7.3-0.20250909100113-20db23663e45/go.mod h1:uCENf5EVhNVvXTvhB+jXiwRB2NdLlz8cymseOM4qmI0=
 github.com/jaytaylor/html2text v0.0.0-20230321000545-74c2419ad056 h1:iCHtR9CQyktQ5+f3dMVZfwD2KWJUgm7M0gdL9NGr8KA=
 github.com/jaytaylor/html2text v0.0.0-20230321000545-74c2419ad056/go.mod h1:CVKlgaMiht+LXvHG173ujK6JUhZXKb2u/BQtjPDIvyk=
 github.com/jessevdk/go-flags v1.6.1 h1:Cvu5U8UGrLay1rZfv/zP7iLpSHGUZ/Ou68T0iX1bBK4=

internal/config/rule.go

@@ -12,7 +12,7 @@ type SourceRulesInfo struct {
 	//
 	// 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.
+	// by 1, effectively starting at 1.
 	//
 	// The Version is not unique across different sources, but it is unique for a specific source at a specific time.
 	Version uint64
@@ -36,6 +36,10 @@ 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{})
 
+	if r.RulesBySource == nil {
+		r.RulesBySource = make(map[int64]*SourceRulesInfo)
+	}
+
 	incrementalApplyPending(
 		r,
 		&r.Rules, &r.configChange.Rules,
@@ -49,17 +53,16 @@ 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 {
+			if sourceInfo, ok := r.RulesBySource[newElement.SourceID]; ok {
 				sourceInfo.RuleIDs = append(sourceInfo.RuleIDs, newElement.ID)
+			} else {
+				r.RulesBySource[newElement.SourceID] = &SourceRulesInfo{RuleIDs: []int64{newElement.ID}}
 			}
+
+			updatedSources[newElement.SourceID] = struct{}{}
+
 			return nil
 		},
 		func(curElement, update *rule.Rule) error {
@@ -79,6 +82,7 @@ func (r *RuntimeConfig) applyPendingRules() {
 
 			// ObjectFilter{,Expr} are being initialized by config.IncrementalConfigurableInitAndValidatable.
 			curElement.ObjectFilterExpr = update.ObjectFilterExpr
+
 			updatedSources[curElement.SourceID] = struct{}{}
 
 			return nil
@@ -88,10 +92,10 @@ func (r *RuntimeConfig) applyPendingRules() {
 				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.
-				}
 			}
+
+			updatedSources[delElement.SourceID] = struct{}{}
+
 			return nil
 		},
 	)
@@ -101,11 +105,8 @@ func (r *RuntimeConfig) applyPendingRules() {
 	// 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++
-			}
+		if sourceInfo, ok := r.RulesBySource[sourceID]; ok {
+			sourceInfo.Version++
 		}
 	}
 

internal/config/runtime.go

@@ -165,6 +165,15 @@ func (r *RuntimeConfig) GetRuleEscalation(escalationID int64) *rule.Escalation {
 	return nil
 }
 
+// RulesVersionString formats a rule version.
+func (r *RuntimeConfig) RulesVersionString(version uint64) string {
+	if version > 0 {
+		return fmt.Sprintf("%x", version)
+	}
+
+	return source.EmptyRulesVersion
+}
+
 // 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.
@@ -177,11 +186,11 @@ func (r *RuntimeConfig) GetRulesVersionFor(srcId int64) string {
 
 	if r.RulesBySource != nil {
 		if sourceInfo, ok := r.RulesBySource[srcId]; ok && sourceInfo.Version > 0 {
-			return fmt.Sprintf("%x", sourceInfo.Version)
+			return r.RulesVersionString(sourceInfo.Version)
 		}
 	}
 
-	return source.EmptyRulesVersion
+	return r.RulesVersionString(0)
 }
 
 // GetContact returns *recipient.Contact by the given username (case-insensitive).

internal/event/event.go

@@ -5,8 +5,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strconv"
-	"strings"
 	"time"
 
 	"github.com/icinga/icinga-go-library/database"
@@ -33,8 +31,6 @@ type Event struct {
 	SourceId int64     `json:"-"`
 	ID       int64     `json:"-"`
 
-	MatchedRules map[int64]struct{} `json:"-"` // MatchedRules contains the event rule IDs received from source.
-
 	*baseEv.Event `json:",inline"`
 }
 
@@ -85,28 +81,6 @@ func (e *Event) String() string {
 	return fmt.Sprintf("[time=%s type=%q severity=%s]", e.Time, e.Type, e.Severity.String())
 }
 
-// LoadMatchedRulesFromString parses a comma-separated string of rule IDs and loads them into the event's MatchedRules.
-//
-// Returns an error if any of the rule IDs cannot be parsed as an int64.
-func (e *Event) LoadMatchedRulesFromString(ruleIdsStr string) error {
-	if e.MatchedRules == nil {
-		e.MatchedRules = make(map[int64]struct{})
-	}
-
-	if ruleIdsStr == "" {
-		return nil // No rule IDs to load, nothing to do.
-	}
-
-	for _, ruleIdStr := range strings.Split(ruleIdsStr, ",") {
-		ruleId, err := strconv.ParseInt(ruleIdStr, 10, 64)
-		if err != nil {
-			return fmt.Errorf("cannot parse rule ID %q: %w", ruleIdStr, err)
-		}
-		e.MatchedRules[ruleId] = struct{}{}
-	}
-	return nil
-}
-
 func (e *Event) FullString() string {
 	var b bytes.Buffer
 	_, _ = fmt.Fprintf(&b, "Event:\n")

internal/incident/db_types.go

@@ -3,7 +3,7 @@ package incident
 import (
 	"context"
 	"github.com/icinga/icinga-go-library/database"
-	baseEv "github.com/icinga/icinga-go-library/notifications/event"
+	"github.com/icinga/icinga-go-library/notifications/event"
 	"github.com/icinga/icinga-go-library/types"
 	"github.com/icinga/icinga-notifications/internal/recipient"
 	"github.com/jmoiron/sqlx"
@@ -74,8 +74,8 @@ type HistoryRow struct {
 	Time              types.UnixMilli   `db:"time"`
 	Type              HistoryEventType  `db:"type"`
 	ChannelID         types.Int         `db:"channel_id"`
-	NewSeverity       baseEv.Severity   `db:"new_severity"`
-	OldSeverity       baseEv.Severity   `db:"old_severity"`
+	NewSeverity       event.Severity    `db:"new_severity"`
+	OldSeverity       event.Severity    `db:"old_severity"`
 	NewRecipientRole  ContactRole       `db:"new_recipient_role"`
 	OldRecipientRole  ContactRole       `db:"old_recipient_role"`
 	Message           types.String      `db:"message"`

internal/incident/incident.go

@@ -394,25 +394,16 @@ func (i *Incident) handleMuteUnmute(ctx context.Context, tx *sqlx.Tx, ev *event.
 }
 
 // applyMatchingRules walks through the rule IDs obtained from source and generates a RuleMatched history entry.
-//
-// Unknown event rule IDs are ignored, and a warning is logged. Otherwise, if the rule is not already matched on this
-// incident's Object with previous events, it is added to the incident rules and a corresponding history is generated.
-//
-// Returns an error on any database failure.
 func (i *Incident) applyMatchingRules(ctx context.Context, tx *sqlx.Tx, ev *event.Event) error {
 	if i.Rules == nil {
 		i.Rules = make(map[int64]struct{})
 	}
 
-	for ruleID := range ev.MatchedRules {
-		r, ok := i.runtimeConfig.Rules[ruleID]
+	for _, ruleId := range ev.RuleIds {
+		r, ok := i.runtimeConfig.Rules[ruleId]
 		if !ok {
-			// Usually, sources aren't expected to deliberately send rule IDs that don't exist, but if they do,
-			// we warn about it and move on. Other causes of this could be a rule being deleted just right after
-			// the version validation by the listener, and before the incident acquired a read lock on the runtime
-			// config in the ProcessEvent method.
-			i.logger.Warnw("Event refers to non-existing event rule, might got deleted", zap.Int64("rule_id", ruleID))
-			continue
+			i.logger.Errorw("Event refers to non-existing event rule, might got deleted", zap.Int64("rule_id", ruleId))
+			return fmt.Errorf("cannot apply unknown rule %d", ruleId)
 		}
 
 		if _, ok := i.Rules[r.ID]; !ok {

internal/listener/listener.go

@@ -9,7 +9,6 @@ import (
 	"github.com/icinga/icinga-go-library/database"
 	"github.com/icinga/icinga-go-library/logging"
 	baseEv "github.com/icinga/icinga-go-library/notifications/event"
-	"github.com/icinga/icinga-go-library/notifications/source"
 	"github.com/icinga/icinga-notifications/internal"
 	"github.com/icinga/icinga-notifications/internal/config"
 	"github.com/icinga/icinga-notifications/internal/daemon"
@@ -124,38 +123,31 @@ func (l *Listener) ProcessEvent(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	src, validAuth := l.sourceFromAuthOrAbort(w, r)
-	if !validAuth {
+	src, isAuthenticated := l.sourceFromAuthOrAbort(w, r)
+	if !isAuthenticated {
+		// Listener.sourceFromAuthOrAbort writes 401 response by itself; no abort() necessary.
 		return
 	}
 
-	ruleIdsStr := r.Header.Get(source.XIcingaRulesId)
-	ruleVersion := r.Header.Get(source.XIcingaRulesVersion)
+	var ev event.Event
+	if err := json.NewDecoder(r.Body).Decode(&ev); err != nil {
+		abort(http.StatusBadRequest, nil, "cannot parse JSON body: %v", err)
+		return
+	}
 
 	// 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(src.ID); ruleVersion != latestRuleVersion {
+	if latestRuleVersion := l.runtimeConfig.GetRulesVersionFor(src.ID); ev.RulesVersion != latestRuleVersion {
 		w.WriteHeader(http.StatusPreconditionFailed)
 		l.writeSourceRulesInfo(w, src)
 
 		l.logger.Debugw("Abort event processing due to outdated rules version",
 			zap.String("current_version", latestRuleVersion),
-			zap.String("provided_version", ruleVersion),
+			zap.String("provided_version", ev.RulesVersion),
 			zap.String("source", src.Name))
 		return
 	}
 
-	var ev event.Event
-	if err := ev.LoadMatchedRulesFromString(ruleIdsStr); err != nil {
-		abort(http.StatusBadRequest, nil, "cannot parse %s header: %v", source.XIcingaRulesId, err)
-		return
-	}
-
-	if err := json.NewDecoder(r.Body).Decode(&ev); err != nil {
-		abort(http.StatusBadRequest, nil, "cannot parse JSON body: %v", err)
-		return
-	}
-
 	ev.Time = time.Now()
 	ev.SourceId = src.ID
 	if ev.Type == baseEv.TypeUnknown {
@@ -301,12 +293,11 @@ func (l *Listener) DumpRules(w http.ResponseWriter, r *http.Request) {
 	}
 
 	l.runtimeConfig.RLock()
-	rules := l.runtimeConfig.Rules
-	l.runtimeConfig.RUnlock()
+	defer l.runtimeConfig.RUnlock()
 
 	enc := json.NewEncoder(w)
 	enc.SetIndent("", "  ")
-	_ = enc.Encode(rules)
+	_ = enc.Encode(l.runtimeConfig.Rules)
 }
 
 // writeSourceRulesInfo writes the rules information for a specific source to the response writer.
@@ -319,14 +310,13 @@ func (l *Listener) writeSourceRulesInfo(w http.ResponseWriter, source *config.So
 	}
 
 	var resp Response
-	resp.Version = l.runtimeConfig.GetRulesVersionFor(source.ID)
 
 	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 {
+		if sourceInfo, ok := l.runtimeConfig.RulesBySource[source.ID]; ok {
+			resp.Version = l.runtimeConfig.RulesVersionString(sourceInfo.Version)
 			resp.Rules = make(map[int64]*rule.Rule, len(sourceInfo.RuleIDs))
 			for _, rID := range sourceInfo.RuleIDs {
 				resp.Rules[rID] = l.runtimeConfig.Rules[rID]

internal/rule/rule.go

@@ -22,15 +22,6 @@ type Rule struct {
 
 // 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
-	// 	}
-
-	// 	r.ObjectFilter = f
-	// }
-
 	return nil
 }
 

schema/mysql/schema.sql

@@ -217,7 +217,7 @@ CREATE TABLE source (
     -- The hash is a PHP password_hash with PASSWORD_DEFAULT algorithm, defaulting to bcrypt. This check roughly ensures
     -- that listener_password_hash can only be populated with bcrypt hashes.
     -- https://icinga.com/docs/icinga-web/latest/doc/20-Advanced-Topics/#manual-user-creation-for-database-authentication-backend
-    CONSTRAINT ck_source_bcrypt_listener_password_hash CHECK (listener_password_hash LIKE '$2y$%'),
+    CONSTRAINT ck_source_bcrypt_listener_password_hash CHECK (listener_password_hash LIKE '$2_$%'),
 
     CONSTRAINT pk_source PRIMARY KEY (id)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;