feat: notification watchdog

adityathebe · adityathebe · commit ac586a2183b7 · 2026-04-16T13:14:49.000+05:45
diff --git a/api/event.go b/api/event.go
@@ -8,6 +8,9 @@ import (
 	"github.com/google/uuid"
 )
 
+// A dummy event that lives only within the application (never in the event queue)
+const EventWatchdog = "notification.watchdog"
+
 // List of all sync events in the `event_queue` table.
 //
 // These events are generated by the database in response to updates on some of the tables.
diff --git a/api/v1/notification_types.go b/api/v1/notification_types.go
@@ -115,6 +115,12 @@ type NotificationSpec struct {
 	// It uses the repeat interval as the window for suppression
 	// as well as the wait for period.
 	Inhibitions []NotificationInihibition `json:"inhibitions,omitempty"`
+
+	// WatchdogInterval is the interval at a watchdog notification is sent.
+	// A watchdog notification contains the statistics of this notification.
+	//
+	// Format: duration string (e.g., "30s", "2m")
+	WatchdogInterval *string `json:"watchdogInterval,omitempty"`
 }
 
 type NotificationInihibition struct {
diff --git a/api/v1/zz_generated.deepcopy.go b/api/v1/zz_generated.deepcopy.go
diff --git a/cmd/server.go b/cmd/server.go
@@ -83,7 +83,7 @@ func launchKopper(ctx context.Context) {
 	}
 
 	if v1.NotificationReconciler, err = kopper.SetupReconciler(ctx, mgr,
-		db.PersistNotificationFromCRD,
+		PersistNotificationFromCRD,
 		db.DeleteNotification,
 		db.DeleteStaleNotification,
 		"notification.mission-control.flanksource.com",
@@ -159,6 +159,14 @@ func launchKopper(ctx context.Context) {
 	}
 }
 
+func PersistNotificationFromCRD(ctx context.Context, obj *v1.Notification) error {
+	if err := db.PersistNotificationFromCRD(ctx, obj); err != nil {
+		return err
+	}
+
+	return notification.SyncWatchdogJob(ctx, jobs.FuncScheduler, string(obj.GetUID()), obj.Spec.WatchdogInterval)
+}
+
 var Serve = &cobra.Command{
 	Use:    "serve",
 	PreRun: PreRun,
diff --git a/config/crds/mission-control.flanksource.com_notifications.yaml b/config/crds/mission-control.flanksource.com_notifications.yaml
@@ -1380,6 +1380,13 @@ spec:
                   Only applies to Kubernetes health notifications and cannot be turned off.
                   Defaults to 30s.
 
+                  Format: duration string (e.g., "30s", "2m")
+                type: string
+              watchdogInterval:
+                description: |-
+                  WatchdogInterval is the interval at a watchdog notification is sent.
+                  A watchdog notification contains the statistics of this notification.
+
                   Format: duration string (e.g., "30s", "2m")
                 type: string
             required:
diff --git a/config/schemas/notification.schema.json b/config/schemas/notification.schema.json
@@ -362,6 +362,10 @@
           },
           "type": "array",
           "description": "Inhibit controls notification suppression for related resources.\nIt uses the repeat interval as the window for suppression\nas well as the wait for period."
+        },
+        "watchdogInterval": {
+          "type": "string",
+          "description": "WatchdogInterval is the interval at a watchdog notification is sent.\nA watchdog notification contains the statistics of this notification.\n\nFormat: duration string (e.g., \"30s\", \"2m\")"
         }
       },
       "additionalProperties": false,
diff --git a/db/notifications.go b/db/notifications.go
@@ -76,6 +76,14 @@ func PersistNotificationFromCRD(ctx context.Context, obj *v1.Notification) error
 		}
 	}
 
+	if obj.Spec.WatchdogInterval != nil && *obj.Spec.WatchdogInterval != "" {
+		if parsed, err := text.ParseDuration(*obj.Spec.WatchdogInterval); err != nil {
+			return err
+		} else {
+			dbObj.WatchdogInterval = parsed
+		}
+	}
+
 	if len(obj.Spec.GroupBy) > 0 && obj.Spec.WaitFor != nil && *obj.Spec.WaitFor == "" {
 		return fmt.Errorf("groupBy provided with an empty waitFor. either remove the groupBy or set a waitFor period")
 	}
diff --git a/go.mod b/go.mod
@@ -88,7 +88,6 @@ require (
 	go.opentelemetry.io/otel/trace v1.43.0
 	golang.org/x/sync v0.20.0
 	google.golang.org/genai v1.34.0
-	gorm.io/hints v1.1.2
 	k8s.io/api v0.35.2
 )
 
diff --git a/go.sum b/go.sum
@@ -894,10 +894,6 @@ github.com/mattn/go-localereader v0.0.1/go.mod h1:8fBrzywKY7BI3czFoHkuzRoWE9C+Ei
 github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
 github.com/mattn/go-runewidth v0.0.20 h1:WcT52H91ZUAwy8+HUkdM3THM6gXqXuLJi9O3rjcQQaQ=
 github.com/mattn/go-runewidth v0.0.20/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs=
-github.com/mattn/go-sqlite3 v1.14.15/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
-github.com/mattn/go-sqlite3 v1.14.16/go.mod h1:2eHXhiwb8IkHr+BDWZGa96P6+rkvnG63S2DGjv9HUNg=
-github.com/mattn/go-sqlite3 v1.14.30 h1:bVreufq3EAIG1Quvws73du3/QgdeZ3myglJlrzSYYCY=
-github.com/mattn/go-sqlite3 v1.14.30/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.4/go.mod h1:BSXmuO+STAnVfrANrmjBb36TMTDstsz7MSK+HVaYKv4=
 github.com/mfridman/tparse v0.18.0 h1:wh6dzOKaIwkUGyKgOntDW4liXSo37qg5AXbIhkMV3vE=
@@ -1754,18 +1750,12 @@ gorm.io/driver/mysql v1.6.0 h1:eNbLmNTpPpTOVZi8MMxCi2aaIm0ZpInbORNXDwyLGvg=
 gorm.io/driver/mysql v1.6.0/go.mod h1:D/oCC2GWK3M/dqoLxnOlaNKmXz8WNTfcS9y5ovaSqKo=
 gorm.io/driver/postgres v1.6.0 h1:2dxzU8xJ+ivvqTRph34QX+WrRaJlmfyPqXmoGVjMBa4=
 gorm.io/driver/postgres v1.6.0/go.mod h1:vUw0mrGgrTK+uPHEhAdV4sfFELrByKVGnaVRkXDhtWo=
-gorm.io/driver/sqlite v1.5.0/go.mod h1:kDMDfntV9u/vuMmz8APHtHF0b4nyBB7sfCieC6G8k8I=
-gorm.io/driver/sqlite v1.6.0 h1:WHRRrIiulaPiPFmDcod6prc4l2VGVWHz80KspNsxSfQ=
-gorm.io/driver/sqlite v1.6.0/go.mod h1:AO9V1qIQddBESngQUKWL9yoH93HIeA1X6V633rBwyT8=
 gorm.io/driver/sqlserver v1.6.3 h1:UR+nWCuphPnq7UxnL57PSrlYjuvs+sf1N59GgFX7uAI=
 gorm.io/driver/sqlserver v1.6.3/go.mod h1:VZeNn7hqX1aXoN5TPAFGWvxWG90xtA8erGn2gQmpc6U=
-gorm.io/gorm v1.24.7-0.20230306060331-85eaf9eeda11/go.mod h1:L4uxeKpfBml98NYqVqwAdmV1a2nBtAec/cf3fpucW/k=
 gorm.io/gorm v1.25.0/go.mod h1:L4uxeKpfBml98NYqVqwAdmV1a2nBtAec/cf3fpucW/k=
 gorm.io/gorm v1.30.0/go.mod h1:8Z33v652h4//uMA76KjeDH8mJXPm1QNCYrMeatR0DOE=
 gorm.io/gorm v1.31.1 h1:7CA8FTFz/gRfgqgpeKIBcervUn3xSyPUmr6B2WXJ7kg=
 gorm.io/gorm v1.31.1/go.mod h1:XyQVbO2k6YkOis7C2437jSit3SsDK72s7n7rsSHd+Gs=
-gorm.io/hints v1.1.2 h1:b5j0kwk5p4+3BtDtYqqfY+ATSxjj+6ptPgVveuynn9o=
-gorm.io/hints v1.1.2/go.mod h1:/ARdpUHAtyEMCh5NNi3tI7FsGh+Cj/MIUlvNxCNCFWg=
 gorm.io/plugin/dbresolver v1.6.2 h1:F4b85TenghUeITqe3+epPSUtHH7RIk3fXr5l83DF8Pc=
 gorm.io/plugin/dbresolver v1.6.2/go.mod h1:tctw63jdrOezFR9HmrKnPkmig3m5Edem9fdxk9bQSzM=
 gorm.io/plugin/prometheus v0.1.0 h1:kDQwAfCUsT9D6jDUpIp7pnc7bCJu/6voM8I/BmFjxUQ=
diff --git a/jobs/jobs.go b/jobs/jobs.go
@@ -110,6 +110,10 @@ func Start(ctx context.Context, mcpServer *server.MCPServer) {
 			shutdown.ShutdownAndExit(1, fmt.Sprintf("failed to schedule job SyncCRDStatusJob: %v", err))
 		}
 
+		if err := notification.InitWatchdogNotifications(ctx, FuncScheduler); err != nil {
+			shutdown.ShutdownAndExit(1, fmt.Sprintf("failed to schedule job InitWatchdogNotifications: %v", err))
+		}
+
 		if err := notification.InitCRDStatusUpdates(ctx); err != nil {
 			logger.Errorf("failed to start notification status update queue: %v", err)
 		}
diff --git a/notification/cel.go b/notification/cel.go
@@ -43,6 +43,8 @@ type celVariables struct {
 	Permalink  string
 	SilenceURL string
 
+	Summary models.NotificationSummary
+
 	GroupedResources []string
 }
 
@@ -119,6 +121,8 @@ func (t *celVariables) AsMap(ctx context.Context, opts ...celAsMapOption) map[st
 
 		"comment": lo.FromPtr(t.Comment).AsMap(),
 		"author":  lo.FromPtr(t.Author).AsMap(),
+
+		"summary": t.Summary.AsMap(),
 	}
 
 	if len(t.GroupedResources) > 0 {
diff --git a/notification/events.go b/notification/events.go
@@ -786,6 +786,17 @@ func GetEnvForEvent(ctx context.Context, event models.Event) (*celVariables, err
 		EventTime:   event.CreatedAt,
 	}
 
+	if event.Name == api.EventWatchdog {
+		notificationID := event.Properties["id"]
+
+		stats, err := query.GetNotificationStats(ctx, notificationID)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get notification statistics: %w", err)
+		}
+
+		env.Summary = stats[0]
+	}
+
 	if strings.HasPrefix(event.Name, "check.") {
 		checkID := event.EventID.String()
 		lastRuntime := event.Properties["last_runtime"]
diff --git a/notification/job.go b/notification/job.go
@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/flanksource/commons/collections"
@@ -14,10 +15,10 @@ import (
 	"github.com/flanksource/duty/models"
 	"github.com/flanksource/duty/query"
 	"github.com/google/uuid"
+	"github.com/robfig/cron/v3"
 	"github.com/samber/lo"
 	"gorm.io/gorm"
 	"gorm.io/gorm/clause"
-	"gorm.io/hints"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -28,7 +29,12 @@ import (
 	"github.com/flanksource/incident-commander/events"
 )
 
-var CRDStatusUpdateQueue *collections.Queue[string]
+var (
+	CRDStatusUpdateQueue *collections.Queue[string]
+
+	// Keeps track of all watchdog jobs
+	watchdogJobs sync.Map
+)
 
 func InitCRDStatusUpdates(ctx context.Context) error {
 	var err error
@@ -81,26 +87,9 @@ func SyncCRDStatus(ctx context.Context, ids ...string) error {
 		return errors.New("notification reconciler is not initialized")
 	}
 
-	var summary []struct {
-		Name         string
-		Namespace    string
-		Sent         int
-		Failed       int
-		Pending      int
-		UpdatedAt    time.Time
-		Error        string
-		LastFailedAt time.Time
-	}
-
-	q := ctx.DB().Clauses(hints.CommentBefore("select", "notification_crd_sync")).
-		Table("notifications_summary").
-		Where("name != '' AND namespace != '' AND source = ?", models.SourceCRD)
-
-	if len(ids) > 0 {
-		q = q.Where("id in ?", ids)
-	}
-	if err := q.Find(&summary).Error; err != nil {
-		return fmt.Errorf("error querying notifications_summary: %w", err)
+	summary, err := query.GetNotificationStats(ctx, ids...)
+	if err != nil {
+		return ctx.Oops().Wrapf(err, "failed to get notification stats")
 	}
 
 	for _, s := range summary {
@@ -116,6 +105,7 @@ func SyncCRDStatus(ctx context.Context, ids ...string) error {
 			return fmt.Errorf("error in patchCRDStatus: %w", err)
 		}
 	}
+
 	return nil
 }
 
@@ -507,3 +497,130 @@ func isKubernetesConfigItem(ctx context.Context, configID string) (bool, error)
 
 	return len(scraperSpec.Kubernetes) != 0, nil
 }
+
+func InitWatchdogNotifications(ctx context.Context, scheduler *cron.Cron) error {
+	var notifications []models.Notification
+	if err := ctx.DB().
+		Where("watchdog_interval IS NOT NULL AND watchdog_interval != 0").
+		Where("deleted_at IS NULL").
+		Find(&notifications).Error; err != nil {
+		return fmt.Errorf("failed to get notifications with watchdog interval: %w", err)
+	}
+
+	ctx.Debugf("initializing watchdog jobs for %d notifications", len(notifications))
+	for _, n := range notifications {
+		_ = scheduleWatchdogJob(ctx, scheduler, n.ID.String(), n.WatchdogInterval.String())
+	}
+
+	return nil
+}
+
+func SyncWatchdogJob(ctx context.Context, scheduler *cron.Cron, notificationID string, interval *string) error {
+	var scheduleChanged bool
+	var existingJob *job.Job
+
+	if j, ok := watchdogJobs.Load(notificationID); ok {
+		existingJob = j.(*job.Job)
+		if interval != nil {
+			scheduleChanged = existingJob.Schedule != fmt.Sprintf("@every %s", *interval)
+		}
+
+		if interval == nil || scheduleChanged {
+			ctx.Debugf("deleting existing watchdog job for %s", notificationID)
+			existingJob.Unschedule()
+			watchdogJobs.Delete(notificationID)
+		}
+	}
+
+	if interval != nil && (scheduleChanged || existingJob == nil) {
+		return scheduleWatchdogJob(ctx, scheduler, notificationID, *interval)
+	}
+
+	return nil
+}
+
+func scheduleWatchdogJob(ctx context.Context, scheduler *cron.Cron, notificationID string, interval string) error {
+	ctx.Debugf("scheduling watchdog job for %s with interval %s", notificationID, interval)
+	job := WatchdogNotificationJob(ctx, notificationID, interval)
+	if err := job.AddToScheduler(scheduler); err != nil {
+		return fmt.Errorf("failed to add watchdog job to scheduler: %w", err)
+	}
+	watchdogJobs.Store(notificationID, job)
+	return nil
+}
+
+func WatchdogNotificationJob(ctx context.Context, notificationID string, interval string) *job.Job {
+	schedule := fmt.Sprintf("@every %s", interval)
+	return &job.Job{
+		Name:          "NotificationWatchdog",
+		Retention:     job.RetentionFailed,
+		JitterDisable: false,
+		ResourceID:    notificationID,
+		ResourceType:  "notification",
+		JobHistory:    true,
+		RunNow:        false,
+		Context:       ctx,
+		Singleton:     true,
+		Schedule:      schedule,
+		Fn: func(ctx job.JobRuntime) error {
+			return SendWatchdogNotification(ctx.Context.WithSubject(api.SystemUserID.String()), notificationID)
+		},
+	}
+}
+
+// SendWatchdogNotification sends a watchdog notification containing statistics
+// for the specified notification
+func SendWatchdogNotification(ctx context.Context, notificationID string) error {
+	notificationUUID, err := uuid.Parse(notificationID)
+	if err != nil {
+		return fmt.Errorf("failed to parse notification ID: %w", err)
+	}
+
+	notification, err := GetNotification(ctx, notificationID)
+	if err != nil {
+		return fmt.Errorf("failed to get notification: %w", err)
+	}
+
+	if notification.PlaybookID == nil {
+		// NOTE: Watchdog notifications aren't sent to playbook recievers.
+
+		// Manually craft a payload (unlike other payloads that are generated from events)
+		// This allows us to bypass the event queue and process synchronously
+		payload := NotificationEventPayload{
+			ResourceID:     notificationUUID,
+			EventName:      api.EventWatchdog,
+			EventCreatedAt: time.Now(),
+			PersonID:       notification.PersonID,
+			NotificationID: notificationUUID,
+			TeamID:         notification.TeamID,
+			Properties:     fmt.Appendf(nil, `{"id": "%s"}`, notificationUUID.String()),
+		}
+		if len(notification.CustomNotifications) > 0 {
+			payload.CustomService = &notification.CustomNotifications[0]
+		}
+
+		if err := sendNotification(ctx, payload); err != nil {
+			return fmt.Errorf("failed to send watchdog notification to primary recipient: %w", err)
+		}
+	}
+
+	if notification.FallbackPlaybookID == nil {
+		// Also, send to fallback recipient
+		payload := NotificationEventPayload{
+			ResourceID:     notificationUUID,
+			EventName:      api.EventWatchdog,
+			EventCreatedAt: time.Now(),
+			NotificationID: notificationUUID,
+			PersonID:       notification.FallbackPersonID,
+			TeamID:         notification.FallbackTeamID,
+			CustomService:  notification.FallbackCustomNotification,
+			Properties:     fmt.Appendf(nil, `{"id": "%s"}`, notificationUUID.String()),
+		}
+
+		if err := sendNotification(ctx, payload); err != nil {
+			return fmt.Errorf("failed to send watchdog notification to fallback recipient: %w", err)
+		}
+	}
+
+	return nil
+}
diff --git a/notification/message.go b/notification/message.go

Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,9 @@ import (`
`8`	`8`	`"github.com/google/uuid"`
`9`	`9`	`)`
`10`	`10`
	`11`	`+// A dummy event that lives only within the application (never in the event queue)`
	`12`	`+const EventWatchdog = "notification.watchdog"`
	`13`	`+`
`11`	`14`	// List of all sync events in the `event_queue` table.
`12`	`15`	`//`
`13`	`16`	`// These events are generated by the database in response to updates on some of the tables.`
Original file line number	Diff line number	Diff line change
`@@ -76,6 +76,14 @@ func PersistNotificationFromCRD(ctx context.Context, obj *v1.Notification) error`
`76`	`76`	`}`
`77`	`77`	`}`
`78`	`78`
	`79`	`+ if obj.Spec.WatchdogInterval != nil && *obj.Spec.WatchdogInterval != "" {`
	`80`	`+ if parsed, err := text.ParseDuration(*obj.Spec.WatchdogInterval); err != nil {`
	`81`	`+ return err`
	`82`	`+ } else {`
	`83`	`+ dbObj.WatchdogInterval = parsed`
	`84`	`+ }`
	`85`	`+ }`
	`86`	`+`
`79`	`87`	`if len(obj.Spec.GroupBy) > 0 && obj.Spec.WaitFor != nil && *obj.Spec.WaitFor == "" {`
`80`	`88`	`return fmt.Errorf("groupBy provided with an empty waitFor. either remove the groupBy or set a waitFor period")`
`81`	`89`	`}`
Original file line number	Diff line number	Diff line change
`@@ -88,7 +88,6 @@ require (`
`88`	`88`	`go.opentelemetry.io/otel/trace v1.43.0`
`89`	`89`	`golang.org/x/sync v0.20.0`
`90`	`90`	`google.golang.org/genai v1.34.0`
`91`		`- gorm.io/hints v1.1.2`
`92`	`91`	`k8s.io/api v0.35.2`
`93`	`92`	`)`
`94`	`93`
Original file line number	Diff line number	Diff line change
`@@ -110,6 +110,10 @@ func Start(ctx context.Context, mcpServer *server.MCPServer) {`
`110`	`110`	`shutdown.ShutdownAndExit(1, fmt.Sprintf("failed to schedule job SyncCRDStatusJob: %v", err))`
`111`	`111`	`}`
`112`	`112`
	`113`	`+ if err := notification.InitWatchdogNotifications(ctx, FuncScheduler); err != nil {`
	`114`	`+ shutdown.ShutdownAndExit(1, fmt.Sprintf("failed to schedule job InitWatchdogNotifications: %v", err))`
	`115`	`+ }`
	`116`	`+`
`113`	`117`	`if err := notification.InitCRDStatusUpdates(ctx); err != nil {`
`114`	`118`	`logger.Errorf("failed to start notification status update queue: %v", err)`
`115`	`119`	`}`