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

Provisioning: Fix miscellaneous issues with setting and displaying sync status (#113529)

Browse Source 
* Provisioning: Preserve in progress job data

* Refactor code and cover more situations

* Fix linting

* Fix issue with remove path operation for started time

* Cleanup

* prettier

---------

Co-authored-by: Roberto Jimenez Sanchez <roberto.jimenez@grafana.com>
This commit is contained in:
Alex Khomenko
2025-11-07 12:27:25 +01:00
committed by GitHub
co-authored by Roberto Jimenez Sanchez
parent c784de6ef5
commit 8cb5f5646a
5 changed files with 157 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/registry/apis/provisioning/controller/repository.go
+34 -23
View File
@@ -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 {
pkg/registry/apis/provisioning/jobs.go
+22 -14
View File
@@ -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
}
pkg/registry/apis/provisioning/jobs/sync/worker.go
+20 -11
View File
@@ -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{}{
{
pkg/registry/apis/provisioning/jobs/sync/worker_test.go
+38 -28
View File
@@ -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)
public/app/features/provisioning/Repository/RepositoryPullStatusCard.tsx
+43 -33
View File
@@ -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 }) {
<Text variant="body">{status?.sync.job ?? 'N/A'}</Text>
</div>
{/* Last Ref */}
<Text color="secondary">
<Trans i18nKey="provisioning.repository-overview.last-ref">Last Ref:</Trans>
</Text>
<div className={styles.spanTwo}>
{hasUrl && lastCommitUrl ? (
<TextLink href={lastCommitUrl} external>
<div className={cx(styles.historicalData, { [styles.historicalDataOverlay]: isWorking })}>
{/* Last Ref */}
<Text color="secondary">
<Trans i18nKey="provisioning.repository-overview.last-ref">Last Ref:</Trans>
</Text>
<div className={styles.spanTwo}>
{hasUrl && lastCommitUrl ? (
<TextLink href={lastCommitUrl} external>
{status?.sync.lastRef
? status.sync.lastRef.substring(0, 7)
: t('provisioning.repository-overview.not-available', 'N/A')}
</TextLink>
) : (
<Text variant="body">
{status?.sync.lastRef
? status.sync.lastRef.substring(0, 7)
: t('provisioning.repository-overview.not-available', 'N/A')}
</Text>
</TextLink>
) : (
<Text variant="body">
{status?.sync.lastRef
? status.sync.lastRef.substring(0, 7)
: t('provisioning.repository-overview.not-available', 'N/A')}
</Text>
)}
</div>
<Text color="secondary">
<Trans i18nKey="provisioning.repository-overview.finished">Last successful pull:</Trans>
</Text>
<div className={styles.spanTwo}>
<Text variant="body">{formatTimestamp(status?.sync.finished)}</Text>
</div>
{!!status?.sync?.message?.length && (
<>
<Text color="secondary">
<Trans i18nKey="provisioning.repository-overview.messages">Messages:</Trans>
</Text>
<div className={styles.spanTwo}>
<MessageList messages={status.sync.message} variant="body" />
</div>
</>
)}
</div>
<Text color="secondary">
<Trans i18nKey="provisioning.repository-overview.finished">Last successful pull:</Trans>
</Text>
<div className={styles.spanTwo}>
<Text variant="body">{formatTimestamp(status?.sync.finished)}</Text>
</div>
{!!status?.sync?.message?.length && (
<>
<Text color="secondary">
<Trans i18nKey="provisioning.repository-overview.messages">Messages:</Trans>
</Text>
<div className={styles.spanTwo}>
<MessageList messages={status.sync.message} variant="body" />
</div>
</>
)}
</Grid>
</Card.Description>
<Card.Actions>
@@ -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,
}),
};
};