diff --git a/pkg/registry/apis/provisioning/jobs/metrics.go b/pkg/registry/apis/provisioning/jobs/metrics.go index dc946838659..01067feccb3 100644 --- a/pkg/registry/apis/provisioning/jobs/metrics.go +++ b/pkg/registry/apis/provisioning/jobs/metrics.go @@ -1,6 +1,8 @@ package jobs import ( + "time" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/utils" "github.com/prometheus/client_golang/prometheus" ) @@ -9,6 +11,10 @@ type JobMetrics struct { registry prometheus.Registerer processedTotal *prometheus.CounterVec durationHist *prometheus.HistogramVec + + incrementalSyncPhaseDurationHist *prometheus.HistogramVec // phases of incremental sync + fullSyncPhaseDurationHist *prometheus.HistogramVec // phases of full sync + syncDurationHist *prometheus.HistogramVec // total sync durations } type QueueMetrics struct { @@ -72,12 +78,44 @@ func RegisterJobMetrics(registry prometheus.Registerer) JobMetrics { }, []string{"action", "resources_changed_bucket"}, ) - registry.MustRegister(durationHist) + + incrementalSyncPhaseDurationHist := prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "grafana_provisioning_jobs_incremental_sync_phase_duration_seconds", + Help: "Duration of job phases for incremental sync", + Buckets: prometheus.ExponentialBucketsRange(0.01, 10*60, 8), // 1ms -> 10m + }, + []string{"phase"}, + ) + registry.MustRegister(incrementalSyncPhaseDurationHist) + + fullSyncPhaseDurationHist := prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "grafana_provisioning_jobs_full_sync_phase_duration_seconds", + Help: "Duration of job phases for full sync", + Buckets: prometheus.ExponentialBucketsRange(0.01, 10*60, 8), // 1ms -> 10m + }, + []string{"phase"}, + ) + registry.MustRegister(fullSyncPhaseDurationHist) + + syncDurationHist := prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "grafana_provisioning_jobs_sync_duration_seconds", + Help: "Duration of sync (full or incremental)", + Buckets: prometheus.ExponentialBucketsRange(0.01, 10*60, 8), // 1ms -> 10m + }, + []string{"type"}, + ) + registry.MustRegister(syncDurationHist) return JobMetrics{ - registry: registry, - processedTotal: processedTotal, - durationHist: durationHist, + registry: registry, + processedTotal: processedTotal, + durationHist: durationHist, + incrementalSyncPhaseDurationHist: incrementalSyncPhaseDurationHist, + fullSyncPhaseDurationHist: fullSyncPhaseDurationHist, + syncDurationHist: syncDurationHist, } } @@ -90,6 +128,18 @@ func (m *JobMetrics) RecordJob(jobAction string, outcome string, resourceCountCh } } +func (m *JobMetrics) RecordIncrementalSyncPhase(phase IncrementalSyncPhase, duration time.Duration) { + m.incrementalSyncPhaseDurationHist.WithLabelValues(phase.String()).Observe(duration.Seconds()) +} + +func (m *JobMetrics) RecordFullSyncPhase(phase FullSyncPhase, duration time.Duration) { + m.fullSyncPhaseDurationHist.WithLabelValues(phase.String()).Observe(duration.Seconds()) +} + +func (m *JobMetrics) RecordSyncDuration(syncType SyncType, duration time.Duration) { + m.syncDurationHist.WithLabelValues(syncType.String()).Observe(duration.Seconds()) +} + func recordConcurrentDriverMetric(registry prometheus.Registerer, numDrivers int) { concurrentDriver := prometheus.NewGaugeVec( prometheus.GaugeOpts{ @@ -101,3 +151,72 @@ func recordConcurrentDriverMetric(registry prometheus.Registerer, numDrivers int registry.MustRegister(concurrentDriver) concurrentDriver.WithLabelValues().Set(float64(numDrivers)) } + +type SyncType int + +const ( + SyncTypeUnknown SyncType = iota // to prevent zero value being valid + SyncTypeFull + SyncTypeIncremental +) + +func (t SyncType) String() string { + switch t { + case SyncTypeFull: + return "full" + case SyncTypeIncremental: + return "incremental" + default: + return "unknown" + } +} + +type FullSyncPhase int + +const ( + FullSyncPhaseUnknown FullSyncPhase = iota // to prevent zero value being valid + FullSyncPhaseCompare + FullSyncPhaseFileDeletions + FullSyncPhaseFolderDeletions + FullSyncPhaseFolderCreations + FullSyncPhaseFileCreations +) + +func (p FullSyncPhase) String() string { + switch p { + case FullSyncPhaseCompare: + return "compare" + case FullSyncPhaseFileDeletions: + return "file_deletions" + case FullSyncPhaseFolderDeletions: + return "folder_deletions" + case FullSyncPhaseFolderCreations: + return "folder_creations" + case FullSyncPhaseFileCreations: + return "file_creations" + default: + return "unknown" + } +} + +type IncrementalSyncPhase int + +const ( + IncrementalSyncPhaseUnknown IncrementalSyncPhase = iota // to prevent zero value being valid + IncrementalSyncPhaseCompare + IncrementalSyncPhaseApply + IncrementalSyncPhaseCleanup +) + +func (p IncrementalSyncPhase) String() string { + switch p { + case IncrementalSyncPhaseCompare: + return "compare" + case IncrementalSyncPhaseApply: + return "apply" + case IncrementalSyncPhaseCleanup: + return "cleanup" + default: + return "unknown" + } +} diff --git a/pkg/registry/apis/provisioning/jobs/sync/full.go b/pkg/registry/apis/provisioning/jobs/sync/full.go index 14d53c00afc..10aad46693b 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/full.go +++ b/pkg/registry/apis/provisioning/jobs/sync/full.go @@ -29,11 +29,16 @@ func FullSync( progress jobs.JobProgressRecorder, tracer tracing.Tracer, maxSyncWorkers int, + metrics jobs.JobMetrics, ) error { + syncStart := time.Now() cfg := repo.Config() ctx, span := tracer.Start(ctx, "provisioning.sync.full") defer span.End() + defer func() { + metrics.RecordSyncDuration(jobs.SyncTypeFull, time.Since(syncStart)) + }() ensureFolderCtx, ensureFolderSpan := tracer.Start(ctx, "provisioning.sync.full.ensure_folder_exists") // Ensure the configured folder exists and is managed by the repository @@ -51,19 +56,23 @@ func FullSync( ensureFolderSpan.End() compareCtx, compareSpan := tracer.Start(ctx, "provisioning.sync.full.compare") - changes, err := compare(compareCtx, repo, repositoryResources, currentRef) + var changes []ResourceFileChange + err := instrumentedFullSyncPhase(jobs.FullSyncPhaseCompare, func() (err error) { + changes, err = compare(compareCtx, repo, repositoryResources, currentRef) + return + }, metrics) + compareSpan.End() + if err != nil { - 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, tracer, maxSyncWorkers) + return applyChanges(ctx, changes, clients, repositoryResources, progress, tracer, maxSyncWorkers, metrics) } func applyChange(ctx context.Context, change ResourceFileChange, clients resources.ResourceClients, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) { @@ -156,7 +165,15 @@ func applyChange(ctx context.Context, change ResourceFileChange, clients resourc writeSpan.End() } -func applyChanges(ctx context.Context, changes []ResourceFileChange, clients resources.ResourceClients, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer, maxSyncWorkers int) error { +// instrument a function with a phase and metrics +func instrumentedFullSyncPhase(phase jobs.FullSyncPhase, fn func() error, metrics jobs.JobMetrics) error { + phaseStart := time.Now() + err := fn() + metrics.RecordFullSyncPhase(phase, time.Since(phaseStart)) + return err +} + +func applyChanges(ctx context.Context, changes []ResourceFileChange, clients resources.ResourceClients, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer, maxSyncWorkers int, metrics jobs.JobMetrics) error { progress.SetTotal(ctx, len(changes)) _, applyChangesSpan := tracer.Start(ctx, "provisioning.sync.full.apply_changes", @@ -201,25 +218,35 @@ func applyChanges(ctx context.Context, changes []ResourceFileChange, clients res ) if len(fileDeletions) > 0 { - if err := applyResourcesInParallel(ctx, fileDeletions, clients, repositoryResources, progress, tracer, maxSyncWorkers); err != nil { + if err := instrumentedFullSyncPhase(jobs.FullSyncPhaseFileDeletions, func() error { + return applyResourcesInParallel(ctx, fileDeletions, clients, repositoryResources, progress, tracer, maxSyncWorkers) + }, metrics); err != nil { return err } } if len(folderDeletions) > 0 { - if err := applyFoldersSerially(ctx, folderDeletions, clients, repositoryResources, progress, tracer); err != nil { + if err := instrumentedFullSyncPhase(jobs.FullSyncPhaseFolderDeletions, func() error { + return applyFoldersSerially(ctx, folderDeletions, clients, repositoryResources, progress, tracer) + }, metrics); err != nil { return err } } if len(folderCreations) > 0 { - if err := applyFoldersSerially(ctx, folderCreations, clients, repositoryResources, progress, tracer); err != nil { + if err := instrumentedFullSyncPhase(jobs.FullSyncPhaseFolderCreations, func() error { + return applyFoldersSerially(ctx, folderCreations, clients, repositoryResources, progress, tracer) + }, metrics); err != nil { return err } } if len(fileCreations) > 0 { - return applyResourcesInParallel(ctx, fileCreations, clients, repositoryResources, progress, tracer, maxSyncWorkers) + if err := instrumentedFullSyncPhase(jobs.FullSyncPhaseFileCreations, func() error { + return applyResourcesInParallel(ctx, fileCreations, clients, repositoryResources, progress, tracer, maxSyncWorkers) + }, metrics); err != nil { + return err + } } 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 c3e0a68a8c6..a9b18943898 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 @@ -28,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, 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, maxSyncWorkers int) error { - ret := _m.Called(ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer) +// Execute provides a mock function with given fields: ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer, maxSyncWorkers, metrics +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, metrics jobs.JobMetrics) error { + ret := _m.Called(ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer, maxSyncWorkers, metrics) 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, tracing.Tracer) error); ok { - r0 = rf(ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer) + if rf, ok := ret.Get(0).(func(context.Context, repository.Reader, CompareFn, resources.ResourceClients, string, resources.RepositoryResources, jobs.JobProgressRecorder, tracing.Tracer, int, jobs.JobMetrics) error); ok { + r0 = rf(ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer, maxSyncWorkers, metrics) } else { r0 = ret.Error(0) } @@ -60,13 +60,15 @@ type MockFullSyncFn_Execute_Call struct { // - repositoryResources resources.RepositoryResources // - progress jobs.JobProgressRecorder // - 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)} +// - maxSyncWorkers int +// - metrics jobs.JobMetrics +func (_e *MockFullSyncFn_Expecter) Execute(ctx interface{}, repo interface{}, compare interface{}, clients interface{}, currentRef interface{}, repositoryResources interface{}, progress interface{}, tracer interface{}, maxSyncWorkers interface{}, metrics interface{}) *MockFullSyncFn_Execute_Call { + return &MockFullSyncFn_Execute_Call{Call: _e.mock.On("Execute", ctx, repo, compare, clients, currentRef, repositoryResources, progress, tracer, maxSyncWorkers, metrics)} } -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 { +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, maxSyncWorkers int, metrics jobs.JobMetrics)) *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), args[7].(tracing.Tracer)) + 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), args[8].(int), args[9].(jobs.JobMetrics)) }) return _c } @@ -76,7 +78,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, tracing.Tracer) 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, int, jobs.JobMetrics) 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 904b2645256..aaa61ee61db 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 ( "testing" "time" + "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -45,7 +46,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(), 10) + err := FullSync(ctx, repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) require.EqualError(t, err, "context canceled") } @@ -64,7 +65,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(), 10) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) require.EqualError(t, err, "compare changes: some error") } @@ -84,7 +85,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(), 10) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) require.NoError(t, err) } @@ -115,7 +116,7 @@ func TestFullSync_SuccessfulFolderCreation(t *testing.T) { Path: "", }, "").Return(nil) - err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) require.NoError(t, err) } @@ -144,7 +145,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(), 10) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) require.Error(t, err) require.Contains(t, err.Error(), "create root folder: folder creation failed") } @@ -173,7 +174,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(), 10) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) require.Error(t, err) require.Contains(t, err.Error(), "compare changes: compare error") } @@ -744,7 +745,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(), 10) + err := FullSync(context.Background(), repo, compareFn.Execute, clients, "current-ref", repoResources, progress, tracing.NewNoopTracerService(), 10, jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) 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 0c4ddde659b..daa94d94636 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/incremental.go +++ b/pkg/registry/apis/provisioning/jobs/sync/incremental.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "time" "github.com/grafana/grafana/apps/provisioning/pkg/repository" "github.com/grafana/grafana/apps/provisioning/pkg/safepath" @@ -16,7 +17,8 @@ import ( ) // Convert git changes into resource file changes -func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer) error { +func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer, metrics jobs.JobMetrics) error { + syncStart := time.Now() if previousRef == currentRef { progress.SetFinalMessage(ctx, "same commit as last time") return nil @@ -24,7 +26,11 @@ func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef ctx, span := tracer.Start(ctx, "provisioning.sync.incremental") defer span.End() + defer func() { + metrics.RecordSyncDuration(jobs.SyncTypeIncremental, time.Since(syncStart)) + }() + compareStart := time.Now() compareCtx, compareSpan := tracer.Start(ctx, "provisioning.sync.incremental.compare_files") diff, err := repo.CompareFiles(compareCtx, previousRef, currentRef) if err != nil { @@ -33,6 +39,7 @@ func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef return tracing.Error(span, fmt.Errorf("compare files error: %w", err)) } compareSpan.End() + metrics.RecordIncrementalSyncPhase(jobs.IncrementalSyncPhaseCompare, time.Since(compareStart)) if len(diff) < 1 { progress.SetFinalMessage(ctx, "no changes detected between commits") @@ -41,18 +48,41 @@ func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef progress.SetTotal(ctx, len(diff)) progress.SetMessage(ctx, "replicating versioned changes") + applyStart := time.Now() + affectedFolders, err := applyIncrementalChanges(ctx, diff, repositoryResources, progress, tracer, span) + metrics.RecordIncrementalSyncPhase(jobs.IncrementalSyncPhaseApply, time.Since(applyStart)) + if err != nil { + return err + } + progress.SetMessage(ctx, "versioned changes replicated") + + if len(affectedFolders) > 0 { + cleanupStart := time.Now() + span.AddEvent("checking if impacted folders should be deleted", trace.WithAttributes(attribute.Int("affected_folders", len(affectedFolders)))) + err := cleanupOrphanedFolders(ctx, repo, affectedFolders, repositoryResources, tracer) + metrics.RecordIncrementalSyncPhase(jobs.IncrementalSyncPhaseCleanup, time.Since(cleanupStart)) + if err != nil { + return tracing.Error(span, fmt.Errorf("cleanup orphaned folders: %w", err)) + } + } + + return nil +} + +func applyIncrementalChanges(ctx context.Context, diff []repository.VersionedFileChange, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer, span trace.Span) (affectedFolders map[string]string, err error) { // this will keep track of any folders that had resources deleted from it // with key-value as path:grafana uid. // after cleaning up all resources, we will look to see if the foldrs are // now empty, and if so, delete them. - affectedFolders := make(map[string]string) + affectedFolders = make(map[string]string) + for _, change := range diff { if ctx.Err() != nil { - return ctx.Err() + return nil, ctx.Err() } if err := progress.TooManyErrors(); err != nil { - return tracing.Error(span, err) + return nil, tracing.Error(span, err) } if err := resources.IsPathSupported(change.Path); err != nil { @@ -68,7 +98,7 @@ func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef if err != nil { ensureFolderSpan.RecordError(err) ensureFolderSpan.End() - return tracing.Error(span, fmt.Errorf("unable to create empty file folder: %w", err)) + return nil, tracing.Error(span, fmt.Errorf("unable to create empty file folder: %w", err)) } progress.Record(ensureFolderCtx, jobs.JobResourceResult{ @@ -145,16 +175,7 @@ func IncrementalSync(ctx context.Context, repo repository.Versioned, previousRef progress.Record(ctx, result) } - progress.SetMessage(ctx, "versioned changes replicated") - - if len(affectedFolders) > 0 { - span.AddEvent("checking if impacted folders should be deleted", trace.WithAttributes(attribute.Int("affected_folders", len(affectedFolders)))) - if err := cleanupOrphanedFolders(ctx, repo, affectedFolders, repositoryResources, tracer); err != nil { - return tracing.Error(span, fmt.Errorf("cleanup orphaned folders: %w", err)) - } - } - - return nil + return affectedFolders, nil } // cleanupOrphanedFolders removes folders that no longer contain any resources in git after deletions have occurred. 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 ff032146788..417f8d89efe 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 @@ -28,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, 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) +// Execute provides a mock function with given fields: ctx, repo, previousRef, currentRef, repositoryResources, progress, tracer, metrics +func (_m *MockIncrementalSyncFn) Execute(ctx context.Context, repo repository.Versioned, previousRef string, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer, metrics jobs.JobMetrics) error { + ret := _m.Called(ctx, repo, previousRef, currentRef, repositoryResources, progress, tracer, metrics) 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, tracing.Tracer) error); ok { - r0 = rf(ctx, repo, previousRef, currentRef, repositoryResources, progress, tracer) + if rf, ok := ret.Get(0).(func(context.Context, repository.Versioned, string, string, resources.RepositoryResources, jobs.JobProgressRecorder, tracing.Tracer, jobs.JobMetrics) error); ok { + r0 = rf(ctx, repo, previousRef, currentRef, repositoryResources, progress, tracer, metrics) } else { r0 = ret.Error(0) } @@ -59,13 +59,14 @@ type MockIncrementalSyncFn_Execute_Call struct { // - repositoryResources resources.RepositoryResources // - progress jobs.JobProgressRecorder // - 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)} +// - metrics jobs.JobMetrics +func (_e *MockIncrementalSyncFn_Expecter) Execute(ctx interface{}, repo interface{}, previousRef interface{}, currentRef interface{}, repositoryResources interface{}, progress interface{}, tracer interface{}, metrics interface{}) *MockIncrementalSyncFn_Execute_Call { + return &MockIncrementalSyncFn_Execute_Call{Call: _e.mock.On("Execute", ctx, repo, previousRef, currentRef, repositoryResources, progress, tracer, metrics)} } -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 { +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, metrics jobs.JobMetrics)) *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), args[6].(tracing.Tracer)) + 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), args[7].(jobs.JobMetrics)) }) return _c } @@ -75,7 +76,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, tracing.Tracer) error) *MockIncrementalSyncFn_Execute_Call { +func (_c *MockIncrementalSyncFn_Execute_Call) RunAndReturn(run func(context.Context, repository.Versioned, string, string, resources.RepositoryResources, jobs.JobProgressRecorder, tracing.Tracer, jobs.JobMetrics) 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 95d595e8c6c..f694d7f5068 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/incremental_test.go +++ b/pkg/registry/apis/provisioning/jobs/sync/incremental_test.go @@ -5,6 +5,7 @@ import ( "fmt" "testing" + "github.com/prometheus/client_golang/prometheus" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/runtime/schema" @@ -31,7 +32,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, tracing.NewNoopTracerService()) + err := IncrementalSync(ctx, repo, "old-ref", "new-ref", repoResources, progress, tracing.NewNoopTracerService(), jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) require.EqualError(t, err, "context canceled") } @@ -388,7 +389,7 @@ func TestIncrementalSync(t *testing.T) { tt.setupMocks(repo, repoResources, progress) - err := IncrementalSync(context.Background(), repo, tt.previousRef, tt.currentRef, repoResources, progress, tracing.NewNoopTracerService()) + err := IncrementalSync(context.Background(), repo, tt.previousRef, tt.currentRef, repoResources, progress, tracing.NewNoopTracerService(), jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) if tt.expectedError != "" { require.EqualError(t, err, tt.expectedError) @@ -511,7 +512,7 @@ func TestIncrementalSync_CleanupOrphanedFolders(t *testing.T) { tt.setupMocks(repo, repoResources, progress) - err := IncrementalSync(context.Background(), repo, "old-ref", "new-ref", repoResources, progress, tracing.NewNoopTracerService()) + err := IncrementalSync(context.Background(), repo, "old-ref", "new-ref", repoResources, progress, tracing.NewNoopTracerService(), jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry())) 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 55272332add..7f677a8ce3b 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/sync.go +++ b/pkg/registry/apis/provisioning/jobs/sync/sync.go @@ -12,13 +12,13 @@ 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, maxSyncWorkers int) 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, metrics jobs.JobMetrics) 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, tracer tracing.Tracer) error +type IncrementalSyncFn func(ctx context.Context, repo repository.Versioned, previousRef, currentRef string, repositoryResources resources.RepositoryResources, progress jobs.JobProgressRecorder, tracer tracing.Tracer, metrics jobs.JobMetrics) error //go:generate mockery --name Syncer --structname MockSyncer --inpackage --filename syncer_mock.go --with-expecter type Syncer interface { @@ -30,15 +30,17 @@ type syncer struct { fullSync FullSyncFn incrementalSync IncrementalSyncFn tracer tracing.Tracer + metrics jobs.JobMetrics maxSyncWorkers int } -func NewSyncer(compare CompareFn, fullSync FullSyncFn, incrementalSync IncrementalSyncFn, tracer tracing.Tracer, maxSyncWorkers int) Syncer { +func NewSyncer(compare CompareFn, fullSync FullSyncFn, incrementalSync IncrementalSyncFn, tracer tracing.Tracer, maxSyncWorkers int, metrics jobs.JobMetrics) Syncer { return &syncer{ compare: compare, fullSync: fullSync, incrementalSync: incrementalSync, tracer: tracer, + metrics: metrics, maxSyncWorkers: maxSyncWorkers, } } @@ -57,11 +59,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, r.tracer) + return currentRef, r.incrementalSync(ctx, versionedRepo, cfg.Status.Sync.LastRef, currentRef, repositoryResources, progress, r.tracer, r.metrics) } } progress.SetMessage(ctx, "full sync") - return currentRef, r.fullSync(ctx, repo, r.compare, clients, currentRef, repositoryResources, progress, r.tracer, r.maxSyncWorkers) + return currentRef, r.fullSync(ctx, repo, r.compare, clients, currentRef, repositoryResources, progress, r.tracer, r.maxSyncWorkers, r.metrics) } diff --git a/pkg/registry/apis/provisioning/jobs/sync/sync_test.go b/pkg/registry/apis/provisioning/jobs/sync/sync_test.go index e0d1bbee7e1..d8c40689874 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/sync_test.go +++ b/pkg/registry/apis/provisioning/jobs/sync/sync_test.go @@ -10,6 +10,7 @@ import ( "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/prometheus/client_golang/prometheus" mock "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -66,7 +67,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, mock.Anything).Return(nil) + fullSyncFn.EXPECT().Execute(mock.Anything, mock.Anything, mock.Anything, mock.Anything, "new-ref", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) }, expectedMessages: []string{"full sync"}, }, @@ -88,7 +89,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, mock.Anything).Return(nil) + incrementalSyncFn.EXPECT().Execute(mock.Anything, mock.Anything, "old-ref", "new-ref", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil) }, expectedRef: "new-ref", expectedMessages: []string{"incremental sync"}, @@ -131,7 +132,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, 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, mock.Anything).Return(fmt.Errorf("incremental sync failed")) }, expectedRef: "new-ref", expectedMessages: []string{"incremental sync"}, @@ -161,6 +162,7 @@ func TestSyncer_Sync(t *testing.T) { incrementalSyncFn.Execute, tracing.NewNoopTracerService(), 10, + jobs.RegisterJobMetrics(prometheus.NewPedanticRegistry()), ) ref, err := syncer.Sync(context.Background(), repo, tt.options, repoResources, clients, progress) diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 485f6bfe5d5..2f31c2b6aa8 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, 10) + syncer := sync.NewSyncer(sync.Compare, sync.FullSync, sync.IncrementalSync, b.tracer, 10, metrics) syncWorker := sync.NewSyncWorker( b.clients, b.repositoryResources,