Documentation
¶
Index ¶
- Variables
- func ChannelConfigMap(ch models.NotificationChannel) (map[string]any, error)
- func DecodePolicyEventTypes(raw json.RawMessage) ([]event.Type, error)
- func MatchPolicies(ctx context.Context, db *gorm.DB, evt event.Event) ([]models.NotificationPolicy, error)
- func RegisterMetrics()
- func ValidChannelTypes() map[models.ChannelType]struct{}
- func ValidEventTypes() map[event.Type]struct{}
- type AIAgentLeaderCheck
- type AIAgentSender
- type EmailSender
- type PagerDutySender
- type Payload
- type PolicyFilter
- type Sender
- type SlackSender
- type Subscriber
- type Watcher
- type WebhookSender
Constants ¶
This section is empty.
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]any, 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 MatchPolicies ¶
func MatchPolicies(ctx context.Context, db *gorm.DB, evt event.Event) ([]models.NotificationPolicy, error)
MatchPolicies returns the enabled policies that would deliver evt.
Exported because delivery is not always fire-and-forget: an escalation must be able to report whether anything will actually route it, rather than recording itself as delivered into a deployment with no matching policy (cmd/start/incident_ops.go Escalate). Sharing this function is what keeps that answer honest — a separate re-implementation would drift from the filters the subscriber actually applies.
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 ¶
ValidEventTypes returns the set of event types valid for notification policies.
Types ¶
type AIAgentLeaderCheck ¶
AIAgentLeaderCheck reports whether this node hosts the cluster leader. The ai_agent sender is leader-gated so that — unlike the per-node notification subscriber it runs inside — a matched policy opens exactly one incident per failure on an N-node cluster (the incident store's atomic conditional insert is the backstop; the leader gate avoids the wasted work). A nil check means "always act" (single-node / tests).
type AIAgentSender ¶
type AIAgentSender struct {
// contains filtered or unexported fields
}
AIAgentSender makes the reserved ChannelTypeAIAgent = "ai_agent" a real dispatch path (agent-in-the-loop D3). It is a second, policy-driven route into the incident manager: the same NotificationPolicy matching that fans a failure event out to Slack can fan it to the agent. Rather than re-publishing onto the bus (which the leader-gated incident subscriber already consumes — and which would risk a match/republish loop), the sender opens/appends the incident directly via the shared incident store, converging on the same pipeline as the job-owner-facing metadata.remediation opt-in.
func NewAIAgentSender ¶
func NewAIAgentSender(db *gorm.DB, leaderCheck AIAgentLeaderCheck, cooldown time.Duration) *AIAgentSender
NewAIAgentSender constructs an ai_agent sender over db. leaderCheck may be nil (single-node / tests), in which case the sender always acts. cooldown is the configured CAESIUM_AGENT_INCIDENT_COOLDOWN so flap suppression applies to this dispatch path too, matching the main subscriber.
func (*AIAgentSender) Send ¶
func (s *AIAgentSender) Send(ctx context.Context, ch models.NotificationChannel, payload Payload) error
Send routes a matched FAILURE event into the incident manager by opening or appending an incident for the failing job/task. Success events (a policy could fan run_completed / task_succeeded to an ai_agent channel) are skipped — they must never manufacture an incident for a healthy run.
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 ¶
NewWatcher creates a new timeout/SLA watcher.
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
Source Files