kubevpn/pkg/dev/options.go

package dev

import (
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/containerd/containerd/platforms"
	"github.com/docker/cli/cli/command"
	"github.com/docker/docker/api/types/container"
	typescontainer "github.com/docker/docker/api/types/container"
	"github.com/docker/docker/api/types/network"
	"github.com/docker/docker/api/types/strslice"
	"github.com/docker/docker/client"
	"github.com/docker/go-connections/nat"
	"github.com/google/uuid"
	specs "github.com/opencontainers/image-spec/specs-go/v1"
	log "github.com/sirupsen/logrus"
	"github.com/spf13/pflag"
	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	cmdutil "k8s.io/kubectl/pkg/cmd/util"
	"k8s.io/kubectl/pkg/polymorphichelpers"
	"k8s.io/kubectl/pkg/util/podutils"
	"k8s.io/utils/ptr"

	"github.com/wencaiwulue/kubevpn/v2/pkg/config"
	"github.com/wencaiwulue/kubevpn/v2/pkg/daemon"
	"github.com/wencaiwulue/kubevpn/v2/pkg/daemon/rpc"
	"github.com/wencaiwulue/kubevpn/v2/pkg/handler"
	"github.com/wencaiwulue/kubevpn/v2/pkg/inject"
	"github.com/wencaiwulue/kubevpn/v2/pkg/util"
)

type ConnectMode string

const (
	ConnectModeContainer ConnectMode = "container"
	ConnectModeHost      ConnectMode = "host"
)

type Options struct {
	Headers        map[string]string
	Namespace      string
	Workload       string
	ContainerName  string
	NoProxy        bool
	ExtraRouteInfo handler.ExtraRouteInfo
	ConnectMode    ConnectMode
	Engine         config.Engine

	// docker options
	DevImage string

	RunOptions       RunOptions
	ContainerOptions *ContainerOptions

	// inner
	cli       *client.Client
	dockerCli *command.DockerCli

	factory    cmdutil.Factory
	clientset  *kubernetes.Clientset
	restclient *rest.RESTClient
	config     *rest.Config

	// rollback
	rollbackFuncList []func() error
}

func (option *Options) Main(ctx context.Context, sshConfig *util.SshConfig, flags *pflag.FlagSet, transferImage bool) error {
	mode := typescontainer.NetworkMode(option.ContainerOptions.netMode.NetworkMode())
	if mode.IsContainer() {
		log.Infof("network mode container is %s", mode.ConnectedContainer())
		inspect, err := option.cli.ContainerInspect(ctx, mode.ConnectedContainer())
		if err != nil {
			log.Errorf("can not inspect container %s, err: %v", mode.ConnectedContainer(), err)
			return err
		}
		if inspect.State == nil {
			return fmt.Errorf("can not get container status, please make container name is valid")
		}
		if !inspect.State.Running {
			return fmt.Errorf("container %s status is %s, expect is running, please make sure your outer docker name is correct", mode.ConnectedContainer(), inspect.State.Status)
		}
		log.Infof("container %s is running", mode.ConnectedContainer())
	} else if mode.IsDefault() && util.RunningInContainer() {
		hostname, err := os.Hostname()
		if err != nil {
			return err
		}
		log.Infof("hostname is %s", hostname)
		err = option.ContainerOptions.netMode.Set(fmt.Sprintf("container:%s", hostname))
		if err != nil {
			return err
		}
	}

	config, hostConfig, err := Parse(flags, option.ContainerOptions)
	// just in case the Parse does not exit
	if err != nil {
		return err
	}

	// Connect to cluster, in container or host
	err = option.Connect(ctx, sshConfig, transferImage, hostConfig.PortBindings)
	if err != nil {
		log.Errorf("Connect to cluster failed, err: %v", err)
		return err
	}

	return option.Dev(ctx, config, hostConfig)
}

// Connect to cluster network on docker container or host
func (option *Options) Connect(ctx context.Context, sshConfig *util.SshConfig, transferImage bool, portBindings nat.PortMap) error {
	switch option.ConnectMode {
	case ConnectModeHost:
		daemonCli := daemon.GetClient(false)
		if daemonCli == nil {
			return fmt.Errorf("get nil daemon client")
		}
		kubeConfigBytes, ns, err := util.ConvertToKubeConfigBytes(option.factory)
		if err != nil {
			return err
		}
		logLevel := log.ErrorLevel
		if config.Debug {
			logLevel = log.DebugLevel
		}
		// not needs to ssh jump in daemon, because dev mode will hang up until user exit,
		// so just ssh jump in client is enough
		req := &rpc.ConnectRequest{
			KubeconfigBytes:      string(kubeConfigBytes),
			Namespace:            ns,
			Headers:              option.Headers,
			Workloads:            []string{option.Workload},
			ExtraRoute:           option.ExtraRouteInfo.ToRPC(),
			Engine:               string(option.Engine),
			OriginKubeconfigPath: util.GetKubeConfigPath(option.factory),
			TransferImage:        transferImage,
			Image:                config.Image,
			Level:                int32(logLevel),
			SshJump:              sshConfig.ToRPC(),
		}
		if option.NoProxy {
			req.Workloads = nil
		}
		option.AddRollbackFunc(func() error {
			_ = disconnect(ctx, daemonCli, &rpc.DisconnectRequest{
				KubeconfigBytes: ptr.To(string(kubeConfigBytes)),
				Namespace:       ptr.To(ns),
				SshJump:         sshConfig.ToRPC(),
			})
			return nil
		})
		var resp rpc.Daemon_ConnectClient
		resp, err = daemonCli.Proxy(ctx, req)
		if err != nil {
			log.Errorf("Connect to cluster error: %s", err.Error())
			return err
		}
		for {
			resp, err := resp.Recv()
			if err == io.EOF {
				return nil
			} else if err != nil {
				return err
			}
			_, _ = fmt.Fprint(os.Stdout, resp.Message)
		}

	case ConnectModeContainer:
		runConfig, err := option.CreateConnectContainer(portBindings)
		if err != nil {
			return err
		}
		var id string
		log.Infof("starting container connect to cluster")
		id, err = run(ctx, option.cli, option.dockerCli, runConfig)
		if err != nil {
			return err
		}
		option.AddRollbackFunc(func() error {
			_ = option.cli.ContainerKill(context.Background(), id, "SIGTERM")
			_ = runLogsSinceNow(option.dockerCli, id, true)
			return nil
		})
		err = runLogsWaitRunning(ctx, option.dockerCli, id)
		if err != nil {
			// interrupt by signal KILL
			if errors.Is(err, context.Canceled) {
				return nil
			}
			return err
		}
		log.Infof("container connect to cluster successfully")
		err = option.ContainerOptions.netMode.Set(fmt.Sprintf("container:%s", id))
		return err
	default:
		return fmt.Errorf("unsupport connect mode: %s", option.ConnectMode)
	}
}

func (option *Options) Dev(ctx context.Context, cConfig *Config, hostConfig *HostConfig) error {
	templateSpec, err := option.GetPodTemplateSpec()
	if err != nil {
		return err
	}

	sortBy := func(pods []*v1.Pod) sort.Interface {
		for i := 0; i < len(pods); i++ {
			if pods[i].DeletionTimestamp != nil {
				pods = append(pods[:i], pods[i+1:]...)
				i--
			}
		}
		return sort.Reverse(podutils.ActivePods(pods))
	}
	label := labels.SelectorFromSet(templateSpec.Labels).String()
	firstPod, _, err := polymorphichelpers.GetFirstPod(option.clientset.CoreV1(), option.Namespace, label, time.Second*5, sortBy)
	if err != nil {
		log.Errorf("get first running pod from k8s: %v", err)
		return err
	}

	env, err := util.GetEnv(ctx, option.clientset, option.config, option.Namespace, firstPod.Name)
	if err != nil {
		log.Errorf("get env from k8s: %v", err)
		return err
	}
	volume, err := util.GetVolume(ctx, option.factory, option.Namespace, firstPod.Name)
	if err != nil {
		log.Errorf("get volume from k8s: %v", err)
		return err
	}
	option.AddRollbackFunc(func() error {
		return util.RemoveDir(volume)
	})
	dns, err := util.GetDNS(ctx, option.clientset, option.config, option.Namespace, firstPod.Name)
	if err != nil {
		log.Errorf("get dns from k8s: %v", err)
		return err
	}

	inject.RemoveContainers(templateSpec)
	if option.ContainerName != "" {
		var index = -1
		for i, c := range templateSpec.Spec.Containers {
			if option.ContainerName == c.Name {
				index = i
				break
			}
		}
		if index != -1 {
			templateSpec.Spec.Containers[0], templateSpec.Spec.Containers[index] = templateSpec.Spec.Containers[index], templateSpec.Spec.Containers[0]
		}
	}
	configList := ConvertPodToContainer(option.Namespace, *templateSpec, env, volume, dns)
	MergeDockerOptions(configList, option, cConfig, hostConfig)

	mode := container.NetworkMode(option.ContainerOptions.netMode.NetworkMode())
	if len(option.ContainerOptions.netMode.Value()) != 0 {
		log.Infof("network mode is %s", option.ContainerOptions.netMode.NetworkMode())
		for _, runConfig := range configList[:] {
			// remove expose port
			runConfig.config.ExposedPorts = nil
			runConfig.hostConfig.NetworkMode = mode
			if mode.IsContainer() {
				runConfig.hostConfig.PidMode = typescontainer.PidMode(option.ContainerOptions.netMode.NetworkMode())
			}
			runConfig.hostConfig.PortBindings = nil

			// remove dns
			runConfig.hostConfig.DNS = nil
			runConfig.hostConfig.DNSOptions = nil
			runConfig.hostConfig.DNSSearch = nil
			runConfig.hostConfig.PublishAllPorts = false
			runConfig.config.Hostname = ""
		}
	} else {
		var networkID string
		networkID, err = createNetwork(ctx, option.cli)
		if err != nil {
			log.Errorf("create network for %s: %v", option.Workload, err)
			return err
		}
		log.Infof("create docker network %s", networkID)

		configList[len(configList)-1].networkingConfig.EndpointsConfig = map[string]*network.EndpointSettings{
			configList[len(configList)-1].name: {NetworkID: networkID},
		}
		var portMap = nat.PortMap{}
		var portSet = nat.PortSet{}
		for _, runConfig := range configList {
			for k, v := range runConfig.hostConfig.PortBindings {
				if oldValue, ok := portMap[k]; ok {
					portMap[k] = append(oldValue, v...)
				} else {
					portMap[k] = v
				}
			}
			for k, v := range runConfig.config.ExposedPorts {
				portSet[k] = v
			}
		}
		configList[len(configList)-1].hostConfig.PortBindings = portMap
		configList[len(configList)-1].config.ExposedPorts = portSet

		// skip last, use last container network
		for _, runConfig := range configList[:len(configList)-1] {
			// remove expose port
			runConfig.config.ExposedPorts = nil
			runConfig.hostConfig.NetworkMode = typescontainer.NetworkMode("container:" + configList[len(configList)-1].name)
			runConfig.hostConfig.PidMode = typescontainer.PidMode("container:" + configList[len(configList)-1].name)
			runConfig.hostConfig.PortBindings = nil

			// remove dns
			runConfig.hostConfig.DNS = nil
			runConfig.hostConfig.DNSOptions = nil
			runConfig.hostConfig.DNSSearch = nil
			runConfig.hostConfig.PublishAllPorts = false
			runConfig.config.Hostname = ""
		}
	}

	option.AddRollbackFunc(func() error {
		return configList.Remove(ctx, option.cli)
	})
	return configList.Run(ctx, volume, option.cli, option.dockerCli)
}

func disconnect(ctx context.Context, daemonClient rpc.DaemonClient, req *rpc.DisconnectRequest) error {
	resp, err := daemonClient.Disconnect(ctx, req)
	if err != nil {
		return err
	}
	for {
		recv, err := resp.Recv()
		if err == io.EOF {
			return nil
		} else if err != nil {
			return err
		}
		_, _ = fmt.Fprint(os.Stdout, recv.Message)
	}
}

func (option *Options) CreateConnectContainer(portBindings nat.PortMap) (*RunConfig, error) {
	portMap, portSet, err := option.GetExposePort(portBindings)
	if err != nil {
		return nil, err
	}

	var kubeconfigPath = os.Getenv(config.EnvSSHJump)
	if kubeconfigPath != "" {
		kubeconfigPath, err = util.ConvertK8sApiServerToDomain(kubeconfigPath)
	} else {
		kubeconfigPath, err = util.GetKubeconfigPath(option.factory)
	}
	if err != nil {
		return nil, err
	}

	var entrypoint []string
	if option.NoProxy {
		entrypoint = []string{"kubevpn", "connect", "--foreground", "-n", option.Namespace, "--kubeconfig", "/root/.kube/config", "--image", config.Image, "--engine", string(option.Engine)}
	} else {
		entrypoint = []string{"kubevpn", "proxy", option.Workload, "--foreground", "-n", option.Namespace, "--kubeconfig", "/root/.kube/config", "--image", config.Image, "--engine", string(option.Engine)}
		for k, v := range option.Headers {
			entrypoint = append(entrypoint, "--headers", fmt.Sprintf("%s=%s", k, v))
		}
	}
	for _, v := range option.ExtraRouteInfo.ExtraCIDR {
		entrypoint = append(entrypoint, "--extra-cidr", v)
	}
	for _, v := range option.ExtraRouteInfo.ExtraDomain {
		entrypoint = append(entrypoint, "--extra-domain", v)
	}
	if option.ExtraRouteInfo.ExtraNodeIP {
		entrypoint = append(entrypoint, "--extra-node-ip")
	}

	runConfig := &container.Config{
		User:         "root",
		ExposedPorts: portSet,
		Env:          []string{},
		Cmd:          []string{},
		Healthcheck:  nil,
		Image:        config.Image,
		Entrypoint:   entrypoint,
	}
	hostConfig := &container.HostConfig{
		Binds:         []string{fmt.Sprintf("%s:%s", kubeconfigPath, "/root/.kube/config")},
		LogConfig:     container.LogConfig{},
		PortBindings:  portMap,
		AutoRemove:    true,
		Privileged:    true,
		RestartPolicy: container.RestartPolicy{},
		CapAdd:        strslice.StrSlice{"SYS_PTRACE", "SYS_ADMIN"}, // for dlv
		// https://stackoverflow.com/questions/24319662/from-inside-of-a-docker-container-how-do-i-connect-to-the-localhost-of-the-mach
		// couldn't get current server API group list: Get "https://host.docker.internal:62844/api?timeout=32s": tls: failed to verify certificate: x509: certificate is valid for kubernetes.default.svc.cluster.local, kubernetes.default.svc, kubernetes.default, kubernetes, istio-sidecar-injector.istio-system.svc, proxy-exporter.kube-system.svc, not host.docker.internal
		ExtraHosts:  []string{"host.docker.internal:host-gateway", "kubernetes:host-gateway"},
		SecurityOpt: []string{"apparmor=unconfined", "seccomp=unconfined"},
		Sysctls:     map[string]string{"net.ipv6.conf.all.disable_ipv6": strconv.Itoa(0)},
		Resources:   container.Resources{},
	}
	newUUID, err := uuid.NewUUID()
	if err != nil {
		return nil, err
	}
	suffix := strings.ReplaceAll(newUUID.String(), "-", "")[:5]
	name := util.Join(option.Namespace, "kubevpn", suffix)
	networkID, err := createNetwork(context.Background(), option.cli)
	if err != nil {
		return nil, err
	}
	var platform *specs.Platform
	if option.RunOptions.Platform != "" {
		plat, _ := platforms.Parse(option.RunOptions.Platform)
		platform = &plat
	}
	c := &RunConfig{
		config:           runConfig,
		hostConfig:       hostConfig,
		networkingConfig: &network.NetworkingConfig{EndpointsConfig: map[string]*network.EndpointSettings{name: {NetworkID: networkID}}},
		platform:         platform,
		name:             name,
		Options:          RunOptions{Pull: PullImageMissing},
	}
	return c, nil
}

func (option *Options) AddRollbackFunc(f func() error) {
	option.rollbackFuncList = append(option.rollbackFuncList, f)
}

func (option *Options) GetRollbackFuncList() []func() error {
	return option.rollbackFuncList
}

func AddDockerFlags(options *Options, p *pflag.FlagSet) {
	p.SetInterspersed(false)

	// These are flags not stored in Config/HostConfig
	p.StringVar(&options.RunOptions.Pull, "pull", PullImageMissing, `Pull image before running ("`+PullImageAlways+`"|"`+PullImageMissing+`"|"`+PullImageNever+`")`)
	p.BoolVar(&options.RunOptions.SigProxy, "sig-proxy", true, "Proxy received signals to the process")

	// Add an explicit help that doesn't have a `-h` to prevent the conflict
	// with hostname
	p.Bool("help", false, "Print usage")

	command.AddPlatformFlag(p, &options.RunOptions.Platform)
	options.ContainerOptions = addFlags(p)
}

func (option *Options) GetExposePort(portBinds nat.PortMap) (nat.PortMap, nat.PortSet, error) {
	templateSpec, err := option.GetPodTemplateSpec()
	if err != nil {
		return nil, nil, err
	}

	var portMap = nat.PortMap{}
	var portSet = nat.PortSet{}
	for _, c := range templateSpec.Spec.Containers {
		for _, port := range c.Ports {
			p := nat.Port(fmt.Sprintf("%d/%s", port.ContainerPort, strings.ToLower(string(port.Protocol))))
			if port.HostPort != 0 {
				binding := []nat.PortBinding{{HostPort: strconv.FormatInt(int64(port.HostPort), 10)}}
				portMap[p] = binding
			} else {
				binding := []nat.PortBinding{{HostPort: strconv.FormatInt(int64(port.ContainerPort), 10)}}
				portMap[p] = binding
			}
			portSet[p] = struct{}{}
		}
	}

	for port, bindings := range portBinds {
		portMap[port] = bindings
		portSet[port] = struct{}{}
	}

	return portMap, portSet, nil
}

func (option *Options) InitClient(f cmdutil.Factory) (err error) {
	option.factory = f
	if option.config, err = option.factory.ToRESTConfig(); err != nil {
		return
	}
	if option.restclient, err = option.factory.RESTClient(); err != nil {
		return
	}
	if option.clientset, err = option.factory.KubernetesClientSet(); err != nil {
		return
	}
	if option.Namespace, _, err = option.factory.ToRawKubeConfigLoader().Namespace(); err != nil {
		return
	}
	if option.cli, option.dockerCli, err = util.GetClient(); err != nil {
		return err
	}
	return
}

func (option *Options) GetPodTemplateSpec() (*v1.PodTemplateSpec, error) {
	object, err := util.GetUnstructuredObject(option.factory, option.Namespace, option.Workload)
	if err != nil {
		log.Errorf("get unstructured object error: %v", err)
		return nil, err
	}

	u := object.Object.(*unstructured.Unstructured)
	var templateSpec *v1.PodTemplateSpec
	templateSpec, _, err = util.GetPodTemplateSpecPath(u)
	return templateSpec, err
}