diff --git a/pkg/registry/apis/provisioning/controller/job.go b/pkg/registry/apis/provisioning/controller/job.go new file mode 100644 index 00000000000..c3af398d2c1 --- /dev/null +++ b/pkg/registry/apis/provisioning/controller/job.go @@ -0,0 +1,58 @@ +package controller + +import ( + "k8s.io/client-go/tools/cache" + + "github.com/grafana/grafana-app-sdk/logging" + informer "github.com/grafana/grafana/apps/provisioning/pkg/generated/informers/externalversions/provisioning/v0alpha1" +) + +const ( + jobControllerLoggerName = "provisioning-job-controller" +) + +// JobController manages job create notifications. +type JobController struct { + jobSynced cache.InformerSynced + logger logging.Logger + + // notification channel for job create events (replaces InsertNotifications) + notifications chan struct{} +} + +// NewJobController creates a new JobController. +func NewJobController( + jobInformer informer.JobInformer, +) (*JobController, error) { + jc := &JobController{ + jobSynced: jobInformer.Informer().HasSynced, + logger: logging.DefaultLogger.With("logger", jobControllerLoggerName), + notifications: make(chan struct{}, 1), + } + + _, err := jobInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { + // Send notification for job create events (replaces InsertNotifications) + jc.sendNotification() + }, + }) + if err != nil { + return nil, err + } + + return jc, nil +} + +// InsertNotifications returns a channel that receives notifications when jobs are created. +// This replaces the InsertNotifications method from persistentstore.go. +func (jc *JobController) InsertNotifications() chan struct{} { + return jc.notifications +} + +func (jc *JobController) sendNotification() { + select { + case jc.notifications <- struct{}{}: + default: + // Don't block if there's already a notification waiting + } +} diff --git a/pkg/registry/apis/provisioning/controller/job_test.go b/pkg/registry/apis/provisioning/controller/job_test.go new file mode 100644 index 00000000000..8c84abefb0f --- /dev/null +++ b/pkg/registry/apis/provisioning/controller/job_test.go @@ -0,0 +1,85 @@ +package controller + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" + provisioningfake "github.com/grafana/grafana/apps/provisioning/pkg/generated/clientset/versioned/fake" + provisioninginformers "github.com/grafana/grafana/apps/provisioning/pkg/generated/informers/externalversions" +) + +func TestJobController_New(t *testing.T) { + client := provisioningfake.NewSimpleClientset() + informerFactory := provisioninginformers.NewSharedInformerFactory(client, time.Second) + jobInformer := informerFactory.Provisioning().V0alpha1().Jobs() + + controller, err := NewJobController(jobInformer) + + require.NoError(t, err) + assert.NotNil(t, controller) + assert.NotNil(t, controller.notifications) +} + +func TestJobController_InsertNotifications(t *testing.T) { + client := provisioningfake.NewSimpleClientset() + informerFactory := provisioninginformers.NewSharedInformerFactory(client, time.Second) + jobInformer := informerFactory.Provisioning().V0alpha1().Jobs() + + controller, err := NewJobController(jobInformer) + require.NoError(t, err) + + notifications := controller.InsertNotifications() + assert.NotNil(t, notifications) + + // Test that notification is sent + controller.sendNotification() + + select { + case <-notifications: + // Success - notification received + case <-time.After(time.Second): + t.Fatal("Expected notification but didn't receive one") + } +} + +func TestJobController_NotificationOnJobCreate(t *testing.T) { + client := provisioningfake.NewSimpleClientset() + informerFactory := provisioninginformers.NewSharedInformerFactory(client, time.Second) + jobInformer := informerFactory.Provisioning().V0alpha1().Jobs() + + controller, err := NewJobController(jobInformer) + require.NoError(t, err) + + // Start informer and wait for cache sync + ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) + defer cancel() + + informerFactory.Start(ctx.Done()) + informerFactory.WaitForCacheSync(ctx.Done()) + + // Get notifications channel + notifications := controller.InsertNotifications() + + // Create a job - this should trigger a notification + _, err = client.ProvisioningV0alpha1().Jobs("default").Create(ctx, &provisioning.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-job", + Namespace: "default", + }, + }, metav1.CreateOptions{}) + require.NoError(t, err) + + // Wait for notification + select { + case <-notifications: + // Success - notification received + case <-time.After(time.Second * 2): + t.Fatal("Expected notification but didn't receive one") + } +} diff --git a/pkg/registry/apis/provisioning/jobs/concurrent_driver.go b/pkg/registry/apis/provisioning/jobs/concurrent_driver.go index 23935113a36..4f508c25a8d 100644 --- a/pkg/registry/apis/provisioning/jobs/concurrent_driver.go +++ b/pkg/registry/apis/provisioning/jobs/concurrent_driver.go @@ -20,6 +20,7 @@ type ConcurrentJobDriver struct { repoGetter RepoGetter historicJobs History workers []Worker + notifications chan struct{} } // NewConcurrentJobDriver creates a new concurrent job driver that spawns multiple job drivers. @@ -29,6 +30,7 @@ func NewConcurrentJobDriver( store Store, repoGetter RepoGetter, historicJobs History, + notifications chan struct{}, workers ...Worker, ) (*ConcurrentJobDriver, error) { if numDrivers <= 0 { @@ -63,6 +65,7 @@ func NewConcurrentJobDriver( repoGetter: repoGetter, historicJobs: historicJobs, workers: workers, + notifications: notifications, }, nil } @@ -117,6 +120,7 @@ func (c *ConcurrentJobDriver) Run(ctx context.Context) error { c.store, c.repoGetter, c.historicJobs, + c.notifications, c.workers..., ) if err != nil { diff --git a/pkg/registry/apis/provisioning/jobs/driver.go b/pkg/registry/apis/provisioning/jobs/driver.go index a348d9988b0..f64eebe3528 100644 --- a/pkg/registry/apis/provisioning/jobs/driver.go +++ b/pkg/registry/apis/provisioning/jobs/driver.go @@ -36,10 +36,6 @@ type Store interface { // 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 - // InsertNotifications returns a channel that will have a value sent to it when a new job is inserted. - // This is used to wake up the job driver when a new job is inserted. - InsertNotifications() chan struct{} - // Update saves the job back to the store. Update(ctx context.Context, job *provisioning.Job) (*provisioning.Job, error) @@ -77,6 +73,9 @@ type jobDriver struct { // Workers process the job. // Only the first worker who supports the job will process it; the rest are ignored. workers []Worker + + // notifications channel for job create events + notifications chan struct{} } func NewJobDriver( @@ -84,6 +83,7 @@ func NewJobDriver( store Store, repoGetter RepoGetter, historicJobs History, + notifications chan struct{}, workers ...Worker, ) (*jobDriver, error) { return &jobDriver{ @@ -94,6 +94,7 @@ func NewJobDriver( repoGetter: repoGetter, historicJobs: historicJobs, workers: workers, + notifications: notifications, }, nil } @@ -120,7 +121,7 @@ func (d *jobDriver) Run(ctx context.Context) error { return ctx.Err() case <-jobTicker.C: d.processJobsUntilDoneOrError(ctx) - case <-d.store.InsertNotifications(): + case <-d.notifications: d.processJobsUntilDoneOrError(ctx) } } diff --git a/pkg/registry/apis/provisioning/jobs/persistentstore.go b/pkg/registry/apis/provisioning/jobs/persistentstore.go index 2dfcf216b6b..b25fdff68e2 100644 --- a/pkg/registry/apis/provisioning/jobs/persistentstore.go +++ b/pkg/registry/apis/provisioning/jobs/persistentstore.go @@ -87,28 +87,17 @@ type persistentStore struct { // expiry is the time after which a job is considered abandoned. // If a job is abandoned, it will have its claim cleaned up periodically. expiry time.Duration - - // notifications has a signal sent to it when a new job is inserted. If a value already exists, nothing is sent. - // - // This is very similar to the concept of a Waker in Rust: - notifications chan struct{} } -func NewJobStore( - jobStore jobStorage, - expiry time.Duration, -) (*persistentStore, error) { +func NewJobStore(jobStore jobStorage, expiry time.Duration) (*persistentStore, error) { if expiry <= 0 { expiry = time.Second * 30 } return &persistentStore{ jobStore: jobStore, - - clock: time.Now, - expiry: expiry, - - notifications: make(chan struct{}, 1), + clock: time.Now, + expiry: expiry, }, nil } @@ -446,19 +435,9 @@ func (s *persistentStore) Insert(ctx context.Context, namespace string, spec pro return nil, apifmt.Errorf("unexpected object type %T", obj) } - select { - case s.notifications <- struct{}{}: - default: - // We don't want to block if there is already a notification waiting. - } - return created, nil } -func (s *persistentStore) InsertNotifications() chan struct{} { - return s.notifications -} - // generateJobName creates and updates the job's name to one that fits it. func (s *persistentStore) generateJobName(job *provisioning.Job) { switch job.Spec.Action { diff --git a/pkg/registry/apis/provisioning/jobs/store_mock.go b/pkg/registry/apis/provisioning/jobs/store_mock.go index 114a621561e..79d91719605 100644 --- a/pkg/registry/apis/provisioning/jobs/store_mock.go +++ b/pkg/registry/apis/provisioning/jobs/store_mock.go @@ -1,4 +1,4 @@ -// Code generated by mockery v2.53.4. DO NOT EDIT. +// Code generated by mockery v2.52.4. DO NOT EDIT. package jobs @@ -182,53 +182,6 @@ func (_c *MockStore_Complete_Call) RunAndReturn(run func(context.Context, *v0alp return _c } -// InsertNotifications provides a mock function with no fields -func (_m *MockStore) InsertNotifications() chan struct{} { - ret := _m.Called() - - if len(ret) == 0 { - panic("no return value specified for InsertNotifications") - } - - var r0 chan struct{} - if rf, ok := ret.Get(0).(func() chan struct{}); ok { - r0 = rf() - } else { - if ret.Get(0) != nil { - r0 = ret.Get(0).(chan struct{}) - } - } - - return r0 -} - -// MockStore_InsertNotifications_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'InsertNotifications' -type MockStore_InsertNotifications_Call struct { - *mock.Call -} - -// InsertNotifications is a helper method to define mock.On call -func (_e *MockStore_Expecter) InsertNotifications() *MockStore_InsertNotifications_Call { - return &MockStore_InsertNotifications_Call{Call: _e.mock.On("InsertNotifications")} -} - -func (_c *MockStore_InsertNotifications_Call) Run(run func()) *MockStore_InsertNotifications_Call { - _c.Call.Run(func(args mock.Arguments) { - run() - }) - return _c -} - -func (_c *MockStore_InsertNotifications_Call) Return(_a0 chan struct{}) *MockStore_InsertNotifications_Call { - _c.Call.Return(_a0) - return _c -} - -func (_c *MockStore_InsertNotifications_Call) RunAndReturn(run func() chan struct{}) *MockStore_InsertNotifications_Call { - _c.Call.Return(run) - return _c -} - // Update provides a mock function with given fields: ctx, job func (_m *MockStore) Update(ctx context.Context, job *v0alpha1.Job) (*v0alpha1.Job, error) { ret := _m.Called(ctx, job) diff --git a/pkg/registry/apis/provisioning/register.go b/pkg/registry/apis/provisioning/register.go index 4ad94af2687..ee8ad2dd02e 100644 --- a/pkg/registry/apis/provisioning/register.go +++ b/pkg/registry/apis/provisioning/register.go @@ -603,7 +603,9 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH // Informer with resync interval used for health check and reconciliation sharedInformerFactory := informers.NewSharedInformerFactory(c, 60*time.Second) repoInformer := sharedInformerFactory.Provisioning().V0alpha1().Repositories() + jobInformer := sharedInformerFactory.Provisioning().V0alpha1().Jobs() go repoInformer.Informer().Run(postStartHookCtx.Done()) + go jobInformer.Informer().Run(postStartHookCtx.Done()) b.client = c.ProvisioningV0alpha1() @@ -675,6 +677,12 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH syncWorker, } + // Create JobController to handle job create notifications + jobController, err := controller.NewJobController(jobInformer) + if err != nil { + return err + } + // Add any extra workers for _, extra := range b.extras { workers = append(workers, extra.GetJobWorkers()...) @@ -688,6 +696,7 @@ func (b *APIBuilder) GetPostStartHooks() (map[string]genericapiserver.PostStartH 30*time.Second, // Periodically look for new jobs 30*time.Second, // Lease renewal interval b.jobs, b, b.jobHistory, + jobController.InsertNotifications(), workers..., ) if err != nil {