Make alerting notifcations sync (grafana#6158)

* tech(routines): move the async logic from notification to alerting notifier

* tech(notification): reduce code dupe

* fix(notification): dont touch the response unless its an error

* feat(alerting): make alerting exeuction async but flow sync

* tech(alerting): remove commented code

* tech(alerting): remove unused code

* tech(alerting): fix typo

* tech(alerting): implement Context on EvalContext

* tech(alerting): wait for all alerts to return

* feat(alerting): dont allow alert responses to cancel

* Revert "feat(alerting): dont allow alert responses to cancel"

This reverts commit 324b006.

* feat(alerting): give alerts some time to finish before closing down

Author: bergquist

pkg/api/metrics.go

@@ -1,6 +1,7 @@
 package api
 
 import (
+	"context"
 	"encoding/json"
 	"net/http"
 
@@ -31,7 +32,7 @@ func QueryMetrics(c *middleware.Context, reqDto dtos.MetricRequest) Response {
 		})
 	}
 
-	resp, err := tsdb.HandleRequest(request)
+	resp, err := tsdb.HandleRequest(context.TODO(), request)
 	if err != nil {
 		return ApiError(500, "Metric request error", err)
 	}

pkg/bus/bus.go

@@ -1,18 +1,22 @@
 package bus
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 )
 
 type HandlerFunc interface{}
+type CtxHandlerFunc func() 
 type Msg interface{}
 
 type Bus interface {
 	Dispatch(msg Msg) error
+	DispatchCtx(ctx context.Context, msg Msg) error
 	Publish(msg Msg) error
 
 	AddHandler(handler HandlerFunc)
+	AddCtxHandler(handler HandlerFunc)
 	AddEventListener(handler HandlerFunc)
 	AddWildcardListener(handler HandlerFunc)
 }
@@ -34,6 +38,27 @@ func New() Bus {
 	return bus
 }
 
+func (b *InProcBus) DispatchCtx(ctx context.Context, msg Msg) error {
+	var msgName = reflect.TypeOf(msg).Elem().Name()
+
+	var handler = b.handlers[msgName]
+	if handler == nil {
+		return fmt.Errorf("handler not found for %s", msgName)
+	}
+
+	var params = make([]reflect.Value, 2)
+	params[0] = reflect.ValueOf(ctx)
+	params[1] = reflect.ValueOf(msg)
+
+	ret := reflect.ValueOf(handler).Call(params)
+	err := ret[0].Interface()
+	if err == nil {
+		return nil
+	} else {
+		return err.(error)
+	}
+}
+
 func (b *InProcBus) Dispatch(msg Msg) error {
 	var msgName = reflect.TypeOf(msg).Elem().Name()
 
@@ -90,6 +115,12 @@ func (b *InProcBus) AddHandler(handler HandlerFunc) {
 	b.handlers[queryTypeName] = handler
 }
 
+func (b *InProcBus) AddCtxHandler(handler HandlerFunc) {
+	handlerType := reflect.TypeOf(handler)
+	queryTypeName := handlerType.In(1).Elem().Name()
+	b.handlers[queryTypeName] = handler
+}
+
 func (b *InProcBus) AddEventListener(handler HandlerFunc) {
 	handlerType := reflect.TypeOf(handler)
 	eventName := handlerType.In(0).Elem().Name()
@@ -105,6 +136,11 @@ func AddHandler(implName string, handler HandlerFunc) {
 	globalBus.AddHandler(handler)
 }
 
+// Package level functions
+func AddCtxHandler(implName string, handler HandlerFunc) {
+	globalBus.AddCtxHandler(handler)
+}
+
 // Package level functions
 func AddEventListener(handler HandlerFunc) {
 	globalBus.AddEventListener(handler)
@@ -118,6 +154,10 @@ func Dispatch(msg Msg) error {
 	return globalBus.Dispatch(msg)
 }
 
+func DispatchCtx(ctx context.Context, msg Msg) error {
+	return globalBus.DispatchCtx(ctx, msg)
+}
+
 func Publish(msg Msg) error {
 	return globalBus.Publish(msg)
 }

pkg/models/notifications.go

@@ -12,13 +12,24 @@ type SendEmailCommand struct {
 	Info     string
 }
 
+type SendEmailCommandSync struct {
+	SendEmailCommand
+}
+
 type SendWebhook struct {
 	Url      string
 	User     string
 	Password string
 	Body     string
 }
 
+type SendWebhookSync struct {
+	Url      string
+	User     string
+	Password string
+	Body     string
+}
+
 type SendResetPasswordEmailCommand struct {
 	User *User
 }

pkg/services/alerting/conditions/query.go

@@ -82,7 +82,7 @@ func (c *QueryCondition) executeQuery(context *alerting.EvalContext, timeRange *
 	req := c.getRequestForAlertRule(getDsInfo.Result, timeRange)
 	result := make(tsdb.TimeSeriesSlice, 0)
 
-	resp, err := c.HandleRequest(req)
+	resp, err := c.HandleRequest(context.Context, req)
 	if err != nil {
 		return nil, fmt.Errorf("tsdb.HandleRequest() error %v", err)
 	}

pkg/services/alerting/conditions/query_test.go

@@ -1,6 +1,7 @@
 package conditions
 
 import (
+	"context"
 	"testing"
 
 	null "gopkg.in/guregu/null.v3"
@@ -137,7 +138,7 @@ func (ctx *queryConditionTestContext) exec() {
 
 	ctx.condition = condition
 
-	condition.HandleRequest = func(req *tsdb.Request) (*tsdb.Response, error) {
+	condition.HandleRequest = func(context context.Context, req *tsdb.Request) (*tsdb.Response, error) {
 		return &tsdb.Response{
 			Results: map[string]*tsdb.QueryResult{
 				"A": {Series: ctx.series},

pkg/services/alerting/engine.go

@@ -11,7 +11,6 @@ import (
 
 type Engine struct {
 	execQueue     chan *Job
-	resultQueue   chan *EvalContext
 	clock         clock.Clock
 	ticker        *Ticker
 	scheduler     Scheduler
@@ -25,7 +24,6 @@ func NewEngine() *Engine {
 	e := &Engine{
 		ticker:        NewTicker(time.Now(), time.Second*0, clock.New()),
 		execQueue:     make(chan *Job, 1000),
-		resultQueue:   make(chan *EvalContext, 1000),
 		scheduler:     NewScheduler(),
 		evalHandler:   NewEvalHandler(),
 		ruleReader:    NewRuleReader(),
@@ -39,23 +37,17 @@ func NewEngine() *Engine {
 func (e *Engine) Run(ctx context.Context) error {
 	e.log.Info("Initializing Alerting")
 
-	g, ctx := errgroup.WithContext(ctx)
+	alertGroup, ctx := errgroup.WithContext(ctx)
 
-	g.Go(func() error { return e.alertingTicker(ctx) })
-	g.Go(func() error { return e.execDispatcher(ctx) })
-	g.Go(func() error { return e.resultDispatcher(ctx) })
+	alertGroup.Go(func() error { return e.alertingTicker(ctx) })
+	alertGroup.Go(func() error { return e.runJobDispatcher(ctx) })
 
-	err := g.Wait()
+	err := alertGroup.Wait()
 
 	e.log.Info("Stopped Alerting", "reason", err)
 	return err
 }
 
-func (e *Engine) Stop() {
-	close(e.execQueue)
-	close(e.resultQueue)
-}
-
 func (e *Engine) alertingTicker(grafanaCtx context.Context) error {
 	defer func() {
 		if err := recover(); err != nil {
@@ -81,69 +73,58 @@ func (e *Engine) alertingTicker(grafanaCtx context.Context) error {
 	}
 }
 
-func (e *Engine) execDispatcher(grafanaCtx context.Context) error {
+func (e *Engine) runJobDispatcher(grafanaCtx context.Context) error {
+	dispatcherGroup, alertCtx := errgroup.WithContext(grafanaCtx)
+
 	for {
 		select {
 		case <-grafanaCtx.Done():
-			close(e.resultQueue)
-			return grafanaCtx.Err()
+			return dispatcherGroup.Wait()
 		case job := <-e.execQueue:
-			go e.executeJob(grafanaCtx, job)
+			dispatcherGroup.Go(func() error { return e.processJob(alertCtx, job) })
 		}
 	}
 }
 
-func (e *Engine) executeJob(grafanaCtx context.Context, job *Job) error {
+var (
+	unfinishedWorkTimeout time.Duration = time.Second * 5
+	alertTimeout          time.Duration = time.Second * 30
+)
+
+func (e *Engine) processJob(grafanaCtx context.Context, job *Job) error {
 	defer func() {
 		if err := recover(); err != nil {
-			e.log.Error("Execute Alert Panic", "error", err, "stack", log.Stack(1))
+			e.log.Error("Alert Panic", "error", err, "stack", log.Stack(1))
 		}
 	}()
 
-	done := make(chan *EvalContext, 1)
+	alertCtx, cancelFn := context.WithTimeout(context.TODO(), alertTimeout)
+
+	job.Running = true
+	evalContext := NewEvalContext(alertCtx, job.Rule)
+
+	done := make(chan struct{})
+
 	go func() {
-		job.Running = true
-		context := NewEvalContext(job.Rule)
-		e.evalHandler.Eval(context)
-		job.Running = false
-		done <- context
+		e.evalHandler.Eval(evalContext)
+		e.resultHandler.Handle(evalContext)
 		close(done)
 	}()
 
+	var err error = nil
 	select {
-
 	case <-grafanaCtx.Done():
-		return grafanaCtx.Err()
-	case evalContext := <-done:
-		e.resultQueue <- evalContext
-	}
-
-	return nil
-}
-
-func (e *Engine) resultDispatcher(grafanaCtx context.Context) error {
-	for {
 		select {
-		case <-grafanaCtx.Done():
-			//handle all responses before shutting down.
-			for result := range e.resultQueue {
-				e.handleResponse(result)
-			}
-
-			return grafanaCtx.Err()
-		case result := <-e.resultQueue:
-			e.handleResponse(result)
+		case <-time.After(unfinishedWorkTimeout):
+			cancelFn()
+			err = grafanaCtx.Err()
+		case <-done:
 		}
+	case <-done:
 	}
-}
 
-func (e *Engine) handleResponse(result *EvalContext) {
-	defer func() {
-		if err := recover(); err != nil {
-			e.log.Error("Panic in resultDispatcher", "error", err, "stack", log.Stack(1))
-		}
-	}()
-
-	e.log.Debug("Alert Rule Result", "ruleId", result.Rule.Id, "firing", result.Firing)
-	e.resultHandler.Handle(result)
+	e.log.Debug("Job Execution completed", "timeMs", evalContext.GetDurationMs(), "alertId", evalContext.Rule.Id, "name", evalContext.Rule.Name, "firing", evalContext.Firing)
+	job.Running = false
+	cancelFn()
+	return err
 }

pkg/services/alerting/eval_context.go

@@ -1,6 +1,7 @@
 package alerting
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -20,14 +21,30 @@ type EvalContext struct {
 	StartTime       time.Time
 	EndTime         time.Time
 	Rule            *Rule
-	DoneChan        chan bool
-	CancelChan      chan bool
 	log             log.Logger
 	dashboardSlug   string
 	ImagePublicUrl  string
 	ImageOnDiskPath string
 	NoDataFound     bool
 	RetryCount      int
+
+	Context context.Context
+}
+
+func (evalContext *EvalContext) Deadline() (deadline time.Time, ok bool) {
+	return evalContext.Deadline()
+}
+
+func (evalContext *EvalContext) Done() <-chan struct{} {
+	return evalContext.Context.Done()
+}
+
+func (evalContext *EvalContext) Err() error {
+	return evalContext.Context.Err()
+}
+
+func (evalContext *EvalContext) Value(key interface{}) interface{} {
+	return evalContext.Context.Value(key)
 }
 
 type StateDescription struct {
@@ -94,14 +111,13 @@ func (c *EvalContext) GetRuleUrl() (string, error) {
 	}
 }
 
-func NewEvalContext(rule *Rule) *EvalContext {
+func NewEvalContext(alertCtx context.Context, rule *Rule) *EvalContext {
 	return &EvalContext{
+		Context:     alertCtx,
 		StartTime:   time.Now(),
 		Rule:        rule,
 		Logs:        make([]*ResultLogEntry, 0),
 		EvalMatches: make([]*EvalMatch, 0),
-		DoneChan:    make(chan bool, 1),
-		CancelChan:  make(chan bool, 1),
 		log:         log.New("alerting.evalContext"),
 		RetryCount:  0,
 	}