Preparation for sending non-state notifications

cmd/icinga-notifications/main.go

@@ -54,8 +54,8 @@ func main() {
 
 	go runtimeConfig.PeriodicUpdates(ctx, 1*time.Second)
 
-	err = incident.LoadOpenIncidents(ctx, db, logs.GetChildLogger("incident"), runtimeConfig)
-	if err != nil {
+	logger.Info("Loading all active incidents from database")
+	if err = incident.LoadOpenIncidents(ctx, db, runtimeConfig); err != nil {
 		logger.Fatalf("Cannot load incidents from database: %+v", err)
 	}
 
@@ -68,7 +68,7 @@ func main() {
 	// When Icinga Notifications is started by systemd, we've to notify systemd that we're ready.
 	_ = sdnotify.Ready()
 
-	if err := listener.NewListener(db, runtimeConfig, logs).Run(ctx); err != nil {
+	if err := listener.NewListener(runtimeConfig).Run(ctx); err != nil {
 		logger.Errorf("Listener has finished with an error: %+v", err)
 	} else {
 		logger.Info("Listener has finished")

go.mod

@@ -9,7 +9,7 @@ require (
 	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.20250909100113-20db23663e45
+	github.com/icinga/icinga-go-library v0.7.3-0.20250910094214-649cb701c5b1
 	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

@@ -37,6 +37,8 @@ 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.20250909100113-20db23663e45 h1:Wz6ttTYgYB7y8FH7snBSnnllLuzhE0QSp6m3P9b/QfM=
 github.com/icinga/icinga-go-library v0.7.3-0.20250909100113-20db23663e45/go.mod h1:uCENf5EVhNVvXTvhB+jXiwRB2NdLlz8cymseOM4qmI0=
+github.com/icinga/icinga-go-library v0.7.3-0.20250910094214-649cb701c5b1 h1:P8WtinCY5yneqr9cSAgyOhUDkb5XxQLQ3kBBMqZmvZg=
+github.com/icinga/icinga-go-library v0.7.3-0.20250910094214-649cb701c5b1/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/channel/channel.go

@@ -3,15 +3,12 @@ package channel
 import (
 	"context"
 	"errors"
-	"fmt"
+
+	"github.com/icinga/icinga-go-library/notifications/event"
 	"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"
 )
 
 type Channel struct {
@@ -159,41 +156,33 @@ func (c *Channel) Restart(logger *zap.SugaredLogger) {
 	c.restartCh <- newConfig{c.Type, c.Config}
 }
 
-// Notify prepares and sends the notification request, returns a non-error on fails, nil on success
-func (c *Channel) Notify(contact *recipient.Contact, i contracts.Incident, ev *event.Event, icingaweb2Url string) error {
-	p := c.getPlugin()
-	if p == nil {
-		return errors.New("plugin could not be started")
+// Notify sends the provided notification request using the channel's plugin.
+//
+// First, it applies some basic validation to ensure that the request is well-formed, for example
+// that it contains an event and a contact. If the request is invalid, an error is returned. Next,
+// it retrieves the channel's plugin using getPlugin, and if the plugin could not be started or is
+// otherwise unavailable, an error is returned.
+//
+// Finally, it calls the plugin's [Plugin.SendNotification] method with the provided request and
+// returns any error that occurs during this process.
+func (c *Channel) Notify(req *plugin.NotificationRequest) error {
+	if req.Event == nil {
+		return errors.New("invalid notification request: Event is nil")
 	}
-
-	contactStruct := &plugin.Contact{FullName: contact.FullName}
-	for _, addr := range contact.Addresses {
-		contactStruct.Addresses = append(contactStruct.Addresses, &plugin.Address{Type: addr.Type, Address: addr.Address})
+	if req.Object == nil {
+		return errors.New("invalid notification request: Object is nil")
+	}
+	if req.Contact == nil {
+		return errors.New("invalid notification request: Contact is nil")
+	}
+	// If this is a state event, an incident must be provided as well, otherwise something is really wrong.
+	if req.Incident == nil && req.Event.Type == event.TypeState {
+		return errors.New("invalid notification request: cannot send state notification without an incident")
 	}
 
-	baseUrl, _ := url.Parse(icingaweb2Url)
-	incidentUrl := baseUrl.JoinPath("/notifications/incident")
-	incidentUrl.RawQuery = fmt.Sprintf("id=%d", i.ID())
-	object := i.IncidentObject()
-
-	req := &plugin.NotificationRequest{
-		Contact: contactStruct,
-		Object: &plugin.Object{
-			Name: object.DisplayName(),
-			Url:  ev.URL,
-			Tags: object.Tags,
-		},
-		Incident: &plugin.Incident{
-			Id:       i.ID(),
-			Url:      incidentUrl.String(),
-			Severity: i.IncidentSeverity(),
-		},
-		Event: &plugin.Event{
-			Time:     ev.Time,
-			Type:     ev.Type,
-			Username: ev.Username,
-			Message:  ev.Message,
-		},
+	p := c.getPlugin()
+	if p == nil {
+		return errors.New("plugin could not be started")
 	}
 
 	return p.SendNotification(req)

internal/config/rule.go

@@ -2,8 +2,10 @@ package config
 
 import (
 	"fmt"
-	"github.com/icinga/icinga-notifications/internal/rule"
 	"slices"
+	"time"
+
+	"github.com/icinga/icinga-notifications/internal/rule"
 )
 
 // SourceRulesInfo holds information about the rules associated with a specific source.
@@ -176,3 +178,106 @@ func (r *RuntimeConfig) applyPendingRules() {
 			return nil
 		})
 }
+
+// EvalOptions specifies optional callbacks that are executed upon certain filter evaluation events.
+//
+// The EvalOptions type is used to configure the behaviour of the evaluation process when evaluating
+// filter expressions against a set of [rule.Escalation] entries. It allows you to hook into specific
+// events during the evaluation process and perform custom actions based on your requirements.
+type EvalOptions struct {
+	// OnPreEvaluate can be used to perform some actions before evaluating the filter for the current entry.
+	//
+	// This callback receives the current [rule.Escalation] entry as an argument, which is about to be
+	// evaluated. If this callback returns "false", the filter evaluation for the current entry is skipped,
+	// and the evaluation continues with the next one. If it returns "true" or is nil, the filter evaluation
+	// proceeds as normal.
+	//
+	// Note that if you skip the evaluation of an entry using this callback, the OnFilterMatch callback
+	// will not be triggered for that entry, even if its filter would have matched on the filterable object.
+	OnPreEvaluate func(*rule.Escalation) bool
+
+	// OnError is called when an error occurs during the filter evaluation.
+	//
+	// This callback receives the current [rule.Escalation] entry and the error that occurred as arguments.
+	// By default, the evaluation continues even if some entries fail, but you can override this behaviour
+	// by returning "false" in your handler, which aborts the evaluation prematurely. If you return "true"
+	// or if this callback is nil, the evaluation continues with the remaining entries.
+	//
+	// Note that if you choose to abort the evaluation by returning "false", the OnAllConfigEvaluated callback
+	// will not be triggered, as the evaluation did not complete successfully.
+	OnError func(*rule.Escalation, error) bool
+
+	// OnFilterMatch is called when the filter for an entry matches successfully.
+	//
+	// This callback receives the current [rule.Escalation] entry as an argument. If this callback returns
+	// an error, the evaluation is aborted prematurely, and the error is returned. Otherwise, the evaluation
+	// continues with the remaining entries.
+	//
+	// Note that if you return an error from this callback, the OnAllConfigEvaluated callback will not be triggered,
+	// as the evaluation did not complete successfully.
+	OnFilterMatch func(*rule.Escalation) error
+
+	// OnAllConfigEvaluated is called after all configured entries have been evaluated.
+	//
+	// This callback receives a value of type [time.Duration] derived from the evaluation process as an argument.
+	// This callback is guaranteed to be called if none of the individual evaluation callbacks return prematurely
+	// with an error. If any of the callbacks return prematurely, this callback will not be triggered.
+	//
+	// The [time.Duration] argument can be used to indicate a duration after which a re-evaluation might be necessary,
+	// based on the evaluation results. This is optional and can be ignored if not needed.
+	OnAllConfigEvaluated func(time.Duration)
+}
+
+// RuleEntries is a map of rule.Escalation entries, keyed by their ID.
+//
+// This type is used to store the results of evaluating rule.Escalation entries against a filterable object.
+// It allows for efficient lookups and ensures that each entry is unique based on its ID.
+type RuleEntries map[int64]*rule.Escalation
+
+// Evaluate evaluates the rule.Escalation entries associated with the given rule IDs against the provided filterable object.
+//
+// The Evaluate method iterates over the specified rule IDs, retrieves the corresponding rule.Escalation entries
+// from the RuntimeConfig, and evaluates their filters against the provided filterable object. It uses the
+// provided EvalOptions to handle various events during the evaluation process.
+//
+// Returns an error if any of the evaluation callbacks return an error or if there are issues during the evaluation.
+func (re RuleEntries) Evaluate(r *RuntimeConfig, filterable *rule.EscalationFilter, rules map[int64]struct{}, opts EvalOptions) error {
+	retryAfter := rule.RetryNever
+
+	for ruleID := range rules {
+		ru := r.Rules[ruleID]
+		if ru == nil {
+			// It would be appropriate to have a debug log here, but unfortunately we don't have access to a logger.
+			continue
+		}
+
+		for _, entry := range ru.Escalations {
+			if opts.OnPreEvaluate != nil && !opts.OnPreEvaluate(entry) {
+				continue
+			}
+
+			if matched, err := entry.Eval(filterable); err != nil {
+				if opts.OnError != nil && !opts.OnError(entry, err) {
+					return err
+				}
+			} else if !matched {
+				incidentAgeFilter := filterable.ReevaluateAfter(entry.Condition)
+				retryAfter = min(retryAfter, incidentAgeFilter)
+			} else {
+				if opts.OnFilterMatch != nil {
+					if err := opts.OnFilterMatch(entry); err != nil {
+						return err
+					}
+				}
+
+				re[entry.ID] = entry
+			}
+		}
+	}
+
+	if opts.OnAllConfigEvaluated != nil {
+		opts.OnAllConfigEvaluated(retryAfter)
+	}
+
+	return nil
+}

internal/config/rule_test.go

@@ -0,0 +1,150 @@
+package config
+
+import (
+	"fmt"
+	"maps"
+	"testing"
+	"time"
+
+	"github.com/icinga/icinga-go-library/notifications/event"
+	"github.com/icinga/icinga-notifications/internal/filter"
+	"github.com/icinga/icinga-notifications/internal/rule"
+	"github.com/stretchr/testify/require"
+)
+
+const defaultDivisor = 3 // Every third rule gets a valid escalation condition.
+
+func TestRuleEntries(t *testing.T) {
+	t.Parallel()
+
+	runtimeConfigTest := new(RuntimeConfig)
+	runtimeConfigTest.Rules = make(map[int64]*rule.Rule)
+	for i := 1; i <= 50; i++ {
+		runtimeConfigTest.Rules[int64(i)] = makeRule(t, i)
+	}
+
+	t.Run("Evaluate", func(t *testing.T) {
+		t.Parallel()
+
+		runtime := new(RuntimeConfig)
+		runtime.Rules = maps.Clone(runtimeConfigTest.Rules)
+
+		e := make(RuleEntries)
+
+		expectedLen := 0
+		filterContext := &rule.EscalationFilter{IncidentSeverity: event.SeverityEmerg}
+		assertEntries := func(rules map[int64]struct{}, expectedLen *int, expectError bool, opts EvalOptions) {
+			if expectError {
+				require.Error(t, e.Evaluate(runtime, filterContext, rules, opts))
+			} else {
+				require.NoError(t, e.Evaluate(runtime, filterContext, rules, opts))
+			}
+			require.Len(t, e, *expectedLen)
+			clear(e) // Clear the entries for the next run.
+		}
+		expectedLen = len(runtime.Rules)/defaultDivisor - 5 // 15/3 => (5) valid entries are going to be deleted below.
+
+		// Drop some random rules from the runtime config to simulate a runtime config deletion!
+		maps.DeleteFunc(runtime.Rules, func(ruleID int64, _ *rule.Rule) bool { return ruleID > 35 && ruleID%defaultDivisor == 0 })
+
+		opts := EvalOptions{
+			OnPreEvaluate: func(re *rule.Escalation) bool {
+				if re.RuleID > 35 && re.RuleID%defaultDivisor == 0 { // Those rules are deleted from our runtime config.
+					require.Failf(t, "OnPreEvaluate() shouldn't have been called", "rule %d was deleted from runtime config", re.RuleID)
+				}
+				require.Nilf(t, e[re.ID], "Evaluate() shouldn't evaluate entry %d twice", re.ID)
+				return true
+			},
+			OnError: func(re *rule.Escalation, err error) bool {
+				require.EqualError(t, err, `unknown severity "evaluable"`)
+				return true
+			},
+			OnFilterMatch: func(re *rule.Escalation) error {
+				require.Nilf(t, e[re.ID], "OnPreEvaluate() shouldn't evaluate %d twice", re.ID)
+				return nil
+			},
+		}
+		rules := ruleIDs(runtime)
+		assertEntries(rules, &expectedLen, false, opts)
+
+		lenBeforeError := new(int)
+		opts.OnError = func(re *rule.Escalation, err error) bool {
+			if *lenBeforeError != 0 {
+				require.Fail(t, "OnError() shouldn't have been called again")
+			}
+			require.EqualError(t, err, `unknown severity "evaluable"`)
+
+			*lenBeforeError = len(e)
+			return false // This should let the evaluation fail completely!
+		}
+		assertEntries(rules, lenBeforeError, true, opts)
+
+		*lenBeforeError = 0
+		opts.OnError = nil
+		opts.OnFilterMatch = func(re *rule.Escalation) error {
+			if *lenBeforeError != 0 {
+				require.Fail(t, "OnFilterMatch() shouldn't have been called again")
+			}
+
+			*lenBeforeError = len(e)
+			return fmt.Errorf("OnFilterMatch() failed badly") // This should let the evaluation fail completely!
+		}
+		assertEntries(rules, lenBeforeError, true, opts)
+
+		expectedLen = 0
+		filterContext.IncidentSeverity = 1 // OK
+		filterContext.IncidentAge = 5 * time.Minute
+
+		opts.OnFilterMatch = nil
+		opts.OnPreEvaluate = func(re *rule.Escalation) bool { return re.RuleID < 5 }
+		opts.OnAllConfigEvaluated = func(result time.Duration) {
+			// The filter string of the escalation condition is incident_age>=10m and the actual incident age is 5m.
+			require.Equal(t, 5*time.Minute, result)
+		}
+		assertEntries(rules, &expectedLen, false, opts)
+	})
+}
+
+// makeRule creates a rule with some escalation entries.
+//
+// Every rule gets one invalid escalation condition that always fails to evaluate.
+// Additionally, every third (defaultDivisor) rule gets a valid escalation condition that matches
+// on `incident_severity>warning||incident_age>=10m` to simulate some real-world conditions.
+func makeRule(t *testing.T, i int) *rule.Rule {
+	r := new(rule.Rule)
+	r.ID = int64(i)
+	r.Name = fmt.Sprintf("rule-%d", i)
+	r.Escalations = make(map[int64]*rule.Escalation)
+
+	invalidSeverity, err := filter.Parse("incident_severity=evaluable")
+	require.NoError(t, err, "parsing incident_severity=evaluable shouldn't fail")
+
+	redundant := new(rule.Escalation)
+	redundant.ID = r.ID * 150 // It must be large enough to avoid colliding with others!
+	redundant.RuleID = r.ID
+	redundant.Condition = invalidSeverity
+
+	r.Escalations[redundant.ID] = redundant
+	if i%defaultDivisor == 0 {
+		escalationCond, err := filter.Parse("incident_severity>warning||incident_age>=10m")
+		require.NoError(t, err, "parsing incident_severity>warning||incident_age>=10m shouldn't fail")
+
+		entry := new(rule.Escalation)
+		entry.ID = r.ID * 2
+		entry.RuleID = r.ID
+		entry.Condition = escalationCond
+
+		r.Escalations[entry.ID] = entry
+	}
+
+	return r
+}
+
+// ruleIDs extracts the rule IDs from the given RuntimeConfig and returns them as a map[int64]struct{}.
+func ruleIDs(r *RuntimeConfig) map[int64]struct{} {
+	ids := make(map[int64]struct{}, len(r.Rules))
+	for id := range r.Rules {
+		ids[id] = struct{}{}
+	}
+	return ids
+}