From 163a88056e29efa2ca2e466f36a04cd3da07c13f Mon Sep 17 00:00:00 2001 From: Costa Alexoglou Date: Thu, 16 Oct 2025 19:17:04 +0200 Subject: [PATCH] feat: parallel processChange (#112198) * feat: parallel processChange * all except move work * fix: tests and order of operations * fix: tests * chore: review feedback * chore: review feedback --- .../apis/provisioning/jobs/sync/full.go | 301 +++++++++++++----- .../jobs/sync/full_sync_fn_mock.go | 2 +- .../apis/provisioning/jobs/sync/full_test.go | 41 +-- .../apis/provisioning/jobs/sync/sync.go | 8 +- .../apis/provisioning/jobs/sync/sync_test.go | 1 + .../apis/provisioning/jobs/sync/worker.go | 4 + .../provisioning/jobs/sync/worker_test.go | 5 +- pkg/registry/apis/provisioning/register.go | 3 +- .../apis/provisioning/resources/parser.go | 57 +++- .../apis/provisioning/resources/resources.go | 53 ++- .../apis/provisioning/resources/tree.go | 32 +- 11 files changed, 381 insertions(+), 126 deletions(-) diff --git a/pkg/registry/apis/provisioning/jobs/sync/full.go b/pkg/registry/apis/provisioning/jobs/sync/full.go index 075f2fbac5f..0c5d9c5bb2d 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/full.go +++ b/pkg/registry/apis/provisioning/jobs/sync/full.go @@ -3,6 +3,7 @@ package sync import ( "context" "fmt" + "sync" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" @@ -25,6 +26,7 @@ func FullSync( repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer, + maxSyncWorkers int, ) error { cfg := repo.Config() @@ -59,10 +61,100 @@ func FullSync( return nil } - return applyChanges(ctx, changes, clients, repositoryResources, progress, tracer) + return applyChanges(ctx, changes, clients, repositoryResources, progress, tracer, maxSyncWorkers) } -func applyChanges(ctx context.Context, changes []ResourceFileChange, clients resources.ResourceClients, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) error { +func applyChange(ctx context.Context, change ResourceFileChange, clients resources.ResourceClients, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) { + if ctx.Err() != nil { + return + } + + 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, + } + + if change.Existing == nil || change.Existing.Name == "" { + result.Error = fmt.Errorf("processing deletion for file %s: missing existing reference", change.Path) + progress.Record(deleteCtx, result) + deleteSpan.RecordError(result.Error) + deleteSpan.End() + return + } + result.Name = change.Existing.Name + result.Group = change.Existing.Group + + versionlessGVR := schema.GroupVersionResource{ + Group: change.Existing.Group, + Resource: change.Existing.Resource, + } + + // TODO: should we use the clients or the resource manager instead? + client, gvk, err := clients.ForResource(deleteCtx, versionlessGVR) + if err != nil { + result.Kind = versionlessGVR.Resource // could not find a kind + result.Error = fmt.Errorf("get client for deleted object: %w", err) + progress.Record(deleteCtx, result) + deleteSpan.End() + return + } + result.Kind = gvk.Kind + + 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, gvk.Kind, change.Existing.Name, err) + } + progress.Record(deleteCtx, result) + deleteSpan.End() + return + } + + // Handle folders based on action type + if safepath.IsDir(change.Path) { + // For non-deletions, ensure folder exists + ensureFolderCtx, ensureFolderSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes.ensure_folder_exists") + result := jobs.JobResourceResult{ + Path: change.Path, + Action: change.Action, + Group: resources.FolderKind.Group, + Kind: resources.FolderKind.Kind, + } + + 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) + return + } + + result.Name = folder + progress.Record(ensureFolderCtx, result) + ensureFolderSpan.End() + return + } + + 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, + Name: name, + Group: gvk.Group, + Kind: gvk.Kind, + } + + if err != nil { + writeSpan.RecordError(err) + result.Error = fmt.Errorf("writing resource from file %s: %w", change.Path, err) + } + progress.Record(writeCtx, result) + writeSpan.End() +} + +func applyChanges(ctx context.Context, changes []ResourceFileChange, clients resources.ResourceClients, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer, maxSyncWorkers int) error { progress.SetTotal(ctx, len(changes)) _, applyChangesSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes", @@ -70,98 +162,139 @@ func applyChanges(ctx context.Context, changes []ResourceFileChange, clients res ) defer applyChangesSpan.End() + // Separate changes into four categories for proper ordering: + // 1. File deletions (must happen before folder deletions) + // 2. Folder deletions + // 3. Folder creations (must happen before file creations) + // 4. File creations (must happen after folder creations) + var fileDeletions []ResourceFileChange + var folderDeletions []ResourceFileChange + var folderCreations []ResourceFileChange + var fileCreations []ResourceFileChange + for _, change := range changes { - if ctx.Err() != nil { - return ctx.Err() + isFolder := safepath.IsDir(change.Path) + isDeleted := change.Action == repository.FileActionDeleted + + if isDeleted { + if isFolder { + folderDeletions = append(folderDeletions, change) + } else { + fileDeletions = append(fileDeletions, change) + } + } else { + if isFolder { + folderCreations = append(folderCreations, change) + } else { + fileCreations = append(fileCreations, change) + } } + } - if err := progress.TooManyErrors(); err != nil { - return tracing.Error(applyChangesSpan, err) + applyChangesSpan.SetAttributes( + attribute.Int("file_deletions", len(fileDeletions)), + attribute.Int("folder_deletions", len(folderDeletions)), + attribute.Int("folder_creations", len(folderCreations)), + attribute.Int("file_creations", len(fileCreations)), + ) + + if len(fileDeletions) > 0 { + if err := applyResourcesInParallel(ctx, fileDeletions, clients, repositoryResources, progress, tracer, maxSyncWorkers); err != nil { + return 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, - } - - if change.Existing == nil || change.Existing.Name == "" { - result.Error = fmt.Errorf("processing deletion for file %s: missing existing reference", change.Path) - progress.Record(deleteCtx, result) - deleteSpan.RecordError(result.Error) - deleteSpan.End() - continue - } - result.Name = change.Existing.Name - result.Group = change.Existing.Group - - versionlessGVR := schema.GroupVersionResource{ - Group: change.Existing.Group, - Resource: change.Existing.Resource, - } - - // TODO: should we use the clients or the resource manager instead? - client, gvk, err := clients.ForResource(deleteCtx, versionlessGVR) - if err != nil { - result.Kind = versionlessGVR.Resource // could not find a kind - result.Error = fmt.Errorf("get client for deleted object: %w", err) - progress.Record(deleteCtx, result) - continue - } - result.Kind = gvk.Kind - - 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, gvk.Kind, change.Existing.Name, err) - } - progress.Record(deleteCtx, result) - deleteSpan.End() - continue + if len(folderDeletions) > 0 { + if err := applyFoldersSerially(ctx, folderDeletions, clients, repositoryResources, progress, tracer); err != nil { + return err } + } - // 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, - Group: resources.FolderKind.Group, - Kind: resources.FolderKind.Kind, - } - - 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(ensureFolderCtx, result) - ensureFolderSpan.End() - - continue + if len(folderCreations) > 0 { + if err := applyFoldersSerially(ctx, folderCreations, clients, repositoryResources, progress, tracer); err != nil { + return err } + } - 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, - Name: name, - Group: gvk.Group, - Kind: gvk.Kind, - } - - if err != nil { - writeSpan.RecordError(err) - result.Error = fmt.Errorf("writing resource from file %s: %w", change.Path, err) - } - progress.Record(writeCtx, result) - writeSpan.End() + if len(fileCreations) > 0 { + return applyResourcesInParallel(ctx, fileCreations, clients, repositoryResources, progress, tracer, maxSyncWorkers) } return nil } + +func applyFoldersSerially(ctx context.Context, folders []ResourceFileChange, clients resources.ResourceClients, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) error { + folderCtx, folderCancel := context.WithCancel(ctx) + defer folderCancel() + + for _, folder := range folders { + if folderCtx.Err() != nil { + return folderCtx.Err() + } + + if err := progress.TooManyErrors(); err != nil { + return err + } + + applyChange(folderCtx, folder, clients, repositoryResources, progress, tracer) + } + + return nil +} + +func applyResourcesInParallel(ctx context.Context, resources []ResourceFileChange, clients resources.ResourceClients, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer, maxSyncWorkers int) error { + if len(resources) == 0 { + return nil + } + + workerCtx, cancel := context.WithCancel(ctx) + defer cancel() + + changeChan := make(chan ResourceFileChange, len(resources)) + var wg sync.WaitGroup + + for i := 0; i < maxSyncWorkers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case change, ok := <-changeChan: + if !ok { + return + } + + if err := progress.TooManyErrors(); err != nil { + cancel() + return + } + if workerCtx.Err() != nil { + return + } + + applyChange(workerCtx, change, clients, repositoryResources, progress, tracer) + + case <-workerCtx.Done(): + return + } + } + }() + } + + for _, change := range resources { + select { + case changeChan <- change: + case <-workerCtx.Done(): + goto done + } + } +done: + close(changeChan) + wg.Wait() + + if err := progress.TooManyErrors(); err != nil { + return err + } + + return ctx.Err() +} 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 3f10550e3f8..c3e0a68a8c6 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 @@ -29,7 +29,7 @@ func (_m *MockFullSyncFn) EXPECT() *MockFullSyncFn_Expecter { } // 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 { +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, maxSyncWorkers int) error { ret := _m.Called(ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer) if len(ret) == 0 { diff --git a/pkg/registry/apis/provisioning/jobs/sync/full_test.go b/pkg/registry/apis/provisioning/jobs/sync/full_test.go index 90e5f4fb0c1..76cf514f23d 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/full_test.go +++ b/pkg/registry/apis/provisioning/jobs/sync/full_test.go @@ -43,7 +43,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, tracing.NewNoopTracerService()) + err := FullSync(ctx, repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10) require.EqualError(t, err, "context canceled") } @@ -62,7 +62,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, tracing.NewNoopTracerService()) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10) require.EqualError(t, err, "compare changes: some error") } @@ -82,7 +82,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, tracing.NewNoopTracerService()) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10) require.NoError(t, err) } @@ -113,7 +113,7 @@ func TestFullSync_SuccessfulFolderCreation(t *testing.T) { Path: "", }, "").Return(nil) - err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService()) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10) require.NoError(t, err) } @@ -142,7 +142,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, tracing.NewNoopTracerService()) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10) require.Error(t, err) require.Contains(t, err.Error(), "create root folder: folder creation failed") } @@ -171,7 +171,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, tracing.NewNoopTracerService()) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10) require.Error(t, err) require.Contains(t, err.Error(), "compare changes: compare error") } @@ -202,20 +202,23 @@ func TestFullSync_ApplyChanges(t *testing.T) { //nolint:gocyclo }, }, setupMocks: func(repo *repository.MockRepository, repoResources *resources.MockRepositoryResources, clients *resources.MockResourceClients, progress *jobs.MockJobProgressRecorder, compareFn *MockCompareFn) { - // First call returns nil, second call returns error - progress.On("TooManyErrors").Return(nil).Once() - progress.On("TooManyErrors").Return(fmt.Errorf("too many errors")).Once() + callCount := 0 + progress.On("TooManyErrors").Return(func() error { + callCount++ + if callCount > 1 { + return fmt.Errorf("too many errors") + } + return nil + }) - repoResources.On("WriteResourceFromFile", mock.Anything, "dashboards/one.json", ""). - Return("test-dashboard", schema.GroupVersionKind{Kind: "Dashboard", Group: "dashboards"}, nil) + repoResources.On("WriteResourceFromFile", mock.Anything, mock.MatchedBy(func(path string) bool { + return path == "dashboards/one.json" || path == "dashboards/two.json" || path == "dashboards/three.json" + }), "").Return("test-dashboard", schema.GroupVersionKind{Kind: "Dashboard", Group: "dashboards"}, nil).Maybe() - progress.On("Record", mock.Anything, jobs.JobResourceResult{ - Action: repository.FileActionCreated, - Path: "dashboards/one.json", - Name: "test-dashboard", - Kind: "Dashboard", - Group: "dashboards", - }).Return() + progress.On("Record", mock.Anything, mock.MatchedBy(func(result jobs.JobResourceResult) bool { + return result.Action == repository.FileActionCreated && + (result.Path == "dashboards/one.json" || result.Path == "dashboards/two.json" || result.Path == "dashboards/three.json") + })).Return().Maybe() }, expectedError: "too many errors", }, @@ -701,7 +704,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, tracing.NewNoopTracerService()) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10) if tt.expectedError != "" { require.EqualError(t, err, tt.expectedError, tt.description) } else { diff --git a/pkg/registry/apis/provisioning/jobs/sync/sync.go b/pkg/registry/apis/provisioning/jobs/sync/sync.go index fa58d426cba..55272332add 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/sync.go +++ b/pkg/registry/apis/provisioning/jobs/sync/sync.go @@ -12,7 +12,7 @@ import ( ) //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, tracer tracing.Tracer) 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, maxSyncWorkers int) 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) @@ -30,14 +30,16 @@ type syncer struct { fullSync FullSyncFn incrementalSync IncrementalSyncFn tracer tracing.Tracer + maxSyncWorkers int } -func NewSyncer(compare CompareFn, fullSync FullSyncFn, incrementalSync IncrementalSyncFn, tracer tracing.Tracer) Syncer { +func NewSyncer(compare CompareFn, fullSync FullSyncFn, incrementalSync IncrementalSyncFn, tracer tracing.Tracer, maxSyncWorkers int) Syncer { return &syncer{ compare: compare, fullSync: fullSync, incrementalSync: incrementalSync, tracer: tracer, + maxSyncWorkers: maxSyncWorkers, } } @@ -61,5 +63,5 @@ func (r *syncer) Sync(ctx context.Context, repo repository.ReaderWriter, options progress.SetMessage(ctx, "full sync") - return currentRef, r.fullSync(ctx, repo, r.compare, clients, currentRef, repositoryResources, progress, r.tracer) + return currentRef, r.fullSync(ctx, repo, r.compare, clients, currentRef, repositoryResources, progress, r.tracer, r.maxSyncWorkers) } diff --git a/pkg/registry/apis/provisioning/jobs/sync/sync_test.go b/pkg/registry/apis/provisioning/jobs/sync/sync_test.go index 5f04ed350e1..e0d1bbee7e1 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/sync_test.go +++ b/pkg/registry/apis/provisioning/jobs/sync/sync_test.go @@ -160,6 +160,7 @@ func TestSyncer_Sync(t *testing.T) { fullSyncFn.Execute, incrementalSyncFn.Execute, tracing.NewNoopTracerService(), + 10, ) 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 c5df4553e04..dfb376ca470 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/worker.go +++ b/pkg/registry/apis/provisioning/jobs/sync/worker.go @@ -41,6 +41,8 @@ type SyncWorker struct { metrics jobs.JobMetrics tracer tracing.Tracer + + maxSyncWorkers int } func NewSyncWorker( @@ -51,6 +53,7 @@ func NewSyncWorker( syncer Syncer, metrics jobs.JobMetrics, tracer tracing.Tracer, + maxSyncWorkers int, ) *SyncWorker { return &SyncWorker{ clients: clients, @@ -60,6 +63,7 @@ func NewSyncWorker( syncer: syncer, metrics: metrics, tracer: tracer, + maxSyncWorkers: maxSyncWorkers, } } diff --git a/pkg/registry/apis/provisioning/jobs/sync/worker_test.go b/pkg/registry/apis/provisioning/jobs/sync/worker_test.go index bd8b094b4f6..bf0b60bea01 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/worker_test.go +++ b/pkg/registry/apis/provisioning/jobs/sync/worker_test.go @@ -46,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, tracing.NewNoopTracerService()) + worker := NewSyncWorker(nil, nil, nil, nil, nil, metrics, tracing.NewNoopTracerService(), 10) result := worker.IsSupported(context.Background(), tt.job) require.Equal(t, tt.expected, result) }) @@ -65,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()), tracing.NewNoopTracerService()) + worker := NewSyncWorker(nil, nil, fakeDualwrite, nil, nil, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()), tracing.NewNoopTracerService(), 10) 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") } @@ -545,6 +545,7 @@ func TestSyncWorker_Process(t *testing.T) { syncer, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()), tracing.NewNoopTracerService(), + 10, ) // Create test job diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 02a84545988..4382302fc18 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -696,7 +696,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH metrics, ) - syncer := sync.NewSyncer(sync.Compare, sync.FullSync, sync.IncrementalSync, b.tracer) + syncer := sync.NewSyncer(sync.Compare, sync.FullSync, sync.IncrementalSync, b.tracer, 10) syncWorker := sync.NewSyncWorker( b.clients, b.repositoryResources, @@ -705,6 +705,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH syncer, metrics, b.tracer, + 10, ) signerFactory := signature.NewSignerFactory(b.clients) legacyResources := migrate.NewLegacyResourcesMigrator( diff --git a/pkg/registry/apis/provisioning/resources/parser.go b/pkg/registry/apis/provisioning/resources/parser.go index 44e48041bd5..c8c65baf656 100644 --- a/pkg/registry/apis/provisioning/resources/parser.go +++ b/pkg/registry/apis/provisioning/resources/parser.go @@ -7,6 +7,7 @@ import ( "fmt" "path" + "go.opentelemetry.io/otel/attribute" "gopkg.in/yaml.v3" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -22,6 +23,7 @@ import ( "github.com/grafana/grafana/pkg/apimachinery/apis/common/v0alpha1" "github.com/grafana/grafana/pkg/apimachinery/identity" "github.com/grafana/grafana/pkg/apimachinery/utils" + "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/util" ) @@ -302,10 +304,15 @@ func (f *ParsedResource) Run(ctx context.Context) error { } // Always use the provisioning identity when writing - ctx, _, err := identity.WithProvisioningIdentity(ctx, f.Obj.GetNamespace()) + identityCtx, _, err := identity.WithProvisioningIdentity(ctx, f.Obj.GetNamespace()) + ctx, identitySpan := tracing.Start(identityCtx, "provisioning.resources.run_resource.set_identity") + if err != nil { + identitySpan.RecordError(err) + identitySpan.End() return err } + identitySpan.End() fieldValidation := "Strict" if f.GVR == DashboardResource { @@ -318,42 +325,57 @@ func (f *ParsedResource) Run(ctx context.Context) error { Identity: f.Repo.Name, } + actionsCtx, actionsSpan := tracing.Start(ctx, "provisioning.resources.run_resource.actions") + defer actionsSpan.End() + // Handle deletion action if f.Action == provisioning.ResourceActionDelete { + deleteCtx, deleteSpan := tracing.Start(actionsCtx, "provisioning.resources.run_resource.delete") + deleteSpan.SetAttributes(attribute.String("resource.name", f.Obj.GetName())) + // If we don't have existing resource from DryRun, fetch it now if f.DryRunResponse == nil { - f.Existing, err = f.Client.Get(ctx, f.Obj.GetName(), metav1.GetOptions{}) + f.Existing, err = f.Client.Get(deleteCtx, f.Obj.GetName(), metav1.GetOptions{}) if err != nil { + deleteSpan.RecordError(err) if apierrors.IsNotFound(err) { // Resource doesn't exist, nothing to delete - this is fine + deleteSpan.End() return nil } + deleteSpan.End() return fmt.Errorf("failed to get existing resource for delete: %w", err) } } // Check ownership with the existing resource if err := CheckResourceOwnership(f.Existing, f.Obj.GetName(), requestingManager); err != nil { + deleteSpan.RecordError(err) + deleteSpan.End() return err } // Perform the actual delete - err = f.Client.Delete(ctx, f.Obj.GetName(), metav1.DeleteOptions{}) + err = f.Client.Delete(deleteCtx, f.Obj.GetName(), metav1.DeleteOptions{}) if apierrors.IsNotFound(err) { err = nil // ignorable - resource was already deleted } + if err != nil { + deleteSpan.RecordError(err) + } // Set the deleted resource as the result if err == nil && f.Existing != nil { f.Upsert = f.Existing.DeepCopy() } + deleteSpan.End() return err } // If we don't have existing resource from DryRun, fetch it now if f.DryRunResponse == nil { - f.Existing, _ = f.Client.Get(ctx, f.Obj.GetName(), metav1.GetOptions{}) + f.Existing, _ = f.Client.Get(actionsCtx, f.Obj.GetName(), metav1.GetOptions{}) } // Check ownership with the existing resource (if any) @@ -364,9 +386,16 @@ func (f *ParsedResource) Run(ctx context.Context) error { // If we have already tried loading existing, start with create if f.DryRunResponse != nil && f.Existing == nil { f.Action = provisioning.ResourceActionCreate - f.Upsert, err = f.Client.Create(ctx, f.Obj, metav1.CreateOptions{ + createCtx, createSpan := tracing.Start(actionsCtx, "provisioning.resources.run_resource.create") + createSpan.SetAttributes(attribute.String("resource.name", f.Obj.GetName())) + f.Upsert, err = f.Client.Create(createCtx, f.Obj, metav1.CreateOptions{ FieldValidation: fieldValidation, }) + if err != nil { + createSpan.RecordError(err) + } + createSpan.End() + if err == nil { return nil // it worked, return } @@ -374,14 +403,28 @@ func (f *ParsedResource) Run(ctx context.Context) error { // Try update, otherwise create f.Action = provisioning.ResourceActionUpdate - f.Upsert, err = f.Client.Update(ctx, f.Obj, metav1.UpdateOptions{ + + updateCtx, updateSpan := tracing.Start(actionsCtx, "provisioning.resources.run_resource.update") + updateSpan.SetAttributes(attribute.String("resource.name", f.Obj.GetName())) + f.Upsert, err = f.Client.Update(updateCtx, f.Obj, metav1.UpdateOptions{ FieldValidation: fieldValidation, }) + if err != nil { + updateSpan.RecordError(err) + } + updateSpan.End() + if apierrors.IsNotFound(err) { f.Action = provisioning.ResourceActionCreate - f.Upsert, err = f.Client.Create(ctx, f.Obj, metav1.CreateOptions{ + fallbackCreateCtx, fallbackCreateSpan := tracing.Start(actionsCtx, "provisioning.resources.run_resource.create_fallback") + fallbackCreateSpan.SetAttributes(attribute.String("resource.name", f.Obj.GetName())) + f.Upsert, err = f.Client.Create(fallbackCreateCtx, f.Obj, metav1.CreateOptions{ FieldValidation: fieldValidation, }) + if err != nil { + fallbackCreateSpan.RecordError(err) + } + fallbackCreateSpan.End() } return err } diff --git a/pkg/registry/apis/provisioning/resources/resources.go b/pkg/registry/apis/provisioning/resources/resources.go index b66e16917da..4954e6b51b6 100644 --- a/pkg/registry/apis/provisioning/resources/resources.go +++ b/pkg/registry/apis/provisioning/resources/resources.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "slices" + "sync" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -17,6 +18,7 @@ import ( "github.com/grafana/grafana/apps/provisioning/pkg/safepath" "github.com/grafana/grafana/pkg/apimachinery/utils" "github.com/grafana/grafana/pkg/infra/slugify" + "github.com/grafana/grafana/pkg/infra/tracing" ) var ( @@ -55,6 +57,7 @@ type ResourcesManager struct { parser Parser clients ResourceClients resourcesLookup map[resourceID]string // the path with this k8s name + mu sync.RWMutex } func NewResourcesManager(repo repository.ReaderWriter, folders *FolderManager, parser Parser, clients ResourceClients) *ResourcesManager { @@ -67,6 +70,25 @@ func NewResourcesManager(repo repository.ReaderWriter, folders *FolderManager, p } } +// findResource checks if a resource exists in the lookup map (read operation) +func (r *ResourcesManager) findResource(id resourceID) (string, bool) { + r.mu.RLock() + defer r.mu.RUnlock() + path, found := r.resourcesLookup[id] + return path, found +} + +func (r *ResourcesManager) addResource(id resourceID, path string) { + r.mu.Lock() + defer r.mu.Unlock() + + if _, found := r.resourcesLookup[id]; found { + return + } + + r.resourcesLookup[id] = path +} + // CheckResourceOwnership validates that the requesting manager can modify the existing resource // Returns an error if the existing resource is owned by a different manager that doesn't allow edits // If existingResource is nil, no ownership conflict exists (new resource) @@ -193,15 +215,23 @@ func (r *ResourcesManager) WriteResourceFileFromObject(ctx context.Context, obj func (r *ResourcesManager) WriteResourceFromFile(ctx context.Context, path string, ref string) (string, schema.GroupVersionKind, error) { // Read the referenced file - fileInfo, err := r.repo.Read(ctx, path, ref) + readCtx, readSpan := tracing.Start(ctx, "provisioning.resources.write_resource_from_file.read_file") + fileInfo, err := r.repo.Read(readCtx, path, ref) if err != nil { + readSpan.RecordError(err) + readSpan.End() return "", schema.GroupVersionKind{}, fmt.Errorf("failed to read file: %w", err) } + readSpan.End() - parsed, err := r.parser.Parse(ctx, fileInfo) + parseCtx, parseSpan := tracing.Start(ctx, "provisioning.resources.write_resource_from_file.parse_file") + parsed, err := r.parser.Parse(parseCtx, fileInfo) if err != nil { + parseSpan.RecordError(err) + parseSpan.End() return "", schema.GroupVersionKind{}, fmt.Errorf("failed to parse file: %w", err) } + parseSpan.End() if parsed.Obj.GetName() == "" { return "", schema.GroupVersionKind{}, ErrMissingName @@ -213,27 +243,36 @@ func (r *ResourcesManager) WriteResourceFromFile(ctx context.Context, path strin Resource: parsed.GVR.Resource, Group: parsed.GVK.Group, } - existing, found := r.resourcesLookup[id] - if found { + + if existing, found := r.findResource(id); found { return "", parsed.GVK, fmt.Errorf("duplicate resource name: %s, %s and %s: %w", parsed.Obj.GetName(), path, existing, ErrDuplicateName) } - r.resourcesLookup[id] = path + r.addResource(id, path) // For resources that exist in folders, set the header annotation if slices.Contains(SupportsFolderAnnotation, parsed.GVR.GroupResource()) { // Make sure the parent folders exist - folder, err := r.folders.EnsureFolderPathExist(ctx, path) + folderCtx, folderSpan := tracing.Start(ctx, "provisioning.resources.write_resource_from_file.ensure_folder") + folder, err := r.folders.EnsureFolderPathExist(folderCtx, path) if err != nil { + folderSpan.RecordError(err) + folderSpan.End() return "", parsed.GVK, fmt.Errorf("failed to ensure folder path exists: %w", err) } parsed.Meta.SetFolder(folder) + folderSpan.End() } // Clear any saved identifiers parsed.Meta.SetUID("") parsed.Meta.SetResourceVersion("") - err = parsed.Run(ctx) + runCtx, runSpan := tracing.Start(ctx, "provisioning.resources.write_resource_from_file.run_resource") + err = parsed.Run(runCtx) + if err != nil { + runSpan.RecordError(err) + } + runSpan.End() return parsed.Obj.GetName(), parsed.GVK, err } diff --git a/pkg/registry/apis/provisioning/resources/tree.go b/pkg/registry/apis/provisioning/resources/tree.go index 77d97573a5f..0fc25fd15ca 100644 --- a/pkg/registry/apis/provisioning/resources/tree.go +++ b/pkg/registry/apis/provisioning/resources/tree.go @@ -3,6 +3,7 @@ package resources import ( "context" "fmt" + "sync" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" @@ -30,11 +31,18 @@ type folderTree struct { tree map[string]string folders map[string]Folder count int + mu sync.RWMutex } // In determines if the given folder is in the tree at all. That is, it answers "does the folder even exist in the Grafana instance?" // An empty folder string means the root folder, and is special-cased to always return true. func (t *folderTree) In(folder string) bool { + t.mu.RLock() + defer t.mu.RUnlock() + return t.in(folder) +} + +func (t *folderTree) in(folder string) bool { _, ok := t.tree[folder] return ok || folder == "" } @@ -47,7 +55,19 @@ func (t *folderTree) In(folder string) bool { // If In(folder) or In(baseFolder) is false, this will return ok=false, because it would be undefined behaviour. // If baseFolder is not a parent of folder, ok=false is returned. func (t *folderTree) DirPath(folder, baseFolder string) (fid Folder, ok bool) { - if !t.In(folder) || !t.In(baseFolder) { + t.mu.RLock() + defer t.mu.RUnlock() + return t.dirPath(folder, baseFolder) +} + +// dirPath is the internal implementation that assumes the mutex is already held +// Needed to avoid deadlock when called from other methods that hold locks like Walk() +func (t *folderTree) dirPath(folder, baseFolder string) (fid Folder, ok bool) { + // Inline In() logic to avoid deadlock when called from other methods that hold locks + folderInTree := t.in(folder) + baseFolderInTree := t.in(baseFolder) + + if !folderInTree || !baseFolderInTree { return Folder{}, false } if folder == "" && baseFolder != "" { @@ -76,21 +96,27 @@ func (t *folderTree) DirPath(folder, baseFolder string) (fid Folder, ok bool) { } func (t *folderTree) Add(folder Folder, parent string) { + t.mu.Lock() + defer t.mu.Unlock() t.tree[folder.ID] = parent t.folders[folder.ID] = folder t.count++ } func (t *folderTree) Count() int { + t.mu.RLock() + defer t.mu.RUnlock() return t.count } type WalkFunc func(ctx context.Context, folder Folder, parent string) error func (t *folderTree) Walk(ctx context.Context, fn WalkFunc) error { + t.mu.RLock() + defer t.mu.RUnlock() toWalk := make([]Folder, 0, len(t.folders)) for _, folder := range t.folders { - folder, _ := t.DirPath(folder.ID, "") + folder, _ := t.dirPath(folder.ID, "") toWalk = append(toWalk, folder) } @@ -123,6 +149,8 @@ func (t *folderTree) AddUnstructured(item *unstructured.Unstructured) error { Title: meta.FindTitle(item.GetName()), ID: item.GetName(), } + t.mu.Lock() + defer t.mu.Unlock() t.tree[folder.ID] = meta.GetFolder() t.folders[folder.ID] = folder t.count++