diff --git a/pkg/registry/apis/provisioning/controller/repository.go b/pkg/registry/apis/provisioning/controller/repository.go index 681619711e1..b0ba13385b6 100644 --- a/pkg/registry/apis/provisioning/controller/repository.go +++ b/pkg/registry/apis/provisioning/controller/repository.go @@ -404,32 +404,47 @@ func (rc *RepositoryController) addSyncJob(ctx context.Context, obj *provisionin return nil } -func (rc *RepositoryController) determineSyncStatus(obj *provisioning.Repository, syncOptions *provisioning.SyncJobOptions, healthStatus provisioning.HealthStatus) *provisioning.SyncStatus { +func (rc *RepositoryController) determineSyncStatusOps(obj *provisioning.Repository, syncOptions *provisioning.SyncJobOptions, healthStatus provisioning.HealthStatus) []map[string]interface{} { const unhealthyMessage = "Repository is unhealthy" hasUnhealthyMessage := len(obj.Status.Sync.Message) > 0 && obj.Status.Sync.Message[0] == unhealthyMessage + var patchOperations []map[string]interface{} + switch { case syncOptions != nil: - return &provisioning.SyncStatus{ - State: provisioning.JobStatePending, - LastRef: obj.Status.Sync.LastRef, - Started: time.Now().UnixMilli(), - } + // We will try to trigger a new sync job if we have sync options + patchOperations = append(patchOperations, map[string]interface{}{ + "op": "replace", + "path": "/status/sync/state", + "value": provisioning.JobStatePending, + }) + patchOperations = append(patchOperations, map[string]interface{}{ + "op": "replace", + "path": "/status/sync/started", + "value": int64(0), + }) case healthStatus.Healthy && hasUnhealthyMessage: // if the repository is healthy and the message is set, clear it // FIXME: is this the clearest way to do this? Should we introduce another status or way of way of handling more // specific errors? - return &provisioning.SyncStatus{ - LastRef: obj.Status.Sync.LastRef, - } + patchOperations = append(patchOperations, map[string]interface{}{ + "op": "replace", + "path": "/status/sync/message", + "value": []string{}, + }) case !healthStatus.Healthy && !hasUnhealthyMessage: // if the repository is unhealthy and the message is not already set, set it - return &provisioning.SyncStatus{ - State: provisioning.JobStateError, - Message: []string{unhealthyMessage}, - LastRef: obj.Status.Sync.LastRef, - } - default: - return nil + patchOperations = append(patchOperations, map[string]interface{}{ + "op": "replace", + "path": "/status/sync/state", + "value": provisioning.JobStateError, + }) + patchOperations = append(patchOperations, map[string]interface{}{ + "op": "replace", + "path": "/status/sync/message", + "value": []string{unhealthyMessage}, + }) } + + return patchOperations } //nolint:gocyclo @@ -509,13 +524,7 @@ func (rc *RepositoryController) process(item *queueItem) error { // determine the sync strategy and sync status to apply syncOptions := rc.determineSyncStrategy(ctx, obj, repo, shouldResync, healthStatus) - if syncStatus := rc.determineSyncStatus(obj, syncOptions, healthStatus); syncStatus != nil { - patchOperations = append(patchOperations, map[string]interface{}{ - "op": "replace", - "path": "/status/sync", - "value": syncStatus, - }) - } + patchOperations = append(patchOperations, rc.determineSyncStatusOps(obj, syncOptions, healthStatus)...) // Apply all patch operations if len(patchOperations) > 0 { @@ -525,6 +534,8 @@ func (rc *RepositoryController) process(item *queueItem) error { } } + // QUESTION: should we trigger the sync job after we have applied all patch operations or before? + // Is there are risk of race condition here? // Trigger sync job after we have applied all patch operations if syncOptions != nil { if err := rc.addSyncJob(ctx, obj, syncOptions); err != nil { diff --git a/pkg/registry/apis/provisioning/jobs.go b/pkg/registry/apis/provisioning/jobs.go index b1e174a04b7..94b705414f8 100644 --- a/pkg/registry/apis/provisioning/jobs.go +++ b/pkg/registry/apis/provisioning/jobs.go @@ -132,28 +132,36 @@ func (c *jobsConnector) Connect( } spec.Repository = name - // If a sync job is being created, we should update its status to pending. + job, err := c.jobs.GetJobQueue().Insert(ctx, cfg.Namespace, spec) + if err != nil { + responder.Error(err) + return + } + + // For pull jobs update the sync status + // patch the sync status 'state' to 'pending', and reset the 'started' field, leaving other fields unchanged. + // Intentionally maintain the previous job name until the jobs is picked up. if spec.Pull != nil { - err = c.statusPatcherProvider.GetStatusPatcher().Patch(ctx, cfg, map[string]interface{}{ - "op": "replace", - "path": "/status/sync", - "value": &provisioning.SyncStatus{ - State: provisioning.JobStatePending, - LastRef: cfg.Status.Sync.LastRef, - Started: time.Now().UnixMilli(), + err = c.statusPatcherProvider.GetStatusPatcher().Patch(ctx, cfg, + map[string]interface{}{ + "op": "replace", + "path": "/status/sync/state", + "value": provisioning.JobStatePending, }, - }) + map[string]interface{}{ + // Use "replace" instead of "remove" since "remove" fails if the path does not exist (RFC 6902). + // "started" field uses "omitempty", so it may be missing in the JSON. + "op": "replace", + "path": "/status/sync/started", + "value": int64(0), + }, + ) if err != nil { responder.Error(err) return } } - job, err := c.jobs.GetJobQueue().Insert(ctx, cfg.Namespace, spec) - if err != nil { - responder.Error(err) - return - } responder.Object(http.StatusAccepted, job) }), 30*time.Second), nil } diff --git a/pkg/registry/apis/provisioning/jobs/sync/worker.go b/pkg/registry/apis/provisioning/jobs/sync/worker.go index dfb376ca470..05e6340b193 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/worker.go +++ b/pkg/registry/apis/provisioning/jobs/sync/worker.go @@ -110,25 +110,35 @@ func (r *SyncWorker) Process(ctx context.Context, repo repository.Repository, jo } syncStatus := job.Status.ToSyncStatus(job.Name) - // Preserve last ref as we use replace operation + // Preserve last ref lastRef := repo.Config().Status.Sync.LastRef syncStatus.LastRef = lastRef - if syncStatus.State == "" { - syncStatus.State = provisioning.JobStateWorking - } + // Ensure the sync state is set to 'working' if not already set or still pending. + // FIXME: This should not be needed as the progress recorder should have set it to 'working' by now. + syncStatus.State = provisioning.JobStateWorking - // Update sync status at start using JSON patch + // Update sync status at start using granular JSON patch operations + // Only patch fields that are actually being set to avoid overwriting with zero values patchOperations := []map[string]interface{}{ { "op": "replace", - "path": "/status/sync", - "value": syncStatus, + "path": "/status/sync/state", + "value": syncStatus.State, + }, + { + "op": "replace", + "path": "/status/sync/job", + "value": syncStatus.JobID, + }, + { + "op": "replace", + "path": "/status/sync/started", + "value": syncStatus.Started, }, } progress.SetMessage(ctx, "update sync status at start") - statusCtx, statusSpan := r.tracer.Start(ctx, "provisioning.sync.update_start_status") if err := r.patchStatus(statusCtx, cfg, patchOperations...); err != nil { statusSpan.End() @@ -174,14 +184,13 @@ func (r *SyncWorker) Process(ctx context.Context, repo repository.Repository, jo } syncSpan.End() - // Create sync status and set hash if successful - if syncStatus.State == provisioning.JobStateSuccess { + if syncStatus.State != provisioning.JobStateError { syncStatus.LastRef = currentRef } else { + // Preserve the original lastRef on error syncStatus.LastRef = lastRef } - // Update final status using JSON patch progress.SetMessage(ctx, "update status and stats") patchOperations = []map[string]interface{}{ { diff --git a/pkg/registry/apis/provisioning/jobs/sync/worker_test.go b/pkg/registry/apis/provisioning/jobs/sync/worker_test.go index bf0b60bea01..a5013539957 100644 --- a/pkg/registry/apis/provisioning/jobs/sync/worker_test.go +++ b/pkg/registry/apis/provisioning/jobs/sync/worker_test.go @@ -115,17 +115,18 @@ func TestSyncWorker_Process(t *testing.T) { rw.MockRepository.On("Config").Return(repoConfig) pr.On("SetMessage", mock.Anything, "update sync status at start").Return() - rpf.On("Execute", mock.Anything, repoConfig, mock.MatchedBy(func(patch map[string]interface{}) bool { - if patch["op"] != "replace" || patch["path"] != "/status/sync" { - return false - } - - if patch["value"].(provisioning.SyncStatus).LastRef != "existing-ref" || patch["value"].(provisioning.SyncStatus).JobID != "test-job" { - return false - } - - return true - })).Return(errors.New("failed to patch status")) + // Expect granular patches for state, job, and started fields + rpf.On("Execute", mock.Anything, repoConfig, + mock.MatchedBy(func(patch map[string]interface{}) bool { + return patch["op"] == "replace" && patch["path"] == "/status/sync/state" + }), + mock.MatchedBy(func(patch map[string]interface{}) bool { + return patch["op"] == "replace" && patch["path"] == "/status/sync/job" + }), + mock.MatchedBy(func(patch map[string]interface{}) bool { + return patch["op"] == "replace" && patch["path"] == "/status/sync/started" + }), + ).Return(errors.New("failed to patch status")) }, expectedError: "update repo with job status at start: failed to patch status", }, @@ -151,9 +152,9 @@ func TestSyncWorker_Process(t *testing.T) { // Storage is migrated ds.On("ReadFromUnified", mock.Anything, mock.Anything).Return(true, nil).Twice() - // Initial status update succeeds + // Initial status update succeeds - expect granular patches pr.On("SetMessage", mock.Anything, "update sync status at start").Return() - rpf.On("Execute", mock.Anything, repoConfig, mock.Anything).Return(nil).Once() + rpf.On("Execute", mock.Anything, repoConfig, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() // Repository resources creation fails rrf.On("Client", mock.Anything, mock.Anything).Return(nil, errors.New("failed to create repository resources client")) @@ -188,9 +189,9 @@ func TestSyncWorker_Process(t *testing.T) { // Storage is migrated ds.On("ReadFromUnified", mock.Anything, mock.Anything).Return(true, nil).Twice() - // Initial status update succeeds + // Initial status update succeeds - expect granular patches pr.On("SetMessage", mock.Anything, "update sync status at start").Return() - rpf.On("Execute", mock.Anything, repoConfig, mock.Anything).Return(nil).Once() + rpf.On("Execute", mock.Anything, repoConfig, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() // Repository resources creation succeeds rrf.On("Client", mock.Anything, mock.Anything).Return(&resources.MockRepositoryResources{}, nil) @@ -224,9 +225,9 @@ func TestSyncWorker_Process(t *testing.T) { // Storage is migrated ds.On("ReadFromUnified", mock.Anything, mock.Anything).Return(true, nil).Twice() - // Initial status update + // Initial status update - expect granular patches pr.On("SetMessage", mock.Anything, "update sync status at start").Return() - rpf.On("Execute", mock.Anything, repoConfig, mock.Anything).Return(nil) + rpf.On("Execute", mock.Anything, repoConfig, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() // Setup resources and clients mockRepoResources := resources.NewMockRepositoryResources(t) @@ -254,7 +255,7 @@ func TestSyncWorker_Process(t *testing.T) { } syncStatus := patch["value"].(provisioning.SyncStatus) return syncStatus.LastRef == "new-ref" && syncStatus.State == provisioning.JobStateSuccess - })).Return(nil) + })).Return(nil).Once() }, expectedError: "", }, @@ -277,9 +278,9 @@ func TestSyncWorker_Process(t *testing.T) { // Storage is migrated ds.On("ReadFromUnified", mock.Anything, mock.Anything).Return(true, nil).Twice() - // Initial status update + // Initial status update - expect granular patches pr.On("SetMessage", mock.Anything, "update sync status at start").Return() - rpf.On("Execute", mock.Anything, repoConfig, mock.Anything).Return(nil) + rpf.On("Execute", mock.Anything, repoConfig, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() // Setup resources and clients mockRepoResources := resources.NewMockRepositoryResources(t) @@ -308,7 +309,7 @@ func TestSyncWorker_Process(t *testing.T) { patch["path"] == "/status/sync" && syncStatus.LastRef == "existing-ref" && // LastRef should not change on failure syncStatus.State == provisioning.JobStateError - })).Return(nil) + })).Return(nil).Once() }, expectedError: "sync operation failed", }, @@ -334,7 +335,9 @@ func TestSyncWorker_Process(t *testing.T) { pr.On("SetMessage", mock.Anything, mock.Anything).Return() pr.On("StrictMaxErrors", 20).Return() pr.On("Complete", mock.Anything, mock.Anything).Return(provisioning.JobStatus{State: provisioning.JobStateSuccess}) - rpf.On("Execute", mock.Anything, mock.Anything, mock.Anything).Return(nil) + // Initial patch with granular updates, final patch with full sync status + rpf.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() + rpf.On("Execute", mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() s.On("Sync", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("new-ref", nil) }, expectedError: "", @@ -355,10 +358,13 @@ func TestSyncWorker_Process(t *testing.T) { mockRepoResources.On("Stats", mock.Anything).Return(nil, nil) rrf.On("Client", mock.Anything, mock.Anything).Return(mockRepoResources, nil) - // Verify only sync status is patched + // Initial patch with granular updates + rpf.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() + + // Verify only sync status is patched for final update rpf.On("Execute", mock.Anything, mock.Anything, mock.MatchedBy(func(patch map[string]interface{}) bool { return patch["path"] == "/status/sync" - })).Return(nil) + })).Return(nil).Once() // Simple mocks for other calls mockClients := resources.NewMockResourceClients(t) @@ -381,7 +387,8 @@ func TestSyncWorker_Process(t *testing.T) { } rw.MockRepository.On("Config").Return(repoConfig) ds.On("ReadFromUnified", mock.Anything, mock.Anything).Return(true, nil).Twice() - rpf.On("Execute", mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() + // Initial patch with granular updates + rpf.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() mockRepoResources := resources.NewMockRepositoryResources(t) stats := &provisioning.ResourceStats{ @@ -468,10 +475,13 @@ func TestSyncWorker_Process(t *testing.T) { mockRepoResources.On("Stats", mock.Anything).Return(stats, nil) rrf.On("Client", mock.Anything, mock.Anything).Return(mockRepoResources, nil) + // Initial patch with granular updates + rpf.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() + // Verify only sync status is patched (multiple stats should be ignored) rpf.On("Execute", mock.Anything, mock.Anything, mock.MatchedBy(func(patch map[string]interface{}) bool { return patch["path"] == "/status/sync" - })).Return(nil) + })).Return(nil).Once() // Simple mocks for other calls mockClients := resources.NewMockResourceClients(t) @@ -495,8 +505,8 @@ func TestSyncWorker_Process(t *testing.T) { rw.MockRepository.On("Config").Return(repoConfig) ds.On("ReadFromUnified", mock.Anything, mock.Anything).Return(true, nil).Twice() - // Initial status patch succeeds - rpf.On("Execute", mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() + // Initial status patch succeeds - expect granular patches + rpf.On("Execute", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() // Setup resources and clients mockRepoResources := resources.NewMockRepositoryResources(t) diff --git a/public/app/features/provisioning/Repository/RepositoryPullStatusCard.tsx b/public/app/features/provisioning/Repository/RepositoryPullStatusCard.tsx index 8758f8de911..7733d77db4f 100644 --- a/public/app/features/provisioning/Repository/RepositoryPullStatusCard.tsx +++ b/public/app/features/provisioning/Repository/RepositoryPullStatusCard.tsx @@ -1,4 +1,4 @@ -import { css } from '@emotion/css'; +import { css, cx } from '@emotion/css'; import { t, Trans } from '@grafana/i18n'; import { Badge, Card, Grid, Text, TextLink, useStyles2 } from '@grafana/ui'; @@ -17,6 +17,8 @@ export function RepositoryPullStatusCard({ repo }: { repo: Repository }) { const statusColor = getStatusColor(status?.sync.state); const statusIcon = getStatusIcon(status?.sync.state); + const isWorking = status?.sync.state === 'working' || status?.sync.state === 'pending'; + const { url: lastCommitUrl, hasUrl } = getRepoCommitUrl(repo.spec, status?.sync.lastRef); return ( @@ -42,45 +44,45 @@ export function RepositoryPullStatusCard({ repo }: { repo: Repository }) { {status?.sync.job ?? 'N/A'} - {/* Last Ref */} - - Last Ref: - -
- {hasUrl && lastCommitUrl ? ( - +
+ {/* Last Ref */} + + Last Ref: + +
+ {hasUrl && lastCommitUrl ? ( + + {status?.sync.lastRef + ? status.sync.lastRef.substring(0, 7) + : t('provisioning.repository-overview.not-available', 'N/A')} + + ) : ( {status?.sync.lastRef ? status.sync.lastRef.substring(0, 7) : t('provisioning.repository-overview.not-available', 'N/A')} - - ) : ( - - {status?.sync.lastRef - ? status.sync.lastRef.substring(0, 7) - : t('provisioning.repository-overview.not-available', 'N/A')} - + )} +
+ + + Last successful pull: + +
+ {formatTimestamp(status?.sync.finished)} +
+ + {!!status?.sync?.message?.length && ( + <> + + Messages: + +
+ +
+ )}
- - - Last successful pull: - -
- {formatTimestamp(status?.sync.finished)} -
- - {!!status?.sync?.message?.length && ( - <> - - Messages: - -
- -
- - )} @@ -95,5 +97,13 @@ const getStyles = () => { spanTwo: css({ gridColumn: 'span 2', }), + historicalData: css({ + gridColumn: '1 / -1', + display: 'grid', + gridTemplateColumns: 'subgrid', + }), + historicalDataOverlay: css({ + opacity: 0.6, + }), }; };