diff --git a/go.mod b/go.mod index 04cbd663..80c4e86f 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,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.9.1-0.20260623130323-faef23cbd4c9 + github.com/icinga/icinga-go-library v0.9.1-0.20260630080726-15d47aff49f8 github.com/jhillyerd/enmime v1.3.0 github.com/jmoiron/sqlx v1.4.0 github.com/okzk/sdnotify v0.0.0-20180710141335-d9becc38acbd diff --git a/go.sum b/go.sum index 0f2349ae..d0a4e17e 100644 --- a/go.sum +++ b/go.sum @@ -36,8 +36,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.9.1-0.20260623130323-faef23cbd4c9 h1:HyeACIYSpROzA9bUghELjzsCwnHzM48M6oD2zgKMlfA= -github.com/icinga/icinga-go-library v0.9.1-0.20260623130323-faef23cbd4c9/go.mod h1:XFsuZEn2hrE3YXQ3NHqnI2UkjdZgJJWb/7D/+VH6MsE= +github.com/icinga/icinga-go-library v0.9.1-0.20260630080726-15d47aff49f8 h1:YZ+N5qKN9tB6JPUno3oVcVE9TAayAIOZAG8hmVkgYBQ= +github.com/icinga/icinga-go-library v0.9.1-0.20260630080726-15d47aff49f8/go.mod h1:XFsuZEn2hrE3YXQ3NHqnI2UkjdZgJJWb/7D/+VH6MsE= 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= diff --git a/internal/channel/channel.go b/internal/channel/channel.go index 4f00ce92..e3676327 100644 --- a/internal/channel/channel.go +++ b/internal/channel/channel.go @@ -188,10 +188,8 @@ func (c *Channel) Notify(contact *recipient.Contact, i contracts.Incident, ev *e Severity: i.IncidentSeverity(), }, Event: &plugin.Event{ - Time: ev.Time, - Type: ev.Type, - Username: ev.Username, - Message: ev.Message, + Time: ev.Time, + Message: ev.Message, }, } diff --git a/internal/event/event.go b/internal/event/event.go index 53062716..40b9e982 100644 --- a/internal/event/event.go +++ b/internal/event/event.go @@ -1,7 +1,6 @@ package event import ( - "errors" "fmt" "net/url" "regexp" @@ -10,20 +9,12 @@ import ( "time" baseEv "github.com/icinga/icinga-go-library/notifications/event" - "github.com/icinga/icinga-go-library/types" "github.com/icinga/icinga-notifications/internal/pool" "github.com/icinga/icinga-notifications/internal/utils" "github.com/theory/jsonpath" "go.uber.org/zap/zapcore" ) -// ErrSuperfluousStateChange indicates a superfluous state change being ignored and stopping further processing. -var ErrSuperfluousStateChange = errors.New("ignoring superfluous state change") - -// ErrSuperfluousMuteUnmuteEvent indicates that a superfluous mute or unmute event is being ignored and is -// triggered when trying to mute/unmute an already muted/unmuted incident. -var ErrSuperfluousMuteUnmuteEvent = errors.New("ignoring superfluous (un)mute event") - // Event received of a specified Type for internal processing. // // This is a representation of an event received from an external source with additional metadata with sole @@ -68,53 +59,42 @@ func (e *Event) CompleteURL(icingaWebBaseUrl *url.URL) { // Validate validates the current event state. // Returns an error if it detects a misconfigured field. func (e *Event) Validate() error { - if len(e.Tags) == 0 { - return fmt.Errorf("invalid event: tags must not be empty") + if err := e.Event.Validate(); err != nil { + return err } - for tag := range e.Tags { - if len(tag) > 255 { - return fmt.Errorf("invalid event: tag %q is too long, at most 255 chars allowed, %d given", tag, len(tag)) - } + if e.Time.IsZero() { + return fmt.Errorf("invalid event: time must not be empty") } if e.SourceId == 0 { return fmt.Errorf("invalid event: source ID must not be empty") } - if e.Severity != baseEv.SeverityNone && e.Type != baseEv.TypeState { - return fmt.Errorf("invalid event: if 'severity' is set, 'type' must be set to %q", baseEv.TypeState) - } - if e.Type == baseEv.TypeMute && (!e.Mute.Valid || !e.Mute.Bool) { - return fmt.Errorf("invalid event: 'mute' must be true if 'type' is set to %q", baseEv.TypeMute) - } - if e.Type == baseEv.TypeUnmute && (!e.Mute.Valid || e.Mute.Bool) { - return fmt.Errorf("invalid event: 'mute' must be false if 'type' is set to %q", baseEv.TypeUnmute) - } - if e.Mute.Valid && e.Mute.Bool && e.MuteReason == "" { - return fmt.Errorf("invalid event: 'mute_reason' must not be empty if 'mute' is set") - } - - if e.Type == baseEv.TypeUnknown { - return errors.New("invalid event: missing type") - } - return nil } -// SetMute alters the event mute and mute reason. -func (e *Event) SetMute(muted bool, reason string) { - e.Mute = types.Bool{Valid: true, Bool: muted} - e.MuteReason = reason -} - // MarshalLogObject implements the [zapcore.ObjectMarshaler] interface to allow logging the event as a structured object. func (e *Event) MarshalLogObject(encoder zapcore.ObjectEncoder) error { encoder.AddString("name", e.Name) encoder.AddTime("time", e.Time) encoder.AddInt64("source_id", e.SourceId) - encoder.AddString("type", e.Type.String()) encoder.AddString("severity", e.Severity.String()) + + if e.Muted.Valid { + encoder.AddBool("muted", e.Muted.Bool) + encoder.AddString("muted_reason", e.MutedReason) + } + if e.OpenOrEscalate() { + encoder.AddBool("incident", true) + } + if e.CloseIncident() { + encoder.AddBool("close", true) + } + if e.NotifyRecipients() { + encoder.AddBool("notify", true) + } + return encoder.AddObject("tags", zapcore.ObjectMarshalerFunc(func(objectEncoder zapcore.ObjectEncoder) error { for key, value := range e.Tags { objectEncoder.AddString(key, value) @@ -130,6 +110,10 @@ func (e *Event) MarshalLogObject(encoder zapcore.ObjectEncoder) error { // field and are not part of the CompleteRelations field. For filter columns that do have matching // nodes, it caches the evaluated nodes for potential later use during rules evaluation. func (e *Event) ExtractMissingRelations(filterColumns ...[]string) []string { + if len(filterColumns) == 0 { + return nil + } + if e.evaluatedRelations == nil { e.evaluatedRelations = make(map[string]jsonpath.NodeList) } diff --git a/internal/incident/incident.go b/internal/incident/incident.go index 0ec7b15f..3c1c935e 100644 --- a/internal/incident/incident.go +++ b/internal/incident/incident.go @@ -2,7 +2,11 @@ package incident import ( "context" + "errors" "fmt" + "sync" + "time" + "github.com/icinga/icinga-go-library/database" baseEv "github.com/icinga/icinga-go-library/notifications/event" "github.com/icinga/icinga-go-library/types" @@ -15,8 +19,6 @@ import ( "github.com/icinga/icinga-notifications/internal/rule" "github.com/jmoiron/sqlx" "go.uber.org/zap" - "sync" - "time" ) type ruleID = int64 @@ -30,6 +32,7 @@ type Incident struct { Severity baseEv.Severity `db:"severity"` // MuteReason indicates whether this incident is currently muted; its non-null value contains the mute reason. MuteReason types.String `db:"mute_reason"` + Message types.String `db:"message"` Object *object.Object `db:"-"` @@ -127,17 +130,6 @@ func (i *Incident) ProcessEvent(ctx context.Context, ev *event.Event) error { i.runtimeConfig.RLock() defer i.runtimeConfig.RUnlock() - // These event types are not like the others used to mute an object/incident, such as DowntimeStart, which - // uniquely identify themselves why an incident is being muted, but are rather super generic types, and as - // such, we are ignoring superfluous ones that don't have any effect on that incident. - if i.IsMuted() && ev.Type == baseEv.TypeMute { - i.logger.Debugw("Ignoring superfluous mute event", zap.Object("event", ev)) - return event.ErrSuperfluousMuteUnmuteEvent - } else if !i.IsMuted() && ev.Type == baseEv.TypeUnmute { - i.logger.Debugw("Ignoring superfluous unmute event", zap.Object("event", ev)) - return event.ErrSuperfluousMuteUnmuteEvent - } - tx, err := i.db.BeginTxx(ctx, nil) if err != nil { i.logger.Errorw("Cannot start a db transaction", zap.Error(err)) @@ -150,15 +142,22 @@ func (i *Incident) ProcessEvent(ctx context.Context, ev *event.Event) error { return fmt.Errorf("cannot sync event object: %w", err) } + triggerNotifications := true isNew := i.StartedAt.Time().IsZero() if isNew { - err = i.processIncidentOpenedEvent(ctx, tx, ev) - if err != nil { + if err := i.processIncidentOpenedEvent(ctx, tx, ev); err != nil { return err } i.logger = i.logger.With(zap.String("incident", i.String())) } else { + if sevChanged, err := i.processSeverityChangedEvent(ctx, tx, ev); err != nil { + return err + } else { + // In case the severity didn't change, we need to check whether we can trigger notifications nonetheless. + triggerNotifications = sevChanged || ev.NotifyRecipients() || (ev.Muted.Valid && ev.IsMuted() != i.IsMuted()) + } + // For all existing incidents, we have to reload the recipients from the database to ensure that we have the // most up-to-date recipient list, since the recipients or recipients roles might have changed since users // are allowed to subscribe or manage incidents from within Icinga Notifications Web. @@ -167,13 +166,7 @@ func (i *Incident) ProcessEvent(ctx context.Context, ev *event.Event) error { } } - if ev.Type == baseEv.TypeState { - if !isNew { - if err := i.processSeverityChangedEvent(ctx, tx, ev); err != nil { - return err - } - } - + if ev.OpenOrEscalate() { if err := i.applyMatchingRules(ctx, tx, ev); err != nil { return err } @@ -184,9 +177,18 @@ func (i *Incident) ProcessEvent(ctx context.Context, ev *event.Event) error { return err } - if err := i.triggerEscalations(ctx, tx, ev, escalations); err != nil { + if err := i.triggerEscalations(ctx, tx, escalations); err != nil { return err } + + // If we have managed to trigger any new escalations, we must trigger notifications as well, + // even if the event itself doesn't request it. + triggerNotifications = triggerNotifications || len(escalations) > 0 + + if !isNew { + // Even if the severity didn't change, we want to update the message nonetheless. + i.Message = types.MakeString(ev.Message, types.TransformEmptyStringToNull) + } } // The unmute history entry, on the other hand, must be inserted first, so that the notifications generated @@ -198,9 +200,11 @@ func (i *Incident) ProcessEvent(ctx context.Context, ev *event.Event) error { } var notifications []*NotificationEntry - notifications, err = i.generateNotifications(ctx, tx, ev, i.getRecipientsChannel(ev.Time)) - if err != nil { - return err + if triggerNotifications { + notifications, err = i.generateNotifications(ctx, tx, ev, i.getRecipientsChannel(ev.Time)) + if err != nil { + return err + } } // So that the incident muted history appears logically after the just generated notifications, we must insert @@ -210,6 +214,17 @@ func (i *Incident) ProcessEvent(ctx context.Context, ev *event.Event) error { return err } + if ev.CloseIncident() { + if err := i.Close(ctx, tx, false); err != nil { + return err + } + } + + if err := i.Sync(ctx, tx); err != nil { + i.logger.Errorw("Failed to update incident", zap.Error(err)) + return err + } + if err = tx.Commit(); err != nil { i.logger.Errorw("Cannot commit db transaction", zap.Error(err)) return err @@ -250,7 +265,7 @@ func (i *Incident) RetriggerEscalations(ev *event.Event) { var notifications []*NotificationEntry ctx := context.Background() err = i.db.ExecTx(ctx, nil, func(ctx context.Context, tx *sqlx.Tx) error { - if err = i.triggerEscalations(ctx, tx, ev, escalations); err != nil { + if err = i.triggerEscalations(ctx, tx, escalations); err != nil { return err } @@ -274,37 +289,18 @@ func (i *Incident) RetriggerEscalations(ev *event.Event) { } } -func (i *Incident) processSeverityChangedEvent(ctx context.Context, tx *sqlx.Tx, ev *event.Event) error { - oldSeverity := i.Severity - newSeverity := ev.Severity - if oldSeverity == newSeverity { - err := fmt.Errorf("%w: %s state event from source %d", event.ErrSuperfluousStateChange, ev.Severity.String(), ev.SourceId) - return err - } - - i.logger.Infof("Incident severity changed from %s to %s", oldSeverity.String(), newSeverity.String()) - - hr := &HistoryRow{ - IncidentID: i.Id, - Time: types.UnixMilli(time.Now()), - Type: IncidentSeverityChanged, - NewSeverity: newSeverity, - OldSeverity: oldSeverity, - Message: types.MakeString(ev.Message, types.TransformEmptyStringToNull), - } - - if err := hr.Sync(ctx, i.db, tx); err != nil { - i.logger.Errorw("Failed to insert incident severity changed history", zap.Error(err)) - return err - } - - if newSeverity == baseEv.SeverityOK { +// Close closes the current incident if not already recovered. +// +// If the incident is already recovered, this is a no-op. Returns an error if fails +// to persist the recovery time or insert the generated history to the database. +func (i *Incident) Close(ctx context.Context, tx *sqlx.Tx, persist bool) error { + if i.RecoveredAt.Time().IsZero() { i.RecoveredAt = types.UnixMilli(time.Now()) - i.logger.Info("All sources recovered, closing incident") + i.logger.Info("Received request to close the incident, marking it as recovered") RemoveCurrent(i.Object) - hr = &HistoryRow{ + hr := &HistoryRow{ IncidentID: i.Id, Time: i.RecoveredAt, Type: Closed, @@ -318,20 +314,56 @@ func (i *Incident) processSeverityChangedEvent(ctx context.Context, tx *sqlx.Tx, if i.timer != nil { i.timer.Stop() } + if persist { + return i.Sync(ctx, tx) + } } + return nil +} - i.Severity = newSeverity - if err := i.Sync(ctx, tx); err != nil { - i.logger.Errorw("Failed to update incident severity", zap.Error(err)) - return err +// ErrSeverityChangeWithoutIncidentFlag is returned when an event tries to change the severity of an incident +// but does not set the 'incident' flag. +var ErrSeverityChangeWithoutIncidentFlag = errors.New("cannot change severity of an incident with an event that doesn't set the 'incident' flag") + +// processSeverityChangedEvent processes the given event as a severity changed event, if the severity has actually changed. +// +// If the severity has not changed, this is a no-op, otherwise returns true. +// +// Returns an error if fails to persist the generated history or [ErrSeverityChangeWithoutIncidentFlag] +// if the event does not set the 'incident' flag. +func (i *Incident) processSeverityChangedEvent(ctx context.Context, tx *sqlx.Tx, ev *event.Event) (bool, error) { + sevChanged := ev.Severity != baseEv.SeverityNone && i.Severity != ev.Severity + if sevChanged { + if !ev.OpenOrEscalate() { + i.logger.Errorw("Cannot change incident severity with an event that doesn't set the 'incident' flag") + return false, ErrSeverityChangeWithoutIncidentFlag + } + i.logger.Infof("Incident severity changed from %s to %s", i.Severity.String(), ev.Severity.String()) + + hr := &HistoryRow{ + IncidentID: i.Id, + Time: types.UnixMilli(time.Now()), + Type: IncidentSeverityChanged, + NewSeverity: ev.Severity, + OldSeverity: i.Severity, + Message: types.MakeString(ev.Message, types.TransformEmptyStringToNull), + } + + if err := hr.Sync(ctx, i.db, tx); err != nil { + i.logger.Errorw("Failed to insert incident severity changed history", zap.Error(err)) + return false, err + } + + i.Severity = ev.Severity } - return nil + return sevChanged, nil } func (i *Incident) processIncidentOpenedEvent(ctx context.Context, tx *sqlx.Tx, ev *event.Event) error { i.StartedAt = types.UnixMilli(ev.Time) i.Severity = ev.Severity + i.Message = types.MakeString(ev.Message, types.TransformEmptyStringToNull) if err := i.Sync(ctx, tx); err != nil { i.logger.Errorw("Cannot insert incident to the database", zap.Error(err)) return err @@ -361,22 +393,19 @@ func (i *Incident) processIncidentOpenedEvent(ctx context.Context, tx *sqlx.Tx, // If the incident is already unmuted, or the event does not unmute, this is a no-op. // Returns an error if fails to persist the cleared mute reason or insert the generated history to the database. func (i *Incident) handleUnmute(ctx context.Context, tx *sqlx.Tx, ev *event.Event) error { - if !i.IsMuted() || !ev.Mute.Valid || ev.Mute.Bool { + if !i.IsMuted() || !ev.Muted.Valid || ev.IsMuted() { return nil } i.logger.Infow("Unmuting incident", zap.String("reason", i.MuteReason.String)) i.MuteReason = types.String{} - if err := i.Sync(ctx, tx); err != nil { - return fmt.Errorf("failed to persist incident unmute state: %w", err) - } hr := &HistoryRow{ IncidentID: i.Id, Time: types.UnixMilli(time.Now()), Type: Unmuted, - Message: types.MakeString(ev.MuteReason, types.TransformEmptyStringToNull), + Message: types.MakeString(ev.MutedReason, types.TransformEmptyStringToNull), } return hr.Sync(ctx, i.db, tx) } @@ -387,17 +416,13 @@ func (i *Incident) handleUnmute(ctx context.Context, tx *sqlx.Tx, ev *event.Even // If the incident is already muted, or the event does not mute, this is a no-op. // Returns an error if fails to persist the mute reason or insert the generated history to the database. func (i *Incident) handleMute(ctx context.Context, tx *sqlx.Tx, ev *event.Event) error { - if i.IsMuted() || !ev.Mute.Valid || !ev.Mute.Bool { + if i.IsMuted() || !ev.IsMuted() { return nil } - i.MuteReason = types.MakeString(ev.MuteReason, types.TransformEmptyStringToNull) + i.MuteReason = types.MakeString(ev.MutedReason, types.TransformEmptyStringToNull) i.logger.Infow("Muting incident", zap.String("reason", i.MuteReason.String)) - if err := i.Sync(ctx, tx); err != nil { - return fmt.Errorf("failed to persist incident mute state: %w", err) - } - hr := &HistoryRow{ IncidentID: i.Id, Time: types.UnixMilli(time.Now()), @@ -520,8 +545,8 @@ func (i *Incident) evaluateEscalations(eventTime time.Time) ([]*rule.Escalation, i.RetriggerEscalations(&event.Event{ Time: nextEvalAt, Event: baseEv.Event{ - Type: baseEv.TypeIncidentAge, - Message: fmt.Sprintf("Incident reached age %v", nextEvalAt.Sub(i.StartedAt.Time())), + Incident: types.MakeBool(true), + Message: fmt.Sprintf("Incident reached age %v", nextEvalAt.Sub(i.StartedAt.Time())), }, }) }) @@ -532,7 +557,7 @@ func (i *Incident) evaluateEscalations(eventTime time.Time) ([]*rule.Escalation, // triggerEscalations triggers the given escalations and generates incident history items for each of them. // Returns an error on database failure. -func (i *Incident) triggerEscalations(ctx context.Context, tx *sqlx.Tx, ev *event.Event, escalations []*rule.Escalation) error { +func (i *Incident) triggerEscalations(ctx context.Context, tx *sqlx.Tx, escalations []*rule.Escalation) error { for _, escalation := range escalations { r := i.runtimeConfig.Rules[escalation.RuleID] if r == nil { @@ -619,16 +644,14 @@ func (i *Incident) notifyContact(contact *recipient.Contact, ev *event.Event, ch return fmt.Errorf("could not find config for channel ID: %d", chID) } - i.logger.Infow(fmt.Sprintf("Notify contact %q via %q of type %q", contact.FullName, ch.Name, ch.Type), - zap.Int64("channel_id", chID), zap.String("event_type", ev.Type.String())) + i.logger.Infof("Notifying contact %q via %q of type %q", contact.FullName, ch.Name, ch.Type) if err := ch.Notify(contact, i, ev, daemon.Config().IcingaWeb2UrlParsed); err != nil { i.logger.Errorw("Failed to send notification via channel plugin", zap.String("type", ch.Type), zap.Error(err)) return err } - i.logger.Infow("Successfully sent a notification via channel plugin", zap.String("type", ch.Type), - zap.String("contact", contact.FullName), zap.String("event_type", ev.Type.String())) + i.logger.Infow("Successfully sent notification", zap.String("type", ch.Type), zap.Stringer("contact", contact)) return nil } diff --git a/internal/incident/incidents.go b/internal/incident/incidents.go index 56ed75ea..c19674bd 100644 --- a/internal/incident/incidents.go +++ b/internal/incident/incidents.go @@ -3,6 +3,9 @@ package incident import ( "context" "fmt" + "sync" + "time" + "github.com/icinga/icinga-go-library/com" "github.com/icinga/icinga-go-library/database" "github.com/icinga/icinga-go-library/logging" @@ -15,8 +18,6 @@ import ( "github.com/pkg/errors" "go.uber.org/zap" "golang.org/x/sync/errgroup" - "sync" - "time" ) var ( @@ -131,8 +132,8 @@ func LoadOpenIncidents(ctx context.Context, db *database.DB, logger *logging.Log i.RetriggerEscalations(&event.Event{ Time: time.Now(), Event: baseEv.Event{ - Type: baseEv.TypeIncidentAge, - Message: fmt.Sprintf("Incident reached age %v (daemon was restarted)", time.Since(i.StartedAt.Time())), + Incident: types.MakeBool(true), + Message: fmt.Sprintf("Incident reached age %v (daemon was restarted)", time.Since(i.StartedAt.Time())), }, }) } @@ -195,12 +196,17 @@ func GetCurrentIncidentsForSource(sourceID int64) []*Incident { return result } +// ErrOpenIncidentWithoutSeverity is returned when an event tries to open a new incident without a severity. +var ErrOpenIncidentWithoutSeverity = errors.New("cannot open or escalate an incident without a severity") + // ProcessEvent from an event.Event. // // 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 event.ErrSuperfluousStateChange. +// It might return [ErrOpenIncidentWithoutSeverity] if the event is trying to open an incident without a severity or +// [ErrSeverityChangeWithoutIncidentFlag] if the event is trying to change the severity of an incident without the +// incident flag set. In both cases, the listener should map these errors to a 400 Bad Request response to the source. func ProcessEvent( ctx context.Context, db *database.DB, @@ -208,19 +214,21 @@ func ProcessEvent( runtimeConfig *config.RuntimeConfig, ev *event.Event, ) error { + o := object.Get(db, ev) + if ev.OpenOrEscalate() && ev.Severity == baseEv.SeverityNone && !HasCurrent(o) { + return ErrOpenIncidentWithoutSeverity + } + currentIncident := GetCurrent( db, - object.Get(db, ev), + o, logs.GetChildLogger("incident"), runtimeConfig, - CanOpenNewIncident(ev)) + ev.OpenOrEscalate()) if currentIncident == nil { - if ev.Severity == baseEv.SeverityOK { - return fmt.Errorf("%w: ok state event from source %d", event.ErrSuperfluousStateChange, ev.SourceId) - } - if CanOpenNewIncident(ev) { - panic(fmt.Sprintf("cannot process event %v with a non-OK state %v without a known incident", ev, ev.Severity)) + if ev.OpenOrEscalate() { + panic(fmt.Sprintf("BUG: incident should have been created for event %v, but it was not", ev)) } return nil } @@ -228,11 +236,6 @@ func ProcessEvent( return currentIncident.ProcessEvent(ctx, ev) } -// CanOpenNewIncident returns true if the given event can open a new incident if there is no active one yet. -func CanOpenNewIncident(ev *event.Event) bool { - return ev.Severity != baseEv.SeverityNone && ev.Severity != baseEv.SeverityOK -} - // HasCurrent returns true if there is an active incident for the given object. func HasCurrent(obj *object.Object) bool { currentIncidentsMu.Lock() diff --git a/internal/incident/incidents_test.go b/internal/incident/incidents_test.go index f99e0fc4..089ad700 100644 --- a/internal/incident/incidents_test.go +++ b/internal/incident/incidents_test.go @@ -2,6 +2,9 @@ package incident import ( "context" + "testing" + "time" + "github.com/icinga/icinga-go-library/database" "github.com/icinga/icinga-go-library/logging" baseEv "github.com/icinga/icinga-go-library/notifications/event" @@ -13,14 +16,18 @@ import ( "github.com/jmoiron/sqlx" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" "go.uber.org/zap/zaptest" - "testing" - "time" ) -func TestLoadOpenIncidents(t *testing.T) { - ctx := context.Background() - db := testutils.GetTestDB(ctx, t) +func TestIncidents(t *testing.T) { + t.Parallel() + + db := testutils.GetTestDB(t.Context(), t) + logs := logging.NewLoggingWithFactory("testing", zapcore.DebugLevel, time.Second, func(level zap.AtomicLevel) zapcore.Core { + return zaptest.NewLogger(t, zaptest.Level(level.Level())).Core() + }) // Insert a dummy source for our test cases! source := &config.Source{ @@ -31,7 +38,7 @@ func TestLoadOpenIncidents(t *testing.T) { source.ChangedAt = types.UnixMilli(time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC)) source.Deleted = types.Bool{Bool: false, Valid: true} - err := db.ExecTx(ctx, nil, func(ctx context.Context, tx *sqlx.Tx) error { + err := db.ExecTx(t.Context(), nil, func(ctx context.Context, tx *sqlx.Tx) error { id, err := database.InsertObtainID(ctx, tx, database.BuildInsertStmtWithout(db, source, "id"), source) require.NoError(t, err, "populating source table should not fail") @@ -40,59 +47,192 @@ func TestLoadOpenIncidents(t *testing.T) { }) require.NoError(t, err, "db.ExecTx should not fail") - // Reduce the default placeholders per statement to a meaningful number, so that we can - // test some parallelism when loading the incidents. - db.Options.MaxPlaceholdersPerStatement = 100 + t.Cleanup(func() { + // Cleanup all the database tables, so that one can just re-run the tests without having either to re-create + // the database or to manually clean it up. We can't use t.Context() here though, as it will be canceled before + // our cleanup function gets called, so we need to create a new context with a timeout for the cleanup. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + cleanupDB(ctx, db, t) + }) - // Due to the 10*maxPlaceholders constraint, only 10 goroutines are going to process simultaneously. - // Therefore, reduce the default maximum number of connections per table to 4 in order to fully simulate - // semaphore lock wait cycles for a given table. - db.Options.MaxConnectionsPerTable = 4 + runtimeConfig := config.NewRuntimeConfig(logs, db) + require.NoError(t, runtimeConfig.UpdateFromDatabase(t.Context())) - testData := make(map[string]*Incident, 10*db.Options.MaxPlaceholdersPerStatement) - for j := 1; j <= 10*db.Options.MaxPlaceholdersPerStatement; j++ { - i := makeIncident(ctx, db, t, source.ID, false) - testData[i.ObjectID.String()] = i - } + t.Run("LoadOpenIncidents", func(t *testing.T) { + // Reduce the default placeholders per statement to a meaningful number, so that we can + // test some parallelism when loading the incidents. + db.Options.MaxPlaceholdersPerStatement = 100 - t.Run("WithNoRecoveredIncidents", func(t *testing.T) { - assertIncidents(ctx, db, t, testData) - }) + // Due to the 10*maxPlaceholders constraint, only 10 goroutines are going to process simultaneously. + // Therefore, reduce the default maximum number of connections per table to 4 in order to fully simulate + // semaphore lock wait cycles for a given table. + db.Options.MaxConnectionsPerTable = 4 - t.Run("WithSomeRecoveredIncidents", func(t *testing.T) { - tx, err := db.BeginTxx(ctx, nil) - require.NoError(t, err, "starting a transaction should not fail") + testData := make(map[string]*Incident, 10*db.Options.MaxPlaceholdersPerStatement) + for j := 1; j <= 10*db.Options.MaxPlaceholdersPerStatement; j++ { + i := makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, withIncident(), withSeverity(baseEv.SeverityCrit))) + testData[i.ObjectID.String()] = i + } - // Drop all cached incidents before re-loading them! - for _, i := range GetCurrentIncidents() { - RemoveCurrent(i.Object) + t.Run("WithNoRecoveredIncidents", func(t *testing.T) { + assertIncidents(t.Context(), db, logs.GetChildLogger("incident"), runtimeConfig, t, testData) + }) - // Mark some of the existing incidents as recovered. - if i.Id%20 == 0 { // 1000 / 20 => 50 existing incidents will be marked as recovered! - i.RecoveredAt = types.UnixMilli(time.Now()) + t.Run("WithSomeRecoveredIncidents", func(t *testing.T) { + for _, i := range GetCurrentIncidents() { + // Mark some of the existing incidents as recovered. + if i.Id%20 == 0 { // 1000 / 20 => 50 existing incidents will be marked as recovered! + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, + withIncident(), withClose(), withTags(i.Object.Tags)))) + require.NotZero(t, i.RecoveredAt) + require.False(t, HasCurrent(i.Object)) // Incident is gone (closed)? + delete(testData, i.ObjectID.String()) + } else { + RemoveCurrent(i.Object) // Drop it from the cache to simulate a daemon reload. + } + } - require.NoError(t, i.Sync(ctx, tx), "failed to update/insert incident") + assert.Len(t, GetCurrentIncidents(), 0, "there should be no cached incidents") - // Drop it from our test data as it's recovered - delete(testData, i.ObjectID.String()) - } - } - require.NoError(t, tx.Commit(), "committing a transaction should not fail") + for j := 1; j <= db.Options.MaxPlaceholdersPerStatement/2; j++ { + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, + withIncident(), withClose(), withSeverity(baseEv.SeverityAlert)))) - assert.Len(t, GetCurrentIncidents(), 0, "there should be no cached incidents") + if j%2 == 0 { + // Add some extra new not recovered incidents to fully simulate a daemon reload. + i := makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, + withIncident(), withSeverity(baseEv.SeverityWarning))) + testData[i.ObjectID.String()] = i + } + } - for j := 1; j <= db.Options.MaxPlaceholdersPerStatement/2; j++ { - // We don't need to cache recovered incidents in memory. - _ = makeIncident(ctx, db, t, source.ID, true) + assertIncidents(t.Context(), db, logs.GetChildLogger("incident"), runtimeConfig, t, testData) - if j%2 == 0 { - // Add some extra new not recovered incidents to fully simulate a daemon reload. - i := makeIncident(ctx, db, t, source.ID, false) - testData[i.ObjectID.String()] = i + // Close all remaining incidents to clean up the database for the next test run. + for _, i := range GetCurrentIncidents() { + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, + withIncident(), withClose(), withTags(i.Object.Tags)))) } - } + assert.Len(t, GetCurrentIncidents(), 0, "there should be no cached incidents") + testData = make(map[string]*Incident) // Reset test data for the next test run. + }) + }) + + t.Run("Severity Change", func(t *testing.T) { + t.Parallel() + + i := makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, withIncident(), withSeverity(baseEv.SeverityDebug))) + assert.NotZero(t, i.ID()) + assert.Zero(t, i.RecoveredAt) + assert.Equal(t, baseEv.SeverityDebug, i.Severity) - assertIncidents(ctx, db, t, testData) + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, + withIncident(), withSeverity(baseEv.SeverityEmerg), withTags(i.Object.Tags)))) + assert.Equal(t, baseEv.SeverityEmerg, i.Severity) + + err := ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, + withMuted(false), withSeverity(baseEv.SeverityNotice), withTags(i.Object.Tags))) + require.ErrorIs(t, err, ErrSeverityChangeWithoutIncidentFlag) + assert.Equal(t, baseEv.SeverityEmerg, i.Severity) + }) + + t.Run("Incident Open", func(t *testing.T) { + t.Parallel() + + i := makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, withIncident(), withSeverity(baseEv.SeverityDebug))) + assert.NotZero(t, i.ID()) + assert.Zero(t, i.RecoveredAt) + assert.Equal(t, baseEv.SeverityDebug, i.Severity) + + // Attempting to open an incident without a severity should fail. + err := ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, withIncident())) + require.ErrorIs(t, err, ErrOpenIncidentWithoutSeverity) + + i = makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, + withIncident(), withSeverity(baseEv.SeverityEmerg), withMsg("Incident opened!"))) + assert.NotZero(t, i.ID()) + assert.Equal(t, baseEv.SeverityEmerg, i.Severity) + assert.Equal(t, "Incident opened!", i.Message.String) + + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, + withIncident(), + withSeverity(baseEv.SeverityEmerg), + withMsg("Incident updated!"), + withTags(i.Object.Tags)))) + assert.Equal(t, baseEv.SeverityEmerg, i.Severity) + assert.Equal(t, "Incident updated!", i.Message.String) + + // We shouldn't be able to update the incident message without the incident flag set. + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, withMuted(false), withMsg("YOLO!")))) + assert.Equal(t, "Incident updated!", i.Message.String) + }) + + t.Run("Close Flag", func(t *testing.T) { + t.Parallel() + + // Incident opened and closed immediately, so it won't be in the cache anymore. + require.Nil(t, makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, + withIncident(), withClose(), withSeverity(baseEv.SeverityDebug)))) + + i := makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, + withIncident(), withSeverity(baseEv.SeverityInfo))) + assert.Zero(t, i.RecoveredAt) + assert.Equal(t, baseEv.SeverityInfo, i.Severity) + + // Closing incident with a new severity will update the severity and mark it as recovered. + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, + withIncident(), withClose(), withSeverity(baseEv.SeverityEmerg), withTags(i.Object.Tags)))) + assert.NotZero(t, i.RecoveredAt) + assert.Equal(t, baseEv.SeverityEmerg, i.Severity) + + i = makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, withIncident(), withSeverity(baseEv.SeverityWarning))) + assert.Zero(t, i.RecoveredAt) + assert.Equal(t, baseEv.SeverityWarning, i.Severity) + + // Closing incident without providing a severity will keep the existing severity and mark it as recovered. + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, + withIncident(), withClose(), withTags(i.Object.Tags)))) + assert.NotZero(t, i.RecoveredAt) + assert.Equal(t, baseEv.SeverityWarning, i.Severity) + }) + + t.Run("Notify Flag", func(t *testing.T) { + t.Skipf("Skipping Notify Flag test, as it requires to verify whether notifications were sent") + }) + + t.Run("Muted Flag", func(t *testing.T) { + t.Parallel() + + i := makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, + withIncident(), withSeverity(baseEv.SeverityDebug), withMuted(true))) + assert.Equal(t, baseEv.SeverityDebug, i.Severity) + assert.True(t, i.IsMuted()) + assert.Equal(t, "You're gonna have a bad time!", i.MuteReason.String) + + // Unmute it with the incident flag still set... + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, + withIncident(), withMuted(false), withTags(i.Object.Tags)))) + assert.Equal(t, baseEv.SeverityDebug, i.Severity) + assert.False(t, i.IsMuted()) + assert.Equal(t, "", i.MuteReason.String) + + i = makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, + withIncident(), withSeverity(baseEv.SeverityDebug), withMuted(true))) + assert.Equal(t, baseEv.SeverityDebug, i.Severity) + assert.True(t, i.IsMuted()) + assert.Equal(t, "You're gonna have a bad time!", i.MuteReason.String) + + // Unmute it without the incident flag set... + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, makeEvent(t, source.ID, + withMuted(false), withTags(i.Object.Tags)))) + assert.Equal(t, baseEv.SeverityDebug, i.Severity) + assert.False(t, i.IsMuted()) + assert.Equal(t, "", i.MuteReason.String) + + // Muted flag without the incident flag has no effect on non-existing incidents. + i = makeIncident(db, logs, runtimeConfig, t, makeEvent(t, source.ID, withMuted(true))) + require.Nil(t, i) }) } @@ -100,20 +240,23 @@ func TestLoadOpenIncidents(t *testing.T) { // // The incident loading process is limited to a maximum duration of 10 seconds and will be // aborted and causes the entire test suite to fail immediately, if it takes longer. -func assertIncidents(ctx context.Context, db *database.DB, t *testing.T, testData map[string]*Incident) { - logger := logging.NewLogger(zaptest.NewLogger(t).Sugar(), time.Hour) - +func assertIncidents(ctx context.Context, db *database.DB, l *logging.Logger, rc *config.RuntimeConfig, t *testing.T, testData map[string]*Incident) { // Since we have been using object.FromEvent() to persist the test objects to the database, // these will be automatically added to the objects cache as well. So clear the cache before // reloading the incidents, otherwise it will panic in object.RestoreObjects(). object.ClearCache() + // Clear the incidents for the same reasons as above. + currentIncidentsMu.Lock() + currentIncidents = make(map[*object.Object]*Incident) + currentIncidentsMu.Unlock() + // The incident loading process may hang due to unknown bugs or semaphore lock waits. // Therefore, give it maximum time of 10s to finish normally, otherwise give up and fail. ctx, cancelFunc := context.WithDeadline(ctx, time.Now().Add(10*time.Second)) defer cancelFunc() - err := LoadOpenIncidents(ctx, db, logger, &config.RuntimeConfig{}) + err := LoadOpenIncidents(ctx, db, l, rc) require.NoError(t, err, "failed to load not recovered incidents") incidents := GetCurrentIncidents() @@ -141,38 +284,76 @@ func assertIncidents(ctx context.Context, db *database.DB, t *testing.T, testDat } } -// makeIncident returns a fully initialised recovered/un-recovered incident. +// cleanupDB removes all test data from the database tables used by the incident package. // -// This will firstly create and synchronise a new object from a freshly generated dummy event with distinct -// tags and name, and ensures that no error is returned, otherwise it will cause the entire test suite to fail. -// Once the object has been successfully synchronised, an incident is created and synced with the database. -func makeIncident(ctx context.Context, db *database.DB, t *testing.T, sourceID int64, recovered bool) *Incident { +// If we introduce new tests in the future that require additional database tables, we need +// to add them to the list of tables to clean up here. +func cleanupDB(ctx context.Context, db *database.DB, t *testing.T) { + switch db.DriverName() { + case database.PostgreSQL: + // As opposed to MySQL, we can just use truncate to clean up all tables in one go. + _, err := db.ExecContext(ctx, `TRUNCATE TABLE source RESTART IDENTITY CASCADE`) + require.NoError(t, err) + case database.MySQL: + // InnoDB doesn't support truncating tables with foreign key constraints, so we need to delete + // each table one by one in the correct order, not to violate any foreign key constraints. + tables := []string{ + "incident_history", + "incident", + "object_id_tag", + "object", + "source", + } + + for _, table := range tables { + _, err := db.ExecContext(ctx, "DELETE FROM "+table) + require.NoErrorf(t, err, "failed to clean up table %s", table) + } + default: + t.Fatalf("unsupported database driver: %s", db.DriverName()) + } +} + +// makeIncident creates a new incident by processing the given event and returns the resulting incident object. +// +// The incident is guaranteed to be fully initialized and ready for assertions but might be nil if it's immediately closed. +func makeIncident(db *database.DB, logs *logging.Logging, runtimeConfig *config.RuntimeConfig, t *testing.T, ev *event.Event) *Incident { + require.NoError(t, ProcessEvent(t.Context(), db, logs, runtimeConfig, ev)) + return GetCurrent(db, object.Get(db, ev), logs.GetChildLogger("incident"), runtimeConfig, false) +} + +// makeEvent returns a fully initialized event based on the given parameters. +func makeEvent(t *testing.T, sourceID int64, opts ...eventOption) *event.Event { ev := &event.Event{ - Time: time.Time{}, + Time: time.Now().Add(-2 * time.Hour).Truncate(time.Second), SourceId: sourceID, - Event: baseEv.Event{ - Name: testutils.MakeRandomString(t), - Tags: map[string]string{ // Always generate unique object tags not to produce same object ID! - "host": testutils.MakeRandomString(t), - "service": testutils.MakeRandomString(t), - }, - }, + Event: baseEv.Event{Name: testutils.MakeRandomString(t)}, + } + for _, opt := range opts { + opt(ev) + } + if ev.Tags == nil { + ev.Tags = map[string]string{ // Always generate unique object tags not to produce same object ID! + "host": testutils.MakeRandomString(t), + "service": testutils.MakeRandomString(t), + } } - i := NewIncident(db, object.Get(db, ev), &config.RuntimeConfig{}, nil) - i.StartedAt = types.UnixMilli(time.Now().Add(-2 * time.Hour).Truncate(time.Second)) - i.Severity = baseEv.SeverityCrit - if recovered { - i.Severity = baseEv.SeverityOK - i.RecoveredAt = types.UnixMilli(time.Now()) + if ev.Muted.Valid { + ev.MutedReason = "You're gonna have a bad time!" } + require.NoError(t, ev.Validate(), "failed to validate event") + return ev +} - require.NoError(t, db.ExecTx(ctx, nil, func(ctx context.Context, tx *sqlx.Tx) error { - if err := i.Object.SyncFromEvent(ctx, tx, ev); err != nil { - return err - } - return i.Sync(ctx, tx) - })) +// eventOption is a functional option type for modifying an event. +type eventOption func(*event.Event) - return i +func withIncident() eventOption { return func(ev *event.Event) { ev.Incident = types.MakeBool(true) } } +func withClose() eventOption { return func(ev *event.Event) { ev.Close = types.MakeBool(true) } } +func withMuted(v bool) eventOption { return func(ev *event.Event) { ev.Muted = types.MakeBool(v) } } +func withTags(tags map[string]string) eventOption { return func(ev *event.Event) { ev.Tags = tags } } +func withMsg(msg string) eventOption { return func(ev *event.Event) { ev.Message = msg } } +func withSeverity(sev baseEv.Severity) eventOption { + return func(ev *event.Event) { ev.Severity = sev } } diff --git a/internal/incident/sync.go b/internal/incident/sync.go index 7686a240..f012c7bd 100644 --- a/internal/incident/sync.go +++ b/internal/incident/sync.go @@ -16,12 +16,13 @@ import ( ) // Upsert implements the contracts.Upserter interface. -func (i *Incident) Upsert() interface{} { +func (i *Incident) Upsert() any { return &struct { Severity baseEv.Severity `db:"severity"` RecoveredAt types.UnixMilli `db:"recovered_at"` MuteReason types.String `db:"mute_reason"` - }{Severity: i.Severity, RecoveredAt: i.RecoveredAt, MuteReason: i.MuteReason} + Message types.String `db:"message"` + }{} } // Sync initiates an *incident.IncidentRow from the current incident state and syncs it with the database. diff --git a/internal/listener/listener.go b/internal/listener/listener.go index b5542789..e20aea9f 100644 --- a/internal/listener/listener.go +++ b/internal/listener/listener.go @@ -15,7 +15,6 @@ import ( "github.com/icinga/icinga-notifications/internal/daemon" "github.com/icinga/icinga-notifications/internal/event" "github.com/icinga/icinga-notifications/internal/incident" - "github.com/icinga/icinga-notifications/internal/object" "go.uber.org/zap" "net/http" "time" @@ -200,13 +199,6 @@ func (l *Listener) ProcessEvent(w http.ResponseWriter, r *http.Request) { ev.CompleteURL(daemon.Config().IcingaWeb2UrlParsed) ev.Time = time.Now() ev.SourceId = src.ID - if ev.Type == baseEv.TypeUnknown { - ev.Type = baseEv.TypeState - } else if !ev.Mute.Valid && ev.Type == baseEv.TypeMute { - ev.SetMute(true, ev.MuteReason) - } else if !ev.Mute.Valid && ev.Type == baseEv.TypeUnmute { - ev.SetMute(false, ev.MuteReason) - } if err := ev.Validate(); err != nil { l.abort(w, http.StatusBadRequest, src, "%v", err) @@ -215,16 +207,20 @@ func (l *Listener) ProcessEvent(w http.ResponseWriter, r *http.Request) { l.logger.Debugw("Processing event", zap.String("source", src.Name), zap.Object("event", &ev)) - filterColumns, hasRulesWithoutFilter := l.runtimeConfig.GetRulesFilterColumnsForSource(src) - missingRelations := ev.ExtractMissingRelations(filterColumns...) - if len(missingRelations) > 0 && ShouldRejectRequestOnIncompleteRelations(r, &ev, hasRulesWithoutFilter) { - l.sendMissingAttrsError(w, src, missingRelations) - return + // Submitting an event without the "incident" field won't cause any new event rules to be evaluated or + // escalations to be triggered, but only updates the state of an existing incident without a severity change. + if ev.OpenOrEscalate() { + filterColumns, hasRulesWithoutFilter := l.runtimeConfig.GetRulesFilterColumnsForSource(src) + missingRelations := ev.ExtractMissingRelations(filterColumns...) + if len(missingRelations) > 0 && ShouldRejectRequestOnIncompleteRelations(r, hasRulesWithoutFilter) { + l.sendMissingAttrsError(w, src, missingRelations) + return + } } err := incident.ProcessEvent(context.Background(), l.db, l.logs, l.runtimeConfig, &ev) - if errors.Is(err, event.ErrSuperfluousStateChange) || errors.Is(err, event.ErrSuperfluousMuteUnmuteEvent) { - l.abort(w, http.StatusNotAcceptable, src, "%v", err) + if errors.Is(err, incident.ErrSeverityChangeWithoutIncidentFlag) || errors.Is(err, incident.ErrOpenIncidentWithoutSeverity) { + l.abort(w, http.StatusBadRequest, src, "%v", err) return } else if err != nil { l.logger.Errorw("Failed to successfully process event", zap.String("source", src.Name), zap.Error(err)) @@ -430,15 +426,17 @@ func (l *Listener) sendMissingAttrsError(w http.ResponseWriter, src *config.Sour // ShouldRejectRequestOnIncompleteRelations determines whether a request with incomplete relations should be rejected. // // This function always returns true if the client explicitly requested to reject such events by setting the -// [notifications.XIcingaRejectIfRelationsIncomplete] HTTP header. Otherwise, it only returns true when the -// src doesn't have any rules without an object filter and the event doesn't cause a new incident to be opened -// and there's no active one yet for the event's source object. -func ShouldRejectRequestOnIncompleteRelations(r *http.Request, ev *event.Event, hasRulesWithoutFilter bool) bool { +// [notifications.XIcingaRejectIfRelationsIncomplete] HTTP header. Otherwise, if there are rules without a filter, +// it returns false as such rules match unconditionally and thus don't necessarily require the missing relations. +// +// Note that this function assumes that it's guarded by a [baseEv.Event.OpenOrEscalate] check, since only +// events that open or escalate an incident are subject to relation completeness checks in the first place. +func ShouldRejectRequestOnIncompleteRelations(r *http.Request, hasRulesWithoutFilter bool) bool { if r.Header.Get(notifications.XIcingaRejectIfRelationsIncomplete) == "true" { return true } if hasRulesWithoutFilter { return false } - return !incident.CanOpenNewIncident(ev) && !incident.HasCurrent(object.GetFromCache(object.ID(ev.SourceId, ev.Tags))) + return true } diff --git a/internal/object/object.go b/internal/object/object.go index befd220c..153b8032 100644 --- a/internal/object/object.go +++ b/internal/object/object.go @@ -64,8 +64,7 @@ func ClearCache() { // // Returns error on any database failure. func (o *Object) SyncFromEvent(ctx context.Context, tx *sqlx.Tx, ev *event.Event) error { - newObject := *o - newObject.Name = ev.Name + newObject := Object{ID: o.ID, SourceID: o.SourceID, Name: ev.Name} newObject.URL = types.MakeString(ev.URL, types.TransformEmptyStringToNull) stmt, _ := o.db.BuildUpsertStmt(o) @@ -78,7 +77,8 @@ func (o *Object) SyncFromEvent(ctx context.Context, tx *sqlx.Tx, ev *event.Event return fmt.Errorf("failed to upsert object id tags: %w", err) } - *o = newObject + o.Name = newObject.Name + o.URL = newObject.URL return nil } diff --git a/schema/mysql/schema.sql b/schema/mysql/schema.sql index 3531035a..e9e96995 100644 --- a/schema/mysql/schema.sql +++ b/schema/mysql/schema.sql @@ -353,6 +353,7 @@ CREATE TABLE incident ( severity enum('ok', 'debug', 'info', 'notice', 'warning', 'err', 'crit', 'alert', 'emerg'), -- mute_reason indicates whether this incident is currently muted, and its non-null value is mapped to true. mute_reason mediumtext, + message longtext, -- contains the latest plugin output of the respective object. CONSTRAINT pk_incident PRIMARY KEY (id), CONSTRAINT ck_incident_severity_notnull CHECK (severity IS NOT NULL), diff --git a/schema/mysql/upgrades/incident-message.sql b/schema/mysql/upgrades/incident-message.sql new file mode 100644 index 00000000..3ed448a0 --- /dev/null +++ b/schema/mysql/upgrades/incident-message.sql @@ -0,0 +1 @@ +ALTER TABLE incident ADD COLUMN message longtext; diff --git a/schema/pgsql/schema.sql b/schema/pgsql/schema.sql index b6a2eac4..a5a3ed10 100644 --- a/schema/pgsql/schema.sql +++ b/schema/pgsql/schema.sql @@ -381,6 +381,7 @@ CREATE TABLE incident ( severity severity NOT NULL, -- mute_reason indicates whether this incident is currently muted, and its non-null value is mapped to true. mute_reason text, + message text, -- contains the latest plugin output of the respective object. CONSTRAINT pk_incident PRIMARY KEY (id), CONSTRAINT fk_incident_object FOREIGN KEY (object_id) REFERENCES object(id) diff --git a/schema/pgsql/upgrades/incident-message.sql b/schema/pgsql/upgrades/incident-message.sql new file mode 100644 index 00000000..643c6233 --- /dev/null +++ b/schema/pgsql/upgrades/incident-message.sql @@ -0,0 +1 @@ +ALTER TABLE incident ADD COLUMN message text;