Permalink
Fetching contributors…
Cannot retrieve contributors at this time
730 lines (621 sloc) 13.8 KB
package sqlite
import (
"database/sql"
"encoding/json"
"fmt"
"sync"
_ "github.com/mattn/go-sqlite3"
"github.com/prometheus/common/log"
"github.com/prometheus/common/model"
"github.com/prometheus/alertmanager/provider"
"github.com/prometheus/alertmanager/types"
)
const createAlertsTable = `
CREATE TABLE IF NOT EXISTS alerts (
id integer PRIMARY KEY AUTOINCREMENT,
fingerprint integer,
labels blob,
annotations blob,
starts_at timestamp,
ends_at timestamp,
updated_at timestamp,
timeout integer
);
CREATE INDEX IF NOT EXISTS fingerprint ON alerts (fingerprint);
CREATE INDEX IF NOT EXISTS alerts_start ON alerts (starts_at);
CREATE INDEX IF NOT EXISTS alerts_end ON alerts (ends_at);
CREATE INDEX IF NOT EXISTS alerts_updated ON alerts (updated_at);
`
var dbmtx sync.Mutex
type Alerts struct {
db *sql.DB
mtx sync.RWMutex
listeners map[int]chan *types.Alert
next int
insertCh chan *types.Alert
}
func NewAlerts(db *sql.DB) (*Alerts, error) {
dbmtx.Lock()
defer dbmtx.Unlock()
tx, err := db.Begin()
if err != nil {
return nil, err
}
if _, err := tx.Exec(createAlertsTable); err != nil {
tx.Rollback()
return nil, err
}
tx.Commit()
alerts := &Alerts{
db: db,
listeners: map[int]chan *types.Alert{},
insertCh: make(chan *types.Alert, 100),
}
return alerts, nil
}
// Subscribe implements the Alerts interface.
func (a *Alerts) Subscribe() provider.AlertIterator {
var (
ch = make(chan *types.Alert, 200)
done = make(chan struct{})
)
alerts, err := a.getPending()
i := a.next
a.next++
a.mtx.Lock()
a.listeners[i] = ch
a.mtx.Unlock()
go func() {
defer func() {
a.mtx.Lock()
delete(a.listeners, i)
close(ch)
a.mtx.Unlock()
}()
for _, a := range alerts {
select {
case ch <- a:
case <-done:
return
}
}
<-done
}()
return provider.NewAlertIterator(ch, done, err)
}
// GetPending implements the Alerts interface.
func (a *Alerts) GetPending() provider.AlertIterator {
var (
ch = make(chan *types.Alert, 200)
done = make(chan struct{})
)
alerts, err := a.getPending()
go func() {
defer close(ch)
for _, a := range alerts {
select {
case ch <- a:
case <-done:
return
}
}
}()
return provider.NewAlertIterator(ch, done, err)
}
func (a *Alerts) getPending() ([]*types.Alert, error) {
dbmtx.Lock()
defer dbmtx.Unlock()
// Get the last instance for each alert.
rows, err := a.db.Query(`
SELECT a1.labels, a1.annotations, a1.starts_at, a1.ends_at, a1.updated_at, a1.timeout
FROM alerts AS a1
LEFT OUTER JOIN alerts AS a2
ON a1.fingerprint = a2.fingerprint AND a1.updated_at < a2.updated_at
WHERE a2.fingerprint IS NULL;
`)
if err != nil {
return nil, err
}
var alerts []*types.Alert
for rows.Next() {
var (
labels []byte
annotations []byte
al types.Alert
)
if err := rows.Scan(
&labels,
&annotations,
&al.StartsAt,
&al.EndsAt,
&al.UpdatedAt,
&al.Timeout,
); err != nil {
return nil, err
}
if err := json.Unmarshal(labels, &al.Labels); err != nil {
return nil, err
}
if err := json.Unmarshal(annotations, &al.Annotations); err != nil {
return nil, err
}
alerts = append(alerts, &al)
}
if err := rows.Err(); err != nil {
return nil, err
}
return alerts, nil
}
// GetPending implements the Alerts interface.
func (a *Alerts) GetAll() provider.AlertIterator {
var (
ch = make(chan *types.Alert, 200)
done = make(chan struct{})
)
alerts, err := a.getAll()
go func() {
defer close(ch)
for _, a := range alerts {
select {
case ch <- a:
case <-done:
return
}
}
}()
return provider.NewAlertIterator(ch, done, err)
}
func (a *Alerts) getAll() ([]*types.Alert, error) {
dbmtx.Lock()
defer dbmtx.Unlock()
// Get the last instance for each alert.
rows, err := a.db.Query(`
SELECT a1.updated_at, a1.labels, a1.annotations, a1.starts_at, a1.ends_at, a1.updated_at, a1.timeout
FROM alerts AS a1
ORDER BY a1.updated_at DESC;
`)
if err != nil {
return nil, err
}
var alerts []*types.Alert
for rows.Next() {
var (
labels []byte
annotations []byte
al types.Alert
)
if err := rows.Scan(
&labels,
&annotations,
&al.StartsAt,
&al.EndsAt,
&al.UpdatedAt,
&al.Timeout,
); err != nil {
return nil, err
}
if err := json.Unmarshal(labels, &al.Labels); err != nil {
return nil, err
}
if err := json.Unmarshal(annotations, &al.Annotations); err != nil {
return nil, err
}
alerts = append(alerts, &al)
}
if err := rows.Err(); err != nil {
return nil, err
}
return alerts, nil
}
// Get implements the Alerts interface.
func (a *Alerts) Get(model.Fingerprint) (*types.Alert, error) {
return nil, nil
}
// Put implements the Alerts interface.
func (a *Alerts) Put(alerts ...*types.Alert) error {
dbmtx.Lock()
defer dbmtx.Unlock()
tx, err := a.db.Begin()
if err != nil {
return err
}
// The insert invariant requires that there are no two alerts with the same
// fingerprint that have overlapping activity range ([StartsAt:EndsAt]).
// Such alerts are merged into a single one with the union of both intervals
// as its new activity interval.
// The exact merge procedure is defined on the Alert structure. Here, we just
// care about finding intersecting alerts for each new inserts, deleting them
// if existant, and insert the new alert we retrieved by merging.
overlap, err := tx.Prepare(`
SELECT id, annotations, starts_at, ends_at, updated_at, timeout
FROM alerts
WHERE fingerprint == $1 AND (
(starts_at <= $2 AND ends_at >= $2) OR
(starts_at <= $3 AND ends_at >= $3)
)
`)
if err != nil {
tx.Rollback()
return err
}
defer overlap.Close()
delOverlap, err := tx.Prepare(`
DELETE FROM alerts WHERE id IN (
SELECT id FROM alerts
WHERE fingerprint == $1 AND (
(starts_at <= $2 AND ends_at >= $2) OR
(starts_at <= $3 AND ends_at >= $3)
)
)
`)
if err != nil {
tx.Rollback()
return err
}
defer delOverlap.Close()
insert, err := tx.Prepare(`
INSERT INTO alerts(fingerprint, labels, annotations, starts_at, ends_at, updated_at, timeout)
VALUES ($1, $2, $3, $4, $5, $6, $7)
`)
if err != nil {
tx.Rollback()
return err
}
defer insert.Close()
for _, alert := range alerts {
fp := alert.Fingerprint()
// Retrieve all intersecting alerts and delete them.
olaps, err := overlap.Query(int64(fp), alert.StartsAt, alert.EndsAt)
if err != nil {
tx.Rollback()
return err
}
var (
overlapIDs []int64
merges []*types.Alert
)
for olaps.Next() {
var (
id int64
na types.Alert
ann []byte
)
if err := olaps.Scan(
&id,
&ann,
&na.StartsAt,
&na.EndsAt,
&na.UpdatedAt,
&na.Timeout,
); err != nil {
tx.Rollback()
return err
}
if err := json.Unmarshal(ann, &na.Annotations); err != nil {
tx.Rollback()
return err
}
na.Labels = alert.Labels
merges = append(merges, &na)
overlapIDs = append(overlapIDs, id)
}
if err := olaps.Err(); err != nil {
tx.Rollback()
return err
}
// Merge them.
for _, ma := range merges {
alert = alert.Merge(ma)
}
// Delete the old ones.
if _, err := delOverlap.Exec(int64(fp), alert.StartsAt, alert.EndsAt); err != nil {
tx.Rollback()
return err
}
// Insert the final alert.
labels, err := json.Marshal(alert.Labels)
if err != nil {
tx.Rollback()
return err
}
annotations, err := json.Marshal(alert.Annotations)
if err != nil {
tx.Rollback()
return err
}
_, err = insert.Exec(
int64(fp),
labels,
annotations,
alert.StartsAt,
alert.EndsAt,
alert.UpdatedAt,
alert.Timeout,
)
if err != nil {
tx.Rollback()
return err
}
a.mtx.RLock()
for _, ch := range a.listeners {
ch <- alert
}
a.mtx.RUnlock()
}
tx.Commit()
return nil
}
const createNotifyInfoTable = `
CREATE TABLE IF NOT EXISTS notify_info (
alert bigint,
receiver text,
resolved integer,
timestamp timestamp
);
CREATE INDEX IF NOT EXISTS notify_done ON notify_info (resolved);
CREATE UNIQUE INDEX IF NOT EXISTS alert_receiver ON notify_info (alert,receiver);
`
type Notifies struct {
db *sql.DB
}
func NewNotifies(db *sql.DB) (*Notifies, error) {
dbmtx.Lock()
defer dbmtx.Unlock()
tx, err := db.Begin()
if err != nil {
return nil, err
}
if _, err := tx.Exec(createNotifyInfoTable); err != nil {
tx.Rollback()
return nil, err
}
tx.Commit()
return &Notifies{db: db}, nil
}
// Get implements the Notifies interface.
func (n *Notifies) Get(dest string, fps ...model.Fingerprint) ([]*types.NotifyInfo, error) {
dbmtx.Lock()
defer dbmtx.Unlock()
var result []*types.NotifyInfo
for _, fp := range fps {
row := n.db.QueryRow(`
SELECT alert, receiver, resolved, timestamp
FROM notify_info
WHERE receiver == $1 AND alert == $2
`, dest, int64(fp))
var alertFP int64
var ni types.NotifyInfo
err := row.Scan(
&alertFP,
&ni.Receiver,
&ni.Resolved,
&ni.Timestamp,
)
if err == sql.ErrNoRows {
result = append(result, nil)
continue
}
if err != nil {
return nil, err
}
ni.Alert = model.Fingerprint(alertFP)
result = append(result, &ni)
}
return result, nil
}
// Set implements the Notifies interface.
func (n *Notifies) Set(ns ...*types.NotifyInfo) error {
dbmtx.Lock()
defer dbmtx.Unlock()
tx, err := n.db.Begin()
if err != nil {
return err
}
insert, err := tx.Prepare(`
INSERT INTO notify_info(alert, receiver, resolved, timestamp)
VALUES ($1, $2, $3, $4);
`)
if err != nil {
tx.Rollback()
return err
}
defer insert.Close()
del, err := tx.Prepare(`
DELETE FROM notify_info
WHERE alert == $1 AND receiver == $2
`)
if err != nil {
tx.Rollback()
return err
}
defer del.Close()
for _, ni := range ns {
if _, err := del.Exec(int64(ni.Alert), ni.Receiver); err != nil {
tx.Rollback()
return fmt.Errorf("deleting old notify failed: %s", err)
}
if _, err := insert.Exec(
int64(ni.Alert),
ni.Receiver,
ni.Resolved,
ni.Timestamp,
); err != nil {
tx.Rollback()
return fmt.Errorf("inserting new notify failed: %s", err)
}
}
tx.Commit()
return nil
}
const createSilencesTable = `
CREATE TABLE IF NOT EXISTS silences (
id integer PRIMARY KEY AUTOINCREMENT,
matchers blob,
starts_at timestamp,
ends_at timestamp,
created_at timestamp,
created_by text,
comment text
);
CREATE INDEX IF NOT EXISTS silences_start ON silences (starts_at);
CREATE INDEX IF NOT EXISTS silences_end ON silences (ends_at);
`
type Silences struct {
db *sql.DB
marker types.Marker
}
// NewSilences returns a new Silences based on the provided SQL DB.
func NewSilences(db *sql.DB, mk types.Marker) (*Silences, error) {
dbmtx.Lock()
defer dbmtx.Unlock()
tx, err := db.Begin()
if err != nil {
return nil, err
}
if _, err := tx.Exec(createSilencesTable); err != nil {
tx.Rollback()
return nil, err
}
tx.Commit()
return &Silences{db: db, marker: mk}, nil
}
// Mutes implements the Muter interface.
func (s *Silences) Mutes(lset model.LabelSet) bool {
sils, err := s.All()
if err != nil {
log.Errorf("retrieving silences failed: %s", err)
// In doubt, do not silence anything.
return false
}
for _, sil := range sils {
if sil.Mutes(lset) {
s.marker.SetSilenced(lset.Fingerprint(), sil.ID)
return true
}
}
s.marker.SetSilenced(lset.Fingerprint())
return false
}
// All implements the Silences interface.
func (s *Silences) All() ([]*types.Silence, error) {
dbmtx.Lock()
defer dbmtx.Unlock()
rows, err := s.db.Query(`
SELECT id, matchers, starts_at, ends_at, created_at, created_by, comment
FROM silences
ORDER BY starts_at DESC
`)
if err != nil {
return nil, err
}
defer rows.Close()
var silences []*types.Silence
for rows.Next() {
var (
sil model.Silence
matchers []byte
)
if err := rows.Scan(
&sil.ID,
&matchers,
&sil.StartsAt,
&sil.EndsAt,
&sil.CreatedAt,
&sil.CreatedBy,
&sil.Comment,
); err != nil {
return nil, err
}
if err := json.Unmarshal(matchers, &sil.Matchers); err != nil {
return nil, err
}
silences = append(silences, types.NewSilence(&sil))
}
if err := rows.Err(); err != nil {
return nil, err
}
return silences, nil
}
// Set impelements the Silences interface.
func (s *Silences) Set(sil *types.Silence) (uint64, error) {
dbmtx.Lock()
defer dbmtx.Unlock()
mb, err := json.Marshal(sil.Silence.Matchers)
if err != nil {
return 0, err
}
tx, err := s.db.Begin()
if err != nil {
return 0, err
}
res, err := tx.Exec(`
INSERT INTO silences(matchers, starts_at, ends_at, created_at, created_by, comment)
VALUES ($1, $2, $3, $4, $5, $6)
`,
mb,
sil.StartsAt,
sil.EndsAt,
sil.CreatedAt,
sil.CreatedBy,
sil.Comment,
)
if err != nil {
tx.Rollback()
return 0, err
}
sid, err := res.LastInsertId()
if err != nil {
tx.Rollback()
return 0, err
}
tx.Commit()
return uint64(sid), nil
}
// Del implements the Silences interface.
func (s *Silences) Del(sid uint64) error {
dbmtx.Lock()
defer dbmtx.Unlock()
tx, err := s.db.Begin()
if err != nil {
return err
}
if _, err := tx.Exec(`DELETE FROM silences WHERE id == $1`, sid); err != nil {
tx.Rollback()
return err
}
tx.Commit()
return nil
}
// Get implements the Silences interface.
func (s *Silences) Get(sid uint64) (*types.Silence, error) {
dbmtx.Lock()
defer dbmtx.Unlock()
row := s.db.QueryRow(`
SELECT id, matchers, starts_at, ends_at, created_at, created_by, comment
FROM silences
WHERE id == $1
`, sid)
var (
sil model.Silence
matchers []byte
)
err := row.Scan(
&sil.ID,
&matchers,
&sil.StartsAt,
&sil.EndsAt,
&sil.CreatedAt,
&sil.CreatedBy,
&sil.Comment,
)
if err == sql.ErrNoRows {
return nil, provider.ErrNotFound
}
if err != nil {
return nil, err
}
if err := json.Unmarshal(matchers, &sil.Matchers); err != nil {
return nil, err
}
return types.NewSilence(&sil), nil
}