2c884c5612
Некоторые проверки не удались
CI / test (push) Successful in 2m5s
Docker / Build and publish worker image (push) Failing after 31s
371 строка
10 KiB
Go
371 строка
10 KiB
Go
package notifier
|
|
|
|
import (
|
|
"log"
|
|
"time"
|
|
|
|
"gorm.io/gorm"
|
|
|
|
"rocketgit.ru/rsmon/worker/app/models"
|
|
)
|
|
|
|
const (
|
|
eventStateEnded = "ended"
|
|
eventStateCurrent = "current"
|
|
messageKindDown = "down"
|
|
messageKindUp = "up"
|
|
)
|
|
|
|
// DEBUG provides functionality.
|
|
const DEBUG = false
|
|
|
|
// Run starts the notification scheduler loop.
|
|
func Run() {
|
|
_ = models.LogCheck("notify")
|
|
|
|
events := make([]models.Event, 0)
|
|
|
|
tx := models.DB().Begin()
|
|
|
|
q := tx
|
|
// q = q.Set("gorm:query_option", "FOR UPDATE")
|
|
err := models.EventScope(q).Find(&events).Error
|
|
if err != nil {
|
|
tx.Rollback()
|
|
log.Println(err)
|
|
return
|
|
// panic(err)
|
|
}
|
|
|
|
type SendItem struct {
|
|
Notification models.Notification
|
|
Contact models.Contact
|
|
Events []models.Event
|
|
}
|
|
|
|
eventsByNotification := make(map[int64]map[int64]*SendItem, 0)
|
|
|
|
hasPossible := make(map[int64]bool)
|
|
|
|
eventIDs := make(map[int64]bool, 0)
|
|
contactIDs := make(map[int64]bool, 0)
|
|
alreadySentDown := make(map[int64]map[int64]bool, 0)
|
|
alreadySentUp := make(map[int64]map[int64]bool, 0)
|
|
for _, e := range events { //nolint:gocritic // range copy is acceptable here
|
|
eventIDs[e.ID] = true
|
|
if e.Monitor == nil {
|
|
println("event has no monitor")
|
|
e.State = "broken"
|
|
err := tx.Save(&e).Error
|
|
if err != nil {
|
|
tx.Rollback()
|
|
log.Println(err)
|
|
return
|
|
// panic(err)
|
|
}
|
|
continue
|
|
}
|
|
if e.Monitor.Group == nil {
|
|
println("monitor has no group")
|
|
e.State = "broken"
|
|
err := tx.Save(&e).Error
|
|
if err != nil {
|
|
tx.Rollback()
|
|
log.Println(err)
|
|
return
|
|
// panic(err)
|
|
}
|
|
continue
|
|
}
|
|
for _, n := range e.Monitor.Group.Notifications { //nolint:gocritic // range copy is acceptable here
|
|
for _, c := range n.Contacts { //nolint:gocritic // range copy is acceptable here
|
|
contactIDs[c.ID] = true
|
|
alreadySentDown[c.ID] = make(map[int64]bool, 0)
|
|
alreadySentUp[c.ID] = make(map[int64]bool, 0)
|
|
}
|
|
}
|
|
}
|
|
|
|
sentMessages := make([]models.Message, 0)
|
|
|
|
eventIDsSlice := make([]int64, 0)
|
|
for k := range eventIDs {
|
|
eventIDsSlice = append(eventIDsSlice, k)
|
|
}
|
|
contactIDsSlice := make([]int64, 0)
|
|
for k := range contactIDs {
|
|
contactIDsSlice = append(contactIDsSlice, k)
|
|
}
|
|
|
|
err = tx.Preload("Events").
|
|
// Where("kind = ?", "down").
|
|
Where("id IN (SELECT message_id FROM event_messages WHERE event_id IN (?))", eventIDsSlice).
|
|
Where("contact_id IN (?)", contactIDsSlice).Find(&sentMessages).
|
|
Error
|
|
if err != nil {
|
|
tx.Rollback()
|
|
log.Println(err)
|
|
return
|
|
// panic(err)
|
|
}
|
|
|
|
for _, m := range sentMessages { //nolint:gocritic // range copy is acceptable here
|
|
for _, evt := range m.Events { //nolint:gocritic // range copy is acceptable here
|
|
switch m.Kind {
|
|
case messageKindDown:
|
|
alreadySentDown[m.ContactID][evt.ID] = true
|
|
case messageKindUp:
|
|
alreadySentUp[m.ContactID][evt.ID] = true
|
|
}
|
|
}
|
|
}
|
|
|
|
for _, e := range events { //nolint:gocritic // range copy is acceptable here
|
|
if e.Monitor == nil {
|
|
println("event has no monitor")
|
|
continue
|
|
}
|
|
if e.Monitor.Group == nil {
|
|
println("monitor has no group")
|
|
continue
|
|
}
|
|
underMaintenance, maintenanceErr := models.MonitorUnderMaintenance(e.MonitorID, time.Now().UTC())
|
|
if maintenanceErr != nil {
|
|
log.Printf("notifier: maintenance lookup for monitor %d: %v", e.MonitorID, maintenanceErr)
|
|
} else if underMaintenance {
|
|
// Keep the event pending. Marking it old here would silently drop a
|
|
// failure that remains unresolved after the maintenance window ends.
|
|
hasPossible[e.ID] = true
|
|
continue
|
|
}
|
|
|
|
if e.State == eventStateEnded {
|
|
hasPossible[e.ID] = false
|
|
} else {
|
|
hasPossible[e.ID] = true
|
|
}
|
|
|
|
if DEBUG {
|
|
log.Println("run event", e.Inspect())
|
|
}
|
|
for _, n := range e.Monitor.Group.Notifications { //nolint:gocritic // range copy is acceptable here
|
|
if !n.Enabled {
|
|
if DEBUG {
|
|
log.Println("event", e.ID, "notification", n.ID, "Enabled = false")
|
|
}
|
|
continue
|
|
}
|
|
|
|
if e.State == eventStateCurrent {
|
|
if !n.NotifyDown {
|
|
if DEBUG {
|
|
log.Println("event", e.ID, "notification", n.ID, "NotifyDown = false")
|
|
}
|
|
continue
|
|
}
|
|
} else if e.State == eventStateEnded {
|
|
if !n.NotifyRestore {
|
|
if DEBUG {
|
|
log.Println("event", e.ID, "notification", n.ID, "NotifyRestore = false")
|
|
}
|
|
continue
|
|
}
|
|
}
|
|
|
|
if _, ok := eventsByNotification[n.ID]; !ok {
|
|
eventsByNotification[n.ID] = make(map[int64]*SendItem, 0)
|
|
}
|
|
|
|
for _, c := range n.Contacts { //nolint:gocritic // range copy is acceptable here
|
|
if e.State == eventStateCurrent {
|
|
if _, sent := alreadySentDown[c.ID][e.ID]; sent {
|
|
if DEBUG {
|
|
log.Println("event", e.ID, "notification", n.ID, "already sent")
|
|
}
|
|
continue
|
|
}
|
|
} else if e.State == eventStateEnded {
|
|
if _, sent := alreadySentDown[c.ID][e.ID]; !sent {
|
|
if DEBUG {
|
|
log.Println("event", e.ID, "dont notify up", n.ID, "- no down was sent")
|
|
}
|
|
continue
|
|
}
|
|
|
|
if _, sent := alreadySentUp[c.ID][e.ID]; sent {
|
|
if DEBUG {
|
|
log.Println("event", e.ID, "notification", n.ID, "already sent")
|
|
}
|
|
continue
|
|
}
|
|
}
|
|
if _, ok := eventsByNotification[n.ID][c.ID]; !ok {
|
|
si := SendItem{
|
|
Notification: n,
|
|
Contact: c,
|
|
Events: make([]models.Event, 0),
|
|
}
|
|
// log.Println("create", n.ID, c.ID)
|
|
// spew.Dump(si)
|
|
eventsByNotification[n.ID][c.ID] = &si
|
|
}
|
|
sendItem := eventsByNotification[n.ID][c.ID]
|
|
sendItem.Events = append(sendItem.Events, e)
|
|
}
|
|
}
|
|
}
|
|
|
|
for _, eventsByContact := range eventsByNotification {
|
|
for _, sendItem := range eventsByContact {
|
|
n := sendItem.Notification
|
|
c := sendItem.Contact
|
|
|
|
tn := time.Now()
|
|
|
|
requredEvents := make([]models.Event, 0)
|
|
possibleEvents := make([]models.Event, 0)
|
|
laterEvents := make([]models.Event, 0)
|
|
|
|
for _, e := range sendItem.Events { //nolint:gocritic // range copy is acceptable here
|
|
dur := e.GetDuration(tn)
|
|
var delay int64
|
|
if n.AlertDelay != nil {
|
|
delay = *n.AlertDelay
|
|
} else {
|
|
delay = 300
|
|
}
|
|
|
|
if !n.EnabledNow(&tn) {
|
|
if DEBUG {
|
|
log.Println("notification", n.ID, "is not enabled at this time")
|
|
}
|
|
laterEvents = append(laterEvents, e)
|
|
}
|
|
|
|
if e.State == eventStateCurrent && e.Errors > 4 { //nolint:gocritic // complex condition chain
|
|
if DEBUG {
|
|
log.Println("min errors count to force send reached:", e.Errors)
|
|
}
|
|
requredEvents = append(requredEvents, e)
|
|
} else if e.State == eventStateCurrent && e.Errors < 2 {
|
|
if DEBUG {
|
|
log.Println("event possbile to notify in aggregation, but errs count not reached:", e.Errors)
|
|
}
|
|
possibleEvents = append(possibleEvents, e)
|
|
} else if e.State == eventStateEnded && e.Oks > 4 {
|
|
if DEBUG {
|
|
log.Println("min oks count to force send reached:", e.Oks)
|
|
}
|
|
requredEvents = append(requredEvents, e)
|
|
} else if e.State == eventStateEnded && e.Oks < 2 {
|
|
if DEBUG {
|
|
log.Println("event possbile to notify in aggregation, but oks count not reached:", e.Oks)
|
|
}
|
|
possibleEvents = append(possibleEvents, e)
|
|
} else if dur < delay {
|
|
if DEBUG {
|
|
log.Println("event possbile to notify in aggregation, but alert_delay not reached: delay", delay, "duration", dur, "so", (delay - dur), "left") //nolint:lll
|
|
}
|
|
possibleEvents = append(possibleEvents, e)
|
|
} else {
|
|
if DEBUG {
|
|
log.Println("event required to notify, alert_delay reached: delay", delay, "duration", dur, "so", (delay - dur), "left") //nolint:lll
|
|
}
|
|
requredEvents = append(requredEvents, e)
|
|
}
|
|
}
|
|
if len(requredEvents) > 0 {
|
|
performEvents(tx, &n, &c, append(requredEvents, possibleEvents...))
|
|
} else {
|
|
if len(possibleEvents) > 0 || len(laterEvents) > 0 {
|
|
// log.Println("notification", n.ID, "no required events, but will send later")
|
|
for _, evt := range possibleEvents { //nolint:gocritic // range copy is acceptable here
|
|
hasPossible[evt.ID] = true
|
|
}
|
|
for _, evt := range laterEvents { //nolint:gocritic // range copy is acceptable here
|
|
hasPossible[evt.ID] = true
|
|
}
|
|
} else {
|
|
log.Println("notification", n.ID, "no events left")
|
|
}
|
|
}
|
|
|
|
// spew.Dump(sendItem.Notification)
|
|
// spew.Dump(sendItem.Contact)
|
|
// spew.Dump(sendItem.Events)
|
|
}
|
|
}
|
|
|
|
for _, e := range events { //nolint:gocritic // range copy is acceptable here
|
|
if !hasPossible[e.ID] {
|
|
// log.Println("event has no possible notifications left to send, mark as done")
|
|
e.State = "old"
|
|
err := tx.Save(&e).Error
|
|
if err != nil {
|
|
tx.Rollback()
|
|
log.Println(err)
|
|
return
|
|
// panic(err)
|
|
}
|
|
}
|
|
}
|
|
|
|
tx.Commit()
|
|
}
|
|
|
|
func performEvents(tx *gorm.DB, n *models.Notification, c *models.Contact, events []models.Event) {
|
|
eventIDs := make([]int64, 0, len(events))
|
|
for _, evt := range events { //nolint:gocritic // range copy is acceptable here
|
|
eventIDs = append(eventIDs, evt.ID)
|
|
}
|
|
log.Println("performing events:", n.ID, c.ID, eventIDs)
|
|
|
|
eventsByKind := make(map[string][]models.Event, 0)
|
|
for _, evt := range events { //nolint:gocritic // range copy is acceptable here
|
|
var kind string
|
|
switch evt.State {
|
|
case eventStateCurrent:
|
|
kind = messageKindDown
|
|
case eventStateEnded:
|
|
kind = messageKindUp
|
|
default:
|
|
log.Println("unknown event state: " + evt.State)
|
|
tx.Rollback()
|
|
return
|
|
}
|
|
if _, ok := eventsByKind[kind]; !ok {
|
|
eventsByKind[kind] = make([]models.Event, 0)
|
|
}
|
|
eventsByKind[kind] = append(eventsByKind[kind], evt)
|
|
}
|
|
|
|
for kind, evts := range eventsByKind {
|
|
// spew.Dump(kind, evts)
|
|
message := models.Message{
|
|
NotificationID: n.ID,
|
|
ContactID: c.ID,
|
|
Events: evts,
|
|
Kind: kind,
|
|
State: models.TaskStateQueued,
|
|
}
|
|
err := tx.Save(&message).Error
|
|
if err != nil {
|
|
// panic(err)
|
|
tx.Rollback()
|
|
log.Println(err)
|
|
return
|
|
}
|
|
|
|
// Worker notification tasks are the only delivery path. If enqueue fails,
|
|
// keep the message as an explicit error instead of relying on the retired
|
|
// in-process sender loop.
|
|
if _, err := enqueueNotificationTaskFromMessageTx(tx, n, c, &message); err != nil {
|
|
errText := err.Error()
|
|
log.Printf("notifier: enqueue task for message %d failed: %v", message.ID, err)
|
|
if saveErr := tx.Model(&message).Updates(map[string]interface{}{"state": "error", "error": &errText}).Error; saveErr != nil {
|
|
log.Printf("notifier: mark message %d error failed: %v", message.ID, saveErr)
|
|
}
|
|
}
|
|
}
|
|
}
|