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

Background Services: Add wrapper to support startup order (#110225)

Browse Source
This commit is contained in:
Todd Treece
2025-09-10 09:06:25 +01:00
committed by GitHub
parent 60d26ac0ce
commit 7c242eeaef
17 changed files with 790 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/server/server.go
+51 -24
View File
@@ -19,9 +19,12 @@ import (
_ "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/tracing"
"github.com/grafana/grafana/pkg/infra/usagestats/statscollector"
"github.com/grafana/grafana/pkg/registry"
"github.com/grafana/grafana/pkg/registry/backgroundsvcs/adapter"
"github.com/grafana/grafana/pkg/services/accesscontrol"
"github.com/grafana/grafana/pkg/services/featuremgmt"
"github.com/grafana/grafana/pkg/services/provisioning"
"github.com/grafana/grafana/pkg/setting"
)
@@ -40,10 +43,11 @@ type Options struct {
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,
tracerProvider *tracing.TracingService,
promReg prometheus.Registerer,
) (*Server, error) {
statsCollectorService.RegisterProviders(usageStatsProvidersRegistry.GetServices())
s, err := newServer(opts, cfg, httpServer, roleRegistry, provisioningService, backgroundServiceProvider, promReg)
s, err := newServer(opts, cfg, httpServer, roleRegistry, provisioningService, backgroundServiceProvider, tracerProvider, promReg)
if err != nil {
return nil, err
}
@@ -57,27 +61,29 @@ 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,
tracerProvider *tracing.TracingService,
promReg prometheus.Registerer,
) (*Server, error) {
rootCtx, shutdownFn := context.WithCancel(context.Background())
childRoutines, childCtx := errgroup.WithContext(rootCtx)
s := &Server{
promReg: promReg,
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(),
promReg: promReg,
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,
backgroundServiceRegistry: backgroundServiceProvider,
tracerProvider: tracerProvider,
}
return s, nil
@@ -97,11 +103,13 @@ 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
backgroundServiceRegistry registry.BackgroundServiceRegistry
tracerProvider *tracing.TracingService
HTTPServer *api.HTTPServer
roleRegistry accesscontrol.RoleRegistry
@@ -134,16 +142,35 @@ func (s *Server) Init() error {
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() error {
if s.cfg.IsFeatureToggleEnabled(featuremgmt.FlagDskitBackgroundServices) {
s.log.Debug("Running background services with dskit wrapper")
return s.dskitRun()
}
s.log.Debug("Running standard background services")
return s.backgroundServicesRun()
}
func (s *Server) dskitRun() error {
defer close(s.shutdownFinished)
if err := s.Init(); err != nil {
return err
}
managerAdapter := adapter.NewManagerAdapter(s.backgroundServiceRegistry)
s.notifySystemd("READY=1")
return managerAdapter.Run(s.context)
}
func (s *Server) backgroundServicesRun() error {
defer close(s.shutdownFinished)
if err := s.Init(); err != nil {
return err
}
services := s.backgroundServices
services := s.backgroundServiceRegistry.GetServices()
// Start background services.
for _, svc := range services {
pkg/server/server_test.go
+3 -2
View File
@@ -10,6 +10,7 @@ import (
"github.com/prometheus/client_golang/prometheus"
"github.com/stretchr/testify/require"
"github.com/grafana/grafana/pkg/infra/tracing"
"github.com/grafana/grafana/pkg/registry"
"github.com/grafana/grafana/pkg/registry/backgroundsvcs"
"github.com/grafana/grafana/pkg/services/accesscontrol/acimpl"
@@ -49,7 +50,7 @@ func (s *testService) IsDisabled() bool {
func testServer(t *testing.T, services ...registry.BackgroundService) *Server {
t.Helper()
s, err := newServer(Options{}, setting.NewCfg(), nil, &acimpl.Service{}, nil, backgroundsvcs.NewBackgroundServiceRegistry(services...), prometheus.NewRegistry())
s, err := newServer(Options{}, setting.NewCfg(), nil, &acimpl.Service{}, nil, backgroundsvcs.NewBackgroundServiceRegistry(services...), tracing.NewNoopTracerService(), prometheus.NewRegistry())
require.NoError(t, err)
// Required to skip configuration initialization that causes
// DI errors in this test.
@@ -75,7 +76,7 @@ func TestServer_Shutdown(t *testing.T) {
defer close(ch)
// Wait until all services launched.
for _, svc := range s.backgroundServices {
for _, svc := range s.backgroundServiceRegistry.GetServices() {
if !svc.(*testService).isDisabled {
<-svc.(*testService).started
}
pkg/server/wire_gen.go
+2 -2
View File
@@ -864,7 +864,7 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api
registration := authnimpl.ProvideRegistration(cfg, authnService, orgService, userAuthTokenService, acimplService, permissionRegistry, apikeyService, userService, authService, ossUserProtectionImpl, loginattemptimplService, quotaService, authinfoimplService, renderingService, featureToggles, oauthtokenService, socialService, remoteCache, ldapImpl, ossImpl, tracingService, tempuserService, notificationService)
backgroundServiceRegistry := backgroundsvcs.ProvideBackgroundServiceRegistry(httpServer, alertNG, cleanUpService, grafanaLive, gateway, notificationService, pluginstoreService, renderingService, userAuthTokenService, tracingService, provisioningServiceImpl, usageStats, statscollectorService, grafanaService, pluginsService, internalMetricsService, secretsService, remoteCache, storageService, searchService, entityEventsService, serviceAccountsService, grpcserverProvider, secretMigrationProviderImpl, loginattemptimplService, supportbundlesimplService, metricService, keyRetriever, angulardetectorsproviderDynamic, apiserverService, anonDeviceService, ssosettingsimplService, pluginexternalService, plugininstallerService, zanzanaReconciler, appregistryService, dashboardUpdater, dashboardServiceImpl, worker, serviceImpl, serviceAccountsProxy, healthService, reflectionService, apiService, apiregistryService, idimplService, teamAPI, ssosettingsimplService, cloudmigrationService, registration)
usageStatsProvidersRegistry := usagestatssvcs.ProvideUsageStatsProvidersRegistry(acimplService, userService)
server, err := New(opts, cfg, httpServer, acimplService, provisioningServiceImpl, backgroundServiceRegistry, usageStatsProvidersRegistry, statscollectorService, registerer)
server, err := New(opts, cfg, httpServer, acimplService, provisioningServiceImpl, backgroundServiceRegistry, usageStatsProvidersRegistry, statscollectorService, tracingService, registerer)
if err != nil {
return nil, err
}
@@ -1453,7 +1453,7 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac
registration := authnimpl.ProvideRegistration(cfg, authnService, orgService, userAuthTokenService, acimplService, permissionRegistry, apikeyService, userService, authService, ossUserProtectionImpl, loginattemptimplService, quotaService, authinfoimplService, renderingService, featureToggles, oauthtokentestService, socialService, remoteCache, ldapImpl, ossImpl, tracingService, tempuserService, notificationServiceMock)
backgroundServiceRegistry := backgroundsvcs.ProvideBackgroundServiceRegistry(httpServer, alertNG, cleanUpService, grafanaLive, gateway, notificationService, pluginstoreService, renderingService, userAuthTokenService, tracingService, provisioningServiceImpl, usageStats, statscollectorService, grafanaService, pluginsService, internalMetricsService, secretsService, remoteCache, storageService, searchService, entityEventsService, serviceAccountsService, grpcserverProvider, secretMigrationProviderImpl, loginattemptimplService, supportbundlesimplService, metricService, keyRetriever, angulardetectorsproviderDynamic, apiserverService, anonDeviceService, ssosettingsimplService, pluginexternalService, plugininstallerService, zanzanaReconciler, appregistryService, dashboardUpdater, dashboardServiceImpl, worker, serviceImpl, serviceAccountsProxy, healthService, reflectionService, apiService, apiregistryService, idimplService, teamAPI, ssosettingsimplService, cloudmigrationService, registration)
usageStatsProvidersRegistry := usagestatssvcs.ProvideUsageStatsProvidersRegistry(acimplService, userService)
server, err := New(opts, cfg, httpServer, acimplService, provisioningServiceImpl, backgroundServiceRegistry, usageStatsProvidersRegistry, statscollectorService, registerer)
server, err := New(opts, cfg, httpServer, acimplService, provisioningServiceImpl, backgroundServiceRegistry, usageStatsProvidersRegistry, statscollectorService, tracingService, registerer)
if err != nil {
return nil, err
}