Skip to content

Commit ba9bcec

Browse files
zhangzujianzhangzujian
authored
bind to pod ips when env variable ENABLE_BIND_LOCAL_IP is set to true (#5049)
Signed-off-by: zhangzujian <zhangzujian.7@gmail.com>
1 parent 6f4100d commit ba9bcec

File tree

10 files changed

+117
-97
lines changed

10 files changed

+117
-97
lines changed

‎cmd/controller/controller.go‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"net/http"
88
"net/http/pprof"
99
"os"
10+
"slices"
1011
"time"
1112

1213
v1 "k8s.io/api/authorization/v1"
@@ -53,8 +54,8 @@ func CmdMain() {
5354
ctrl.SetLogger(klog.NewKlogr())
5455
ctx := signals.SetupSignalHandler()
5556
go func() {
56-
metricsAddr := util.GetDefaultListenAddr()
57-
servePprofInMetricsServer := config.EnableMetrics && metricsAddr == "0.0.0.0"
57+
metricsAddrs := util.GetDefaultListenAddr()
58+
servePprofInMetricsServer := config.EnableMetrics && slices.Contains(metricsAddrs, "0.0.0.0")
5859
if config.EnablePprof && !servePprofInMetricsServer {
5960
mux := http.NewServeMux()
6061
mux.HandleFunc("/debug/pprof/", pprof.Index)
@@ -87,15 +88,19 @@ func CmdMain() {
8788
if config.EnableMetrics {
8889
metrics.InitKlogMetrics()
8990
metrics.InitClientGoMetrics()
90-
addr := util.JoinHostPort(metricsAddr, config.PprofPort)
91-
if err := metrics.Run(ctx, config.KubeRestConfig, addr, config.SecureServing, servePprofInMetricsServer); err != nil {
92-
util.LogFatalAndExit(err, "failed to run metrics server")
91+
for _, metricsAddr := range metricsAddrs {
92+
addr := util.JoinHostPort(metricsAddr, config.PprofPort)
93+
go func() {
94+
if err := metrics.Run(ctx, config.KubeRestConfig, addr, config.SecureServing, servePprofInMetricsServer); err != nil {
95+
util.LogFatalAndExit(err, "failed to run metrics server")
96+
}
97+
}()
9398
}
9499
} else {
95100
klog.Info("metrics server is disabled")
96-
listerner, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.ParseIP(metricsAddr), Port: int(config.PprofPort)})
101+
listerner, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.ParseIP(metricsAddrs[0]), Port: int(config.PprofPort)})
97102
if err != nil {
98-
util.LogFatalAndExit(err, "failed to listen on %s", util.JoinHostPort(metricsAddr, config.PprofPort))
103+
util.LogFatalAndExit(err, "failed to listen on %s", util.JoinHostPort(metricsAddrs[0], config.PprofPort))
99104
}
100105
mux := http.NewServeMux()
101106
mux.HandleFunc("/healthz", util.DefaultHealthCheckHandler)

‎cmd/daemon/cniserver.go‎

Lines changed: 26 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import (
88
"net/http/pprof"
99
"os"
1010
"path/filepath"
11+
"slices"
1112
"strings"
1213
"time"
1314

@@ -99,24 +100,25 @@ func main() {
99100
go ctl.Run(stopCh)
100101
go daemon.RunServer(config, ctl)
101102

102-
addr := util.GetDefaultListenAddr()
103+
addrs := util.GetDefaultListenAddr()
103104
if config.EnableVerboseConnCheck {
104-
go func() {
105-
connListenaddr := util.JoinHostPort(addr, config.TCPConnCheckPort)
106-
if err := util.TCPConnectivityListen(connListenaddr); err != nil {
107-
util.LogFatalAndExit(err, "failed to start TCP listen on addr %s", addr)
108-
}
109-
}()
110-
111-
go func() {
112-
connListenaddr := util.JoinHostPort(addr, config.UDPConnCheckPort)
113-
if err := util.UDPConnectivityListen(connListenaddr); err != nil {
114-
util.LogFatalAndExit(err, "failed to start UDP listen on addr %s", addr)
115-
}
116-
}()
105+
for _, addr := range addrs {
106+
go func() {
107+
connListenaddr := util.JoinHostPort(addr, config.TCPConnCheckPort)
108+
if err := util.TCPConnectivityListen(connListenaddr); err != nil {
109+
util.LogFatalAndExit(err, "failed to start TCP listen on addr %s", addr)
110+
}
111+
}()
112+
go func() {
113+
connListenaddr := util.JoinHostPort(addr, config.UDPConnCheckPort)
114+
if err := util.UDPConnectivityListen(connListenaddr); err != nil {
115+
util.LogFatalAndExit(err, "failed to start UDP listen on addr %s", addr)
116+
}
117+
}()
118+
}
117119
}
118120

119-
servePprofInMetricsServer := config.EnableMetrics && addr == "0.0.0.0"
121+
servePprofInMetricsServer := config.EnableMetrics && slices.Contains(addrs, "0.0.0.0")
120122
if config.EnablePprof && !servePprofInMetricsServer {
121123
mux := http.NewServeMux()
122124
mux.HandleFunc("/debug/pprof/", pprof.Index)
@@ -149,15 +151,19 @@ func main() {
149151
if config.EnableMetrics {
150152
daemon.InitMetrics()
151153
metrics.InitKlogMetrics()
152-
listenAddr := util.JoinHostPort(addr, config.PprofPort)
153-
if err = metrics.Run(ctx, nil, listenAddr, config.SecureServing, servePprofInMetricsServer); err != nil {
154-
util.LogFatalAndExit(err, "failed to run metrics server")
154+
for _, addr := range addrs {
155+
listenAddr := util.JoinHostPort(addr, config.PprofPort)
156+
go func() {
157+
if err := metrics.Run(ctx, nil, listenAddr, config.SecureServing, servePprofInMetricsServer); err != nil {
158+
util.LogFatalAndExit(err, "failed to run metrics server")
159+
}
160+
}()
155161
}
156162
} else {
157163
klog.Info("metrics server is disabled")
158-
listerner, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.ParseIP(addr), Port: int(config.PprofPort)})
164+
listerner, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.ParseIP(addrs[0]), Port: int(config.PprofPort)})
159165
if err != nil {
160-
util.LogFatalAndExit(err, "failed to listen on %s", util.JoinHostPort(addr, config.PprofPort))
166+
util.LogFatalAndExit(err, "failed to listen on %s", util.JoinHostPort(addrs[0], config.PprofPort))
161167
}
162168
mux := http.NewServeMux()
163169
mux.HandleFunc("/healthz", util.DefaultHealthCheckHandler)

‎cmd/ovn_monitor/ovn_monitor.go‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -33,23 +33,27 @@ func CmdMain() {
3333
ctrl.SetLogger(klog.NewKlogr())
3434
ctx := signals.SetupSignalHandler()
3535

36-
metricsAddr := util.GetDefaultListenAddr()
36+
metricsAddrs := util.GetDefaultListenAddr()
3737
if config.EnableMetrics {
3838
exporter := ovn.NewExporter(config)
3939
if err = exporter.StartConnection(); err != nil {
4040
klog.Errorf("%s failed to connect db socket properly: %s", ovn.GetExporterName(), err)
4141
go exporter.TryClientConnection()
4242
}
4343
exporter.StartOvnMetrics()
44-
addr := util.JoinHostPort(metricsAddr, config.MetricsPort)
45-
if err = metrics.Run(ctx, nil, addr, config.SecureServing, false); err != nil {
46-
util.LogFatalAndExit(err, "failed to run metrics server")
44+
for _, metricsAddr := range metricsAddrs {
45+
addr := util.JoinHostPort(metricsAddr, config.MetricsPort)
46+
go func() {
47+
if err := metrics.Run(ctx, nil, addr, config.SecureServing, false); err != nil {
48+
util.LogFatalAndExit(err, "failed to run metrics server")
49+
}
50+
}()
4751
}
4852
} else {
4953
klog.Info("metrics server is disabled")
50-
listerner, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.ParseIP(util.GetDefaultListenAddr()), Port: int(config.MetricsPort)})
54+
listerner, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.ParseIP(metricsAddrs[0]), Port: int(config.MetricsPort)})
5155
if err != nil {
52-
util.LogFatalAndExit(err, "failed to listen on %s", util.JoinHostPort(metricsAddr, config.MetricsPort))
56+
util.LogFatalAndExit(err, "failed to listen on %s", util.JoinHostPort(metricsAddrs[0], config.MetricsPort))
5357
}
5458
mux := http.NewServeMux()
5559
mux.HandleFunc("/healthz", util.DefaultHealthCheckHandler)

‎pkg/daemon/config.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,8 @@ type Configuration struct {
4949
KubeClient kubernetes.Interface
5050
KubeOvnClient clientset.Interface
5151
NodeName string
52+
NodeIPv4 string
53+
NodeIPv6 string
5254
ServiceClusterIPRange string
5355
ClusterRouter string
5456
NodeSwitch string
@@ -213,6 +215,7 @@ func (config *Configuration) initNicConfig(nicBridgeMappings map[string]string)
213215
klog.Errorf("Failed to find node info, err: %v", err)
214216
return err
215217
}
218+
config.NodeIPv4, config.NodeIPv6 = util.GetNodeInternalIP(*node)
216219
if nodeTunnelName := node.GetAnnotations()[util.TunnelInterfaceAnnotation]; nodeTunnelName != "" {
217220
config.Iface = nodeTunnelName
218221
klog.Infof("Find node tunnel interface name: %v", nodeTunnelName)

‎pkg/daemon/gateway_linux.go‎

Lines changed: 7 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -642,14 +642,11 @@ func (c *Controller) setIptables() error {
642642
{Table: MANGLE, Chain: OvnPostrouting, Rule: strings.Fields(`-p tcp -m set --match-set ovn60subnets src -m tcp --tcp-flags RST RST -m state --state INVALID -j DROP`)},
643643
}
644644
)
645-
protocols := make([]string, 2)
646-
isDual := false
645+
protocols := make([]string, 0, 2)
647646
if c.protocol == kubeovnv1.ProtocolDual {
648-
protocols[0] = kubeovnv1.ProtocolIPv4
649-
protocols[1] = kubeovnv1.ProtocolIPv6
650-
isDual = true
647+
protocols = append(protocols, kubeovnv1.ProtocolIPv4, kubeovnv1.ProtocolIPv6)
651648
} else {
652-
protocols[0] = c.protocol
649+
protocols = append(protocols, c.protocol)
653650
}
654651

655652
for _, protocol := range protocols {
@@ -837,7 +834,7 @@ func (c *Controller) setIptables() error {
837834
return err
838835
}
839836

840-
if err = c.reconcileTProxyIPTableRules(protocol, isDual); err != nil {
837+
if err = c.reconcileTProxyIPTableRules(protocol); err != nil {
841838
klog.Error(err)
842839
return err
843840
}
@@ -868,7 +865,7 @@ func (c *Controller) setIptables() error {
868865
return nil
869866
}
870867

871-
func (c *Controller) reconcileTProxyIPTableRules(protocol string, isDual bool) error {
868+
func (c *Controller) reconcileTProxyIPTableRules(protocol string) error {
872869
if !c.config.EnableTProxy {
873870
return nil
874871
}
@@ -902,19 +899,13 @@ func (c *Controller) reconcileTProxyIPTableRules(protocol string, isDual bool) e
902899
}
903900

904901
for _, probePort := range ports.SortedList() {
905-
hostIP := pod.Status.HostIP
902+
hostIP := c.config.NodeIPv4
906903
prefixLen := 32
907904
if protocol == kubeovnv1.ProtocolIPv6 {
908905
prefixLen = 128
906+
hostIP = c.config.NodeIPv6
909907
}
910908

911-
if isDual || os.Getenv("ENABLE_BIND_LOCAL_IP") == "false" {
912-
if protocol == kubeovnv1.ProtocolIPv4 {
913-
hostIP = "0.0.0.0"
914-
} else if protocol == kubeovnv1.ProtocolIPv6 {
915-
hostIP = "::"
916-
}
917-
}
918909
tproxyOutputRules = append(tproxyOutputRules, util.IPTableRule{Table: MANGLE, Chain: OvnOutput, Rule: strings.Fields(fmt.Sprintf(`-d %s/%d -p tcp -m tcp --dport %d -j MARK --set-xmark %s`, podIP, prefixLen, probePort, tProxyOutputMarkMask))})
919910
tproxyPreRoutingRules = append(tproxyPreRoutingRules, util.IPTableRule{Table: MANGLE, Chain: OvnPrerouting, Rule: strings.Fields(fmt.Sprintf(`-d %s/%d -p tcp -m tcp --dport %d -j TPROXY --on-port %d --on-ip %s --tproxy-mark %s`, podIP, prefixLen, probePort, util.TProxyListenPort, hostIP, tProxyPreRoutingMarkMask))})
920911
}

‎pkg/daemon/tproxy_linux.go‎

Lines changed: 26 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -30,31 +30,34 @@ var (
3030
)
3131

3232
func (c *Controller) StartTProxyForwarding() {
33-
protocol := "tcp"
34-
addr := util.GetDefaultListenAddr()
35-
if util.CheckProtocol(addr) == kubeovnv1.ProtocolIPv6 {
36-
protocol = "tcp6"
37-
}
33+
for _, addr := range util.GetDefaultListenAddr() {
34+
protocol := "tcp"
35+
if util.CheckProtocol(addr) == kubeovnv1.ProtocolIPv6 {
36+
protocol = "tcp6"
37+
}
3838

39-
tcpListener, err := goTProxy.ListenTCP(protocol, &net.TCPAddr{IP: net.ParseIP(addr), Port: util.TProxyListenPort})
40-
if err != nil {
41-
klog.Fatalf("Encountered error while binding listener: %s", err)
42-
return
43-
}
39+
go func() {
40+
tcpListener, err := goTProxy.ListenTCP(protocol, &net.TCPAddr{IP: net.ParseIP(addr), Port: util.TProxyListenPort})
41+
if err != nil {
42+
klog.Fatalf("Encountered error while binding listener: %s", err)
43+
return
44+
}
4445

45-
defer func() {
46-
if err := tcpListener.Close(); err != nil {
47-
klog.Errorf("Error tcpListener Close err: %v", err)
48-
}
49-
}()
46+
defer func() {
47+
if err := tcpListener.Close(); err != nil {
48+
klog.Errorf("Error tcpListener Close err: %v", err)
49+
}
50+
}()
5051

51-
for {
52-
conn, err := tcpListener.Accept()
53-
if err != nil {
54-
klog.Fatalf("Unrecoverable error while accepting connection: %s", err)
55-
return
56-
}
57-
go handleRedirectFlow(conn)
52+
for {
53+
conn, err := tcpListener.Accept()
54+
if err != nil {
55+
klog.Fatalf("Unrecoverable error while accepting connection: %s", err)
56+
return
57+
}
58+
go handleRedirectFlow(conn)
59+
}
60+
}()
5861
}
5962
}
6063

@@ -90,7 +93,7 @@ func getProbePorts(pod *corev1.Pod) set.Set[int32] {
9093
}
9194

9295
ports.Delete(0)
93-
klog.Infof("probe ports for pod %s/%s: %v", pod.Namespace, pod.Name, ports.SortedList())
96+
klog.V(3).Infof("probe ports for pod %s/%s: %v", pod.Namespace, pod.Name, ports.SortedList())
9497
return ports
9598
}
9699

‎pkg/util/net.go‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -692,13 +692,15 @@ func UDPConnectivityListen(endpoint string) error {
692692
return nil
693693
}
694694

695-
func GetDefaultListenAddr() string {
695+
func GetDefaultListenAddr() []string {
696696
if os.Getenv("ENABLE_BIND_LOCAL_IP") == "true" {
697-
if ips := strings.Split(os.Getenv("POD_IPS"), ","); len(ips) == 1 {
698-
return ips[0]
697+
if podIPs := os.Getenv("POD_IPS"); podIPs != "" {
698+
return strings.Split(podIPs, ",")
699699
}
700+
klog.Error("environment variable POD_IPS is not set, cannot bind to local ip")
701+
klog.FlushAndExit(klog.ExitFlushTimeout, 1)
700702
}
701-
return "0.0.0.0"
703+
return []string{"0.0.0.0"}
702704
}
703705

704706
func ContainsUppercase(s string) bool {

‎pkg/util/net_test.go‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1390,14 +1390,21 @@ func TestUDPConnectivityCheck(t *testing.T) {
13901390
}
13911391

13921392
func TestGetDefaultListenAddr(t *testing.T) {
1393-
addr := GetDefaultListenAddr()
1394-
require.Equal(t, addr, "0.0.0.0")
1393+
require.Equal(t, GetDefaultListenAddr(), []string{"0.0.0.0"})
13951394
err := os.Setenv("ENABLE_BIND_LOCAL_IP", "true")
13961395
require.NoError(t, err)
13971396
err = os.Setenv("POD_IPS", "10.10.10.10")
13981397
require.NoError(t, err)
1399-
addr = GetDefaultListenAddr()
1400-
require.Equal(t, addr, "10.10.10.10")
1398+
require.Equal(t, GetDefaultListenAddr(), []string{"10.10.10.10"})
1399+
err = os.Setenv("POD_IPS", "fd00::1")
1400+
require.NoError(t, err)
1401+
require.Equal(t, GetDefaultListenAddr(), []string{"fd00::1"})
1402+
err = os.Setenv("POD_IPS", "10.10.10.10,fd00::1")
1403+
require.NoError(t, err)
1404+
require.Equal(t, GetDefaultListenAddr(), []string{"10.10.10.10", "fd00::1"})
1405+
err = os.Setenv("ENABLE_BIND_LOCAL_IP", "false")
1406+
require.NoError(t, err)
1407+
require.Equal(t, GetDefaultListenAddr(), []string{"0.0.0.0"})
14011408
}
14021409

14031410
func TestContainsUppercase(t *testing.T) {

‎test/e2e/kube-ovn/pod/vpc_pod_probe.go‎

Lines changed: 10 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,15 @@
11
package pod
22

33
import (
4+
"context"
45
"fmt"
56
"math/rand/v2"
67
"net"
78
"strconv"
89
"strings"
910

1011
corev1 "k8s.io/api/core/v1"
12+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
1113
"k8s.io/apimachinery/pkg/util/intstr"
1214

1315
"github.com/onsi/ginkgo/v2"
@@ -185,9 +187,12 @@ func checkTProxyRules(f *framework.Framework, pod *corev1.Pod, probePort int32,
185187
ginkgo.GinkgoHelper()
186188

187189
nodeName := pod.Spec.NodeName
188-
isZeroIP := false
189-
if len(pod.Status.PodIPs) == 2 {
190-
isZeroIP = true
190+
node, err := f.ClientSet.CoreV1().Nodes().Get(context.Background(), nodeName, metav1.GetOptions{})
191+
framework.ExpectNoError(err)
192+
193+
nodeIPv4, nodeIPv6 := util.GetNodeInternalIP(*node)
194+
if len(pod.Status.PodIPs) == 2 && f.VersionPriorTo(1, 13) {
195+
nodeIPv4, nodeIPv6 = net.IPv4zero.String(), net.IPv6zero.String()
191196
}
192197

193198
for _, podIP := range pod.Status.PodIPs {
@@ -196,12 +201,8 @@ func checkTProxyRules(f *framework.Framework, pod *corev1.Pod, probePort int32,
196201
fmt.Sprintf(`-A OVN-OUTPUT -d %s/32 -p tcp -m tcp --dport %d -j MARK --set-xmark %s`, podIP.IP, probePort, tProxyOutputMarkMask),
197202
}
198203
iptables.CheckIptablesRulesOnNode(f, nodeName, util.Mangle, util.OvnOutput, apiv1.ProtocolIPv4, expectedRules, exist)
199-
hostIP := pod.Status.HostIP
200-
if isZeroIP {
201-
hostIP = net.IPv4zero.String()
202-
}
203204
expectedRules = []string{
204-
fmt.Sprintf(`-A OVN-PREROUTING -d %s/32 -p tcp -m tcp --dport %d -j TPROXY --on-port %d --on-ip %s --tproxy-mark %s`, podIP.IP, probePort, util.TProxyListenPort, hostIP, tProxyPreRoutingMarkMask),
205+
fmt.Sprintf(`-A OVN-PREROUTING -d %s/32 -p tcp -m tcp --dport %d -j TPROXY --on-port %d --on-ip %s --tproxy-mark %s`, podIP.IP, probePort, util.TProxyListenPort, nodeIPv4, tProxyPreRoutingMarkMask),
205206
}
206207
iptables.CheckIptablesRulesOnNode(f, nodeName, util.Mangle, util.OvnPrerouting, apiv1.ProtocolIPv4, expectedRules, exist)
207208
} else if util.CheckProtocol(podIP.IP) == apiv1.ProtocolIPv6 {
@@ -210,12 +211,8 @@ func checkTProxyRules(f *framework.Framework, pod *corev1.Pod, probePort int32,
210211
}
211212
iptables.CheckIptablesRulesOnNode(f, nodeName, util.Mangle, util.OvnOutput, apiv1.ProtocolIPv6, expectedRules, exist)
212213

213-
hostIP := pod.Status.HostIP
214-
if isZeroIP {
215-
hostIP = "::"
216-
}
217214
expectedRules = []string{
218-
fmt.Sprintf(`-A OVN-PREROUTING -d %s/128 -p tcp -m tcp --dport %d -j TPROXY --on-port %d --on-ip %s --tproxy-mark %s`, podIP.IP, probePort, util.TProxyListenPort, hostIP, tProxyPreRoutingMarkMask),
215+
fmt.Sprintf(`-A OVN-PREROUTING -d %s/128 -p tcp -m tcp --dport %d -j TPROXY --on-port %d --on-ip %s --tproxy-mark %s`, podIP.IP, probePort, util.TProxyListenPort, nodeIPv6, tProxyPreRoutingMarkMask),
219216
}
220217
iptables.CheckIptablesRulesOnNode(f, nodeName, util.Mangle, util.OvnPrerouting, apiv1.ProtocolIPv6, expectedRules, exist)
221218
}

0 commit comments

Comments
 (0)