From 191ddb127a91771878643888e57550f47f48a7a3 Mon Sep 17 00:00:00 2001 From: Todd Treece <360020+toddtreece@users.noreply.github.com> Date: Wed, 10 Sep 2025 14:33:03 -0400 Subject: [PATCH] Background Services: Lifecycle o11y (#110570) --- pkg/infra/tracing/tracing.go | 23 +- pkg/modules/modules.go | 119 ++-- pkg/modules/tracing/listener.go | 117 ++++ pkg/modules/tracing/listener_test.go | 617 ++++++++++++++++++ pkg/modules/tracing/manager.go | 70 ++ pkg/modules/tracing/manager_test.go | 342 ++++++++++ pkg/modules/tracing/service.go | 35 - .../backgroundsvcs/adapter/dependencies.go | 29 +- .../backgroundsvcs/adapter/manager.go | 152 +++-- .../backgroundsvcs/adapter/manager_test.go | 479 ++++++++++---- .../backgroundsvcs/adapter/service_test.go | 137 +--- pkg/server/module_server.go | 2 +- pkg/server/server.go | 32 +- pkg/services/apiserver/service.go | 1 - 14 files changed, 1736 insertions(+), 419 deletions(-) create mode 100644 pkg/modules/tracing/listener.go create mode 100644 pkg/modules/tracing/listener_test.go create mode 100644 pkg/modules/tracing/manager.go create mode 100644 pkg/modules/tracing/manager_test.go delete mode 100644 pkg/modules/tracing/service.go diff --git a/pkg/infra/tracing/tracing.go b/pkg/infra/tracing/tracing.go index 43cdc9a9d56..f8b340d12df 100644 --- a/pkg/infra/tracing/tracing.go +++ b/pkg/infra/tracing/tracing.go @@ -29,11 +29,13 @@ import ( "github.com/go-kit/log/level" + "github.com/grafana/dskit/services" "github.com/grafana/grafana/pkg/apimachinery/errutil" "github.com/grafana/grafana/pkg/infra/log" ) const ( + ServiceName = "tracing" envJaegerAgentHost = "JAEGER_AGENT_HOST" envJaegerAgentPort = "JAEGER_AGENT_PORT" ) @@ -48,6 +50,8 @@ const ( ) type TracingService struct { + services.NamedService + cfg *TracingConfig log log.Logger @@ -93,6 +97,7 @@ func ProvideService(tracingCfg *TracingConfig) (*TracingService, error) { cfg: tracingCfg, log: log.New("tracing"), } + ots.NamedService = services.NewBasicService(ots.starting, ots.running, ots.stopping).WithName(ServiceName) if err := ots.initOpentelemetryTracer(); err != nil { return nil, err @@ -306,27 +311,35 @@ func (ots *TracingService) initOpentelemetryTracer() error { return nil } -func (ots *TracingService) Run(ctx context.Context) error { +func (ots *TracingService) starting(ctx context.Context) error { otel.SetErrorHandler(otel.ErrorHandlerFunc(func(err error) { err = level.Error(ots.log).Log("msg", "OpenTelemetry handler returned an error", "err", err) if err != nil { ots.log.Error("OpenTelemetry log returning error", err) } })) + return nil +} +func (ots *TracingService) running(ctx context.Context) error { <-ctx.Done() + return nil +} +func (ots *TracingService) stopping(_ error) error { ots.log.Info("Closing tracing") if ots.tracerProvider == nil { return nil } - ctxShutdown, cancel := context.WithTimeout(ctx, time.Second*5) + ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) defer cancel() + return ots.tracerProvider.Shutdown(ctx) +} - if err := ots.tracerProvider.Shutdown(ctxShutdown); err != nil { +func (ots *TracingService) Run(ctx context.Context) error { + if err := ots.StartAsync(ctx); err != nil { return err } - - return nil + return ots.AwaitTerminated(ctx) } func (ots *TracingService) Inject(ctx context.Context, header http.Header, _ trace.Span) { diff --git a/pkg/modules/modules.go b/pkg/modules/modules.go index d72ec0867a8..286f8e5e350 100644 --- a/pkg/modules/modules.go +++ b/pkg/modules/modules.go @@ -3,12 +3,13 @@ package modules import ( "context" "errors" - "strings" "github.com/grafana/dskit/modules" "github.com/grafana/dskit/services" "github.com/grafana/grafana/pkg/infra/log" + infratracing "github.com/grafana/grafana/pkg/infra/tracing" + "github.com/grafana/grafana/pkg/modules/tracing" ) type Engine interface { @@ -16,60 +17,57 @@ type Engine interface { Shutdown(context.Context, string) error } +type Registry interface { + RegisterModule(name string, fn func() (services.Service, error)) + RegisterInvisibleModule(name string, fn func() (services.Service, error)) +} + type Manager interface { - RegisterModule(name string, fn initFn) - RegisterInvisibleModule(name string, fn initFn) + Registry + Engine } var _ Engine = (*service)(nil) -var _ Manager = (*service)(nil) +var _ Registry = (*service)(nil) // service manages the registration and lifecycle of modules. type service struct { + services.NamedService + log log.Logger targets []string dependencyMap map[string][]string - moduleManager *modules.Manager + moduleManager *tracing.ModuleManagerWrapper serviceManager *services.Manager serviceMap map[string]services.Service } func New( - targets []string, -) *service { - logger := log.New("modules") - - return &service{ - 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{ + s := &service{ log: logger, targets: targets, dependencyMap: dependencyMap, - - moduleManager: manager, + moduleManager: tracing.WrapModuleManager(modules.NewManager(logger)), serviceMap: map[string]services.Service{}, } + s.NamedService = services.NewBasicService(s.starting, s.running, s.stopping).WithName("modules.service") + return s } -// Run starts all registered modules. -func (m *service) Run(ctx context.Context) error { - var err error +func (m *service) WithDependencies(dependencyMap map[string][]string) *service { + m.dependencyMap = dependencyMap + return m +} +func (m *service) starting(ctx context.Context) error { + var err error + m.moduleManager.SetContext(ctx) + _, span := infratracing.Start(ctx, "modules.service.starting") + defer span.End() for mod, targets := range m.dependencyMap { if !m.moduleManager.IsModuleRegistered(mod) { continue @@ -110,16 +108,38 @@ func (m *service) Run(ctx context.Context) error { listener := newServiceListener(m.log, m) m.serviceManager.AddListener(listener) - - m.log.Debug("Starting module service manager", "targets", strings.Join(m.targets, ",")) - // wait until a service fails or stop signal was received - err = m.serviceManager.StartAsync(ctx) - if err != nil { + if err := m.serviceManager.StartAsync(ctx); err != nil { return err } + return m.serviceManager.AwaitHealthy(ctx) +} + +func (m *service) running(ctx context.Context) error { + _, span := infratracing.Start(ctx, "modules.service.running") + defer span.End() + + // If no service manager was created (no modules registered), just wait for context + if m.serviceManager == nil { + <-ctx.Done() + return nil + } stopCtx := context.Background() - if err = m.serviceManager.AwaitStopped(stopCtx); err != nil { + return m.serviceManager.AwaitStopped(stopCtx) +} + +func (m *service) stopping(failureReason error) error { + spanCtx, span := infratracing.Start(context.Background(), "modules.service.stopping") + defer span.End() + m.log.Debug("Stopping module service manager", "reason", failureReason) + + // If no service manager was created (no modules registered), nothing to stop + if m.serviceManager == nil { + return nil + } + + m.serviceManager.StopAsync() + if err := m.serviceManager.AwaitStopped(spanCtx); err != nil { m.log.Error("Failed to stop module service manager", "error", err) return err } @@ -136,31 +156,36 @@ func (m *service) Run(ctx context.Context) error { return nil } -// Shutdown stops all modules and waits for them to stop. -func (m *service) Shutdown(ctx context.Context, reason string) error { - if m.serviceManager == nil { - m.log.Debug("No modules registered, nothing to stop...") - return nil +// Run starts all registered modules. +func (m *service) Run(ctx context.Context) error { + spanCtx, span := infratracing.Start(ctx, "modules.service.Run") + defer span.End() + if err := m.StartAsync(spanCtx); err != nil { + return err } - m.serviceManager.StopAsync() - m.log.Info("Awaiting services to be stopped...", "reason", reason) - return m.serviceManager.AwaitStopped(ctx) + stopCtx := context.Background() + return m.AwaitTerminated(stopCtx) } -type initFn func() (services.Service, error) +// Shutdown stops all modules and waits for them to stop. +func (m *service) Shutdown(ctx context.Context, reason string) error { + spanCtx, span := infratracing.Start(ctx, "modules.service.Shutdown") + defer span.End() + m.StopAsync() + return m.AwaitTerminated(spanCtx) +} // RegisterModule registers a module with the dskit module manager. -func (m *service) RegisterModule(name string, fn initFn) { +func (m *service) RegisterModule(name string, fn func() (services.Service, error)) { m.moduleManager.RegisterModule(name, fn) } // RegisterInvisibleModule registers an invisible module with the dskit module manager. // Invisible modules are not visible to the user, and are intended to be used as dependencies. -func (m *service) RegisterInvisibleModule(name string, fn initFn) { - m.moduleManager.RegisterModule(name, fn, modules.UserInvisibleModule) +func (m *service) RegisterInvisibleModule(name string, fn func() (services.Service, error)) { + m.moduleManager.RegisterInvisibleModule(name, fn) } -// IsModuleEnabled returns true if the module is enabled. func (m *service) IsModuleEnabled(name string) bool { return stringsContain(m.targets, name) } diff --git a/pkg/modules/tracing/listener.go b/pkg/modules/tracing/listener.go new file mode 100644 index 00000000000..428543fe63a --- /dev/null +++ b/pkg/modules/tracing/listener.go @@ -0,0 +1,117 @@ +package tracing + +import ( + "context" + "fmt" + + "github.com/grafana/dskit/services" + "github.com/grafana/grafana/pkg/infra/tracing" + "github.com/grafana/grafana/pkg/semconv" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" +) + +var _ services.Listener = (*Listener)(nil) + +// Listener implements dskit's services.Listener interface to add comprehensive tracing +// for service state transitions. It creates individual spans for each service state +// (Starting, Running, Stopping) to track their durations, providing detailed timing +// information about service lifecycle performance. +type Listener struct { + serviceName string + + ctx context.Context + parentSpan trace.Span + stateSpan trace.Span +} + +// NewListener creates a new tracing listener for the given service. +func NewListener(ctx context.Context, serviceName string) *Listener { + l := &Listener{ + ctx: ctx, + serviceName: serviceName, + } + return l +} + +// Starting is called when the service transitions from NEW to STARTING. +func (l *Listener) Starting() { + // Create the parent span when the service starts + spanCtx, span := tracing.Start(l.ctx, l.serviceName) + l.ctx = spanCtx + l.parentSpan = span + + l.startSpan(services.Starting, nil) +} + +// Running is called when the service transitions from STARTING to RUNNING. +func (l *Listener) Running() { + l.endSpan(nil) + l.startSpan(services.Running, nil) +} + +// Stopping is called when the service transitions to the STOPPING state. +func (l *Listener) Stopping(from services.State) { + l.endSpan(nil) + l.startSpan(services.Stopping, &from) +} + +// Terminated is called when the service transitions to the TERMINATED state. +func (l *Listener) Terminated(from services.State) { + l.endSpan(nil) + l.endParentSpan(from, nil) +} + +// Failed is called when the service transitions to the FAILED state. +func (l *Listener) Failed(from services.State, failure error) { + l.endSpan(failure) + l.endParentSpan(from, failure) +} + +// startSpan creates and stores a span for the given state +func (l *Listener) startSpan(toState services.State, fromState *services.State) { + spanName := fmt.Sprintf("%s Service", toState.String()) + _, span := tracing.Start(l.ctx, spanName, semconv.GrafanaServiceName(l.serviceName)) + attributes := []attribute.KeyValue{ + semconv.GrafanaServiceName(l.serviceName), + } + if fromState != nil { + attributes = append(attributes, attribute.String("modules.tracing.from_state", fromState.String())) + } + span.SetAttributes(attributes...) + l.stateSpan = span +} + +// endSpan safely ends and removes a span for the given state +// If err is provided, it will be recorded on the span before ending +func (l *Listener) endSpan(err error) { + if l.stateSpan == nil || !l.stateSpan.IsRecording() { + return + } + if err != nil { + l.stateSpan.SetStatus(codes.Error, err.Error()) + l.stateSpan.RecordError(err) + } else { + l.stateSpan.SetStatus(codes.Ok, "") + } + l.stateSpan.End() + l.stateSpan = nil +} + +// endParentSpan safely ends and removes the parent span +// If err is provided, it will be recorded on the parent span before ending +func (l *Listener) endParentSpan(from services.State, err error) { + if l.parentSpan == nil || !l.parentSpan.IsRecording() { + return + } + l.parentSpan.SetAttributes(attribute.String("modules.tracing.final_state", from.String())) + if err != nil { + l.parentSpan.SetStatus(codes.Error, err.Error()) + l.parentSpan.RecordError(err) + } else { + l.parentSpan.SetStatus(codes.Ok, "") + } + l.parentSpan.End() + l.parentSpan = nil +} diff --git a/pkg/modules/tracing/listener_test.go b/pkg/modules/tracing/listener_test.go new file mode 100644 index 00000000000..571dd719ff8 --- /dev/null +++ b/pkg/modules/tracing/listener_test.go @@ -0,0 +1,617 @@ +package tracing_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/grafana/dskit/services" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + + tracingmodule "github.com/grafana/grafana/pkg/modules/tracing" +) + +// setupTestTracer creates a test tracer with in-memory span recording +func setupTestTracer(t *testing.T) (*tracetest.InMemoryExporter, *trace.TracerProvider, func()) { + t.Helper() + exporter := tracetest.NewInMemoryExporter() + tp := trace.NewTracerProvider( + trace.WithSyncer(exporter), + trace.WithSampler(trace.AlwaysSample()), + ) + + // Set the global tracer provider for the tracing package to use + otel.SetTracerProvider(tp) + + cleanup := func() { + err := tp.Shutdown(context.Background()) + require.NoError(t, err) + } + + return exporter, tp, cleanup +} + +// createTracingContext creates a context with a root span to enable tracing +func createTracingContext(t *testing.T, tp *trace.TracerProvider) (context.Context, func()) { + t.Helper() + ctx := context.Background() + tracer := tp.Tracer("test-tracer") + ctx, rootSpan := tracer.Start(ctx, "test-root") + + cleanup := func() { + rootSpan.End() + } + + return ctx, cleanup +} + +func TestListener_Starting(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + + serviceName := "test-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + // Call Starting to create a span + listener.Starting() + + // End the starting span by transitioning to another state + listener.Running() + + // Give a moment for span to be recorded + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 1) + + // First span should be the Starting Service span + startingSpan := spans[0] + require.Equal(t, "Starting Service", startingSpan.Name) + require.True(t, startingSpan.SpanContext.IsValid()) + + // Check that the span has the expected attributes + found := false + for _, attr := range startingSpan.Attributes { + if attr.Key == "grafana.service.name" && attr.Value.AsString() == serviceName { + found = true + break + } + } + require.True(t, found, "Expected grafana.service.name attribute not found") +} + +func TestListener_Running(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + serviceName := "test-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + // Start with Starting state + listener.Starting() + + // Transition to Running + listener.Running() + + // End the running span by stopping + listener.Stopping(services.Running) + + // Give a moment for spans to be recorded + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 2) + + // First span should be the completed Starting span + startingSpan := spans[0] + require.Equal(t, "Starting Service", startingSpan.Name) + require.True(t, startingSpan.EndTime.After(startingSpan.StartTime)) + + // Second span should be the Running span (still active) + runningSpan := spans[1] + require.Equal(t, "Running Service", runningSpan.Name) +} + +func TestListener_Stopping(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + serviceName := "test-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + // Start with Running state + listener.Starting() + listener.Running() + + // Transition to Stopping + listener.Stopping(services.Running) + + // End the stopping span by terminating + listener.Terminated(services.Stopping) + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 4) // Starting, Running, Stopping, Parent + + // Check that Stopping span was started (should be the 3rd span, index 2) + stoppingSpan := spans[2] + require.Equal(t, "Stopping Service", stoppingSpan.Name) +} + +func TestListener_Terminated(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + serviceName := "test-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + // Go through normal lifecycle + listener.Starting() + listener.Running() + listener.Stopping(services.Running) + + // Terminate + listener.Terminated(services.Stopping) + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 4) // Starting, Running, Stopping, Parent - all should be ended + + // All spans should be completed + for _, span := range spans { + require.True(t, span.EndTime.After(span.StartTime), "Span %s should be ended", span.Name) + } +} + +func TestListener_Failed(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + serviceName := "test-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + // Start and then fail + listener.Starting() + + testError := errors.New("service failed") + listener.Failed(services.Starting, testError) + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 2) // Starting, Parent spans + + // The Starting span should have the error recorded + startingSpan := spans[0] + require.Equal(t, "Starting Service", startingSpan.Name) + require.True(t, startingSpan.EndTime.After(startingSpan.StartTime)) + + // Check that the error was recorded + found := false + for _, event := range startingSpan.Events { + if event.Name == "exception" { + found = true + break + } + } + require.True(t, found, "Expected exception event not found in span") +} + +func TestListener_ServiceLifecycleIntegration(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + serviceName := "integration-test-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + // Simulate complete service lifecycle + listener.Starting() + time.Sleep(5 * time.Millisecond) // Simulate startup time + + listener.Running() + time.Sleep(10 * time.Millisecond) // Simulate running time + + listener.Stopping(services.Running) + time.Sleep(5 * time.Millisecond) // Simulate shutdown time + + listener.Terminated(services.Stopping) + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 4) // Starting, Running, Stopping, Parent + + // Verify span names and order (excluding parent span which is last) + expectedNames := []string{ + "Starting Service", + "Running Service", + "Stopping Service", + } + + // Check the first 3 spans (service state spans) + for i := 0; i < 3; i++ { + span := spans[i] + require.Equal(t, expectedNames[i], span.Name) + require.True(t, span.EndTime.After(span.StartTime), "Span %s should be ended", span.Name) + + // Verify service name attribute + found := false + for _, attr := range span.Attributes { + if attr.Key == "grafana.service.name" && attr.Value.AsString() == serviceName { + found = true + break + } + } + require.True(t, found, "Expected grafana.service.name attribute not found in span %s", span.Name) + } + + // Check the parent span (last span) + parentSpan := spans[3] + require.Equal(t, serviceName, parentSpan.Name) + require.True(t, parentSpan.EndTime.After(parentSpan.StartTime), "Parent span should be ended") + + // Verify timing relationships between state spans + require.True(t, spans[0].EndTime.Before(spans[1].StartTime) || spans[0].EndTime.Equal(spans[1].StartTime), + "Starting span should end before or when Running span starts") + require.True(t, spans[1].EndTime.Before(spans[2].StartTime) || spans[1].EndTime.Equal(spans[2].StartTime), + "Running span should end before or when Stopping span starts") +} + +func TestListener_ErrorRecording(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + serviceName := "error-test-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + // Start service and then fail with error + listener.Starting() + listener.Running() + + testError := errors.New("critical service failure") + listener.Failed(services.Running, testError) + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 3) // Starting, Running, Parent spans + + // The Running span should have the error recorded + runningSpan := spans[1] + require.Equal(t, "Running Service", runningSpan.Name) + + // Check for exception event + hasException := false + for _, event := range runningSpan.Events { + if event.Name == "exception" { + hasException = true + // Check for error message in attributes + for _, attr := range event.Attributes { + if attr.Key == "exception.message" { + require.Equal(t, testError.Error(), attr.Value.AsString()) + } + } + } + } + require.True(t, hasException, "Expected exception event in failed span") +} + +func TestListener_SpanAttributes(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + + serviceName := "attribute-test-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + // Test that all state transitions include proper attributes + listener.Starting() + listener.Running() + listener.Stopping(services.Running) + listener.Terminated(services.Stopping) + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 4) // Starting, Running, Stopping, Parent + + // Check Starting span attributes + startingSpan := spans[0] + require.Equal(t, "Starting Service", startingSpan.Name) + hasServiceName := false + for _, attr := range startingSpan.Attributes { + if attr.Key == "grafana.service.name" && attr.Value.AsString() == serviceName { + hasServiceName = true + } + } + require.True(t, hasServiceName, "Starting span should have service name attribute") + + // Check Stopping span has from_state attribute + stoppingSpan := spans[2] + require.Equal(t, "Stopping Service", stoppingSpan.Name) + hasFromState2 := false + hasServiceName2 := false + for _, attr := range stoppingSpan.Attributes { + if attr.Key == "modules.tracing.from_state" && attr.Value.AsString() == "Running" { + hasFromState2 = true + } + if attr.Key == "grafana.service.name" && attr.Value.AsString() == serviceName { + hasServiceName2 = true + } + } + require.True(t, hasFromState2, "Stopping span should have from_state attribute") + require.True(t, hasServiceName2, "Stopping span should have service name attribute") + + // Check parent span has final_state attribute + parentSpan := spans[3] + require.Equal(t, serviceName, parentSpan.Name) + hasFinalState := false + for _, attr := range parentSpan.Attributes { + if attr.Key == "modules.tracing.final_state" && attr.Value.AsString() == "Stopping" { + hasFinalState = true + } + } + require.True(t, hasFinalState, "Parent span should have final_state attribute") +} + +func TestListener_SpanStatusCodes(t *testing.T) { + t.Parallel() + + t.Run("successful lifecycle has OK status", func(t *testing.T) { + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + + serviceName := "status-test-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + listener.Starting() + listener.Running() + listener.Terminated(services.Running) + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 3) // Starting, Running, Parent + + // All spans should have OK status + for _, span := range spans { + require.Equal(t, "Ok", span.Status.Code.String(), "Span %s should have OK status", span.Name) + } + }) + + t.Run("failed service has Error status", func(t *testing.T) { + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + + serviceName := "error-status-test-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + listener.Starting() + testError := errors.New("service startup failed") + listener.Failed(services.Starting, testError) + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 2) // Starting, Parent + + // Both spans should have Error status + for _, span := range spans { + require.Equal(t, "Error", span.Status.Code.String(), "Span %s should have Error status", span.Name) + require.Equal(t, testError.Error(), span.Status.Description, "Span %s should have error description", span.Name) + } + }) +} + +func TestListener_ContextPropagation(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + // Create a parent context with a span + ctx := context.Background() + tracer := tp.Tracer("test-tracer") + parentCtx, parentSpan := tracer.Start(ctx, "parent-operation") + defer parentSpan.End() + + serviceName := "context-test-service" + listener := tracingmodule.NewListener(parentCtx, serviceName) + + listener.Starting() + listener.Running() + listener.Terminated(services.Running) + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.GreaterOrEqual(t, len(spans), 3, "Should have at least service spans") + + // Find the service parent span + var serviceParentSpan *tracetest.SpanStub + for i := range spans { + if spans[i].Name == serviceName { + serviceParentSpan = &spans[i] + break + } + } + require.NotNil(t, serviceParentSpan, "Should find service parent span") + + // The service parent span should be a child of our test parent span + require.Equal(t, parentSpan.SpanContext().TraceID(), serviceParentSpan.SpanContext.TraceID(), + "Service spans should be in the same trace as parent context") +} + +func TestListener_EmptyServiceName(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + + // Test with empty service name + listener := tracingmodule.NewListener(ctx, "") + + listener.Starting() + listener.Terminated(services.Starting) + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 2) // Starting, Parent + + // Parent span should have empty name + parentSpan := spans[1] + require.Equal(t, "", parentSpan.Name) +} + +func TestListener_LongRunningService(t *testing.T) { + t.Parallel() + + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + + serviceName := "long-running-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + startTime := time.Now() + + listener.Starting() + time.Sleep(20 * time.Millisecond) // Simulate startup time + + listener.Running() + time.Sleep(50 * time.Millisecond) // Simulate running time + + listener.Stopping(services.Running) + time.Sleep(10 * time.Millisecond) // Simulate shutdown time + + listener.Terminated(services.Stopping) + + endTime := time.Now() + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + require.Len(t, spans, 4) // Starting, Running, Stopping, Parent + + // Verify timing relationships + startingSpan := spans[0] + runningSpan := spans[1] + stoppingSpan := spans[2] + parentSpan := spans[3] + + // Each span should have reasonable duration + require.True(t, startingSpan.EndTime.After(startingSpan.StartTime)) + require.True(t, runningSpan.EndTime.After(runningSpan.StartTime)) + require.True(t, stoppingSpan.EndTime.After(stoppingSpan.StartTime)) + require.True(t, parentSpan.EndTime.After(parentSpan.StartTime)) + + // Parent span should encompass the entire lifecycle + require.True(t, parentSpan.StartTime.Before(startTime.Add(10*time.Millisecond)) || + parentSpan.StartTime.Equal(startTime.Add(10*time.Millisecond)), + "Parent span should start around the beginning") + require.True(t, parentSpan.EndTime.After(endTime.Add(-10*time.Millisecond)), + "Parent span should end around the end") +} + +func TestListener_EarlyTermination(t *testing.T) { + t.Parallel() + + t.Run("New to Terminated without Starting", func(t *testing.T) { + // This is a valid dskit transition: when StopAsync() is called on a service in New state, + // it goes directly to Terminated without ever calling Starting() + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + + serviceName := "early-terminated-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + // Call Terminated directly from New state (valid dskit behavior) + require.NotPanics(t, func() { + listener.Terminated(services.New) + }, "Terminated should not panic when called without Starting") + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + // Should have no spans since Starting() was never called to create the parent span + require.Len(t, spans, 0, "Should have no spans since Starting() was never called") + }) + + t.Run("New to Failed without Starting", func(t *testing.T) { + // This is a valid dskit transition: service can fail during initialization + // before Starting() is called + exporter, tp, cleanup := setupTestTracer(t) + defer cleanup() + + ctx, ctxCleanup := createTracingContext(t, tp) + defer ctxCleanup() + + serviceName := "early-failed-service" + listener := tracingmodule.NewListener(ctx, serviceName) + + testError := errors.New("initialization failure") + // Call Failed directly from New state (valid dskit behavior) + require.NotPanics(t, func() { + listener.Failed(services.New, testError) + }, "Failed should not panic when called without Starting") + + time.Sleep(10 * time.Millisecond) + + spans := exporter.GetSpans() + // Should have no spans since Starting() was never called to create the parent span + require.Len(t, spans, 0, "Should have no spans since Starting() was never called") + }) +} diff --git a/pkg/modules/tracing/manager.go b/pkg/modules/tracing/manager.go new file mode 100644 index 00000000000..02a1b17f18c --- /dev/null +++ b/pkg/modules/tracing/manager.go @@ -0,0 +1,70 @@ +package tracing + +import ( + "context" + + "github.com/grafana/dskit/modules" + "github.com/grafana/dskit/services" +) + +type initFn func() (services.Service, error) + +type ModuleManagerWrapper struct { + ready chan struct{} + ctx context.Context + *modules.Manager +} + +func WrapModuleManager(moduleManager *modules.Manager) *ModuleManagerWrapper { + return &ModuleManagerWrapper{ + ready: make(chan struct{}), + Manager: moduleManager, + } +} + +// RegisterModule registers a module without any options +func (m *ModuleManagerWrapper) RegisterModule(name string, fn initFn) { + var wrappedFn initFn + if fn != nil { + wrappedFn = m.wrapInitFn(fn) + } + m.Manager.RegisterModule(name, wrappedFn) +} + +// RegisterInvisibleModule registers a module with the UserInvisibleModule option +func (m *ModuleManagerWrapper) RegisterInvisibleModule(name string, fn initFn) { + var wrappedFn initFn + if fn != nil { + wrappedFn = m.wrapInitFn(fn) + } + m.Manager.RegisterModule(name, wrappedFn, modules.UserInvisibleModule) +} + +func (m *ModuleManagerWrapper) SetContext(ctx context.Context) { + if m.ctx != nil { + return + } + m.ctx = ctx + close(m.ready) +} + +func (m *ModuleManagerWrapper) getContext() context.Context { + <-m.ready + return m.ctx +} + +func (m *ModuleManagerWrapper) wrapInitFn(fn initFn) initFn { + return func() (services.Service, error) { + service, err := fn() + if err != nil { + return nil, err + } + if service == nil { + return nil, nil + } + if namedService, ok := service.(services.NamedService); ok { + namedService.AddListener(NewListener(m.getContext(), namedService.ServiceName())) + } + return service, nil + } +} diff --git a/pkg/modules/tracing/manager_test.go b/pkg/modules/tracing/manager_test.go new file mode 100644 index 00000000000..41ccdb5c677 --- /dev/null +++ b/pkg/modules/tracing/manager_test.go @@ -0,0 +1,342 @@ +package tracing + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/grafana/dskit/modules" + "github.com/grafana/dskit/services" + "github.com/stretchr/testify/require" +) + +func TestModuleManagerWrapper_RegisterModule(t *testing.T) { + t.Run("registers module with wrapped init function", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + // Set context so getContext doesn't block + ctx := context.Background() + wrapper.SetContext(ctx) + + called := false + mockService := &mockNamedService{name: "test-service"} + initFn := func() (services.Service, error) { + called = true + return mockService, nil + } + + wrapper.RegisterModule("test-module", initFn) + + // Verify the module was registered + require.True(t, manager.IsModuleRegistered("test-module")) + + // Initialize the module to test the wrapped init function + service, err := manager.InitModuleServices("test-module") + require.NoError(t, err) + require.True(t, called) + require.NotNil(t, service) + + // Verify listener was added to the service + require.Len(t, mockService.listeners, 1) + }) + + t.Run("propagates init function errors", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + // Set context so getContext doesn't block + ctx := context.Background() + wrapper.SetContext(ctx) + + expectedErr := errors.New("init error") + initFn := func() (services.Service, error) { + return nil, expectedErr + } + + wrapper.RegisterModule("test-module", initFn) + + // Try to initialize the module + _, err := manager.InitModuleServices("test-module") + require.Error(t, err) + require.Contains(t, err.Error(), expectedErr.Error()) + }) + + t.Run("handles nil init function", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + // Set context so getContext doesn't block + ctx := context.Background() + wrapper.SetContext(ctx) + + // Register module with nil init function + wrapper.RegisterModule("nil-module", nil) + + // Verify the module was registered + require.True(t, manager.IsModuleRegistered("nil-module")) + + // Initialize the module - should work with nil function + service, err := manager.InitModuleServices("nil-module") + require.NoError(t, err) + require.Empty(t, service) // Should return empty map for nil function + }) +} + +func TestModuleManagerWrapper_RegisterInvisibleModule(t *testing.T) { + t.Run("registers invisible module with wrapped init function", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + // Set context so getContext doesn't block + ctx := context.Background() + wrapper.SetContext(ctx) + + called := false + mockService := &mockNamedService{name: "invisible-service"} + initFn := func() (services.Service, error) { + called = true + return mockService, nil + } + + wrapper.RegisterInvisibleModule("invisible-module", initFn) + + // Verify the module was registered + require.True(t, manager.IsModuleRegistered("invisible-module")) + + // Initialize the module to test the wrapped init function + service, err := manager.InitModuleServices("invisible-module") + require.NoError(t, err) + require.True(t, called) + require.NotNil(t, service) + + // Verify listener was added to the service + require.Len(t, mockService.listeners, 1) + }) + + t.Run("handles nil init function", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + // Set context so getContext doesn't block + ctx := context.Background() + wrapper.SetContext(ctx) + + // Register invisible module with nil init function + wrapper.RegisterInvisibleModule("nil-invisible-module", nil) + + // Verify the module was registered + require.True(t, manager.IsModuleRegistered("nil-invisible-module")) + + // Initialize the module - should work with nil function + service, err := manager.InitModuleServices("nil-invisible-module") + require.NoError(t, err) + require.Empty(t, service) // Should return empty map for nil function + }) +} + +func TestModuleManagerWrapper_SetContext(t *testing.T) { + t.Run("sets context and closes ready channel", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + ctx := context.Background() + + // Verify ready channel is not closed initially + select { + case <-wrapper.ready: + t.Fatal("ready channel should not be closed initially") + default: + // Expected + } + + wrapper.SetContext(ctx) + + // Verify context is set and ready channel is closed + require.Equal(t, ctx, wrapper.ctx) + select { + case <-wrapper.ready: + // Expected - channel should be closed + case <-time.After(100 * time.Millisecond): + t.Fatal("ready channel should be closed after SetContext") + } + }) + + t.Run("ignores subsequent calls", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + ctx1 := context.Background() + type contextKey string + ctx2 := context.WithValue(context.Background(), contextKey("key"), "value") + + wrapper.SetContext(ctx1) + wrapper.SetContext(ctx2) // Should be ignored + + require.Equal(t, ctx1, wrapper.ctx) + }) +} + +func TestModuleManagerWrapper_getContext(t *testing.T) { + t.Run("waits for ready channel and returns context", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + ctx := context.Background() + + // Start goroutine to get context + resultCh := make(chan context.Context) + go func() { + resultCh <- wrapper.getContext() + }() + + // Verify it's waiting + select { + case <-resultCh: + t.Fatal("getContext should wait for ready channel") + case <-time.After(50 * time.Millisecond): + // Expected - should be waiting + } + + // Set context + wrapper.SetContext(ctx) + + // Verify getContext returns the context + select { + case result := <-resultCh: + require.Equal(t, ctx, result) + case <-time.After(100 * time.Millisecond): + t.Fatal("getContext should return after SetContext") + } + }) +} + +func TestModuleManagerWrapper_wrapInitFn(t *testing.T) { + t.Run("adds listener to NamedService", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + ctx := context.Background() + wrapper.SetContext(ctx) + + mockService := &mockNamedService{name: "test-service"} + initFn := func() (services.Service, error) { + return mockService, nil + } + + wrappedFn := wrapper.wrapInitFn(initFn) + service, err := wrappedFn() + + require.NoError(t, err) + require.Equal(t, mockService, service) + + // Verify listener was added to the service + require.Len(t, mockService.listeners, 1) + }) + + t.Run("handles regular service without NamedService interface", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + ctx := context.Background() + wrapper.SetContext(ctx) + + mockService := &mockService{} + initFn := func() (services.Service, error) { + return mockService, nil + } + + wrappedFn := wrapper.wrapInitFn(initFn) + service, err := wrappedFn() + + require.NoError(t, err) + require.Equal(t, mockService, service) + // No listener should be added for non-NamedService + }) + + t.Run("propagates init function errors", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + ctx := context.Background() + wrapper.SetContext(ctx) + + expectedErr := errors.New("init error") + initFn := func() (services.Service, error) { + return nil, expectedErr + } + + wrappedFn := wrapper.wrapInitFn(initFn) + service, err := wrappedFn() + + require.Error(t, err) + require.Equal(t, expectedErr, err) + require.Nil(t, service) + }) + + t.Run("handles nil service return", func(t *testing.T) { + manager := modules.NewManager(nil) + wrapper := WrapModuleManager(manager) + + ctx := context.Background() + wrapper.SetContext(ctx) + + initFn := func() (services.Service, error) { + return nil, nil // Return nil service with no error + } + + wrappedFn := wrapper.wrapInitFn(initFn) + service, err := wrappedFn() + + require.NoError(t, err) + require.Nil(t, service) + }) +} + +// Mock implementations for testing + +type mockService struct { + listeners []services.Listener +} + +func (m *mockService) AddListener(listener services.Listener) func() { + m.listeners = append(m.listeners, listener) + return func() {} +} + +func (m *mockService) AwaitRunning(ctx context.Context) error { + return nil +} + +func (m *mockService) AwaitTerminated(ctx context.Context) error { + return nil +} + +func (m *mockService) FailureCase() error { + return nil +} + +func (m *mockService) ServiceName() string { + return "mock-service" +} + +func (m *mockService) StartAsync(ctx context.Context) error { + return nil +} + +func (m *mockService) State() services.State { + return services.New +} + +func (m *mockService) StopAsync() { +} + +type mockNamedService struct { + mockService + name string +} + +func (m *mockNamedService) ServiceName() string { + return m.name +} diff --git a/pkg/modules/tracing/service.go b/pkg/modules/tracing/service.go deleted file mode 100644 index 38fe155edc1..00000000000 --- a/pkg/modules/tracing/service.go +++ /dev/null @@ -1,35 +0,0 @@ -package tracing - -import ( - "context" - - "github.com/grafana/dskit/services" - "github.com/grafana/grafana/pkg/semconv" - "go.opentelemetry.io/otel/trace" -) - -var _ services.NamedService = &ServiceTracer{} - -// ServiceTracer wraps service.NamedService and adds tracing. -// Currently it is limited to the starting -> running state transition. -type ServiceTracer struct { - services.NamedService - tracer trace.Tracer -} - -// NewServiceTracer creates a new ServiceTracer. -func NewServiceTracer(tracerProvider trace.TracerProvider, service services.NamedService) *ServiceTracer { - tracer := tracerProvider.Tracer("pkg/modules/tracing") - return &ServiceTracer{NamedService: service, tracer: tracer} -} - -func (s *ServiceTracer) StartAsync(ctx context.Context) error { - spanCtx, span := s.tracer.Start(ctx, "Service Start", trace.WithAttributes(semconv.GrafanaServiceName(s.ServiceName()))) - go func() { - if err := s.AwaitRunning(spanCtx); err != nil { - span.RecordError(err) - } - span.End() - }() - return s.NamedService.StartAsync(ctx) -} diff --git a/pkg/registry/backgroundsvcs/adapter/dependencies.go b/pkg/registry/backgroundsvcs/adapter/dependencies.go index 939121e03fb..20e31754490 100644 --- a/pkg/registry/backgroundsvcs/adapter/dependencies.go +++ b/pkg/registry/backgroundsvcs/adapter/dependencies.go @@ -1,24 +1,35 @@ package adapter -import "github.com/grafana/grafana/pkg/modules" +import ( + "github.com/grafana/grafana/pkg/infra/tracing" + "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. + // Tracing is the module name for the tracing service. + Tracing = tracing.ServiceName + + // GrafanaAPIServer is the module name for the embedded Grafana API server service. + GrafanaAPIServer = modules.GrafanaAPIServer + + // BackgroundServices is the module name for the background services module. + // This module is an alias for any background service that is not explicitly listed in the dependency map. BackgroundServices = "background-services" - // Core is an alias for a set of services that must be running before most other services can start. + // Core is the module name for the core module. + // This module is an alias for a set of service dependencies 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. +// It defines the startup order and dependencies between different modules. // Background services are automatically added as dependencies to the BackgroundServices module -// unless they are explicitly listed in this map with custom dependencies. +// unless they are explicitly listed in this map. func dependencyMap() map[string][]string { return map[string][]string{ - modules.GrafanaAPIServer: {}, - Core: {modules.GrafanaAPIServer}, - BackgroundServices: {Core}, + Tracing: {}, + GrafanaAPIServer: {Tracing}, + Core: {GrafanaAPIServer}, + BackgroundServices: {Core}, } } diff --git a/pkg/registry/backgroundsvcs/adapter/manager.go b/pkg/registry/backgroundsvcs/adapter/manager.go index 47cf2bfdc63..d43035f3d62 100644 --- a/pkg/registry/backgroundsvcs/adapter/manager.go +++ b/pkg/registry/backgroundsvcs/adapter/manager.go @@ -2,10 +2,8 @@ package adapter import ( "context" - "reflect" - "sync" + "time" - "github.com/grafana/dskit/modules" "github.com/grafana/dskit/services" "github.com/grafana/grafana/pkg/infra/log" @@ -14,10 +12,16 @@ import ( "github.com/grafana/grafana/pkg/registry" ) +var ( + stopTimeout = 30 * time.Second +) + type managerAdapter struct { - reg registry.BackgroundServiceRegistry - manager grafanamodules.Engine - mu sync.RWMutex // protects manager field from concurrent access + services.NamedService + + reg registry.BackgroundServiceRegistry + manager grafanamodules.Manager + dependencyMap map[string][]string } // NewManagerAdapter creates a new manager adapter that bridges Grafana's background @@ -26,77 +30,83 @@ type managerAdapter struct { // - Coordinated service initialization // - Observable service states and health monitoring // - Graceful shutdown with proper cleanup ordering +// +// Services implementing CanBeDisabled that are disabled will be skipped. func NewManagerAdapter(reg registry.BackgroundServiceRegistry) *managerAdapter { - return &managerAdapter{ - reg: reg, + m := &managerAdapter{ + reg: reg, + dependencyMap: dependencyMap(), } + m.NamedService = services.NewBasicService(m.starting, m.running, m.stopping).WithName("backgroundsvcs.managerAdapter") + return m +} + +func (m *managerAdapter) starting(ctx context.Context) error { + spanCtx, span := tracing.Start(ctx, "backgroundsvcs.managerAdapter.starting") + defer span.End() + logger := log.New("backgroundsvcs.managerAdapter").FromContext(spanCtx) + manager := grafanamodules.New(logger, []string{BackgroundServices}).WithDependencies(m.dependencyMap) + + for _, bgSvc := range m.reg.GetServices() { + //only wrap background services that are not already a NamedService + namedService, ok := bgSvc.(services.NamedService) + if !ok { + namedService = asNamedService(bgSvc) + } + + // skip disabled services + if s, ok := bgSvc.(registry.CanBeDisabled); ok && s.IsDisabled() { + logger.Debug("Skipping disabled service", "service", namedService.ServiceName()) + continue + } + + // register the service as an invisible module + manager.RegisterInvisibleModule(namedService.ServiceName(), func() (services.Service, error) { + return namedService, nil + }) + + // add the service as a background service dependency if it's not already in the dependency map + if _, ok := m.dependencyMap[namedService.ServiceName()]; !ok { + m.dependencyMap[namedService.ServiceName()] = []string{Core} + m.dependencyMap[BackgroundServices] = append(m.dependencyMap[BackgroundServices], namedService.ServiceName()) + } + } + + manager.RegisterModule(Core, nil) + manager.RegisterModule(BackgroundServices, nil) + + m.manager = manager + return nil +} + +func (m *managerAdapter) running(ctx context.Context) error { + spanCtx, span := tracing.Start(ctx, "backgroundsvcs.managerAdapter.running") + defer span.End() + return m.manager.Run(spanCtx) +} + +func (m *managerAdapter) stopping(failure error) error { + ctx, cancel := context.WithTimeout(context.Background(), stopTimeout) + defer cancel() + spanCtx, span := tracing.Start(ctx, "backgroundsvcs.managerAdapter.stopping") + defer span.End() + reason := "" + if failure != nil { + reason = failure.Error() + } + return m.manager.Shutdown(spanCtx, reason) } // 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()) - } +func (m *managerAdapter) Run(ctx context.Context) error { + if err := m.StartAsync(ctx); err != nil { + return err } - - // 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) + return m.AwaitTerminated(ctx) } -// 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) +// Shutdown calls calls the underlying manager's Shutdown +func (m *managerAdapter) Shutdown(ctx context.Context, reason string) error { + m.StopAsync() + return m.AwaitTerminated(ctx) } diff --git a/pkg/registry/backgroundsvcs/adapter/manager_test.go b/pkg/registry/backgroundsvcs/adapter/manager_test.go index 7ea140344ee..d511b546cbb 100644 --- a/pkg/registry/backgroundsvcs/adapter/manager_test.go +++ b/pkg/registry/backgroundsvcs/adapter/manager_test.go @@ -2,6 +2,8 @@ package adapter import ( "context" + "errors" + "reflect" "testing" "time" @@ -11,6 +13,8 @@ import ( "github.com/grafana/grafana/pkg/registry" ) +const testTimeout = 200 * time.Millisecond + func TestNewManagerAdapter(t *testing.T) { reg := &mockBackgroundServiceRegistry{} adapter := NewManagerAdapter(reg) @@ -18,170 +22,344 @@ func TestNewManagerAdapter(t *testing.T) { require.NotNil(t, adapter) require.Equal(t, reg, adapter.reg) require.Nil(t, adapter.manager) + require.NotNil(t, adapter.dependencyMap) } -func TestManagerAdapter_Run(t *testing.T) { +func TestManagerAdapter_Starting(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 + adapter.dependencyMap = map[string][]string{ + BackgroundServices: {}, } - reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{mockSvc}} - adapter := NewManagerAdapter(reg) - - ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) defer cancel() - err := adapter.Run(ctx) + err := adapter.StartAsync(ctx) + require.NoError(t, err) + + err = adapter.AwaitRunning(ctx) + require.NoError(t, err) + }) + + t.Run("registers enabled services and skips disabled", func(t *testing.T) { + enabledSvc := &mockService{} + + // Create a different type for the disabled service to distinguish them + type disabledMockService struct{ mockService } + disabledSvc := &disabledMockService{mockService{disabled: true}} + + namedSvc := &mockNamedService{name: "test-service"} + + reg := &mockBackgroundServiceRegistry{ + services: []registry.BackgroundService{enabledSvc, disabledSvc, namedSvc}, + } + adapter := NewManagerAdapter(reg) + adapter.dependencyMap = map[string][]string{ + BackgroundServices: {}, + } + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + err := adapter.StartAsync(ctx) + require.NoError(t, err) + + err = adapter.AwaitRunning(ctx) + require.NoError(t, err) + + // Verify dependency map was updated correctly + // Should have entries for enabled services but not disabled ones + require.Contains(t, adapter.dependencyMap, "*adapter.mockNamedService") // Named service + require.Contains(t, adapter.dependencyMap, reflect.TypeOf(enabledSvc).String()) // Wrapped service + require.NotContains(t, adapter.dependencyMap, reflect.TypeOf(disabledSvc).String()) // Disabled service should not be in map + + // Check that BackgroundServices depends on the enabled services + bgDeps := adapter.dependencyMap[BackgroundServices] + require.Contains(t, bgDeps, "*adapter.mockNamedService") + require.Contains(t, bgDeps, reflect.TypeOf(enabledSvc).String()) + require.NotContains(t, bgDeps, reflect.TypeOf(disabledSvc).String()) + }) + + t.Run("handles services that are already NamedService", func(t *testing.T) { + namedSvc := &mockNamedService{name: "already-named"} + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{namedSvc}} + adapter := NewManagerAdapter(reg) + adapter.dependencyMap = map[string][]string{ + BackgroundServices: {}, + } + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + err := adapter.StartAsync(ctx) + require.NoError(t, err) + + err = adapter.AwaitRunning(ctx) + require.NoError(t, err) + + // Verify named service was added to dependency map + require.Contains(t, adapter.dependencyMap, "*adapter.mockNamedService") + }) + + t.Run("service already in dependency map is not added again", func(t *testing.T) { + namedSvc := &mockNamedService{name: "existing-service"} + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{namedSvc}} + adapter := NewManagerAdapter(reg) + + // Pre-populate the dependency map with the service using the actual service name that will be used + serviceName := "*adapter.mockNamedService" + adapter.dependencyMap[serviceName] = []string{"custom-dependency"} + initialBgDeps := append([]string{}, adapter.dependencyMap[BackgroundServices]...) + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + err := adapter.starting(ctx) require.NoError(t, err) require.NotNil(t, adapter.manager) - require.True(t, mockSvc.runCalled) + + // Verify the existing dependency was not overwritten + require.Equal(t, []string{"custom-dependency"}, adapter.dependencyMap[serviceName]) + + // Verify BackgroundServices dependencies were not modified (should not contain the service twice) + finalBgDeps := adapter.dependencyMap[BackgroundServices] + require.Equal(t, initialBgDeps, finalBgDeps) + }) + + t.Run("service without NamedService interface gets wrapped", func(t *testing.T) { + // Create a service that doesn't implement NamedService + plainSvc := &mockService{} + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{plainSvc}} + adapter := NewManagerAdapter(reg) + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + err := adapter.starting(ctx) + require.NoError(t, err) + require.NotNil(t, adapter.manager) + + // Verify the service was wrapped and added to dependency map using its type name + expectedServiceName := reflect.TypeOf(plainSvc).String() + require.Contains(t, adapter.dependencyMap, expectedServiceName) + + // Verify it was added to BackgroundServices dependencies + bgDeps := adapter.dependencyMap[BackgroundServices] + require.Contains(t, bgDeps, expectedServiceName) + }) + + t.Run("service without CanBeDisabled interface is always enabled", func(t *testing.T) { + // Create a service that doesn't implement CanBeDisabled + simpleSvc := &simpleBackgroundService{} + + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{simpleSvc}} + adapter := NewManagerAdapter(reg) + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + err := adapter.starting(ctx) + require.NoError(t, err) + require.NotNil(t, adapter.manager) + + // Verify the service was added (since it doesn't implement CanBeDisabled, it's always enabled) + expectedServiceName := reflect.TypeOf(simpleSvc).String() + require.Contains(t, adapter.dependencyMap, expectedServiceName) + }) + + t.Run("real manager integration test", func(t *testing.T) { + testSvc := &mockService{} + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{testSvc}} + adapter := NewManagerAdapter(reg) + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + // Use the real manager - this tests actual integration + err := adapter.starting(ctx) + require.NoError(t, err) + require.NotNil(t, adapter.manager) + + // Verify the service was registered in dependency map + expectedServiceName := reflect.TypeOf(testSvc).String() + require.Contains(t, adapter.dependencyMap, expectedServiceName) }) } -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}, - } +func TestManagerAdapter_Running(t *testing.T) { + t.Run("runs with real manager", func(t *testing.T) { + mock := &mockNamedService{name: "mock"} + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{ + mock, + }} adapter := NewManagerAdapter(reg) + adapter.dependencyMap = map[string][]string{ + BackgroundServices: {}, + } - ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) defer cancel() - err := adapter.Run(ctx) + err := adapter.StartAsync(ctx) require.NoError(t, err) - require.NotNil(t, adapter.manager) - // Verify the service was called - require.True(t, mockSvc.runCalled) + err = adapter.AwaitRunning(ctx) + require.NoError(t, err) }) - 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}, - } + t.Run("running delegates to manager", func(t *testing.T) { + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{}} adapter := NewManagerAdapter(reg) - ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) defer cancel() - err := adapter.Run(ctx) + // Initialize with real manager + err := adapter.starting(ctx) + require.NoError(t, err) + + // Test running method directly - this will likely fail due to missing production modules + // but it covers the running method code path + err = adapter.running(ctx) + if err != nil { + require.Contains(t, err.Error(), "no such module") + } + }) +} + +func TestManagerAdapter_Stopping(t *testing.T) { + t.Run("stopping method delegates to manager", func(t *testing.T) { + mock := &mockNamedService{name: "test-service"} + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{mock}} + adapter := NewManagerAdapter(reg) + adapter.dependencyMap = map[string][]string{ + BackgroundServices: {}, + } + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + // Initialize the manager first - need to go through starting to initialize manager + err := adapter.starting(ctx) require.NoError(t, err) require.NotNil(t, adapter.manager) - // Verify the service was called - require.True(t, mockSvc.runCalled) + // Test the stopping method directly - this covers the stopping function + err = adapter.stopping(nil) + require.NoError(t, err) }) - 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 + t.Run("stopping with failure reason", func(t *testing.T) { + mock := &mockNamedService{name: "test-service"} + reg := &mockBackgroundServiceRegistry{services: []registry.BackgroundService{mock}} + adapter := NewManagerAdapter(reg) + adapter.dependencyMap = map[string][]string{ + BackgroundServices: {}, } - reg := &mockBackgroundServiceRegistry{ - services: []registry.BackgroundService{disabledSvc, enabledSvc}, - } + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + // Initialize the manager first - need to go through starting to initialize manager + err := adapter.starting(ctx) + require.NoError(t, err) + require.NotNil(t, adapter.manager) + + // Test stopping with failure reason + failure := errors.New("test failure") + err = adapter.stopping(failure) + require.NoError(t, err) + }) +} + +func TestManagerAdapter_Run(t *testing.T) { + t.Run("successful run lifecycle", func(t *testing.T) { + reg := &mockBackgroundServiceRegistry{} adapter := NewManagerAdapter(reg) - ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout) + // Create a mock basic service that we can control + mockBasicService := &mockBasicService{} + adapter.NamedService = mockBasicService + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) defer cancel() err := adapter.Run(ctx) require.NoError(t, err) - require.NotNil(t, adapter.manager) + require.True(t, mockBasicService.startAsyncCalled) + require.True(t, mockBasicService.awaitTerminatedCalled) + }) - // Verify only enabled service was called - require.False(t, disabledSvc.runCalled) - require.True(t, enabledSvc.runCalled) + t.Run("returns StartAsync error", func(t *testing.T) { + reg := &mockBackgroundServiceRegistry{} + adapter := NewManagerAdapter(reg) + + expectedErr := errors.New("start error") + mockBasicService := &mockBasicService{startAsyncError: expectedErr} + adapter.NamedService = mockBasicService + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + err := adapter.Run(ctx) + require.Error(t, err) + require.Equal(t, expectedErr, err) + require.True(t, mockBasicService.startAsyncCalled) + require.False(t, mockBasicService.awaitTerminatedCalled) + }) + + t.Run("returns AwaitTerminated error", func(t *testing.T) { + reg := &mockBackgroundServiceRegistry{} + adapter := NewManagerAdapter(reg) + + expectedErr := errors.New("await error") + mockBasicService := &mockBasicService{awaitTerminatedError: expectedErr} + adapter.NamedService = mockBasicService + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + err := adapter.Run(ctx) + require.Error(t, err) + require.Equal(t, expectedErr, err) + require.True(t, mockBasicService.startAsyncCalled) + require.True(t, mockBasicService.awaitTerminatedCalled) }) } func TestManagerAdapter_Shutdown(t *testing.T) { - t.Run("shutdown with nil manager returns nil", func(t *testing.T) { + t.Run("calls StopAsync and AwaitTerminated", func(t *testing.T) { reg := &mockBackgroundServiceRegistry{} adapter := NewManagerAdapter(reg) - err := adapter.Shutdown(context.Background(), "test shutdown") - require.NoError(t, err) - }) + mockBasicService := &mockBasicService{} + adapter.NamedService = mockBasicService - 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) + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) 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") + err := adapter.Shutdown(ctx, "test shutdown") require.NoError(t, err) + require.True(t, mockBasicService.stopAsyncCalled) + require.True(t, mockBasicService.awaitTerminatedCalled) }) -} -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}, - } + t.Run("returns AwaitTerminated error", func(t *testing.T) { + reg := &mockBackgroundServiceRegistry{} adapter := NewManagerAdapter(reg) - require.NotNil(t, adapter) - require.Equal(t, reg, adapter.reg) - require.Len(t, reg.GetServices(), 2) + expectedErr := errors.New("await error") + mockBasicService := &mockBasicService{awaitTerminatedError: expectedErr} + adapter.NamedService = mockBasicService + + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + err := adapter.Shutdown(ctx, "test shutdown") + require.Error(t, err) + require.Equal(t, expectedErr, err) }) } @@ -193,6 +371,29 @@ func (m *mockBackgroundServiceRegistry) GetServices() []registry.BackgroundServi return m.services } +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 +} + type mockNamedService struct { mockService name string @@ -205,3 +406,63 @@ func (m *mockNamedService) ServiceName() string { func (m *mockNamedService) State() services.State { return services.New } + +func (m *mockNamedService) AddListener(listener services.Listener) func() { + return func() {} +} + +func (m *mockNamedService) FailureCase() error { + return nil +} + +type mockBasicService struct { + startAsyncCalled bool + startAsyncError error + awaitTerminatedCalled bool + awaitTerminatedError error + stopAsyncCalled bool +} + +func (m *mockBasicService) StartAsync(ctx context.Context) error { + m.startAsyncCalled = true + return m.startAsyncError +} + +func (m *mockBasicService) AwaitRunning(ctx context.Context) error { + return nil +} + +func (m *mockBasicService) StopAsync() { + m.stopAsyncCalled = true +} + +func (m *mockBasicService) AwaitTerminated(ctx context.Context) error { + m.awaitTerminatedCalled = true + return m.awaitTerminatedError +} + +func (m *mockBasicService) State() services.State { + return services.New +} + +func (m *mockBasicService) ServiceName() string { + return "mockBasicService" +} + +func (m *mockBasicService) AddListener(listener services.Listener) func() { + return func() {} +} + +func (m *mockBasicService) FailureCase() error { + return nil +} + +// simpleBackgroundService only implements BackgroundService, not CanBeDisabled +type simpleBackgroundService struct { + runCalled bool +} + +func (s *simpleBackgroundService) Run(ctx context.Context) error { + s.runCalled = true + return nil +} diff --git a/pkg/registry/backgroundsvcs/adapter/service_test.go b/pkg/registry/backgroundsvcs/adapter/service_test.go index 463ae130e76..368c596d395 100644 --- a/pkg/registry/backgroundsvcs/adapter/service_test.go +++ b/pkg/registry/backgroundsvcs/adapter/service_test.go @@ -5,143 +5,38 @@ import ( "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) { + t.Run("creates service adapter with correct properties", func(t *testing.T) { mockSvc := &mockService{} adapter := asNamedService(mockSvc) require.NotNil(t, adapter) require.NotNil(t, adapter.BasicService) + require.Equal(t, mockSvc, adapter.service) 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 + t.Run("different service types get different names", func(t *testing.T) { mockSvc1 := &mockService{} adapter1 := asNamedService(mockSvc1) - // Create a different type for comparison type anotherMockService struct{ mockService } mockSvc2 := &anotherMockService{} adapter2 := asNamedService(mockSvc2) @@ -212,29 +107,3 @@ func TestServiceAdapter_ErrorHandling(t *testing.T) { 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/module_server.go b/pkg/server/module_server.go index 0a87e4afe67..6d26f9aee8f 100644 --- a/pkg/server/module_server.go +++ b/pkg/server/module_server.go @@ -155,7 +155,7 @@ func (s *ModuleServer) Run() error { s.notifySystemd("READY=1") s.log.Debug("Waiting on services...") - m := modules.New(s.cfg.Target) + m := modules.New(s.log, s.cfg.Target) // only run the instrumentation server module if were not running a module that already contains an http server m.RegisterInvisibleModule(modules.InstrumentationServer, func() (services.Service, error) { diff --git a/pkg/server/server.go b/pkg/server/server.go index 4d184dea2bb..51a05827369 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -11,9 +11,10 @@ import ( "strconv" "sync" - "golang.org/x/sync/errgroup" - + "github.com/grafana/dskit/modules" "github.com/prometheus/client_golang/prometheus" + "go.opentelemetry.io/otel/trace" + "golang.org/x/sync/errgroup" "github.com/grafana/grafana/pkg/api" _ "github.com/grafana/grafana/pkg/extensions" @@ -23,6 +24,7 @@ import ( "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/semconv" "github.com/grafana/grafana/pkg/services/accesscontrol" "github.com/grafana/grafana/pkg/services/featuremgmt" "github.com/grafana/grafana/pkg/services/provisioning" @@ -94,7 +96,7 @@ func newServer(opts Options, cfg *setting.Cfg, httpServer *api.HTTPServer, roleR // ModuleServer to launch specific modules. type Server struct { context context.Context - shutdownFn context.CancelFunc + shutdownFn func() childRoutines *errgroup.Group log log.Logger cfg *setting.Cfg @@ -152,18 +154,33 @@ func (s *Server) Run() error { } 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) + ctx, span := s.tracerProvider.Start(s.context, "server.dskitRun") + defer span.End() + + // override the shutdownFn (context cancel func) for now until the feature flag is removed. + // this is a temporary solution to ensure that the services are shutdown properly. + cancelFn := s.shutdownFn + s.shutdownFn = func() { + defer close(s.shutdownFinished) + s.log.Debug("Shutting down background services") + if err := managerAdapter.Shutdown(s.context, modules.ErrStopProcess.Error()); err != nil { + s.log.Error("Failed to shutdown background services", "error", err) + } + cancelFn() + } + + return managerAdapter.Run(ctx) } func (s *Server) backgroundServicesRun() error { + ctx, span := s.tracerProvider.Start(s.context, "server.backgroundServicesRun") + defer span.End() defer close(s.shutdownFinished) if err := s.Init(); err != nil { @@ -187,7 +204,8 @@ func (s *Server) backgroundServicesRun() error { default: } s.log.Debug("Starting background service", "service", serviceName) - err := service.Run(s.context) + span.AddEvent(fmt.Sprintf("%s start", serviceName), trace.WithAttributes(semconv.GrafanaServiceName(serviceName))) + err := service.Run(ctx) // 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. diff --git a/pkg/services/apiserver/service.go b/pkg/services/apiserver/service.go index 73de7770a2a..69381a2bc99 100644 --- a/pkg/services/apiserver/service.go +++ b/pkg/services/apiserver/service.go @@ -581,7 +581,6 @@ func (s *service) running(ctx context.Context) error { return err } case <-ctx.Done(): - return ctx.Err() } return nil }