diff --git a/pkg/services/alerting/alert_rule_reader.gi.go b/pkg/services/alerting/alert_rule_reader.go similarity index 100% rename from pkg/services/alerting/alert_rule_reader.gi.go rename to pkg/services/alerting/alert_rule_reader.go diff --git a/pkg/services/alerting/alerting.go b/pkg/services/alerting/alerting.go index ae44a6dd557..19453a29988 100644 --- a/pkg/services/alerting/alerting.go +++ b/pkg/services/alerting/alerting.go @@ -8,6 +8,7 @@ import ( "github.com/grafana/grafana/pkg/log" m "github.com/grafana/grafana/pkg/models" "github.com/grafana/grafana/pkg/setting" + "sync" ) func Init() { @@ -25,6 +26,7 @@ func Init() { type Scheduler struct { jobs []*AlertJob runQueue chan *AlertJob + mtx sync.RWMutex alertRuleFetcher RuleReader @@ -41,42 +43,45 @@ func NewScheduler() *Scheduler { } } -func (s *Scheduler) heartBeat() { +func (this *Scheduler) heartBeat() { //write heartBeat to db. //get the modulus position of active servers - log.Info("Heartbeat: Sending heartbeat from " + s.serverId) - s.clusterSize = 1 - s.serverPosition = 1 + log.Info("Heartbeat: Sending heartbeat from " + this.serverId) + this.clusterSize = 1 + this.serverPosition = 1 } -func (s *Scheduler) Dispatch(reader RuleReader) { +func (this *Scheduler) Dispatch(reader RuleReader) { reschedule := time.NewTicker(time.Second * 10) secondTicker := time.NewTicker(time.Second) heartbeat := time.NewTicker(time.Second * 5) - s.heartBeat() - s.updateJobs(reader.Fetch) + this.heartBeat() + this.updateJobs(reader.Fetch) for { select { case <-secondTicker.C: - s.queueJobs() + this.queueJobs() case <-reschedule.C: - s.updateJobs(reader.Fetch) + this.updateJobs(reader.Fetch) case <-heartbeat.C: - s.heartBeat() + this.heartBeat() } } } -func (s *Scheduler) updateJobs(f func() []m.AlertRule) { +func (this *Scheduler) updateJobs(f func() []m.AlertRule) { log.Debug("Scheduler: UpdateJobs()") jobs := make([]*AlertJob, 0) rules := f() - for i := s.serverPosition - 1; i < len(rules); i += s.clusterSize { + this.mtx.Lock() + defer this.mtx.Unlock() + + for i := this.serverPosition - 1; i < len(rules); i += this.clusterSize { rule := rules[i] jobs = append(jobs, &AlertJob{ id: rule.Id, @@ -89,34 +94,44 @@ func (s *Scheduler) updateJobs(f func() []m.AlertRule) { log.Debug("Scheduler: Selected %d jobs", len(jobs)) - s.jobs = jobs + this.jobs = jobs } -func (s *Scheduler) queueJobs() { +func (this *Scheduler) queueJobs() { now := time.Now().Unix() - for _, job := range s.jobs { - if now%job.frequency == 0 { + for _, job := range this.jobs { + if now%job.frequency == 0 && job.running == false { log.Info("Scheduler: Putting job on to run queue: %s", job.name) - s.runQueue <- job + this.runQueue <- job } } } -func (s *Scheduler) Executor(executor Executor) { - for job := range s.runQueue { - log.Info("Executor: queue length %d", len(s.runQueue)) +func (this *Scheduler) Executor(executor Executor) { + for job := range this.runQueue { + log.Info("Executor: queue length %d", len(this.runQueue)) log.Info("Executor: executing %s", job.name) - go executor.Execute(job.rule) + go Measure(executor, job) } } +func Measure(exec Executor, rule *AlertJob) { + now := time.Now() + rule.running = true + exec.Execute(rule.rule) + rule.running = true + elapsed := time.Since(now) + log.Info("Schedular: exeuction took %v milli seconds", elapsed.Nanoseconds()/1000000) +} + type AlertJob struct { id int64 name string frequency int64 offset int64 delay bool + running bool rule m.AlertRule } diff --git a/pkg/services/alerting/alerting_test.go b/pkg/services/alerting/alerting_test.go index c2f41a3692a..e3f199a16a3 100644 --- a/pkg/services/alerting/alerting_test.go +++ b/pkg/services/alerting/alerting_test.go @@ -59,5 +59,24 @@ func TestAlertingScheduler(t *testing.T) { So(len(scheduler.jobs), ShouldEqual, 1) So(scheduler.jobs[0].id, ShouldEqual, 6) }) + + Convey("more servers then alerts", func() { + mockFn := func() []m.AlertRule { + return []m.AlertRule{ + {Id: 1, Title: "test 1"}, + } + } + + scheduler := &Scheduler{ + jobs: make([]*AlertJob, 0), + runQueue: make(chan *AlertJob, 1000), + serverId: "", + serverPosition: 3, + clusterSize: 3, + } + + scheduler.updateJobs(mockFn) + So(len(scheduler.jobs), ShouldEqual, 0) + }) }) } diff --git a/pkg/services/alerting/notifier.go b/pkg/services/alerting/notifier.go new file mode 100644 index 00000000000..d806a5d69ca --- /dev/null +++ b/pkg/services/alerting/notifier.go @@ -0,0 +1 @@ +package alerting