test: add unit test

Makefile

@@ -73,11 +73,8 @@ vet: ## Run go vet against code.
 lint:
 	golangci-lint run -v --timeout=5m
 
-ENVTEST_ASSETS_DIR=$(shell pwd)/testbin
-test: manifests generate fmt vet ## Run tests.
-	mkdir -p ${ENVTEST_ASSETS_DIR}
-	test -f ${ENVTEST_ASSETS_DIR}/setup-envtest.sh || curl -sSLo ${ENVTEST_ASSETS_DIR}/setup-envtest.sh https://raw.githubusercontent.com/kubernetes-sigs/controller-runtime/v0.7.0/hack/setup-envtest.sh
-	source ${ENVTEST_ASSETS_DIR}/setup-envtest.sh; fetch_envtest_tools $(ENVTEST_ASSETS_DIR); setup_envtest_env $(ENVTEST_ASSETS_DIR); go test ./... -coverprofile cover.out
+test: ## Run tests.
+	go test -v -cover ./... -coverprofile=coverage.out
 
 ##@ Build
 

api/v1alpha1/groupversion_info.go

@@ -28,8 +28,8 @@ var (
 	// KindNerveXJob is kind of NerveXJob
 	KindNerveXJob = "NerveXJob"
 
-	// KindALConfig if kind of KindALConfig
-	KindALConfig = "ActorLearnerConfig"
+	// KindAGConfig is kind of AGConfig
+	KindAGConfig = "ActorLearnerConfig"
 
 	// GroupVersion is group version used to register these objects
 	GroupVersion = schema.GroupVersion{Group: "nervex.sensetime.com", Version: "v1alpha1"}

config/manager/nervex_operator.yaml

@@ -49,8 +49,8 @@ spec:
         resources:
           limits:
             cpu: 100m
-            memory: 30Mi
+            memory: 500Mi
           requests:
             cpu: 100m
-            memory: 20Mi
+            memory: 500Mi
       terminationGracePeriodSeconds: 10

config/manager/nervex_server.yaml

@@ -37,10 +37,10 @@ spec:
         resources:
           limits:
             cpu: 100m
-            memory: 30Mi
+            memory: 500Mi
           requests:
             cpu: 100m
-            memory: 20Mi
+            memory: 500Mi
       terminationGracePeriodSeconds: 10
 
 ---

controllers/nervexjob_controller_test.go

+package controllers
+
+import (
+	"context"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	nervexv1alpha1 "go-sensephoenix.sensetime.com/nervex-operator/api/v1alpha1"
+	nervexutil "go-sensephoenix.sensetime.com/nervex-operator/utils"
+	testutil "go-sensephoenix.sensetime.com/nervex-operator/utils/testutils"
+)
+
+var _ = Describe("NerveXJob Controller", func() {
+
+	Context("When creating a NerveXJob", func() {
+		It("Should be succeeded", func() {
+			By("Creating a NerveXJob")
+			ctx := context.Background()
+			nervexjob := testutil.NewNerveXJob()
+			err := k8sClient.Create(ctx, nervexjob, &client.CreateOptions{})
+			Expect(err).ShouldNot(HaveOccurred())
+
+			By("Successfully created")
+			key := types.NamespacedName{Namespace: nervexjob.Namespace, Name: nervexjob.Name}
+			createdNvxjob := nervexv1alpha1.NerveXJob{}
+			Eventually(func() bool {
+				err := k8sClient.Get(ctx, key, &createdNvxjob)
+				if err != nil {
+					return false
+				}
+				return true
+			}, timeout, interval).Should(BeTrue())
+
+			By("Checking coordinator and aggregator are created")
+			for _, replicaName := range []string{
+				nervexutil.ReplicaPodName(nervexjob.Name, "coordinator"),
+				nervexutil.ReplicaPodName(nervexjob.Name, "aggregator"),
+			} {
+				var pod corev1.Pod
+				podKey := types.NamespacedName{Namespace: nervexjob.Namespace, Name: replicaName}
+				Eventually(func() bool {
+					err = k8sClient.Get(ctx, podKey, &pod)
+					if err != nil {
+						return false
+					}
+					return true
+				}, timeout, interval).Should(BeTrue())
+			}
+
+			By("Update coordinator and aggregator to Running")
+			for _, replicaName := range []string{
+				nervexutil.ReplicaPodName(nervexjob.Name, "coordinator"),
+				nervexutil.ReplicaPodName(nervexjob.Name, "aggregator"),
+			} {
+				podKey := types.NamespacedName{Namespace: nervexjob.Namespace, Name: replicaName}
+				err = testutil.UpdatePodPhase(ctx, k8sClient, podKey, corev1.PodRunning)
+				Expect(err).NotTo(HaveOccurred())
+			}
+
+			By("Checking the created NerveXJob has enough coordinator and aggregator")
+			for _, rtype := range []nervexv1alpha1.ReplicaType{nervexv1alpha1.ReplicaTypeCoordinator, nervexv1alpha1.ReplicaTypeAggregator} {
+				Eventually(func() int {
+					err := k8sClient.Get(ctx, key, &createdNvxjob)
+					if err != nil {
+						return -1
+					}
+					if createdNvxjob.Status.ReplicaStatus == nil {
+						return -1
+					}
+					return int(createdNvxjob.Status.ReplicaStatus[rtype].Active)
+				}, timeout, interval).Should(Equal(1))
+			}
+
+			By("Update coordinator and aggregator to Succeeded")
+			for _, replicaName := range []string{
+				nervexutil.ReplicaPodName(nervexjob.Name, "coordinator"),
+				nervexutil.ReplicaPodName(nervexjob.Name, "aggregator"),
+			} {
+				podKey := types.NamespacedName{Namespace: nervexjob.Namespace, Name: replicaName}
+				err = testutil.UpdatePodPhase(ctx, k8sClient, podKey, corev1.PodSucceeded)
+				Expect(err).NotTo(HaveOccurred())
+			}
+
+			By("Successfully succeeded")
+			Eventually(func() nervexv1alpha1.Phase {
+				err := k8sClient.Get(ctx, key, &createdNvxjob)
+				if err != nil {
+					return ""
+				}
+				return createdNvxjob.Status.Phase
+			}, timeout, interval).Should(Equal(nervexv1alpha1.JobSucceeded))
+
+			By("Checking the coordinator are succeeded")
+			for _, rtype := range []nervexv1alpha1.ReplicaType{nervexv1alpha1.ReplicaTypeCoordinator, nervexv1alpha1.ReplicaTypeAggregator} {
+				Consistently(func() int {
+					err := k8sClient.Get(ctx, key, &createdNvxjob)
+					if err != nil {
+						return -1
+					}
+					return int(createdNvxjob.Status.ReplicaStatus[rtype].Succeeded)
+				}, duration, interval).Should(Equal(1))
+			}
+		})
+	})
+})

controllers/suite_test.go

@@ -17,12 +17,17 @@ limitations under the License.
 package controllers
 
 import (
+	"context"
 	"path/filepath"
 	"testing"
+	"time"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 	"sigs.k8s.io/controller-runtime/pkg/envtest/printer"
@@ -30,13 +35,20 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
 	nervexv1alpha1 "go-sensephoenix.sensetime.com/nervex-operator/api/v1alpha1"
+	testutil "go-sensephoenix.sensetime.com/nervex-operator/utils/testutils"
 	//+kubebuilder:scaffold:imports
 )
 
 // These tests use Ginkgo (BDD-style Go testing framework). Refer to
 // http://onsi.github.io/ginkgo/ to learn more about Ginkgo.
 
-// var cfg *rest.Config
+const (
+	timeout  = 30 * time.Second
+	interval = 250 * time.Millisecond
+	duration = 10 * time.Second
+)
+
+var cfg *rest.Config
 var k8sClient client.Client
 var testEnv *envtest.Environment
 
@@ -70,6 +82,47 @@ var _ = BeforeSuite(func() {
 	Expect(err).NotTo(HaveOccurred())
 	Expect(k8sClient).NotTo(BeNil())
 
+	By("Apply AggregatorConfig")
+	ctx := context.Background()
+	agconfig := testutil.NewAggregatorConfig()
+
+	err = k8sClient.Create(ctx, testutil.NewNamespace(agconfig.Namespace), &client.CreateOptions{})
+	Expect(err).NotTo(HaveOccurred())
+
+	err = k8sClient.Create(ctx, agconfig, &client.CreateOptions{})
+	Expect(err).NotTo(HaveOccurred())
+
+	By("Agconfig successfully created")
+	key := types.NamespacedName{Namespace: agconfig.Namespace, Name: agconfig.Name}
+	createdAg := nervexv1alpha1.AggregatorConfig{}
+	Eventually(func() bool {
+		err := k8sClient.Get(ctx, key, &createdAg)
+		if err != nil {
+			return false
+		}
+		return true
+	}, timeout, interval).Should(BeTrue())
+
+	// create controller manager
+	k8sManager, err := ctrl.NewManager(cfg, ctrl.Options{
+		Scheme: scheme.Scheme,
+	})
+	Expect(err).NotTo(HaveOccurred())
+
+	err = (&NerveXJobReconciler{
+		Scheme:   k8sManager.GetScheme(),
+		Client:   k8sManager.GetClient(),
+		Log:      ctrl.Log.WithName("controllers").WithName("NerveXJob"),
+		AGConfig: key.String(),
+	}).SetupWithManager(k8sManager)
+
+	Expect(err).NotTo(HaveOccurred())
+
+	// starting manager
+	go func() {
+		err = k8sManager.Start(ctrl.SetupSignalHandler())
+		Expect(err).NotTo(HaveOccurred())
+	}()
 }, 60)
 
 var _ = AfterSuite(func() {

utils/testutils/const.go

+package testutils
+
+const (
+	DefaultAGConfigNamespace = "nervex-system"
+	DefaultAGConfigName      = "aggregator-config"
+
+	NerveXJobName        = "nervexjob-example"
+	NerveXJobNamespace   = "default"
+	NerveXJobImage       = "alpine:latest"
+	DefaultSleepDuration = "2s"
+)

utils/testutils/nervexjob.go

+package testutils
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	nervexv1alpha1 "go-sensephoenix.sensetime.com/nervex-operator/api/v1alpha1"
+	nervexutil "go-sensephoenix.sensetime.com/nervex-operator/utils"
+)
+
+func NewNerveXJob() *nervexv1alpha1.NerveXJob {
+	return &nervexv1alpha1.NerveXJob{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       nervexv1alpha1.KindNerveXJob,
+			APIVersion: nervexv1alpha1.GroupVersion.String(),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      NerveXJobName,
+			Namespace: NerveXJobNamespace,
+		},
+		Spec: nervexv1alpha1.NerveXJobSpec{
+			Coordinator: nervexv1alpha1.CoordinatorSpec{
+				Template: corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{
+								Name:    nervexutil.DefaultCoordinatorContainerName,
+								Image:   NerveXJobImage,
+								Command: []string{"/bin/sh", "-c", "sleep", DefaultSleepDuration},
+							},
+						},
+					},
+				},
+			},
+			Collector: nervexv1alpha1.CollectorSpec{
+				Template: corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{
+								Name:    nervexutil.DefaultCollectorContainerName,
+								Image:   NerveXJobImage,
+								Command: []string{"/bin/sh", "-c", "sleep", DefaultSleepDuration},
+							},
+						},
+					},
+				},
+			},
+			Learner: nervexv1alpha1.LearnerSpec{
+				Template: corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{
+								Name:    nervexutil.DefaultLearnerContainerName,
+								Image:   NerveXJobImage,
+								Command: []string{"/bin/sh", "-c", "sleep", DefaultSleepDuration},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func NewNamespace(namespace string) *corev1.Namespace {
+	return &corev1.Namespace{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: "v1",
+			Kind:       "Namespace",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: namespace,
+		},
+	}
+}
+
+func NewAggregatorConfig() *nervexv1alpha1.AggregatorConfig {
+	return &nervexv1alpha1.AggregatorConfig{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: nervexv1alpha1.GroupVersion.String(),
+			Kind:       nervexv1alpha1.KindAGConfig,
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      DefaultAGConfigName,
+			Namespace: DefaultAGConfigNamespace,
+		},
+		Spec: nervexv1alpha1.AggregatorConfigSpec{
+			Aggregator: nervexv1alpha1.AggregatorSpec{
+				Template: corev1.PodTemplateSpec{
+					Spec: corev1.PodSpec{
+						Containers: []corev1.Container{
+							{
+								Name:    nervexutil.DefaultAggregatorContainerName,
+								Image:   NerveXJobImage,
+								Command: []string{"/bin/sh", "-c", "sleep", DefaultSleepDuration},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}

utils/testutils/pod.go

+package testutils
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func UpdatePodPhase(ctx context.Context, k8sClient client.Client, podKey types.NamespacedName, phase corev1.PodPhase) error {
+	var pod corev1.Pod
+	err := k8sClient.Get(ctx, podKey, &pod)
+	if err != nil {
+		return err
+	}
+	pod.Status.Phase = phase
+	err = k8sClient.Status().Update(ctx, &pod, &client.UpdateOptions{})
+	if err != nil {
+		return err
+	}
+	return nil
+}

utils/util.go

@@ -99,7 +99,7 @@ func BuildPodFromTemplate(
 	// generate name is the NerveXJob name
 	jobName := ownRefer.Name
 	portEnv := ""
-	podName := fmt.Sprintf("%s-%s", jobName, replicaType)
+	podName := ReplicaPodName(jobName, replicaType)
 	switch replicaType {
 	case CollectorName:
 		portEnv = "COLLECTOR_PORT"
@@ -150,6 +150,10 @@ func BuildPodFromTemplate(
 	return pod, port, nil
 }
 
+func ReplicaPodName(name, replicaType string) string {
+	return fmt.Sprintf("%s-%s", name, replicaType)
+}
+
 func SetPodEnv(pod *corev1.Pod, envs map[string]string) {
 	// add env
 	for i := range pod.Spec.Containers {