Skip to content

Commit da40f33

Browse files
wencaiwuluewencaiwulue
authoredApr 27, 2025··
Merge pull request #566 from kubenetworks/hotfix/fix-bugs
hotfix: fix bugs
·
v2.7.7v2.7.3
2 parents bed0a91 + c4540b1 commit da40f33

File tree

6 files changed

+45
-50
lines changed

6 files changed

+45
-50
lines changed
 

‎cmd/kubevpn/cmds/upgrade.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -49,8 +49,8 @@ func CmdUpgrade(cmdutil.Factory) *cobra.Command {
4949
}
5050
_, _ = fmt.Fprintln(os.Stdout, fmt.Sprintf("Current version is: %s less than latest version: %s, needs to upgrade", config.Version, latestVersion))
5151
_ = os.Setenv(envLatestUrl, url)
52-
_ = quit(cmd.Context(), false)
5352
_ = quit(cmd.Context(), true)
53+
_ = quit(cmd.Context(), false)
5454
}
5555
return upgrade.Main(cmd.Context(), client, url)
5656
},

‎pkg/core/ssh.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ func (s *sshHandler) Handle(ctx context.Context, conn net.Conn) {
3838
}),
3939
Handler: ssh.Handler(func(s ssh.Session) {
4040
io.WriteString(s, "Remote forwarding available...\n")
41-
select {}
41+
<-s.Context().Done()
4242
}),
4343
ReversePortForwardingCallback: ssh.ReversePortForwardingCallback(func(ctx ssh.Context, host string, port uint32) bool {
4444
plog.G(ctx).Infoln("attempt to bind", host, port, "granted")

‎pkg/core/tcphandler.go

+6-6
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,7 @@ func (h *UDPOverTCPHandler) removeFromRouteMapTCP(ctx context.Context, tcpConn n
128128
})
129129
}
130130

131-
var _ net.PacketConn = (*UDPConnOverTCP)(nil)
131+
var _ net.Conn = (*UDPConnOverTCP)(nil)
132132

133133
// UDPConnOverTCP fake udp connection over tcp connection
134134
type UDPConnOverTCP struct {
@@ -141,20 +141,20 @@ func newUDPConnOverTCP(ctx context.Context, conn net.Conn) (net.Conn, error) {
141141
return &UDPConnOverTCP{ctx: ctx, Conn: conn}, nil
142142
}
143143

144-
func (c *UDPConnOverTCP) ReadFrom(b []byte) (int, net.Addr, error) {
144+
func (c *UDPConnOverTCP) Read(b []byte) (int, error) {
145145
select {
146146
case <-c.ctx.Done():
147-
return 0, nil, c.ctx.Err()
147+
return 0, c.ctx.Err()
148148
default:
149149
datagram, err := readDatagramPacket(c.Conn, b)
150150
if err != nil {
151-
return 0, nil, err
151+
return 0, err
152152
}
153-
return int(datagram.DataLength), nil, nil
153+
return int(datagram.DataLength), nil
154154
}
155155
}
156156

157-
func (c *UDPConnOverTCP) WriteTo(b []byte, _ net.Addr) (int, error) {
157+
func (c *UDPConnOverTCP) Write(b []byte) (int, error) {
158158
buf := config.LPool.Get().([]byte)[:]
159159
n := copy(buf, b)
160160
defer config.LPool.Put(buf)

‎pkg/core/tunhandler.go

+1-6
Original file line numberDiff line numberDiff line change
@@ -37,12 +37,7 @@ func TunHandler(forward *Forwarder, node *Node) Handler {
3737

3838
func (h *tunHandler) Handle(ctx context.Context, tun net.Conn) {
3939
if remote := h.node.Remote; remote != "" {
40-
remoteAddr, err := net.ResolveUDPAddr("udp", remote)
41-
if err != nil {
42-
plog.G(ctx).Errorf("Failed to resolve udp addr %s: %v", remote, err)
43-
return
44-
}
45-
h.HandleClient(ctx, tun, remoteAddr)
40+
h.HandleClient(ctx, tun)
4641
} else {
4742
h.HandleServer(ctx, tun)
4843
}

‎pkg/core/tunhandlerclient.go

+25-35
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ import (
1414
"github.com/wencaiwulue/kubevpn/v2/pkg/util"
1515
)
1616

17-
func (h *tunHandler) HandleClient(ctx context.Context, tun net.Conn, remoteAddr *net.UDPAddr) {
17+
func (h *tunHandler) HandleClient(ctx context.Context, tun net.Conn) {
1818
device := &ClientDevice{
1919
tun: tun,
2020
tunInbound: make(chan *Packet, MaxSize),
@@ -23,7 +23,7 @@ func (h *tunHandler) HandleClient(ctx context.Context, tun net.Conn, remoteAddr
2323
}
2424

2525
defer device.Close()
26-
go device.handlePacket(ctx, remoteAddr, h.forward)
26+
go device.handlePacket(ctx, h.forward)
2727
go device.readFromTun(ctx)
2828
go device.writeToTun(ctx)
2929
go heartbeats(ctx, device.tun)
@@ -43,56 +43,40 @@ type ClientDevice struct {
4343
forward *Forwarder
4444
}
4545

46-
func (d *ClientDevice) handlePacket(ctx context.Context, remoteAddr *net.UDPAddr, forward *Forwarder) {
46+
func (d *ClientDevice) handlePacket(ctx context.Context, forward *Forwarder) {
4747
for ctx.Err() == nil {
48-
packetConn, err := getRemotePacketConn(ctx, forward)
48+
conn, err := forwardConn(ctx, forward)
4949
if err != nil {
50-
plog.G(ctx).Errorf("Failed to get remote conn from %s -> %s: %s", d.tun.LocalAddr(), remoteAddr, err)
50+
plog.G(ctx).Errorf("Failed to get remote conn from %s -> %s: %s", d.tun.LocalAddr(), forward.node.Remote, err)
5151
time.Sleep(time.Second * 1)
5252
continue
5353
}
54-
err = handlePacketClient(ctx, d.tunInbound, d.tunOutbound, packetConn, remoteAddr)
54+
err = handlePacketClient(ctx, d.tunInbound, d.tunOutbound, conn)
5555
if err != nil {
56-
plog.G(ctx).Errorf("Failed to transport data to remote %s: %v", remoteAddr, err)
56+
plog.G(ctx).Errorf("Failed to transport data to remote %s: %v", conn.RemoteAddr(), err)
5757
}
5858
}
5959
}
6060

61-
func getRemotePacketConn(ctx context.Context, forwarder *Forwarder) (net.PacketConn, error) {
61+
func forwardConn(ctx context.Context, forwarder *Forwarder) (net.Conn, error) {
6262
conn, err := forwarder.DialContext(ctx)
6363
if err != nil {
6464
return nil, errors.Wrap(err, "failed to dial forwarder")
6565
}
66-
67-
if packetConn, ok := conn.(net.PacketConn); !ok {
68-
return nil, errors.Errorf("failed to cast packet conn to PacketConn")
69-
} else {
70-
return packetConn, nil
71-
}
66+
return conn, nil
7267
}
7368

74-
func handlePacketClient(ctx context.Context, tunInbound <-chan *Packet, tunOutbound chan<- *Packet, packetConn net.PacketConn, remoteAddr net.Addr) error {
69+
func handlePacketClient(ctx context.Context, tunInbound <-chan *Packet, tunOutbound chan<- *Packet, conn net.Conn) error {
7570
errChan := make(chan error, 2)
76-
defer packetConn.Close()
71+
defer conn.Close()
7772

7873
go func() {
7974
defer util.HandleCrash()
8075
for packet := range tunInbound {
81-
if packet.src.Equal(packet.dst) {
82-
util.SafeWrite(tunOutbound, packet, func(v *Packet) {
83-
var p = "unknown"
84-
if _, _, protocol, err := util.ParseIP(v.data[:v.length]); err == nil {
85-
p = layers.IPProtocol(protocol).String()
86-
}
87-
config.LPool.Put(v.data[:])
88-
plog.G(context.Background()).Errorf("Drop packet, SRC: %s, DST: %s, Protocol: %s, Length: %d", v.src, v.dst, p, v.length)
89-
})
90-
continue
91-
}
92-
_, err := packetConn.WriteTo(packet.data[:packet.length], remoteAddr)
76+
_, err := conn.Write(packet.data[:packet.length])
9377
config.LPool.Put(packet.data[:])
9478
if err != nil {
95-
util.SafeWrite(errChan, errors.Wrap(err, fmt.Sprintf("failed to write packet to remote %s", remoteAddr)))
79+
util.SafeWrite(errChan, errors.Wrap(err, "failed to write packet to remote"))
9680
return
9781
}
9882
}
@@ -102,10 +86,10 @@ func handlePacketClient(ctx context.Context, tunInbound <-chan *Packet, tunOutbo
10286
defer util.HandleCrash()
10387
for {
10488
buf := config.LPool.Get().([]byte)[:]
105-
n, _, err := packetConn.ReadFrom(buf[:])
89+
n, err := conn.Read(buf[:])
10690
if err != nil {
10791
config.LPool.Put(buf[:])
108-
util.SafeWrite(errChan, errors.Wrap(err, fmt.Sprintf("failed to read packet from remote %s", remoteAddr)))
92+
util.SafeWrite(errChan, errors.Wrap(err, fmt.Sprintf("failed to read packet from remote %s", conn.RemoteAddr())))
10993
return
11094
}
11195
if n == 0 {
@@ -115,7 +99,7 @@ func handlePacketClient(ctx context.Context, tunInbound <-chan *Packet, tunOutbo
11599
}
116100
util.SafeWrite(tunOutbound, NewPacket(buf[:], n, nil, nil), func(v *Packet) {
117101
config.LPool.Put(v.data[:])
118-
plog.G(context.Background()).Errorf("Drop packet, LocalAddr: %s, Remote: %s, Length: %d", packetConn.LocalAddr(), remoteAddr, v.length)
102+
plog.G(context.Background()).Errorf("Drop packet, LocalAddr: %s, Remote: %s, Length: %d", conn.LocalAddr(), conn.RemoteAddr(), v.length)
119103
})
120104
}
121105
}()
@@ -150,10 +134,16 @@ func (d *ClientDevice) readFromTun(ctx context.Context) {
150134
continue
151135
}
152136
plog.G(context.Background()).Debugf("SRC: %s, DST: %s, Protocol: %s, Length: %d", src, dst, layers.IPProtocol(protocol).String(), n)
153-
util.SafeWrite(d.tunInbound, NewPacket(buf[:], n, src, dst), func(v *Packet) {
137+
packet := NewPacket(buf[:], n, src, dst)
138+
f := func(v *Packet) {
154139
config.LPool.Put(v.data[:])
155140
plog.G(context.Background()).Errorf("Drop packet, SRC: %s, DST: %s, Protocol: %s, Length: %d", v.src, v.dst, layers.IPProtocol(protocol).String(), v.length)
156-
})
141+
}
142+
if packet.src.Equal(packet.dst) {
143+
util.SafeWrite(d.tunOutbound, packet, f)
144+
continue
145+
}
146+
util.SafeWrite(d.tunInbound, packet, f)
157147
}
158148
}
159149

@@ -188,7 +178,7 @@ func heartbeats(ctx context.Context, tun net.Conn) {
188178
return
189179
}
190180

191-
ticker := time.NewTicker(time.Second * 60)
181+
ticker := time.NewTicker(config.KeepAliveTime)
192182
defer ticker.Stop()
193183

194184
for ; ctx.Err() == nil; <-ticker.C {

‎pkg/handler/connect.go

+11-1
Original file line numberDiff line numberDiff line change
@@ -955,18 +955,28 @@ func (c *ConnectOptions) upgradeDeploy(ctx context.Context) error {
955955
if len(deploy.Spec.Template.Spec.Containers) == 0 {
956956
return fmt.Errorf("can not found any container in deploy %s", deploy.Name)
957957
}
958+
// check running pod, sometime deployment is rolling back, so need to check running pod
959+
list, err := c.GetRunningPodList(ctx)
960+
if err != nil {
961+
return err
962+
}
958963

959964
clientVer := config.Version
960965
clientImg := config.Image
961966
serverImg := deploy.Spec.Template.Spec.Containers[0].Image
967+
runningPodImg := list[0].Spec.Containers[0].Image
962968

963969
isNeedUpgrade, err := util.IsNewer(clientVer, clientImg, serverImg)
964-
if !isNeedUpgrade {
970+
isPodNeedUpgrade, err1 := util.IsNewer(clientVer, clientImg, runningPodImg)
971+
if !isNeedUpgrade && !isPodNeedUpgrade {
965972
return nil
966973
}
967974
if err != nil {
968975
return err
969976
}
977+
if err1 != nil {
978+
return err1
979+
}
970980

971981
// 1) update secret
972982
err = upgradeSecretSpec(ctx, c.factory, c.Namespace)

0 commit comments

Comments
 (0)