feat: Add CronController tests

cmd/execution-controller/main.go

@@ -107,6 +107,17 @@ func main() {
 		klog.Fatalf("cannot initialize controller manager: %v", err)
 	}
 
+	// Set up stores.
+	for _, factory := range GetStoreFactories() {
+		klog.Infof("setting up store %v", factory.Name())
+		store, err := factory.New(ctrlContext)
+		if err != nil {
+			klog.Fatalf("cannot initialize store %v: %v", factory.Name(), err)
+		}
+		mgr.AddStore(store)
+		ctrlContext.Stores().Register(store)
+	}
+
 	// Set up controllers.
 	for _, factory := range GetControllerFactories() {
 		concurrencySpec := options.ControllerConcurrency
@@ -121,17 +132,6 @@ func main() {
 		mgr.Add(controller)
 	}
 
-	// Set up stores.
-	for _, factory := range GetStoreFactories() {
-		klog.Infof("setting up store %v", factory.Name())
-		store, err := factory.New(ctrlContext)
-		if err != nil {
-			klog.Fatalf("cannot initialize store %v: %v", factory.Name(), err)
-		}
-		mgr.AddStore(store)
-		ctrlContext.Stores().Register(store)
-	}
-
 	ctx := ctrl.SetupSignalHandler()
 
 	// Start HTTP server in background.

pkg/execution/controllers/croncontroller/control.go

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2022 The Furiko Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package croncontroller
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/pkg/errors"
+	corev1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
+
+	execution "github.com/furiko-io/furiko/apis/execution/v1alpha1"
+	executionv1alpha1 "github.com/furiko-io/furiko/pkg/generated/clientset/versioned/typed/execution/v1alpha1"
+	"github.com/furiko-io/furiko/pkg/utils/logvalues"
+)
+
+type ExecutionControlInterface interface {
+	CreateJob(ctx context.Context, rjc *execution.JobConfig, rj *execution.Job) error
+}
+
+// ExecutionControl is a wrapper around the Execution clientset.
+type ExecutionControl struct {
+	name     string
+	client   executionv1alpha1.ExecutionV1alpha1Interface
+	recorder record.EventRecorder
+}
+
+var _ ExecutionControlInterface = (*ExecutionControl)(nil)
+
+func NewExecutionControl(
+	name string,
+	client executionv1alpha1.ExecutionV1alpha1Interface,
+	recorder record.EventRecorder,
+) *ExecutionControl {
+	return &ExecutionControl{
+		name:     name,
+		client:   client,
+		recorder: recorder,
+	}
+}
+
+func (c *ExecutionControl) CreateJob(ctx context.Context, rjc *execution.JobConfig, rj *execution.Job) error {
+	createdRj, err := c.client.Jobs(rj.GetNamespace()).Create(ctx, rj, metav1.CreateOptions{})
+
+	// If we fail to create the Job due to an invalid error (e.g. from webhook), do
+	// not retry and instead store as an event.
+	if kerrors.IsInvalid(err) {
+		// Get detailed error information if possible.
+		errMessage := err.Error()
+		if err, ok := err.(kerrors.APIStatus); ok {
+			if details := err.Status().Details; details != nil && len(details.Causes) > 0 {
+				causes := make([]string, 0, len(details.Causes))
+				for _, cause := range details.Causes {
+					causes = append(causes, fmt.Sprintf("%s: %s", cause.Field, cause.Message))
+				}
+				errMessage = strings.Join(causes, ", ")
+			}
+		}
+
+		klog.ErrorS(err, "croncontroller: failed to create job", logvalues.
+			Values("worker", c.name, "namespace", rj.GetNamespace(), "name", rj.GetName()).
+			Level(4, "job", rj).
+			Build()...,
+		)
+		c.recorder.Eventf(rjc, corev1.EventTypeWarning, "CronScheduleFailed",
+			"Failed to create Job %v from cron schedule: %v", rj.GetName(), errMessage)
+		return nil
+	}
+
+	if err != nil {
+		return errors.Wrapf(err, "cannot create job")
+	}
+
+	klog.InfoS("croncontroller: created job", logvalues.
+		Values("worker", c.name, "namespace", createdRj.GetNamespace(), "name", createdRj.GetName()).
+		Level(4, "job", createdRj).
+		Build()...,
+	)
+
+	c.recorder.Eventf(rjc, corev1.EventTypeNormal, "CronSchedule",
+		"Created Job %v from cron schedule", createdRj.GetName())
+
+	return nil
+}

pkg/execution/controllers/croncontroller/control_test.go

@@ -0,0 +1,124 @@
+/*
+ * Copyright 2022 The Furiko Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package croncontroller_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/clock"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+	ktesting "k8s.io/client-go/testing"
+	"k8s.io/client-go/tools/record"
+
+	execution "github.com/furiko-io/furiko/apis/execution/v1alpha1"
+	"github.com/furiko-io/furiko/pkg/execution/controllers/croncontroller"
+	"github.com/furiko-io/furiko/pkg/runtime/controllercontext/mock"
+	"github.com/furiko-io/furiko/pkg/utils/ktime"
+)
+
+const (
+	jobNamespace = "test"
+	jobName      = "my-sample-job"
+)
+
+var (
+	fakeJob = &execution.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      jobName,
+			Namespace: jobNamespace,
+		},
+		Spec: execution.JobSpec{
+			Type: execution.JobTypeAdhoc,
+			Template: &execution.JobTemplateSpec{
+				Task: execution.JobTaskSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name:  "container",
+									Image: "hello-world",
+									Args: []string{
+										"echo",
+										"Hello world!",
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+)
+
+func TestExecutionControl(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
+	defer cancel()
+
+	ktime.Clock = clock.NewFakeClock(time.Now())
+	c := mock.NewContext()
+	fakeClient := c.MockClientsets().FurikoMock()
+	client := fakeClient.ExecutionV1alpha1()
+	recorder := record.NewFakeRecorder(1024)
+	control := croncontroller.NewExecutionControl("test", client, recorder)
+	err := c.Start(ctx)
+	assert.NoError(t, err)
+
+	// Create job with client with recorded event
+	assert.Len(t, recorder.Events, 0)
+	err = control.CreateJob(ctx, jobConfigAllow, fakeJob)
+	assert.NoError(t, err)
+	assert.Len(t, recorder.Events, 1)
+
+	// Should be created
+	_, err = client.Jobs(fakeJob.Namespace).Get(ctx, fakeJob.Name, metav1.GetOptions{})
+	assert.NoError(t, err)
+
+	// Creating duplicate Job should throw errpr
+	err = control.CreateJob(ctx, jobConfigAllow, fakeJob)
+	assert.True(t, kerrors.IsAlreadyExists(err))
+
+	// Delete the Job now
+	err = client.Jobs(fakeJob.Namespace).Delete(ctx, fakeJob.Name, metav1.DeleteOptions{})
+	assert.NoError(t, err)
+
+	// Add Reactor to simulate InvalidError
+	fakeClient.PrependReactor("create", "jobs",
+		func(action ktesting.Action) (handled bool, ret runtime.Object, err error) {
+			return true, nil, kerrors.NewInvalid(execution.GVKJob.GroupKind(), fakeJob.Name, field.ErrorList{
+				field.Invalid(field.NewPath("metadata", "name"), fakeJob.Name, "invalid name"),
+			})
+		})
+
+	// Create Job with InvalidError, should not throw error but recorded an event
+	assert.Len(t, recorder.Events, 1)
+	err = control.CreateJob(ctx, jobConfigAllow, fakeJob)
+	assert.NoError(t, err)
+	assert.Len(t, recorder.Events, 2)
+
+	// Should not be created
+	_, err = client.Jobs(fakeJob.Namespace).Get(ctx, fakeJob.Name, metav1.GetOptions{})
+	assert.Error(t, err)
+	assert.True(t, kerrors.IsNotFound(err))
+}

pkg/execution/controllers/croncontroller/controller.go

@@ -21,6 +21,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
@@ -67,30 +68,23 @@ type Context struct {
 	controllercontext.ContextInterface
 	jobInformer       executioninformers.JobInformer
 	jobconfigInformer executioninformers.JobConfigInformer
-	hasSynced         []cache.InformerSynced
+	HasSynced         []cache.InformerSynced
 	queue             workqueue.RateLimitingInterface
 	updatedConfigs    chan *execution.JobConfig
-	recorder          record.EventRecorder
 }
 
+// NewContext returns a new Context.
 func NewContext(context controllercontext.ContextInterface) *Context {
 	c := &Context{ContextInterface: context}
 
-	// Create recorder.
-	eventBroadcaster := record.NewBroadcaster()
-	eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{
-		Interface: c.Clientsets().Kubernetes().CoreV1().Events(""),
-	})
-	c.recorder = eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerName})
-
 	// Create workqueue.
 	ratelimiter := workqueue.DefaultControllerRateLimiter()
 	c.queue = workqueue.NewNamedRateLimitingQueue(ratelimiter, controllerName)
 
 	// Bind informers.
 	c.jobInformer = c.Informers().Furiko().Execution().V1alpha1().Jobs()
 	c.jobconfigInformer = c.Informers().Furiko().Execution().V1alpha1().JobConfigs()
-	c.hasSynced = []cache.InformerSynced{
+	c.HasSynced = []cache.InformerSynced{
 		c.jobInformer.Informer().HasSynced,
 		c.jobconfigInformer.Informer().HasSynced,
 	}
@@ -100,6 +94,15 @@ func NewContext(context controllercontext.ContextInterface) *Context {
 	return c
 }
 
+// NewRecorder returns a new Recorder for the controller.
+func NewRecorder(context controllercontext.ContextInterface) record.EventRecorder {
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{
+		Interface: context.Clientsets().Kubernetes().CoreV1().Events(""),
+	})
+	return eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerName})
+}
+
 func NewController(
 	ctrlContext controllercontext.ContextInterface,
 	concurrency *configv1.Concurrency,
@@ -110,7 +113,18 @@ func NewController(
 
 	ctrl.cronWorker = NewCronWorker(ctrl.Context)
 	ctrl.informerWorker = NewInformerWorker(ctrl.Context)
-	ctrl.reconciler = reconciler.NewController(NewReconciler(ctrl.Context, concurrency), ctrl.queue)
+
+	client := NewExecutionControl(
+		(&Reconciler{}).Name(),
+		ctrlContext.Clientsets().Furiko().ExecutionV1alpha1(),
+		NewRecorder(ctrl.Context),
+	)
+	store, err := ctrlContext.Stores().ActiveJobStore()
+	if err != nil {
+		return nil, errors.Wrapf(err, "cannot load ActiveJobStore")
+	}
+	recon := NewReconciler(ctrl.Context, client, store, concurrency)
+	ctrl.reconciler = reconciler.NewController(recon, ctrl.queue)
 
 	return ctrl, nil
 }
@@ -125,7 +139,7 @@ func (c *Controller) Run(ctx context.Context) error {
 		ctx,
 		controllerName,
 		waitForCacheSyncTimeout,
-		c.hasSynced...,
+		c.HasSynced...,
 	); err != nil {
 		klog.ErrorS(err, "croncontroller: cache sync timeout")
 		return err