diff --git a/pkg/registry/apis/provisioning/jobs/concurrent_driver.go b/pkg/registry/apis/provisioning/jobs/concurrent_driver.go index dce8d65a97e..39f705aca31 100644 --- a/pkg/registry/apis/provisioning/jobs/concurrent_driver.go +++ b/pkg/registry/apis/provisioning/jobs/concurrent_driver.go @@ -14,7 +14,6 @@ import ( type ConcurrentJobDriver struct { numDrivers int jobTimeout time.Duration - cleanupInterval time.Duration jobInterval time.Duration leaseRenewalInterval time.Duration store Store @@ -27,7 +26,7 @@ type ConcurrentJobDriver struct { // NewConcurrentJobDriver creates a new concurrent job driver that spawns multiple job drivers. func NewConcurrentJobDriver( numDrivers int, - jobTimeout, cleanupInterval, jobInterval, leaseRenewalInterval time.Duration, + jobTimeout, jobInterval, leaseRenewalInterval time.Duration, store Store, repoGetter RepoGetter, historicJobs HistoryWriter, @@ -45,24 +44,12 @@ func NewConcurrentJobDriver( if leaseRenewalInterval < 5*time.Second { leaseRenewalInterval = 5 * time.Second } - // For lease-based cleanup, run at most every 3-4 lease renewal intervals - // to detect expired leases promptly but not too aggressively - if cleanupInterval <= 0 { - cleanupInterval = leaseRenewalInterval * 3 - } - if cleanupInterval < 30*time.Second { - cleanupInterval = 30 * time.Second // Minimum cleanup interval - } - if cleanupInterval > 5*time.Minute { - cleanupInterval = 5 * time.Minute // Maximum cleanup interval - } recordConcurrentDriverMetric(registry, numDrivers) return &ConcurrentJobDriver{ numDrivers: numDrivers, jobTimeout: jobTimeout, - cleanupInterval: cleanupInterval, jobInterval: jobInterval, leaseRenewalInterval: leaseRenewalInterval, store: store, @@ -73,43 +60,17 @@ func NewConcurrentJobDriver( }, nil } -// Run starts multiple job drivers concurrently and handles cleanup coordination. +// Run starts multiple job drivers concurrently. // This is a blocking function that will run until the context is canceled or an error occurs. // // Note: This function intentionally does NOT create a tracing span because it runs indefinitely -// until shutdown. Individual job processing and cleanup operations already have their own spans. +// until shutdown. Individual job processing operations already have their own spans. func (c *ConcurrentJobDriver) Run(ctx context.Context) error { logger := logging.FromContext(ctx).With("logger", "concurrent-job-driver", "num_drivers", c.numDrivers) - logger.Info("start concurrent job driver", "num_drivers", c.numDrivers, "cleanup_interval", c.cleanupInterval) - - // Set up cleanup ticker - runs more frequently with lease-based approach - cleanupTicker := time.NewTicker(c.cleanupInterval) - defer cleanupTicker.Stop() - - // Initial cleanup - if err := c.store.Cleanup(ctx); err != nil { - logger.Error("failed initial cleanup", "error", err) - } + logger.Info("start concurrent job driver", "num_drivers", c.numDrivers) var wg sync.WaitGroup - errChan := make(chan error, c.numDrivers+1) // +1 for cleanup goroutine - - // Start cleanup goroutine - wg.Add(1) - go func() { - defer wg.Done() - for { - select { - case <-cleanupTicker.C: - if err := c.store.Cleanup(ctx); err != nil { - logger.Error("failed cleanup", "error", err) - } - case <-ctx.Done(): - logger.Debug("cleanup routine stopped") - return - } - } - }() + errChan := make(chan error, c.numDrivers) // Start driver goroutines for i := 0; i < c.numDrivers; i++ { diff --git a/pkg/registry/apis/provisioning/jobs/driver.go b/pkg/registry/apis/provisioning/jobs/driver.go index b1cb0ee6da1..5241c211517 100644 --- a/pkg/registry/apis/provisioning/jobs/driver.go +++ b/pkg/registry/apis/provisioning/jobs/driver.go @@ -31,14 +31,10 @@ type Store interface { // The err may be ErrNoJobs if there are no jobs to claim. Claim(ctx context.Context) (job *provisioning.Job, rollback func(), err error) - // Complete marks a job as completed and moves it to the historic job store. - // When in the historic store, there is no more claim on the job. + // Complete marks a job as completed and removes it from the active job store. + // Callers are responsible for writing the job to history after calling this. Complete(ctx context.Context, job *provisioning.Job) error - // Cleanup should be called periodically to clean up abandoned jobs. - // An abandoned job is one that has been claimed by a worker, but the worker has not updated the job in a while. - Cleanup(ctx context.Context) error - // Update saves the job back to the store. Update(ctx context.Context, job *provisioning.Job) (*provisioning.Job, error) @@ -48,6 +44,10 @@ type Store interface { // Get retrieves a job by name for conflict resolution. Get(ctx context.Context, namespace, name string) (*provisioning.Job, error) + + // ListExpiredJobs lists jobs with expired leases (claim timestamp older than the given time). + // Returns jobs in batches up to the specified limit. + ListExpiredJobs(ctx context.Context, expiredBefore time.Time, limit int) ([]*provisioning.Job, error) } // jobDriver drives jobs to completion and manages the job queue. diff --git a/pkg/registry/apis/provisioning/jobs/expired_job_cleanup.go b/pkg/registry/apis/provisioning/jobs/expired_job_cleanup.go new file mode 100644 index 00000000000..d6cc1dc2e25 --- /dev/null +++ b/pkg/registry/apis/provisioning/jobs/expired_job_cleanup.go @@ -0,0 +1,177 @@ +package jobs + +import ( + "context" + "time" + + "go.opentelemetry.io/otel/attribute" + + "github.com/grafana/grafana-app-sdk/logging" + "github.com/grafana/grafana/apps/provisioning/pkg/apifmt" + provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" + "github.com/grafana/grafana/pkg/apimachinery/identity" + "github.com/grafana/grafana/pkg/infra/tracing" +) + +// JobCleanupController handles cleanup of expired/abandoned jobs. +type JobCleanupController struct { + store Store + historicJobs HistoryWriter + clock func() time.Time + expiry time.Duration + cleanupInterval time.Duration +} + +// NewJobCleanupController creates a new job cleanup controller. +func NewJobCleanupController( + store Store, + historicJobs HistoryWriter, + expiry time.Duration, +) *JobCleanupController { + // Calculate cleanup interval based on expiry duration + // Run cleanup every 3-4 expiry intervals to detect expired leases promptly but not too aggressively + cleanupInterval := expiry * 3 + + // Enforce minimum and maximum bounds + if cleanupInterval < 30*time.Second { + cleanupInterval = 30 * time.Second + } + if cleanupInterval > 5*time.Minute { + cleanupInterval = 5 * time.Minute + } + + return &JobCleanupController{ + store: store, + historicJobs: historicJobs, + clock: time.Now, + expiry: expiry, + cleanupInterval: cleanupInterval, + } +} + +// Run starts the cleanup loop that runs at an appropriate interval. +// This is a blocking function that runs until the context is canceled. +func (c *JobCleanupController) Run(ctx context.Context) error { + logger := logging.FromContext(ctx).With("logger", "job-cleanup-controller") + ctx = logging.Context(ctx, logger) + + // Set up provisioning identity to access jobs across all namespaces + ctx, _, err := identity.WithProvisioningIdentity(ctx, "*") + if err != nil { + return apifmt.Errorf("failed to grant provisioning identity for cleanup: %w", err) + } + + logger.Info("starting job cleanup controller", "cleanup_interval", c.cleanupInterval, "expiry", c.expiry) + + // Initial cleanup + if err := c.Cleanup(ctx); err != nil { + logger.Error("failed to clean up jobs at start", "error", err) + } + + ticker := time.NewTicker(c.cleanupInterval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + if err := c.Cleanup(ctx); err != nil { + logger.Error("failed to cleanup jobs", "error", err) + } + case <-ctx.Done(): + logger.Info("job cleanup controller stopping") + return ctx.Err() + } + } +} + +// Cleanup finds jobs with expired leases and marks them as failed. +// This should be called periodically to clean up jobs from crashed workers. +func (c *JobCleanupController) Cleanup(ctx context.Context) error { + ctx, span := tracing.Start(ctx, "provisioning.jobs.cleanup") + defer span.End() + + startTime := c.clock() + logger := logging.FromContext(ctx) + + // Find jobs with expired leases + expiredBefore := c.clock().Add(-c.expiry) + + // Process in batches of 100 to avoid overwhelming the system + const batchSize = 100 + jobs, err := c.store.ListExpiredJobs(ctx, expiredBefore, batchSize) + if err != nil { + span.RecordError(err) + return apifmt.Errorf("failed to list jobs with expired leases: %w", err) + } + + // If no jobs found, cleanup is complete + if len(jobs) == 0 { + duration := c.clock().Sub(startTime) + span.SetAttributes( + attribute.Int("count", 0), + attribute.Int64("duration_ms", duration.Milliseconds()), + ) + return nil + } + + logger.Info("cleaning up expired jobs", "count", len(jobs)) + + for _, job := range jobs { + if err := c.cleanUpExpiredJob(ctx, job); err != nil { + // Log error but continue processing other jobs + logger.Error("failed to clean up expired job", "error", err, "job", job.GetName(), "namespace", job.GetNamespace()) + } + } + + duration := c.clock().Sub(startTime) + logger.Info("cleanup complete", "duration", duration, "count", len(jobs)) + + span.SetAttributes( + attribute.Int("count", len(jobs)), + attribute.Int64("duration_ms", duration.Milliseconds()), + ) + + return nil +} + +// cleanUpExpiredJob marks a single expired job as failed and archives it. +func (c *JobCleanupController) cleanUpExpiredJob(ctx context.Context, job *provisioning.Job) error { + ctx, span := tracing.Start(ctx, "provisioning.jobs.cleanup.complete_expired_job") + defer span.End() + + // Mark job as failed due to lease expiry + jobCopy := job.DeepCopy() + jobCopy.Status.State = provisioning.JobStateError + jobCopy.Status.Message = "Job failed due to lease expiry - worker may have crashed or lost connection" + jobCopy.Status.Finished = c.clock().UnixMilli() + + span.SetAttributes( + attribute.String("job.name", jobCopy.GetName()), + attribute.String("job.namespace", jobCopy.GetNamespace()), + attribute.String("job.repository", jobCopy.Spec.Repository), + attribute.String("job.action", string(jobCopy.Spec.Action)), + ) + + jobLogger := logging.FromContext(ctx).With("namespace", jobCopy.GetNamespace(), "job", jobCopy.GetName(), "action", jobCopy.Spec.Action) + + // Delete from active job store first + if err := c.store.Complete(ctx, jobCopy); err != nil { + span.RecordError(err) + return apifmt.Errorf("failed to complete expired job: %w", err) + } + + // Remove the claim label before archiving + if jobCopy.Labels != nil { + delete(jobCopy.Labels, LabelJobClaim) + } + + // Write to history after deleting from active store (matching driver.go pattern) + if err := c.historicJobs.WriteJob(ctx, jobCopy); err != nil { + span.RecordError(err) + jobLogger.Warn("failed to write expired job to history", "error", err) + // Job was already deleted, so we can't recover from this + } + + jobLogger.Debug("cleaned up expired job") + return nil +} diff --git a/pkg/registry/apis/provisioning/jobs/expired_job_cleanup_test.go b/pkg/registry/apis/provisioning/jobs/expired_job_cleanup_test.go new file mode 100644 index 00000000000..6018e5d91e0 --- /dev/null +++ b/pkg/registry/apis/provisioning/jobs/expired_job_cleanup_test.go @@ -0,0 +1,465 @@ +package jobs + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func TestNewJobCleanupController(t *testing.T) { + store := &MockStore{} + historyWriter := &MockHistoryWriter{} + + t.Run("creates controller with default cleanup interval", func(t *testing.T) { + expiry := 30 * time.Second + controller := NewJobCleanupController(store, historyWriter, expiry) + + assert.NotNil(t, controller) + assert.Equal(t, expiry, controller.expiry) + // Cleanup interval should be 3x expiry = 90 seconds + assert.Equal(t, 90*time.Second, controller.cleanupInterval) + }) + + t.Run("enforces minimum cleanup interval", func(t *testing.T) { + // With expiry of 5 seconds, 3x = 15 seconds, but minimum is 30 seconds + expiry := 5 * time.Second + controller := NewJobCleanupController(store, historyWriter, expiry) + + assert.Equal(t, 30*time.Second, controller.cleanupInterval) + }) + + t.Run("enforces maximum cleanup interval", func(t *testing.T) { + // With expiry of 5 minutes, 3x = 15 minutes, but maximum is 5 minutes + expiry := 5 * time.Minute + controller := NewJobCleanupController(store, historyWriter, expiry) + + assert.Equal(t, 5*time.Minute, controller.cleanupInterval) + }) +} + +func TestJobCleanupController_Cleanup(t *testing.T) { + t.Run("no expired jobs returns nil", func(t *testing.T) { + store := &MockStore{} + historyWriter := &MockHistoryWriter{} + + controller := NewJobCleanupController(store, historyWriter, 30*time.Second) + ctx := context.Background() + + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{}, nil) + + err := controller.Cleanup(ctx) + + assert.NoError(t, err) + store.AssertExpectations(t) + // store.AssertNotCalled(t, "Complete") - not needed with combined Store mock + historyWriter.AssertNotCalled(t, "WriteJob") + }) + + t.Run("error listing expired jobs returns error", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + controller := NewJobCleanupController(store, historyWriter, 30*time.Second) + ctx := context.Background() + + expectedErr := errors.New("list failed") + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return(nil, expectedErr) + + err := controller.Cleanup(ctx) + + assert.Error(t, err) + assert.Contains(t, err.Error(), "failed to list jobs with expired leases") + store.AssertExpectations(t) + }) + + t.Run("successfully cleans up expired job", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + controller := NewJobCleanupController(store, historyWriter, 30*time.Second) + ctx := context.Background() + + job := &provisioning.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-job", + Namespace: "test-ns", + Labels: map[string]string{ + LabelJobClaim: "123456789", + }, + }, + Spec: provisioning.JobSpec{ + Repository: "test-repo", + Action: provisioning.JobActionPull, + }, + } + + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{job}, nil) + store.On("Complete", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool { + return j.Status.State == provisioning.JobStateError && + j.Status.Message == "Job failed due to lease expiry - worker may have crashed or lost connection" + })).Return(nil) + historyWriter.On("WriteJob", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool { + // Verify claim label was removed before writing to history + _, hasLabel := j.Labels[LabelJobClaim] + return !hasLabel && j.Status.State == provisioning.JobStateError + })).Return(nil) + + err := controller.Cleanup(ctx) + + assert.NoError(t, err) + store.AssertExpectations(t) + historyWriter.AssertExpectations(t) + }) + + t.Run("continues on complete error", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + controller := NewJobCleanupController(store, historyWriter, 30*time.Second) + ctx := context.Background() + + job1 := &provisioning.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "job-1", + Namespace: "test-ns", + Labels: map[string]string{LabelJobClaim: "123"}, + }, + Spec: provisioning.JobSpec{ + Repository: "repo-1", + Action: provisioning.JobActionPull, + }, + } + job2 := &provisioning.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "job-2", + Namespace: "test-ns", + Labels: map[string]string{LabelJobClaim: "456"}, + }, + Spec: provisioning.JobSpec{ + Repository: "repo-2", + Action: provisioning.JobActionPull, + }, + } + + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{job1, job2}, nil) + + // First job fails to complete + store.On("Complete", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool { + return j.Name == "job-1" + })).Return(errors.New("complete failed")) + + // Second job succeeds + store.On("Complete", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool { + return j.Name == "job-2" + })).Return(nil) + historyWriter.On("WriteJob", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool { + return j.Name == "job-2" + })).Return(nil) + + err := controller.Cleanup(ctx) + + // Should not return error, continues processing + assert.NoError(t, err) + store.AssertExpectations(t) + historyWriter.AssertExpectations(t) + }) + + t.Run("continues on history write error", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + controller := NewJobCleanupController(store, historyWriter, 30*time.Second) + ctx := context.Background() + + job := &provisioning.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-job", + Namespace: "test-ns", + Labels: map[string]string{LabelJobClaim: "123"}, + }, + Spec: provisioning.JobSpec{ + Repository: "test-repo", + Action: provisioning.JobActionPull, + }, + } + + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{job}, nil) + store.On("Complete", mock.Anything, mock.Anything).Return(nil) + historyWriter.On("WriteJob", mock.Anything, mock.Anything).Return(errors.New("write failed")) + + err := controller.Cleanup(ctx) + + // Should not return error, just log warning + assert.NoError(t, err) + store.AssertExpectations(t) + historyWriter.AssertExpectations(t) + }) + + t.Run("sets job status correctly", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + fixedTime := time.Date(2024, 1, 1, 12, 0, 0, 0, time.UTC) + controller := NewJobCleanupController(store, historyWriter, 30*time.Second) + controller.clock = func() time.Time { return fixedTime } + ctx := context.Background() + + job := &provisioning.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-job", + Namespace: "test-ns", + Labels: map[string]string{LabelJobClaim: "123"}, + }, + Spec: provisioning.JobSpec{ + Repository: "test-repo", + Action: provisioning.JobActionPull, + }, + } + + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{job}, nil) + store.On("Complete", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool { + assert.Equal(t, provisioning.JobStateError, j.Status.State) + assert.Equal(t, "Job failed due to lease expiry - worker may have crashed or lost connection", j.Status.Message) + assert.Equal(t, fixedTime.UnixMilli(), j.Status.Finished) + return true + })).Return(nil) + historyWriter.On("WriteJob", mock.Anything, mock.Anything).Return(nil) + + err := controller.Cleanup(ctx) + + assert.NoError(t, err) + store.AssertExpectations(t) + }) + + t.Run("removes claim label before writing to history", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + controller := NewJobCleanupController(store, historyWriter, 30*time.Second) + ctx := context.Background() + + job := &provisioning.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-job", + Namespace: "test-ns", + Labels: map[string]string{ + LabelJobClaim: "123456789", + "other-label": "value", + }, + }, + Spec: provisioning.JobSpec{ + Repository: "test-repo", + Action: provisioning.JobActionPull, + }, + } + + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{job}, nil) + store.On("Complete", mock.Anything, mock.Anything).Return(nil) + historyWriter.On("WriteJob", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool { + _, hasClaim := j.Labels[LabelJobClaim] + _, hasOther := j.Labels["other-label"] + assert.False(t, hasClaim, "claim label should be removed") + assert.True(t, hasOther, "other labels should be preserved") + return !hasClaim && hasOther + })).Return(nil) + + err := controller.Cleanup(ctx) + + assert.NoError(t, err) + historyWriter.AssertExpectations(t) + }) + + t.Run("processes multiple expired jobs", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + controller := NewJobCleanupController(store, historyWriter, 30*time.Second) + ctx := context.Background() + + jobs := []*provisioning.Job{ + { + ObjectMeta: metav1.ObjectMeta{ + Name: "job-1", + Namespace: "ns-1", + Labels: map[string]string{LabelJobClaim: "111"}, + }, + Spec: provisioning.JobSpec{Repository: "repo-1", Action: provisioning.JobActionPull}, + }, + { + ObjectMeta: metav1.ObjectMeta{ + Name: "job-2", + Namespace: "ns-2", + Labels: map[string]string{LabelJobClaim: "222"}, + }, + Spec: provisioning.JobSpec{Repository: "repo-2", Action: provisioning.JobActionPush}, + }, + { + ObjectMeta: metav1.ObjectMeta{ + Name: "job-3", + Namespace: "ns-3", + Labels: map[string]string{LabelJobClaim: "333"}, + }, + Spec: provisioning.JobSpec{Repository: "repo-3", Action: provisioning.JobActionMigrate}, + }, + } + + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return(jobs, nil) + for _, job := range jobs { + store.On("Complete", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool { + return j.Name == job.Name + })).Return(nil) + historyWriter.On("WriteJob", mock.Anything, mock.MatchedBy(func(j *provisioning.Job) bool { + return j.Name == job.Name + })).Return(nil) + } + + err := controller.Cleanup(ctx) + + assert.NoError(t, err) + store.AssertExpectations(t) + historyWriter.AssertExpectations(t) + // Verify all 3 jobs were processed + store.AssertNumberOfCalls(t, "Complete", 3) + historyWriter.AssertNumberOfCalls(t, "WriteJob", 3) + }) +} + +func TestJobCleanupController_Run(t *testing.T) { + t.Run("runs cleanup on start and periodically", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + // Use short expiry to get short cleanup interval for testing + controller := NewJobCleanupController(store, historyWriter, 10*time.Second) + // Override to even shorter for test + controller.cleanupInterval = 50 * time.Millisecond + + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + + // Expect initial cleanup + periodic cleanups (at least 2, maybe more depending on timing) + callCount := 0 + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100). + Return([]*provisioning.Job{}, nil). + Run(func(args mock.Arguments) { + callCount++ + }). + Maybe() // Allow variable number of calls due to timing + + err := controller.Run(ctx) + + // Should return context.DeadlineExceeded when context times out + require.Error(t, err) + assert.Equal(t, context.DeadlineExceeded, err) + + // Verify cleanup was called at least 3 times (initial + 2 periodic) + assert.GreaterOrEqual(t, callCount, 3, "should have run cleanup at least 3 times") + }) + + t.Run("stops when context is cancelled", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + controller := NewJobCleanupController(store, historyWriter, 30*time.Second) + controller.cleanupInterval = 1 * time.Second + + ctx, cancel := context.WithCancel(context.Background()) + + // Expect initial cleanup + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100).Return([]*provisioning.Job{}, nil).Once() + + // Cancel after initial cleanup + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + + err := controller.Run(ctx) + + assert.Error(t, err) + assert.Equal(t, context.Canceled, err) + store.AssertExpectations(t) + }) + + t.Run("continues running after cleanup error", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + controller := NewJobCleanupController(store, historyWriter, 10*time.Second) + controller.cleanupInterval = 50 * time.Millisecond + + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + + // Track successful calls after initial failure + successCount := 0 + // First cleanup fails + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100). + Return(nil, errors.New("first failure")).Once() + // Subsequent cleanups succeed + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100). + Run(func(args mock.Arguments) { + successCount++ + }). + Return([]*provisioning.Job{}, nil). + Maybe() + + err := controller.Run(ctx) + + // Should still run and return context error + require.Error(t, err) + assert.Equal(t, context.DeadlineExceeded, err) + // Verify it was called successfully at least once after first failure + assert.GreaterOrEqual(t, successCount, 1, "should have retried after first failure") + }) + + t.Run("logs error when periodic cleanup fails", func(t *testing.T) { + store := &MockStore{} + + historyWriter := &MockHistoryWriter{} + + controller := NewJobCleanupController(store, historyWriter, 10*time.Second) + controller.cleanupInterval = 50 * time.Millisecond + + ctx, cancel := context.WithTimeout(context.Background(), 125*time.Millisecond) + defer cancel() + + // Initial cleanup succeeds + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100). + Return([]*provisioning.Job{}, nil).Once() + + // First periodic cleanup fails (this tests the error logging in ticker case) + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100). + Return(nil, errors.New("periodic failure")).Once() + + // Subsequent cleanups succeed + store.On("ListExpiredJobs", mock.Anything, mock.Anything, 100). + Return([]*provisioning.Job{}, nil). + Maybe() + + err := controller.Run(ctx) + + // Should still run and return context error, not the cleanup error + require.Error(t, err) + assert.Equal(t, context.DeadlineExceeded, err) + store.AssertExpectations(t) + }) +} diff --git a/pkg/registry/apis/provisioning/jobs/history_reader_mock.go b/pkg/registry/apis/provisioning/jobs/history_reader_mock.go new file mode 100644 index 00000000000..284c446192e --- /dev/null +++ b/pkg/registry/apis/provisioning/jobs/history_reader_mock.go @@ -0,0 +1,158 @@ +// Code generated by mockery v2.53.4. DO NOT EDIT. + +package jobs + +import ( + context "context" + + v0alpha1 "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" + mock "github.com/stretchr/testify/mock" +) + +// MockHistoryReader is an autogenerated mock type for the HistoryReader type +type MockHistoryReader struct { + mock.Mock +} + +type MockHistoryReader_Expecter struct { + mock *mock.Mock +} + +func (_m *MockHistoryReader) EXPECT() *MockHistoryReader_Expecter { + return &MockHistoryReader_Expecter{mock: &_m.Mock} +} + +// GetJob provides a mock function with given fields: ctx, namespace, repo, uid +func (_m *MockHistoryReader) GetJob(ctx context.Context, namespace string, repo string, uid string) (*v0alpha1.Job, error) { + ret := _m.Called(ctx, namespace, repo, uid) + + if len(ret) == 0 { + panic("no return value specified for GetJob") + } + + var r0 *v0alpha1.Job + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, string, string, string) (*v0alpha1.Job, error)); ok { + return rf(ctx, namespace, repo, uid) + } + if rf, ok := ret.Get(0).(func(context.Context, string, string, string) *v0alpha1.Job); ok { + r0 = rf(ctx, namespace, repo, uid) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*v0alpha1.Job) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, string, string, string) error); ok { + r1 = rf(ctx, namespace, repo, uid) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// MockHistoryReader_GetJob_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetJob' +type MockHistoryReader_GetJob_Call struct { + *mock.Call +} + +// GetJob is a helper method to define mock.On call +// - ctx context.Context +// - namespace string +// - repo string +// - uid string +func (_e *MockHistoryReader_Expecter) GetJob(ctx interface{}, namespace interface{}, repo interface{}, uid interface{}) *MockHistoryReader_GetJob_Call { + return &MockHistoryReader_GetJob_Call{Call: _e.mock.On("GetJob", ctx, namespace, repo, uid)} +} + +func (_c *MockHistoryReader_GetJob_Call) Run(run func(ctx context.Context, namespace string, repo string, uid string)) *MockHistoryReader_GetJob_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(string), args[2].(string), args[3].(string)) + }) + return _c +} + +func (_c *MockHistoryReader_GetJob_Call) Return(_a0 *v0alpha1.Job, _a1 error) *MockHistoryReader_GetJob_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *MockHistoryReader_GetJob_Call) RunAndReturn(run func(context.Context, string, string, string) (*v0alpha1.Job, error)) *MockHistoryReader_GetJob_Call { + _c.Call.Return(run) + return _c +} + +// RecentJobs provides a mock function with given fields: ctx, namespace, repo +func (_m *MockHistoryReader) RecentJobs(ctx context.Context, namespace string, repo string) (*v0alpha1.JobList, error) { + ret := _m.Called(ctx, namespace, repo) + + if len(ret) == 0 { + panic("no return value specified for RecentJobs") + } + + var r0 *v0alpha1.JobList + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, string, string) (*v0alpha1.JobList, error)); ok { + return rf(ctx, namespace, repo) + } + if rf, ok := ret.Get(0).(func(context.Context, string, string) *v0alpha1.JobList); ok { + r0 = rf(ctx, namespace, repo) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*v0alpha1.JobList) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, string, string) error); ok { + r1 = rf(ctx, namespace, repo) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// MockHistoryReader_RecentJobs_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RecentJobs' +type MockHistoryReader_RecentJobs_Call struct { + *mock.Call +} + +// RecentJobs is a helper method to define mock.On call +// - ctx context.Context +// - namespace string +// - repo string +func (_e *MockHistoryReader_Expecter) RecentJobs(ctx interface{}, namespace interface{}, repo interface{}) *MockHistoryReader_RecentJobs_Call { + return &MockHistoryReader_RecentJobs_Call{Call: _e.mock.On("RecentJobs", ctx, namespace, repo)} +} + +func (_c *MockHistoryReader_RecentJobs_Call) Run(run func(ctx context.Context, namespace string, repo string)) *MockHistoryReader_RecentJobs_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(string), args[2].(string)) + }) + return _c +} + +func (_c *MockHistoryReader_RecentJobs_Call) Return(_a0 *v0alpha1.JobList, _a1 error) *MockHistoryReader_RecentJobs_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *MockHistoryReader_RecentJobs_Call) RunAndReturn(run func(context.Context, string, string) (*v0alpha1.JobList, error)) *MockHistoryReader_RecentJobs_Call { + _c.Call.Return(run) + return _c +} + +// NewMockHistoryReader creates a new instance of MockHistoryReader. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. +// The first argument is typically a *testing.T value. +func NewMockHistoryReader(t interface { + mock.TestingT + Cleanup(func()) +}) *MockHistoryReader { + mock := &MockHistoryReader{} + mock.Mock.Test(t) + + t.Cleanup(func() { mock.AssertExpectations(t) }) + + return mock +} diff --git a/pkg/registry/apis/provisioning/jobs/history_writer_mock.go b/pkg/registry/apis/provisioning/jobs/history_writer_mock.go new file mode 100644 index 00000000000..8ec0e63018e --- /dev/null +++ b/pkg/registry/apis/provisioning/jobs/history_writer_mock.go @@ -0,0 +1,84 @@ +// Code generated by mockery v2.53.4. DO NOT EDIT. + +package jobs + +import ( + context "context" + + v0alpha1 "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" + mock "github.com/stretchr/testify/mock" +) + +// MockHistoryWriter is an autogenerated mock type for the HistoryWriter type +type MockHistoryWriter struct { + mock.Mock +} + +type MockHistoryWriter_Expecter struct { + mock *mock.Mock +} + +func (_m *MockHistoryWriter) EXPECT() *MockHistoryWriter_Expecter { + return &MockHistoryWriter_Expecter{mock: &_m.Mock} +} + +// WriteJob provides a mock function with given fields: ctx, job +func (_m *MockHistoryWriter) WriteJob(ctx context.Context, job *v0alpha1.Job) error { + ret := _m.Called(ctx, job) + + if len(ret) == 0 { + panic("no return value specified for WriteJob") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, *v0alpha1.Job) error); ok { + r0 = rf(ctx, job) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// MockHistoryWriter_WriteJob_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'WriteJob' +type MockHistoryWriter_WriteJob_Call struct { + *mock.Call +} + +// WriteJob is a helper method to define mock.On call +// - ctx context.Context +// - job *v0alpha1.Job +func (_e *MockHistoryWriter_Expecter) WriteJob(ctx interface{}, job interface{}) *MockHistoryWriter_WriteJob_Call { + return &MockHistoryWriter_WriteJob_Call{Call: _e.mock.On("WriteJob", ctx, job)} +} + +func (_c *MockHistoryWriter_WriteJob_Call) Run(run func(ctx context.Context, job *v0alpha1.Job)) *MockHistoryWriter_WriteJob_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(*v0alpha1.Job)) + }) + return _c +} + +func (_c *MockHistoryWriter_WriteJob_Call) Return(_a0 error) *MockHistoryWriter_WriteJob_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *MockHistoryWriter_WriteJob_Call) RunAndReturn(run func(context.Context, *v0alpha1.Job) error) *MockHistoryWriter_WriteJob_Call { + _c.Call.Return(run) + return _c +} + +// NewMockHistoryWriter creates a new instance of MockHistoryWriter. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations. +// The first argument is typically a *testing.T value. +func NewMockHistoryWriter(t interface { + mock.TestingT + Cleanup(func()) +}) *MockHistoryWriter { + mock := &MockHistoryWriter{} + mock.Mock.Test(t) + + t.Cleanup(func() { mock.AssertExpectations(t) }) + + return mock +} diff --git a/pkg/registry/apis/provisioning/jobs/loki_client_mock.go b/pkg/registry/apis/provisioning/jobs/loki_client_mock.go index 9043c329875..557350d2661 100644 --- a/pkg/registry/apis/provisioning/jobs/loki_client_mock.go +++ b/pkg/registry/apis/provisioning/jobs/loki_client_mock.go @@ -134,8 +134,7 @@ func (_c *MockLokiClient_RangeQuery_Call) RunAndReturn(run func(context.Context, func NewMockLokiClient(t interface { mock.TestingT Cleanup(func()) -}, -) *MockLokiClient { +}) *MockLokiClient { mock := &MockLokiClient{} mock.Mock.Test(t) diff --git a/pkg/registry/apis/provisioning/jobs/persistentstore.go b/pkg/registry/apis/provisioning/jobs/persistentstore.go index b7c7e47bd35..8b164320bd3 100644 --- a/pkg/registry/apis/provisioning/jobs/persistentstore.go +++ b/pkg/registry/apis/provisioning/jobs/persistentstore.go @@ -306,11 +306,11 @@ func (s *persistentStore) Complete(ctx context.Context, job *provisioning.Job) e return apifmt.Errorf("failed to get provisioning identity for '%s': %w", job.GetNamespace(), err) } - // We need to delete the job from the job store and create it in the historic job store. - // We are fine with the job being lost if the historic job store fails to create it. + // Delete the job from the active job store. + // Callers are responsible for writing the job to history after calling this. // // We will assume that the caller is the claimant. If this is not true, an error is returned. - // This is a best-effort operation; if the job is not in the claimed state, we will still attempt to move it to the historic job store. + // This is a best-effort operation; if the job is not in the claimed state, we will still attempt to delete it. err = s.client.Jobs(job.GetNamespace()).Delete(ctx, job.GetName(), metav1.DeleteOptions{}) if err != nil { span.RecordError(err) @@ -329,6 +329,56 @@ func (s *persistentStore) Complete(ctx context.Context, job *provisioning.Job) e return nil } +// ListExpiredJobs lists jobs with expired leases (claim timestamp older than the given time). +// Returns jobs in batches up to the specified limit. +func (s *persistentStore) ListExpiredJobs(ctx context.Context, expiredBefore time.Time, limit int) ([]*provisioning.Job, error) { + ctx, span := tracing.Start(ctx, "provisioning.jobs.list_expired_jobs") + defer span.End() + + logger := logging.FromContext(ctx).With("operation", "list_expired_jobs") + + // Set up provisioning identity to access jobs across all namespaces + ctx, _, err := identity.WithProvisioningIdentity(ctx, "*") + if err != nil { + span.RecordError(err) + return nil, apifmt.Errorf("failed to grant provisioning identity for listing expired jobs: %w", err) + } + + // Find jobs with expired leases (older than expiredBefore) + expiry := expiredBefore.UnixMilli() + logger.Debug("searching for expired jobs", "expiry_threshold", expiredBefore.Format(time.RFC3339)) + + requirement, err := labels.NewRequirement(LabelJobClaim, selection.LessThan, []string{strconv.FormatInt(expiry, 10)}) + if err != nil { + span.RecordError(err) + return nil, apifmt.Errorf("could not create requirement: %w", err) + } + + span.SetAttributes( + attribute.String("expiry_threshold", expiredBefore.Format(time.RFC3339)), + attribute.Int("limit", limit), + ) + + jobList, err := s.client.Jobs("").List(ctx, metav1.ListOptions{ + LabelSelector: labels.NewSelector().Add(*requirement).String(), + Limit: int64(limit), + }) + if err != nil { + span.RecordError(err) + return nil, apifmt.Errorf("failed to list jobs with expired leases: %w", err) + } + + result := make([]*provisioning.Job, len(jobList.Items)) + for i := range jobList.Items { + result[i] = &jobList.Items[i] + } + + span.SetAttributes(attribute.Int("jobs_found", len(result))) + logger.Debug("found expired jobs", "count", len(result)) + + return result, nil +} + // RenewLease renews the lease for a claimed job, extending its expiry time. // Returns an error if the lease cannot be renewed (e.g., job was completed or lease expired). func (s *persistentStore) RenewLease(ctx context.Context, job *provisioning.Job) error { @@ -405,164 +455,6 @@ func (s *persistentStore) RenewLease(ctx context.Context, job *provisioning.Job) return nil } -// Cleanup finds jobs with expired leases and marks them as failed. -// This replaces the old cleanup mechanism and should be called more frequently. -func (s *persistentStore) Cleanup(ctx context.Context) error { - ctx, span := tracing.Start(ctx, "provisioning.jobs.cleanup") - defer span.End() - - startTime := s.clock() - logger := logging.FromContext(ctx).With("operation", "cleanup") - - // List expired jobs - jobs, err := s.listExpiredJobs(ctx) - if err != nil { - span.RecordError(err) - return err - } - - // If no jobs found, cleanup is complete - if len(jobs) == 0 { - duration := s.clock().Sub(startTime) - logger.Info("cleanup complete - no expired jobs found", "duration", duration) - span.SetAttributes( - attribute.Int("count", 0), - attribute.Int64("duration_ms", duration.Milliseconds()), - ) - return nil - } - - logger.Info("found expired jobs", "count", len(jobs)) - - // Clean up each expired job - for _, job := range jobs { - if err := s.cleanUpExpiredJob(ctx, job); err != nil { - span.RecordError(err) - return err - } - } - - duration := s.clock().Sub(startTime) - logger.Info("cleanup complete", - "duration", duration, - "count", len(jobs), - ) - - span.SetAttributes( - attribute.Int("count", len(jobs)), - attribute.Int64("duration_ms", duration.Milliseconds()), - ) - - return nil -} - -// listExpiredJobs returns jobs with expired leases. -func (s *persistentStore) listExpiredJobs(ctx context.Context) ([]provisioning.Job, error) { - logger := logging.FromContext(ctx) - - // Set up provisioning identity to access jobs across all namespaces - ctx, _, err := identity.WithProvisioningIdentity(ctx, "*") // "*" grants access to all namespaces - if err != nil { - return nil, apifmt.Errorf("failed to grant provisioning identity for cleanup: %w", err) - } - - // Find jobs with expired leases (older than expiry time) - expiry := s.clock().Add(-s.expiry).UnixMilli() - expiryTime := time.UnixMilli(expiry) - logger.Debug("search for expired jobs", "expiry_threshold", expiryTime.Format(time.RFC3339)) - - requirement, err := labels.NewRequirement(LabelJobClaim, selection.LessThan, []string{strconv.FormatInt(expiry, 10)}) - if err != nil { - return nil, apifmt.Errorf("could not create requirement: %w", err) - } - - listCtx, listSpan := tracing.Start(ctx, "provisioning.jobs.cleanup.list_expired_jobs") - defer listSpan.End() - - listSpan.SetAttributes( - attribute.String("expiry_threshold", expiryTime.Format(time.RFC3339)), - attribute.Int64("expiry_duration_seconds", int64(s.expiry.Seconds())), - ) - - timeoutCtx, cancel := context.WithTimeout(listCtx, 5*time.Second) - defer cancel() - - jobList, err := s.client.Jobs("").List(timeoutCtx, metav1.ListOptions{ - LabelSelector: labels.NewSelector().Add(*requirement).String(), - Limit: 100, // Process in batches - }) - if err != nil { - listSpan.RecordError(err) - return nil, apifmt.Errorf("failed to list jobs with expired leases: %w", err) - } - - listSpan.SetAttributes(attribute.Int("jobs_found", len(jobList.Items))) - return jobList.Items, nil -} - -// cleanUpExpiredJob marks a single expired job as failed and archives it. -func (s *persistentStore) cleanUpExpiredJob(ctx context.Context, job provisioning.Job) error { - // Calculate how long the job has been expired - var expiredFor time.Duration - var claimTimestamp time.Time - if claimTime, exists := job.Labels[LabelJobClaim]; exists { - claimMillis, parseErr := strconv.ParseInt(claimTime, 10, 64) - if parseErr == nil { - claimTimestamp = time.UnixMilli(claimMillis) - expiredFor = s.clock().Sub(claimTimestamp) - } - } - - logger := logging.FromContext(ctx).With( - "job", job.GetName(), - "namespace", job.GetNamespace(), - "repository", job.Spec.Repository, - "action", job.Spec.Action, - "expired_for", expiredFor, - ) - - if !claimTimestamp.IsZero() { - logger = logger.With("claim_time", claimTimestamp.Format(time.RFC3339)) - } - - jobCtx, jobSpan := tracing.Start(ctx, "provisioning.jobs.cleanup.complete_expired_job") - defer jobSpan.End() - - jobSpan.SetAttributes( - attribute.String("job.name", job.GetName()), - attribute.String("job.namespace", job.GetNamespace()), - attribute.String("job.repository", job.Spec.Repository), - attribute.String("job.action", string(job.Spec.Action)), - attribute.String("job.expired_for", expiredFor.String()), - ) - - // Mark job as failed due to lease expiry and archive it - jobCopy := job.DeepCopy() - jobCopy.Status.State = provisioning.JobStateError - jobCopy.Status.Message = "Job failed due to lease expiry - worker may have crashed or lost connection" - - // Set namespace context for the completion - jobCtx, _, err := identity.WithProvisioningIdentity(jobCtx, job.GetNamespace()) - if err != nil { - jobSpan.RecordError(err) - return apifmt.Errorf("failed to get provisioning identity for '%s': %w", job.GetNamespace(), err) - } - - // Use Complete to properly archive the failed job - if err := s.Complete(jobCtx, jobCopy); err != nil { - if apierrors.IsNotFound(err) { - // Job was already completed/deleted by another process - this is expected - logger.Warn("job already completed or deleted by another process") - return nil - } - jobSpan.RecordError(err) - return apifmt.Errorf("failed to complete expired job '%s' in '%s': %w", job.GetName(), job.GetNamespace(), err) - } - - logger.Info("clean up expired job complete") - return nil -} - func (s *persistentStore) Insert(ctx context.Context, namespace string, spec provisioning.JobSpec) (*provisioning.Job, error) { ctx, span := tracing.Start(ctx, "provisioning.jobs.insert") defer span.End() diff --git a/pkg/registry/apis/provisioning/jobs/store_mock.go b/pkg/registry/apis/provisioning/jobs/store_mock.go index aebcac9bbbe..99a1945c6a8 100644 --- a/pkg/registry/apis/provisioning/jobs/store_mock.go +++ b/pkg/registry/apis/provisioning/jobs/store_mock.go @@ -1,12 +1,14 @@ -// Code generated by mockery v2.52.4. DO NOT EDIT. +// Code generated by mockery v2.53.4. DO NOT EDIT. package jobs import ( context "context" + time "time" + + mock "github.com/stretchr/testify/mock" v0alpha1 "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" - mock "github.com/stretchr/testify/mock" ) // MockStore is an autogenerated mock type for the Store type @@ -89,52 +91,6 @@ func (_c *MockStore_Claim_Call) RunAndReturn(run func(context.Context) (*v0alpha return _c } -// Cleanup provides a mock function with given fields: ctx -func (_m *MockStore) Cleanup(ctx context.Context) error { - ret := _m.Called(ctx) - - if len(ret) == 0 { - panic("no return value specified for Cleanup") - } - - var r0 error - if rf, ok := ret.Get(0).(func(context.Context) error); ok { - r0 = rf(ctx) - } else { - r0 = ret.Error(0) - } - - return r0 -} - -// MockStore_Cleanup_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Cleanup' -type MockStore_Cleanup_Call struct { - *mock.Call -} - -// Cleanup is a helper method to define mock.On call -// - ctx context.Context -func (_e *MockStore_Expecter) Cleanup(ctx interface{}) *MockStore_Cleanup_Call { - return &MockStore_Cleanup_Call{Call: _e.mock.On("Cleanup", ctx)} -} - -func (_c *MockStore_Cleanup_Call) Run(run func(ctx context.Context)) *MockStore_Cleanup_Call { - _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context)) - }) - return _c -} - -func (_c *MockStore_Cleanup_Call) Return(_a0 error) *MockStore_Cleanup_Call { - _c.Call.Return(_a0) - return _c -} - -func (_c *MockStore_Cleanup_Call) RunAndReturn(run func(context.Context) error) *MockStore_Cleanup_Call { - _c.Call.Return(run) - return _c -} - // Complete provides a mock function with given fields: ctx, job func (_m *MockStore) Complete(ctx context.Context, job *v0alpha1.Job) error { ret := _m.Called(ctx, job) @@ -182,9 +138,9 @@ func (_c *MockStore_Complete_Call) RunAndReturn(run func(context.Context, *v0alp return _c } -// Get provides a mock function with given fields: ctx, name -func (_m *MockStore) Get(ctx context.Context, name string) (*v0alpha1.Job, error) { - ret := _m.Called(ctx, name) +// Get provides a mock function with given fields: ctx, namespace, name +func (_m *MockStore) Get(ctx context.Context, namespace string, name string) (*v0alpha1.Job, error) { + ret := _m.Called(ctx, namespace, name) if len(ret) == 0 { panic("no return value specified for Get") @@ -192,19 +148,19 @@ func (_m *MockStore) Get(ctx context.Context, name string) (*v0alpha1.Job, error var r0 *v0alpha1.Job var r1 error - if rf, ok := ret.Get(0).(func(context.Context, string) (*v0alpha1.Job, error)); ok { - return rf(ctx, name) + if rf, ok := ret.Get(0).(func(context.Context, string, string) (*v0alpha1.Job, error)); ok { + return rf(ctx, namespace, name) } - if rf, ok := ret.Get(0).(func(context.Context, string) *v0alpha1.Job); ok { - r0 = rf(ctx, name) + if rf, ok := ret.Get(0).(func(context.Context, string, string) *v0alpha1.Job); ok { + r0 = rf(ctx, namespace, name) } else { if ret.Get(0) != nil { r0 = ret.Get(0).(*v0alpha1.Job) } } - if rf, ok := ret.Get(1).(func(context.Context, string) error); ok { - r1 = rf(ctx, name) + if rf, ok := ret.Get(1).(func(context.Context, string, string) error); ok { + r1 = rf(ctx, namespace, name) } else { r1 = ret.Error(1) } @@ -219,14 +175,15 @@ type MockStore_Get_Call struct { // Get is a helper method to define mock.On call // - ctx context.Context +// - namespace string // - name string -func (_e *MockStore_Expecter) Get(ctx interface{}, name interface{}) *MockStore_Get_Call { - return &MockStore_Get_Call{Call: _e.mock.On("Get", ctx, name)} +func (_e *MockStore_Expecter) Get(ctx interface{}, namespace interface{}, name interface{}) *MockStore_Get_Call { + return &MockStore_Get_Call{Call: _e.mock.On("Get", ctx, namespace, name)} } -func (_c *MockStore_Get_Call) Run(run func(ctx context.Context, name string)) *MockStore_Get_Call { +func (_c *MockStore_Get_Call) Run(run func(ctx context.Context, namespace string, name string)) *MockStore_Get_Call { _c.Call.Run(func(args mock.Arguments) { - run(args[0].(context.Context), args[1].(string)) + run(args[0].(context.Context), args[1].(string), args[2].(string)) }) return _c } @@ -236,7 +193,67 @@ func (_c *MockStore_Get_Call) Return(_a0 *v0alpha1.Job, _a1 error) *MockStore_Ge return _c } -func (_c *MockStore_Get_Call) RunAndReturn(run func(context.Context, string) (*v0alpha1.Job, error)) *MockStore_Get_Call { +func (_c *MockStore_Get_Call) RunAndReturn(run func(context.Context, string, string) (*v0alpha1.Job, error)) *MockStore_Get_Call { + _c.Call.Return(run) + return _c +} + +// ListExpiredJobs provides a mock function with given fields: ctx, expiredBefore, limit +func (_m *MockStore) ListExpiredJobs(ctx context.Context, expiredBefore time.Time, limit int) ([]*v0alpha1.Job, error) { + ret := _m.Called(ctx, expiredBefore, limit) + + if len(ret) == 0 { + panic("no return value specified for ListExpiredJobs") + } + + var r0 []*v0alpha1.Job + var r1 error + if rf, ok := ret.Get(0).(func(context.Context, time.Time, int) ([]*v0alpha1.Job, error)); ok { + return rf(ctx, expiredBefore, limit) + } + if rf, ok := ret.Get(0).(func(context.Context, time.Time, int) []*v0alpha1.Job); ok { + r0 = rf(ctx, expiredBefore, limit) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]*v0alpha1.Job) + } + } + + if rf, ok := ret.Get(1).(func(context.Context, time.Time, int) error); ok { + r1 = rf(ctx, expiredBefore, limit) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// MockStore_ListExpiredJobs_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'ListExpiredJobs' +type MockStore_ListExpiredJobs_Call struct { + *mock.Call +} + +// ListExpiredJobs is a helper method to define mock.On call +// - ctx context.Context +// - expiredBefore time.Time +// - limit int +func (_e *MockStore_Expecter) ListExpiredJobs(ctx interface{}, expiredBefore interface{}, limit interface{}) *MockStore_ListExpiredJobs_Call { + return &MockStore_ListExpiredJobs_Call{Call: _e.mock.On("ListExpiredJobs", ctx, expiredBefore, limit)} +} + +func (_c *MockStore_ListExpiredJobs_Call) Run(run func(ctx context.Context, expiredBefore time.Time, limit int)) *MockStore_ListExpiredJobs_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(time.Time), args[2].(int)) + }) + return _c +} + +func (_c *MockStore_ListExpiredJobs_Call) Return(_a0 []*v0alpha1.Job, _a1 error) *MockStore_ListExpiredJobs_Call { + _c.Call.Return(_a0, _a1) + return _c +} + +func (_c *MockStore_ListExpiredJobs_Call) RunAndReturn(run func(context.Context, time.Time, int) ([]*v0alpha1.Job, error)) *MockStore_ListExpiredJobs_Call { _c.Call.Return(run) return _c } diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index e2030ef0818..c30b43eeb60 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -775,13 +775,21 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH } repoGetter := resources.NewRepositoryGetter(b.repoFactory, b.client) + + // Create job cleanup controller + jobExpiry := 30 * time.Second + jobCleanupController := jobs.NewJobCleanupController( + b.jobs, + jobHistoryWriter, + jobExpiry, + ) + // This is basically our own JobQueue system driver, err := jobs.NewConcurrentJobDriver( 3, // 3 drivers for now 20*time.Minute, // Max time for each job - time.Minute, // Cleanup jobs 30*time.Second, // Periodically look for new jobs - 30*time.Second, // Lease renewal interval + jobExpiry, // Lease renewal interval b.jobs, repoGetter, jobHistoryWriter, jobController.InsertNotifications(), b.registry, @@ -797,6 +805,12 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH } }() + go func() { + if err := jobCleanupController.Run(postStartHookCtx.Context); err != nil { + logging.FromContext(postStartHookCtx.Context).Error("job cleanup controller failed", "error", err) + } + }() + repoController, err := controller.NewRepositoryController( b.GetClient(), repoInformer, @@ -821,7 +835,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH if b.jobHistoryLoki == nil { // Create HistoryJobController for cleanup of old job history entries // Separate informer factory for HistoryJob cleanup with resync interval - historyJobExpiration := 30 * time.Second + historyJobExpiration := 10 * time.Minute historyJobInformerFactory := informers.NewSharedInformerFactory(c, historyJobExpiration) historyJobInformer := historyJobInformerFactory.Provisioning().V0alpha1().HistoricJobs() go historyJobInformer.Informer().Run(postStartHookCtx.Done())