From 7c242eeaef3c1ca1e772685c6aeda1d6b480c90a Mon Sep 17 00:00:00 2001 From: Todd Treece <360020+toddtreece@users.noreply.github.com> Date: Wed, 10 Sep 2025 04:06:25 -0400 Subject: [PATCH] Background Services: Add wrapper to support startup order (#110225) --- .../src/types/featureToggles.gen.ts | 5 + pkg/modules/modules.go | 28 +- .../backgroundsvcs/adapter/dependencies.go | 24 ++ .../adapter/dependencies_test.go | 28 ++ pkg/registry/backgroundsvcs/adapter/doc.go | 12 + .../backgroundsvcs/adapter/manager.go | 102 ++++++++ .../backgroundsvcs/adapter/manager_test.go | 207 +++++++++++++++ .../backgroundsvcs/adapter/service.go | 58 +++++ .../backgroundsvcs/adapter/service_test.go | 240 ++++++++++++++++++ pkg/server/server.go | 75 ++++-- pkg/server/server_test.go | 5 +- pkg/server/wire_gen.go | 4 +- pkg/services/apiserver/service.go | 4 +- pkg/services/featuremgmt/registry.go | 10 + pkg/services/featuremgmt/toggles_gen.csv | 1 + pkg/services/featuremgmt/toggles_gen.go | 4 + pkg/services/featuremgmt/toggles_gen.json | 19 ++ 17 files changed, 790 insertions(+), 36 deletions(-) create mode 100644 pkg/registry/backgroundsvcs/adapter/dependencies.go create mode 100644 pkg/registry/backgroundsvcs/adapter/dependencies_test.go create mode 100644 pkg/registry/backgroundsvcs/adapter/doc.go create mode 100644 pkg/registry/backgroundsvcs/adapter/manager.go create mode 100644 pkg/registry/backgroundsvcs/adapter/manager_test.go create mode 100644 pkg/registry/backgroundsvcs/adapter/service.go create mode 100644 pkg/registry/backgroundsvcs/adapter/service_test.go diff --git a/packages/grafana-data/src/types/featureToggles.gen.ts b/packages/grafana-data/src/types/featureToggles.gen.ts index c9138982378..6eba9a2b7d6 100644 --- a/packages/grafana-data/src/types/featureToggles.gen.ts +++ b/packages/grafana-data/src/types/featureToggles.gen.ts @@ -1143,4 +1143,9 @@ export interface FeatureToggles { * @default false */ prometheusTypeMigration?: boolean; + /** + * Enables dskit background service wrapper + * @default false + */ + dskitBackgroundServices?: boolean; } diff --git a/pkg/modules/modules.go b/pkg/modules/modules.go index ff397e8ce50..d72ec0867a8 100644 --- a/pkg/modules/modules.go +++ b/pkg/modules/modules.go @@ -26,8 +26,9 @@ var _ Manager = (*service)(nil) // service manages the registration and lifecycle of modules. type service struct { - log log.Logger - targets []string + log log.Logger + targets []string + dependencyMap map[string][]string moduleManager *modules.Manager serviceManager *services.Manager @@ -40,19 +41,36 @@ func New( logger := log.New("modules") return &service{ - log: logger, - targets: targets, + log: logger, + targets: targets, + dependencyMap: dependencyMap, moduleManager: modules.NewManager(logger), serviceMap: map[string]services.Service{}, } } +func NewWithManager( + logger log.Logger, + targets []string, + manager *modules.Manager, + dependencyMap map[string][]string, +) *service { + return &service{ + log: logger, + targets: targets, + dependencyMap: dependencyMap, + + moduleManager: manager, + serviceMap: map[string]services.Service{}, + } +} + // Run starts all registered modules. func (m *service) Run(ctx context.Context) error { var err error - for mod, targets := range dependencyMap { + for mod, targets := range m.dependencyMap { if !m.moduleManager.IsModuleRegistered(mod) { continue } diff --git a/pkg/registry/backgroundsvcs/adapter/dependencies.go b/pkg/registry/backgroundsvcs/adapter/dependencies.go new file mode 100644 index 00000000000..939121e03fb --- /dev/null +++ b/pkg/registry/backgroundsvcs/adapter/dependencies.go @@ -0,0 +1,24 @@ +package adapter + +import "github.com/grafana/grafana/pkg/modules" + +const ( + // BackgroundServices is an alias for any background service that is not explicitly listed in the dependency map. + // This includes most background services in the BackgroundServiceRegistry. + BackgroundServices = "background-services" + + // Core is an alias for a set of services that must be running before most other services can start. + Core = "core" +) + +// dependencyMap returns the module dependency relationships for the background service system. +// It defines the startup order and dependencies between different module groups. +// Background services are automatically added as dependencies to the BackgroundServices module +// unless they are explicitly listed in this map with custom dependencies. +func dependencyMap() map[string][]string { + return map[string][]string{ + modules.GrafanaAPIServer: {}, + Core: {modules.GrafanaAPIServer}, + BackgroundServices: {Core}, + } +} diff --git a/pkg/registry/backgroundsvcs/adapter/dependencies_test.go b/pkg/registry/backgroundsvcs/adapter/dependencies_test.go new file mode 100644 index 00000000000..5773d855d93 --- /dev/null +++ b/pkg/registry/backgroundsvcs/adapter/dependencies_test.go @@ -0,0 +1,28 @@ +package adapter + +import ( + "slices" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestDependencyMap(t *testing.T) { + t.Run("ensure every module except `BackgroundServices` is a dependency of another module", func(t *testing.T) { + deps := dependencyMap() + for module := range deps { + // it's safe to ignore the `BackgroundServices` module + if module == BackgroundServices { + continue + } + found := false + for _, moduleDeps := range deps { + if slices.Contains(moduleDeps, module) { + found = true + break + } + } + require.True(t, found, "module %s is not a dependency of any other module", module) + } + }) +} diff --git a/pkg/registry/backgroundsvcs/adapter/doc.go b/pkg/registry/backgroundsvcs/adapter/doc.go new file mode 100644 index 00000000000..19ba77b24dc --- /dev/null +++ b/pkg/registry/backgroundsvcs/adapter/doc.go @@ -0,0 +1,12 @@ +// This package provides an adapter layer between Grafana's background service registry +// and dskit's module and service managers. It enables Grafana background services to integrate +// with dskit's well-defined service states (New → Starting → Running → Stopping → Terminated) +// and module initialization order, allowing them to benefit from: +// +// - Coordinated service initialization +// - Observable service states and health monitoring +// - Graceful shutdown with proper cleanup ordering +// +// Background services that don't already implement the dskit's NamedService interface are +// automatically wrapped with dskit's BasicService and registered with dskit's module Manager. +package adapter diff --git a/pkg/registry/backgroundsvcs/adapter/manager.go b/pkg/registry/backgroundsvcs/adapter/manager.go new file mode 100644 index 00000000000..47cf2bfdc63 --- /dev/null +++ b/pkg/registry/backgroundsvcs/adapter/manager.go @@ -0,0 +1,102 @@ +package adapter + +import ( + "context" + "reflect" + "sync" + + "github.com/grafana/dskit/modules" + "github.com/grafana/dskit/services" + + "github.com/grafana/grafana/pkg/infra/log" + "github.com/grafana/grafana/pkg/infra/tracing" + grafanamodules "github.com/grafana/grafana/pkg/modules" + "github.com/grafana/grafana/pkg/registry" +) + +type managerAdapter struct { + reg registry.BackgroundServiceRegistry + manager grafanamodules.Engine + mu sync.RWMutex // protects manager field from concurrent access +} + +// NewManagerAdapter creates a new manager adapter that bridges Grafana's background +// service registry with dskit's module and service patterns. The adapter converts background +// services to dskit services and manages them using dskit's module Manager, which provides: +// - Coordinated service initialization +// - Observable service states and health monitoring +// - Graceful shutdown with proper cleanup ordering +func NewManagerAdapter(reg registry.BackgroundServiceRegistry) *managerAdapter { + return &managerAdapter{ + reg: reg, + } +} + +// Run initializes and starts all background services using dskit's module and service patterns. +// +// 1. Convert each registry.BackgroundService to a dskit service.NamedService (unless it already implements NamedService) +// 2. Register the services with the dskit module Manager +// 3. If the service is not already present in the dependency map, add it as a dependency of the `BackgroundServices` module +// 4. Initialize all services in the order of the dependency map +// +// Services implementing CanBeDisabled that are disabled will be skipped. +// The method blocks until the context is cancelled or a service fails. +func (r *managerAdapter) Run(ctx context.Context) error { + spanCtx, span := tracing.Start(ctx, "backgroundsvcs.adapter.Run") + defer span.End() + + logger := log.New("backgroundsvcs.adapter").FromContext(spanCtx) + manager := modules.NewManager(logger) + + deps := dependencyMap() + + for _, bgSvc := range r.reg.GetServices() { + if s, ok := bgSvc.(registry.CanBeDisabled); ok && s.IsDisabled() { + logger.Debug("service is disabled, skipping", "service", reflect.TypeOf(bgSvc).String()) + continue + } + namedService, ok := bgSvc.(services.NamedService) + if !ok { + // if the service is not a NamedService, try to convert it + namedService = asNamedService(bgSvc) + } + manager.RegisterModule(namedService.ServiceName(), func() (services.Service, error) { + return namedService, nil + }, modules.UserInvisibleModule) + + // add the service as a background service dependency if it's not already in the dependency map + if _, ok := deps[namedService.ServiceName()]; !ok { + deps[namedService.ServiceName()] = []string{Core} + deps[BackgroundServices] = append(deps[BackgroundServices], namedService.ServiceName()) + } + } + + // any modules in the dependency map that haven't been registered should be registered. + // this should only include modules like all and core. + for modName := range deps { + if manager.IsModuleRegistered(modName) { + continue + } + logger.Debug("registering virtual module", "module", modName) + manager.RegisterModule(modName, nil) + } + + r.mu.Lock() + r.manager = grafanamodules.NewWithManager(logger, []string{BackgroundServices}, manager, deps) + r.mu.Unlock() + + logger.Debug("starting background services") + return r.manager.Run(spanCtx) +} + +// Shutdown calls calls the underlying manager's Shutdown method if it has been initialized. +func (r *managerAdapter) Shutdown(ctx context.Context, reason string) error { + r.mu.RLock() + manager := r.manager + r.mu.RUnlock() + + if manager == nil { + return nil + } + return manager.Shutdown(ctx, reason) +} diff --git a/pkg/registry/backgroundsvcs/adapter/manager_test.go b/pkg/registry/backgroundsvcs/adapter/manager_test.go new file mode 100644 index 00000000000..7ea140344ee --- /dev/null +++ b/pkg/registry/backgroundsvcs/adapter/manager_test.go @@ -0,0 +1,207 @@ +package adapter + +import ( + "context" + "testing" + "time" + + "github.com/grafana/dskit/services" + "github.com/stretchr/testify/require" + + "github.com/grafana/grafana/pkg/registry" +) + +func TestNewManagerAdapter(t *testing.T) { + reg := &mockBackgroundServiceRegistry{} + adapter := NewManagerAdapter(reg) + + require.NotNil(t, adapter) + require.Equal(t, reg, adapter.reg) + require.Nil(t, adapter.manager) +} + +func TestManagerAdapter_Run(t *testing.T) { + t.Run("empty registry initializes manager", func(t *testing.T) { + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{}} + adapter := NewManagerAdapter(reg) + + // Test that Run initializes the manager properly + ctx, cancel := context.WithCancel(context.Background()) + cancel() // Cancel immediately to avoid hanging + + err := adapter.Run(ctx) + require.NoError(t, err) + require.NotNil(t, adapter.manager) + }) + + t.Run("services are registered and called", func(t *testing.T) { + mockSvc := &mockService{} + // Make the service block until context is cancelled + mockSvc.runFunc = func(ctx context.Context) error { + <-ctx.Done() + return nil + } + + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{mockSvc}} + adapter := NewManagerAdapter(reg) + + ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + defer cancel() + + err := adapter.Run(ctx) + require.NoError(t, err) + require.NotNil(t, adapter.manager) + require.True(t, mockSvc.runCalled) + }) +} + +func TestManagerAdapter_Run_ServiceTypes(t *testing.T) { + t.Run("service without NamedService interface gets converted", func(t *testing.T) { + mockSvc := &mockService{} + mockSvc.runFunc = func(ctx context.Context) error { + <-ctx.Done() + return nil + } + + reg := &mockBackgroundServiceRegistry{ + services: []registry.BackgroundService{mockSvc}, + } + adapter := NewManagerAdapter(reg) + + ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + defer cancel() + + err := adapter.Run(ctx) + require.NoError(t, err) + require.NotNil(t, adapter.manager) + + // Verify the service was called + require.True(t, mockSvc.runCalled) + }) + + t.Run("service with NamedService interface is used directly", func(t *testing.T) { + mockSvc := &mockNamedService{name: "custom-service"} + mockSvc.runFunc = func(ctx context.Context) error { + <-ctx.Done() + return nil + } + + reg := &mockBackgroundServiceRegistry{ + services: []registry.BackgroundService{mockSvc}, + } + adapter := NewManagerAdapter(reg) + + ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + defer cancel() + + err := adapter.Run(ctx) + require.NoError(t, err) + require.NotNil(t, adapter.manager) + + // Verify the service was called + require.True(t, mockSvc.runCalled) + }) + + t.Run("disabled service is skipped", func(t *testing.T) { + disabledSvc := &mockService{} + disabledSvc.disabled = true + enabledSvc := &mockService{} + enabledSvc.runFunc = func(ctx context.Context) error { + <-ctx.Done() + return nil + } + + reg := &mockBackgroundServiceRegistry{ + services: []registry.BackgroundService{disabledSvc, enabledSvc}, + } + adapter := NewManagerAdapter(reg) + + ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + defer cancel() + + err := adapter.Run(ctx) + require.NoError(t, err) + require.NotNil(t, adapter.manager) + + // Verify only enabled service was called + require.False(t, disabledSvc.runCalled) + require.True(t, enabledSvc.runCalled) + }) +} + +func TestManagerAdapter_Shutdown(t *testing.T) { + t.Run("shutdown with nil manager returns nil", func(t *testing.T) { + reg := &mockBackgroundServiceRegistry{} + adapter := NewManagerAdapter(reg) + + err := adapter.Shutdown(context.Background(), "test shutdown") + require.NoError(t, err) + }) + + t.Run("shutdown with initialized manager", func(t *testing.T) { + mockSvc := &mockService{} + mockSvc.runFunc = func(ctx context.Context) error { + <-ctx.Done() + return nil + } + + reg := &mockBackgroundServiceRegistry{ + services: []registry.BackgroundService{mockSvc}, + } + adapter := NewManagerAdapter(reg) + + // Initialize the manager + ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + defer cancel() + + go func() { + err := adapter.Run(ctx) + require.NoError(t, err) + }() + + // Give it a moment to initialize, then test shutdown + require.Eventually(t, func() bool { + return adapter.manager != nil + }, testContextTimeout, 5*time.Millisecond) + + err := adapter.Shutdown(context.Background(), "test shutdown") + require.NoError(t, err) + }) +} + +func TestManagerAdapter_MultipleServices(t *testing.T) { + t.Run("multiple services are registered", func(t *testing.T) { + mockSvc1 := &mockService{} + mockSvc2 := &mockService{} + + reg := &mockBackgroundServiceRegistry{ + services: []registry.BackgroundService{mockSvc1, mockSvc2}, + } + adapter := NewManagerAdapter(reg) + + require.NotNil(t, adapter) + require.Equal(t, reg, adapter.reg) + require.Len(t, reg.GetServices(), 2) + }) +} + +type mockBackgroundServiceRegistry struct { + services []registry.BackgroundService +} + +func (m *mockBackgroundServiceRegistry) GetServices() []registry.BackgroundService { + return m.services +} + +type mockNamedService struct { + mockService + name string +} + +func (m *mockNamedService) ServiceName() string { + return m.name +} + +func (m *mockNamedService) State() services.State { + return services.New +} diff --git a/pkg/registry/backgroundsvcs/adapter/service.go b/pkg/registry/backgroundsvcs/adapter/service.go new file mode 100644 index 00000000000..037d7f392b2 --- /dev/null +++ b/pkg/registry/backgroundsvcs/adapter/service.go @@ -0,0 +1,58 @@ +package adapter + +import ( + "context" + "errors" + "reflect" + + "github.com/grafana/dskit/services" + "github.com/grafana/grafana/pkg/registry" +) + +var _ services.NamedService = &serviceAdapter{} + +// serviceAdapter adapts a Grafana background service to dskit's NamedService interface. +// It wraps the background service with dskit's BasicService to provide the standard +// service state model: New → Starting → Running → Stopping → Terminated/Failed. +// +// The adapter uses dskit's BasicService with a custom RunningFn: +// - Starting phase: No-op, transitions immediately to Running +// - Running phase: Delegates to the wrapped service's Run method +// - Stopping phase: No-op, transitions immediately to Terminated/Failed +type serviceAdapter struct { + *services.BasicService + name string + service registry.BackgroundService +} + +// asNamedService converts a Grafana background service into a dskit NamedService. +// The returned service starts in the New state and can be managed using dskit's +// standard service operations (StartAsync, AwaitRunning, StopAsync, AwaitTerminated). +// +// The service name is derived from the Go type name using reflection, ensuring +// each service type has a unique identifier within the dskit module system. +func asNamedService(service registry.BackgroundService) *serviceAdapter { + name := reflect.TypeOf(service).String() + a := &serviceAdapter{ + name: name, + service: service, + } + a.BasicService = services.NewBasicService(nil, a.run, nil).WithName(name) + return a +} + +// run implements the RunningFn for dskit's BasicService. +// This phase keeps the service in Running state by delegating to the wrapped +// background service's Run method. If the background service completes without +// error, the adapter waits for context cancellation (service stop) before +// transitioning to Stopping state, ensuring proper dskit service lifecycle. +func (a *serviceAdapter) run(ctx context.Context) error { + err := a.service.Run(ctx) + if err != nil && !errors.Is(err, context.Canceled) { + return err + } + // wait for context cancellation to transition to Stopping state. + // this prevents the service from causing it's dependents to stop prematurely. + <-ctx.Done() + return nil +} diff --git a/pkg/registry/backgroundsvcs/adapter/service_test.go b/pkg/registry/backgroundsvcs/adapter/service_test.go new file mode 100644 index 00000000000..463ae130e76 --- /dev/null +++ b/pkg/registry/backgroundsvcs/adapter/service_test.go @@ -0,0 +1,240 @@ +package adapter + +import ( + "context" + "errors" + "reflect" + "testing" + "time" + + "github.com/grafana/dskit/services" + "github.com/grafana/grafana/pkg/registry" + "github.com/stretchr/testify/require" +) + +const ( + // Constants for timeout-based tests + testContextTimeout = 50 * time.Millisecond + expectedMinDuration = 45 * time.Millisecond +) + +func TestAsNamedService(t *testing.T) { + t.Run("creates service adapter with correct name", func(t *testing.T) { + mockSvc := &mockService{} + adapter := asNamedService(mockSvc) + + require.NotNil(t, adapter) + require.NotNil(t, adapter.BasicService) + + expectedName := reflect.TypeOf(mockSvc).String() + require.Equal(t, expectedName, adapter.name) + require.Equal(t, expectedName, adapter.ServiceName()) + require.Equal(t, mockSvc, adapter.service) + }) + + t.Run("implements NamedService interface", func(t *testing.T) { + mockSvc := &mockService{} + adapter := asNamedService(mockSvc) + + // Verify it implements the interface + require.Implements(t, (*services.NamedService)(nil), adapter) + + // Verify it has the expected methods + require.NotEmpty(t, adapter.ServiceName()) + require.Equal(t, services.New, adapter.State()) + }) + + t.Run("creates BasicService", func(t *testing.T) { + mockSvc := &mockService{} + adapter := asNamedService(mockSvc) + + require.NotNil(t, adapter.BasicService) + }) +} +func TestServiceAdapter_Run(t *testing.T) { + t.Run("run calls underlying service and waits for context", func(t *testing.T) { + mockSvc := &mockService{} + mockSvc.runFunc = func(ctx context.Context) error { + // Simulate service running until context is cancelled + <-ctx.Done() + return nil + } + + adapter := asNamedService(mockSvc) + + ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + defer cancel() + + err := adapter.run(ctx) + require.NoError(t, err) + require.True(t, mockSvc.runCalled) + }) + + t.Run("run returns error from underlying service", func(t *testing.T) { + expectedErr := errors.New("service error") + mockSvc := &mockService{} + mockSvc.runError = expectedErr + + adapter := asNamedService(mockSvc) + + err := adapter.run(context.Background()) + require.Error(t, err) + require.Equal(t, expectedErr, err) + require.True(t, mockSvc.runCalled) + }) + + t.Run("run waits for context cancellation after service completes", func(t *testing.T) { + mockSvc := &mockService{} + // Service completes immediately, adapter should wait for context + + adapter := asNamedService(mockSvc) + + ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + defer cancel() + + start := time.Now() + err := adapter.run(ctx) + duration := time.Since(start) + + require.NoError(t, err) + require.GreaterOrEqual(t, duration, expectedMinDuration) // Should wait for context timeout + require.True(t, mockSvc.runCalled) + }) + + t.Run("run with immediately cancelled context", func(t *testing.T) { + mockSvc := &mockService{} + // Service completes immediately + + adapter := asNamedService(mockSvc) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // Cancel immediately + + err := adapter.run(ctx) + require.NoError(t, err) + require.True(t, mockSvc.runCalled) + }) +} + +func TestServiceAdapter_Integration(t *testing.T) { + t.Run("full lifecycle with BasicService", func(t *testing.T) { + mockSvc := &mockService{} + mockSvc.runFunc = func(ctx context.Context) error { + <-ctx.Done() + return nil + } + + adapter := asNamedService(mockSvc) + + // Test that we can start the service + require.Equal(t, services.New, adapter.State()) + + // The BasicService should be properly configured + require.NotNil(t, adapter.BasicService) + require.Contains(t, adapter.ServiceName(), "mockService") + + require.False(t, mockSvc.runCalled) + }) + + t.Run("service name reflects actual type", func(t *testing.T) { + // Test with different service types + mockSvc1 := &mockService{} + adapter1 := asNamedService(mockSvc1) + + // Create a different type for comparison + type anotherMockService struct{ mockService } + mockSvc2 := &anotherMockService{} + adapter2 := asNamedService(mockSvc2) + + require.Contains(t, adapter1.ServiceName(), "mockService") + require.Contains(t, adapter2.ServiceName(), "anotherMockService") + require.NotEqual(t, adapter1.ServiceName(), adapter2.ServiceName()) + }) +} + +func TestServiceAdapter_ErrorHandling(t *testing.T) { + t.Run("generic error", func(t *testing.T) { + expectedErr := errors.New("generic error") + mockSvc := &mockService{} + mockSvc.runError = expectedErr + + adapter := asNamedService(mockSvc) + + t.Cleanup(func() { + adapter.StopAsync() + err := adapter.AwaitTerminated(context.Background()) + require.ErrorIs(t, err, expectedErr) + }) + + err := adapter.StartAsync(context.Background()) + require.NoError(t, err) + err = adapter.AwaitRunning(context.Background()) + require.ErrorIs(t, err, expectedErr) + require.True(t, mockSvc.runCalled) + }) + + t.Run("context error", func(t *testing.T) { + mockSvc := &mockService{} + mockSvc.runError = context.Canceled + + adapter := asNamedService(mockSvc) + + t.Cleanup(func() { + adapter.StopAsync() + err := adapter.AwaitTerminated(context.Background()) + require.NoError(t, err) + }) + + err := adapter.StartAsync(context.Background()) + require.NoError(t, err) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + err = adapter.AwaitRunning(ctx) + require.ErrorIs(t, err, context.Canceled) + }) + + t.Run("timeout error", func(t *testing.T) { + expectedErr := context.DeadlineExceeded + mockSvc := &mockService{} + mockSvc.runError = expectedErr + + adapter := asNamedService(mockSvc) + + t.Cleanup(func() { + adapter.StopAsync() + err := adapter.AwaitTerminated(context.Background()) + require.ErrorIs(t, err, expectedErr) + }) + err := adapter.StartAsync(context.Background()) + require.NoError(t, err) + err = adapter.AwaitRunning(context.Background()) + require.ErrorIs(t, err, expectedErr) + require.True(t, mockSvc.runCalled) + }) +} + +var _ registry.CanBeDisabled = &mockService{} +var _ registry.BackgroundService = &mockService{} + +type mockService struct { + runFunc func(ctx context.Context) error + runCalled bool + runContext context.Context + runError error + disabled bool +} + +func (m *mockService) Run(ctx context.Context) error { + m.runCalled = true + m.runContext = ctx + + if m.runFunc != nil { + return m.runFunc(ctx) + } + + return m.runError +} + +func (m *mockService) IsDisabled() bool { + return m.disabled +} diff --git a/pkg/server/server.go b/pkg/server/server.go index ca1263d9e4d..4d184dea2bb 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -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 { diff --git a/pkg/server/server_test.go b/pkg/server/server_test.go index 5e1d57b2fa3..b64ef62851f 100644 --- a/pkg/server/server_test.go +++ b/pkg/server/server_test.go @@ -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 } diff --git a/pkg/server/wire_gen.go b/pkg/server/wire_gen.go index a9aa865be5a..01cb3754d3d 100644 --- a/pkg/server/wire_gen.go +++ b/pkg/server/wire_gen.go @@ -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 } diff --git a/pkg/services/apiserver/service.go b/pkg/services/apiserver/service.go index c943c2ed8fc..73de7770a2a 100644 --- a/pkg/services/apiserver/service.go +++ b/pkg/services/apiserver/service.go @@ -36,7 +36,6 @@ import ( "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/middleware" "github.com/grafana/grafana/pkg/modules" - servicetracing "github.com/grafana/grafana/pkg/modules/tracing" "github.com/grafana/grafana/pkg/plugins" "github.com/grafana/grafana/pkg/registry" "github.com/grafana/grafana/pkg/registry/apis/datasource" @@ -172,8 +171,7 @@ func ProvideService( dualWriterMetrics: grafanarest.NewDualWriterMetrics(reg), } // This will be used when running as a dskit service - service := services.NewBasicService(s.start, s.running, nil).WithName(modules.GrafanaAPIServer) - s.NamedService = servicetracing.NewServiceTracer(tracing.GetTracerProvider(), service) + s.NamedService = services.NewBasicService(s.start, s.running, nil).WithName(modules.GrafanaAPIServer) // TODO: this is very hacky // We need to register the routes in ProvideService to make sure diff --git a/pkg/services/featuremgmt/registry.go b/pkg/services/featuremgmt/registry.go index 1403ef2e92c..c472625ee26 100644 --- a/pkg/services/featuremgmt/registry.go +++ b/pkg/services/featuremgmt/registry.go @@ -1983,6 +1983,16 @@ var ( Owner: grafanaPartnerPluginsSquad, Expression: "false", }, + { + Name: "dskitBackgroundServices", + Description: "Enables dskit background service wrapper", + HideFromAdminPage: true, + HideFromDocs: true, + Stage: FeatureStageExperimental, + RequiresRestart: true, + Owner: grafanaPluginsPlatformSquad, + Expression: "false", + }, } ) diff --git a/pkg/services/featuremgmt/toggles_gen.csv b/pkg/services/featuremgmt/toggles_gen.csv index 64593d8684d..0fbac6f8625 100644 --- a/pkg/services/featuremgmt/toggles_gen.csv +++ b/pkg/services/featuremgmt/toggles_gen.csv @@ -254,3 +254,4 @@ alertingTriage,experimental,@grafana/alerting-squad,false,false,true graphiteBackendMode,privatePreview,@grafana/partner-datasources,false,false,false azureResourcePickerUpdates,preview,@grafana/partner-datasources,false,false,true prometheusTypeMigration,experimental,@grafana/partner-datasources,false,true,false +dskitBackgroundServices,experimental,@grafana/plugins-platform-backend,false,true,false diff --git a/pkg/services/featuremgmt/toggles_gen.go b/pkg/services/featuremgmt/toggles_gen.go index 0fa4b06a1e9..363f0241847 100644 --- a/pkg/services/featuremgmt/toggles_gen.go +++ b/pkg/services/featuremgmt/toggles_gen.go @@ -1026,4 +1026,8 @@ const ( // FlagPrometheusTypeMigration // Checks for deprecated Prometheus authentication methods (SigV4 and Azure), installs the relevant data source, and migrates the Prometheus data sources FlagPrometheusTypeMigration = "prometheusTypeMigration" + + // FlagDskitBackgroundServices + // Enables dskit background service wrapper + FlagDskitBackgroundServices = "dskitBackgroundServices" ) diff --git a/pkg/services/featuremgmt/toggles_gen.json b/pkg/services/featuremgmt/toggles_gen.json index e36ed285fe3..4dfb4eb567e 100644 --- a/pkg/services/featuremgmt/toggles_gen.json +++ b/pkg/services/featuremgmt/toggles_gen.json @@ -1169,6 +1169,25 @@ "codeowner": "@grafana/observability-metrics" } }, + { + "metadata": { + "name": "dskitBackgroundServices", + "resourceVersion": "1757339637779", + "creationTimestamp": "2025-09-03T12:20:24Z", + "annotations": { + "grafana.app/updatedTimestamp": "2025-09-08 13:53:57.77994 +0000 UTC" + } + }, + "spec": { + "description": "Enables dskit background service wrapper", + "stage": "experimental", + "codeowner": "@grafana/plugins-platform-backend", + "requiresRestart": true, + "hideFromAdminPage": true, + "hideFromDocs": true, + "expression": "false" + } + }, { "metadata": { "name": "editPanelCSVDragAndDrop",