tech(goroutines): sync state between different go routines

pkg/cmd/grafana-server/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"io/ioutil"
@@ -12,6 +13,8 @@ import (
 	"syscall"
 	"time"
 
+	"golang.org/x/sync/errgroup"
+
 	"github.com/grafana/grafana/pkg/log"
 	"github.com/grafana/grafana/pkg/login"
 	"github.com/grafana/grafana/pkg/metrics"
@@ -57,26 +60,33 @@ func main() {
 	setting.BuildCommit = commit
 	setting.BuildStamp = buildstampInt64
 
-	go listenToSystemSignals()
+	appContext, cancelFn := context.WithCancel(context.Background())
+	grafanaGroup, _ := errgroup.WithContext(appContext)
+
+	go listenToSystemSignals(cancelFn, grafanaGroup)
 
 	flag.Parse()
 	writePIDFile()
 	initRuntime()
+	initSql()
 	metrics.Init()
 	search.Init()
 	login.Init()
 	social.NewOAuthService()
 	eventpublisher.Init()
 	plugins.Init()
-	alertingInit.Init()
-	backgroundtasks.Init()
+
+	grafanaGroup.Go(func() error { return alertingInit.Init(appContext) })
+	grafanaGroup.Go(func() error { return backgroundtasks.Init(appContext) })
 
 	if err := notifications.Init(); err != nil {
 		log.Fatal(3, "Notification service failed to initialize", err)
 	}
 
-	StartServer()
-	exitChan <- 0
+	exitCode := StartServer()
+
+	grafanaGroup.Wait()
+	exitChan <- exitCode
 }
 
 func initRuntime() {
@@ -94,7 +104,9 @@ func initRuntime() {
 	logger.Info("Starting Grafana", "version", version, "commit", commit, "compiled", time.Unix(setting.BuildStamp, 0))
 
 	setting.LogConfigurationInfo()
+}
 
+func initSql() {
 	sqlstore.NewEngine()
 	sqlstore.EnsureAdminUser()
 }
@@ -117,7 +129,7 @@ func writePIDFile() {
 	}
 }
 
-func listenToSystemSignals() {
+func listenToSystemSignals(cancel context.CancelFunc, grafanaGroup *errgroup.Group) {
 	signalChan := make(chan os.Signal, 1)
 	code := 0
 
@@ -125,7 +137,7 @@ func listenToSystemSignals() {
 
 	select {
 	case sig := <-signalChan:
-		log.Info("Received signal %s. shutting down", sig)
+		log.Info2("Received system signal. Shutting down", "signal", sig)
 	case code = <-exitChan:
 		switch code {
 		case 0:
@@ -135,6 +147,8 @@ func listenToSystemSignals() {
 		}
 	}
 
+	cancel()
+	grafanaGroup.Wait()
 	log.Close()
 	os.Exit(code)
 }

pkg/cmd/grafana-server/web.go

@@ -6,7 +6,6 @@ package main
 import (
 	"fmt"
 	"net/http"
-	"os"
 	"path"
 
 	"gopkg.in/macaron.v1"
@@ -79,7 +78,7 @@ func mapStatic(m *macaron.Macaron, rootDir string, dir string, prefix string) {
 	))
 }
 
-func StartServer() {
+func StartServer() int {
 	logger = log.New("server")
 
 	var err error
@@ -95,11 +94,13 @@ func StartServer() {
 		err = http.ListenAndServeTLS(listenAddr, setting.CertFile, setting.KeyFile, m)
 	default:
 		logger.Error("Invalid protocol", "protocol", setting.Protocol)
-		os.Exit(1)
+		return 1
 	}
 
 	if err != nil {
 		logger.Error("Fail to start server", "error", err)
-		os.Exit(1)
+		return 1
 	}
+
+	return 0
 }

pkg/services/alerting/engine.go

@@ -1,10 +1,12 @@
 package alerting
 
 import (
+	"context"
 	"time"
 
 	"github.com/benbjohnson/clock"
 	"github.com/grafana/grafana/pkg/log"
+	"golang.org/x/sync/errgroup"
 )
 
 type Engine struct {
@@ -34,12 +36,16 @@ func NewEngine() *Engine {
 	return e
 }
 
-func (e *Engine) Start() {
+func (e *Engine) Start(grafanaCtx context.Context) error {
 	e.log.Info("Starting Alerting Engine")
 
-	go e.alertingTicker()
-	go e.execDispatcher()
-	go e.resultDispatcher()
+	g, _ := errgroup.WithContext(grafanaCtx)
+
+	g.Go(func() error { return e.alertingTicker(grafanaCtx) })
+	g.Go(func() error { return e.execDispatcher(grafanaCtx) })
+	g.Go(func() error { return e.resultDispatcher(grafanaCtx) })
+
+	return g.Wait()
 }
 
 func (e *Engine) Stop() {
@@ -47,7 +53,7 @@ func (e *Engine) Stop() {
 	close(e.resultQueue)
 }
 
-func (e *Engine) alertingTicker() {
+func (e *Engine) alertingTicker(grafanaCtx context.Context) error {
 	defer func() {
 		if err := recover(); err != nil {
 			e.log.Error("Scheduler Panic: stopping alertingTicker", "error", err, "stack", log.Stack(1))
@@ -58,6 +64,8 @@ func (e *Engine) alertingTicker() {
 
 	for {
 		select {
+		case <-grafanaCtx.Done():
+			return grafanaCtx.Err()
 		case tick := <-e.ticker.C:
 			// TEMP SOLUTION update rules ever tenth tick
 			if tickIndex%10 == 0 {
@@ -70,31 +78,56 @@ func (e *Engine) alertingTicker() {
 	}
 }
 
-func (e *Engine) execDispatcher() {
-	for job := range e.execQueue {
-		e.log.Debug("Starting executing alert rule", "alert id", job.Rule.Id)
-		go e.executeJob(job)
+func (e *Engine) execDispatcher(grafanaCtx context.Context) error {
+	for {
+		select {
+		case <-grafanaCtx.Done():
+			close(e.resultQueue)
+			return grafanaCtx.Err()
+		case job := <-e.execQueue:
+			go e.executeJob(grafanaCtx, job)
+		}
 	}
 }
 
-func (e *Engine) executeJob(job *Job) {
+func (e *Engine) executeJob(grafanaCtx context.Context, job *Job) {
 	defer func() {
 		if err := recover(); err != nil {
 			e.log.Error("Execute Alert Panic", "error", err, "stack", log.Stack(1))
 		}
 	}()
 
-	job.Running = true
-	context := NewEvalContext(job.Rule)
-	e.evalHandler.Eval(context)
-	job.Running = false
+	done := make(chan *EvalContext, 1)
+	go func() {
+		job.Running = true
+		context := NewEvalContext(job.Rule)
+		e.evalHandler.Eval(context)
+		job.Running = false
+		done <- context
+		close(done)
+	}()
 
-	e.resultQueue <- context
+	select {
+	case evalContext := <-done:
+		e.resultQueue <- evalContext
+	case <-grafanaCtx.Done():
+
+	}
 }
 
-func (e *Engine) resultDispatcher() {
-	for result := range e.resultQueue {
-		go e.handleResponse(result)
+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)
+		}
 	}
 }
 

pkg/services/alerting/init/init.go

@@ -1,6 +1,8 @@
 package init
 
 import (
+	"context"
+
 	"github.com/grafana/grafana/pkg/services/alerting"
 	_ "github.com/grafana/grafana/pkg/services/alerting/conditions"
 	_ "github.com/grafana/grafana/pkg/services/alerting/notifiers"
@@ -11,11 +13,11 @@ import (
 
 var engine *alerting.Engine
 
-func Init() {
+func Init(ctx context.Context) error {
 	if !setting.AlertingEnabled {
-		return
+		return nil
 	}
 
 	engine = alerting.NewEngine()
-	engine.Start()
+	return engine.Start(ctx)
 }

pkg/services/backgroundtasks/background_tasks.go

@@ -4,8 +4,11 @@
 package backgroundtasks
 
 import (
+	"context"
 	"time"
 
+	"golang.org/x/sync/errgroup"
+
 	"github.com/grafana/grafana/pkg/bus"
 	"github.com/grafana/grafana/pkg/log"
 	"github.com/grafana/grafana/pkg/models"
@@ -15,11 +18,14 @@ var (
 	tlog log.Logger = log.New("ticker")
 )
 
-func Init() {
-	go start()
+func Init(ctx context.Context) error {
+	g, _ := errgroup.WithContext(ctx)
+	g.Go(func() error { return start(ctx) })
+
+	return g.Wait()
 }
 
-func start() {
+func start(ctx context.Context) error {
 	go cleanup(time.Now())
 
 	ticker := time.NewTicker(time.Hour * 1)
@@ -27,6 +33,8 @@ func start() {
 		select {
 		case tick := <-ticker.C:
 			go cleanup(tick)
+		case <-ctx.Done():
+			return ctx.Err()
 		}
 	}
 }