public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[App-Platform,Git-Sync]: Add sync job & phase metrics (#113253)

Browse Source 
* [git-sync] add sync job & phase metrics

* refactoring sync jobs for clarity and ease of instrumentation

* make gofmt
This commit is contained in:
Owen Diehl
2025-10-31 10:24:17 -07:00
committed by GitHub
parent d64cc987b8
commit b0ecc36fd6
10 changed files with 243 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/registry/apis/provisioning/jobs/metrics.go
+123 -4
View File
@@ -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"
}
}
pkg/registry/apis/provisioning/jobs/sync/full.go
+36 -9
View File
@@ -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
pkg/registry/apis/provisioning/jobs/sync/full_sync_fn_mock.go
+12 -10
View File
@@ -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
}
pkg/registry/apis/provisioning/jobs/sync/full_test.go
+8 -7
View File
@@ -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 {
pkg/registry/apis/provisioning/jobs/sync/incremental.go
+36 -15
View File
@@ -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.
pkg/registry/apis/provisioning/jobs/sync/incremental_sync_fn_mock.go
+11 -10
View File
@@ -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
}
pkg/registry/apis/provisioning/jobs/sync/incremental_test.go
+4 -3
View File
@@ -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)
pkg/registry/apis/provisioning/jobs/sync/sync.go
+7 -5
View File
@@ -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)
}
pkg/registry/apis/provisioning/jobs/sync/sync_test.go
+5 -3
View File
@@ -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)
pkg/registry/apis/provisioning/register.go
+1 -1
View File
@@ -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,