From 34b31aeef84aae39e4ec63f4afa04ea3bda6ef29 Mon Sep 17 00:00:00 2001 From: bergquist Date: Tue, 27 Sep 2016 16:38:19 +0200 Subject: [PATCH] tech(goroutines): sync state between different go routines --- pkg/cmd/grafana-server/main.go | 28 ++++++-- pkg/cmd/grafana-server/web.go | 9 +-- pkg/services/alerting/engine.go | 69 ++++++++++++++----- pkg/services/alerting/init/init.go | 8 ++- .../backgroundtasks/background_tasks.go | 14 +++- vendor/golang.org/x/sync/LICENSE | 27 ++++++++ vendor/golang.org/x/sync/PATENTS | 22 ++++++ vendor/golang.org/x/sync/errgroup/errgroup.go | 67 ++++++++++++++++++ vendor/vendor.json | 18 +++-- 9 files changed, 221 insertions(+), 41 deletions(-) create mode 100644 vendor/golang.org/x/sync/LICENSE create mode 100644 vendor/golang.org/x/sync/PATENTS create mode 100644 vendor/golang.org/x/sync/errgroup/errgroup.go diff --git a/pkg/cmd/grafana-server/main.go b/pkg/cmd/grafana-server/main.go index 42c8dfedacf..eaf9a290d48 100644 --- a/pkg/cmd/grafana-server/main.go +++ b/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) } diff --git a/pkg/cmd/grafana-server/web.go b/pkg/cmd/grafana-server/web.go index 4c294a814ca..44177ccc5b4 100644 --- a/pkg/cmd/grafana-server/web.go +++ b/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 } diff --git a/pkg/services/alerting/engine.go b/pkg/services/alerting/engine.go index 19befe87ed8..c3841790d88 100644 --- a/pkg/services/alerting/engine.go +++ b/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) + } } } diff --git a/pkg/services/alerting/init/init.go b/pkg/services/alerting/init/init.go index b9cba2fd353..66f272b2266 100644 --- a/pkg/services/alerting/init/init.go +++ b/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) } diff --git a/pkg/services/backgroundtasks/background_tasks.go b/pkg/services/backgroundtasks/background_tasks.go index 5c4a7d197a8..796bff17290 100644 --- a/pkg/services/backgroundtasks/background_tasks.go +++ b/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() } } } diff --git a/vendor/golang.org/x/sync/LICENSE b/vendor/golang.org/x/sync/LICENSE new file mode 100644 index 00000000000..6a66aea5eaf --- /dev/null +++ b/vendor/golang.org/x/sync/LICENSE @@ -0,0 +1,27 @@ +Copyright (c) 2009 The Go Authors. All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright +notice, this list of conditions and the following disclaimer. + * Redistributions in binary form must reproduce the above +copyright notice, this list of conditions and the following disclaimer +in the documentation and/or other materials provided with the +distribution. + * Neither the name of Google Inc. nor the names of its +contributors may be used to endorse or promote products derived from +this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/vendor/golang.org/x/sync/PATENTS b/vendor/golang.org/x/sync/PATENTS new file mode 100644 index 00000000000..733099041f8 --- /dev/null +++ b/vendor/golang.org/x/sync/PATENTS @@ -0,0 +1,22 @@ +Additional IP Rights Grant (Patents) + +"This implementation" means the copyrightable works distributed by +Google as part of the Go project. + +Google hereby grants to You a perpetual, worldwide, non-exclusive, +no-charge, royalty-free, irrevocable (except as stated in this section) +patent license to make, have made, use, offer to sell, sell, import, +transfer and otherwise run, modify and propagate the contents of this +implementation of Go, where such license applies only to those patent +claims, both currently owned or controlled by Google and acquired in +the future, licensable by Google that are necessarily infringed by this +implementation of Go. This grant does not include claims that would be +infringed only as a consequence of further modification of this +implementation. If you or your agent or exclusive licensee institute or +order or agree to the institution of patent litigation against any +entity (including a cross-claim or counterclaim in a lawsuit) alleging +that this implementation of Go or any code incorporated within this +implementation of Go constitutes direct or contributory patent +infringement, or inducement of patent infringement, then any patent +rights granted to you under this License for this implementation of Go +shall terminate as of the date such litigation is filed. diff --git a/vendor/golang.org/x/sync/errgroup/errgroup.go b/vendor/golang.org/x/sync/errgroup/errgroup.go new file mode 100644 index 00000000000..533438d91c1 --- /dev/null +++ b/vendor/golang.org/x/sync/errgroup/errgroup.go @@ -0,0 +1,67 @@ +// Copyright 2016 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +// Package errgroup provides synchronization, error propagation, and Context +// cancelation for groups of goroutines working on subtasks of a common task. +package errgroup + +import ( + "sync" + + "golang.org/x/net/context" +) + +// A Group is a collection of goroutines working on subtasks that are part of +// the same overall task. +// +// A zero Group is valid and does not cancel on error. +type Group struct { + cancel func() + + wg sync.WaitGroup + + errOnce sync.Once + err error +} + +// WithContext returns a new Group and an associated Context derived from ctx. +// +// The derived Context is canceled the first time a function passed to Go +// returns a non-nil error or the first time Wait returns, whichever occurs +// first. +func WithContext(ctx context.Context) (*Group, context.Context) { + ctx, cancel := context.WithCancel(ctx) + return &Group{cancel: cancel}, ctx +} + +// Wait blocks until all function calls from the Go method have returned, then +// returns the first non-nil error (if any) from them. +func (g *Group) Wait() error { + g.wg.Wait() + if g.cancel != nil { + g.cancel() + } + return g.err +} + +// Go calls the given function in a new goroutine. +// +// The first call to return a non-nil error cancels the group; its error will be +// returned by Wait. +func (g *Group) Go(f func() error) { + g.wg.Add(1) + + go func() { + defer g.wg.Done() + + if err := f(); err != nil { + g.errOnce.Do(func() { + g.err = err + if g.cancel != nil { + g.cancel() + } + }) + } + }() +} diff --git a/vendor/vendor.json b/vendor/vendor.json index 46aec8bd61b..31d781dfd07 100644 --- a/vendor/vendor.json +++ b/vendor/vendor.json @@ -15,12 +15,6 @@ "revisionTime": "2016-09-17T18:44:01Z" }, { - "checksumSHA1": "WHc3uByvGaMcnSoI21fhzYgbOgg=", - "path": "golang.org/x/net/context/ctxhttp", - "revision": "71a035914f99bb58fe82eac0f1289f10963d876c", - "revisionTime": "2016-09-12T21:59:12Z" - }, - { "checksumSHA1": "6AYg4fjEvFuAVN3wHakGApjhZAM=", "path": "github.com/smartystreets/assertions", "revision": "2063fd1cc7c975db70502811a34b06ad034ccdf2", @@ -55,6 +49,18 @@ "path": "github.com/smartystreets/goconvey/convey/reporting", "revision": "5db88ed452e937f2fd557de6f4f1af7f2eabed0b", "revisionTime": "2016-08-23T18:01:44Z" + }, + { + "checksumSHA1": "WHc3uByvGaMcnSoI21fhzYgbOgg=", + "path": "golang.org/x/net/context/ctxhttp", + "revision": "71a035914f99bb58fe82eac0f1289f10963d876c", + "revisionTime": "2016-09-12T21:59:12Z" + }, + { + "checksumSHA1": "S0DP7Pn7sZUmXc55IzZnNvERu6s=", + "path": "golang.org/x/sync/errgroup", + "revision": "316e794f7b5e3df4e95175a45a5fb8b12f85cb4f", + "revisionTime": "2016-07-15T18:54:39Z" } ], "rootPath": "github.com/grafana/grafana"