diff --git a/internal/controller/job_controller_integration_test.go b/internal/controller/job_controller_integration_test.go new file mode 100644 index 0000000..8606a00 --- /dev/null +++ b/internal/controller/job_controller_integration_test.go @@ -0,0 +1,202 @@ +package controller + +import ( + "context" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + executionv1alpha1 "git.ddupan.top/panxiao81/ayatori/api/execution/v1alpha1" + kubernetesadapter "git.ddupan.top/panxiao81/ayatori/internal/adapter/kubernetes" + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/envtest" + metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" +) + +const ( + integrationNamespace = "controller-integration" + integrationClass = "integration" +) + +//nolint:modernize // Kubernetes API structs expose ObjectMeta through embedded TypeMeta fields. +func TestJobControllerIntegration(t *testing.T) { + if os.Getenv("KUBEBUILDER_ASSETS") == "" { + t.Skip("KUBEBUILDER_ASSETS is unset; run make test to execute controller integration tests") + } + + scheme := runtime.NewScheme() + for _, addToScheme := range []func(*runtime.Scheme) error{ + corev1.AddToScheme, + batchv1.AddToScheme, + executionv1alpha1.AddToScheme, + } { + if err := addToScheme(scheme); err != nil { + t.Fatal(err) + } + } + + crdPath, err := filepath.Abs("../../config/crd/bases") + if err != nil { + t.Fatal(err) + } + environment := &envtest.Environment{CRDDirectoryPaths: []string{crdPath}} + config, err := environment.Start() + if err != nil { + t.Fatalf("start envtest: %v", err) + } + t.Cleanup(func() { + if err := environment.Stop(); err != nil { + t.Errorf("stop envtest: %v", err) + } + }) + + manager, err := ctrl.NewManager(config, ctrl.Options{ + Scheme: scheme, + Metrics: metricsserver.Options{BindAddress: "0"}, + }) + if err != nil { + t.Fatal(err) + } + if err := (&JobReconciler{Client: manager.GetClient()}).SetupWithManager(manager); err != nil { + t.Fatal(err) + } + + managerContext, cancelManager := context.WithCancel(context.Background()) + t.Cleanup(cancelManager) + managerErrors := make(chan error, 1) + go func() { + managerErrors <- manager.Start(managerContext) + }() + if !manager.GetCache().WaitForCacheSync(managerContext) { + t.Fatal("manager cache did not synchronize") + } + + directClient, err := client.New(config, client.Options{Scheme: scheme}) + if err != nil { + t.Fatal(err) + } + ctx := context.Background() + objects := []client.Object{ + &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: integrationNamespace, Labels: map[string]string{testLabelKey: testLabelEnabled}}}, + &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: testSAName, Namespace: integrationNamespace}}, + &executionv1alpha1.KubernetesExecutionParameters{ + ObjectMeta: metav1.ObjectMeta{Name: integrationClass}, + Spec: executionv1alpha1.KubernetesExecutionParametersSpec{ + ServiceAccountName: testSAName, + ImagePullPolicy: corev1.PullIfNotPresent, + }, + }, + &executionv1alpha1.JobClass{ + ObjectMeta: metav1.ObjectMeta{Name: integrationClass}, + Spec: executionv1alpha1.JobClassSpec{ + ControllerName: kubernetesadapter.ControllerName, + ParametersRef: executionv1alpha1.ParametersReference{ + Group: executionv1alpha1.GroupVersion.Group, + Kind: "KubernetesExecutionParameters", + Name: integrationClass, + }, + AllowedNamespaces: &metav1.LabelSelector{MatchLabels: map[string]string{testLabelKey: testLabelEnabled}}, + }, + }, + } + for _, object := range objects { + if err := directClient.Create(ctx, object); err != nil { + t.Fatalf("create %T: %v", object, err) + } + } + + job := &executionv1alpha1.Job{ + ObjectMeta: metav1.ObjectMeta{Name: testJobName, Namespace: integrationNamespace}, + Spec: executionv1alpha1.JobSpec{ + JobClassName: integrationClass, + DesiredState: executionv1alpha1.JobDesiredStateRunning, + Task: executionv1alpha1.TaskSpec{Image: "alpine:3.22", Command: []string{"true"}}, + }, + } + if err := directClient.Create(ctx, job); err != nil { + t.Fatal(err) + } + + backend := &batchv1.Job{} + eventually(t, 10*time.Second, func() (bool, error) { + err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, backend) + return err == nil, client.IgnoreNotFound(err) + }) + if backend.Labels[kubernetesadapter.JobUIDLabel] != string(job.UID) { + t.Fatalf("backend identity label = %q, want %q", backend.Labels[kubernetesadapter.JobUIDLabel], job.UID) + } + + eventually(t, 10*time.Second, func() (bool, error) { + if err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, job); err != nil { + return false, err + } + return conditionStatus(job, executionv1alpha1.JobConditionScheduled) == metav1.ConditionTrue, nil + }) + + completed := metav1.Now() + if err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, backend); err != nil { + t.Fatal(err) + } + backend.Status.StartTime = &completed + backend.Status.CompletionTime = &completed + backend.Status.Conditions = []batchv1.JobCondition{ + {Type: batchv1.JobSuccessCriteriaMet, Status: corev1.ConditionTrue, Reason: "CompletionsReached"}, + {Type: batchv1.JobComplete, Status: corev1.ConditionTrue, Reason: "Completed"}, + } + if err := directClient.Status().Update(ctx, backend); err != nil { + t.Fatal(err) + } + + eventually(t, 10*time.Second, func() (bool, error) { + if err := directClient.Get(ctx, types.NamespacedName{Namespace: job.Namespace, Name: job.Name}, job); err != nil { + return false, err + } + return conditionStatus(job, executionv1alpha1.JobConditionSucceeded) == metav1.ConditionTrue, nil + }) + if job.Status.StartTime == nil || job.Status.CompletionTime == nil || job.Status.Execution == nil { + t.Fatalf("controller did not persist execution status: %#v", job.Status) + } + + cancelManager() + select { + case err := <-managerErrors: + if err != nil { + t.Fatalf("manager stopped with error: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("manager did not stop") + } +} + +func eventually(t *testing.T, timeout time.Duration, check func() (bool, error)) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + ready, err := check() + if err != nil { + t.Fatal(err) + } + if ready { + return + } + time.Sleep(100 * time.Millisecond) + } + t.Fatal(fmt.Errorf("condition was not met within %s", timeout)) +} + +func conditionStatus(job *executionv1alpha1.Job, conditionType string) metav1.ConditionStatus { + condition := meta.FindStatusCondition(job.Status.Conditions, conditionType) + if condition == nil { + return metav1.ConditionUnknown + } + return condition.Status +} diff --git a/internal/controller/job_controller_test.go b/internal/controller/job_controller_test.go index 124ab94..fca0afc 100644 --- a/internal/controller/job_controller_test.go +++ b/internal/controller/job_controller_test.go @@ -18,7 +18,13 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client/fake" ) -const defaultClassName = "default" +const ( + defaultClassName = "default" + testJobName = "hello" + testSAName = "runner" + testLabelKey = "execution" + testLabelEnabled = "enabled" +) //nolint:modernize // controller-runtime and Kubernetes API structs expose promoted embedded fields. func TestJobReconcilerKubernetesLifecycle(t *testing.T) { @@ -26,7 +32,7 @@ func TestJobReconcilerKubernetesLifecycle(t *testing.T) { now := time.Unix(1_700_000_000, 0) reconciler, kubeClient := testReconciler(t, now, validObjects()...) request := ctrl.Request{} - request.NamespacedName = types.NamespacedName{Namespace: "ci", Name: "hello"} + request.NamespacedName = types.NamespacedName{Namespace: "ci", Name: testJobName} if _, err := reconciler.Reconcile(ctx, request); err != nil { t.Fatalf("add finalizer: %v", err) @@ -180,18 +186,18 @@ func TestJobReconcilerKeepsConfirmedSuccessDuringCancellation(t *testing.T) { //nolint:modernize // Kubernetes API structs expose ObjectMeta through embedded TypeMeta fields. func validObjects() []client.Object { return []client.Object{ - &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ci", Labels: map[string]string{"execution": "enabled"}}}, - &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: "runner", Namespace: "ci"}}, + &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "ci", Labels: map[string]string{testLabelKey: testLabelEnabled}}}, + &corev1.ServiceAccount{ObjectMeta: metav1.ObjectMeta{Name: testSAName, Namespace: "ci"}}, &executionv1alpha1.JobClass{ ObjectMeta: metav1.ObjectMeta{Name: defaultClassName, UID: types.UID("class-uid")}, Spec: executionv1alpha1.JobClassSpec{ ControllerName: kubernetesadapter.ControllerName, ParametersRef: executionv1alpha1.ParametersReference{Group: executionv1alpha1.GroupVersion.Group, Kind: "KubernetesExecutionParameters", Name: defaultClassName}, - AllowedNamespaces: &metav1.LabelSelector{MatchLabels: map[string]string{"execution": "enabled"}}, + AllowedNamespaces: &metav1.LabelSelector{MatchLabels: map[string]string{testLabelKey: testLabelEnabled}}, }, }, &executionv1alpha1.Job{ - ObjectMeta: metav1.ObjectMeta{Name: "hello", Namespace: "ci", UID: types.UID("ayatori-job-uid")}, + ObjectMeta: metav1.ObjectMeta{Name: testJobName, Namespace: "ci", UID: types.UID("ayatori-job-uid")}, Spec: executionv1alpha1.JobSpec{ JobClassName: defaultClassName, DesiredState: executionv1alpha1.JobDesiredStateRunning, Task: executionv1alpha1.TaskSpec{Image: "alpine:3.22", Command: []string{"true"}}, @@ -199,7 +205,7 @@ func validObjects() []client.Object { }, &executionv1alpha1.KubernetesExecutionParameters{ ObjectMeta: metav1.ObjectMeta{Name: defaultClassName, UID: types.UID("parameters-uid")}, - Spec: executionv1alpha1.KubernetesExecutionParametersSpec{ServiceAccountName: "runner", ImagePullPolicy: corev1.PullIfNotPresent}, + Spec: executionv1alpha1.KubernetesExecutionParametersSpec{ServiceAccountName: testSAName, ImagePullPolicy: corev1.PullIfNotPresent}, }, } }