Merge pull request #31 from TaoZou1/gc

GC support

cmd/main.go

@@ -71,7 +71,7 @@ func main() {
 		securityReconcile.Service = service
 	}
 
-	if err = securityReconcile.SetupWithManager(mgr); err != nil {
+	if err = securityReconcile.Start(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "SecurityPolicy")
 		os.Exit(1)
 	}

go.mod

@@ -8,7 +8,7 @@ require (
 	github.com/vmware/vsphere-automation-sdk-go/runtime v0.4.0
 	github.com/vmware/vsphere-automation-sdk-go/services/nsxt v0.6.0
 	golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac
-	gopkg.in/ini.v1 v1.62.0
+	gopkg.in/ini.v1 v1.66.2
 	k8s.io/api v0.22.1
 	k8s.io/apimachinery v0.22.1
 	k8s.io/client-go v0.22.1
@@ -17,39 +17,54 @@ require (
 )
 
 require (
+	github.com/agiledragon/gomonkey/v2 v2.4.0 // indirect
 	github.com/cilium/ebpf v0.7.0 // indirect
+	github.com/cosiner/argv v0.1.0 // indirect
+	github.com/cpuguy83/go-md2man/v2 v2.0.1 // indirect
+	github.com/deckarep/golang-set v1.8.0 // indirect
+	github.com/derekparker/trie v0.0.0-20200317170641-1fdf38b7b0e9 // indirect
 	github.com/go-delve/delve v1.8.0 // indirect
 	github.com/go-logr/zapr v0.4.0 // indirect
+	github.com/google/go-dap v0.6.0 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
-	github.com/mattn/go-isatty v0.0.3 // indirect
+	github.com/inconshreveable/mousetrap v1.0.0 // indirect
+	github.com/mattn/go-colorable v0.1.12 // indirect
+	github.com/mattn/go-isatty v0.0.14 // indirect
+	github.com/mattn/go-runewidth v0.0.13 // indirect
+	github.com/peterh/liner v1.2.1 // indirect
+	github.com/rivo/uniseg v0.2.0 // indirect
+	github.com/russross/blackfriday/v2 v2.1.0 // indirect
+	github.com/shurcooL/sanitized_anchor_name v1.0.0 // indirect
+	github.com/spf13/cobra v1.3.0 // indirect
+	go.starlark.net v0.0.0-20211203141949-70c0e40ae128 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
 	go.uber.org/zap v1.19.0 // indirect
-	golang.org/x/arch v0.0.0-20190927153633-4e8777c89be4 // indirect
+	golang.org/x/arch v0.0.0-20210923205945-b76863e36670 // indirect
 )
 
 require (
 	github.com/beevik/etree v1.1.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/cespare/xxhash/v2 v2.1.1 // indirect
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/evanphx/json-patch v4.11.0+incompatible // indirect
-	github.com/fsnotify/fsnotify v1.4.9 // indirect
+	github.com/fsnotify/fsnotify v1.5.1 // indirect
 	github.com/gibson042/canonicaljson-go v1.0.3 // indirect
 	github.com/go-logr/logr v0.4.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt v3.2.2+incompatible
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
-	github.com/google/go-cmp v0.5.5 // indirect
+	github.com/google/go-cmp v0.5.6 // indirect
 	github.com/google/gofuzz v1.1.0 // indirect
 	github.com/google/uuid v1.2.0 // indirect
 	github.com/googleapis/gnostic v0.5.5 // indirect
 	github.com/imdario/mergo v0.3.12 // indirect
-	github.com/json-iterator/go v1.1.11 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
-	github.com/modern-go/reflect2 v1.0.1 // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.11.0 // indirect
@@ -61,8 +76,8 @@ require (
 	github.com/vmware/govmomi v0.27.2
 	github.com/vmware/vsphere-automation-sdk-go/lib v0.4.0 // indirect
 	golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f // indirect
-	golang.org/x/oauth2 v0.0.0-20210402161424-2e8d93401602 // indirect
-	golang.org/x/sys v0.0.0-20211209171907-798191bca915 // indirect
+	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
+	golang.org/x/sys v0.0.0-20220111092808-5a964db01320 // indirect
 	golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/tools v0.1.8 // indirect

pkg/config/config.go

@@ -153,7 +153,12 @@ func (operatorConfig *NSXOperatorConfig) validate() error {
 func (operatorConfig *NSXOperatorConfig) validateVersion() error {
 	nsxVersion := &NsxVersion{}
 	host := operatorConfig.NsxApiManagers[0]
-	tokenProvider, _ := jwt.NewTokenProvider(operatorConfig.VCEndPoint, operatorConfig.HttpsPort, operatorConfig.SsoDomain, nil)
+	var tokenProvider auth.TokenProvider
+	if err := operatorConfig.VCConfig.validate(); err == nil {
+		tokenProvider, _ = jwt.NewTokenProvider(operatorConfig.VCEndPoint, operatorConfig.HttpsPort, operatorConfig.SsoDomain, nil)
+	} else {
+		tokenProvider = nil
+	}
 	if err := nsxVersion.getVersion(host, operatorConfig.NsxApiUser, operatorConfig.NsxApiPassword, tokenProvider); err != nil {
 		return err
 	}
@@ -165,13 +170,13 @@ func (operatorConfig *NSXOperatorConfig) validateVersion() error {
 
 func (vcConfig *VCConfig) validate() error {
 	if len(vcConfig.VCEndPoint) == 0 {
-		err := errors.New("Invalid field " + "VcEndPoint")
+		err := errors.New("invalid field " + "VcEndPoint")
 		log.Error(err, "validate VcConfig failed", "VcEndPoint", vcConfig.VCEndPoint)
 		return err
 	}
 
 	if len(vcConfig.SsoDomain) == 0 {
-		err := errors.New("Invalid field " + "SsoDomain")
+		err := errors.New("invalid field " + "SsoDomain")
 		log.Error(err, "validate VcConfig failed", "SsoDomain", vcConfig.SsoDomain)
 		return err
 	}

pkg/controllers/securitypolicy_controller.go

@@ -5,15 +5,20 @@ package controllers
 
 import (
 	"context"
+	"time"
 
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
 	"github.com/vmware-tanzu/nsx-operator/pkg/apis/v1alpha1"
 	_ "github.com/vmware-tanzu/nsx-operator/pkg/nsx/ratelimiter"
 	"github.com/vmware-tanzu/nsx-operator/pkg/nsx/services"
+	"github.com/vmware-tanzu/nsx-operator/pkg/util"
 )
 
 var (
@@ -22,32 +27,100 @@ var (
 
 // SecurityPolicyReconciler reconciles a SecurityPolicy object
 type SecurityPolicyReconciler struct {
-	client.Client
+	Client  client.Client
 	Scheme  *runtime.Scheme
 	Service *services.SecurityPolicyService
 }
 
 func (r *SecurityPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-
 	obj := &v1alpha1.SecurityPolicy{}
-	if err := r.Get(ctx, req.NamespacedName, obj); err != nil {
-		log.Error(err, "unable to fetch object")
-		return ctrl.Result{}, err
+	log.Info("reconciling securitypolicy CR", "securitypolicy", req.NamespacedName)
+
+	if err := r.Client.Get(ctx, req.NamespacedName, obj); err != nil {
+		log.Error(err, "unable to fetch security policy CR")
+		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 
-	log.Info("reconciling securitypolicy CR", "securitypolicy", req.NamespacedName)
+	if obj.ObjectMeta.DeletionTimestamp.IsZero() {
+		if !controllerutil.ContainsFinalizer(obj, util.FinalizerName) {
+			controllerutil.AddFinalizer(obj, util.FinalizerName)
+			if err := r.Client.Update(ctx, obj); err != nil {
+				return ctrl.Result{}, err
+			}
+		}
+		if err := r.Service.CreateOrUpdateSecurityPolicy(obj); err != nil {
+			log.Error(err, "failed to create or update security policy CR", "securitypolicy", req.NamespacedName)
+			return ctrl.Result{}, err
+		}
+	} else {
+		if containsString(obj.GetFinalizers(), util.FinalizerName) {
+			if err := r.Service.DeleteSecurityPolicy(obj.UID); err != nil {
+				return ctrl.Result{}, err
+			}
 
-	if err := r.Service.CreateOrUpdateSecurityPolicy(obj); err != nil {
-		log.Error(err, "failed to create or update security policy", "securitypolicy", req.NamespacedName)
-		return ctrl.Result{}, err
+			controllerutil.RemoveFinalizer(obj, util.FinalizerName)
+			if err := r.Client.Update(ctx, obj); err != nil {
+				return ctrl.Result{}, err
+			}
+		}
 	}
 
 	return ctrl.Result{}, nil
 }
 
-// SetupWithManager sets up the controller with the Manager.
-func (r *SecurityPolicyReconciler) SetupWithManager(mgr ctrl.Manager) error {
+func containsString(source []string, target string) bool {
+	for _, item := range source {
+		if item == target {
+			return true
+		}
+	}
+	return false
+}
+
+func (r *SecurityPolicyReconciler) setupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1alpha1.SecurityPolicy{}).
 		Complete(r)
 }
+
+// Start setup manager and launch GC
+func (r *SecurityPolicyReconciler) Start(mgr ctrl.Manager) error {
+	err := r.setupWithManager(mgr)
+	if err != nil {
+		return err
+	}
+
+	go r.GarbageCollector()
+	return nil
+}
+
+func (r *SecurityPolicyReconciler) GarbageCollector() {
+	ctx := context.Background()
+	log.V(1).Info("garbage collector started")
+	for {
+		time.Sleep(util.GCInterval)
+		nsxPolicySet := r.Service.ListSecurityPolicy()
+		if len(nsxPolicySet) == 0 {
+			continue
+		}
+		policyList := &v1alpha1.SecurityPolicyList{}
+		err := r.Client.List(ctx, policyList)
+		if err != nil {
+			log.Error(err, "failed to list security policy CR")
+			continue
+		}
+
+		CRPolicySet := sets.NewString()
+		for _, policy := range policyList.Items {
+			CRPolicySet.Insert(string(policy.UID))
+		}
+
+		for elem := range nsxPolicySet {
+			if CRPolicySet.Has(elem) {
+				continue
+			}
+			log.V(1).Info("GC collected SecurityPolicy CR", "UID", elem)
+			r.Service.DeleteSecurityPolicy(types.UID(elem))
+		}
+	}
+}

pkg/nsx/cluster.go

@@ -91,7 +91,6 @@ func (cluster *Cluster) NewRestConnector() (*policyclient.RestConnector, *Header
 func (cluster *Cluster) createTransport(tokenProvider auth.TokenProvider, idle time.Duration) *Transport {
 	// TODO: support the case if InsecureSkipVerify is false
 	tlsConfig := tls.Config{InsecureSkipVerify: true}
-	tlsConfig.BuildNameToCertificate()
 	tr := &http.Transport{
 		TLSClientConfig: &tlsConfig,
 		IdleConnTimeout: idle * time.Second,
@@ -107,7 +106,6 @@ func (cluster *Cluster) createHTTPClient(tr *Transport, timeout time.Duration) h
 
 func (cluster *Cluster) createNoBalancerClient(timeout, idle time.Duration) http.Client {
 	tlsConfig := tls.Config{InsecureSkipVerify: true}
-	tlsConfig.BuildNameToCertificate()
 	transport := &http.Transport{
 		TLSClientConfig: &tlsConfig,
 		IdleConnTimeout: idle * time.Second,

pkg/nsx/endpoint.go

@@ -4,6 +4,7 @@
 package nsx
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -114,9 +115,9 @@ func (ep *Endpoint) keepAlive() error {
 			return err
 		}
 	} else {
-		log.V(1).Info("token is invalid, using user/password to keep alive")
-		req, err = http.NewRequest("GET", fmt.Sprintf(healthURL, ep.Scheme(), ep.Host()), nil)
+		log.V(1).Info("no token provider, using user/password to keep alive")
 		req.SetBasicAuth(ep.user, ep.password)
+		req.Header.Add("X-Xsrf-Token", ep.XSRFToken())
 		if err != nil {
 			log.Error(err, "keepalive request creation failed")
 			return err
@@ -128,17 +129,23 @@ func (ep *Endpoint) keepAlive() error {
 		return err
 	}
 	body, err := ioutil.ReadAll(resp.Body)
-	log.V(1).Info("received HTTP response", "request", req, "response", string(body))
+	if err != nil {
+		log.Error(err, "failed to read response", "endpoint", ep.Host())
+		return err
+	}
+	log.V(1).Info("received HTTP response", "response", string(body))
 	defer resp.Body.Close()
-	if err = util.InitErrorFromResponse(ep.Host(), resp); err == nil {
+	if resp.StatusCode == http.StatusOK {
 		var a epHealthy
-		if err = json.Unmarshal(body, &a); err == nil && a.Healthy == true {
+		if err = json.Unmarshal(body, &a); err == nil && a.Healthy {
 			ep.setStatus(UP)
 			return nil
 		}
 		log.Error(err, "failed to validate API cluster", "endpoint", ep.Host(), "healthy", a)
 		return err
 	}
+	resp.Body = ioutil.NopCloser(bytes.NewReader(body))
+	err = util.InitErrorFromResponse(ep.Host(), resp)
 
 	if util.ShouldRegenerate(err) {
 		log.Error(err, "failed to validate API cluster due to an exception that calls for regeneration", "endpoint", ep.Host())
@@ -259,9 +266,6 @@ func (ep *Endpoint) createAuthSession(certProvider auth.ClientCertProvider, toke
 		log.V(1).Info("skipping session creation with client certificate auth")
 		return nil
 	}
-	u := &url.URL{Host: ep.Host(), Scheme: ep.Scheme()}
-	var req *http.Request
-	var err error
 	if tokenProvider != nil {
 		_, err := tokenProvider.GetToken(true)
 		if err != nil {
@@ -270,21 +274,22 @@ func (ep *Endpoint) createAuthSession(certProvider auth.ClientCertProvider, toke
 		}
 		log.V(1).Info("Skipping session create with JWT based auth")
 		return nil
-	} else {
-		postValues := url.Values{}
-		postValues.Add("j_username", username)
-		postValues.Add("j_password", password)
-		req, err = http.NewRequest("POST", fmt.Sprintf("%s://%s/api/session/create", u.Scheme, u.Host), strings.NewReader(postValues.Encode()))
-		if err != nil {
-			log.Error(err, "failed to generate request for session creation", "endpoint", ep.Host())
-			return err
-		}
+	}
+
+	u := &url.URL{Host: ep.Host(), Scheme: ep.Scheme()}
+	postValues := url.Values{}
+	postValues.Add("j_username", username)
+	postValues.Add("j_password", password)
+	req, err := http.NewRequest("POST", fmt.Sprintf("%s://%s/api/session/create", u.Scheme, u.Host), strings.NewReader(postValues.Encode()))
+	if err != nil {
+		log.Error(err, "failed to generate request for session creation", "endpoint", ep.Host())
+		return err
 	}
 
 	req.Header.Add("Accept", "application/json")
 	req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
 
-	log.V(1).Info("creating session", "endpoint", ep, "request", req)
+	log.V(1).Info("creating auth session", "endpoint", ep, "request header", req.Header)
 	resp, err := ep.noBalancerClient.Do(req)
 	if err != nil {
 		log.Error(err, "session creation failed", "endpoint", u.Host)