Move Rule evaluation to Source

cmd/channels/email/main.go

@@ -8,9 +8,9 @@ import (
 	"github.com/emersion/go-sasl"
 	"github.com/emersion/go-smtp"
 	"github.com/google/uuid"
+	"github.com/icinga/icinga-go-library/notifications/plugin"
 	"github.com/icinga/icinga-go-library/types"
 	"github.com/icinga/icinga-notifications/internal"
-	"github.com/icinga/icinga-notifications/pkg/plugin"
 	"github.com/jhillyerd/enmime"
 	"net"
 	"net/mail"

cmd/channels/rocketchat/main.go

@@ -5,8 +5,8 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/icinga/icinga-go-library/notifications/plugin"
 	"github.com/icinga/icinga-notifications/internal"
-	"github.com/icinga/icinga-notifications/pkg/plugin"
 	"net/http"
 	"time"
 )

cmd/channels/webhook/main.go

@@ -4,8 +4,8 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"github.com/icinga/icinga-go-library/notifications/plugin"
 	"github.com/icinga/icinga-notifications/internal"
-	"github.com/icinga/icinga-notifications/pkg/plugin"
 	"io"
 	"net/http"
 	"slices"

cmd/icinga-notifications/main.go

@@ -9,7 +9,6 @@ import (
 	"github.com/icinga/icinga-notifications/internal/channel"
 	"github.com/icinga/icinga-notifications/internal/config"
 	"github.com/icinga/icinga-notifications/internal/daemon"
-	"github.com/icinga/icinga-notifications/internal/icinga2"
 	"github.com/icinga/icinga-notifications/internal/incident"
 	"github.com/icinga/icinga-notifications/internal/listener"
 	"github.com/icinga/icinga-notifications/internal/object"
@@ -48,20 +47,11 @@ func main() {
 
 	channel.UpsertPlugins(ctx, conf.ChannelsDir, logs.GetChildLogger("channel"), db)
 
-	icinga2Launcher := &icinga2.Launcher{
-		Ctx:           ctx,
-		Logs:          logs,
-		Db:            db,
-		RuntimeConfig: nil, // Will be set below as it is interconnected..
-	}
-
-	runtimeConfig := config.NewRuntimeConfig(icinga2Launcher.Launch, logs, db)
+	runtimeConfig := config.NewRuntimeConfig(logs, db)
 	if err := runtimeConfig.UpdateFromDatabase(ctx); err != nil {
 		logger.Fatalf("Failed to load config from database %+v", err)
 	}
 
-	icinga2Launcher.RuntimeConfig = runtimeConfig
-
 	go runtimeConfig.PeriodicUpdates(ctx, 1*time.Second)
 
 	err = incident.LoadOpenIncidents(ctx, db, logs.GetChildLogger("incident"), runtimeConfig)
@@ -75,9 +65,6 @@ func main() {
 		logger.Fatalf("Failed to restore muted objects: %+v", err)
 	}
 
-	// Wait to load open incidents from the database before either starting Event Stream Clients or starting the Listener.
-	icinga2Launcher.Ready()
-
 	// When Icinga Notifications is started by systemd, we've to notify systemd that we're ready.
 	_ = sdnotify.Ready()
 

config.example.yml

@@ -66,7 +66,6 @@ database:
 #  options:
     #channel:
     #database:
-    #icinga2:
     #incident:
     #listener:
     #runtime-updates:

doc/03-Configuration.md

@@ -95,7 +95,6 @@ Configuration of the logging component used by Icinga Notifications.
 |-----------------|---------------------------------------------------------------------------|
 | channel         | Notification channels, their configuration and output.                    |
 | database        | Database connection status and queries.                                   |
-| icinga2         | Icinga 2 API communications, including the Event Stream.                  |
 | incident        | Incident management and changes.                                          |
 | listener        | HTTP listener for event submission and debugging.                         |
 | runtime-updates | Configuration changes through Icinga Notifications Web from the database. |

doc/20-HTTP-API.md

@@ -5,14 +5,25 @@ via `listen` and `debug-password`.
 
 ## Process Event
 
-One possible source next to the Icinga 2 API is event submission to the Icinga Notifications HTTP API listener.
-After creating a source with type _Other_ in Icinga Notifications Web,
-the specified credentials can be used for HTTP Basic Authentication of a JSON-encoded
-[Event](https://github.com/Icinga/icinga-notifications/blob/main/internal/event/event.go).
+Events can be submitted to Icinga Notifications using the `/process-event` HTTP API endpoint.
+
+After creating a source in Icinga Notifications Web,
+the specified credentials can be used via HTTP Basic Authentication to submit a JSON-encoded
+[`Event`](https://github.com/Icinga/icinga-go-library/blob/main/notifications/event/event.go).
 
 The authentication is performed via HTTP Basic Authentication, expecting `source-${id}` as the username,
 `${id}` being the source's `id` within the database, and the configured password.
 
+Events sent to Icinga Notifications are expected to match rules that describe further event escalations.
+These rules can be created in the web interface.
+Next to an array of `rule_ids`, a `rules_version` must be provided to ensure that the source has no outdated state.
+
+When the submitted `rules_version` is either outdated or empty, the `/process-event` endpoint returns an HTTP 412 response.
+The response's body is a JSON-encoded version of the
+[`RulesInfo`](https://github.com/Icinga/icinga-go-library/blob/main/notifications/source/client.go),
+containing the latest `rules_version` together with all rules for this source.
+After reevaluating these rules, one can resubmit the event with the updated `rules_version`.
+
 ```
 curl -v -u 'source-2:insecureinsecure' -d '@-' 'http://localhost:5680/process-event' <<EOF
 {
@@ -35,7 +46,9 @@ curl -v -u 'source-2:insecureinsecure' -d '@-' 'http://localhost:5680/process-ev
   "type": "state",
   "severity": "crit",
   "username": "",
-  "message": "Something went somewhere very wrong."
+  "message": "Something went somewhere very wrong.",
+  "rules_version": "23",
+  "rule_ids": ["0"]
 }
 EOF
 ```
@@ -62,6 +75,15 @@ The current incidents can be dumped as JSON.
 curl -v -u ':debug-password' 'http://localhost:5680/debug/dump-incidents'
 ```
 
+
+### Dump Rules
+
+The current rules can be dumped as JSON.
+
+```
+curl -v -u ':debug-password' 'http://localhost:5680/debug/dump-rules'
+```
+
 ### Dump Schedules
 
 All schedules with their assignee can be dumped in a human-readable form.

go.mod

@@ -2,13 +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/hashicorp/golang-lru/v2 v2.0.7
-	github.com/icinga/icinga-go-library v0.7.2
+	github.com/icinga/icinga-go-library v0.7.3-0.20251022120618-6600889adc38
 	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,10 +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/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
-github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
-github.com/icinga/icinga-go-library v0.7.2 h1:6ilUeE9F9OqxxJXNR9URWDf6zOqsdhjjR9w1MUXY9Kg=
-github.com/icinga/icinga-go-library v0.7.2/go.mod h1:HZTiYD+N+9FZIVpPdUEJWJnc6sLvrIRO03jvkdkmUEU=
+github.com/icinga/icinga-go-library v0.7.3-0.20251022120618-6600889adc38 h1:5RNrPZCwvqm2/06i9dUCJtcBV+tR8WgUKtHne2sOaA8=
+github.com/icinga/icinga-go-library v0.7.3-0.20251022120618-6600889adc38/go.mod h1:L80M/ufoqFJJjZcdnfsTp6eFl06vm3JuvSWlGcDf708=
 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/channel/channel.go

@@ -4,11 +4,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/icinga/icinga-go-library/notifications/plugin"
 	"github.com/icinga/icinga-notifications/internal/config/baseconf"
 	"github.com/icinga/icinga-notifications/internal/contracts"
 	"github.com/icinga/icinga-notifications/internal/event"
 	"github.com/icinga/icinga-notifications/internal/recipient"
-	"github.com/icinga/icinga-notifications/pkg/plugin"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"net/url"
@@ -187,7 +187,7 @@ func (c *Channel) Notify(contact *recipient.Contact, i contracts.Incident, ev *e
 		Incident: &plugin.Incident{
 			Id:       i.ID(),
 			Url:      incidentUrl.String(),
-			Severity: i.SeverityString(),
+			Severity: i.IncidentSeverity(),
 		},
 		Event: &plugin.Event{
 			Time:     ev.Time,

internal/channel/plugin.go

@@ -7,9 +7,9 @@ import (
 	"fmt"
 	"github.com/icinga/icinga-go-library/database"
 	"github.com/icinga/icinga-go-library/logging"
+	"github.com/icinga/icinga-go-library/notifications/plugin"
+	"github.com/icinga/icinga-go-library/notifications/rpc"
 	"github.com/icinga/icinga-notifications/internal/daemon"
-	"github.com/icinga/icinga-notifications/pkg/plugin"
-	"github.com/icinga/icinga-notifications/pkg/rpc"
 	"go.uber.org/zap"
 	"io"
 	"os"

internal/config/rule.go

@@ -4,10 +4,78 @@ import (
 	"fmt"
 	"github.com/icinga/icinga-notifications/internal/rule"
 	"slices"
+	"time"
 )
 
+// SourceRuleVersion for SourceRulesInfo, consisting of two numbers, one static and one incrementable.
+type SourceRuleVersion struct {
+	Major int64
+	Minor int64
+}
+
+// NewSourceRuleVersion creates a new source version based on the current timestamp and a zero counter.
+func NewSourceRuleVersion() SourceRuleVersion {
+	return SourceRuleVersion{
+		Major: time.Now().UnixMilli(),
+		Minor: 0,
+	}
+}
+
+// Increment the version counter.
+func (sourceVersion *SourceRuleVersion) Increment() {
+	sourceVersion.Minor++
+}
+
+// String implements fmt.Stringer and returns a pretty-printable representation.
+func (sourceVersion *SourceRuleVersion) String() string {
+	return fmt.Sprintf("%x-%x", sourceVersion.Major, sourceVersion.Minor)
+}
+
+// SourceRulesInfo holds information about the rules associated with a specific source.
+type SourceRulesInfo struct {
+	// Version is the version of the rules for the source.
+	//
+	// Multiple source's versions are independent of another.
+	Version SourceRuleVersion
+
+	// 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
+}
+
 // 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{})
+
+	if r.RulesBySource == nil {
+		r.RulesBySource = make(map[int64]*SourceRulesInfo)
+	}
+
+	addToRulesBySource := func(elem *rule.Rule) {
+		if sourceInfo, ok := r.RulesBySource[elem.SourceID]; ok {
+			sourceInfo.RuleIDs = append(sourceInfo.RuleIDs, elem.ID)
+		} else {
+			r.RulesBySource[elem.SourceID] = &SourceRulesInfo{
+				Version: NewSourceRuleVersion(),
+				RuleIDs: []int64{elem.ID},
+			}
+		}
+
+		updatedSources[elem.SourceID] = struct{}{}
+	}
+
+	delFromRulesBySource := func(elem *rule.Rule) {
+		if sourceInfo, ok := r.RulesBySource[elem.SourceID]; ok {
+			sourceInfo.RuleIDs = slices.DeleteFunc(sourceInfo.RuleIDs, func(id int64) bool {
+				return id == elem.ID
+			})
+		}
+
+		updatedSources[elem.SourceID] = struct{}{}
+	}
+
 	incrementalApplyPending(
 		r,
 		&r.Rules, &r.configChange.Rules,
@@ -21,6 +89,9 @@ func (r *RuntimeConfig) applyPendingRules() {
 			}
 
 			newElement.Escalations = make(map[int64]*rule.Escalation)
+
+			addToRulesBySource(newElement)
+
 			return nil
 		},
 		func(curElement, update *rule.Rule) error {
@@ -38,13 +109,35 @@ func (r *RuntimeConfig) applyPendingRules() {
 				curElement.TimePeriod = nil
 			}
 
-			// ObjectFilter{,Expr} are being initialized by config.IncrementalConfigurableInitAndValidatable.
-			curElement.ObjectFilter = update.ObjectFilter
+			if curElement.SourceID != update.SourceID {
+				delFromRulesBySource(curElement)
+				curElement.SourceID = update.SourceID
+				addToRulesBySource(curElement)
+			}
+
+			// ObjectFilterExpr is being initialized by config.IncrementalConfigurableInitAndValidatable.
 			curElement.ObjectFilterExpr = update.ObjectFilterExpr
 
+			updatedSources[curElement.SourceID] = struct{}{}
+
 			return nil
 		},
-		nil)
+		func(delElement *rule.Rule) error {
+			delFromRulesBySource(delElement)
+
+			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 sourceInfo, ok := r.RulesBySource[sourceID]; ok {
+			sourceInfo.Version.Increment()
+		}
+	}
 
 	incrementalApplyPending(
 		r,