246 lines
11 KiB
Go
246 lines
11 KiB
Go
package controller
|
|
|
|
import (
|
|
"context"
|
|
"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/client/fake"
|
|
)
|
|
|
|
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) {
|
|
ctx := context.Background()
|
|
now := time.Unix(1_700_000_000, 0)
|
|
reconciler, kubeClient := testReconciler(t, now, validObjects()...)
|
|
request := ctrl.Request{}
|
|
request.NamespacedName = types.NamespacedName{Namespace: "ci", Name: testJobName}
|
|
|
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
|
t.Fatalf("add finalizer: %v", err)
|
|
}
|
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
|
t.Fatalf("create backend: %v", err)
|
|
}
|
|
|
|
backend := &batchv1.Job{}
|
|
if err := kubeClient.Get(ctx, request.NamespacedName, backend); err != nil {
|
|
t.Fatalf("backend Job was not created: %v", err)
|
|
}
|
|
if backend.Labels[kubernetesadapter.JobUIDLabel] != "ayatori-job-uid" {
|
|
t.Fatalf("backend UID label = %q", backend.Labels[kubernetesadapter.JobUIDLabel])
|
|
}
|
|
|
|
job := getJob(t, ctx, kubeClient, request.NamespacedName)
|
|
if !conditionIs(job, executionv1alpha1.JobConditionAccepted, metav1.ConditionTrue) ||
|
|
!conditionIs(job, executionv1alpha1.JobConditionScheduled, metav1.ConditionTrue) {
|
|
t.Fatalf("Job was not accepted and scheduled: %#v", job.Status.Conditions)
|
|
}
|
|
|
|
started := metav1.NewTime(now.Add(time.Minute))
|
|
backend.Status.StartTime = &started
|
|
backend.Status.Active = 1
|
|
if err := kubeClient.Status().Update(ctx, backend); err != nil {
|
|
t.Fatalf("set backend running: %v", err)
|
|
}
|
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
|
t.Fatalf("observe running backend: %v", err)
|
|
}
|
|
job = getJob(t, ctx, kubeClient, request.NamespacedName)
|
|
if job.Status.StartTime == nil || !conditionIs(job, executionv1alpha1.JobConditionSucceeded, metav1.ConditionUnknown) {
|
|
t.Fatalf("running state was not observed: %#v", job.Status)
|
|
}
|
|
|
|
completed := metav1.NewTime(now.Add(2 * time.Minute))
|
|
backend = &batchv1.Job{}
|
|
if err := kubeClient.Get(ctx, request.NamespacedName, backend); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
backend.Status.Active = 0
|
|
backend.Status.CompletionTime = &completed
|
|
backend.Status.Conditions = []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue, Reason: "Completed"}}
|
|
if err := kubeClient.Status().Update(ctx, backend); err != nil {
|
|
t.Fatalf("set backend complete: %v", err)
|
|
}
|
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
|
t.Fatalf("observe completed backend: %v", err)
|
|
}
|
|
job = getJob(t, ctx, kubeClient, request.NamespacedName)
|
|
if !conditionIs(job, executionv1alpha1.JobConditionSucceeded, metav1.ConditionTrue) || job.Status.CompletionTime == nil {
|
|
t.Fatalf("terminal state was not observed: %#v", job.Status)
|
|
}
|
|
}
|
|
|
|
//nolint:modernize // controller-runtime Request exposes NamespacedName as a promoted embedded field.
|
|
func TestJobReconcilerRejectsMissingClass(t *testing.T) {
|
|
ctx := context.Background()
|
|
job := validObjects()[3].(*executionv1alpha1.Job).DeepCopy()
|
|
job.Spec.JobClassName = "missing"
|
|
reconciler, kubeClient := testReconciler(t, time.Now(), validObjects()[0], validObjects()[1], job)
|
|
request := ctrl.Request{}
|
|
request.NamespacedName = types.NamespacedName{Namespace: job.Namespace, Name: job.Name}
|
|
|
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
result, err := reconciler.Reconcile(ctx, request)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if result.RequeueAfter == 0 {
|
|
t.Fatal("missing JobClass did not schedule a retry")
|
|
}
|
|
stored := getJob(t, ctx, kubeClient, request.NamespacedName)
|
|
accepted := meta.FindStatusCondition(stored.Status.Conditions, executionv1alpha1.JobConditionAccepted)
|
|
if accepted == nil || accepted.Status != metav1.ConditionFalse || accepted.Reason != "JobClassNotFound" {
|
|
t.Fatalf("unexpected Accepted condition: %#v", accepted)
|
|
}
|
|
}
|
|
|
|
//nolint:modernize // controller-runtime Request exposes NamespacedName as a promoted embedded field.
|
|
func TestJobReconcilerObservesExistingExecutionWithoutJobClass(t *testing.T) {
|
|
ctx := context.Background()
|
|
now := time.Unix(1_700_000_000, 0)
|
|
job := validObjects()[3].(*executionv1alpha1.Job).DeepCopy()
|
|
job.Finalizers = []string{jobFinalizer}
|
|
job.Status.Execution = &executionv1alpha1.ExecutionStatus{Adapter: "kubernetes"}
|
|
backend := kubernetesadapter.BuildJob(job, validObjects()[4].(*executionv1alpha1.KubernetesExecutionParameters), executionv1alpha1.ExecutionResourceRequirements{})
|
|
backend.Status.StartTime = &metav1.Time{Time: now}
|
|
backend.Status.Active = 1
|
|
reconciler, kubeClient := testReconciler(t, now, job, backend)
|
|
request := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: job.Namespace, Name: job.Name}}
|
|
|
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
stored := getJob(t, ctx, kubeClient, request.NamespacedName)
|
|
if stored.Status.StartTime == nil || !conditionIs(stored, executionv1alpha1.JobConditionSucceeded, metav1.ConditionUnknown) {
|
|
t.Fatalf("existing execution was not observed without its JobClass: %#v", stored.Status)
|
|
}
|
|
}
|
|
|
|
//nolint:modernize // controller-runtime Request exposes NamespacedName as a promoted embedded field.
|
|
func TestJobReconcilerCancelsBeforeScheduling(t *testing.T) {
|
|
ctx := context.Background()
|
|
job := validObjects()[3].(*executionv1alpha1.Job).DeepCopy()
|
|
job.Spec.DesiredState = executionv1alpha1.JobDesiredStateCancelled
|
|
reconciler, kubeClient := testReconciler(t, time.Unix(1_700_000_000, 0), job)
|
|
request := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: job.Namespace, Name: job.Name}}
|
|
|
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
stored := getJob(t, ctx, kubeClient, request.NamespacedName)
|
|
condition := meta.FindStatusCondition(stored.Status.Conditions, executionv1alpha1.JobConditionSucceeded)
|
|
if condition == nil || condition.Status != metav1.ConditionFalse || condition.Reason != "Cancelled" {
|
|
t.Fatalf("unexpected cancellation condition: %#v", condition)
|
|
}
|
|
if stored.Status.CompletionTime == nil {
|
|
t.Fatal("cancelled Job has no completionTime")
|
|
}
|
|
}
|
|
|
|
//nolint:modernize // controller-runtime Request exposes NamespacedName as a promoted embedded field.
|
|
func TestJobReconcilerKeepsConfirmedSuccessDuringCancellation(t *testing.T) {
|
|
ctx := context.Background()
|
|
now := time.Unix(1_700_000_000, 0)
|
|
job := validObjects()[3].(*executionv1alpha1.Job).DeepCopy()
|
|
job.Spec.DesiredState = executionv1alpha1.JobDesiredStateCancelled
|
|
job.Finalizers = []string{jobFinalizer}
|
|
backend := kubernetesadapter.BuildJob(job, validObjects()[4].(*executionv1alpha1.KubernetesExecutionParameters), executionv1alpha1.ExecutionResourceRequirements{})
|
|
backend.Status.CompletionTime = &metav1.Time{Time: now}
|
|
backend.Status.Conditions = []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue}}
|
|
reconciler, kubeClient := testReconciler(t, now, job, backend)
|
|
request := ctrl.Request{NamespacedName: types.NamespacedName{Namespace: job.Namespace, Name: job.Name}}
|
|
|
|
if _, err := reconciler.Reconcile(ctx, request); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
stored := getJob(t, ctx, kubeClient, request.NamespacedName)
|
|
if !conditionIs(stored, executionv1alpha1.JobConditionSucceeded, metav1.ConditionTrue) {
|
|
t.Fatalf("confirmed success was overwritten by cancellation: %#v", stored.Status.Conditions)
|
|
}
|
|
if err := kubeClient.Get(ctx, request.NamespacedName, &batchv1.Job{}); err != nil {
|
|
t.Fatalf("successful backend was deleted: %v", err)
|
|
}
|
|
}
|
|
|
|
//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{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{testLabelKey: testLabelEnabled}},
|
|
},
|
|
},
|
|
&executionv1alpha1.Job{
|
|
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"}},
|
|
},
|
|
},
|
|
&executionv1alpha1.KubernetesExecutionParameters{
|
|
ObjectMeta: metav1.ObjectMeta{Name: defaultClassName, UID: types.UID("parameters-uid")},
|
|
Spec: executionv1alpha1.KubernetesExecutionParametersSpec{ServiceAccountName: testSAName, ImagePullPolicy: corev1.PullIfNotPresent},
|
|
},
|
|
}
|
|
}
|
|
|
|
func testReconciler(t *testing.T, now time.Time, objects ...client.Object) (*JobReconciler, client.Client) {
|
|
t.Helper()
|
|
scheme := runtime.NewScheme()
|
|
if err := corev1.AddToScheme(scheme); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := batchv1.AddToScheme(scheme); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := executionv1alpha1.AddToScheme(scheme); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
kubeClient := fake.NewClientBuilder().
|
|
WithScheme(scheme).
|
|
WithStatusSubresource(&executionv1alpha1.Job{}, &batchv1.Job{}).
|
|
WithObjects(objects...).
|
|
Build()
|
|
return &JobReconciler{Client: kubeClient, Now: func() time.Time { return now }}, kubeClient
|
|
}
|
|
|
|
func getJob(t *testing.T, ctx context.Context, kubeClient client.Client, key types.NamespacedName) *executionv1alpha1.Job {
|
|
t.Helper()
|
|
job := &executionv1alpha1.Job{}
|
|
if err := kubeClient.Get(ctx, key, job); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return job
|
|
}
|
|
|
|
func conditionIs(job *executionv1alpha1.Job, conditionType string, status metav1.ConditionStatus) bool {
|
|
condition := meta.FindStatusCondition(job.Status.Conditions, conditionType)
|
|
return condition != nil && condition.Status == status
|
|
}
|