From a95c9983004db5b06b81bd41eecb6309a3436f46 Mon Sep 17 00:00:00 2001 From: Chris Goller Date: Thu, 3 Nov 2016 19:44:28 -0500 Subject: [PATCH] Update to store alert information in boltdb --- bolt/alerts.go | 115 +++++++++++++++++++++++++ bolt/client.go | 15 ++-- bolt/internal/internal.go | 27 ++++++ bolt/internal/internal.pb.go | 127 +++++++++++++++------------ bolt/internal/internal.proto | 5 ++ chronograf.go | 32 ++----- kapacitor/alerts.go | 27 +++++- kapacitor/alerts_test.go | 2 +- kapacitor/tickscripts_test.go | 51 +++++------ kapacitor/triggers.go | 4 +- kapacitor/triggers_test.go | 8 +- kapacitor/vars.go | 12 +-- server/kapacitors.go | 158 +++++++++++++++++++++++++++++----- server/server.go | 1 + server/service.go | 1 + 15 files changed, 427 insertions(+), 158 deletions(-) create mode 100644 bolt/alerts.go diff --git a/bolt/alerts.go b/bolt/alerts.go new file mode 100644 index 0000000000..d24abb6b03 --- /dev/null +++ b/bolt/alerts.go @@ -0,0 +1,115 @@ +package bolt + +import ( + "context" + + "github.com/boltdb/bolt" + "github.com/influxdata/chronograf" + "github.com/influxdata/chronograf/bolt/internal" +) + +// Ensure AlertsStore implements chronograf.AlertsStore. +var _ chronograf.AlertRulesStore = &AlertsStore{} + +var AlertsBucket = []byte("Alerts") + +type AlertsStore struct { + client *Client +} + +// All returns all known alerts +func (s *AlertsStore) All(ctx context.Context) ([]chronograf.AlertRule, error) { + var srcs []chronograf.AlertRule + if err := s.client.db.View(func(tx *bolt.Tx) error { + if err := tx.Bucket(AlertsBucket).ForEach(func(k, v []byte) error { + var src chronograf.AlertRule + if err := internal.UnmarshalAlertRule(v, &src); err != nil { + return err + } + srcs = append(srcs, src) + return nil + }); err != nil { + return err + } + return nil + }); err != nil { + return nil, err + } + + return srcs, nil + +} + +// Add creates a new Alerts in the AlertsStore. +func (s *AlertsStore) Add(ctx context.Context, src chronograf.AlertRule) (chronograf.AlertRule, error) { + if err := s.client.db.Update(func(tx *bolt.Tx) error { + b := tx.Bucket(AlertsBucket) + if v, err := internal.MarshalAlertRule(&src); err != nil { + return err + } else if err := b.Put([]byte(src.ID), v); err != nil { + return err + } + return nil + }); err != nil { + return chronograf.AlertRule{}, err + } + + return src, nil +} + +// Delete removes the Alerts from the AlertsStore +func (s *AlertsStore) Delete(ctx context.Context, src chronograf.AlertRule) error { + _, err := s.Get(ctx, src.ID) + if err != nil { + return err + } + if err := s.client.db.Update(func(tx *bolt.Tx) error { + if err := tx.Bucket(AlertsBucket).Delete([]byte(src.ID)); err != nil { + return err + } + return nil + }); err != nil { + return err + } + + return nil +} + +// Get returns a Alerts if the id exists. +func (s *AlertsStore) Get(ctx context.Context, id string) (chronograf.AlertRule, error) { + var src chronograf.AlertRule + if err := s.client.db.View(func(tx *bolt.Tx) error { + if v := tx.Bucket(AlertsBucket).Get([]byte(id)); v == nil { + return chronograf.ErrAlertNotFound + } else if err := internal.UnmarshalAlertRule(v, &src); err != nil { + return err + } + return nil + }); err != nil { + return chronograf.AlertRule{}, err + } + + return src, nil +} + +// Update a Alerts +func (s *AlertsStore) Update(ctx context.Context, src chronograf.AlertRule) error { + if err := s.client.db.Update(func(tx *bolt.Tx) error { + // Get an existing alerts with the same ID. + b := tx.Bucket(AlertsBucket) + if v := b.Get([]byte(src.ID)); v == nil { + return chronograf.ErrAlertNotFound + } + + if v, err := internal.MarshalAlertRule(&src); err != nil { + return err + } else if err := b.Put([]byte(src.ID), v); err != nil { + return err + } + return nil + }); err != nil { + return err + } + + return nil +} diff --git a/bolt/client.go b/bolt/client.go index 542d82905e..93f32d9002 100644 --- a/bolt/client.go +++ b/bolt/client.go @@ -19,6 +19,7 @@ type Client struct { SourcesStore *SourcesStore ServersStore *ServersStore LayoutStore *LayoutStore + AlertsStore *AlertsStore } func NewClient() *Client { @@ -26,6 +27,7 @@ func NewClient() *Client { c.ExplorationStore = &ExplorationStore{client: c} c.SourcesStore = &SourcesStore{client: c} c.ServersStore = &ServersStore{client: c} + c.AlertsStore = &AlertsStore{client: c} c.LayoutStore = &LayoutStore{ client: c, IDs: &uuid.V4{}, @@ -59,20 +61,15 @@ func (c *Client) Open() error { if _, err := tx.CreateBucketIfNotExists(LayoutBucket); err != nil { return err } - + // Always create Alerts bucket. + if _, err := tx.CreateBucketIfNotExists(AlertsBucket); err != nil { + return err + } return nil }); err != nil { return err } - // TODO: Ask @gunnar about these - /* - c.ExplorationStore = &ExplorationStore{client: c} - c.SourcesStore = &SourcesStore{client: c} - c.ServersStore = &ServersStore{client: c} - c.LayoutStore = &LayoutStore{client: c} - */ - return nil } diff --git a/bolt/internal/internal.go b/bolt/internal/internal.go index c48e373929..599214daa3 100644 --- a/bolt/internal/internal.go +++ b/bolt/internal/internal.go @@ -1,6 +1,7 @@ package internal import ( + "encoding/json" "time" "github.com/gogo/protobuf/proto" @@ -161,3 +162,29 @@ func UnmarshalLayout(data []byte, l *chronograf.Layout) error { l.Cells = cells return nil } + +// MarshalAlertRule encodes an alert rule to binary protobuf format. +func MarshalAlertRule(r *chronograf.AlertRule) ([]byte, error) { + j, err := json.Marshal(r) + if err != nil { + return nil, err + } + return proto.Marshal(&AlertRule{ + ID: r.ID, + JSON: string(j), + }) +} + +// UnmarshalAlertRule decodes an alert rule from binary protobuf data. +func UnmarshalAlertRule(data []byte, r *chronograf.AlertRule) error { + var pb AlertRule + if err := proto.Unmarshal(data, &pb); err != nil { + return err + } + + err := json.Unmarshal([]byte(pb.JSON), r) + if err != nil { + return err + } + return nil +} diff --git a/bolt/internal/internal.pb.go b/bolt/internal/internal.pb.go index a71075cc4f..ea37458e13 100644 --- a/bolt/internal/internal.pb.go +++ b/bolt/internal/internal.pb.go @@ -15,6 +15,7 @@ It has these top-level messages: Layout Cell Query + AlertRule */ package internal @@ -34,13 +35,13 @@ var _ = math.Inf const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package type Exploration struct { - ID int64 `protobuf:"varint,1,opt,name=ID,proto3" json:"ID,omitempty"` - Name string `protobuf:"bytes,2,opt,name=Name,proto3" json:"Name,omitempty"` - UserID int64 `protobuf:"varint,3,opt,name=UserID,proto3" json:"UserID,omitempty"` - Data string `protobuf:"bytes,4,opt,name=Data,proto3" json:"Data,omitempty"` - CreatedAt int64 `protobuf:"varint,5,opt,name=CreatedAt,proto3" json:"CreatedAt,omitempty"` - UpdatedAt int64 `protobuf:"varint,6,opt,name=UpdatedAt,proto3" json:"UpdatedAt,omitempty"` - Default bool `protobuf:"varint,7,opt,name=Default,proto3" json:"Default,omitempty"` + ID int64 `protobuf:"varint,1,opt,name=ID,json=iD,proto3" json:"ID,omitempty"` + Name string `protobuf:"bytes,2,opt,name=Name,json=name,proto3" json:"Name,omitempty"` + UserID int64 `protobuf:"varint,3,opt,name=UserID,json=userID,proto3" json:"UserID,omitempty"` + Data string `protobuf:"bytes,4,opt,name=Data,json=data,proto3" json:"Data,omitempty"` + CreatedAt int64 `protobuf:"varint,5,opt,name=CreatedAt,json=createdAt,proto3" json:"CreatedAt,omitempty"` + UpdatedAt int64 `protobuf:"varint,6,opt,name=UpdatedAt,json=updatedAt,proto3" json:"UpdatedAt,omitempty"` + Default bool `protobuf:"varint,7,opt,name=Default,json=default,proto3" json:"Default,omitempty"` } func (m *Exploration) Reset() { *m = Exploration{} } @@ -49,13 +50,13 @@ func (*Exploration) ProtoMessage() {} func (*Exploration) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{0} } type Source struct { - ID int64 `protobuf:"varint,1,opt,name=ID,proto3" json:"ID,omitempty"` - Name string `protobuf:"bytes,2,opt,name=Name,proto3" json:"Name,omitempty"` - Type string `protobuf:"bytes,3,opt,name=Type,proto3" json:"Type,omitempty"` - Username string `protobuf:"bytes,4,opt,name=Username,proto3" json:"Username,omitempty"` - Password string `protobuf:"bytes,5,opt,name=Password,proto3" json:"Password,omitempty"` - URL string `protobuf:"bytes,6,opt,name=URL,proto3" json:"URL,omitempty"` - Default bool `protobuf:"varint,7,opt,name=Default,proto3" json:"Default,omitempty"` + ID int64 `protobuf:"varint,1,opt,name=ID,json=iD,proto3" json:"ID,omitempty"` + Name string `protobuf:"bytes,2,opt,name=Name,json=name,proto3" json:"Name,omitempty"` + Type string `protobuf:"bytes,3,opt,name=Type,json=type,proto3" json:"Type,omitempty"` + Username string `protobuf:"bytes,4,opt,name=Username,json=username,proto3" json:"Username,omitempty"` + Password string `protobuf:"bytes,5,opt,name=Password,json=password,proto3" json:"Password,omitempty"` + URL string `protobuf:"bytes,6,opt,name=URL,json=uRL,proto3" json:"URL,omitempty"` + Default bool `protobuf:"varint,7,opt,name=Default,json=default,proto3" json:"Default,omitempty"` } func (m *Source) Reset() { *m = Source{} } @@ -64,12 +65,12 @@ func (*Source) ProtoMessage() {} func (*Source) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{1} } type Server struct { - ID int64 `protobuf:"varint,1,opt,name=ID,proto3" json:"ID,omitempty"` - Name string `protobuf:"bytes,2,opt,name=Name,proto3" json:"Name,omitempty"` - Username string `protobuf:"bytes,3,opt,name=Username,proto3" json:"Username,omitempty"` - Password string `protobuf:"bytes,4,opt,name=Password,proto3" json:"Password,omitempty"` - URL string `protobuf:"bytes,5,opt,name=URL,proto3" json:"URL,omitempty"` - SrcID int64 `protobuf:"varint,6,opt,name=SrcID,proto3" json:"SrcID,omitempty"` + ID int64 `protobuf:"varint,1,opt,name=ID,json=iD,proto3" json:"ID,omitempty"` + Name string `protobuf:"bytes,2,opt,name=Name,json=name,proto3" json:"Name,omitempty"` + Username string `protobuf:"bytes,3,opt,name=Username,json=username,proto3" json:"Username,omitempty"` + Password string `protobuf:"bytes,4,opt,name=Password,json=password,proto3" json:"Password,omitempty"` + URL string `protobuf:"bytes,5,opt,name=URL,json=uRL,proto3" json:"URL,omitempty"` + SrcID int64 `protobuf:"varint,6,opt,name=SrcID,json=srcID,proto3" json:"SrcID,omitempty"` } func (m *Server) Reset() { *m = Server{} } @@ -78,10 +79,10 @@ func (*Server) ProtoMessage() {} func (*Server) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{2} } type Layout struct { - ID string `protobuf:"bytes,1,opt,name=ID,proto3" json:"ID,omitempty"` - Application string `protobuf:"bytes,2,opt,name=Application,proto3" json:"Application,omitempty"` - Measurement string `protobuf:"bytes,3,opt,name=Measurement,proto3" json:"Measurement,omitempty"` - Cells []*Cell `protobuf:"bytes,4,rep,name=Cells" json:"Cells,omitempty"` + ID string `protobuf:"bytes,1,opt,name=ID,json=iD,proto3" json:"ID,omitempty"` + Application string `protobuf:"bytes,2,opt,name=Application,json=application,proto3" json:"Application,omitempty"` + Measurement string `protobuf:"bytes,3,opt,name=Measurement,json=measurement,proto3" json:"Measurement,omitempty"` + Cells []*Cell `protobuf:"bytes,4,rep,name=Cells,json=cells" json:"Cells,omitempty"` } func (m *Layout) Reset() { *m = Layout{} } @@ -119,9 +120,9 @@ func (m *Cell) GetQueries() []*Query { } type Query struct { - Command string `protobuf:"bytes,1,opt,name=Command,proto3" json:"Command,omitempty"` - DB string `protobuf:"bytes,2,opt,name=DB,proto3" json:"DB,omitempty"` - RP string `protobuf:"bytes,3,opt,name=RP,proto3" json:"RP,omitempty"` + Command string `protobuf:"bytes,1,opt,name=Command,json=command,proto3" json:"Command,omitempty"` + DB string `protobuf:"bytes,2,opt,name=DB,json=dB,proto3" json:"DB,omitempty"` + RP string `protobuf:"bytes,3,opt,name=RP,json=rP,proto3" json:"RP,omitempty"` } func (m *Query) Reset() { *m = Query{} } @@ -129,6 +130,16 @@ func (m *Query) String() string { return proto.CompactTextString(m) } func (*Query) ProtoMessage() {} func (*Query) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{5} } +type AlertRule struct { + ID string `protobuf:"bytes,1,opt,name=ID,json=iD,proto3" json:"ID,omitempty"` + JSON string `protobuf:"bytes,2,opt,name=JSON,json=jSON,proto3" json:"JSON,omitempty"` +} + +func (m *AlertRule) Reset() { *m = AlertRule{} } +func (m *AlertRule) String() string { return proto.CompactTextString(m) } +func (*AlertRule) ProtoMessage() {} +func (*AlertRule) Descriptor() ([]byte, []int) { return fileDescriptorInternal, []int{6} } + func init() { proto.RegisterType((*Exploration)(nil), "internal.Exploration") proto.RegisterType((*Source)(nil), "internal.Source") @@ -136,38 +147,42 @@ func init() { proto.RegisterType((*Layout)(nil), "internal.Layout") proto.RegisterType((*Cell)(nil), "internal.Cell") proto.RegisterType((*Query)(nil), "internal.Query") + proto.RegisterType((*AlertRule)(nil), "internal.AlertRule") } func init() { proto.RegisterFile("internal.proto", fileDescriptorInternal) } var fileDescriptorInternal = []byte{ - // 442 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x8c, 0x93, 0xcf, 0x8e, 0xd3, 0x30, - 0x10, 0xc6, 0xe5, 0x24, 0x4e, 0x9b, 0x29, 0x2a, 0xc8, 0x42, 0xc8, 0x42, 0x1c, 0xa2, 0x88, 0x43, - 0xb9, 0xec, 0x01, 0x9e, 0xa0, 0xdb, 0x70, 0xa8, 0xb4, 0xa0, 0xe2, 0xa5, 0x0f, 0x60, 0x5a, 0xa3, - 0x8d, 0x94, 0x26, 0xc1, 0x71, 0x68, 0x73, 0xe5, 0x0a, 0x8f, 0xc1, 0x1b, 0xf0, 0x82, 0x68, 0x26, - 0xee, 0x9f, 0xc3, 0x6a, 0xd5, 0xdb, 0x7c, 0x33, 0x5f, 0x34, 0x3f, 0x7f, 0x76, 0x60, 0x5a, 0x54, - 0xce, 0xd8, 0x4a, 0x97, 0x37, 0x8d, 0xad, 0x5d, 0x2d, 0xc6, 0x47, 0x9d, 0xfd, 0x63, 0x30, 0xf9, - 0x78, 0x68, 0xca, 0xda, 0x6a, 0x57, 0xd4, 0x95, 0x98, 0x42, 0xb0, 0xcc, 0x25, 0x4b, 0xd9, 0x2c, - 0x54, 0xc1, 0x32, 0x17, 0x02, 0xa2, 0xcf, 0x7a, 0x67, 0x64, 0x90, 0xb2, 0x59, 0xa2, 0xa8, 0x16, - 0xaf, 0x20, 0x5e, 0xb7, 0xc6, 0x2e, 0x73, 0x19, 0x92, 0xcf, 0x2b, 0xf4, 0xe6, 0xda, 0x69, 0x19, - 0x0d, 0x5e, 0xac, 0xc5, 0x1b, 0x48, 0x16, 0xd6, 0x68, 0x67, 0xb6, 0x73, 0x27, 0x39, 0xd9, 0xcf, - 0x0d, 0x9c, 0xae, 0x9b, 0xad, 0x9f, 0xc6, 0xc3, 0xf4, 0xd4, 0x10, 0x12, 0x46, 0xb9, 0xf9, 0xae, - 0xbb, 0xd2, 0xc9, 0x51, 0xca, 0x66, 0x63, 0x75, 0x94, 0xd9, 0x5f, 0x06, 0xf1, 0x7d, 0xdd, 0xd9, - 0x8d, 0xb9, 0x0a, 0x58, 0x40, 0xf4, 0xb5, 0x6f, 0x0c, 0xe1, 0x26, 0x8a, 0x6a, 0xf1, 0x1a, 0xc6, - 0x88, 0x5d, 0xa1, 0x77, 0x00, 0x3e, 0x69, 0x9c, 0xad, 0x74, 0xdb, 0xee, 0x6b, 0xbb, 0x25, 0xe6, - 0x44, 0x9d, 0xb4, 0x78, 0x01, 0xe1, 0x5a, 0xdd, 0x11, 0x6c, 0xa2, 0xb0, 0x7c, 0x02, 0xf3, 0x0f, - 0x62, 0x1a, 0xfb, 0xd3, 0xd8, 0xab, 0x30, 0x2f, 0x91, 0xc2, 0x27, 0x90, 0xa2, 0xc7, 0x91, 0xf8, - 0x19, 0xe9, 0x25, 0xf0, 0x7b, 0xbb, 0x59, 0xe6, 0x3e, 0xd3, 0x41, 0x64, 0xbf, 0x18, 0xc4, 0x77, - 0xba, 0xaf, 0x3b, 0x77, 0x81, 0x93, 0x10, 0x4e, 0x0a, 0x93, 0x79, 0xd3, 0x94, 0xc5, 0x86, 0x5e, - 0x81, 0xa7, 0xba, 0x6c, 0xa1, 0xe3, 0x93, 0xd1, 0x6d, 0x67, 0xcd, 0xce, 0x54, 0xce, 0xf3, 0x5d, - 0xb6, 0xc4, 0x5b, 0xe0, 0x0b, 0x53, 0x96, 0xad, 0x8c, 0xd2, 0x70, 0x36, 0x79, 0x3f, 0xbd, 0x39, - 0x3d, 0x3a, 0x6c, 0xab, 0x61, 0x98, 0xfd, 0x66, 0x10, 0x61, 0x25, 0x9e, 0x01, 0x3b, 0x10, 0x01, - 0x57, 0xec, 0x80, 0xaa, 0xa7, 0xb5, 0x5c, 0xb1, 0x1e, 0xd5, 0x9e, 0x56, 0x70, 0xc5, 0xf6, 0xa8, - 0x1e, 0xe8, 0xd0, 0x5c, 0xb1, 0x07, 0xf1, 0x0e, 0x46, 0x3f, 0x3a, 0x63, 0x0b, 0xd3, 0x4a, 0x4e, - 0x8b, 0x9e, 0x9f, 0x17, 0x7d, 0xe9, 0x8c, 0xed, 0xd5, 0x71, 0x8e, 0x1f, 0x16, 0xfe, 0xa6, 0x58, - 0x81, 0x91, 0x53, 0xb4, 0xa3, 0x21, 0x72, 0xac, 0xb3, 0x39, 0x70, 0xfa, 0x06, 0x2f, 0x71, 0x51, - 0xef, 0x76, 0xba, 0xda, 0xfa, 0x54, 0x8e, 0x12, 0xa3, 0xca, 0x6f, 0x7d, 0x22, 0x41, 0x7e, 0x8b, - 0x5a, 0xad, 0xfc, 0xf9, 0x03, 0xb5, 0xfa, 0x16, 0xd3, 0x2f, 0xf5, 0xe1, 0x7f, 0x00, 0x00, 0x00, - 0xff, 0xff, 0x85, 0xa7, 0xa7, 0xb1, 0x64, 0x03, 0x00, 0x00, + // 486 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0x8c, 0x93, 0xcf, 0x8e, 0xd3, 0x3c, + 0x14, 0xc5, 0xe5, 0x26, 0x4e, 0x9a, 0xdb, 0x4f, 0xfd, 0x90, 0x85, 0x50, 0x84, 0x58, 0x54, 0x11, + 0x8b, 0xb2, 0x19, 0x24, 0x78, 0x82, 0x4e, 0xc3, 0xa2, 0xa8, 0x74, 0x8a, 0x4b, 0x1f, 0xc0, 0x24, + 0x17, 0x4d, 0x50, 0xfe, 0xe1, 0xd8, 0xb4, 0xd9, 0xb2, 0x85, 0xc7, 0xe0, 0x0d, 0x78, 0x41, 0x64, + 0xd7, 0x21, 0x23, 0x81, 0x46, 0xb3, 0x3c, 0xf7, 0xdc, 0xe8, 0xfe, 0xee, 0xb9, 0x0e, 0xcc, 0x8b, + 0x5a, 0xa1, 0xac, 0x45, 0x79, 0xd5, 0xca, 0x46, 0x35, 0x6c, 0x3a, 0xe8, 0xe4, 0x17, 0x81, 0xd9, + 0x9b, 0x73, 0x5b, 0x36, 0x52, 0xa8, 0xa2, 0xa9, 0xd9, 0x1c, 0x26, 0x9b, 0x34, 0x26, 0x0b, 0xb2, + 0xf4, 0xf8, 0xa4, 0x48, 0x19, 0x03, 0x7f, 0x27, 0x2a, 0x8c, 0x27, 0x0b, 0xb2, 0x8c, 0xb8, 0x5f, + 0x8b, 0x0a, 0xd9, 0x13, 0x08, 0x8e, 0x1d, 0xca, 0x4d, 0x1a, 0x7b, 0xb6, 0x2f, 0xd0, 0x56, 0x99, + 0xde, 0x54, 0x28, 0x11, 0xfb, 0x97, 0xde, 0x5c, 0x28, 0xc1, 0x9e, 0x41, 0xb4, 0x96, 0x28, 0x14, + 0xe6, 0x2b, 0x15, 0x53, 0xdb, 0x1e, 0x65, 0x43, 0xc1, 0xb8, 0xc7, 0x36, 0x77, 0x6e, 0x70, 0x71, + 0xf5, 0x50, 0x60, 0x31, 0x84, 0x29, 0x7e, 0x12, 0xba, 0x54, 0x71, 0xb8, 0x20, 0xcb, 0x29, 0x0f, + 0xf3, 0x8b, 0x4c, 0x7e, 0x12, 0x08, 0x0e, 0x8d, 0x96, 0x19, 0x3e, 0x08, 0x98, 0x81, 0xff, 0xa1, + 0x6f, 0xd1, 0xe2, 0x46, 0xdc, 0x57, 0x7d, 0x8b, 0xec, 0x29, 0x4c, 0xcd, 0x12, 0xc6, 0x77, 0xc0, + 0x53, 0xed, 0xb4, 0xf1, 0xf6, 0xa2, 0xeb, 0x4e, 0x8d, 0xcc, 0x2d, 0x73, 0xc4, 0xa7, 0xad, 0xd3, + 0xec, 0x11, 0x78, 0x47, 0xbe, 0xb5, 0xb0, 0x11, 0xf7, 0x34, 0xdf, 0xde, 0x83, 0xf9, 0xc3, 0x60, + 0xa2, 0xfc, 0x8a, 0xf2, 0x41, 0x98, 0x77, 0x91, 0xbc, 0x7b, 0x90, 0xfc, 0x7f, 0x23, 0xd1, 0x11, + 0xe9, 0x31, 0xd0, 0x83, 0xcc, 0x36, 0xa9, 0xcb, 0x94, 0x76, 0x46, 0x24, 0xdf, 0x08, 0x04, 0x5b, + 0xd1, 0x37, 0x5a, 0xdd, 0xc1, 0x89, 0x2c, 0xce, 0x02, 0x66, 0xab, 0xb6, 0x2d, 0x8b, 0xcc, 0xbe, + 0x02, 0x47, 0x35, 0x13, 0x63, 0xc9, 0x74, 0xbc, 0x43, 0xd1, 0x69, 0x89, 0x15, 0xd6, 0xca, 0xf1, + 0xcd, 0xaa, 0xb1, 0xc4, 0x9e, 0x03, 0x5d, 0x63, 0x59, 0x76, 0xb1, 0xbf, 0xf0, 0x96, 0xb3, 0x57, + 0xf3, 0xab, 0x3f, 0x8f, 0xce, 0x94, 0x39, 0xcd, 0x8c, 0x99, 0x7c, 0x27, 0xe0, 0x1b, 0xcd, 0xfe, + 0x03, 0x72, 0xb6, 0x04, 0x94, 0x93, 0xb3, 0x51, 0xbd, 0x1d, 0x4b, 0x39, 0xe9, 0x8d, 0x3a, 0xd9, + 0x11, 0x94, 0x93, 0x93, 0x51, 0xb7, 0x76, 0x69, 0xca, 0xc9, 0x2d, 0x7b, 0x01, 0xe1, 0x17, 0x8d, + 0xb2, 0xc0, 0x2e, 0xa6, 0x76, 0xd0, 0xff, 0xe3, 0xa0, 0xf7, 0x1a, 0x65, 0xcf, 0x07, 0xdf, 0x7c, + 0x58, 0xb8, 0x4b, 0x91, 0xc2, 0x44, 0x6e, 0xa3, 0x0d, 0xc7, 0xc8, 0x93, 0x15, 0x50, 0xfb, 0x8d, + 0x39, 0xe2, 0xba, 0xa9, 0x2a, 0x51, 0xe7, 0x2e, 0x95, 0x30, 0xbb, 0x48, 0x13, 0x55, 0x7a, 0xed, + 0x12, 0x99, 0xe4, 0xd7, 0x46, 0xf3, 0xbd, 0xdb, 0x7f, 0x22, 0xf7, 0xc9, 0x4b, 0x88, 0x56, 0x25, + 0x4a, 0xc5, 0x75, 0x89, 0x7f, 0xe5, 0xca, 0xc0, 0x7f, 0x7b, 0xb8, 0xd9, 0x0d, 0x67, 0xfe, 0x7c, + 0xb8, 0xd9, 0x7d, 0x0c, 0xec, 0x3f, 0xf8, 0xfa, 0x77, 0x00, 0x00, 0x00, 0xff, 0xff, 0xe0, 0xb6, + 0x3a, 0xf1, 0x95, 0x03, 0x00, 0x00, } diff --git a/bolt/internal/internal.proto b/bolt/internal/internal.proto index b482fe86cd..221b140c28 100644 --- a/bolt/internal/internal.proto +++ b/bolt/internal/internal.proto @@ -52,3 +52,8 @@ message Query { string DB = 2; // DB the database for the query (optional) string RP = 3; // RP is a retention policy and optional; } + +message AlertRule { + string ID = 1; // ID is the unique ID of this alert rule + string JSON = 2; // JSON byte representation of the alert +} diff --git a/chronograf.go b/chronograf.go index 649bcb022a..ff2dc4311a 100644 --- a/chronograf.go +++ b/chronograf.go @@ -13,6 +13,7 @@ const ( ErrSourceNotFound = Error("source not found") ErrServerNotFound = Error("server not found") ErrLayoutNotFound = Error("layout not found") + ErrAlertNotFound = Error("alert not found") ErrAuthentication = Error("user not authenticated") ) @@ -122,34 +123,15 @@ type Ticker interface { Generate(AlertRule) (TICKScript, error) } -// DeadmanValue specifies the timeout duration of a deadman alert. -type DeadmanValue struct { - Period string `json:"period, omitempty"` // Period is the max time data can be missed before an alert -} - -// RelativeValue specifies the trigger logic for a relative value change alert. -type RelativeValue struct { - Change string `json:"change,omitempty"` // Change specifies if the change is a percent or absolute - Period string `json:"period,omitempty"` // Period is the window to search for alerting criteria - Shift string `json:"shift,omitempty"` // Shift is the amount of time to look into the past for the alert to compare to the present - Operator string `json:"operator,omitempty"` // Operator for alert comparison - Value string `json:"value,omitempty"` // Value is the boundary value when alert goes critical -} - -// ThresholdValue specifies the trigger logic for a threshold change alert. -type ThresholdValue struct { - Period string `json:"period,omitempty"` // Period is the window to search for the alerting criteria +// TriggerValues specifies the alerting logic for a specific trigger type +type TriggerValues struct { + Change string `json:"change,omitempty"` // Change specifies if the change is a percent or absolute + Period string `json:"period,omitempty"` // Period is the window to search for alerting criteria + Shift string `json:"shift,omitempty"` // Shift is the amount of time to look into the past for the alert to compare to the present Operator string `json:"operator,omitempty"` // Operator for alert comparison + Value string `json:"value,omitempty"` // Value is the boundary value when alert goes critical Percentile string `json:"percentile,omitempty"` // Percentile is defined only when Relation is not "Once" Relation string `json:"relation,omitempty"` // Relation defines the logic about how often the threshold is met to be an alert. - Value string `json:"value,omitempty"` // Value is the boundary value when alert goes critical -} - -// TriggerValues specifies which of the trigger types defines the alerting logic. One of these whould not be nil. -type TriggerValues struct { - Deadman *DeadmanValue `json:"deadman,omitempty"` - Relative *RelativeValue `json:"relative,omitempty"` - Threshold *ThresholdValue `json:"threshold,omitempty"` } // QueryConfig represents UI query from the data explorer diff --git a/kapacitor/alerts.go b/kapacitor/alerts.go index f84f54c864..861ea598e4 100644 --- a/kapacitor/alerts.go +++ b/kapacitor/alerts.go @@ -6,14 +6,37 @@ import ( "github.com/influxdata/chronograf" ) +func kapaService(alert string) (string, error) { + switch alert { + case "hipchat": + return "hipChat", nil + case "opsgenie": + return "opsGenie", nil + case "pagerduty": + return "pagerDuty", nil + case "victorops": + return "victorOps", nil + case "smtp": + return "email", nil + case "sensu", "slack", "email", "talk", "telegram": + return alert, nil + default: + return "", fmt.Errorf("Unsupport alert %s", alert) + } +} + // AlertServices generates alert chaining methods to be attached to an alert from all rule Services func AlertServices(rule chronograf.AlertRule) (string, error) { alert := "" for _, service := range rule.Alerts { - if err := ValidateAlert(service); err != nil { + srv, err := kapaService(service) + if err != nil { return "", err } - alert = alert + fmt.Sprintf(".%s()", service) + if err := ValidateAlert(srv); err != nil { + return "", err + } + alert = alert + fmt.Sprintf(".%s()", srv) } return alert, nil } diff --git a/kapacitor/alerts_test.go b/kapacitor/alerts_test.go index 651027af67..5dbb33b77e 100644 --- a/kapacitor/alerts_test.go +++ b/kapacitor/alerts_test.go @@ -16,7 +16,7 @@ func TestAlertServices(t *testing.T) { { name: "Test several valid services", rule: chronograf.AlertRule{ - Alerts: []string{"slack", "victorOps", "email"}, + Alerts: []string{"slack", "victorops", "email"}, }, want: `alert() .slack() diff --git a/kapacitor/tickscripts_test.go b/kapacitor/tickscripts_test.go index 960e16161b..848c7999e1 100644 --- a/kapacitor/tickscripts_test.go +++ b/kapacitor/tickscripts_test.go @@ -1,6 +1,7 @@ package kapacitor import ( + "encoding/json" "fmt" "testing" @@ -11,15 +12,13 @@ func TestGenerate(t *testing.T) { alert := chronograf.AlertRule{ Name: "name", Trigger: "relative", - Alerts: []string{"slack", "victorOps", "email"}, + Alerts: []string{"slack", "victorops", "email"}, TriggerValues: chronograf.TriggerValues{ - Relative: &chronograf.RelativeValue{ - Change: "change", - Period: "10m", - Shift: "1m", - Operator: "greater than", - Value: "90", - }, + Change: "change", + Period: "10m", + Shift: "1m", + Operator: "greater than", + Value: "90", }, Every: "30s", Query: chronograf.QueryConfig{ @@ -67,15 +66,13 @@ func TestThreshold(t *testing.T) { alert := chronograf.AlertRule{ Name: "name", Trigger: "threshold", - Alerts: []string{"slack", "victorOps", "email"}, + Alerts: []string{"slack", "victorops", "email"}, TriggerValues: chronograf.TriggerValues{ - Threshold: &chronograf.ThresholdValue{ - Relation: "once", - Period: "10m", - Percentile: "", // TODO: if relation is not once then this will have a number - Operator: "greater than", - Value: "90", - }, + Relation: "once", + Period: "10m", + Percentile: "", // TODO: if relation is not once then this will have a number + Operator: "greater than", + Value: "90", }, Every: "30s", Message: "message", @@ -222,15 +219,13 @@ func TestRelative(t *testing.T) { alert := chronograf.AlertRule{ Name: "name", Trigger: "relative", - Alerts: []string{"slack", "victorOps", "email"}, + Alerts: []string{"slack", "victorops", "email"}, TriggerValues: chronograf.TriggerValues{ - Relative: &chronograf.RelativeValue{ - Change: "change", - Period: "10m", - Shift: "1m", - Operator: "greater than", - Value: "90", - }, + Change: "change", + Period: "10m", + Shift: "1m", + Operator: "greater than", + Value: "90", }, Every: "30s", Message: "message", @@ -382,6 +377,8 @@ trigger fmt.Printf("%s", got) t.Errorf("%q. Relative() = %v, want %v", tt.name, got, tt.want) } + b, _ := json.Marshal(tt.alert) + fmt.Printf("%s", string(b)) } } @@ -389,11 +386,9 @@ func TestDeadman(t *testing.T) { alert := chronograf.AlertRule{ Name: "name", Trigger: "deadman", - Alerts: []string{"slack", "victorOps", "email"}, + Alerts: []string{"slack", "victorops", "email"}, TriggerValues: chronograf.TriggerValues{ - Deadman: &chronograf.DeadmanValue{ - Period: "10m", - }, + Period: "10m", }, Every: "30s", Message: "message", diff --git a/kapacitor/triggers.go b/kapacitor/triggers.go index d0f2926646..d0c233c493 100644 --- a/kapacitor/triggers.go +++ b/kapacitor/triggers.go @@ -59,13 +59,13 @@ func Trigger(rule chronograf.AlertRule) (string, error) { case "deadman": return DeadmanTrigger, nil case "relative": - op, err := kapaOperator(rule.TriggerValues.Relative.Operator) + op, err := kapaOperator(rule.TriggerValues.Operator) if err != nil { return "", err } return fmt.Sprintf(RelativeTrigger, op), nil case "threshold": - op, err := kapaOperator(rule.TriggerValues.Threshold.Operator) + op, err := kapaOperator(rule.TriggerValues.Operator) if err != nil { return "", err } diff --git a/kapacitor/triggers_test.go b/kapacitor/triggers_test.go index db1733514a..8665d5a3e8 100644 --- a/kapacitor/triggers_test.go +++ b/kapacitor/triggers_test.go @@ -35,9 +35,7 @@ func TestTrigger(t *testing.T) { rule: chronograf.AlertRule{ Trigger: "relative", TriggerValues: chronograf.TriggerValues{ - Relative: &chronograf.RelativeValue{ - Operator: "greater than", - }, + Operator: "greater than", }, }, want: `var past = data @@ -68,9 +66,7 @@ var trigger = past rule: chronograf.AlertRule{ Trigger: "threshold", TriggerValues: chronograf.TriggerValues{ - Threshold: &chronograf.ThresholdValue{ - Operator: "greater than", - }, + Operator: "greater than", }, }, want: `var trigger = data diff --git a/kapacitor/vars.go b/kapacitor/vars.go index b9d2f7dd2c..73b4d09c45 100644 --- a/kapacitor/vars.go +++ b/kapacitor/vars.go @@ -41,8 +41,8 @@ func Vars(rule chronograf.AlertRule) (string, error) { ` return fmt.Sprintf(vars, common, - rule.TriggerValues.Threshold.Period, - rule.TriggerValues.Threshold.Value), nil + rule.TriggerValues.Period, + rule.TriggerValues.Value), nil case "relative": vars := ` %s @@ -52,9 +52,9 @@ func Vars(rule chronograf.AlertRule) (string, error) { ` return fmt.Sprintf(vars, common, - rule.TriggerValues.Relative.Period, - rule.TriggerValues.Relative.Shift, - rule.TriggerValues.Relative.Value, + rule.TriggerValues.Period, + rule.TriggerValues.Shift, + rule.TriggerValues.Value, ), nil case "deadman": vars := ` @@ -65,7 +65,7 @@ func Vars(rule chronograf.AlertRule) (string, error) { return fmt.Sprintf(vars, common, "0.0", // deadman threshold hardcoded to zero - rule.TriggerValues.Deadman.Period, + rule.TriggerValues.Period, ), nil default: return "", fmt.Errorf("Unknown trigger mechanism") diff --git a/server/kapacitors.go b/server/kapacitors.go index 3090595915..bf5535db01 100644 --- a/server/kapacitors.go +++ b/server/kapacitors.go @@ -293,21 +293,53 @@ func (h *Service) KapacitorTasksPost(w http.ResponseWriter, r *http.Request) { ID: &uuid.V4{}, } - var rule chronograf.AlertRule - task, err := c.Create(ctx, rule) + var req chronograf.AlertRule + if err = json.NewDecoder(r.Body).Decode(&req); err != nil { + invalidJSON(w) + return + } + // TODO: validate this data + /* + if err := req.Valid(); err != nil { + invalidData(w, err) + return + } + */ + + task, err := c.Create(ctx, req) if err != nil { Error(w, http.StatusInternalServerError, err.Error()) return } - // TODO: Set the tickscript the store - // TODO: possibly use the Href in update to the store - _ = task.TICKScript - _ = task.ID - _ = task.Href - // TODO: Add the task from the store - // TODO: Return POST response - w.WriteHeader(http.StatusNoContent) + req.ID = task.ID + rule, err := h.AlertRulesStore.Add(ctx, req) + if err != nil { + Error(w, http.StatusInternalServerError, err.Error()) + return + } + res := alertResponse{ + AlertRule: rule, + Links: alertLinks{ + Self: fmt.Sprintf("/chronograf/v1/sources/%d/kapacitors/%d/tasks/%s", srv.SrcID, srv.ID, req.ID), + Kapacitor: fmt.Sprintf("/chronograf/v1/sources/%d/kapacitors/%d/proxy?path=%s", srv.SrcID, srv.ID, url.QueryEscape(task.Href)), + }, + TICKScript: string(task.TICKScript), + } + + w.Header().Add("Location", res.Links.Self) + encodeJSON(w, http.StatusCreated, res, h.Logger) +} + +type alertLinks struct { + Self string `json:"self"` + Kapacitor string `json:"kapacitor"` +} + +type alertResponse struct { + chronograf.AlertRule + TICKScript string `json:"tickscript"` + Links alertLinks `json:"links"` } // KapacitorTasksPut proxies PATCH to kapacitor @@ -338,19 +370,40 @@ func (h *Service) KapacitorTasksPut(w http.ResponseWriter, r *http.Request) { Password: srv.Password, Ticker: &kapa.Alert{}, } - // TODO: Pull rule from PUT parameters - var rule chronograf.AlertRule - task, err := c.Update(ctx, c.Href(tid), rule) + var req chronograf.AlertRule + if err = json.NewDecoder(r.Body).Decode(&req); err != nil { + invalidJSON(w) + return + } + // TODO: validate this data + /* + if err := req.Valid(); err != nil { + invalidData(w, err) + return + } + */ + + req.ID = tid + task, err := c.Update(ctx, c.Href(tid), req) if err != nil { Error(w, http.StatusInternalServerError, err.Error()) return } - // TODO: Set the tickscript in the update to the store - // TODO: possibly use the Href in update to the store - _ = task.TICKScript - // TODO: Update the task from the store - // TODO: Return Patch response - w.WriteHeader(http.StatusNoContent) + + if err := h.AlertRulesStore.Update(ctx, req); err != nil { + Error(w, http.StatusInternalServerError, err.Error()) + return + } + + res := alertResponse{ + AlertRule: req, + Links: alertLinks{ + Self: fmt.Sprintf("/chronograf/v1/sources/%d/kapacitors/%d/tasks/%s", srv.SrcID, srv.ID, req.ID), + Kapacitor: fmt.Sprintf("/chronograf/v1/sources/%d/kapacitors/%d/proxy?path=%s", srv.SrcID, srv.ID, url.QueryEscape(task.Href)), + }, + TICKScript: string(task.TICKScript), + } + encodeJSON(w, http.StatusOK, res, h.Logger) } // KapacitorTasksGet retrieves all tasks @@ -373,7 +426,40 @@ func (h *Service) KapacitorTasksGet(w http.ResponseWriter, r *http.Request) { notFound(w, id) return } - // TODO: GET tasks from store + + rules, err := h.AlertRulesStore.All(ctx) + if err != nil { + Error(w, http.StatusInternalServerError, err.Error()) + return + } + + ticker := &kapa.Alert{} + c := kapa.Client{} + res := allAlertsResponse{ + Tasks: []alertResponse{}, + } + for _, rule := range rules { + tickscript, err := ticker.Generate(rule) + if err != nil { + Error(w, http.StatusInternalServerError, err.Error()) + return + } + + ar := alertResponse{ + AlertRule: rule, + Links: alertLinks{ + Self: fmt.Sprintf("/chronograf/v1/sources/%d/kapacitors/%d/tasks/%s", srv.SrcID, srv.ID, rule.ID), + Kapacitor: fmt.Sprintf("/chronograf/v1/sources/%d/kapacitors/%d/proxy?path=%s", srv.SrcID, srv.ID, url.QueryEscape(c.Href(rule.ID))), + }, + TICKScript: string(tickscript), + } + res.Tasks = append(res.Tasks, ar) + } + encodeJSON(w, http.StatusOK, res, h.Logger) +} + +type allAlertsResponse struct { + Tasks []alertResponse `json:"tasks"` } // KapacitorTasksGet retrieves specific task @@ -397,8 +483,29 @@ func (h *Service) KapacitorTasksID(w http.ResponseWriter, r *http.Request) { return } tid := httprouter.GetParamFromContext(ctx, "tid") - // TODO: GET task from store - _ = tid + rule, err := h.AlertRulesStore.Get(ctx, tid) + if err != nil { + Error(w, http.StatusInternalServerError, err.Error()) + return + } + + ticker := &kapa.Alert{} + c := kapa.Client{} + tickscript, err := ticker.Generate(rule) + if err != nil { + Error(w, http.StatusInternalServerError, err.Error()) + return + } + + res := alertResponse{ + AlertRule: rule, + Links: alertLinks{ + Self: fmt.Sprintf("/chronograf/v1/sources/%d/kapacitors/%d/tasks/%s", srv.SrcID, srv.ID, rule.ID), + Kapacitor: fmt.Sprintf("/chronograf/v1/sources/%d/kapacitors/%d/proxy?path=%s", srv.SrcID, srv.ID, url.QueryEscape(c.Href(rule.ID))), + }, + TICKScript: string(tickscript), + } + encodeJSON(w, http.StatusOK, res, h.Logger) } // KapacitorTasksDelete proxies DELETE to kapacitor @@ -422,7 +529,6 @@ func (h *Service) KapacitorTasksDelete(w http.ResponseWriter, r *http.Request) { return } - // TODO: Delete the task from the store tid := httprouter.GetParamFromContext(ctx, "tid") c := kapa.Client{ URL: srv.URL, @@ -433,5 +539,11 @@ func (h *Service) KapacitorTasksDelete(w http.ResponseWriter, r *http.Request) { Error(w, http.StatusInternalServerError, err.Error()) return } + + if err := h.AlertRulesStore.Delete(ctx, chronograf.AlertRule{ID: tid}); err != nil { + Error(w, http.StatusInternalServerError, err.Error()) + return + } + w.WriteHeader(http.StatusNoContent) } diff --git a/server/server.go b/server/server.go index e27f43f8e4..7107b9246f 100644 --- a/server/server.go +++ b/server/server.go @@ -128,6 +128,7 @@ func openService(boltPath, cannedPath string, logger chronograf.Logger) Service ServersStore: db.ServersStore, TimeSeries: &influx.Client{}, LayoutStore: layouts, + AlertRulesStore: db.AlertsStore, } } diff --git a/server/service.go b/server/service.go index 5486f00eb4..198b342a7c 100644 --- a/server/service.go +++ b/server/service.go @@ -8,6 +8,7 @@ type Service struct { SourcesStore chronograf.SourcesStore ServersStore chronograf.ServersStore LayoutStore chronograf.LayoutStore + AlertRulesStore chronograf.AlertRulesStore TimeSeries chronograf.TimeSeries Logger chronograf.Logger }