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

Chore: Revert dskit service additions (#72608)

Browse Source
This commit is contained in:
Todd Treece
2023-08-03 09:19:01 -04:00
committed by GitHub
parent f6e836302b
commit f3ffc850aa
27 changed files with 309 additions and 690 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/server/server.go
+127 -38
View File
@@ -7,18 +7,21 @@ 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"
"github.com/grafana/grafana/pkg/infra/metrics"
"github.com/grafana/grafana/pkg/infra/usagestats/statscollector"
"github.com/grafana/grafana/pkg/modules"
moduleRegistry "github.com/grafana/grafana/pkg/modules/registry"
"github.com/grafana/grafana/pkg/registry"
"github.com/grafana/grafana/pkg/services/accesscontrol"
"github.com/grafana/grafana/pkg/services/provisioning"
"github.com/grafana/grafana/pkg/setting"
)
@@ -34,17 +37,17 @@ 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,
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, moduleService)
s, err := newServer(opts, cfg, httpServer, roleRegistry, provisioningService, backgroundServiceProvider, moduleService)
if err != nil {
return nil, err
}
if err = s.init(context.Background()); err != nil {
if err := s.init(); err != nil {
return nil, err
}
@@ -52,23 +55,38 @@ func New(opts Options, cfg *setting.Cfg, httpServer *api.HTTPServer, roleRegistr
}
func newServer(opts Options, cfg *setting.Cfg, httpServer *api.HTTPServer, roleRegistry accesscontrol.RoleRegistry,
moduleService modules.Engine) (*Server, error) {
return &Server{
HTTPServer: httpServer,
roleRegistry: roleRegistry,
shutdownFinished: make(chan struct{}),
log: log.New("server"),
cfg: cfg,
pidFile: opts.PidFile,
version: opts.Version,
commit: opts.Commit,
buildBranch: opts.BuildBranch,
moduleService: moduleService,
}, nil
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,
HTTPServer: httpServer,
provisioningService: provisioningService,
roleRegistry: roleRegistry,
shutdownFn: shutdownFn,
shutdownFinished: make(chan struct{}),
log: log.New("server"),
cfg: cfg,
pidFile: opts.PidFile,
version: opts.Version,
commit: opts.Commit,
buildBranch: opts.BuildBranch,
backgroundServices: backgroundServiceProvider.GetServices(),
moduleService: moduleService,
}
return s, 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
@@ -76,18 +94,20 @@ type Server struct {
isInitialized bool
mtx sync.Mutex
pidFile string
version string
commit string
buildBranch string
pidFile string
version string
commit string
buildBranch string
backgroundServices []registry.BackgroundService
HTTPServer *api.HTTPServer
roleRegistry accesscontrol.RoleRegistry
moduleService modules.Engine
HTTPServer *api.HTTPServer
roleRegistry accesscontrol.RoleRegistry
provisioningService provisioning.ProvisioningService
moduleService modules.Engine
}
// init initializes the server and its services.
func (s *Server) init(ctx context.Context) error {
func (s *Server) init() error {
s.mtx.Lock()
defer s.mtx.Unlock()
@@ -101,7 +121,7 @@ func (s *Server) init(ctx context.Context) error {
}
// Initialize dskit modules.
if err := s.moduleService.Init(ctx); err != nil {
if err := s.moduleService.Init(s.context); err != nil {
return err
}
@@ -109,28 +129,65 @@ func (s *Server) init(ctx context.Context) error {
return err
}
return s.roleRegistry.RegisterFixedRoles(ctx)
}
if err := s.roleRegistry.RegisterFixedRoles(s.context); err != nil {
return err
}
// AwaitHealthy waits for the server to become healthy.
func (s *Server) AwaitHealthy(ctx context.Context) error {
return s.moduleService.AwaitHealthy(ctx)
return s.provisioningService.RunInitProvisioners(s.context)
}
// 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(ctx context.Context) error {
func (s *Server) Run() error {
defer close(s.shutdownFinished)
if err := s.init(ctx); err != nil {
if err := s.init(); err != nil {
return err
}
err := s.moduleService.Run(ctx)
if err != nil && !errors.Is(err, context.Canceled) {
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
})
}
return nil
s.notifySystemd("READY=1")
s.log.Debug("Waiting on services...")
return s.childRoutines.Wait()
}
// Shutdown initiates Grafana graceful shutdown. This shuts down all
@@ -140,9 +197,11 @@ 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:
@@ -179,3 +238,33 @@ 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)
}
}
pkg/server/server_test.go
+72 -45
View File
@@ -3,19 +3,52 @@ package server
import (
"context"
"errors"
"fmt"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/grafana/grafana/pkg/modules"
"github.com/grafana/grafana/pkg/registry"
"github.com/grafana/grafana/pkg/registry/backgroundsvcs"
"github.com/grafana/grafana/pkg/services/accesscontrol/acimpl"
"github.com/grafana/grafana/pkg/setting"
)
func testServer(t *testing.T, m *modules.MockModuleEngine) *Server {
type testService struct {
started chan struct{}
runErr error
isDisabled bool
}
func newTestService(runErr error, disabled bool) *testService {
return &testService{
started: make(chan struct{}),
runErr: runErr,
isDisabled: disabled,
}
}
func (s *testService) Run(ctx context.Context) error {
if s.isDisabled {
return fmt.Errorf("Shouldn't run disabled service")
}
if s.runErr != nil {
return s.runErr
}
close(s.started)
<-ctx.Done()
return ctx.Err()
}
func (s *testService) IsDisabled() bool {
return s.isDisabled
}
func testServer(t *testing.T, services ...registry.BackgroundService) *Server {
t.Helper()
s, err := newServer(Options{}, setting.NewCfg(), nil, &acimpl.Service{}, m)
s, err := newServer(Options{}, setting.NewCfg(), nil, &acimpl.Service{}, nil, backgroundsvcs.NewBackgroundServiceRegistry(services...), &MockModuleService{})
require.NoError(t, err)
// Required to skip configuration initialization that causes
// DI errors in this test.
@@ -25,68 +58,62 @@ func testServer(t *testing.T, m *modules.MockModuleEngine) *Server {
func TestServer_Run_Error(t *testing.T) {
testErr := errors.New("boom")
t.Run("Modules Run error bubbles up", func(t *testing.T) {
ctx := context.Background()
s := testServer(t, &modules.MockModuleEngine{
RunFunc: func(c context.Context) error {
require.Equal(t, ctx, c)
return testErr
},
})
err := s.Run(ctx)
require.ErrorIs(t, err, testErr)
})
s := testServer(t, newTestService(nil, false), newTestService(testErr, false))
err := s.Run()
require.ErrorIs(t, err, testErr)
}
func TestServer_Shutdown(t *testing.T) {
ctx := context.Background()
modulesShutdown := false
s := testServer(t, &modules.MockModuleEngine{
ShutdownFunc: func(_ context.Context) error {
modulesShutdown = true
return nil
},
})
s := testServer(t, newTestService(nil, false), newTestService(nil, true))
ch := make(chan error)
go func() {
defer close(ch)
// Wait until all services launched.
for _, svc := range s.backgroundServices {
if !svc.(*testService).isDisabled {
<-svc.(*testService).started
}
}
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
err := s.Shutdown(ctx, "test interrupt")
ch <- err
}()
err := s.Run(ctx)
err := s.Run()
require.NoError(t, err)
err = <-ch
require.NoError(t, err)
require.True(t, modulesShutdown)
}
t.Run("Modules Shutdown error bubbles up", func(t *testing.T) {
testErr := errors.New("boom")
type MockModuleService struct {
initFunc func(context.Context) error
runFunc func(context.Context) error
shutdownFunc func(context.Context) error
}
s = testServer(t, &modules.MockModuleEngine{
ShutdownFunc: func(_ context.Context) error {
return testErr
},
})
func (m *MockModuleService) Init(ctx context.Context) error {
if m.initFunc != nil {
return m.initFunc(ctx)
}
return nil
}
ch = make(chan error)
go func() {
defer close(ch)
ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
err = s.Shutdown(ctx, "test interrupt")
ch <- err
}()
err = s.Run(ctx)
require.NoError(t, err)
func (m *MockModuleService) Run(ctx context.Context) error {
if m.runFunc != nil {
return m.runFunc(ctx)
}
return nil
}
err = <-ch
require.ErrorIs(t, err, testErr)
})
func (m *MockModuleService) Shutdown(ctx context.Context) error {
if m.shutdownFunc != nil {
return m.shutdownFunc(ctx)
}
return nil
}
pkg/server/wire.go
-4
View File
@@ -31,9 +31,7 @@ import (
"github.com/grafana/grafana/pkg/middleware/csrf"
"github.com/grafana/grafana/pkg/middleware/loggermw"
"github.com/grafana/grafana/pkg/modules"
moduleRegistry "github.com/grafana/grafana/pkg/modules/registry"
pluginDashboards "github.com/grafana/grafana/pkg/plugins/manager/dashboards"
"github.com/grafana/grafana/pkg/registry/backgroundsvcs"
"github.com/grafana/grafana/pkg/registry/corekind"
"github.com/grafana/grafana/pkg/services/accesscontrol"
"github.com/grafana/grafana/pkg/services/accesscontrol/acimpl"
@@ -363,8 +361,6 @@ var wireBasicSet = wire.NewSet(
wire.Bind(new(oauthserver.OAuth2Server), new(*oasimpl.OAuth2ServiceImpl)),
loggermw.Provide,
modules.WireSet,
moduleRegistry.WireSet,
backgroundsvcs.ProvideBackgroundServiceRunner,
signingkeysimpl.ProvideEmbeddedSigningKeysService,
wire.Bind(new(signingkeys.Service), new(*signingkeysimpl.Service)),
)