diff --git a/.travis.yml b/.travis.yml index b03ea30..ade911b 100644 --- a/.travis.yml +++ b/.travis.yml @@ -2,7 +2,6 @@ language: go go: - 1.13.x - - master services: - docker diff --git a/pkg/kt/util/kubernetes.go b/pkg/kt/cluster/helper.go similarity index 56% rename from pkg/kt/util/kubernetes.go rename to pkg/kt/cluster/helper.go index fdcaa2b..32387d7 100644 --- a/pkg/kt/util/kubernetes.go +++ b/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 } diff --git a/pkg/kt/cluster/kubernetes.go b/pkg/kt/cluster/kubernetes.go index f4407d2..6bdb1dd 100644 --- a/pkg/kt/cluster/kubernetes.go +++ b/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 -} diff --git a/pkg/kt/cluster/types.go b/pkg/kt/cluster/types.go new file mode 100644 index 0000000..26513b4 --- /dev/null +++ b/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 +} diff --git a/pkg/kt/command/connect.go b/pkg/kt/command/connect.go index 5433f35..70041d9 100644 --- a/pkg/kt/command/connect.go +++ b/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 +} diff --git a/pkg/kt/command/exchange.go b/pkg/kt/command/exchange.go index c9f021f..aa22c71 100644 --- a/pkg/kt/command/exchange.go +++ b/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 +} diff --git a/pkg/kt/command/mesh.go b/pkg/kt/command/mesh.go index c98b7e5..edef13e 100644 --- a/pkg/kt/command/mesh.go +++ b/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 +} diff --git a/pkg/kt/command/run.go b/pkg/kt/command/run.go index 805c694..f34f4f1 100644 --- a/pkg/kt/command/run.go +++ b/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) diff --git a/pkg/kt/connect/exchange.go b/pkg/kt/connect/exchange.go deleted file mode 100644 index bf8b0e5..0000000 --- a/pkg/kt/connect/exchange.go +++ /dev/null @@ -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 -} diff --git a/pkg/kt/connect/inbound.go b/pkg/kt/connect/inbound.go index 0ef3c65..32d6f20 100644 --- a/pkg/kt/connect/inbound.go +++ b/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] diff --git a/pkg/kt/connect/mesh.go b/pkg/kt/connect/mesh.go deleted file mode 100644 index 17b727e..0000000 --- a/pkg/kt/connect/mesh.go +++ /dev/null @@ -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 -} diff --git a/pkg/kt/connect/outbound.go b/pkg/kt/connect/outbound.go index 58f0128..516165d 100644 --- a/pkg/kt/connect/outbound.go +++ b/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 diff --git a/pkg/kt/connect/types.go b/pkg/kt/connect/types.go index db3c73a..a36c1bd 100644 --- a/pkg/kt/connect/types.go +++ b/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 +} diff --git a/pkg/kt/util/strings_test.go b/pkg/kt/util/strings_test.go new file mode 100644 index 0000000..e4e2d8c --- /dev/null +++ b/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) + } + }) + } +}