Merge branch 'feature/testbility-for-cluster'

* feature/testbility-for-cluster:
  improve: remove go master from travis
  improve add test case for util
  fixed code error
  improve: rename shadow interface
  improve: create shadow interface
  improve: clean code
  improve: use kubernetes interface in mesh command
  improve: 🧪 refactor command exchange use kubernetes interface
  feat: #89 refactor command exchange use kubernetes interface
  improve: #89 refactor command connect use kubernetes interface
  improve: 🧪 clean code
  add cluster interface

.travis.yml

@@ -2,7 +2,6 @@ language: go
 
 go:
   - 1.13.x
-  - master
 
 services:
 - docker

pkg/kt/cluster/helper.go

@@ -1,26 +1,24 @@
-package util
+package cluster
 
 import (
 	"fmt"
 	"strings"
 
-	"github.com/deckarep/golang-set"
+	mapset "github.com/deckarep/golang-set"
 	"github.com/rs/zerolog/log"
+	coreV1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/clientcmd"
 )
 
-// GetCirds Get kubernetes cluster resource crids
-func GetCirds(clientset *kubernetes.Clientset, podCIDR string) (cidrs []string, err error) {
-	cidrs, err = getPodCirds(clientset, podCIDR)
+// GetKubernetesClient get Kubernetes client from config
+func GetKubernetesClient(kubeConfig string) (clientset *kubernetes.Clientset, err error) {
+	config, err := clientcmd.BuildConfigFromFlags("", kubeConfig)
 	if err != nil {
-		return
+		return nil, err
 	}
-	serviceCird, err := getServiceCird(clientset)
-	if err != nil {
-		return
-	}
-	cidrs = append(cidrs, serviceCird...)
+	clientset, err = kubernetes.NewForConfig(config)
 	return
 }
 
@@ -46,20 +44,11 @@ func getPodCirds(clientset *kubernetes.Clientset, podCIDR string) (cidrs []strin
 	}
 
 	if len(cidrs) == 0 {
-		log.Info().Msgf("Fail to get pod cidr from node.Spec.PODCIDR, try to get with pod sample")
-		podList, err2 := clientset.CoreV1().Pods("").List(metav1.ListOptions{})
+		samples, err2 := getPodCirdByInstance(clientset)
 		if err2 != nil {
-			log.Printf("Fails to get service info of cluster")
+			err = err2
 			return
 		}
-
-		samples := mapset.NewSet()
-		for _, pod := range podList.Items {
-			if pod.Status.PodIP != "" && pod.Status.PodIP != "None" {
-				samples.Add(getCirdFromSample(pod.Status.PodIP))
-			}
-		}
-
 		for _, sample := range samples.ToSlice() {
 			cidrs = append(cidrs, fmt.Sprint(sample))
 		}
@@ -68,15 +57,26 @@ func getPodCirds(clientset *kubernetes.Clientset, podCIDR string) (cidrs []strin
 	return
 }
 
-func getServiceCird(clientset *kubernetes.Clientset) (cidr []string, err error) {
-	serviceList, err := clientset.CoreV1().Services("").List(metav1.ListOptions{})
+func getPodCirdByInstance(clientset *kubernetes.Clientset) (samples mapset.Set, err error) {
+	log.Info().Msgf("Fail to get pod cidr from node.Spec.PODCIDR, try to get with pod sample")
+	podList, err := clientset.CoreV1().Pods("").List(metav1.ListOptions{})
 	if err != nil {
 		log.Printf("Fails to get service info of cluster")
-		return cidr, err
+		return
 	}
 
+	samples = mapset.NewSet()
+	for _, pod := range podList.Items {
+		if pod.Status.PodIP != "" && pod.Status.PodIP != "None" {
+			samples.Add(getCirdFromSample(pod.Status.PodIP))
+		}
+	}
+	return
+}
+
+func getServiceCird(serviceList []*coreV1.Service) (cidr []string, err error) {
 	samples := mapset.NewSet()
-	for _, service := range serviceList.Items {
+	for _, service := range serviceList {
 		if service.Spec.ClusterIP != "" && service.Spec.ClusterIP != "None" {
 			samples.Add(getCirdFromSample(service.Spec.ClusterIP))
 		}
@@ -85,7 +85,6 @@ func getServiceCird(clientset *kubernetes.Clientset) (cidr []string, err error)
 	for _, sample := range samples.ToSlice() {
 		cidr = append(cidr, fmt.Sprint(sample))
 	}
-
 	return
 }
 

pkg/kt/cluster/kubernetes.go

@@ -8,34 +8,77 @@ import (
 	clusterWatcher "github.com/alibaba/kt-connect/pkg/apiserver/cluster"
 	"github.com/alibaba/kt-connect/pkg/kt/util"
 	"github.com/rs/zerolog/log"
-	appsv1 "k8s.io/api/apps/v1"
-	apiv1 "k8s.io/api/core/v1"
+	appV1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/tools/clientcmd"
 )
 
-// Signal structure
-type Signal struct {
+// Scale scale deployment to
+func (k *Kubernetes) Scale(deployment *appV1.Deployment, replicas *int32) (err error) {
+	log.Printf("scale deployment %s to %d\n", deployment.GetObjectMeta().GetName(), *replicas)
+	client := k.Clientset.AppsV1().Deployments(deployment.GetObjectMeta().GetNamespace())
+	deployment.Spec.Replicas = replicas
+
+	d, err := client.Update(deployment)
+	if err != nil {
+		log.Printf("%s Fails scale deployment %s to %d\n", err.Error(), deployment.GetObjectMeta().GetName(), *replicas)
+		return
+	}
+	log.Printf(" * %s (%d replicas) success", d.Name, *d.Spec.Replicas)
+	return
 }
 
-// GetKubernetesClient get Kubernetes client from config
-func GetKubernetesClient(kubeConfig string) (clientset *kubernetes.Clientset, err error) {
-	config, err := clientcmd.BuildConfigFromFlags("", kubeConfig)
+// Deployment get deployment
+func (k *Kubernetes) Deployment(name, namespace string) (deployment *appV1.Deployment, err error) {
+	deployment, err = k.Clientset.AppsV1().Deployments(namespace).Get(name, metaV1.GetOptions{})
+	return
+}
+
+// CreateShadow create shadow
+func (k *Kubernetes) CreateShadow(name, namespace, image string, labels map[string]string) (podIP, podName string, err error) {
+	return CreateShadow(k.Clientset, name, labels, namespace, image)
+}
+
+// ClusterCrids get cluster cirds
+func (k *Kubernetes) ClusterCrids(podCIDR string) (cidrs []string, err error) {
+	serviceList, err := k.ServiceListener.List(labels.Everything())
 	if err != nil {
-		return nil, err
+		return
+	}
+
+	cidrs, err = getPodCirds(k.Clientset, podCIDR)
+	if err != nil {
+		return
+	}
+
+	serviceCird, err := getServiceCird(serviceList)
+	if err != nil {
+		return
+	}
+	cidrs = append(cidrs, serviceCird...)
+	return
+}
+
+// ServiceHosts get service dns map
+func (k *Kubernetes) ServiceHosts(namespace string) (hosts map[string]string) {
+	services, err := k.ServiceListener.Services(namespace).List(labels.Everything())
+	if err != nil {
+		return
+	}
+	hosts = map[string]string{}
+	for _, service := range services {
+		hosts[service.ObjectMeta.Name] = service.Spec.ClusterIP
 	}
-	clientset, err = kubernetes.NewForConfig(config)
 	return
 }
 
 // ScaleTo scale app
 func ScaleTo(clientSet *kubernetes.Clientset, namespace, name string, replicas int32) (err error) {
 	client := clientSet.AppsV1().Deployments(namespace)
-	deployment, err := client.Get(name, metav1.GetOptions{})
+	deployment, err := client.Get(name, metaV1.GetOptions{})
 	if err != nil {
 		return
 	}
@@ -55,8 +98,8 @@ func ScaleTo(clientSet *kubernetes.Clientset, namespace, name string, replicas i
 // RemoveShadow remove shadow from cluster
 func RemoveShadow(client *kubernetes.Clientset, namespace, name string) {
 	deploymentsClient := client.AppsV1().Deployments(namespace)
-	deletePolicy := metav1.DeletePropagationBackground
-	err := deploymentsClient.Delete(name, &metav1.DeleteOptions{
+	deletePolicy := metaV1.DeletePropagationBackground
+	err := deploymentsClient.Delete(name, &metaV1.DeleteOptions{
 		PropagationPolicy: &deletePolicy,
 	})
 	if err != nil {
@@ -82,7 +125,7 @@ func RemoveService(
 	clientset *kubernetes.Clientset,
 ) (err error) {
 	client := clientset.CoreV1().Services(namespace)
-	return client.Delete(name, &metav1.DeleteOptions{})
+	return client.Delete(name, &metaV1.DeleteOptions{})
 }
 
 // CreateShadow create shadow
@@ -116,12 +159,12 @@ func CreateShadow(
 	return
 }
 
-func waitPodReadyUsingInformer(namespace, name string, clientset *kubernetes.Clientset) (pod apiv1.Pod, err error) {
+func waitPodReadyUsingInformer(namespace, name string, clientset *kubernetes.Clientset) (pod v1.Pod, err error) {
 	podListener, err := clusterWatcher.PodListener(clientset)
 	if err != nil {
 		return
 	}
-	pod = apiv1.Pod{}
+	pod = v1.Pod{}
 	podLabels := labels.NewSelector()
 	log.Info().Msgf("pod label: kt=%s", name)
 	labelKeys := []string{
@@ -195,7 +238,7 @@ func generateService(name, namespace string, labels map[string]string, port int)
 	})
 
 	return &v1.Service{
-		ObjectMeta: metav1.ObjectMeta{
+		ObjectMeta: metaV1.ObjectMeta{
 			Name:      name,
 			Namespace: namespace,
 			Labels:    labels,
@@ -209,23 +252,23 @@ func generateService(name, namespace string, labels map[string]string, port int)
 
 }
 
-func generatorDeployment(namespace, name string, labels map[string]string, image string) *appsv1.Deployment {
-	return &appsv1.Deployment{
-		ObjectMeta: metav1.ObjectMeta{
+func generatorDeployment(namespace, name string, labels map[string]string, image string) *appV1.Deployment {
+	return &appV1.Deployment{
+		ObjectMeta: metaV1.ObjectMeta{
 			Name:      name,
 			Namespace: namespace,
 			Labels:    labels,
 		},
-		Spec: appsv1.DeploymentSpec{
-			Selector: &metav1.LabelSelector{
+		Spec: appV1.DeploymentSpec{
+			Selector: &metaV1.LabelSelector{
 				MatchLabels: labels,
 			},
-			Template: apiv1.PodTemplateSpec{
-				ObjectMeta: metav1.ObjectMeta{
+			Template: v1.PodTemplateSpec{
+				ObjectMeta: metaV1.ObjectMeta{
 					Labels: labels,
 				},
-				Spec: apiv1.PodSpec{
-					Containers: []apiv1.Container{
+				Spec: v1.PodSpec{
+					Containers: []v1.Container{
 						{
 							Name:            "standalone",
 							Image:           image,
@@ -237,22 +280,3 @@ func generatorDeployment(namespace, name string, labels map[string]string, image
 		},
 	}
 }
-
-// LocalHosts LocalHosts
-func LocalHosts(clientset *kubernetes.Clientset, namespace string) (hosts map[string]string) {
-	serviceListener, err := clusterWatcher.ServiceListener(clientset)
-	if err != nil {
-		return
-	}
-
-	services, err := serviceListener.Services(namespace).List(labels.Everything())
-	if err != nil {
-		return
-	}
-
-	hosts = map[string]string{}
-	for _, service := range services {
-		hosts[service.ObjectMeta.Name] = service.Spec.ClusterIP
-	}
-	return
-}

pkg/kt/cluster/types.go

@@ -0,0 +1,43 @@
+package cluster
+
+import (
+	clusterWatcher "github.com/alibaba/kt-connect/pkg/apiserver/cluster"
+	appV1 "k8s.io/api/apps/v1"
+	"k8s.io/client-go/kubernetes"
+	v1 "k8s.io/client-go/listers/core/v1"
+)
+
+// Create kubernetes instance
+func Create(kubeConfig string) (kubernetes Kubernetes, err error) {
+	clientSet, err := GetKubernetesClient(kubeConfig)
+	if err != nil {
+		return
+	}
+	serviceListener, err := clusterWatcher.ServiceListener(clientSet)
+	podListener, err := clusterWatcher.PodListener(clientSet)
+	if err != nil {
+		return
+	}
+	kubernetes = Kubernetes{
+		Clientset:       clientSet,
+		ServiceListener: serviceListener,
+		PodListener:     podListener,
+	}
+	return
+}
+
+// KubernetesInterface kubernetes interface
+type KubernetesInterface interface {
+	Deployment(name, namespace string) (deployment appV1.Deployment, err error)
+	Scale(name, namespace string, replicas *int32) (err error)
+	ServiceHosts(namespace string) (hosts map[string]string)
+	ClusterCrids(podCIDR string) (cidrs []string, err error)
+	CreateShadow(name, namespace, image string, labels map[string]string) (podIP, podName string, err error)
+}
+
+// Kubernetes implements KubernetesInterface
+type Kubernetes struct {
+	Clientset       *kubernetes.Clientset
+	ServiceListener v1.ServiceLister
+	PodListener     v1.PodLister
+}

pkg/kt/command/connect.go

@@ -4,12 +4,12 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/alibaba/kt-connect/pkg/kt/connect"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"github.com/urfave/cli"
 
 	"github.com/alibaba/kt-connect/pkg/kt/cluster"
-	"github.com/alibaba/kt-connect/pkg/kt/connect"
 	"github.com/alibaba/kt-connect/pkg/kt/options"
 	"github.com/alibaba/kt-connect/pkg/kt/util"
 )
@@ -83,50 +83,40 @@ func (action *Action) Connect(options *options.DaemonOptions) (err error) {
 	if err != nil {
 		return
 	}
+
 	log.Info().Msgf("Connect Start At %d", pid)
-	factory := connect.Connect{Options: options}
-	clientSet, err := cluster.GetKubernetesClient(options.KubeConfig)
+
+	kubernetes, err := cluster.Create(options.KubeConfig)
 	if err != nil {
 		return
 	}
 
 	if options.ConnectOptions.Dump2Hosts {
-		hosts := cluster.LocalHosts(clientSet, options.Namespace)
+		hosts := kubernetes.ServiceHosts(options.Namespace)
 		util.DumpHosts(hosts)
 		options.ConnectOptions.Hosts = hosts
 	}
 
 	workload := fmt.Sprintf("kt-connect-daemon-%s", strings.ToLower(util.RandomString(5)))
-	options.RuntimeOptions.Shadow = workload
 
-	labels := map[string]string{
-		"kt":           workload,
-		"kt-component": "connect",
-		"control-by":   "kt",
-	}
-
-	for k, v := range util.String2Map(options.Labels) {
-		labels[k] = v
-	}
-
-	endPointIP, podName, err := cluster.CreateShadow(
-		clientSet,
-		workload,
-		labels,
-		options.Namespace,
-		options.Image,
+	endPointIP, podName, err := kubernetes.CreateShadow(
+		workload, options.Namespace, options.Image, labels(workload, options),
 	)
 
 	if err != nil {
 		return
 	}
 
-	cidrs, err := util.GetCirds(clientSet, options.ConnectOptions.CIDR)
+	// record shadow name will clean up terminal
+	options.RuntimeOptions.Shadow = workload
+
+	cidrs, err := kubernetes.ClusterCrids(options.ConnectOptions.CIDR)
 	if err != nil {
 		return
 	}
 
-	err = factory.StartConnect(podName, endPointIP, cidrs, options.Debug)
+	shadow := connect.Create(options)
+	err = shadow.Outbound(podName, endPointIP, cidrs)
 	if err != nil {
 		return
 	}
@@ -135,3 +125,15 @@ func (action *Action) Connect(options *options.DaemonOptions) (err error) {
 	log.Info().Msgf("Terminal Signal is %s", s)
 	return
 }
+
+func labels(workload string, options *options.DaemonOptions) map[string]string {
+	labels := map[string]string{
+		"kt":           workload,
+		"kt-component": "connect",
+		"control-by":   "kt",
+	}
+	for k, v := range util.String2Map(options.Labels) {
+		labels[k] = v
+	}
+	return labels
+}

pkg/kt/command/exchange.go

@@ -2,6 +2,9 @@ package command
 
 import (
 	"errors"
+	"strings"
+
+	v1 "k8s.io/api/apps/v1"
 
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
@@ -11,8 +14,6 @@ import (
 	"github.com/alibaba/kt-connect/pkg/kt/options"
 	"github.com/alibaba/kt-connect/pkg/kt/util"
 	"github.com/urfave/cli"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // newExchangeCommand return new exchange command
@@ -52,30 +53,57 @@ func (action *Action) Exchange(exchange string, options *options.DaemonOptions)
 	ch := SetUpCloseHandler(options)
 
 	checkConnectRunning(options.RuntimeOptions.PidFile)
-	clientset, err := cluster.GetKubernetesClient(options.KubeConfig)
+
+	kubernetes, err := cluster.Create(options.KubeConfig)
 	if err != nil {
 		return err
 	}
 
-	origin, err := clientset.AppsV1().Deployments(options.Namespace).Get(exchange, metav1.GetOptions{})
+	app, err := kubernetes.Deployment(exchange, options.Namespace)
 	if err != nil {
 		return err
 	}
 
-	replicas := origin.Spec.Replicas
+	// record context inorder to remove after command exit
+	options.RuntimeOptions.Origin = app.GetObjectMeta().GetName()
+	options.RuntimeOptions.Replicas = *app.Spec.Replicas
 
-	// Prepare context inorder to remove after command exit
-	options.RuntimeOptions.Origin = exchange
-	options.RuntimeOptions.Replicas = *replicas
+	workload := app.GetObjectMeta().GetName() + "-kt-" + strings.ToLower(util.RandomString(5))
+	podIP, podName, err := kubernetes.CreateShadow(
+		workload, options.Namespace, options.Image, getExchangeLables(options.Labels, workload, app))
+	log.Info().Msgf("create exchange shadow %s in namespace %s", workload, options.Namespace)
 
-	factory := connect.Connect{}
-	_, err = factory.Exchange(options, origin, clientset, util.String2Map(options.Labels))
 	if err != nil {
 		return err
 	}
 
+	// record data
+	options.RuntimeOptions.Shadow = workload
+
+	down := int32(0)
+	kubernetes.Scale(app, &down)
+
+	shadow := connect.Create(options)
+	shadow.Inbound(options.ExchangeOptions.Expose, podName, podIP)
+
 	s := <-ch
 	log.Info().Msgf("Terminal Signal is %s", s)
 
 	return nil
 }
+
+func getExchangeLables(customLabels string, workload string, origin *v1.Deployment) map[string]string {
+	labels := map[string]string{
+		"kt":           workload,
+		"kt-component": "exchange",
+		"control-by":   "kt",
+	}
+	for k, v := range origin.Spec.Selector.MatchLabels {
+		labels[k] = v
+	}
+	// extra labels must be applied after origin labels
+	for k, v := range util.String2Map(customLabels) {
+		labels[k] = v
+	}
+	return labels
+}

pkg/kt/command/mesh.go

@@ -2,17 +2,25 @@ package command
 
 import (
 	"errors"
+	"strings"
 
 	"github.com/alibaba/kt-connect/pkg/kt/cluster"
 	"github.com/alibaba/kt-connect/pkg/kt/connect"
 	"github.com/alibaba/kt-connect/pkg/kt/options"
 	"github.com/alibaba/kt-connect/pkg/kt/util"
+	v1 "k8s.io/api/apps/v1"
 
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"github.com/urfave/cli"
 )
 
+// ComponentMesh mesh component
+const ComponentMesh = "mesh"
+
+// KubernetesTool kt sign
+const KubernetesTool = "kt"
+
 // newMeshCommand return new mesh command
 func newMeshCommand(options *options.DaemonOptions, action ActionInterface) cli.Command {
 	return cli.Command{
@@ -52,13 +60,31 @@ func (action *Action) Mesh(mesh string, options *options.DaemonOptions) error {
 
 	ch := SetUpCloseHandler(options)
 
-	clientset, err := cluster.GetKubernetesClient(options.KubeConfig)
+	kubernetes, err := cluster.Create(options.KubeConfig)
 	if err != nil {
 		return err
 	}
 
-	factory := connect.Connect{}
-	_, err = factory.Mesh(mesh, options, clientset, util.String2Map(options.Labels))
+	app, err := kubernetes.Deployment(mesh, options.Namespace)
+	if err != nil {
+		return err
+	}
+
+	meshVersion := strings.ToLower(util.RandomString(5))
+	workload := app.GetObjectMeta().GetName() + "-kt-" + meshVersion
+
+	labels := getMeshLabels(workload, meshVersion, app, options)
+
+	podIP, podName, err := kubernetes.CreateShadow(workload, options.Namespace, options.Image, labels)
+	if err != nil {
+		return err
+	}
+
+	// record context data
+	options.RuntimeOptions.Shadow = workload
+
+	shadow := connect.Create(options)
+	err = shadow.Inbound(options.MeshOptions.Expose, podName, podIP)
 
 	if err != nil {
 		return err
@@ -69,3 +95,20 @@ func (action *Action) Mesh(mesh string, options *options.DaemonOptions) error {
 
 	return nil
 }
+
+func getMeshLabels(workload string, meshVersion string, app *v1.Deployment, options *options.DaemonOptions) map[string]string {
+	labels := map[string]string{
+		"kt":           workload,
+		"version":      meshVersion,
+		"kt-component": ComponentMesh,
+		"control-by":   KubernetesTool,
+	}
+	for k, v := range app.Spec.Selector.MatchLabels {
+		labels[k] = v
+	}
+	// extra labels must be applied after origin labels
+	for k, v := range util.String2Map(options.Labels) {
+		labels[k] = v
+	}
+	return labels
+}

pkg/kt/command/run.go

@@ -79,7 +79,12 @@ func (action *Action) Run(service string, options *options.DaemonOptions) error
 	}
 
 	options.RuntimeOptions.Shadow = service
-	connect.RemotePortForward(strconv.Itoa(options.RunOptions.Port), options.KubeConfig, options.Namespace, podName, podIP, options.Debug)
+
+	shadow := connect.Create(options)
+	err = shadow.Inbound(strconv.Itoa(options.RunOptions.Port), podName, podIP)
+	if err != nil {
+		return err
+	}
 
 	log.Info().Msgf("forward remote %s:%v -> 127.0.0.1:%v", podIP, options.RunOptions.Port, options.RunOptions.Port)
 

pkg/kt/connect/exchange.go

@@ -1,65 +0,0 @@
-package connect
-
-import (
-	"strings"
-
-	"github.com/rs/zerolog/log"
-
-	"github.com/alibaba/kt-connect/pkg/kt/cluster"
-	"github.com/alibaba/kt-connect/pkg/kt/options"
-	"github.com/alibaba/kt-connect/pkg/kt/util"
-	v1 "k8s.io/api/apps/v1"
-	"k8s.io/client-go/kubernetes"
-)
-
-// Exchange exchange request to local
-func (c *Connect) Exchange(options *options.DaemonOptions, origin *v1.Deployment, clientset *kubernetes.Clientset, labels map[string]string) (workload string, err error) {
-	workload = origin.GetObjectMeta().GetName() + "-kt-" + strings.ToLower(util.RandomString(5))
-	podIP, podName, err := c.createExchangeShadow(origin, options.Namespace, workload, clientset, labels, options.Image)
-	options.RuntimeOptions.Shadow = workload
-	down := int32(0)
-	scaleTo(origin, options.Namespace, clientset, &down)
-	RemotePortForward(options.ExchangeOptions.Expose, options.KubeConfig, options.Namespace, podName, podIP, options.Debug)
-	return
-}
-
-//ScaleTo Scale
-func scaleTo(deployment *v1.Deployment, namespace string, clientset *kubernetes.Clientset, replicas *int32) (err error) {
-	log.Printf("Try Scale deployment %s to %d\n", deployment.GetObjectMeta().GetName(), *replicas)
-	client := clientset.AppsV1().Deployments(namespace)
-	deployment.Spec.Replicas = replicas
-
-	d, err := client.Update(deployment)
-	if err != nil {
-		log.Printf("%s Fails scale deployment %s to %d\n", err.Error(), deployment.GetObjectMeta().GetName(), *replicas)
-		return err
-	}
-	log.Printf(" * %s (%d replicas) success", d.Name, *d.Spec.Replicas)
-	return nil
-}
-
-func (c *Connect) createExchangeShadow(
-	origin *v1.Deployment,
-	namespace string,
-	workload string,
-	clientset *kubernetes.Clientset,
-	extraLabels map[string]string,
-	image string,
-) (podIP string, podName string, err error) {
-	log.Info().Msgf("Create Exchange shadow %s in namespace %s", workload, namespace)
-	labels := map[string]string{
-		"kt":           workload,
-		"kt-component": "exchange",
-		"control-by":   "kt",
-	}
-	for k, v := range origin.Spec.Selector.MatchLabels {
-		labels[k] = v
-	}
-	// extra labels must be applied after origin labels
-	for k, v := range extraLabels {
-		labels[k] = v
-	}
-
-	podIP, podName, err = cluster.CreateShadow(clientset, workload, labels, namespace, image)
-	return
-}

pkg/kt/connect/inbound.go

@@ -12,9 +12,12 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-// RemotePortForward mapping local port from cluster
-func RemotePortForward(expose, kubeconfig, namespace, target, remoteIP string, debug bool) (err error) {
-	log.Info().Msgf("remote %s forward to local %s", remoteIP, expose)
+// Inbound mapping local port from cluster
+func (s *Shadow) Inbound(exposePort, podName, remoteIP string) (err error) {
+	debug := s.Options.Debug
+	kubeConfig := s.Options.KubeConfig
+	namespace := s.Options.Namespace
+	log.Info().Msgf("remote %s forward to local %s", remoteIP, exposePort)
 	localSSHPort, err := strconv.Atoi(util.GetRandomSSHPort(remoteIP))
 	if err != nil {
 		return
@@ -22,7 +25,7 @@ func RemotePortForward(expose, kubeconfig, namespace, target, remoteIP string, d
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func(wg *sync.WaitGroup) {
-		portforward := kubectl.PortForward(kubeconfig, namespace, target, localSSHPort)
+		portforward := kubectl.PortForward(kubeConfig, namespace, podName, localSSHPort)
 		err = exec.BackgroundRun(portforward, "exchange port forward to local", debug)
 		wg.Done()
 	}(&wg)
@@ -31,9 +34,9 @@ func RemotePortForward(expose, kubeconfig, namespace, target, remoteIP string, d
 		return
 	}
 	log.Printf("SSH Remote port-forward POD %s 22 to 127.0.0.1:%d starting\n", remoteIP, localSSHPort)
-	localPort := expose
-	remotePort := expose
-	ports := strings.SplitN(expose, ":", 2)
+	localPort := exposePort
+	remotePort := exposePort
+	ports := strings.SplitN(exposePort, ":", 2)
 	if len(ports) > 1 {
 		localPort = ports[1]
 		remotePort = ports[0]

pkg/kt/connect/mesh.go

@@ -1,64 +0,0 @@
-package connect
-
-import (
-	"strings"
-
-	"github.com/rs/zerolog/log"
-
-	"github.com/alibaba/kt-connect/pkg/kt/cluster"
-	"github.com/alibaba/kt-connect/pkg/kt/options"
-	"github.com/alibaba/kt-connect/pkg/kt/util"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-)
-
-// Mesh prepare swap deployment
-func (c *Connect) Mesh(swap string, options *options.DaemonOptions, clientset *kubernetes.Clientset, labels map[string]string) (workload string, err error) {
-	workload, podIP, podName, err := c.createMeshShadown(swap, clientset, labels, options.Namespace, options.Image)
-	if err != nil {
-		return
-	}
-	options.RuntimeOptions.Shadow = workload
-	err = RemotePortForward(options.MeshOptions.Expose, options.KubeConfig, options.Namespace, podName, podIP, options.Debug)
-	return
-}
-
-func (c *Connect) createMeshShadown(
-	swap string,
-	clientset *kubernetes.Clientset,
-	extraLabels map[string]string,
-	namespace, image string,
-) (shadowName, podIP, podName string, err error) {
-	deploymentsClient := clientset.AppsV1().Deployments(namespace)
-	origin, err := deploymentsClient.Get(swap, metav1.GetOptions{})
-	if err != nil {
-		return "", "", "", err
-	}
-
-	meshVersion := strings.ToLower(util.RandomString(5))
-	shadowName = origin.GetObjectMeta().GetName() + "-kt-" + meshVersion
-	labels := map[string]string{
-		"kt":           shadowName,
-		"kt-component": "mesh",
-		"control-by":   "kt",
-		"version":      meshVersion,
-	}
-	for k, v := range origin.Spec.Selector.MatchLabels {
-		labels[k] = v
-	}
-	// extra labels must be applied after origin labels
-	for k, v := range extraLabels {
-		labels[k] = v
-	}
-
-	podIP, podName, err = cluster.CreateShadow(clientset, shadowName, labels, namespace, image)
-	if err != nil {
-		return "", "", "", err
-	}
-
-	log.Printf("-----------------------------------------------------------\n")
-	log.Printf("|    Mesh Version '%s' You can update Istio rule       |\n", meshVersion)
-	log.Printf("-----------------------------------------------------------\n")
-
-	return
-}

pkg/kt/connect/outbound.go

@@ -13,25 +13,33 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
-// StartConnect start vpn connection
-func (c *Connect) StartConnect(name, podIP string, cidrs []string, debug bool) (err error) {
+// Outbound start vpn connection
+func (s *Shadow) Outbound(name, podIP string, cidrs []string) (err error) {
+	options := s.Options
 	err = util.PrepareSSHPrivateKey()
 	if err != nil {
 		return
 	}
-	err = exec.BackgroundRun(kubectl.PortForward(c.Options.KubeConfig, c.Options.Namespace, name, c.Options.ConnectOptions.SSHPort), "port-forward", debug)
+	err = exec.BackgroundRun(
+		kubectl.PortForward(
+			options.KubeConfig,
+			options.Namespace,
+			name,
+			options.ConnectOptions.SSHPort),
+		"port-forward",
+		options.Debug)
 	if err != nil {
 		return
 	}
 	time.Sleep(time.Duration(5) * time.Second)
-	if c.Options.ConnectOptions.Method == "socks5" {
+	if options.ConnectOptions.Method == "socks5" {
 		log.Info().Msgf("==============================================================")
-		log.Info().Msgf("Start SOCKS5 Proxy: export http_proxy=socks5://127.0.0.1:%d", c.Options.ConnectOptions.Socke5Proxy)
+		log.Info().Msgf("Start SOCKS5 Proxy: export http_proxy=socks5://127.0.0.1:%d", options.ConnectOptions.Socke5Proxy)
 		log.Info().Msgf("==============================================================")
-		_ = ioutil.WriteFile(".jvmrc", []byte(fmt.Sprintf("-DsocksProxyHost=127.0.0.1\n-DsocksProxyPort=%d", c.Options.ConnectOptions.Socke5Proxy)), 0644)
-		err = exec.BackgroundRun(ssh.DynamicForwardLocalRequestToRemote("127.0.0.1", c.Options.ConnectOptions.SSHPort, c.Options.ConnectOptions.Socke5Proxy), "vpn(ssh)", debug)
+		_ = ioutil.WriteFile(".jvmrc", []byte(fmt.Sprintf("-DsocksProxyHost=127.0.0.1\n-DsocksProxyPort=%d", options.ConnectOptions.Socke5Proxy)), 0644)
+		err = exec.BackgroundRun(ssh.DynamicForwardLocalRequestToRemote("127.0.0.1", options.ConnectOptions.SSHPort, options.ConnectOptions.Socke5Proxy), "vpn(ssh)", options.Debug)
 	} else {
-		err = exec.BackgroundRun(sshuttle.SSHUttle("127.0.0.1", c.Options.ConnectOptions.SSHPort, podIP, c.Options.ConnectOptions.DisableDNS, cidrs, debug), "vpn(sshuttle)", debug)
+		err = exec.BackgroundRun(sshuttle.SSHUttle("127.0.0.1", options.ConnectOptions.SSHPort, podIP, options.ConnectOptions.DisableDNS, cidrs, options.Debug), "vpn(sshuttle)", options.Debug)
 	}
 	if err != nil {
 		return

pkg/kt/connect/types.go

@@ -1,10 +1,22 @@
 package connect
 
-import (
-	"github.com/alibaba/kt-connect/pkg/kt/options"
-)
+import "github.com/alibaba/kt-connect/pkg/kt/options"
 
-// Connect VPN connect interface
-type Connect struct {
+// ShadowInterface shadow interface
+type ShadowInterface interface {
+	Inbound(exposePort, podName, remoteIP string) (err error)
+	Outbound(name, podIP string, cidrs []string) (err error)
+}
+
+// Shadow shadow
+type Shadow struct {
 	Options *options.DaemonOptions
 }
+
+// Create create shadow
+func Create(options *options.DaemonOptions) (shadow Shadow) {
+	shadow = Shadow{
+		Options: options,
+	}
+	return
+}

pkg/kt/util/strings_test.go

@@ -0,0 +1,35 @@
+package util
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestString2Map(t *testing.T) {
+	type args struct {
+		str string
+	}
+	tests := []struct {
+		name string
+		args args
+		want map[string]string
+	}{
+		{
+			name: "should covert to key value",
+			args: args{
+				str: "k1=v1,k2=v2",
+			},
+			want: map[string]string{
+				"k1": "v1",
+				"k2": "v2",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := String2Map(tt.args.str); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("String2Map() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}