Skip to content

Commit c248c0d

Browse files
zhangzujianzhangzujian
committed
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 217788c commit c248c0d

File tree

8 files changed

+101
-80
lines changed

8 files changed

+101
-80
lines changed

‎cmd/controller/controller.go

+12-7
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

+26-20
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

+10-6
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

+3
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

+7-16
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 {
@@ -830,7 +827,7 @@ func (c *Controller) setIptables() error {
830827
return err
831828
}
832829

833-
if err = c.reconcileTProxyIPTableRules(protocol, isDual); err != nil {
830+
if err = c.reconcileTProxyIPTableRules(protocol); err != nil {
834831
klog.Error(err)
835832
return err
836833
}
@@ -861,7 +858,7 @@ func (c *Controller) setIptables() error {
861858
return nil
862859
}
863860

864-
func (c *Controller) reconcileTProxyIPTableRules(protocol string, isDual bool) error {
861+
func (c *Controller) reconcileTProxyIPTableRules(protocol string) error {
865862
if !c.config.EnableTProxy {
866863
return nil
867864
}
@@ -895,19 +892,13 @@ func (c *Controller) reconcileTProxyIPTableRules(protocol string, isDual bool) e
895892
}
896893

897894
for _, probePort := range ports.SortedList() {
898-
hostIP := pod.Status.HostIP
895+
hostIP := c.config.NodeIPv4
899896
prefixLen := 32
900897
if protocol == kubeovnv1.ProtocolIPv6 {
901898
prefixLen = 128
899+
hostIP = c.config.NodeIPv6
902900
}
903901

904-
if isDual || os.Getenv("ENABLE_BIND_LOCAL_IP") == "false" {
905-
if protocol == kubeovnv1.ProtocolIPv4 {
906-
hostIP = "0.0.0.0"
907-
} else if protocol == kubeovnv1.ProtocolIPv6 {
908-
hostIP = "::"
909-
}
910-
}
911902
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))})
912903
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))})
913904
}

‎pkg/daemon/tproxy_linux.go

+26-23
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

+6-4
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

+11-4
Original file line numberDiff line numberDiff line change
@@ -1389,14 +1389,21 @@ func TestUDPConnectivityCheck(t *testing.T) {
13891389
}
13901390

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

14021409
func TestContainsUppercase(t *testing.T) {

0 commit comments

Comments
 (0)