Add NetworkPolicyRecommnedation rest handler

Add unit-test for rest.go
Add unit-test for controller.go

Signed-off-by: Yun-Tang Hsu <hsuy@vmware.com>

build/charts/theia/templates/theia-manager/clusterrole.yaml

@@ -18,5 +18,11 @@ rules:
     verbs: ["get", "list", "watch"]
   - apiGroups: ["crd.theia.antrea.io"]
     resources: ["networkpolicyrecommendations"]
-    verbs: ["get", "list", "watch"]
+    verbs: ["get", "list", "watch", "create", "delete"]
+  - apiGroups: [ "" ]
+    resources: [ "pods" ]
+    verbs: [ "list"]
+  - apiGroups: [ "" ]
+    resources: [ "services", "secrets" ]
+    verbs: [ "get" ]
 {{- end }}

cmd/theia-manager/theia-manager.go

@@ -21,6 +21,7 @@ import (
 	"antrea.io/antrea/pkg/log"
 	"antrea.io/antrea/pkg/signals"
 	"antrea.io/antrea/pkg/util/cipher"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
 
@@ -52,10 +53,16 @@ func run(o *Options) error {
 	if err != nil {
 		return fmt.Errorf("error when generating CRD client: %v", err)
 	}
+	k8sClient, err := createK8sClient()
+	if err != nil {
+		return fmt.Errorf("error when creating K8s client: %v", err)
+	}
 	crdInformerFactory := crdinformers.NewSharedInformerFactory(crdClient, informerDefaultResync)
 	npRecommendationInformer := crdInformerFactory.Crd().V1alpha1().NetworkPolicyRecommendations()
-	npRecoController := networkpolicyrecommendation.NewNPRecommendationController(crdClient, npRecommendationInformer)
-
+	npRecoController, err := networkpolicyrecommendation.NewNPRecommendationController(crdClient, k8sClient, npRecommendationInformer)
+	if err != nil {
+		return fmt.Errorf("error when creating networkPolicyRecommendation controller: %v", err)
+	}
 	cipherSuites, err := cipher.GenerateCipherSuitesList(o.config.APIServer.TLSCipherSuites)
 	if err != nil {
 		return fmt.Errorf("error when generating Cipher Suite list: %v", err)
@@ -77,3 +84,15 @@ func run(o *Options) error {
 	klog.InfoS("Stopping theia manager")
 	return nil
 }
+
+func createK8sClient() (kubernetes.Interface, error) {
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, err
+	}
+	k8sClient, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return nil, err
+	}
+	return k8sClient, nil
+}

pkg/apis/crd/v1alpha1/types.go

@@ -35,7 +35,7 @@ type NetworkPolicyRecommendation struct {
 }
 
 type NetworkPolicyRecommendationSpec struct {
-	Type                string      `json:"type,omitempty"`
+	JobType             string      `json:"jobType,omitempty"`
 	Limit               int         `json:"limit,omitempty"`
 	PolicyType          string      `json:"policyType,omitempty"`
 	StartTime           metav1.Time `json:"startTime,omitempty"`

pkg/apis/intelligence/v1alpha1/types.go

@@ -37,8 +37,8 @@ type NetworkPolicyRecommendation struct {
 	Type                string                            `json:"jobType,omitempty"`
 	Limit               int                               `json:"limit,omitempty"`
 	PolicyType          string                            `json:"policyType,omitempty"`
-	StartInterval       metav1.Time                       `json:"startInterval,omitempty"`
-	EndInterval         metav1.Time                       `json:"endInterval,omitempty"`
+	IntervalStart       metav1.Time                       `json:"intervalStart,omitempty"`
+	IntervalEnd         metav1.Time                       `json:"intervalEnd,omitempty"`
 	NSAllowList         []string                          `json:"nsAllowList,omitempty"`
 	ExcludeLabels       bool                              `json:"excludeLabels,omitempty"`
 	ToServices          bool                              `json:"toServices,omitempty"`
@@ -51,14 +51,13 @@ type NetworkPolicyRecommendation struct {
 }
 
 type NetworkPolicyRecommendationStatus struct {
-	State                 string      `json:"state,omitempty"`
-	SparkApplication      string      `json:"sparkApplication,omitempty"`
-	CompletedStages       int         `json:"completedStages,omitempty"`
-	TotalStages           int         `json:"totalStages,omitempty"`
-	RecommendationOutcome string      `json:"recommendationOutcome,omitempty"`
-	CompletionTimestamp   metav1.Time `json:"completionTimestamp,omitempty"`
-	ErrorCode             string      `json:"errorCode,omitempty"`
-	ErrorMsg              string      `json:"errorMsg,omitempty"`
+	State                    string `json:"state,omitempty"`
+	SparkApplication         string `json:"sparkApplication,omitempty"`
+	CompletedStages          int    `json:"completedStages,omitempty"`
+	TotalStages              int    `json:"totalStages,omitempty"`
+	RecommendedNetworkPolicy string `json:"recommendedNetworkPolicy,omitempty"`
+	ErrorCode                string `json:"errorCode,omitempty"`
+	ErrorMsg                 string `json:"errorMsg,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/apiserver/registry/intelligence/networkpolicyrecommendation/rest.go

@@ -16,13 +16,15 @@ package networkpolicyrecommendation
 
 import (
 	"context"
+	"fmt"
 
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/internalversion"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apiserver/pkg/registry/rest"
 
+	crdv1alpha1 "antrea.io/theia/pkg/apis/crd/v1alpha1"
 	intelligence "antrea.io/theia/pkg/apis/intelligence/v1alpha1"
 	"antrea.io/theia/pkg/querier"
 )
@@ -33,9 +35,11 @@ type REST struct {
 }
 
 var (
-	_ rest.Scoper = &REST{}
-	_ rest.Getter = &REST{}
-	_ rest.Lister = &REST{}
+	_ rest.Scoper          = &REST{}
+	_ rest.Getter          = &REST{}
+	_ rest.Lister          = &REST{}
+	_ rest.Creater         = &REST{}
+	_ rest.GracefulDeleter = &REST{}
 )
 
 // NewREST returns a REST object that will work against API services.
@@ -48,30 +52,14 @@ func (r *REST) New() runtime.Object {
 }
 
 func (r *REST) getNetworkPolicyRecommendation(name string) *intelligence.NetworkPolicyRecommendation {
-	npReco, err := r.npRecommendationQuerier.GetNetworkPolicyRecommendation("flow-visibility", name)
+	npReco, err := r.npRecommendationQuerier.GetNetworkPolicyRecommendation(name)
 	if err != nil {
 		return nil
 	}
-
-	job := new(intelligence.NetworkPolicyRecommendation)
-	job.Name = npReco.Name
-	job.Type = npReco.Spec.Type
-	job.Limit = npReco.Spec.Limit
-	job.PolicyType = npReco.Spec.PolicyType
-	job.StartInterval = npReco.Spec.StartTime
-	job.EndInterval = npReco.Spec.EndTime
-	job.NSAllowList = npReco.Spec.NSAllowList
-	job.ExcludeLabels = npReco.Spec.ExcludeLabels
-	job.ToServices = npReco.Spec.ToServices
-	job.ExecutorInstances = npReco.Spec.ExecutorInstances
-	job.DriverCoreRequest = npReco.Spec.DriverCoreRequest
-	job.DriverMemory = npReco.Spec.DriverMemory
-	job.ExecutorCoreRequest = npReco.Spec.ExecutorCoreRequest
-	job.ExecutorMemory = npReco.Spec.ExecutorMemory
-	return job
+	return npReco
 }
 
-func (r *REST) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
+func (r *REST) Get(_ context.Context, name string, _ *metav1.GetOptions) (runtime.Object, error) {
 	job := r.getNetworkPolicyRecommendation(name)
 	if job == nil {
 		return nil, errors.NewNotFound(intelligence.Resource("networkpolicyrecommendations"), name)
@@ -83,9 +71,12 @@ func (r *REST) NewList() runtime.Object {
 	return &intelligence.NetworkPolicyRecommendationList{}
 }
 
-func (r *REST) List(ctx context.Context, options *internalversion.ListOptions) (runtime.Object, error) {
-	list := new(intelligence.NetworkPolicyRecommendationList)
-	return list, nil
+func (r *REST) List(_ context.Context, _ *internalversion.ListOptions) (runtime.Object, error) {
+	itemList, err := r.npRecommendationQuerier.ListNetworkPolicyRecommendation()
+	if err != nil {
+		return nil, errors.NewBadRequest(fmt.Sprintf("cannot retrieve npr from controller. err:%s", err))
+	}
+	return itemList, nil
 }
 
 func (r *REST) NamespaceScoped() bool {
@@ -95,3 +86,46 @@ func (r *REST) NamespaceScoped() bool {
 func (r *REST) ConvertToTable(ctx context.Context, obj runtime.Object, tableOptions runtime.Object) (*metav1.Table, error) {
 	return rest.NewDefaultTableConvertor(intelligence.Resource("networkpolicyrecommendations")).ConvertToTable(ctx, obj, tableOptions)
 }
+
+func (r *REST) Create(_ context.Context, obj runtime.Object, _ rest.ValidateObjectFunc, _ *metav1.CreateOptions) (runtime.Object, error) {
+	npReco, ok := obj.(*intelligence.NetworkPolicyRecommendation)
+	if !ok {
+		return nil, errors.NewBadRequest(fmt.Sprintf("not a NetworkPolicyRecommendation object: %T", obj))
+	}
+	existNPReco, _ := r.npRecommendationQuerier.GetNetworkPolicyRecommendation(npReco.Name)
+	if existNPReco != nil {
+		return nil, errors.NewBadRequest(fmt.Sprintf("networkPolicyRecommendation job exists, name: %s", npReco.Name))
+	}
+	job := new(crdv1alpha1.NetworkPolicyRecommendation)
+	job.Name = npReco.Name
+	job.Spec.JobType = npReco.Type
+	job.Spec.Limit = npReco.Limit
+	job.Spec.PolicyType = npReco.PolicyType
+	job.Spec.StartTime = npReco.IntervalStart
+	job.Spec.EndTime = npReco.IntervalEnd
+	job.Spec.NSAllowList = npReco.NSAllowList
+	job.Spec.ExcludeLabels = npReco.ExcludeLabels
+	job.Spec.ToServices = npReco.ToServices
+	job.Spec.ExecutorInstances = npReco.ExecutorInstances
+	job.Spec.DriverCoreRequest = npReco.DriverCoreRequest
+	job.Spec.DriverMemory = npReco.DriverMemory
+	job.Spec.ExecutorCoreRequest = npReco.ExecutorCoreRequest
+	job.Spec.ExecutorMemory = npReco.ExecutorMemory
+	_, err := r.npRecommendationQuerier.CreateNetworkPolicyRecommendation(job)
+	if err != nil {
+		return nil, err
+	}
+	return &metav1.Status{Status: metav1.StatusSuccess}, nil
+}
+
+func (r *REST) Delete(_ context.Context, name string, _ rest.ValidateObjectFunc, _ *metav1.DeleteOptions) (runtime.Object, bool, error) {
+	_, err := r.npRecommendationQuerier.GetNetworkPolicyRecommendation(name)
+	if err != nil {
+		return nil, false, errors.NewBadRequest(fmt.Sprintf("networkPolicyRecommendation job doesn't exist, name: %s", name))
+	}
+	err = r.npRecommendationQuerier.DeleteNetworkPolicyRecommendation(name)
+	if err != nil {
+		return nil, false, err
+	}
+	return &metav1.Status{Status: metav1.StatusSuccess}, false, nil
+}