incident: custom superfluous state change error

By introducing ErrSuperfluousStateChange to signal superfluous state
changes and returning a wrapped error, those messages can now be
suppressed (logged with the debug level) for Event Stream processing.

internal/eventstream/client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"errors"
 	"fmt"
 	"github.com/icinga/icinga-notifications/internal/config"
 	"github.com/icinga/icinga-notifications/internal/daemon"
@@ -76,8 +77,6 @@ func NewClientsFromConfig(
 
 	for _, icinga2Api := range conf.Icinga2Apis {
 		logger := logs.GetChildLogger(fmt.Sprintf("eventstream-%d", icinga2Api.NotificationsEventSourceId))
-		callbackLogger := logs.GetChildLogger(fmt.Sprintf("eventstream-callback-%d", icinga2Api.NotificationsEventSourceId))
-
 		client := &Client{
 			ApiHost:          icinga2Api.Host,
 			ApiBasicAuthUser: icinga2Api.AuthUser,
@@ -104,9 +103,16 @@ func NewClientsFromConfig(
 			IcingaWebRoot:                    conf.Icingaweb2URL,
 
 			CallbackFn: func(ev *event.Event) {
+				l := logger.With(zap.Stringer("event", ev))
+
 				err := incident.ProcessEvent(ctx, db, logs, runtimeConfig, ev)
-				if err != nil {
-					callbackLogger.Warnw("Cannot process event", zap.Error(err))
+				switch {
+				case errors.Is(err, incident.ErrSuperfluousStateChange):
+					l.Debugw("Stopped processing event with superfluous state change", zap.Error(err))
+				case err != nil:
+					l.Errorw("Cannot process event", zap.Error(err))
+				default:
+					l.Debug("Successfully processed event over callback")
 				}
 			},
 			Ctx:    ctx,

internal/incident/incident.go

@@ -249,10 +249,8 @@ func (i *Incident) processSeverityChangedEvent(ctx context.Context, tx *sqlx.Tx,
 	oldSeverity := i.Severity
 	newSeverity := ev.Severity
 	if oldSeverity == newSeverity {
-		msg := fmt.Sprintf("Ignoring superfluous %q state event from source %d", ev.Severity.String(), ev.SourceId)
-		i.logger.Warnln(msg)
-
-		return causedByHistoryId, errors.New(msg)
+		err := fmt.Errorf("%w: %s state event from source %d", ErrSuperfluousStateChange, ev.Severity.String(), ev.SourceId)
+		return causedByHistoryId, err
 	}
 
 	i.logger.Infof("Incident severity changed from %s to %s", oldSeverity.String(), newSeverity.String())

internal/incident/incidents.go

@@ -16,6 +16,9 @@ import (
 	"time"
 )
 
+// ErrSuperfluousStateChange indicates a superfluous state change being ignored and stopping further processing.
+var ErrSuperfluousStateChange = errors.New("ignoring superfluous state change")
+
 var (
 	currentIncidents   = make(map[*object.Object]*Incident)
 	currentIncidentsMu sync.Mutex
@@ -138,6 +141,8 @@ func GetCurrentIncidents() map[int64]*Incident {
 //
 // This function first gets this Event's object.Object and its incident.Incident. Then, after performing some safety
 // checks, it calls the Incident.ProcessEvent method.
+//
+// The returned error might be wrapped around ErrSuperfluousStateChange.
 func ProcessEvent(
 	ctx context.Context,
 	db *icingadb.DB,
@@ -170,7 +175,7 @@ func ProcessEvent(
 		case ev.Severity != event.SeverityOK:
 			panic(fmt.Sprintf("cannot process event %v with a non-OK state %v without a known incident", ev, ev.Severity))
 		default:
-			return fmt.Errorf("ignoring superfluous OK state event from source %d", ev.SourceId)
+			return fmt.Errorf("%w: ok state event from source %d", ErrSuperfluousStateChange, ev.SourceId)
 		}
 	}