pkg/handler/connect.go
package handler
import (
"context"
"encoding/json"
"fmt"
"io"
"math"
"math/rand"
"net"
"net/url"
"os/exec"
"reflect"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/containernetworking/cni/pkg/types"
"github.com/distribution/reference"
goversion "github.com/hashicorp/go-version"
"github.com/libp2p/go-netroute"
miekgdns "github.com/miekg/dns"
"github.com/pkg/errors"
log "github.com/sirupsen/logrus"
"google.golang.org/grpc/metadata"
v1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/fields"
pkgruntime "k8s.io/apimachinery/pkg/runtime"
pkgtypes "k8s.io/apimachinery/pkg/types"
utilnet "k8s.io/apimachinery/pkg/util/net"
"k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/cli-runtime/pkg/resource"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
"k8s.io/client-go/util/retry"
"k8s.io/kubectl/pkg/cmd/set"
cmdutil "k8s.io/kubectl/pkg/cmd/util"
"k8s.io/kubectl/pkg/polymorphichelpers"
"k8s.io/kubectl/pkg/scheme"
"k8s.io/kubectl/pkg/util/podutils"
"k8s.io/utils/pointer"
"github.com/wencaiwulue/kubevpn/v2/pkg/config"
"github.com/wencaiwulue/kubevpn/v2/pkg/core"
"github.com/wencaiwulue/kubevpn/v2/pkg/dhcp"
"github.com/wencaiwulue/kubevpn/v2/pkg/dns"
"github.com/wencaiwulue/kubevpn/v2/pkg/driver"
"github.com/wencaiwulue/kubevpn/v2/pkg/inject"
"github.com/wencaiwulue/kubevpn/v2/pkg/tun"
"github.com/wencaiwulue/kubevpn/v2/pkg/util"
)
type ConnectOptions struct {
Namespace string
Headers map[string]string
PortMap []string
Workloads []string
ExtraRouteInfo ExtraRouteInfo
Engine config.Engine
Foreground bool
OriginKubeconfigPath string
Lock *sync.Mutex
ctx context.Context
cancel context.CancelFunc
clientset *kubernetes.Clientset
restclient *rest.RESTClient
config *rest.Config
factory cmdutil.Factory
cidrs []*net.IPNet
dhcp *dhcp.Manager
// needs to give it back to dhcp
localTunIPv4 *net.IPNet
localTunIPv6 *net.IPNet
rollbackFuncList []func() error
dnsConfig *dns.Config
apiServerIPs []net.IP
extraHost []dns.Entry
once sync.Once
}
func (c *ConnectOptions) Context() context.Context {
return c.ctx
}
func (c *ConnectOptions) InitDHCP(ctx context.Context) error {
if c.dhcp == nil {
c.dhcp = dhcp.NewDHCPManager(c.clientset.CoreV1().ConfigMaps(c.Namespace), c.Namespace)
return c.dhcp.InitDHCP(ctx)
}
return nil
}
func (c *ConnectOptions) RentIP(ctx context.Context) (context.Context, error) {
if err := c.InitDHCP(ctx); err != nil {
return nil, err
}
var err error
c.localTunIPv4, c.localTunIPv6, err = c.dhcp.RentIP(ctx)
if err != nil {
return nil, err
}
ctx1 := metadata.AppendToOutgoingContext(
ctx,
config.HeaderIPv4, c.localTunIPv4.String(),
config.HeaderIPv6, c.localTunIPv6.String(),
)
return ctx1, nil
}
func (c *ConnectOptions) GetIPFromContext(ctx context.Context) error {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return fmt.Errorf("can not get IOP from context")
}
ipv4 := md.Get(config.HeaderIPv4)
if len(ipv4) == 0 {
return fmt.Errorf("can not found IPv4 from header: %v", md)
}
ip, ipNet, err := net.ParseCIDR(ipv4[0])
if err != nil {
return fmt.Errorf("cat not convert IPv4 string: %s: %v", ipv4[0], err)
}
c.localTunIPv4 = &net.IPNet{IP: ip, Mask: ipNet.Mask}
log.Debugf("Get IPv4 %s from context", c.localTunIPv4.String())
ipv6 := md.Get(config.HeaderIPv6)
if len(ipv6) == 0 {
return fmt.Errorf("can not found IPv6 from header: %v", md)
}
ip, ipNet, err = net.ParseCIDR(ipv6[0])
if err != nil {
return fmt.Errorf("cat not convert IPv6 string: %s: %v", ipv6[0], err)
}
c.localTunIPv6 = &net.IPNet{IP: ip, Mask: ipNet.Mask}
log.Debugf("Get IPv6 %s from context", c.localTunIPv6.String())
return nil
}
func (c *ConnectOptions) CreateRemoteInboundPod(ctx context.Context) (err error) {
if c.localTunIPv4 == nil || c.localTunIPv6 == nil {
return fmt.Errorf("local tun IP is invalid")
}
for _, workload := range c.Workloads {
log.Infof("Injecting inbound sidecar for %s", workload)
configInfo := util.PodRouteConfig{
LocalTunIPv4: c.localTunIPv4.IP.String(),
LocalTunIPv6: c.localTunIPv6.IP.String(),
}
// todo consider to use ephemeral container
// https://kubernetes.io/docs/concepts/workloads/pods/ephemeral-containers/
// means mesh mode
if len(c.Headers) != 0 || len(c.PortMap) != 0 {
err = inject.InjectVPNAndEnvoySidecar(ctx, c.factory, c.clientset.CoreV1().ConfigMaps(c.Namespace), c.Namespace, workload, configInfo, c.Headers, c.PortMap)
} else {
err = inject.InjectVPNSidecar(ctx, c.factory, c.Namespace, workload, configInfo)
}
if err != nil {
log.Errorf("Injecting inbound sidecar for %s failed: %s", workload, err.Error())
return err
}
}
return
}
func (c *ConnectOptions) DoConnect(ctx context.Context, isLite bool) (err error) {
c.ctx, c.cancel = context.WithCancel(ctx)
log.Info("Starting connect")
m := dhcp.NewDHCPManager(c.clientset.CoreV1().ConfigMaps(c.Namespace), c.Namespace)
if err = m.InitDHCP(c.ctx); err != nil {
log.Errorf("Init DHCP failed: %v", err)
return
}
go c.setupSignalHandler()
if err = c.getCIDR(c.ctx, m); err != nil {
log.Errorf("Failed to get network CIDR: %v", err)
return
}
if err = createOutboundPod(c.ctx, c.factory, c.clientset, c.Namespace); err != nil {
return
}
if err = c.upgradeDeploy(c.ctx); err != nil {
return
}
//if err = c.CreateRemoteInboundPod(c.ctx); err != nil {
// return
//}
if err = c.addExtraNodeIP(c.ctx); err != nil {
log.Errorf("Add extra node IP failed: %v", err)
return
}
var rawTCPForwardPort, gvisorTCPForwardPort, gvisorUDPForwardPort int
rawTCPForwardPort, err = util.GetAvailableTCPPortOrDie()
if err != nil {
return err
}
gvisorTCPForwardPort, err = util.GetAvailableTCPPortOrDie()
if err != nil {
return err
}
gvisorUDPForwardPort, err = util.GetAvailableTCPPortOrDie()
if err != nil {
return err
}
log.Info("Forwarding port...")
if err = c.portForward(c.ctx, []string{
fmt.Sprintf("%d:10800", rawTCPForwardPort),
fmt.Sprintf("%d:10801", gvisorTCPForwardPort),
fmt.Sprintf("%d:10802", gvisorUDPForwardPort),
}); err != nil {
return
}
if util.IsWindows() {
driver.InstallWireGuardTunDriver()
}
forward := fmt.Sprintf("tcp://127.0.0.1:%d", rawTCPForwardPort)
core.GvisorTCPForwardAddr = fmt.Sprintf("tcp://127.0.0.1:%d", gvisorTCPForwardPort)
core.GvisorUDPForwardAddr = fmt.Sprintf("tcp://127.0.0.1:%d", gvisorUDPForwardPort)
if err = c.startLocalTunServe(c.ctx, forward, isLite); err != nil {
log.Errorf("Start local tun service failed: %v", err)
return
}
log.Infof("Adding route...")
if err = c.addRouteDynamic(c.ctx); err != nil {
log.Errorf("Add route dynamic failed: %v", err)
return
}
go c.deleteFirewallRule(c.ctx)
log.Infof("Configuring DNS service...")
if err = c.setupDNS(c.ctx); err != nil {
log.Errorf("Configure DNS failed: %v", err)
return
}
log.Info("Configured DNS service")
return
}
// detect pod is delete event, if pod is deleted, needs to redo port-forward immediately
func (c *ConnectOptions) portForward(ctx context.Context, portPair []string) error {
firstCtx, firstCancelFunc := context.WithCancel(ctx)
defer firstCancelFunc()
var errChan = make(chan error, 1)
go func() {
runtime.ErrorHandlers = []func(error){}
var first = pointer.Bool(true)
for ctx.Err() == nil {
func() {
defer time.Sleep(time.Millisecond * 200)
sortBy := func(pods []*v1.Pod) sort.Interface { return sort.Reverse(podutils.ActivePods(pods)) }
label := fields.OneTermEqualSelector("app", config.ConfigMapPodTrafficManager).String()
_, _, _ = polymorphichelpers.GetFirstPod(c.clientset.CoreV1(), c.Namespace, label, time.Second*10, sortBy)
ctx2, cancelFunc2 := context.WithTimeout(ctx, time.Second*10)
defer cancelFunc2()
podList, err := c.GetRunningPodList(ctx2)
if err != nil {
log.Debugf("Failed to get running pod: %v", err)
if *first {
util.SafeWrite(errChan, err)
}
return
}
childCtx, cancelFunc := context.WithCancel(ctx)
defer cancelFunc()
var readyChan = make(chan struct{})
podName := podList[0].GetName()
// try to detect pod is delete event, if pod is deleted, needs to redo port-forward
go util.CheckPodStatus(childCtx, cancelFunc, podName, c.clientset.CoreV1().Pods(c.Namespace))
go util.CheckPortStatus(childCtx, cancelFunc, readyChan, strings.Split(portPair[1], ":")[0])
if *first {
go func() {
select {
case <-readyChan:
firstCancelFunc()
case <-childCtx.Done():
}
}()
}
var out = log.StandardLogger().WriterLevel(log.DebugLevel)
defer out.Close()
err = util.PortForwardPod(
c.config,
c.restclient,
podName,
c.Namespace,
portPair,
readyChan,
childCtx.Done(),
out,
out,
)
if *first {
util.SafeWrite(errChan, err)
}
first = pointer.Bool(false)
// exit normal, let context.err to judge to exit or not
if err == nil {
log.Debugf("Port forward retrying")
return
}
if strings.Contains(err.Error(), "unable to listen on any of the requested ports") ||
strings.Contains(err.Error(), "address already in use") {
log.Debugf("Port %s already in use, needs to release it manually", portPair)
} else {
log.Debugf("Port-forward occurs error: %v", err)
}
}()
}
}()
ticker := time.NewTicker(time.Second * 60)
defer ticker.Stop()
select {
case <-ticker.C:
return errors.New("wait port forward to be ready timeout")
case err := <-errChan:
return err
case <-firstCtx.Done():
return nil
}
}
func (c *ConnectOptions) startLocalTunServe(ctx context.Context, forwardAddress string, lite bool) (err error) {
log.Debugf("IPv4: %s, IPv6: %s", c.localTunIPv4.IP.String(), c.localTunIPv6.IP.String())
var cidrList []*net.IPNet
if !lite {
cidrList = append(cidrList, config.CIDR)
}
for _, ipNet := range c.cidrs {
cidrList = append(cidrList, ipNet)
}
// add extra-cidr
for _, s := range c.ExtraRouteInfo.ExtraCIDR {
var ipnet *net.IPNet
_, ipnet, err = net.ParseCIDR(s)
if err != nil {
return fmt.Errorf("invalid extra-cidr %s, err: %v", s, err)
}
cidrList = append(cidrList, ipnet)
}
var routes []types.Route
for _, ipNet := range util.RemoveLargerOverlappingCIDRs(cidrList) {
routes = append(routes, types.Route{Dst: *ipNet})
}
tunConfig := tun.Config{
Addr: c.localTunIPv4.String(),
Routes: routes,
}
if enable, _ := util.IsIPv6Enabled(); enable {
tunConfig.Addr6 = c.localTunIPv6.String()
}
localNode := fmt.Sprintf("tun:/127.0.0.1:8422")
node, err := core.ParseNode(localNode)
if err != nil {
log.Errorf("Failed to parse local node %s: %v", localNode, err)
return err
}
node.Values.Add(config.ConfigKubeVPNTransportEngine, string(c.Engine))
chainNode, err := core.ParseNode(forwardAddress)
if err != nil {
log.Errorf("Failed to parse forward node %s: %v", forwardAddress, err)
return err
}
chainNode.Client = &core.Client{
Connector: core.UDPOverTCPTunnelConnector(),
Transporter: core.TCPTransporter(),
}
chain := core.NewChain(5, chainNode)
handler := core.TunHandler(chain, node)
listener, err := tun.Listener(tunConfig)
if err != nil {
log.Errorf("Failed to create tun listener: %v", err)
return err
}
server := core.Server{
Listener: listener,
Handler: handler,
}
go func() {
defer server.Listener.Close()
go func() {
<-ctx.Done()
server.Listener.Close()
}()
for ctx.Err() == nil {
conn, err := server.Listener.Accept()
if err != nil {
if !errors.Is(err, tun.ClosedErr) {
log.Errorf("Failed to accept local tun conn: %v", err)
}
return
}
go server.Handler.Handle(ctx, conn)
}
}()
log.Info("Connected tunnel")
return
}
// Listen all pod, add route if needed
func (c *ConnectOptions) addRouteDynamic(ctx context.Context) error {
tunName, e := c.GetTunDeviceName()
if e != nil {
return e
}
podNs, svcNs, err1 := util.GetNsForListPodAndSvc(ctx, c.clientset, []string{v1.NamespaceAll, c.Namespace})
if err1 != nil {
return err1
}
var addRouteFunc = func(resource, ipStr string) {
ip := net.ParseIP(ipStr)
if ip == nil {
return
}
for _, p := range c.apiServerIPs {
// if pod ip or service ip is equal to apiServer ip, can not add it to route table
if p.Equal(ip) {
return
}
}
var mask net.IPMask
if ip.To4() != nil {
mask = net.CIDRMask(32, 32)
} else {
mask = net.CIDRMask(128, 128)
}
if r, err := netroute.New(); err == nil {
iface, _, _, err := r.Route(ip)
if err == nil && iface.Name == tunName {
return
}
}
errs := tun.AddRoutes(tunName, types.Route{Dst: net.IPNet{IP: ip, Mask: mask}})
if errs != nil {
log.Errorf("Failed to add route, resource: %s, IP: %s, err: %v", resource, ip, errs)
}
}
go func() {
var listDone bool
for ctx.Err() == nil {
err := func() error {
if !listDone {
err := util.ListService(ctx, c.clientset.CoreV1().Services(svcNs), addRouteFunc)
if err != nil {
return err
}
listDone = true
}
err := util.WatchServiceToAddRoute(ctx, c.clientset.CoreV1().Services(svcNs), addRouteFunc)
return err
}()
if utilnet.IsConnectionRefused(err) || apierrors.IsTooManyRequests(err) || apierrors.IsForbidden(err) {
time.Sleep(time.Second * 1)
} else {
time.Sleep(time.Millisecond * 200)
}
}
}()
go func() {
var listDone bool
for ctx.Err() == nil {
err := func() error {
if !listDone {
err := util.ListPod(ctx, c.clientset.CoreV1().Pods(podNs), addRouteFunc)
if err != nil {
return err
}
listDone = true
}
err := util.WatchPodToAddRoute(ctx, c.clientset.CoreV1().Pods(podNs), addRouteFunc)
return err
}()
if utilnet.IsConnectionRefused(err) || apierrors.IsTooManyRequests(err) || apierrors.IsForbidden(err) {
time.Sleep(time.Second * 1)
} else {
time.Sleep(time.Millisecond * 200)
}
}
}()
return nil
}
func (c *ConnectOptions) deleteFirewallRule(ctx context.Context) {
// The reason why delete firewall rule is:
// On windows use 'kubevpn proxy deploy/authors -H user=windows'
// Open terminal 'curl localhost:9080' ok
// Open terminal 'curl localTunIP:9080' not ok
util.DeleteBlockFirewallRule(ctx)
}
func (c *ConnectOptions) setupDNS(ctx context.Context) error {
const port = 53
const portTCP = 10800
pod, err := c.GetRunningPodList(ctx)
if err != nil {
log.Errorf("Get running pod list failed, err: %v", err)
return err
}
log.Debugf("Get DNS service IP from pod...")
relovConf, err := util.GetDNSServiceIPFromPod(ctx, c.clientset, c.config, pod[0].GetName(), c.Namespace)
if err != nil {
log.Errorln(err)
return err
}
if relovConf.Port == "" {
relovConf.Port = strconv.Itoa(port)
}
marshal, _ := json.Marshal(relovConf)
log.Debugf("Get DNS service config: %v", string(marshal))
svc, err := c.clientset.CoreV1().Services(c.Namespace).Get(ctx, config.ConfigMapPodTrafficManager, metav1.GetOptions{})
if err != nil {
return err
}
var conn net.Conn
d := net.Dialer{Timeout: time.Duration(max(2, relovConf.Timeout)) * time.Second}
conn, err = d.DialContext(ctx, "tcp", net.JoinHostPort(svc.Spec.ClusterIP, strconv.Itoa(portTCP)))
if err != nil {
relovConf.Servers = []string{pod[0].Status.PodIP}
err = nil
log.Debugf("DNS service use pod IP %s", pod[0].Status.PodIP)
} else {
relovConf.Servers = []string{svc.Spec.ClusterIP}
_ = conn.Close()
log.Debugf("DNS service use service IP %s", svc.Spec.ClusterIP)
}
log.Debugf("Adding extra hosts...")
if err = c.addExtraRoute(c.ctx, relovConf.Servers[0]); err != nil {
log.Errorf("Add extra route failed: %v", err)
return err
}
ns := []string{c.Namespace}
list, err := c.clientset.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
if err == nil {
for _, item := range list.Items {
if !sets.New[string](ns...).Has(item.Name) {
ns = append(ns, item.Name)
}
}
}
tunName, err := c.GetTunDeviceName()
if err != nil {
return err
}
c.dnsConfig = &dns.Config{
Config: relovConf,
Ns: ns,
TunName: tunName,
Hosts: c.extraHost,
Lock: c.Lock,
}
log.Debugf("Setup DNS...")
if err = c.dnsConfig.SetupDNS(ctx); err != nil {
return err
}
log.Debugf("Dump service in namespace %s into hosts...", c.Namespace)
// dump service in current namespace for support DNS resolve service:port
err = c.dnsConfig.AddServiceNameToHosts(ctx, c.clientset.CoreV1().Services(c.Namespace), c.extraHost...)
return err
}
func Run(ctx context.Context, servers []core.Server) error {
errChan := make(chan error, len(servers))
for i := range servers {
go func(i int) {
errChan <- func() error {
svr := servers[i]
defer svr.Listener.Close()
go func() {
<-ctx.Done()
svr.Listener.Close()
}()
for ctx.Err() == nil {
conn, err := svr.Listener.Accept()
if err != nil {
return err
}
go svr.Handler.Handle(ctx, conn)
}
return ctx.Err()
}()
}(i)
}
select {
case err := <-errChan:
return err
case <-ctx.Done():
return ctx.Err()
}
}
func Parse(r core.Route) ([]core.Server, error) {
servers, err := r.GenerateServers()
if err != nil {
return nil, err
}
if len(servers) == 0 {
return nil, fmt.Errorf("server is empty, server config: %s", strings.Join(r.ServeNodes, ","))
}
return servers, nil
}
func (c *ConnectOptions) InitClient(f cmdutil.Factory) (err error) {
c.factory = f
if c.config, err = c.factory.ToRESTConfig(); err != nil {
return
}
if c.restclient, err = c.factory.RESTClient(); err != nil {
return
}
if c.clientset, err = c.factory.KubernetesClientSet(); err != nil {
return
}
if c.Namespace, _, err = c.factory.ToRawKubeConfigLoader().Namespace(); err != nil {
return
}
return
}
// PreCheckResource transform user parameter to normal, example:
// pod: productpage-7667dfcddb-cbsn5
// replicast: productpage-7667dfcddb
// deployment: productpage
// transform:
// pod/productpage-7667dfcddb-cbsn5 --> deployment/productpage
// service/productpage --> deployment/productpage
// replicaset/productpage-7667dfcddb --> deployment/productpage
//
// pods without controller
// pod/productpage-without-controller --> pod/productpage-without-controller
// service/productpage-without-pod --> controller/controllerName
func (c *ConnectOptions) PreCheckResource() error {
if len(c.Workloads) == 0 {
return nil
}
list, err := util.GetUnstructuredObjectList(c.factory, c.Namespace, c.Workloads)
if err != nil {
return err
}
var resources []string
for _, info := range list {
resources = append(resources, fmt.Sprintf("%s/%s", info.Mapping.GroupVersionKind.GroupKind().String(), info.Name))
}
c.Workloads = resources
// normal workloads, like pod with controller, deployments, statefulset, replicaset etc...
for i, workload := range c.Workloads {
ownerReference, err := util.GetTopOwnerReference(c.factory, c.Namespace, workload)
if err == nil {
c.Workloads[i] = fmt.Sprintf("%s/%s", ownerReference.Mapping.GroupVersionKind.GroupKind().String(), ownerReference.Name)
}
}
// service which associate with pod
for i, workload := range c.Workloads {
object, err := util.GetUnstructuredObject(c.factory, c.Namespace, workload)
if err != nil {
return err
}
if object.Mapping.Resource.Resource != "services" {
continue
}
get, err := c.clientset.CoreV1().Services(c.Namespace).Get(context.Background(), object.Name, metav1.GetOptions{})
if err != nil {
continue
}
if ns, selector, err := polymorphichelpers.SelectorsForObject(get); err == nil {
list, err := c.clientset.CoreV1().Pods(ns).List(context.Background(), metav1.ListOptions{
LabelSelector: selector.String(),
})
// if pod is not empty, using pods to find top controller
if err == nil && list != nil && len(list.Items) != 0 {
ownerReference, err := util.GetTopOwnerReference(c.factory, c.Namespace, fmt.Sprintf("%s/%s", "pods", list.Items[0].Name))
if err == nil {
c.Workloads[i] = fmt.Sprintf("%s/%s", ownerReference.Mapping.GroupVersionKind.GroupKind().String(), ownerReference.Name)
}
} else
// if list is empty, means not create pods, just controllers
{
controller, err := util.GetTopOwnerReferenceBySelector(c.factory, c.Namespace, selector.String())
if err == nil {
if len(controller) > 0 {
c.Workloads[i] = controller.UnsortedList()[0]
}
}
// only a single service, not support it yet
if controller.Len() == 0 {
return fmt.Errorf("not support resources: %s", workload)
}
}
}
}
return nil
}
func (c *ConnectOptions) GetRunningPodList(ctx context.Context) ([]v1.Pod, error) {
label := fields.OneTermEqualSelector("app", config.ConfigMapPodTrafficManager).String()
return util.GetRunningPodList(ctx, c.clientset, c.Namespace, label)
}
// getCIDR
// 1: get pod cidr
// 2: get service cidr
// distinguish service cidr and pod cidr
// https://stackoverflow.com/questions/45903123/kubernetes-set-service-cidr-and-pod-cidr-the-same
// https://stackoverflow.com/questions/44190607/how-do-you-find-the-cluster-service-cidr-of-a-kubernetes-cluster/54183373#54183373
// https://stackoverflow.com/questions/44190607/how-do-you-find-the-cluster-service-cidr-of-a-kubernetes-cluster
func (c *ConnectOptions) getCIDR(ctx context.Context, m *dhcp.Manager) (err error) {
defer func() {
if err == nil {
u, err2 := url.Parse(c.config.Host)
if err2 != nil {
return
}
host, _, err3 := net.SplitHostPort(u.Host)
if err3 != nil {
return
}
var ipList []net.IP
if ip := net.ParseIP(host); ip != nil {
ipList = append(ipList, ip)
}
ips, _ := net.LookupIP(host)
if ips != nil {
ipList = append(ipList, ips...)
}
c.apiServerIPs = ipList
for i := 0; i < len(c.cidrs); i++ {
for _, ip := range ipList {
if c.cidrs[i].Contains(ip) {
c.cidrs = append(c.cidrs[:i], c.cidrs[i+1:]...)
i--
}
}
}
}
}()
// (1) get CIDR from cache
var value string
value, err = m.Get(ctx, config.KeyClusterIPv4POOLS)
if err == nil {
for _, s := range strings.Split(value, " ") {
_, cidr, _ := net.ParseCIDR(s)
if cidr != nil {
c.cidrs = util.RemoveLargerOverlappingCIDRs(append(c.cidrs, cidr))
}
}
if len(c.cidrs) != 0 {
log.Infoln("Got network CIDR from cache")
return
}
}
// (2) get CIDR from cni
c.cidrs, err = util.GetCIDRElegant(ctx, c.clientset, c.config, c.Namespace)
if err == nil {
s := sets.New[string]()
for _, cidr := range c.cidrs {
s.Insert(cidr.String())
}
cidrs := util.GetCIDRFromResourceUgly(ctx, c.clientset, c.Namespace)
for _, cidr := range cidrs {
s.Insert(cidr.String())
}
c.cidrs = util.RemoveLargerOverlappingCIDRs(append(c.cidrs, cidrs...))
_ = m.Set(ctx, config.KeyClusterIPv4POOLS, strings.Join(s.UnsortedList(), " "))
return
}
// (3) fallback to get cidr from node/pod/service
c.cidrs = util.GetCIDRFromResourceUgly(ctx, c.clientset, c.Namespace)
return
}
func (c *ConnectOptions) addExtraRoute(ctx context.Context, nameserver string) error {
if len(c.ExtraRouteInfo.ExtraDomain) == 0 {
return nil
}
tunName, err := c.GetTunDeviceName()
if err != nil {
log.Errorf("Get tun interface failed: %s", err.Error())
return err
}
addRouteFunc := func(resource, ip string) {
if net.ParseIP(ip) == nil {
return
}
var mask net.IPMask
if net.ParseIP(ip).To4() != nil {
mask = net.CIDRMask(32, 32)
} else {
mask = net.CIDRMask(128, 128)
}
errs := tun.AddRoutes(tunName, types.Route{Dst: net.IPNet{IP: net.ParseIP(ip), Mask: mask}})
if errs != nil {
log.Errorf("Failed to add route, domain: %s, IP: %s, err: %v", resource, ip, err)
}
}
// 1) use dig +short query, if ok, just return
podList, err := c.GetRunningPodList(ctx)
if err != nil {
return err
}
var ok = true
for _, domain := range c.ExtraRouteInfo.ExtraDomain {
ip, err := util.Shell(ctx, c.clientset, c.config, podList[0].Name, config.ContainerSidecarVPN, c.Namespace, []string{"dig", "+short", domain})
if err == nil || net.ParseIP(ip) != nil {
addRouteFunc(domain, ip)
c.extraHost = append(c.extraHost, dns.Entry{IP: net.ParseIP(ip).String(), Domain: domain})
} else {
ok = false
}
}
if ok {
return nil
}
// 2) wait until can ping dns server ip ok
// 3) use nslookup to query dns at first, it will speed up mikdns query process
ticker := time.NewTicker(2 * time.Second)
defer ticker.Stop()
go func() {
for _, domain := range c.ExtraRouteInfo.ExtraDomain {
go func(domain string) {
for ; true; <-ticker.C {
func() {
subCtx, c2 := context.WithTimeout(ctx, time.Second*2)
defer c2()
cmd := exec.CommandContext(subCtx, "nslookup", domain, nameserver)
cmd.Stderr = io.Discard
cmd.Stdout = io.Discard
_ = cmd.Start()
_ = cmd.Wait()
}()
}
}(domain)
}
}()
// 4) query with dns client
client := &miekgdns.Client{Net: "udp", Timeout: time.Second * 2}
for _, domain := range c.ExtraRouteInfo.ExtraDomain {
var success = false
for _, qType := range []uint16{miekgdns.TypeA /*, miekgdns.TypeAAAA*/} {
var iErr = errors.New("No retry")
err = retry.OnError(
wait.Backoff{
Steps: 1000,
Duration: time.Millisecond * 30,
},
func(err error) bool {
return err != nil
},
func() error {
var answer *miekgdns.Msg
answer, _, err = client.ExchangeContext(ctx, &miekgdns.Msg{
MsgHdr: miekgdns.MsgHdr{
Id: uint16(rand.Intn(math.MaxUint16 + 1)),
},
Question: []miekgdns.Question{
{
Name: domain + ".",
Qtype: qType,
},
},
}, fmt.Sprintf("%s:%d", nameserver, 53))
if err != nil {
return err
}
if len(answer.Answer) == 0 {
return iErr
}
for _, rr := range answer.Answer {
switch a := rr.(type) {
case *miekgdns.A:
if ip := net.ParseIP(a.A.String()); ip != nil && !ip.IsLoopback() {
addRouteFunc(domain, a.A.String())
c.extraHost = append(c.extraHost, dns.Entry{IP: a.A.String(), Domain: domain})
success = true
}
case *miekgdns.AAAA:
if ip := net.ParseIP(a.AAAA.String()); ip != nil && !ip.IsLoopback() {
addRouteFunc(domain, a.AAAA.String())
c.extraHost = append(c.extraHost, dns.Entry{IP: a.AAAA.String(), Domain: domain})
success = true
}
}
}
return nil
})
if err != nil && err != iErr {
return err
}
if success {
break
}
}
if !success {
return fmt.Errorf("failed to resolve DNS for domain %s", domain)
}
}
return nil
}
func (c *ConnectOptions) addExtraNodeIP(ctx context.Context) error {
if !c.ExtraRouteInfo.ExtraNodeIP {
return nil
}
list, err := c.clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
if err != nil {
return err
}
for _, item := range list.Items {
for _, address := range item.Status.Addresses {
ip := net.ParseIP(address.Address)
if ip != nil {
var mask net.IPMask
if ip.To4() != nil {
mask = net.CIDRMask(32, 32)
} else {
mask = net.CIDRMask(128, 128)
}
c.ExtraRouteInfo.ExtraCIDR = append(c.ExtraRouteInfo.ExtraCIDR, (&net.IPNet{
IP: ip,
Mask: mask,
}).String())
}
}
}
return nil
}
func (c *ConnectOptions) GetClientset() *kubernetes.Clientset {
return c.clientset
}
func (c *ConnectOptions) GetFactory() cmdutil.Factory {
return c.factory
}
func (c *ConnectOptions) GetLocalTunIP() (v4 string, v6 string) {
if c.localTunIPv4 != nil {
v4 = c.localTunIPv4.IP.String()
}
if c.localTunIPv6 != nil {
v6 = c.localTunIPv6.IP.String()
}
return
}
func (c *ConnectOptions) GetClusterID() string {
if c != nil && c.dhcp != nil {
return string(c.dhcp.GetClusterID())
}
return ""
}
func (c *ConnectOptions) upgradeDeploy(ctx context.Context) error {
deploy, err := c.clientset.AppsV1().Deployments(c.Namespace).Get(ctx, config.ConfigMapPodTrafficManager, metav1.GetOptions{})
if err != nil {
return err
}
if len(deploy.Spec.Template.Spec.Containers) == 0 {
return fmt.Errorf("can not found any container in deploy %s", deploy.Name)
}
clientImg := config.Image
serverImg := deploy.Spec.Template.Spec.Containers[0].Image
if clientImg == serverImg {
return nil
}
isNewer, _ := newer(clientImg, serverImg)
if deploy.Status.ReadyReplicas > 0 && !isNewer {
return nil
}
log.Infof("Set image %s --> %s...", serverImg, clientImg)
r := c.factory.NewBuilder().
WithScheme(scheme.Scheme, scheme.Scheme.PrioritizedVersionsAllGroups()...).
NamespaceParam(c.Namespace).DefaultNamespace().
ResourceNames("deployments", deploy.Name).
ContinueOnError().
Latest().
Flatten().
Do()
if err = r.Err(); err != nil {
return err
}
infos, err := r.Infos()
if err != nil {
return err
}
patches := set.CalculatePatches(infos, scheme.DefaultJSONEncoder(), func(obj pkgruntime.Object) ([]byte, error) {
_, err = polymorphichelpers.UpdatePodSpecForObjectFn(obj, func(spec *v1.PodSpec) error {
for i := range spec.Containers {
spec.Containers[i].Image = clientImg
}
return nil
})
if err != nil {
return nil, err
}
return pkgruntime.Encode(scheme.DefaultJSONEncoder(), obj)
})
for _, p := range patches {
if p.Err != nil {
return p.Err
}
}
for _, p := range patches {
_, err = resource.
NewHelper(p.Info.Client, p.Info.Mapping).
DryRun(false).
Patch(p.Info.Namespace, p.Info.Name, pkgtypes.StrategicMergePatchType, p.Patch, nil)
if err != nil {
log.Errorf("Failed to patch image update to pod template: %v", err)
return err
}
err = util.RolloutStatus(ctx, c.factory, c.Namespace, fmt.Sprintf("%s/%s", p.Info.Mapping.Resource.GroupResource().String(), p.Info.Name), time.Minute*60)
if err != nil {
return err
}
}
return nil
}
func newer(clientImgStr, serverImgStr string) (bool, error) {
clientImg, err := reference.ParseNormalizedNamed(clientImgStr)
if err != nil {
return false, err
}
serverImg, err := reference.ParseNormalizedNamed(serverImgStr)
if err != nil {
return false, err
}
if reference.Domain(clientImg) != reference.Domain(serverImg) {
return false, nil
}
serverTag, ok := serverImg.(reference.NamedTagged)
if !ok {
return false, fmt.Errorf("can not convert server image")
}
serverVersion, err := goversion.NewVersion(serverTag.Tag())
if err != nil {
return false, err
}
clientTag, ok := clientImg.(reference.NamedTagged)
if !ok {
return false, fmt.Errorf("can not convert client image")
}
clientVersion, err := goversion.NewVersion(clientTag.Tag())
if err != nil {
return false, err
}
return clientVersion.GreaterThan(serverVersion), nil
}
func (c *ConnectOptions) Equal(a *ConnectOptions) bool {
return c.Engine == a.Engine &&
sets.New[string](c.ExtraRouteInfo.ExtraDomain...).HasAll(a.ExtraRouteInfo.ExtraDomain...) &&
sets.New[string](c.ExtraRouteInfo.ExtraCIDR...).HasAll(c.ExtraRouteInfo.ExtraCIDR...) &&
(reflect.DeepEqual(c.ExtraRouteInfo.ExtraNodeIP, a.ExtraRouteInfo.ExtraNodeIP) || c.ExtraRouteInfo.ExtraNodeIP == true)
}
func (c *ConnectOptions) GetTunDeviceName() (string, error) {
var ips []net.IP
if c.localTunIPv4 != nil {
ips = append(ips, c.localTunIPv4.IP)
}
if c.localTunIPv6 != nil {
ips = append(ips, c.localTunIPv6.IP)
}
device, err := util.GetTunDevice(ips...)
if err != nil {
return "", err
}
return device.Name, nil
}
func (c *ConnectOptions) AddRolloutFunc(f func() error) {
c.rollbackFuncList = append(c.rollbackFuncList, f)
}
func (c *ConnectOptions) getRolloutFunc() []func() error {
return c.rollbackFuncList
}