From e53b5902fae548fedc637900f4af86ca34301f91 Mon Sep 17 00:00:00 2001 From: Stephanie Hingtgen Date: Mon, 22 Sep 2025 22:23:29 -0600 Subject: [PATCH] Provisioning: Wire up tracing and add trace for job sync (#111455) --- pkg/operators/provisioning/jobs_operator.go | 18 ++++- pkg/operators/provisioning/repo_operator.go | 12 ++++ .../provisioning/controller/repository.go | 4 ++ .../apis/provisioning/jobs/sync/full.go | 61 +++++++++++----- .../jobs/sync/full_sync_fn_mock.go | 23 +++--- .../apis/provisioning/jobs/sync/full_test.go | 15 ++-- .../provisioning/jobs/sync/incremental.go | 45 ++++++++---- .../jobs/sync/incremental_sync_fn_mock.go | 23 +++--- .../jobs/sync/incremental_test.go | 5 +- .../apis/provisioning/jobs/sync/sync.go | 13 ++-- .../apis/provisioning/jobs/sync/sync_test.go | 8 ++- .../apis/provisioning/jobs/sync/worker.go | 72 +++++++++++++++---- .../provisioning/jobs/sync/worker_test.go | 6 +- pkg/registry/apis/provisioning/register.go | 4 +- 14 files changed, 223 insertions(+), 86 deletions(-) diff --git a/pkg/operators/provisioning/jobs_operator.go b/pkg/operators/provisioning/jobs_operator.go index fb9fd9c494e..f901988e0e0 100644 --- a/pkg/operators/provisioning/jobs_operator.go +++ b/pkg/operators/provisioning/jobs_operator.go @@ -13,6 +13,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "k8s.io/client-go/tools/cache" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/export" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs/migrate" @@ -34,6 +35,16 @@ func RunJobController(deps server.OperatorDependencies) error { })).With("logger", "provisioning-job-controller") logger.Info("Starting provisioning job controller") + tracingConfig, err := tracing.ProvideTracingConfig(deps.Config) + if err != nil { + return fmt.Errorf("failed to provide tracing config: %w", err) + } + + tracer, err := tracing.ProvideService(tracingConfig) + if err != nil { + return fmt.Errorf("failed to provide tracing service: %w", err) + } + controllerCfg, err := setupJobsControllerFromConfig(deps.Config, deps.Registerer) if err != nil { return fmt.Errorf("failed to setup operator: %w", err) @@ -100,7 +111,7 @@ func RunJobController(deps server.OperatorDependencies) error { return fmt.Errorf("create API client job store: %w", err) } - workers, err := setupWorkers(controllerCfg, deps.Registerer) + workers, err := setupWorkers(controllerCfg, deps.Registerer, tracer) if err != nil { return fmt.Errorf("setup workers: %w", err) } @@ -175,7 +186,7 @@ func setupJobsControllerFromConfig(cfg *setting.Cfg, registry prometheus.Registe }, nil } -func setupWorkers(controllerCfg *jobsControllerConfig, registry prometheus.Registerer) ([]jobs.Worker, error) { +func setupWorkers(controllerCfg *jobsControllerConfig, registry prometheus.Registerer, tracer tracing.Tracer) ([]jobs.Worker, error) { clients := controllerCfg.clients parsers := resources.NewParserFactory(clients) resourceLister := resources.NewResourceLister(controllerCfg.unified) @@ -187,7 +198,7 @@ func setupWorkers(controllerCfg *jobsControllerConfig, registry prometheus.Regis metrics := jobs.RegisterJobMetrics(registry) // Sync - syncer := sync.NewSyncer(sync.Compare, sync.FullSync, sync.IncrementalSync) + syncer := sync.NewSyncer(sync.Compare, sync.FullSync, sync.IncrementalSync, tracer) syncWorker := sync.NewSyncWorker( clients, repositoryResources, @@ -195,6 +206,7 @@ func setupWorkers(controllerCfg *jobsControllerConfig, registry prometheus.Regis statusPatcher.Patch, syncer, metrics, + tracer, ) workers = append(workers, syncWorker) diff --git a/pkg/operators/provisioning/repo_operator.go b/pkg/operators/provisioning/repo_operator.go index a15c2033d57..0ec243bb41e 100644 --- a/pkg/operators/provisioning/repo_operator.go +++ b/pkg/operators/provisioning/repo_operator.go @@ -14,6 +14,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "k8s.io/client-go/tools/cache" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/controller" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" @@ -34,6 +35,16 @@ func RunRepoController(deps server.OperatorDependencies) error { return fmt.Errorf("failed to setup operator: %w", err) } + tracingConfig, err := tracing.ProvideTracingConfig(deps.Config) + if err != nil { + return fmt.Errorf("failed to provide tracing config: %w", err) + } + + tracer, err := tracing.ProvideService(tracingConfig) + if err != nil { + return fmt.Errorf("failed to provide tracing service: %w", err) + } + ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -70,6 +81,7 @@ func RunRepoController(deps server.OperatorDependencies) error { healthChecker, statusPatcher, deps.Registerer, + tracer, ) if err != nil { return fmt.Errorf("failed to create repository controller: %w", err) diff --git a/pkg/registry/apis/provisioning/controller/repository.go b/pkg/registry/apis/provisioning/controller/repository.go index cbc55781819..ccda76e1722 100644 --- a/pkg/registry/apis/provisioning/controller/repository.go +++ b/pkg/registry/apis/provisioning/controller/repository.go @@ -22,6 +22,7 @@ import ( listers "github.com/grafana/grafana/apps/provisioning/pkg/generated/listers/provisioning/v0alpha1" "github.com/grafana/grafana/apps/provisioning/pkg/repository" "github.com/grafana/grafana/pkg/apimachinery/identity" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" "github.com/grafana/grafana/pkg/storage/legacysql/dualwrite" @@ -62,6 +63,7 @@ type RepositoryController struct { queue workqueue.TypedRateLimitingInterface[*queueItem] registry prometheus.Registerer + tracer tracing.Tracer } // NewRepositoryController creates new RepositoryController. @@ -76,6 +78,7 @@ func NewRepositoryController( healthChecker *HealthChecker, statusPatcher StatusPatcher, registry prometheus.Registerer, + tracer tracing.Tracer, ) (*RepositoryController, error) { finalizerMetrics := registerFinalizerMetrics(registry) @@ -101,6 +104,7 @@ func NewRepositoryController( logger: logging.DefaultLogger.With("logger", loggerName), dualwrite: dualwrite, registry: registry, + tracer: tracer, } _, err := repoInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ diff --git a/pkg/registry/apis/provisioning/jobs/sync/full.go b/pkg/registry/apis/provisioning/jobs/sync/full.go index 8114d8a67ce..0e0b107171a 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/full.go +++ b/pkg/registry/apis/provisioning/jobs/sync/full.go @@ -6,8 +6,11 @@ import ( "github.com/grafana/grafana/apps/provisioning/pkg/repository" "github.com/grafana/grafana/apps/provisioning/pkg/safepath" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" ) @@ -20,47 +23,63 @@ func FullSync( currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, + tracer tracing.Tracer, ) error { cfg := repo.Config() + ctx, span := tracer.Start(ctx, "provisioning.sync.full") + defer span.End() + + ensureFolderCtx, ensureFolderSpan := tracer.Start(ctx, "provisioning.sync.full.ensure_folder_exists") // Ensure the configured folder exists and is managed by the repository rootFolder := resources.RootFolder(cfg) if rootFolder != "" { - if err := repositoryResources.EnsureFolderExists(ctx, resources.Folder{ + if err := repositoryResources.EnsureFolderExists(ensureFolderCtx, resources.Folder{ ID: rootFolder, // will not change if exists Title: cfg.Spec.Title, Path: "", // at the root of the repository }, ""); err != nil { - return fmt.Errorf("create root folder: %w", err) + ensureFolderSpan.End() + return tracing.Error(span, fmt.Errorf("create root folder: %w", err)) } } + ensureFolderSpan.End() - changes, err := compare(ctx, repo, repositoryResources, currentRef) + compareCtx, compareSpan := tracer.Start(ctx, "provisioning.sync.full.compare") + changes, err := compare(compareCtx, repo, repositoryResources, currentRef) if err != nil { - return fmt.Errorf("compare changes: %w", err) + compareSpan.End() + return tracing.Error(span, fmt.Errorf("compare changes: %w", err)) } + compareSpan.End() if len(changes) == 0 { progress.SetFinalMessage(ctx, "no changes to sync") return nil } - return applyChanges(ctx, changes, clients, repositoryResources, progress) + return applyChanges(ctx, changes, clients, repositoryResources, progress, tracer) } -func applyChanges(ctx context.Context, changes []ResourceFileChange, clients resources.ResourceClients, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder) error { +func applyChanges(ctx context.Context, changes []ResourceFileChange, clients resources.ResourceClients, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) error { progress.SetTotal(ctx, len(changes)) + _, applyChangesSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes", + trace.WithAttributes(attribute.Int("changes_count", len(changes))), + ) + defer applyChangesSpan.End() + for _, change := range changes { if ctx.Err() != nil { return ctx.Err() } if err := progress.TooManyErrors(); err != nil { - return err + return tracing.Error(applyChangesSpan, err) } if change.Action == repository.FileActionDeleted { + deleteCtx, deleteSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes.delete") result := jobs.JobResourceResult{ Path: change.Path, Action: change.Action, @@ -68,7 +87,9 @@ func applyChanges(ctx context.Context, changes []ResourceFileChange, clients res if change.Existing == nil || change.Existing.Name == "" { result.Error = fmt.Errorf("processing deletion for file %s: missing existing reference", change.Path) - progress.Record(ctx, result) + progress.Record(deleteCtx, result) + deleteSpan.RecordError(result.Error) + deleteSpan.End() continue } @@ -82,22 +103,24 @@ func applyChanges(ctx context.Context, changes []ResourceFileChange, clients res } // TODO: should we use the clients or the resource manager instead? - client, _, err := clients.ForResource(ctx, versionlessGVR) + client, _, err := clients.ForResource(deleteCtx, versionlessGVR) if err != nil { result.Error = fmt.Errorf("get client for deleted object: %w", err) - progress.Record(ctx, result) + progress.Record(deleteCtx, result) continue } - if err := client.Delete(ctx, change.Existing.Name, metav1.DeleteOptions{}); err != nil { + if err := client.Delete(deleteCtx, change.Existing.Name, metav1.DeleteOptions{}); err != nil { result.Error = fmt.Errorf("deleting resource %s/%s %s: %w", change.Existing.Group, change.Existing.Resource, change.Existing.Name, err) } - progress.Record(ctx, result) + progress.Record(deleteCtx, result) + deleteSpan.End() continue } // If folder ensure it exists if safepath.IsDir(change.Path) { + ensureFolderCtx, ensureFolderSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes.ensure_folder_exists") result := jobs.JobResourceResult{ Path: change.Path, Action: change.Action, @@ -105,20 +128,24 @@ func applyChanges(ctx context.Context, changes []ResourceFileChange, clients res Group: resources.FolderResource.Group, } - folder, err := repositoryResources.EnsureFolderPathExist(ctx, change.Path) + folder, err := repositoryResources.EnsureFolderPathExist(ensureFolderCtx, change.Path) if err != nil { result.Error = fmt.Errorf("ensuring folder exists at path %s: %w", change.Path, err) + ensureFolderSpan.RecordError(err) + ensureFolderSpan.End() progress.Record(ctx, result) continue } result.Name = folder - progress.Record(ctx, result) + progress.Record(ensureFolderCtx, result) + ensureFolderSpan.End() continue } - name, gvk, err := repositoryResources.WriteResourceFromFile(ctx, change.Path, "") + writeCtx, writeSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes.write_resource_from_file") + name, gvk, err := repositoryResources.WriteResourceFromFile(writeCtx, change.Path, "") result := jobs.JobResourceResult{ Path: change.Path, Action: change.Action, @@ -128,9 +155,11 @@ func applyChanges(ctx context.Context, changes []ResourceFileChange, clients res } if err != nil { + writeSpan.RecordError(err) result.Error = fmt.Errorf("writing resource from file %s: %w", change.Path, err) } - progress.Record(ctx, result) + progress.Record(writeCtx, result) + writeSpan.End() } return nil diff --git a/pkg/registry/apis/provisioning/jobs/sync/full_sync_fn_mock.go b/pkg/registry/apis/provisioning/jobs/sync/full_sync_fn_mock.go index 03b080cff25..3f10550e3f8 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/full_sync_fn_mock.go +++ b/pkg/registry/apis/provisioning/jobs/sync/full_sync_fn_mock.go @@ -11,6 +11,8 @@ import ( repository "github.com/grafana/grafana/apps/provisioning/pkg/repository" resources "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" + + tracing "github.com/grafana/grafana/pkg/infra/tracing" ) // MockFullSyncFn is an autogenerated mock type for the FullSyncFn type @@ -26,17 +28,17 @@ func (_m *MockFullSyncFn) EXPECT() *MockFullSyncFn_Expecter { return &MockFullSyncFn_Expecter{mock: &_m.Mock} } -// Execute provides a mock function with given fields: ctx, repo, compare, clients, currentRef, repositoryResources, progress -func (_m *MockFullSyncFn) Execute(ctx context.Context, repo repository.Reader, compare CompareFn, clients resources.ResourceClients, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder) error { - ret := _m.Called(ctx, repo, compare, clients, currentRef, repositoryResources, progress) +// Execute provides a mock function with given fields: ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer +func (_m *MockFullSyncFn) Execute(ctx context.Context, repo repository.Reader, compare CompareFn, clients resources.ResourceClients, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) error { + ret := _m.Called(ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer) if len(ret) == 0 { panic("no return value specified for Execute") } var r0 error - if rf, ok := ret.Get(0).(func(context.Context, repository.Reader, CompareFn, resources.ResourceClients, string, resources.RepositoryResources, jobs.JobProgressRecorder) error); ok { - r0 = rf(ctx, repo, compare, clients, currentRef, repositoryResources, progress) + if rf, ok := ret.Get(0).(func(context.Context, repository.Reader, CompareFn, resources.ResourceClients, string, resources.RepositoryResources, jobs.JobProgressRecorder, tracing.Tracer) error); ok { + r0 = rf(ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer) } else { r0 = ret.Error(0) } @@ -57,13 +59,14 @@ type MockFullSyncFn_Execute_Call struct { // - currentRef string // - repositoryResources resources.RepositoryResources // - progress jobs.JobProgressRecorder -func (_e *MockFullSyncFn_Expecter) Execute(ctx interface{}, repo interface{}, compare interface{}, clients interface{}, currentRef interface{}, repositoryResources interface{}, progress interface{}) *MockFullSyncFn_Execute_Call { - return &MockFullSyncFn_Execute_Call{Call: _e.mock.On("Execute", ctx, repo, compare, clients, currentRef, repositoryResources, progress)} +// - tracer tracing.Tracer +func (_e *MockFullSyncFn_Expecter) Execute(ctx interface{}, repo interface{}, compare interface{}, clients interface{}, currentRef interface{}, repositoryResources interface{}, progress interface{}, tracer interface{}) *MockFullSyncFn_Execute_Call { + return &MockFullSyncFn_Execute_Call{Call: _e.mock.On("Execute", ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer)} } -func (_c *MockFullSyncFn_Execute_Call) Run(run func(ctx context.Context, repo repository.Reader, compare CompareFn, clients resources.ResourceClients, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder)) *MockFullSyncFn_Execute_Call { +func (_c *MockFullSyncFn_Execute_Call) Run(run func(ctx context.Context, repo repository.Reader, compare CompareFn, clients resources.ResourceClients, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer)) *MockFullSyncFn_Execute_Call { _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(repository.Reader), args[2].(CompareFn), args[3].(resources.ResourceClients), args[4].(string), args[5].(resources.RepositoryResources), args[6].(jobs.JobProgressRecorder)) + run(args[0].(context.Context), args[1].(repository.Reader), args[2].(CompareFn), args[3].(resources.ResourceClients), args[4].(string), args[5].(resources.RepositoryResources), args[6].(jobs.JobProgressRecorder), args[7].(tracing.Tracer)) }) return _c } @@ -73,7 +76,7 @@ func (_c *MockFullSyncFn_Execute_Call) Return(_a0 error) *MockFullSyncFn_Execute return _c } -func (_c *MockFullSyncFn_Execute_Call) RunAndReturn(run func(context.Context, repository.Reader, CompareFn, resources.ResourceClients, string, resources.RepositoryResources, jobs.JobProgressRecorder) error) *MockFullSyncFn_Execute_Call { +func (_c *MockFullSyncFn_Execute_Call) RunAndReturn(run func(context.Context, repository.Reader, CompareFn, resources.ResourceClients, string, resources.RepositoryResources, jobs.JobProgressRecorder, tracing.Tracer) error) *MockFullSyncFn_Execute_Call { _c.Call.Return(run) return _c } diff --git a/pkg/registry/apis/provisioning/jobs/sync/full_test.go b/pkg/registry/apis/provisioning/jobs/sync/full_test.go index 0f051e20b72..d5b3b610b80 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/full_test.go +++ b/pkg/registry/apis/provisioning/jobs/sync/full_test.go @@ -8,6 +8,7 @@ import ( provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" "github.com/grafana/grafana/apps/provisioning/pkg/repository" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" "github.com/stretchr/testify/mock" @@ -41,7 +42,7 @@ func TestFullSync_ContextCancelled(t *testing.T) { compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]ResourceFileChange{{}}, nil) progress.On("SetTotal", mock.Anything, 1).Return() - err := FullSync(ctx, repo, compareFn.Execute, clients, "current-ref", repoResources, progress) + err := FullSync(ctx, repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService()) require.EqualError(t, err, "context canceled") } @@ -60,7 +61,7 @@ func TestFullSync_Error(t *testing.T) { compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, fmt.Errorf("some error")) - err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService()) require.EqualError(t, err, "compare changes: some error") } @@ -80,7 +81,7 @@ func TestFullSync_NoChanges(t *testing.T) { compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]ResourceFileChange{}, nil) progress.On("SetFinalMessage", mock.Anything, "no changes to sync").Return() - err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService()) require.NoError(t, err) } @@ -111,7 +112,7 @@ func TestFullSync_SuccessfulFolderCreation(t *testing.T) { Path: "", }, "").Return(nil) - err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService()) require.NoError(t, err) } @@ -140,7 +141,7 @@ func TestFullSync_FolderCreationFailed(t *testing.T) { Path: "", }, "").Return(fmt.Errorf("folder creation failed")) - err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService()) require.Error(t, err) require.Contains(t, err.Error(), "create root folder: folder creation failed") } @@ -169,7 +170,7 @@ func TestFullSync_FolderCreationFailedWithInstanceTarget(t *testing.T) { compareFn.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything). Return(nil, fmt.Errorf("compare error")) - err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService()) require.Error(t, err) require.Contains(t, err.Error(), "compare changes: compare error") } @@ -700,7 +701,7 @@ func TestFullSync_ApplyChanges(t *testing.T) { //nolint:gocyclo }) progress.On("SetTotal", mock.Anything, len(tt.changes)).Return() - err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService()) if tt.expectedError != "" { require.EqualError(t, err, tt.expectedError, tt.description) } else { diff --git a/pkg/registry/apis/provisioning/jobs/sync/incremental.go b/pkg/registry/apis/provisioning/jobs/sync/incremental.go index fe66ad058ee..f7448bffa16 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/incremental.go +++ b/pkg/registry/apis/provisioning/jobs/sync/incremental.go @@ -6,21 +6,29 @@ import ( "github.com/grafana/grafana/apps/provisioning/pkg/repository" "github.com/grafana/grafana/apps/provisioning/pkg/safepath" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" ) // Convert git changes into resource file changes -func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder) error { +func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) error { if previousRef == currentRef { progress.SetFinalMessage(ctx, "same commit as last time") return nil } - diff, err := repo.CompareFiles(ctx, previousRef, currentRef) + ctx, span := tracer.Start(ctx, "provisioning.sync.incremental") + defer span.End() + + compareCtx, compareSpan := tracer.Start(ctx, "provisioning.sync.incremental.compare_files") + diff, err := repo.CompareFiles(compareCtx, previousRef, currentRef) if err != nil { - return fmt.Errorf("compare files error: %w", err) + compareSpan.RecordError(err) + compareSpan.End() + return tracing.Error(span, fmt.Errorf("compare files error: %w", err)) } + compareSpan.End() if len(diff) < 1 { progress.SetFinalMessage(ctx, "no changes detected between commits") @@ -35,10 +43,11 @@ func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef return ctx.Err() } if err := progress.TooManyErrors(); err != nil { - return err + return tracing.Error(span, err) } if err := resources.IsPathSupported(change.Path); err != nil { + ensureFolderCtx, ensureFolderSpan := tracer.Start(ctx, "provisioning.sync.incremental.ensure_folder_path_exist") // Maintain the safe segment for empty folders safeSegment := safepath.SafeSegment(change.Path) if !safepath.IsDir(safeSegment) { @@ -46,26 +55,29 @@ func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef } if safeSegment != "" && resources.IsPathSupported(safeSegment) == nil { - folder, err := repositoryResources.EnsureFolderPathExist(ctx, safeSegment) + folder, err := repositoryResources.EnsureFolderPathExist(ensureFolderCtx, safeSegment) if err != nil { - return fmt.Errorf("unable to create empty file folder: %w", err) + ensureFolderSpan.RecordError(err) + ensureFolderSpan.End() + return tracing.Error(span, fmt.Errorf("unable to create empty file folder: %w", err)) } - progress.Record(ctx, jobs.JobResourceResult{ + progress.Record(ensureFolderCtx, jobs.JobResourceResult{ Path: safeSegment, Action: repository.FileActionCreated, Resource: resources.FolderResource.Resource, Group: resources.FolderResource.Group, Name: folder, }) - + ensureFolderSpan.End() continue } - progress.Record(ctx, jobs.JobResourceResult{ + progress.Record(ensureFolderCtx, jobs.JobResourceResult{ Path: change.Path, Action: repository.FileActionIgnored, }) + ensureFolderSpan.End() continue } @@ -76,29 +88,38 @@ func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef switch change.Action { case repository.FileActionCreated, repository.FileActionUpdated: - name, gvk, err := repositoryResources.WriteResourceFromFile(ctx, change.Path, change.Ref) + writeCtx, writeSpan := tracer.Start(ctx, "provisioning.sync.incremental.write_resource_from_file") + name, gvk, err := repositoryResources.WriteResourceFromFile(writeCtx, change.Path, change.Ref) if err != nil { + writeSpan.RecordError(err) result.Error = fmt.Errorf("writing resource from file %s: %w", change.Path, err) } result.Name = name result.Resource = gvk.Kind result.Group = gvk.Group + writeSpan.End() case repository.FileActionDeleted: - name, gvk, err := repositoryResources.RemoveResourceFromFile(ctx, change.Path, change.PreviousRef) + removeCtx, removeSpan := tracer.Start(ctx, "provisioning.sync.incremental.remove_resource_from_file") + name, gvk, err := repositoryResources.RemoveResourceFromFile(removeCtx, change.Path, change.PreviousRef) if err != nil { + removeSpan.RecordError(err) result.Error = fmt.Errorf("removing resource from file %s: %w", change.Path, err) } result.Name = name result.Resource = gvk.Kind result.Group = gvk.Group + removeSpan.End() case repository.FileActionRenamed: - name, gvk, err := repositoryResources.RenameResourceFile(ctx, change.PreviousPath, change.PreviousRef, change.Path, change.Ref) + renameCtx, renameSpan := tracer.Start(ctx, "provisioning.sync.incremental.rename_resource_file") + name, gvk, err := repositoryResources.RenameResourceFile(renameCtx, change.PreviousPath, change.PreviousRef, change.Path, change.Ref) if err != nil { + renameSpan.RecordError(err) result.Error = fmt.Errorf("renaming resource file from %s to %s: %w", change.PreviousPath, change.Path, err) } result.Name = name result.Resource = gvk.Kind result.Group = gvk.Group + renameSpan.End() case repository.FileActionIgnored: // do nothing } diff --git a/pkg/registry/apis/provisioning/jobs/sync/incremental_sync_fn_mock.go b/pkg/registry/apis/provisioning/jobs/sync/incremental_sync_fn_mock.go index c1ae1325475..ff032146788 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/incremental_sync_fn_mock.go +++ b/pkg/registry/apis/provisioning/jobs/sync/incremental_sync_fn_mock.go @@ -11,6 +11,8 @@ import ( repository "github.com/grafana/grafana/apps/provisioning/pkg/repository" resources "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" + + tracing "github.com/grafana/grafana/pkg/infra/tracing" ) // MockIncrementalSyncFn is an autogenerated mock type for the IncrementalSyncFn type @@ -26,17 +28,17 @@ func (_m *MockIncrementalSyncFn) EXPECT() *MockIncrementalSyncFn_Expecter { return &MockIncrementalSyncFn_Expecter{mock: &_m.Mock} } -// Execute provides a mock function with given fields: ctx, repo, previousRef, currentRef, repositoryResources, progress -func (_m *MockIncrementalSyncFn) Execute(ctx context.Context, repo repository.Versioned, previousRef string, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder) error { - ret := _m.Called(ctx, repo, previousRef, currentRef, repositoryResources, progress) +// Execute provides a mock function with given fields: ctx, repo, previousRef, currentRef, repositoryResources, progress, tracer +func (_m *MockIncrementalSyncFn) Execute(ctx context.Context, repo repository.Versioned, previousRef string, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) error { + ret := _m.Called(ctx, repo, previousRef, currentRef, repositoryResources, progress, tracer) if len(ret) == 0 { panic("no return value specified for Execute") } var r0 error - if rf, ok := ret.Get(0).(func(context.Context, repository.Versioned, string, string, resources.RepositoryResources, jobs.JobProgressRecorder) error); ok { - r0 = rf(ctx, repo, previousRef, currentRef, repositoryResources, progress) + if rf, ok := ret.Get(0).(func(context.Context, repository.Versioned, string, string, resources.RepositoryResources, jobs.JobProgressRecorder, tracing.Tracer) error); ok { + r0 = rf(ctx, repo, previousRef, currentRef, repositoryResources, progress, tracer) } else { r0 = ret.Error(0) } @@ -56,13 +58,14 @@ type MockIncrementalSyncFn_Execute_Call struct { // - currentRef string // - repositoryResources resources.RepositoryResources // - progress jobs.JobProgressRecorder -func (_e *MockIncrementalSyncFn_Expecter) Execute(ctx interface{}, repo interface{}, previousRef interface{}, currentRef interface{}, repositoryResources interface{}, progress interface{}) *MockIncrementalSyncFn_Execute_Call { - return &MockIncrementalSyncFn_Execute_Call{Call: _e.mock.On("Execute", ctx, repo, previousRef, currentRef, repositoryResources, progress)} +// - tracer tracing.Tracer +func (_e *MockIncrementalSyncFn_Expecter) Execute(ctx interface{}, repo interface{}, previousRef interface{}, currentRef interface{}, repositoryResources interface{}, progress interface{}, tracer interface{}) *MockIncrementalSyncFn_Execute_Call { + return &MockIncrementalSyncFn_Execute_Call{Call: _e.mock.On("Execute", ctx, repo, previousRef, currentRef, repositoryResources, progress, tracer)} } -func (_c *MockIncrementalSyncFn_Execute_Call) Run(run func(ctx context.Context, repo repository.Versioned, previousRef string, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder)) *MockIncrementalSyncFn_Execute_Call { +func (_c *MockIncrementalSyncFn_Execute_Call) Run(run func(ctx context.Context, repo repository.Versioned, previousRef string, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer)) *MockIncrementalSyncFn_Execute_Call { _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(repository.Versioned), args[2].(string), args[3].(string), args[4].(resources.RepositoryResources), args[5].(jobs.JobProgressRecorder)) + run(args[0].(context.Context), args[1].(repository.Versioned), args[2].(string), args[3].(string), args[4].(resources.RepositoryResources), args[5].(jobs.JobProgressRecorder), args[6].(tracing.Tracer)) }) return _c } @@ -72,7 +75,7 @@ func (_c *MockIncrementalSyncFn_Execute_Call) Return(_a0 error) *MockIncremental return _c } -func (_c *MockIncrementalSyncFn_Execute_Call) RunAndReturn(run func(context.Context, repository.Versioned, string, string, resources.RepositoryResources, jobs.JobProgressRecorder) error) *MockIncrementalSyncFn_Execute_Call { +func (_c *MockIncrementalSyncFn_Execute_Call) RunAndReturn(run func(context.Context, repository.Versioned, string, string, resources.RepositoryResources, jobs.JobProgressRecorder, tracing.Tracer) error) *MockIncrementalSyncFn_Execute_Call { _c.Call.Return(run) return _c } diff --git a/pkg/registry/apis/provisioning/jobs/sync/incremental_test.go b/pkg/registry/apis/provisioning/jobs/sync/incremental_test.go index 290b40c77ce..6562be15f3c 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/incremental_test.go +++ b/pkg/registry/apis/provisioning/jobs/sync/incremental_test.go @@ -6,6 +6,7 @@ import ( "testing" "github.com/grafana/grafana/apps/provisioning/pkg/repository" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" "github.com/stretchr/testify/mock" @@ -29,7 +30,7 @@ func TestIncrementalSync_ContextCancelled(t *testing.T) { progress.On("SetTotal", mock.Anything, 1).Return() progress.On("SetMessage", mock.Anything, "replicating versioned changes").Return() - err := IncrementalSync(ctx, repo, "old-ref", "new-ref", repoResources, progress) + err := IncrementalSync(ctx, repo, "old-ref", "new-ref", repoResources, progress, tracing.NewNoopTracerService()) require.EqualError(t, err, "context canceled") } @@ -386,7 +387,7 @@ func TestIncrementalSync(t *testing.T) { tt.setupMocks(repo, repoResources, progress) - err := IncrementalSync(context.Background(), repo, tt.previousRef, tt.currentRef, repoResources, progress) + err := IncrementalSync(context.Background(), repo, tt.previousRef, tt.currentRef, repoResources, progress, tracing.NewNoopTracerService()) if tt.expectedError != "" { require.EqualError(t, err, tt.expectedError) diff --git a/pkg/registry/apis/provisioning/jobs/sync/sync.go b/pkg/registry/apis/provisioning/jobs/sync/sync.go index 541fd47a5f9..fa58d426cba 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/sync.go +++ b/pkg/registry/apis/provisioning/jobs/sync/sync.go @@ -6,18 +6,19 @@ import ( provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" "github.com/grafana/grafana/apps/provisioning/pkg/repository" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" ) //go:generate mockery --name FullSyncFn --structname MockFullSyncFn --inpackage --filename full_sync_fn_mock.go --with-expecter -type FullSyncFn func(ctx context.Context, repo repository.Reader, compare CompareFn, clients resources.ResourceClients, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder) error +type FullSyncFn func(ctx context.Context, repo repository.Reader, compare CompareFn, clients resources.ResourceClients, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) error //go:generate mockery --name CompareFn --structname MockCompareFn --inpackage --filename compare_fn_mock.go --with-expecter type CompareFn func(ctx context.Context, repo repository.Reader, repositoryResources resources.RepositoryResources, ref string) ([]ResourceFileChange, error) //go:generate mockery --name IncrementalSyncFn --structname MockIncrementalSyncFn --inpackage --filename incremental_sync_fn_mock.go --with-expecter -type IncrementalSyncFn func(ctx context.Context, repo repository.Versioned, previousRef, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder) error +type IncrementalSyncFn func(ctx context.Context, repo repository.Versioned, previousRef, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) error //go:generate mockery --name Syncer --structname MockSyncer --inpackage --filename syncer_mock.go --with-expecter type Syncer interface { @@ -28,13 +29,15 @@ type syncer struct { compare CompareFn fullSync FullSyncFn incrementalSync IncrementalSyncFn + tracer tracing.Tracer } -func NewSyncer(compare CompareFn, fullSync FullSyncFn, incrementalSync IncrementalSyncFn) Syncer { +func NewSyncer(compare CompareFn, fullSync FullSyncFn, incrementalSync IncrementalSyncFn, tracer tracing.Tracer) Syncer { return &syncer{ compare: compare, fullSync: fullSync, incrementalSync: incrementalSync, + tracer: tracer, } } @@ -52,11 +55,11 @@ func (r *syncer) Sync(ctx context.Context, repo repository.ReaderWriter, options if cfg.Status.Sync.LastRef != "" && options.Incremental { progress.SetMessage(ctx, "incremental sync") - return currentRef, r.incrementalSync(ctx, versionedRepo, cfg.Status.Sync.LastRef, currentRef, repositoryResources, progress) + return currentRef, r.incrementalSync(ctx, versionedRepo, cfg.Status.Sync.LastRef, currentRef, repositoryResources, progress, r.tracer) } } progress.SetMessage(ctx, "full sync") - return currentRef, r.fullSync(ctx, repo, r.compare, clients, currentRef, repositoryResources, progress) + return currentRef, r.fullSync(ctx, repo, r.compare, clients, currentRef, repositoryResources, progress, r.tracer) } diff --git a/pkg/registry/apis/provisioning/jobs/sync/sync_test.go b/pkg/registry/apis/provisioning/jobs/sync/sync_test.go index 1751159659a..5f04ed350e1 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/sync_test.go +++ b/pkg/registry/apis/provisioning/jobs/sync/sync_test.go @@ -7,6 +7,7 @@ import ( provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" "github.com/grafana/grafana/apps/provisioning/pkg/repository" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" mock "github.com/stretchr/testify/mock" @@ -65,7 +66,7 @@ func TestSyncer_Sync(t *testing.T) { repo.MockVersioned.On("LatestRef", mock.Anything).Return("new-ref", nil) progress.On("SetMessage", mock.Anything, "full sync").Return() - fullSyncFn.EXPECT().Execute(mock.Anything, mock.Anything, mock.Anything, mock.Anything, "new-ref", mock.Anything, mock.Anything).Return(nil) + fullSyncFn.EXPECT().Execute(mock.Anything, mock.Anything, mock.Anything, mock.Anything, "new-ref", mock.Anything, mock.Anything, mock.Anything).Return(nil) }, expectedMessages: []string{"full sync"}, }, @@ -87,7 +88,7 @@ func TestSyncer_Sync(t *testing.T) { }) repo.MockVersioned.On("LatestRef", mock.Anything).Return("new-ref", nil) progress.On("SetMessage", mock.Anything, "incremental sync").Return() - incrementalSyncFn.EXPECT().Execute(mock.Anything, mock.Anything, "old-ref", "new-ref", mock.Anything, mock.Anything).Return(nil) + incrementalSyncFn.EXPECT().Execute(mock.Anything, mock.Anything, "old-ref", "new-ref", mock.Anything, mock.Anything, mock.Anything).Return(nil) }, expectedRef: "new-ref", expectedMessages: []string{"incremental sync"}, @@ -130,7 +131,7 @@ func TestSyncer_Sync(t *testing.T) { }) repo.MockVersioned.On("LatestRef", mock.Anything).Return("new-ref", nil) progress.On("SetMessage", mock.Anything, "incremental sync").Return() - incrementalSyncFn.On("Execute", mock.Anything, mock.Anything, "old-ref", "new-ref", mock.Anything, mock.Anything).Return(fmt.Errorf("incremental sync failed")) + incrementalSyncFn.On("Execute", mock.Anything, mock.Anything, "old-ref", "new-ref", mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("incremental sync failed")) }, expectedRef: "new-ref", expectedMessages: []string{"incremental sync"}, @@ -158,6 +159,7 @@ func TestSyncer_Sync(t *testing.T) { compareFn.Execute, fullSyncFn.Execute, incrementalSyncFn.Execute, + tracing.NewNoopTracerService(), ) ref, err := syncer.Sync(context.Background(), repo, tt.options, repoResources, clients, progress) diff --git a/pkg/registry/apis/provisioning/jobs/sync/worker.go b/pkg/registry/apis/provisioning/jobs/sync/worker.go index 5443c814bcf..0b0139941c0 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/worker.go +++ b/pkg/registry/apis/provisioning/jobs/sync/worker.go @@ -8,10 +8,13 @@ import ( "github.com/grafana/grafana-app-sdk/logging" provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" "github.com/grafana/grafana/apps/provisioning/pkg/repository" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" "github.com/grafana/grafana/pkg/registry/apis/provisioning/utils" "github.com/grafana/grafana/pkg/storage/legacysql/dualwrite" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" ) //go:generate mockery --name RepositoryPatchFn --structname MockRepositoryPatchFn --inpackage --filename repository_patch_fn_mock.go --with-expecter @@ -36,6 +39,8 @@ type SyncWorker struct { syncer Syncer metrics jobs.JobMetrics + + tracer tracing.Tracer } func NewSyncWorker( @@ -45,6 +50,7 @@ func NewSyncWorker( patchStatus RepositoryPatchFn, syncer Syncer, metrics jobs.JobMetrics, + tracer tracing.Tracer, ) *SyncWorker { return &SyncWorker{ clients: clients, @@ -53,6 +59,7 @@ func NewSyncWorker( storageStatus: storageStatus, syncer: syncer, metrics: metrics, + tracer: tracer, } } @@ -63,23 +70,39 @@ func (r *SyncWorker) IsSupported(ctx context.Context, job provisioning.Job) bool func (r *SyncWorker) Process(ctx context.Context, repo repository.Repository, job provisioning.Job, progress jobs.JobProgressRecorder) error { cfg := repo.Config() logger := logging.FromContext(ctx).With("job", job.GetName(), "namespace", job.GetNamespace()) + ctx, span := r.tracer.Start(ctx, "provisioning.sync.process", + trace.WithAttributes( + attribute.String("job.name", job.GetName()), + attribute.String("job.namespace", job.GetNamespace()), + attribute.String("job.action", string(job.Spec.Action)), + attribute.String("repository.name", cfg.Name), + attribute.String("repository.namespace", cfg.Namespace), + ), + ) + defer span.End() start := time.Now() outcome := utils.ErrorOutcome totalChangesMade := 0 defer func() { r.metrics.RecordJob(string(provisioning.JobActionPull), outcome, totalChangesMade, time.Since(start).Seconds()) + span.SetAttributes( + attribute.String("outcome", outcome), + attribute.Int("changes_made", totalChangesMade), + ) }() // Check if we are onboarding from legacy storage // HACK -- this should be handled outside of this worker if r.storageStatus != nil && dualwrite.IsReadingLegacyDashboardsAndFolders(ctx, r.storageStatus) { - return fmt.Errorf("sync not supported until storage has migrated") + err := fmt.Errorf("sync not supported until storage has migrated") + return tracing.Error(span, err) } rw, ok := repo.(repository.ReaderWriter) if !ok { - return fmt.Errorf("sync job submitted for repository that does not support read-write") + err := fmt.Errorf("sync job submitted for repository that does not support read-write") + return tracing.Error(span, err) } syncStatus := job.Status.ToSyncStatus(job.Name) @@ -97,38 +120,49 @@ func (r *SyncWorker) Process(ctx context.Context, repo repository.Repository, jo } progress.SetMessage(ctx, "update sync status at start") - if err := r.patchStatus(ctx, cfg, patchOperations...); err != nil { + + statusCtx, statusSpan := r.tracer.Start(ctx, "provisioning.sync.update_start_status") + if err := r.patchStatus(statusCtx, cfg, patchOperations...); err != nil { + statusSpan.End() logger.Error("failed to update the repository status at the start of the sync job", "error", err) - return fmt.Errorf("update repo with job status at start: %w", err) + err = fmt.Errorf("update repo with job status at start: %w", err) + return tracing.Error(span, err) } + statusSpan.End() - repositoryResources, err := r.repositoryResources.Client(ctx, rw) + setupCtx, setupSpan := r.tracer.Start(ctx, "provisioning.sync.setup_clients") + repositoryResources, err := r.repositoryResources.Client(setupCtx, rw) if err != nil { + setupSpan.End() logger.Error("failed to create repository resources client", "error", err) - return fmt.Errorf("create repository resources client: %w", err) + err = fmt.Errorf("create repository resources client: %w", err) + return tracing.Error(span, err) } - - clients, err := r.clients.Clients(ctx, cfg.Namespace) + clients, err := r.clients.Clients(setupCtx, cfg.Namespace) if err != nil { + setupSpan.End() logger.Error("failed to get clients for the repository", "error", err) - return fmt.Errorf("get clients for %s: %w", cfg.Name, err) + err = fmt.Errorf("get clients for %s: %w", cfg.Name, err) + return tracing.Error(span, err) } + setupSpan.End() + syncCtx, syncSpan := r.tracer.Start(ctx, "provisioning.sync.execute") progress.SetMessage(ctx, "execute sync job") progress.StrictMaxErrors(20) // make it stop after 20 errors - - currentRef, syncError := r.syncer.Sync(ctx, rw, *job.Spec.Pull, repositoryResources, clients, progress) + currentRef, syncError := r.syncer.Sync(syncCtx, rw, *job.Spec.Pull, repositoryResources, clients, progress) jobStatus := progress.Complete(ctx, syncError) syncStatus = jobStatus.ToSyncStatus(job.Name) - if syncError != nil { logger.Debug("failed to sync the repository", "error", syncError) + _ = tracing.Error(syncSpan, syncError) } else { outcome = utils.SuccessOutcome for _, summary := range jobStatus.Summary { totalChangesMade += int(summary.Create + summary.Update + summary.Delete) } } + syncSpan.End() // Create sync status and set hash if successful if syncStatus.State == provisioning.JobStateSuccess { @@ -147,13 +181,17 @@ func (r *SyncWorker) Process(ctx context.Context, repo repository.Repository, jo }, } + finalCtx, finalSpan := r.tracer.Start(ctx, "provisioning.sync.update_final_status") + // Only add stats patch if stats are not nil - stats, err := repositoryResources.Stats(ctx) + stats, err := repositoryResources.Stats(finalCtx) switch { case err != nil: logger.Error("unable to read stats", "error", err) + finalSpan.SetAttributes(attribute.String("stats.error", err.Error())) case stats == nil: logger.Error("stats are nil") + finalSpan.SetAttributes(attribute.Bool("stats.nil", true)) case len(stats.Managed) == 1: patchOperations = append(patchOperations, map[string]interface{}{ "op": "replace", @@ -162,13 +200,17 @@ func (r *SyncWorker) Process(ctx context.Context, repo repository.Repository, jo }) default: logger.Warn("unexpected number of managed stats", "count", len(stats.Managed)) + finalSpan.SetAttributes(attribute.Int("stats.unexpected_count", len(stats.Managed))) } // Only patch the specific fields we want to update, not the entire status - if err := r.patchStatus(ctx, cfg, patchOperations...); err != nil { + if err := r.patchStatus(finalCtx, cfg, patchOperations...); err != nil { + finalSpan.End() logger.Error("failed to update the repository status at the end of the sync job", "error", err) - return fmt.Errorf("update repo with job final status: %w", err) + err = fmt.Errorf("update repo with job final status: %w", err) + return tracing.Error(span, err) } + finalSpan.End() return syncError } diff --git a/pkg/registry/apis/provisioning/jobs/sync/worker_test.go b/pkg/registry/apis/provisioning/jobs/sync/worker_test.go index 87ea058735d..090d4deac87 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/worker_test.go +++ b/pkg/registry/apis/provisioning/jobs/sync/worker_test.go @@ -7,6 +7,7 @@ import ( provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" "github.com/grafana/grafana/apps/provisioning/pkg/repository" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/registry/apis/provisioning/jobs" "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" "github.com/grafana/grafana/pkg/storage/legacysql/dualwrite" @@ -45,7 +46,7 @@ func TestSyncWorker_IsSupported(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - worker := NewSyncWorker(nil, nil, nil, nil, nil, metrics) + worker := NewSyncWorker(nil, nil, nil, nil, nil, metrics, tracing.NewNoopTracerService()) result := worker.IsSupported(context.Background(), tt.job) require.Equal(t, tt.expected, result) }) @@ -64,7 +65,7 @@ func TestSyncWorker_ProcessNotReaderWriter(t *testing.T) { }) fakeDualwrite := dualwrite.NewMockService(t) fakeDualwrite.On("ReadFromUnified", mock.Anything, mock.Anything).Return(true, nil).Twice() - worker := NewSyncWorker(nil, nil, fakeDualwrite, nil, nil, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) + worker := NewSyncWorker(nil, nil, fakeDualwrite, nil, nil, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()), tracing.NewNoopTracerService()) err := worker.Process(context.Background(), repo, provisioning.Job{}, jobs.NewMockJobProgressRecorder(t)) require.EqualError(t, err, "sync job submitted for repository that does not support read-write") } @@ -533,6 +534,7 @@ func TestSyncWorker_Process(t *testing.T) { repositoryPatchFn.Execute, syncer, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()), + tracing.NewNoopTracerService(), ) // Create test job diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 1904c417071..920c8680b21 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -748,7 +748,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH metrics, ) - syncer := sync.NewSyncer(sync.Compare, sync.FullSync, sync.IncrementalSync) + syncer := sync.NewSyncer(sync.Compare, sync.FullSync, sync.IncrementalSync, b.tracer) syncWorker := sync.NewSyncWorker( b.clients, b.repositoryResources, @@ -756,6 +756,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH b.statusPatcher.Patch, syncer, metrics, + b.tracer, ) signerFactory := signature.NewSignerFactory(b.clients) legacyResources := migrate.NewLegacyResourcesMigrator( @@ -847,6 +848,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH b.GetHealthChecker(), b.statusPatcher, b.registry, + b.tracer, ) if err != nil { return err