Background Services: Lifecycle o11y (#110570)

This commit is contained in:
Todd Treece
2025-09-10 14:33:03 -04:00
committed by GitHub
parent d3c0980c84
commit 191ddb127a
14 changed files with 1736 additions and 419 deletions
+72 -47
View File
@@ -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)
}
+117
View File
@@ -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
}
+617
View File
@@ -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")
})
}
+70
View File
@@ -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
}
}
+342
View File
@@ -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
}
-35
View File
@@ -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)
}