public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Chore: Convert background service registry to dskit module (#64062)

Browse Source 
* Chore: Add initial support for deployment modes

* revert CLI changes and start modules independently

* add modules to codeowners

* additional comments

* add Engine and Manager interface to fix test issues

* convert background service registry to dskit module

* remove extra context from serviceListener logger

Co-authored-by: Will Browne <wbrowne@users.noreply.github.com>

* Remove whitespace

* fix import

* undo ide changes

* only register All by default

* with registry

* add test

* add comments

* re-add debug log

* fix import

* reorganize arg

* undo kind changes

* add provide service test

* fix import

* rejig systemd calls

* update codeowners

---------

Co-authored-by: Todd Treece <todd.treece@grafana.com>
Co-authored-by: Todd Treece <360020+toddtreece@users.noreply.github.com>
This commit is contained in:
Will Browne
2023-07-06 14:45:47 +02:00
committed by GitHub
co-authored by Todd Treece Todd Treece
parent 76974009d0
commit 4818568c65
15 changed files with 287 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/server/server.go
+22 -109
View File
@@ -7,12 +7,9 @@ import (
"net"
"os"
"path/filepath"
"reflect"
"strconv"
"sync"
"golang.org/x/sync/errgroup"
"github.com/grafana/grafana/pkg/api"
_ "github.com/grafana/grafana/pkg/extensions"
"github.com/grafana/grafana/pkg/infra/log"
@@ -38,18 +35,18 @@ type Options struct {
// New returns a new instance of Server.
func New(opts Options, cfg *setting.Cfg, httpServer *api.HTTPServer, roleRegistry accesscontrol.RoleRegistry,
provisioningService provisioning.ProvisioningService, backgroundServiceProvider registry.BackgroundServiceRegistry,
provisioningService provisioning.ProvisioningService,
usageStatsProvidersRegistry registry.UsageStatsProvidersRegistry, statsCollectorService *statscollector.Service,
moduleService modules.Engine,
_ moduleRegistry.Registry, // imported to invoke initialization via Wire
) (*Server, error) {
statsCollectorService.RegisterProviders(usageStatsProvidersRegistry.GetServices())
s, err := newServer(opts, cfg, httpServer, roleRegistry, provisioningService, backgroundServiceProvider, moduleService)
s, err := newServer(opts, cfg, httpServer, roleRegistry, provisioningService, moduleService)
if err != nil {
return nil, err
}
if err := s.init(); err != nil {
if err = s.init(context.Background()); err != nil {
return nil, err
}
@@ -57,19 +54,12 @@ func New(opts Options, cfg *setting.Cfg, httpServer *api.HTTPServer, roleRegistr
}
func newServer(opts Options, cfg *setting.Cfg, httpServer *api.HTTPServer, roleRegistry accesscontrol.RoleRegistry,
provisioningService provisioning.ProvisioningService, backgroundServiceProvider registry.BackgroundServiceRegistry,
moduleService modules.Engine,
) (*Server, error) {
rootCtx, shutdownFn := context.WithCancel(context.Background())
childRoutines, childCtx := errgroup.WithContext(rootCtx)
s := &Server{
context: childCtx,
childRoutines: childRoutines,
provisioningService provisioning.ProvisioningService,
moduleService modules.Engine) (*Server, error) {
return &Server{
HTTPServer: httpServer,
provisioningService: provisioningService,
roleRegistry: roleRegistry,
shutdownFn: shutdownFn,
shutdownFinished: make(chan struct{}),
log: log.New("server"),
cfg: cfg,
@@ -77,18 +67,12 @@ func newServer(opts Options, cfg *setting.Cfg, httpServer *api.HTTPServer, roleR
version: opts.Version,
commit: opts.Commit,
buildBranch: opts.BuildBranch,
backgroundServices: backgroundServiceProvider.GetServices(),
moduleService: moduleService,
}
return s, nil
}, nil
}
// Server is responsible for managing the lifecycle of services.
type Server struct {
context context.Context
shutdownFn context.CancelFunc
childRoutines *errgroup.Group
log log.Logger
cfg *setting.Cfg
shutdownOnce sync.Once
@@ -96,11 +80,10 @@ type Server struct {
isInitialized bool
mtx sync.Mutex
pidFile string
version string
commit string
buildBranch string
backgroundServices []registry.BackgroundService
pidFile string
version string
commit string
buildBranch string
HTTPServer *api.HTTPServer
roleRegistry accesscontrol.RoleRegistry
@@ -109,7 +92,7 @@ type Server struct {
}
// init initializes the server and its services.
func (s *Server) init() error {
func (s *Server) init(ctx context.Context) error {
s.mtx.Lock()
defer s.mtx.Unlock()
@@ -123,7 +106,7 @@ func (s *Server) init() error {
}
// Initialize dskit modules.
if err := s.moduleService.Init(s.context); err != nil {
if err := s.moduleService.Init(ctx); err != nil {
return err
}
@@ -131,65 +114,27 @@ func (s *Server) init() error {
return err
}
if err := s.roleRegistry.RegisterFixedRoles(s.context); err != nil {
if err := s.roleRegistry.RegisterFixedRoles(ctx); err != nil {
return err
}
return s.provisioningService.RunInitProvisioners(s.context)
return s.provisioningService.RunInitProvisioners(ctx)
}
// Run initializes and starts services. This will block until all services have
// exited. To initiate shutdown, call the Shutdown method in another goroutine.
func (s *Server) Run() error {
func (s *Server) Run(ctx context.Context) error {
defer close(s.shutdownFinished)
if err := s.init(); err != nil {
if err := s.init(ctx); err != nil {
return err
}
// Start dskit modules.
s.childRoutines.Go(func() error {
err := s.moduleService.Run(s.context)
if err != nil && !errors.Is(err, context.Canceled) {
return err
}
return nil
})
services := s.backgroundServices
// Start background services.
for _, svc := range services {
if registry.IsDisabled(svc) {
continue
}
service := svc
serviceName := reflect.TypeOf(service).String()
s.childRoutines.Go(func() error {
select {
case <-s.context.Done():
return s.context.Err()
default:
}
s.log.Debug("Starting background service", "service", serviceName)
err := service.Run(s.context)
// Do not return context.Canceled error since errgroup.Group only
// returns the first error to the caller - thus we can miss a more
// interesting error.
if err != nil && !errors.Is(err, context.Canceled) {
s.log.Error("Stopped background service", "service", serviceName, "reason", err)
return fmt.Errorf("%s run error: %w", serviceName, err)
}
s.log.Debug("Stopped background service", "service", serviceName, "reason", err)
return nil
})
err := s.moduleService.Run(ctx)
if err != nil && !errors.Is(err, context.Canceled) {
return err
}
s.notifySystemd("READY=1")
s.log.Debug("Waiting on services...")
return s.childRoutines.Wait()
return nil
}
// Shutdown initiates Grafana graceful shutdown. This shuts down all
@@ -199,11 +144,9 @@ func (s *Server) Shutdown(ctx context.Context, reason string) error {
var err error
s.shutdownOnce.Do(func() {
s.log.Info("Shutdown started", "reason", reason)
if err := s.moduleService.Shutdown(ctx); err != nil {
if err = s.moduleService.Shutdown(ctx); err != nil {
s.log.Error("Failed to shutdown modules", "error", err)
}
// Call cancel func to stop background services.
s.shutdownFn()
// Wait for server to shut down
select {
case <-s.shutdownFinished:
@@ -240,33 +183,3 @@ func (s *Server) writePIDFile() error {
s.log.Info("Writing PID file", "path", s.pidFile, "pid", pid)
return nil
}
// notifySystemd sends state notifications to systemd.
func (s *Server) notifySystemd(state string) {
notifySocket := os.Getenv("NOTIFY_SOCKET")
if notifySocket == "" {
s.log.Debug(
"NOTIFY_SOCKET environment variable empty or unset, can't send systemd notification")
return
}
socketAddr := &net.UnixAddr{
Name: notifySocket,
Net: "unixgram",
}
conn, err := net.DialUnix(socketAddr.Net, nil, socketAddr)
if err != nil {
s.log.Warn("Failed to connect to systemd", "err", err, "socket", notifySocket)
return
}
defer func() {
if err := conn.Close(); err != nil {
s.log.Warn("Failed to close connection", "err", err)
}
}()
_, err = conn.Write([]byte(state))
if err != nil {
s.log.Warn("Failed to write notification to systemd", "err", err)
}
}