notification

Documentation

Index

• Variables
• func ChannelConfigMap(ch models.NotificationChannel) (map[string]interface{}, error)
• func DecodePolicyEventTypes(raw json.RawMessage) ([]event.Type, error)
• func RegisterMetrics()
• func ValidChannelTypes() map[models.ChannelType]struct{}
• func ValidEventTypes() map[event.Type]struct{}
• type EmailSender
  • func NewEmailSender() *EmailSender
  • func (s *EmailSender) Send(ctx context.Context, ch models.NotificationChannel, payload Payload) error
• type PagerDutySender
  • func NewPagerDutySender() *PagerDutySender
  • func (s *PagerDutySender) Send(ctx context.Context, ch models.NotificationChannel, payload Payload) error
• type Payload
• type PolicyFilter
• type Sender
• type SlackSender
  • func NewSlackSender() *SlackSender
  • func (s *SlackSender) Send(ctx context.Context, ch models.NotificationChannel, payload Payload) error
• type Subscriber
  • func NewSubscriber(bus event.Bus, db *gorm.DB) *Subscriber
  • func (s *Subscriber) RegisterSender(ct models.ChannelType, sender Sender)
  • func (s *Subscriber) Start(ctx context.Context) error
  • func (s *Subscriber) StartWithReady(ctx context.Context, ready chan<- struct{}) error
• type Watcher
  • func NewWatcher(db *gorm.DB, bus event.Bus, store *event.Store, interval time.Duration) *Watcher
  • func (w *Watcher) Start(ctx context.Context) error
• type WebhookSender
  • func NewWebhookSender() *WebhookSender
  • func (s *WebhookSender) Send(ctx context.Context, ch models.NotificationChannel, payload Payload) error

Variables

var (

	// NotificationSendsTotal counts notification dispatch attempts by channel type and outcome.
	NotificationSendsTotal = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "caesium_notification_sends_total",
			Help: "Total notification dispatch attempts by channel type and outcome.",
		},
		[]string{"channel_type", "status"},
	)

	// NotificationSendDuration tracks how long each send takes.
	NotificationSendDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "caesium_notification_send_duration_seconds",
			Help:    "Duration of notification send operations in seconds.",
			Buckets: []float64{0.01, 0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10},
		},
		[]string{"channel_type"},
	)

	// TaskFailuresTotal counts task failure events observed by the notification subscriber.
	TaskFailuresTotal = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "caesium_task_failures_total",
			Help: "Total task failure events observed.",
		},
		[]string{"job_alias"},
	)

	// RunFailuresTotal counts run failure events observed by the notification subscriber.
	RunFailuresTotal = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "caesium_run_failures_total",
			Help: "Total run failure events observed.",
		},
		[]string{"job_alias"},
	)

	// RunTimeoutsTotal counts run timeout events.
	RunTimeoutsTotal = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "caesium_run_timeouts_total",
			Help: "Total run timeout events observed.",
		},
		[]string{"job_alias"},
	)

	// SLAMissesTotal counts SLA miss events.
	SLAMissesTotal = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "caesium_sla_misses_total",
			Help: "Total SLA miss events observed.",
		},
		[]string{"job_alias"},
	)
)

Functions

func ChannelConfigMap

func ChannelConfigMap(ch models.NotificationChannel) (map[string]interface{}, error)

ChannelConfigMap extracts the channel's config as a raw map.

func DecodePolicyEventTypes

func DecodePolicyEventTypes(raw json.RawMessage) ([]event.Type, error)

DecodePolicyEventTypes parses the JSON event types from a policy.

func RegisterMetrics

func RegisterMetrics()

RegisterMetrics registers all notification metrics with the default Prometheus registry.

func ValidChannelTypes

func ValidChannelTypes() map[models.ChannelType]struct{}

ValidChannelTypes returns the set of valid channel types.

func ValidEventTypes

func ValidEventTypes() map[event.Type]struct{}

ValidEventTypes returns the set of event types valid for notification policies.

Types

type EmailSender

type EmailSender struct{}

EmailSender sends notifications via SMTP email.

func NewEmailSender

func NewEmailSender() *EmailSender

NewEmailSender creates an email notification sender.

func (*EmailSender) Send

func (s *EmailSender) Send(ctx context.Context, ch models.NotificationChannel, payload Payload) error

type PagerDutySender

type PagerDutySender struct {
	// contains filtered or unexported fields
}

PagerDutySender sends notifications via PagerDuty Events API v2.

func NewPagerDutySender

func NewPagerDutySender() *PagerDutySender

NewPagerDutySender creates a PagerDuty notification sender.

func (*PagerDutySender) Send

func (s *PagerDutySender) Send(ctx context.Context, ch models.NotificationChannel, payload Payload) error

type Payload

type Payload struct {
	EventType  event.Type        `json:"event_type"`
	JobID      uuid.UUID         `json:"job_id"`
	JobAlias   string            `json:"job_alias,omitempty"`
	JobLabels  map[string]string `json:"job_labels,omitempty"`
	RunID      uuid.UUID         `json:"run_id"`
	TaskID     uuid.UUID         `json:"task_id,omitempty"`
	Error      string            `json:"error,omitempty"`
	Timestamp  time.Time         `json:"timestamp"`
	RawPayload json.RawMessage   `json:"payload,omitempty"`
}

Payload is the notification content delivered to channels.

type PolicyFilter

type PolicyFilter struct {
	JobIDs   []uuid.UUID       `json:"job_ids,omitempty"`
	JobAlias string            `json:"job_alias,omitempty"`
	Labels   map[string]string `json:"labels,omitempty"`
}

PolicyFilter defines optional filters on a notification policy.

type Sender

type Sender interface {
	Send(ctx context.Context, channel models.NotificationChannel, payload Payload) error
}

Sender delivers a notification payload to a specific channel type. The channel's Config field contains the per-channel configuration (URL, credentials, routing keys, etc.) as JSON.

type SlackSender

type SlackSender struct {
	// contains filtered or unexported fields
}

SlackSender sends notifications via Slack incoming webhooks.

func NewSlackSender

func NewSlackSender() *SlackSender

NewSlackSender creates a Slack notification sender.

func (*SlackSender) Send

func (s *SlackSender) Send(ctx context.Context, ch models.NotificationChannel, payload Payload) error

type Subscriber

type Subscriber struct {
	// contains filtered or unexported fields
}

Subscriber listens to the event bus and dispatches notifications through matching policies and channels.

func NewSubscriber

func NewSubscriber(bus event.Bus, db *gorm.DB) *Subscriber

NewSubscriber creates a notification subscriber.

func (*Subscriber) RegisterSender

func (s *Subscriber) RegisterSender(ct models.ChannelType, sender Sender)

RegisterSender registers a Sender for a given channel type.

func (*Subscriber) Start

func (s *Subscriber) Start(ctx context.Context) error

Start subscribes to notifiable events and dispatches notifications.

func (*Subscriber) StartWithReady

func (s *Subscriber) StartWithReady(ctx context.Context, ready chan<- struct{}) error

StartWithReady subscribes and signals readiness after subscription is established.

type Watcher

type Watcher struct {
	// contains filtered or unexported fields
}

Watcher periodically scans for timeout and SLA violations, publishing the appropriate events on the bus.

func NewWatcher

func NewWatcher(db *gorm.DB, bus event.Bus, store *event.Store, interval time.Duration) *Watcher

NewWatcher creates a new timeout/SLA watcher.

func (*Watcher) Start

func (w *Watcher) Start(ctx context.Context) error

Start runs the watcher loop until the context is cancelled.

type WebhookSender

type WebhookSender struct {
	// contains filtered or unexported fields
}

WebhookSender sends notifications as HTTP requests.

func NewWebhookSender

func NewWebhookSender() *WebhookSender

NewWebhookSender creates a webhook notification sender.

func (*WebhookSender) Send

func (s *WebhookSender) Send(ctx context.Context, ch models.NotificationChannel, payload Payload) error