Provisioning: Write finished job to history before delete (#103328)

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)

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 {

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)