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

Provisioning: Wire up tracing and add trace for job sync (#111455)

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