Merge pull request #11801 from grafana/provision-service-refactor

Server shutdown flow rewrite & provision service refactor

pkg/api/http_server.go

@@ -26,9 +26,14 @@ import (
 	"github.com/grafana/grafana/pkg/middleware"
 	"github.com/grafana/grafana/pkg/middleware"
 	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/grafana/grafana/pkg/plugins"
+	"github.com/grafana/grafana/pkg/registry"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/grafana/grafana/pkg/setting"
 )
 )
 
 
+func init() {
+	registry.RegisterService(&HTTPServer{})
+}
+
 type HTTPServer struct {
 type HTTPServer struct {
 	log           log.Logger
 	log           log.Logger
 	macaron       *macaron.Macaron
 	macaron       *macaron.Macaron
@@ -41,12 +46,14 @@ type HTTPServer struct {
 	Bus           bus.Bus       `inject:""`
 	Bus           bus.Bus       `inject:""`
 }
 }
 
 
-func (hs *HTTPServer) Init() {
+func (hs *HTTPServer) Init() error {
 	hs.log = log.New("http.server")
 	hs.log = log.New("http.server")
 	hs.cache = gocache.New(5*time.Minute, 10*time.Minute)
 	hs.cache = gocache.New(5*time.Minute, 10*time.Minute)
+
+	return nil
 }
 }
 
 
-func (hs *HTTPServer) Start(ctx context.Context) error {
+func (hs *HTTPServer) Run(ctx context.Context) error {
 	var err error
 	var err error
 
 
 	hs.context = ctx
 	hs.context = ctx
@@ -57,17 +64,18 @@ func (hs *HTTPServer) Start(ctx context.Context) error {
 	hs.streamManager.Run(ctx)
 	hs.streamManager.Run(ctx)
 
 
 	listenAddr := fmt.Sprintf("%s:%s", setting.HttpAddr, setting.HttpPort)
 	listenAddr := fmt.Sprintf("%s:%s", setting.HttpAddr, setting.HttpPort)
-	hs.log.Info("Initializing HTTP Server", "address", listenAddr, "protocol", setting.Protocol, "subUrl", setting.AppSubUrl, "socket", setting.SocketPath)
+	hs.log.Info("HTTP Server Listen", "address", listenAddr, "protocol", setting.Protocol, "subUrl", setting.AppSubUrl, "socket", setting.SocketPath)
 
 
 	hs.httpSrv = &http.Server{Addr: listenAddr, Handler: hs.macaron}
 	hs.httpSrv = &http.Server{Addr: listenAddr, Handler: hs.macaron}
 
 
 	// handle http shutdown on server context done
 	// handle http shutdown on server context done
 	go func() {
 	go func() {
 		<-ctx.Done()
 		<-ctx.Done()
+		// Hacky fix for race condition between ListenAndServe and Shutdown
+		time.Sleep(time.Millisecond * 100)
 		if err := hs.httpSrv.Shutdown(context.Background()); err != nil {
 		if err := hs.httpSrv.Shutdown(context.Background()); err != nil {
 			hs.log.Error("Failed to shutdown server", "error", err)
 			hs.log.Error("Failed to shutdown server", "error", err)
 		}
 		}
-		hs.log.Info("Stopped HTTP Server")
 	}()
 	}()
 
 
 	switch setting.Protocol {
 	switch setting.Protocol {
@@ -106,12 +114,6 @@ func (hs *HTTPServer) Start(ctx context.Context) error {
 	return err
 	return err
 }
 }
 
 
-func (hs *HTTPServer) Shutdown(ctx context.Context) error {
-	err := hs.httpSrv.Shutdown(ctx)
-	hs.log.Info("Stopped HTTP server")
-	return err
-}
-
 func (hs *HTTPServer) listenAndServeTLS(certfile, keyfile string) error {
 func (hs *HTTPServer) listenAndServeTLS(certfile, keyfile string) error {
 	if certfile == "" {
 	if certfile == "" {
 		return fmt.Errorf("cert_file cannot be empty when using HTTPS")
 		return fmt.Errorf("cert_file cannot be empty when using HTTPS")

pkg/cmd/grafana-server/main.go

@@ -39,7 +39,6 @@ var enterprise string
 var configFile = flag.String("config", "", "path to config file")
 var configFile = flag.String("config", "", "path to config file")
 var homePath = flag.String("homepath", "", "path to grafana install/home path, defaults to working directory")
 var homePath = flag.String("homepath", "", "path to grafana install/home path, defaults to working directory")
 var pidFile = flag.String("pidfile", "", "path to pid file")
 var pidFile = flag.String("pidfile", "", "path to pid file")
-var exitChan = make(chan int)
 
 
 func main() {
 func main() {
 	v := flag.Bool("v", false, "prints current version and exits")
 	v := flag.Bool("v", false, "prints current version and exits")
@@ -81,29 +80,20 @@ func main() {
 	setting.Enterprise, _ = strconv.ParseBool(enterprise)
 	setting.Enterprise, _ = strconv.ParseBool(enterprise)
 
 
 	metrics.M_Grafana_Version.WithLabelValues(version).Set(1)
 	metrics.M_Grafana_Version.WithLabelValues(version).Set(1)
-	shutdownCompleted := make(chan int)
-	server := NewGrafanaServer()
 
 
-	go listenToSystemSignals(server, shutdownCompleted)
+	server := NewGrafanaServer()
 
 
-	go func() {
-		code := 0
-		if err := server.Start(); err != nil {
-			log.Error2("Startup failed", "error", err)
-			code = 1
-		}
+	go listenToSystemSignals(server)
 
 
-		exitChan <- code
-	}()
+	err := server.Run()
 
 
-	code := <-shutdownCompleted
-	log.Info2("Grafana shutdown completed.", "code", code)
+	trace.Stop()
 	log.Close()
 	log.Close()
-	os.Exit(code)
+
+	server.Exit(err)
 }
 }
 
 
-func listenToSystemSignals(server *GrafanaServerImpl, shutdownCompleted chan int) {
-	var code int
+func listenToSystemSignals(server *GrafanaServerImpl) {
 	signalChan := make(chan os.Signal, 1)
 	signalChan := make(chan os.Signal, 1)
 	ignoreChan := make(chan os.Signal, 1)
 	ignoreChan := make(chan os.Signal, 1)
 
 
@@ -112,12 +102,6 @@ func listenToSystemSignals(server *GrafanaServerImpl, shutdownCompleted chan int
 
 
 	select {
 	select {
 	case sig := <-signalChan:
 	case sig := <-signalChan:
-		trace.Stop() // Stops trace if profiling has been enabled
-		server.Shutdown(0, fmt.Sprintf("system signal: %s", sig))
-		shutdownCompleted <- 0
-	case code = <-exitChan:
-		trace.Stop() // Stops trace if profiling has been enabled
-		server.Shutdown(code, "startup error")
-		shutdownCompleted <- code
+		server.Shutdown(fmt.Sprintf("System signal: %s", sig))
 	}
 	}
 }
 }

pkg/cmd/grafana-server/server.go

@@ -17,7 +17,6 @@ import (
 	"github.com/grafana/grafana/pkg/middleware"
 	"github.com/grafana/grafana/pkg/middleware"
 	"github.com/grafana/grafana/pkg/registry"
 	"github.com/grafana/grafana/pkg/registry"
 	"github.com/grafana/grafana/pkg/services/dashboards"
 	"github.com/grafana/grafana/pkg/services/dashboards"
-	"github.com/grafana/grafana/pkg/services/provisioning"
 
 
 	"golang.org/x/sync/errgroup"
 	"golang.org/x/sync/errgroup"
 
 
@@ -37,6 +36,7 @@ import (
 	_ "github.com/grafana/grafana/pkg/services/alerting"
 	_ "github.com/grafana/grafana/pkg/services/alerting"
 	_ "github.com/grafana/grafana/pkg/services/cleanup"
 	_ "github.com/grafana/grafana/pkg/services/cleanup"
 	_ "github.com/grafana/grafana/pkg/services/notifications"
 	_ "github.com/grafana/grafana/pkg/services/notifications"
+	_ "github.com/grafana/grafana/pkg/services/provisioning"
 	_ "github.com/grafana/grafana/pkg/services/search"
 	_ "github.com/grafana/grafana/pkg/services/search"
 )
 )
 
 
@@ -59,12 +59,14 @@ type GrafanaServerImpl struct {
 	childRoutines      *errgroup.Group
 	childRoutines      *errgroup.Group
 	log                log.Logger
 	log                log.Logger
 	cfg                *setting.Cfg
 	cfg                *setting.Cfg
+	shutdownReason     string
+	shutdownInProgress bool
 
 
 	RouteRegister api.RouteRegister `inject:""`
 	RouteRegister api.RouteRegister `inject:""`
 	HttpServer    *api.HTTPServer   `inject:""`
 	HttpServer    *api.HTTPServer   `inject:""`
 }
 }
 
 
-func (g *GrafanaServerImpl) Start() error {
+func (g *GrafanaServerImpl) Run() error {
 	g.loadConfiguration()
 	g.loadConfiguration()
 	g.writePIDFile()
 	g.writePIDFile()
 
 
@@ -75,10 +77,6 @@ func (g *GrafanaServerImpl) Start() error {
 	login.Init()
 	login.Init()
 	social.NewOAuthService()
 	social.NewOAuthService()
 
 
-	if err := provisioning.Init(g.context, setting.HomePath, g.cfg.Raw); err != nil {
-		return fmt.Errorf("Failed to provision Grafana from config. error: %v", err)
-	}
-
 	tracingCloser, err := tracing.Init(g.cfg.Raw)
 	tracingCloser, err := tracing.Init(g.cfg.Raw)
 	if err != nil {
 	if err != nil {
 		return fmt.Errorf("Tracing settings is not valid. error: %v", err)
 		return fmt.Errorf("Tracing settings is not valid. error: %v", err)
@@ -90,7 +88,6 @@ func (g *GrafanaServerImpl) Start() error {
 	serviceGraph.Provide(&inject.Object{Value: g.cfg})
 	serviceGraph.Provide(&inject.Object{Value: g.cfg})
 	serviceGraph.Provide(&inject.Object{Value: dashboards.NewProvisioningService()})
 	serviceGraph.Provide(&inject.Object{Value: dashboards.NewProvisioningService()})
 	serviceGraph.Provide(&inject.Object{Value: api.NewRouteRegister(middleware.RequestMetrics, middleware.RequestTracing)})
 	serviceGraph.Provide(&inject.Object{Value: api.NewRouteRegister(middleware.RequestMetrics, middleware.RequestTracing)})
-	serviceGraph.Provide(&inject.Object{Value: api.HTTPServer{}})
 
 
 	// self registered services
 	// self registered services
 	services := registry.GetServices()
 	services := registry.GetServices()
@@ -116,7 +113,7 @@ func (g *GrafanaServerImpl) Start() error {
 		g.log.Info("Initializing " + reflect.TypeOf(service).Elem().Name())
 		g.log.Info("Initializing " + reflect.TypeOf(service).Elem().Name())
 
 
 		if err := service.Init(); err != nil {
 		if err := service.Init(); err != nil {
-			return fmt.Errorf("Service init failed %v", err)
+			return fmt.Errorf("Service init failed: %v", err)
 		}
 		}
 	}
 	}
 
 
@@ -132,14 +129,31 @@ func (g *GrafanaServerImpl) Start() error {
 		}
 		}
 
 
 		g.childRoutines.Go(func() error {
 		g.childRoutines.Go(func() error {
+			// Skip starting new service when shutting down
+			// Can happen when service stop/return during startup
+			if g.shutdownInProgress {
+				return nil
+			}
+
 			err := service.Run(g.context)
 			err := service.Run(g.context)
+
+			// If error is not canceled then the service crashed
+			if err != context.Canceled {
+				g.log.Error("Stopped "+reflect.TypeOf(service).Elem().Name(), "reason", err)
+			} else {
 				g.log.Info("Stopped "+reflect.TypeOf(service).Elem().Name(), "reason", err)
 				g.log.Info("Stopped "+reflect.TypeOf(service).Elem().Name(), "reason", err)
+			}
+
+			// Mark that we are in shutdown mode
+			// So more services are not started
+			g.shutdownInProgress = true
 			return err
 			return err
 		})
 		})
 	}
 	}
 
 
 	sendSystemdNotification("READY=1")
 	sendSystemdNotification("READY=1")
-	return g.startHttpServer()
+
+	return g.childRoutines.Wait()
 }
 }
 
 
 func (g *GrafanaServerImpl) loadConfiguration() {
 func (g *GrafanaServerImpl) loadConfiguration() {
@@ -158,28 +172,29 @@ func (g *GrafanaServerImpl) loadConfiguration() {
 	g.cfg.LogConfigSources()
 	g.cfg.LogConfigSources()
 }
 }
 
 
-func (g *GrafanaServerImpl) startHttpServer() error {
-	g.HttpServer.Init()
-
-	err := g.HttpServer.Start(g.context)
-
-	if err != nil {
-		return fmt.Errorf("Fail to start server. error: %v", err)
-	}
-
-	return nil
-}
-
-func (g *GrafanaServerImpl) Shutdown(code int, reason string) {
-	g.log.Info("Shutdown started", "code", code, "reason", reason)
+func (g *GrafanaServerImpl) Shutdown(reason string) {
+	g.log.Info("Shutdown started", "reason", reason)
+	g.shutdownReason = reason
+	g.shutdownInProgress = true
 
 
 	// call cancel func on root context
 	// call cancel func on root context
 	g.shutdownFn()
 	g.shutdownFn()
 
 
 	// wait for child routines
 	// wait for child routines
-	if err := g.childRoutines.Wait(); err != nil && err != context.Canceled {
-		g.log.Error("Server shutdown completed", "error", err)
+	g.childRoutines.Wait()
+}
+
+func (g *GrafanaServerImpl) Exit(reason error) {
+	// default exit code is 1
+	code := 1
+
+	if reason == context.Canceled && g.shutdownReason != "" {
+		reason = fmt.Errorf(g.shutdownReason)
+		code = 0
 	}
 	}
+
+	g.log.Error("Server shutdown", "reason", reason)
+	os.Exit(code)
 }
 }
 
 
 func (g *GrafanaServerImpl) writePIDFile() {
 func (g *GrafanaServerImpl) writePIDFile() {

pkg/services/provisioning/dashboards/config_reader.go

@@ -69,7 +69,7 @@ func (cr *configReader) readConfig() ([]*DashboardsAsConfig, error) {
 
 
 		parsedDashboards, err := cr.parseConfigs(file)
 		parsedDashboards, err := cr.parseConfigs(file)
 		if err != nil {
 		if err != nil {
-
+			return nil, err
 		}
 		}
 
 
 		if len(parsedDashboards) > 0 {
 		if len(parsedDashboards) > 0 {

pkg/services/provisioning/dashboards/dashboard.go

@@ -10,19 +10,16 @@ import (
 type DashboardProvisioner struct {
 type DashboardProvisioner struct {
 	cfgReader *configReader
 	cfgReader *configReader
 	log       log.Logger
 	log       log.Logger
-	ctx       context.Context
 }
 }
 
 
-func Provision(ctx context.Context, configDirectory string) (*DashboardProvisioner, error) {
+func NewDashboardProvisioner(configDirectory string) *DashboardProvisioner {
 	log := log.New("provisioning.dashboard")
 	log := log.New("provisioning.dashboard")
 	d := &DashboardProvisioner{
 	d := &DashboardProvisioner{
 		cfgReader: &configReader{path: configDirectory, log: log},
 		cfgReader: &configReader{path: configDirectory, log: log},
 		log:       log,
 		log:       log,
-		ctx:       ctx,
 	}
 	}
 
 
-	err := d.Provision(ctx)
-	return d, err
+	return d
 }
 }
 
 
 func (provider *DashboardProvisioner) Provision(ctx context.Context) error {
 func (provider *DashboardProvisioner) Provision(ctx context.Context) error {

pkg/services/provisioning/provisioning.go

@@ -2,30 +2,40 @@ package provisioning
 
 
 import (
 import (
 	"context"
 	"context"
+	"fmt"
 	"path"
 	"path"
-	"path/filepath"
 
 
+	"github.com/grafana/grafana/pkg/registry"
 	"github.com/grafana/grafana/pkg/services/provisioning/dashboards"
 	"github.com/grafana/grafana/pkg/services/provisioning/dashboards"
 	"github.com/grafana/grafana/pkg/services/provisioning/datasources"
 	"github.com/grafana/grafana/pkg/services/provisioning/datasources"
-	ini "gopkg.in/ini.v1"
+	"github.com/grafana/grafana/pkg/setting"
 )
 )
 
 
-func Init(ctx context.Context, homePath string, cfg *ini.File) error {
-	provisioningPath := makeAbsolute(cfg.Section("paths").Key("provisioning").String(), homePath)
+func init() {
+	registry.RegisterService(&ProvisioningService{})
+}
+
+type ProvisioningService struct {
+	Cfg *setting.Cfg `inject:""`
+}
 
 
-	datasourcePath := path.Join(provisioningPath, "datasources")
+func (ps *ProvisioningService) Init() error {
+	datasourcePath := path.Join(ps.Cfg.ProvisioningPath, "datasources")
 	if err := datasources.Provision(datasourcePath); err != nil {
 	if err := datasources.Provision(datasourcePath); err != nil {
-		return err
+		return fmt.Errorf("Datasource provisioning error: %v", err)
 	}
 	}
 
 
-	dashboardPath := path.Join(provisioningPath, "dashboards")
-	_, err := dashboards.Provision(ctx, dashboardPath)
-	return err
+	return nil
 }
 }
 
 
-func makeAbsolute(path string, root string) string {
-	if filepath.IsAbs(path) {
-		return path
+func (ps *ProvisioningService) Run(ctx context.Context) error {
+	dashboardPath := path.Join(ps.Cfg.ProvisioningPath, "dashboards")
+	dashProvisioner := dashboards.NewDashboardProvisioner(dashboardPath)
+
+	if err := dashProvisioner.Provision(ctx); err != nil {
+		return err
 	}
 	}
-	return filepath.Join(root, path)
+
+	<-ctx.Done()
+	return ctx.Err()
 }
 }

pkg/setting/setting.go

@@ -56,7 +56,6 @@ var (
 	HomePath       string
 	HomePath       string
 	DataPath       string
 	DataPath       string
 	PluginsPath    string
 	PluginsPath    string
-	ProvisioningPath string
 	CustomInitPath = "conf/custom.ini"
 	CustomInitPath = "conf/custom.ini"
 
 
 	// Log settings.
 	// Log settings.
@@ -188,6 +187,9 @@ var (
 type Cfg struct {
 type Cfg struct {
 	Raw *ini.File
 	Raw *ini.File
 
 
+	// Paths
+	ProvisioningPath string
+
 	// SMTP email settings
 	// SMTP email settings
 	Smtp SmtpSettings
 	Smtp SmtpSettings
 
 
@@ -517,7 +519,7 @@ func (cfg *Cfg) Load(args *CommandLineArgs) error {
 	Env = iniFile.Section("").Key("app_mode").MustString("development")
 	Env = iniFile.Section("").Key("app_mode").MustString("development")
 	InstanceName = iniFile.Section("").Key("instance_name").MustString("unknown_instance_name")
 	InstanceName = iniFile.Section("").Key("instance_name").MustString("unknown_instance_name")
 	PluginsPath = makeAbsolute(iniFile.Section("paths").Key("plugins").String(), HomePath)
 	PluginsPath = makeAbsolute(iniFile.Section("paths").Key("plugins").String(), HomePath)
-	ProvisioningPath = makeAbsolute(iniFile.Section("paths").Key("provisioning").String(), HomePath)
+	cfg.ProvisioningPath = makeAbsolute(iniFile.Section("paths").Key("provisioning").String(), HomePath)
 	server := iniFile.Section("server")
 	server := iniFile.Section("server")
 	AppUrl, AppSubUrl = parseAppUrlAndSubUrl(server)
 	AppUrl, AppSubUrl = parseAppUrlAndSubUrl(server)
 
 
@@ -728,6 +730,6 @@ func (cfg *Cfg) LogConfigSources() {
 	logger.Info("Path Data", "path", DataPath)
 	logger.Info("Path Data", "path", DataPath)
 	logger.Info("Path Logs", "path", LogsPath)
 	logger.Info("Path Logs", "path", LogsPath)
 	logger.Info("Path Plugins", "path", PluginsPath)
 	logger.Info("Path Plugins", "path", PluginsPath)
-	logger.Info("Path Provisioning", "path", ProvisioningPath)
+	logger.Info("Path Provisioning", "path", cfg.ProvisioningPath)
 	logger.Info("App mode " + Env)
 	logger.Info("App mode " + Env)
 }
 }