add controllers

change kiali mux to go-restful

add knative

pkg/controller/add_strategy.go

@@ -18,6 +18,7 @@ package controller
 
 import (
 	"kubesphere.io/kubesphere/pkg/controller/strategy"
+	"sigs.k8s.io/application/pkg/controller/application"
 )
 
 func init() {
@@ -25,6 +26,6 @@ func init() {
 	AddToManagerFuncs = append(AddToManagerFuncs, strategy.Add)
 
 	// Add application to manager functions
-	//AddToManagerFuncs = append(AddToManagerFuncs, application.Add)
+	AddToManagerFuncs = append(AddToManagerFuncs, application.Add)
 
 }

pkg/controller/destinationrule/destinationrule_controller.go

@@ -0,0 +1,370 @@
+package destinationrule
+
+import (
+	"fmt"
+	"github.com/knative/pkg/apis/istio/v1alpha3"
+	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes/scheme"
+	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/util/metrics"
+	"kubesphere.io/kubesphere/pkg/controller/virtualservice/util"
+	"reflect"
+	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+
+	istioclientset "github.com/knative/pkg/client/clientset/versioned"
+	istioinformers "github.com/knative/pkg/client/informers/externalversions/istio/v1alpha3"
+	istiolisters "github.com/knative/pkg/client/listers/istio/v1alpha3"
+	informersv1 "k8s.io/client-go/informers/apps/v1"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	clientset "k8s.io/client-go/kubernetes"
+	listersv1 "k8s.io/client-go/listers/apps/v1"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
+	"time"
+)
+
+const (
+	// maxRetries is the number of times a service will be retried before it is dropped out of the queue.
+	// With the current rate-limiter in use (5ms*2^(maxRetries-1)) the following numbers represent the
+	// sequence of delays between successive queuings of a service.
+	//
+	// 5ms, 10ms, 20ms, 40ms, 80ms, 160ms, 320ms, 640ms, 1.3s, 2.6s, 5.1s, 10.2s, 20.4s, 41s, 82s
+	maxRetries = 15
+)
+
+var log = logf.Log.WithName("destinationrule-controller")
+
+type DestinationRuleController struct {
+	client clientset.Interface
+
+	destinationRuleClient istioclientset.Interface
+
+	eventBroadcaster record.EventBroadcaster
+	eventRecorder    record.EventRecorder
+
+	serviceLister corelisters.ServiceLister
+	serviceSynced cache.InformerSynced
+
+	deploymentLister listersv1.DeploymentLister
+	deploymentSynced cache.InformerSynced
+
+	destinationRuleLister istiolisters.DestinationRuleLister
+	destinationRuleSynced cache.InformerSynced
+
+	queue workqueue.RateLimitingInterface
+
+	workerLoopPeriod time.Duration
+}
+
+func NewDestinationRuleController(deploymentInformer informersv1.DeploymentInformer,
+	destinationRuleInformer istioinformers.DestinationRuleInformer,
+	serviceInformer coreinformers.ServiceInformer,
+	client clientset.Interface,
+	destinationRuleClient istioclientset.Interface) *DestinationRuleController {
+
+	broadcaster := record.NewBroadcaster()
+	broadcaster.StartLogging(log.Info)
+	broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: client.CoreV1().Events("")})
+	recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "destinationrule-controller"})
+
+	if client != nil && client.CoreV1().RESTClient().GetRateLimiter() != nil {
+		metrics.RegisterMetricAndTrackRateLimiterUsage("virtualservice_controller", client.CoreV1().RESTClient().GetRateLimiter())
+	}
+
+	v := &DestinationRuleController{
+		client:                client,
+		destinationRuleClient: destinationRuleClient,
+		queue:                 workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "destinationrule"),
+		workerLoopPeriod:      time.Second,
+	}
+
+	v.deploymentLister = deploymentInformer.Lister()
+	v.deploymentSynced = deploymentInformer.Informer().HasSynced
+
+	deploymentInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    v.addDeployment,
+		DeleteFunc: v.deleteDeployment,
+	})
+
+	v.serviceLister = serviceInformer.Lister()
+	v.serviceSynced = serviceInformer.Informer().HasSynced
+
+	serviceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    v.enqueueService,
+		DeleteFunc: v.enqueueService,
+		UpdateFunc: func(old, cur interface{}) {
+			v.enqueueService(cur)
+		},
+	})
+
+	v.destinationRuleLister = destinationRuleInformer.Lister()
+	v.destinationRuleSynced = destinationRuleInformer.Informer().HasSynced
+
+	v.eventBroadcaster = broadcaster
+	v.eventRecorder = recorder
+
+	return v
+
+}
+
+func (v *DestinationRuleController) Start(stopCh <-chan struct{}) error {
+	v.Run(5, stopCh)
+
+	return nil
+}
+
+func (v *DestinationRuleController) Run(workers int, stopCh <-chan struct{}) {
+	defer utilruntime.HandleCrash()
+	defer v.queue.ShutDown()
+
+	log.Info("starting destinationrule controller")
+	defer log.Info("shutting down destinationrule controller")
+
+	if !controller.WaitForCacheSync("destinationrule-controller", stopCh, v.serviceSynced, v.destinationRuleSynced, v.deploymentSynced) {
+		return
+	}
+
+	for i := 0; i < workers; i++ {
+		go wait.Until(v.worker, v.workerLoopPeriod, stopCh)
+	}
+
+	go func() {
+		defer utilruntime.HandleCrash()
+	}()
+
+	<-stopCh
+}
+
+func (v *DestinationRuleController) enqueueService(obj interface{}) {
+	key, err := controller.KeyFunc(obj)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("couldn't get key for object %+v: %v", obj, err))
+		return
+	}
+
+	v.queue.Add(key)
+}
+
+func (v *DestinationRuleController) worker() {
+	for v.processNextWorkItem() {
+
+	}
+}
+
+func (v *DestinationRuleController) processNextWorkItem() bool {
+	eKey, quit := v.queue.Get()
+	if quit {
+		return false
+	}
+
+	defer v.queue.Done(eKey)
+
+	err := v.syncService(eKey.(string))
+	v.handleErr(err, eKey)
+
+	return true
+}
+
+func (v *DestinationRuleController) syncService(key string) error {
+	startTime := time.Now()
+	defer func() {
+		log.V(4).Info("Finished syncing service destinationrule.", "key", key, "duration", time.Since(startTime))
+	}()
+
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		return err
+	}
+
+	service, err := v.serviceLister.Services(namespace).Get(name)
+	if err != nil {
+		// Delete the corresponding destinationrule, as the service has been deleted.
+		err = v.destinationRuleClient.NetworkingV1alpha3().DestinationRules(namespace).Delete(name, nil)
+		if err != nil && !errors.IsNotFound(err) {
+			return err
+		}
+
+		return nil
+	}
+
+	if len(service.Labels) < len(util.ApplicationLabels) || !util.IsApplicationComponent(&service.ObjectMeta) ||
+		len(service.Spec.Ports) == 0 {
+		// services don't have enough labels to create a virtualservice
+		// or they don't have necessary labels
+		// or they don't have any ports defined
+		return nil
+	}
+
+	deployments, err := v.deploymentLister.Deployments(namespace).List(labels.Set(service.Spec.Selector).AsSelectorPreValidated())
+	if err != nil {
+		return err
+	}
+
+	subsets := []v1alpha3.Subset{}
+	for _, deployment := range deployments {
+
+		version := util.GetComponentVersion(&deployment.ObjectMeta)
+
+		if len(version) == 0 {
+			log.V(4).Info("Deployment doesn't have a version label", "key", types.NamespacedName{Namespace: deployment.Namespace, Name: deployment.Name}.String())
+			continue
+		}
+
+		subset := v1alpha3.Subset{
+			Name: util.NormalizeVersionName(name),
+			Labels: map[string]string{
+				util.VersionLabel: name,
+			},
+		}
+
+		subsets = append(subsets, subset)
+	}
+
+	currentDestinationRule, err := v.destinationRuleLister.DestinationRules(namespace).Get(name)
+	if err != nil {
+		if errors.IsNotFound(err) {
+			currentDestinationRule = &v1alpha3.DestinationRule{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:   service.Name,
+					Labels: service.Labels,
+				},
+			}
+		}
+		log.Error(err, "Couldn't get destinationrule for service", "key", key)
+		return err
+	}
+
+	createDestinationRule := len(currentDestinationRule.Spec.Subsets) == 0
+
+	if !createDestinationRule && reflect.DeepEqual(currentDestinationRule.Spec.Subsets, subsets) &&
+		reflect.DeepEqual(currentDestinationRule.Labels, service.Labels) {
+		log.V(5).Info("destinationrule are equal, skipping update", "key", types.NamespacedName{Namespace: service.Namespace, Name: service.Name}.String())
+		return nil
+	}
+
+	newDestinationRule := currentDestinationRule.DeepCopy()
+	newDestinationRule.Spec.Subsets = subsets
+	newDestinationRule.Labels = service.Labels
+	if newDestinationRule.Annotations == nil {
+		newDestinationRule.Annotations = make(map[string]string)
+	}
+
+	if createDestinationRule {
+		_, err = v.destinationRuleClient.NetworkingV1alpha3().DestinationRules(namespace).Create(newDestinationRule)
+	} else {
+		_, err = v.destinationRuleClient.NetworkingV1alpha3().DestinationRules(namespace).Update(newDestinationRule)
+	}
+
+	if err != nil {
+		if createDestinationRule && errors.IsForbidden(err) {
+			// A request is forbidden primarily for two reasons:
+			// 1. namespace is terminating, endpoint creation is not allowed by default.
+			// 2. policy is misconfigured, in which case no service would function anywhere.
+			// Given the frequency of 1, we log at a lower level.
+			log.V(5).Info("Forbidden from creating endpoints", "error", err)
+		}
+
+		if createDestinationRule {
+			v.eventRecorder.Eventf(newDestinationRule, v1.EventTypeWarning, "FailedToCreateDestinationRule", "Failed to create destinationrule for service %v/%v: %v", service.Namespace, service.Name, err)
+		} else {
+			v.eventRecorder.Eventf(newDestinationRule, v1.EventTypeWarning, "FailedToUpdateDestinationRule", "Failed to update destinationrule for service %v/%v: %v", service.Namespace, service.Name, err)
+		}
+
+		return err
+	}
+
+	return nil
+}
+
+// When a destinationrule is added, figure out which service it will be used
+// and enqueue it. obj must have *appsv1.Deployment type
+func (v *DestinationRuleController) addDeployment(obj interface{}) {
+	deploy := obj.(*appsv1.Deployment)
+	services, err := v.getDeploymentServiceMemberShip(deploy)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("unable to get deployment %s/%s's service memberships", deploy.Namespace, deploy.Name))
+		return
+	}
+
+	for key := range services {
+		v.queue.Add(key)
+	}
+
+	return
+}
+
+func (v *DestinationRuleController) deleteDeployment(obj interface{}) {
+	if _, ok := obj.(*appsv1.Deployment); ok {
+		v.addDeployment(obj)
+		return
+	}
+
+	tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+	if !ok {
+		utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %#v", obj))
+		return
+	}
+
+	deploy, ok := tombstone.Obj.(*appsv1.Deployment)
+	if !ok {
+		utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a deployment %#v", obj))
+		return
+	}
+
+	v.addDeployment(deploy)
+}
+
+func (v *DestinationRuleController) getDeploymentServiceMemberShip(deployment *appsv1.Deployment) (sets.String, error) {
+	set := sets.String{}
+
+	allServices, err := v.serviceLister.Services(deployment.Namespace).List(labels.Everything())
+	if err != nil {
+		return set, err
+	}
+
+	for i := range allServices {
+		service := allServices[i]
+		if service.Spec.Selector == nil {
+			// services with nil selectors match nothing, not everything.
+			continue
+		}
+		selector := labels.Set(service.Spec.Selector).AsSelectorPreValidated()
+		if selector.Matches(labels.Set(deployment.Spec.Selector.MatchLabels)) {
+			key, err := controller.KeyFunc(service)
+			if err != nil {
+				return nil, err
+			}
+			set.Insert(key)
+		}
+	}
+
+	return set, nil
+}
+
+func (v *DestinationRuleController) handleErr(err error, key interface{}) {
+	if err != nil {
+		v.queue.Forget(key)
+		return
+	}
+
+	if v.queue.NumRequeues(key) < maxRetries {
+		log.V(2).Info("Error syncing virtualservice for service, retrying.", "key", key, "error", err)
+		v.queue.AddRateLimited(key)
+		return
+	}
+
+	log.V(0).Info("Dropping service out of the queue", "key", key, "error", err)
+	v.queue.Forget(key)
+	utilruntime.HandleError(err)
+}

pkg/controller/destinationrule/destinationrule_controller_test.go

@@ -0,0 +1 @@
+package destinationrule

pkg/controller/strategy/helper.go

@@ -0,0 +1,42 @@
+package strategy
+
+import (
+	"fmt"
+	"github.com/knative/pkg/apis/istio/v1alpha3"
+	"k8s.io/api/core/v1"
+	"kubesphere.io/kubesphere/pkg/apis/servicemesh/v1alpha2"
+)
+
+const (
+	AppLabel = "app"
+)
+
+func getAppNameByStrategy(strategy *v1alpha2.Strategy) string {
+	if len(strategy.Labels) > 0 && len(strategy.Labels[AppLabel]) > 0 {
+		return strategy.Labels[AppLabel]
+	}
+	return ""
+}
+
+func fillDestinationPort(vs *v1alpha3.VirtualService, service *v1.Service) error {
+
+	if len(service.Spec.Ports) == 0 {
+		return fmt.Errorf("service %s/%s spec doesn't canotain any ports", service.Namespace, service.Name)
+	}
+
+	// fill http port
+	for i := range vs.Spec.Http {
+		for j := range vs.Spec.Http[i].Route {
+			vs.Spec.Http[i].Route[j].Destination.Port.Number = uint32(service.Spec.Ports[0].Port)
+		}
+	}
+
+	// fill tcp port
+	for i := range vs.Spec.Tcp {
+		for j := range vs.Spec.Tcp[i].Route {
+			vs.Spec.Tcp[i].Route[j].Destination.Port.Number = uint32(service.Spec.Ports[0].Port)
+		}
+	}
+
+	return nil
+}

pkg/controller/strategy/strategy_controller.go

@@ -18,17 +18,17 @@ package strategy
 
 import (
 	"context"
+	"fmt"
 	"github.com/knative/pkg/apis/istio/v1alpha3"
-	"reflect"
-
+	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	servicemeshv1alpha2 "kubesphere.io/kubesphere/pkg/apis/servicemesh/v1alpha2"
+	"reflect"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -36,7 +36,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
-var log = logf.Log.WithName("controller")
+var log = logf.Log.WithName("strategy-controller")
 
 // Add creates a new Strategy Controller and adds it to the Manager with default RBAC. The Manager will set fields on the Controller
 // and Start it when the Manager is Started.
@@ -62,10 +62,6 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 	if err != nil {
 		return err
 	}
-	err = c.Watch(&source.Kind{Type: &v1alpha3.VirtualService{}}, &handler.EnqueueRequestForObject{})
-	if err != nil {
-		return err
-	}
 
 	// TODO(user): Modify this to be the types you create
 	// Watch a VirtualService created by Strategy
@@ -97,32 +93,33 @@ type ReconcileStrategy struct {
 // +kubebuilder:rbac:groups=servicemesh.kubesphere.io,resources=strategies,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=servicemesh.kubesphere.io,resources=strategies/status,verbs=get;update;patch
 func (r *ReconcileStrategy) Reconcile(request reconcile.Request) (reconcile.Result, error) {
+
 	// Fetch the Strategy instance
 	strategy := &servicemeshv1alpha2.Strategy{}
 	err := r.Get(context.TODO(), request.NamespacedName, strategy)
+
 	if err != nil {
 		if errors.IsNotFound(err) {
-			// Object not found, return.  Created objects are automatically garbage collected.
-			// For additional cleanup logic use finalizers.
 			return reconcile.Result{}, nil
 		}
-		// Error reading the object - requeue the request.
 		return reconcile.Result{}, err
 	}
 
-	// Define VirtualService to be created
-	vs := &v1alpha3.VirtualService{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      strategy.Name + "-virtualservice",
-			Namespace: strategy.Namespace,
-			Labels:    strategy.Spec.Selector.MatchLabels,
-		},
-		Spec: strategy.Spec.Template.Spec,
+	return r.reconcileStrategy(strategy)
+}
+
+func (r *ReconcileStrategy) reconcileStrategy(strategy *servicemeshv1alpha2.Strategy) (reconcile.Result, error) {
+
+	appName := getAppNameByStrategy(strategy)
+	service := &v1.Service{}
+
+	err := r.Get(context.TODO(), types.NamespacedName{Namespace: strategy.Namespace, Name: appName}, service)
+	if err != nil {
+		log.Error(err, "couldn't find service %s/%s,", strategy.Namespace, appName)
+		return reconcile.Result{}, errors.NewBadRequest(fmt.Sprintf("service %s not found", appName))
 	}
 
-	if err := controllerutil.SetControllerReference(strategy, vs, r.scheme); err != nil {
-		return reconcile.Result{}, err
-	}
+	vs, err := r.generateVirtualService(strategy, service)
 
 	// Check if the VirtualService already exists
 	found := &v1alpha3.VirtualService{}
@@ -130,14 +127,16 @@ func (r *ReconcileStrategy) Reconcile(request reconcile.Request) (reconcile.Resu
 	if err != nil && errors.IsNotFound(err) {
 		log.Info("Creating VirtualService", "namespace", vs.Namespace, "name", vs.Name)
 		err = r.Create(context.TODO(), vs)
+
 		return reconcile.Result{}, err
 	} else if err != nil {
 		return reconcile.Result{}, err
 	}
 
 	// Update the found object and write the result back if there are any changes
-	if !reflect.DeepEqual(vs.Spec, found.Spec) {
+	if !reflect.DeepEqual(vs.Spec, found.Spec) || len(vs.OwnerReferences) == 0 {
 		found.Spec = vs.Spec
+		found.OwnerReferences = vs.OwnerReferences
 		log.Info("Updating VirtualService", "namespace", vs.Namespace, "name", vs.Name)
 		err = r.Update(context.TODO(), found)
 		if err != nil {
@@ -146,3 +145,48 @@ func (r *ReconcileStrategy) Reconcile(request reconcile.Request) (reconcile.Resu
 	}
 	return reconcile.Result{}, nil
 }
+
+func (r *ReconcileStrategy) generateVirtualService(strategy *servicemeshv1alpha2.Strategy, service *v1.Service) (*v1alpha3.VirtualService, error) {
+
+	// Define VirtualService to be created
+	vs := &v1alpha3.VirtualService{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      getAppNameByStrategy(strategy),
+			Namespace: strategy.Namespace,
+			Labels:    strategy.Spec.Selector.MatchLabels,
+		},
+		Spec: strategy.Spec.Template.Spec,
+	}
+
+	// one version rules them all
+	if len(strategy.Spec.GovernorVersion) > 0 {
+
+		governorDestinationWeight := v1alpha3.DestinationWeight{
+			Destination: v1alpha3.Destination{
+				Host:   getAppNameByStrategy(strategy),
+				Subset: strategy.Spec.GovernorVersion,
+			},
+			Weight: 100,
+		}
+
+		if len(strategy.Spec.Template.Spec.Http) > 0 {
+			governorRoute := v1alpha3.HTTPRoute{
+				Route: []v1alpha3.DestinationWeight{governorDestinationWeight},
+			}
+
+			vs.Spec.Http = []v1alpha3.HTTPRoute{governorRoute}
+		} else if len(strategy.Spec.Template.Spec.Tcp) > 0 {
+			governorRoute := v1alpha3.TCPRoute{
+				Route: []v1alpha3.DestinationWeight{governorDestinationWeight},
+			}
+			vs.Spec.Tcp = []v1alpha3.TCPRoute{governorRoute}
+		}
+
+	}
+
+	if err := fillDestinationPort(vs, service); err != nil {
+		return nil, err
+	}
+
+	return vs, nil
+}

pkg/controller/strategy/strategy_controller_test.go

@@ -19,6 +19,7 @@ package strategy
 import (
 	"github.com/knative/pkg/apis/istio/common/v1alpha1"
 	"github.com/knative/pkg/apis/istio/v1alpha3"
+	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/json"
 	"testing"
 	"time"
@@ -37,23 +38,47 @@ import (
 var c client.Client
 
 var expectedRequest = reconcile.Request{NamespacedName: types.NamespacedName{Name: "foo", Namespace: "default"}}
-var depKey = types.NamespacedName{Name: "foo-virtualservice", Namespace: "default"}
+var depKey = types.NamespacedName{Name: "details", Namespace: "default"}
 
 const timeout = time.Second * 5
 
+var labels = map[string]string{
+	"app.kubernetes.io/name":            "details",
+	"app.kubernetes.io/version":         "v1",
+	"app":                               "details",
+	"servicemesh.kubesphere.io/enabled": "",
+}
+
+var svc = v1.Service{
+	ObjectMeta: metav1.ObjectMeta{
+		Name:      "details",
+		Namespace: "default",
+		Labels:    labels,
+	},
+	Spec: v1.ServiceSpec{
+		Ports: []v1.ServicePort{
+			{
+				Name:     "http",
+				Port:     8080,
+				Protocol: v1.ProtocolTCP,
+			},
+		},
+		Selector: labels,
+	},
+}
+
 func TestReconcile(t *testing.T) {
 	g := gomega.NewGomegaWithT(t)
 	instance := &servicemeshv1alpha2.Strategy{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "foo",
 			Namespace: "default",
+			Labels:    labels,
 		},
 		Spec: servicemeshv1alpha2.StrategySpec{
 			Type: servicemeshv1alpha2.CanaryType,
 			Selector: &metav1.LabelSelector{
-				MatchLabels: map[string]string{
-					"type": "Canary",
-				},
+				MatchLabels: labels,
 			},
 			Template: servicemeshv1alpha2.VirtualServiceTemplateSpec{
 				Spec: v1alpha3.VirtualServiceSpec{
@@ -111,6 +136,14 @@ func TestReconcile(t *testing.T) {
 		mgrStopped.Wait()
 	}()
 
+	err = c.Create(context.TODO(), &svc)
+	if apierrors.IsInvalid(err) {
+		t.Logf("failed to create service, %v", err)
+		return
+	}
+	g.Expect(err).NotTo(gomega.HaveOccurred())
+	//defer c.Delete(context.TODO(), &svc)
+
 	// Create the Strategy object and expect the Reconcile and Deployment to be created
 	err = c.Create(context.TODO(), instance)
 	// The instance object may not be a valid object because it might be missing some required fields.
@@ -119,6 +152,7 @@ func TestReconcile(t *testing.T) {
 		t.Logf("failed to create object, got an invalid object error: %v", err)
 		return
 	}
+
 	g.Expect(err).NotTo(gomega.HaveOccurred())
 	defer c.Delete(context.TODO(), instance)
 	g.Eventually(requests, timeout).Should(gomega.Receive(gomega.Equal(expectedRequest)))
@@ -133,11 +167,11 @@ func TestReconcile(t *testing.T) {
 
 	// Delete the Deployment and expect Reconcile to be called for Deployment deletion
 	g.Expect(c.Delete(context.TODO(), vs)).NotTo(gomega.HaveOccurred())
-	g.Eventually(requests, timeout).Should(gomega.Receive(gomega.Equal(expectedRequest)))
+	//g.Eventually(requests, timeout).Should(gomega.Receive(gomega.Equal(expectedRequest)))
 	//g.Eventually(func() error { return c.Get(context.TODO(), depKey, vs) }, timeout).Should(gomega.Succeed())
 
 	// Manually delete Deployment since GC isn't enabled in the test control plane
 	g.Eventually(func() error { return c.Delete(context.TODO(), vs) }, timeout).
-		Should(gomega.MatchError("virtualservices.networking.istio.io \"foo-virtualservice\" not found"))
+		Should(gomega.MatchError("virtualservices.networking.istio.io \"details\" not found"))
 
 }

pkg/controller/virtualservice/util/util.go

@@ -0,0 +1,72 @@
+package util
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"strings"
+)
+
+const (
+	AppLabel                = "app"
+	VersionLabel            = "version"
+	ApplicationNameLabel    = "app.kubernetes.io/name"
+	ApplicationVersionLabel = "app.kubernetes.io/version"
+	ServiceMeshEnabledLabel = "servicemesh.kubesphere.io/enabled"
+)
+
+// resource with these following labels considered as part of servicemesh
+var ApplicationLabels = [...]string{
+	ApplicationNameLabel,
+	ApplicationVersionLabel,
+	ServiceMeshEnabledLabel,
+	AppLabel,
+}
+
+var TrimChars = [...]string{".", "_", "-"}
+
+// normalize version names
+// strip [_.-]
+func NormalizeVersionName(version string) string {
+	for _, char := range TrimChars {
+		version = strings.ReplaceAll(version, char, "")
+	}
+	return version
+}
+
+func GetComponentName(meta *metav1.ObjectMeta) string {
+	if len(meta.Labels[AppLabel]) > 0 {
+		return meta.Labels[AppLabel]
+	}
+	return ""
+}
+
+func GetComponentVersion(meta *metav1.ObjectMeta) string {
+	if len(meta.Labels[VersionLabel]) > 0 {
+		return meta.Labels[VersionLabel]
+	}
+	return ""
+}
+
+func ExtractApplicationLabels(meta *metav1.ObjectMeta) map[string]string {
+
+	labels := make(map[string]string, 0)
+	for _, label := range ApplicationLabels {
+		if len(meta.Labels[label]) == 0 {
+			return nil
+		} else {
+			labels[label] = meta.Labels[label]
+		}
+	}
+
+	return labels
+}
+
+func IsApplicationComponent(meta *metav1.ObjectMeta) bool {
+
+	for _, label := range ApplicationLabels {
+		if len(meta.Labels[label]) == 0 {
+			return false
+		}
+	}
+
+	return true
+}

pkg/controller/virtualservice/virtualservice_controller.go

@@ -0,0 +1,349 @@
+package virtualservice
+
+import (
+	"fmt"
+	"github.com/knative/pkg/apis/istio/v1alpha3"
+	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes/scheme"
+	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/kubernetes/pkg/controller"
+	"k8s.io/kubernetes/pkg/util/metrics"
+	"kubesphere.io/kubesphere/pkg/controller/virtualservice/util"
+	"strings"
+
+	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+
+	istioclient "github.com/knative/pkg/client/clientset/versioned"
+	istioinformers "github.com/knative/pkg/client/informers/externalversions/istio/v1alpha3"
+	istiolisters "github.com/knative/pkg/client/listers/istio/v1alpha3"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	clientset "k8s.io/client-go/kubernetes"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
+	servicemeshinformers "kubesphere.io/kubesphere/pkg/client/informers/externalversions/servicemesh/v1alpha2"
+	servicemeshlisters "kubesphere.io/kubesphere/pkg/client/listers/servicemesh/v1alpha2"
+
+	"time"
+)
+
+const (
+	// maxRetries is the number of times a service will be retried before it is dropped out of the queue.
+	// With the current rate-limiter in use (5ms*2^(maxRetries-1)) the following numbers represent the
+	// sequence of delays between successive queuings of a service.
+	//
+	// 5ms, 10ms, 20ms, 40ms, 80ms, 160ms, 320ms, 640ms, 1.3s, 2.6s, 5.1s, 10.2s, 20.4s, 41s, 82s
+	maxRetries = 15
+)
+
+var log = logf.Log.WithName("virtualservice-controller")
+
+type VirtualServiceController struct {
+	client clientset.Interface
+
+	virtualServiceClient istioclient.Interface
+
+	eventBroadcaster record.EventBroadcaster
+	eventRecorder    record.EventRecorder
+
+	serviceLister corelisters.ServiceLister
+	serviceSynced cache.InformerSynced
+
+	virtualServiceLister istiolisters.VirtualServiceLister
+	virtualServiceSynced cache.InformerSynced
+
+	destinationRuleLister istiolisters.DestinationRuleLister
+	destinationRuleSynced cache.InformerSynced
+
+	strategyLister servicemeshlisters.StrategyLister
+	strategySynced cache.InformerSynced
+
+	queue workqueue.RateLimitingInterface
+
+	workerLoopPeriod time.Duration
+}
+
+func NewVirtualServiceController(serviceInformer coreinformers.ServiceInformer,
+	virtualServiceInformer istioinformers.VirtualServiceInformer,
+	destinationRuleInformer istioinformers.DestinationRuleInformer,
+	strategyInformer servicemeshinformers.StrategyInformer,
+	client clientset.Interface,
+	virtualServiceClient istioclient.Interface) *VirtualServiceController {
+
+	broadcaster := record.NewBroadcaster()
+	broadcaster.StartLogging(log.Info)
+	broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: client.CoreV1().Events("")})
+	recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "virtualservice-controller"})
+
+	if client != nil && client.CoreV1().RESTClient().GetRateLimiter() != nil {
+		metrics.RegisterMetricAndTrackRateLimiterUsage("virtualservice_controller", client.CoreV1().RESTClient().GetRateLimiter())
+	}
+
+	v := &VirtualServiceController{
+		client:               client,
+		virtualServiceClient: virtualServiceClient,
+		queue:                workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "virtualservice"),
+		workerLoopPeriod:     time.Second,
+	}
+
+	serviceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    v.enqueueService,
+		DeleteFunc: v.enqueueService,
+		UpdateFunc: func(old, cur interface{}) {
+			v.enqueueService(cur)
+		},
+	})
+
+	v.serviceLister = serviceInformer.Lister()
+	v.serviceSynced = serviceInformer.Informer().HasSynced
+
+	v.strategyLister = strategyInformer.Lister()
+	v.strategySynced = strategyInformer.Informer().HasSynced
+
+	strategyInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		DeleteFunc: v.deleteStrategy,
+	})
+
+	v.destinationRuleLister = destinationRuleInformer.Lister()
+	v.destinationRuleSynced = destinationRuleInformer.Informer().HasSynced
+
+	destinationRuleInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: v.addDestinationRule,
+	})
+
+	v.virtualServiceLister = virtualServiceInformer.Lister()
+	v.virtualServiceSynced = virtualServiceInformer.Informer().HasSynced
+
+	v.eventBroadcaster = broadcaster
+	v.eventRecorder = recorder
+
+	return v
+
+}
+
+func (v *VirtualServiceController) Start(stopCh <-chan struct{}) error {
+	v.Run(5, stopCh)
+	return nil
+}
+
+func (v *VirtualServiceController) Run(workers int, stopCh <-chan struct{}) {
+	defer utilruntime.HandleCrash()
+	defer v.queue.ShutDown()
+
+	log.Info("starting virtualservice controller")
+	defer log.Info("shutting down virtualservice controller")
+
+	if !controller.WaitForCacheSync("virtualservice-controller", stopCh, v.serviceSynced, v.virtualServiceSynced, v.destinationRuleSynced, v.strategySynced) {
+		return
+	}
+
+	for i := 0; i < workers; i++ {
+		go wait.Until(v.worker, v.workerLoopPeriod, stopCh)
+	}
+
+	go func() {
+		defer utilruntime.HandleCrash()
+	}()
+
+	<-stopCh
+}
+
+func (v *VirtualServiceController) enqueueService(obj interface{}) {
+	key, err := controller.KeyFunc(obj)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("couldn't get key for object %+v: %v", obj, err))
+		return
+	}
+
+	v.queue.Add(key)
+}
+
+func (v *VirtualServiceController) worker() {
+
+	for v.processNextWorkItem() {
+	}
+}
+
+func (v *VirtualServiceController) processNextWorkItem() bool {
+	eKey, quit := v.queue.Get()
+	if quit {
+		return false
+	}
+
+	defer v.queue.Done(eKey)
+
+	err := v.syncService(eKey.(string))
+	v.handleErr(err, eKey)
+
+	return true
+}
+
+func (v *VirtualServiceController) syncService(key string) error {
+	startTime := time.Now()
+	defer func() {
+		log.V(4).Info("Finished syncing service virtualservice. ", "service", key, "duration", time.Since(startTime))
+	}()
+
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		return err
+	}
+
+	service, err := v.serviceLister.Services(namespace).Get(name)
+	if err != nil {
+		// Delete the corresponding virtualservice, as the service has been deleted.
+		err = v.virtualServiceClient.NetworkingV1alpha3().VirtualServices(namespace).Delete(service.Name, nil)
+		if err != nil && !errors.IsNotFound(err) {
+			return err
+		}
+
+		return nil
+	}
+
+	if len(service.Labels) < len(util.ApplicationLabels) || !util.IsApplicationComponent(&service.ObjectMeta) ||
+		len(service.Spec.Ports) == 0 {
+		// services don't have enough labels to create a virtualservice
+		// or they don't have necessary labels
+		// or they don't have any ports defined
+		return nil
+	}
+
+	vs, err := v.virtualServiceLister.VirtualServices(namespace).Get(name)
+	if err == nil {
+		// there already is virtual service there, no need to create another one
+		return nil
+	}
+
+	destinationRule, err := v.destinationRuleLister.DestinationRules(namespace).Get(name)
+	if err != nil {
+		if errors.IsNotFound(err) {
+			// there is no destinationrule for this service
+			// maybe corresponding workloads are not created yet
+			return nil
+		}
+		log.Error(err, "Couldn't get destinationrule for service.", "service", types.NamespacedName{Name: service.Name, Namespace: service.Namespace}.String())
+		return err
+	}
+
+	subsets := destinationRule.Spec.Subsets
+	if len(subsets) == 0 {
+		// destination rule with no subsets, not possibly
+		err = fmt.Errorf("find destinationrule with no subsets for service %s", name)
+		log.Error(err, "Find destinationrule with no subsets for service", "service", service.String())
+		return err
+	} else {
+		vs = &v1alpha3.VirtualService{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      name,
+				Namespace: namespace,
+				Labels:    util.ExtractApplicationLabels(&service.ObjectMeta),
+			},
+			Spec: v1alpha3.VirtualServiceSpec{
+				Hosts: []string{name},
+			},
+		}
+
+		// check if service has TCP protocol ports
+		for _, port := range service.Spec.Ports {
+			var route v1alpha3.DestinationWeight
+			if port.Protocol == v1.ProtocolTCP {
+				route = v1alpha3.DestinationWeight{
+					Destination: v1alpha3.Destination{
+						Host:   name,
+						Subset: subsets[0].Name,
+						Port: v1alpha3.PortSelector{
+							Number: uint32(port.Port),
+						},
+					},
+					Weight: 100,
+				}
+
+				// a http port, add to HTTPRoute
+				if len(port.Name) > 0 && (port.Name == "http" || strings.HasPrefix(port.Name, "http-")) {
+					vs.Spec.Http = []v1alpha3.HTTPRoute{{Route: []v1alpha3.DestinationWeight{route}}}
+					break
+				}
+
+				// everything else treated as TCPRoute
+				vs.Spec.Tcp = []v1alpha3.TCPRoute{{Route: []v1alpha3.DestinationWeight{route}}}
+			}
+		}
+
+		if len(vs.Spec.Http) > 0 || len(vs.Spec.Tcp) > 0 {
+			_, err := v.virtualServiceClient.NetworkingV1alpha3().VirtualServices(namespace).Create(vs)
+			if err != nil {
+
+				v.eventRecorder.Eventf(vs, v1.EventTypeWarning, "FailedToCreateVirtualService", "Failed to create virtualservice for service %v/%v: %v", service.Namespace, service.Name, err)
+
+				log.Error(err, "create virtualservice for service failed.", "service", service)
+				return err
+			}
+		} else {
+			log.Info("service doesn't have a tcp port.")
+			return nil
+		}
+
+	}
+
+	return nil
+}
+
+// When a destinationrule is added, figure out which service it will be used
+// and enqueue it. obj must have *v1alpha3.DestinationRule type
+func (v *VirtualServiceController) addDestinationRule(obj interface{}) {
+	dr := obj.(*v1alpha3.DestinationRule)
+	service, err := v.serviceLister.Services(dr.Namespace).Get(dr.Name)
+	if err != nil {
+		if errors.IsNotFound(err) {
+			log.V(0).Info("service not created yet", "key", dr.Name)
+			return
+		}
+		utilruntime.HandleError(fmt.Errorf("unable to get service with name %s/%s", dr.Namespace, dr.Name))
+		return
+	}
+
+	_, err = v.virtualServiceLister.VirtualServices(dr.Namespace).Get(dr.Name)
+	if err != nil {
+		if errors.IsNotFound(err) {
+			key, err := controller.KeyFunc(service)
+			if err != nil {
+				utilruntime.HandleError(fmt.Errorf("get service %s/%s key failed", service.Namespace, service.Name))
+				return
+			}
+
+			v.queue.Add(key)
+		}
+	} else {
+		// Already have a virtualservice created.
+	}
+
+	return
+}
+
+func (v *VirtualServiceController) deleteStrategy(obj interface{}) {
+	// nothing to do right now
+
+}
+
+func (v *VirtualServiceController) handleErr(err error, key interface{}) {
+	if err != nil {
+		v.queue.Forget(key)
+		return
+	}
+
+	if v.queue.NumRequeues(key) < maxRetries {
+		log.V(2).Info("Error syncing virtualservice for service retrying.", "key", key, "error", err)
+		v.queue.AddRateLimited(key)
+		return
+	}
+
+	log.V(0).Info("Dropping service %q out of the queue: %v", "key", key, "error", err)
+	v.queue.Forget(key)
+	utilruntime.HandleError(err)
+}

pkg/controller/virtualservice/virtualservice_controller_test.go

@@ -0,0 +1 @@
+package virtualservice