diff --git a/pkg/registry/apis/provisioning/jobs/driver.go b/pkg/registry/apis/provisioning/jobs/driver.go index e599792d28b..6a5ed6ed2c6 100644 --- a/pkg/registry/apis/provisioning/jobs/driver.go +++ b/pkg/registry/apis/provisioning/jobs/driver.go @@ -58,6 +58,9 @@ type jobDriver struct { // RepoGetter lets us access repositories to pass to the worker. repoGetter RepoGetter + // save info about finished jobs + historicJobs History + // Workers process the job. // Only the first worker who supports the job will process it; the rest are ignored. workers []Worker @@ -67,6 +70,7 @@ func NewJobDriver( timeout, cleanupInterval, jobInterval time.Duration, store Store, repoGetter RepoGetter, + historicJobs History, workers ...Worker, ) *jobDriver { return &jobDriver{ @@ -75,6 +79,7 @@ func NewJobDriver( jobInterval: jobInterval, store: store, repoGetter: repoGetter, + historicJobs: historicJobs, workers: workers, } } @@ -170,6 +175,15 @@ func (d *jobDriver) drive(ctx context.Context) error { job.Status.State = provisioning.JobStateSuccess // no error } + // Save the finished job + err = d.historicJobs.WriteJob(ctx, job) + if err != nil { + // We're not going to return this as it is not critical. Not ideal, but not critical. + logger.Warn("failed to create historic job", "historic_job", *job, "error", err) + } else { + logger.Debug("created historic job", "historic_job", *job) + } + // Mark the job as completed. if err := d.store.Complete(ctx, job); err != nil { return apifmt.Errorf("failed to complete job '%s' in '%s': %w", job.GetName(), job.GetNamespace(), err) diff --git a/pkg/registry/apis/provisioning/jobs/persistentstore.go b/pkg/registry/apis/provisioning/jobs/persistentstore.go index 5abf1ffc726..964625e9bf0 100644 --- a/pkg/registry/apis/provisioning/jobs/persistentstore.go +++ b/pkg/registry/apis/provisioning/jobs/persistentstore.go @@ -78,8 +78,7 @@ type jobStorage interface { // When persistentStore claims a job, it will update the status of it. This does a ResourceVersion check to ensure it is atomic; if the job has been claimed by another worker, the claim will fail. // When a job is completed, it is moved to the historic job store by first deleting it from the job store and then creating it in the historic job store. We are fine with the job being lost if the historic job store fails to create it. type persistentStore struct { - jobStore jobStorage - historicJobs History + jobStore jobStorage // clock is a function that returns the current time. clock func() time.Time @@ -95,7 +94,6 @@ type persistentStore struct { func NewStore( jobStore jobStorage, - historicJobs History, expiry time.Duration, ) (*persistentStore, error) { if expiry <= 0 { @@ -103,8 +101,7 @@ func NewStore( } return &persistentStore{ - jobStore: jobStore, - historicJobs: historicJobs, + jobStore: jobStore, clock: time.Now, expiry: expiry, @@ -275,14 +272,6 @@ func (s *persistentStore) Complete(ctx context.Context, job *provisioning.Job) e } delete(job.Labels, LabelJobClaim) - err = s.historicJobs.WriteJob(ctx, job) - if err != nil { - // We're not going to return this as it is not critical. Not ideal, but not critical. - logger.Warn("failed to create historic job", "historic_job", *job, "error", err) - } else { - logger.Debug("created historic job", "historic_job", *job) - } - logger.Debug("job completion done") return nil } @@ -413,8 +402,8 @@ func (s *persistentStore) InsertNotifications() chan struct{} { func (s *persistentStore) generateJobName(job *provisioning.Job) { switch job.Spec.Action { case provisioning.JobActionMigrate, provisioning.JobActionPull: - // Sync and migrate jobs should never run at the same time. Hence, the name encapsulates them both (and the spec differentiates them). - job.Name = job.Spec.Repository + "-syncmigrate" + // Pull and migrate jobs should never run at the same time. Hence, the name encapsulates them both (and the spec differentiates them). + job.Name = job.Spec.Repository + "-sync" case provisioning.JobActionPullRequest: var pr int if job.Spec.PullRequest != nil { diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 99f6632ead4..096768ca037 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -87,6 +87,7 @@ type APIBuilder struct { jobs.Queue jobs.Store } + jobHistory jobs.History tester *RepositoryTester resourceLister resources.ResourceLister repositoryLister listers.RepositoryLister @@ -342,12 +343,12 @@ func (b *APIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserver.APIGroupI return fmt.Errorf("failed to create historic job storage: %w", err) } - jobHistory, err := jobs.NewStorageBackedHistory(historicJobStore) + b.jobHistory, err = jobs.NewStorageBackedHistory(historicJobStore) if err != nil { return fmt.Errorf("failed to create historic job wrapper: %w", err) } - b.jobs, err = jobs.NewStore(realJobStore, jobHistory, time.Second*30) + b.jobs, err = jobs.NewStore(realJobStore, time.Second*30) if err != nil { return fmt.Errorf("failed to create job store: %w", err) } @@ -380,7 +381,7 @@ func (b *APIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserver.APIGroupI storage[provisioning.RepositoryResourceInfo.StoragePath("jobs")] = &jobsConnector{ repoGetter: b, jobs: b.jobs, - historic: jobHistory, + historic: b.jobHistory, } storage[provisioning.RepositoryResourceInfo.StoragePath("render")] = &renderConnector{ blob: b.unified, @@ -588,7 +589,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH return fmt.Errorf("create pull request worker: %w", err) } - driver := jobs.NewJobDriver(time.Second*28, time.Second*30, time.Second*30, b.jobs, b, + driver := jobs.NewJobDriver(time.Second*28, time.Second*30, time.Second*30, b.jobs, b, b.jobHistory, exportWorker, syncWorker, migrationWorker, pullRequestWorker) go driver.Run(postStartHookCtx.Context)