enhance: add FORWARD rules for IPVS FullNAT mode, improve docs and tests, bump version to 0.3.3

Author: gjmzj

cmd/ezlb/main.go

@@ -16,7 +16,7 @@ import (
 var (
 	BuildTime   string
 	BuildCommit string
-	Version     = "0.3.2"
+	Version     = "0.3.3"
 	configPath  string
 	showVersion bool
 )

docs/blog/01.IPVS服务FullNAT下网络报文流转与排查.md

@@ -0,0 +1,21 @@
+# FullNAT 模式下报文流转与排查
+
+## 问题现象
+现在客户端请求 ezlb(IPVS)服务端口失败，tcpdump 抓包发现如下：
+1. 客户端 tcp sync ---> ezlb 服务器，正常
+2. ezlb 服务器对fullNAT后 tcp sync ---> 后端服务器，正常
+3. 后端服务器收到 tcp sync ，返回 sync ack 正常
+4. ezlb 服务器 收到 sync ack 后没有反应了，好像丢弃了
+
+## 根因分析
+IPVS 的工作方式比较特殊：IPVS 工作在 INPUT 链上，去程的包被 IPVS 直接在 INPUT 链上截获并做 NAT 转发，不走 FORWARD 链。但回程的 SYN-ACK 到达 ezlb 后，conntrack 做完 un-SNAT，此时 dst 变成了客户端 IP，这个包需要被转发出去 → 走 FORWARD 链 → 被 DROP！
+
+### 去程
+客户端 → ezlb（INPUT 链，IPVS 在 INPUT 链上做 DNAT）→ IPVS DNAT → POSTROUTING（SNAT）→ 后端
+
+### 返程
+后端 → ezlb（PREROUTING，conntrack un-SNAT）→ FORWARD 链 → POSTROUTING（IPVS un-DNAT）→ 客户端
+
+## 解决方案
+1. iptables -P FORWARD ACCEPT
+2. 新版本ezlb 代码中添加 EZLB-FORWARD iptables 链放行ipvs FullNAT 模式下的返程流量。

docs/deploy/ezlb.service

@@ -5,12 +5,14 @@ After=network.target
 
 [Service]
 Type=simple
-ExecStartPre=/sbin/sysctl -w net.ipv4.ip_forward=1
-ExecStartPre=/sbin/sysctl -w net.ipv4.vs.conntrack=1
 ExecStartPre=/sbin/modprobe ip_vs
 ExecStartPre=/sbin/modprobe ip_vs_rr
 ExecStartPre=/sbin/modprobe ip_vs_wrr
 ExecStartPre=/sbin/modprobe ip_vs_lc
+ExecStartPre=/sbin/sysctl -w net.ipv4.ip_forward=1
+ExecStartPre=/sbin/sysctl -w net.ipv4.vs.conntrack=1
+ExecStartPre=/sbin/sysctl -w net.ipv4.conf.all.rp_filter=0
+ExecStartPre=/sbin/sysctl -w net.ipv4.conf.default.rp_filter=0
 ExecStart=/usr/local/bin/ezlb start -c /etc/ezlb/ezlb.yaml
 ExecReload=/bin/kill -HUP $MAINPID
 Restart=on-failure

docs/load/bench.sh

@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+# ezlb_bench.sh 使用 wrk 对 ezlb 转发的 (VIP:PORT) 做多档位压测，并采集 CPU 与连接状态
+# 使用方式：
+#  1. 在客户端压力机上安装 wrk，并确保能免密 SSH 访问负载均衡器节点（用于抓取 CPU、软中断、socket 状态）。
+#  2. 将 VIP_HOST/VIP_PORT 修改为 ezlb 对外监听的地址端口；如需压测 TCP 其他协议，可换用 wrk --script 或自定义 TCP 脚本。
+#  3. CONCURRENCY_LEVELS 可根据目标并发量增减；DURATION 设置为每档位压测时间（建议 ≥60s）。
+#  4. 运行脚本：chmod +x ezlb_bench.sh && ./ezlb_bench.sh。
+#  5. 结果会生成在 logs/ 目录：wrk_*.log 记录吞吐、延迟；lb_stats_*.log 记录压测期间负载均衡器CPU与连接概况，方便对照分析。
+#
+#  如果需要 UDP 或更底层 TCP 测试，可将 run_wrk 换成 iperf3 -u/netperf 等命令，其他结构保持不变。
+
+VIP_HOST="10.0.0.1"
+VIP_PORT="80"
+DURATION="60s"          # 每档位压测时长
+THREADS=4               # wrk 工作线程数
+CONCURRENCY_LEVELS=("512" "1024" "4096" "8192")
+
+# 被测负载均衡器节点，用于采集状态
+LB_HOST="10.0.0.10"
+
+run_wrk() {
+  local c=$1
+  echo "=== 并发 $c ==="
+  wrk \
+    -t "${THREADS}" \
+    -c "${c}" \
+    -d "${DURATION}" \
+    --latency \
+    "http://${VIP_HOST}:${VIP_PORT}/healthz" \
+    | tee "logs/wrk_${c}.log"
+}
+
+collect_lb_stats() {
+  ssh "${LB_HOST}" "date '+%F %T'; \
+    mpstat 1 5 | tail -n +4; \
+    echo '--- softirqs ---'; \
+    cat /proc/softirqs | head -n 20; \
+    echo '--- ss summary ---'; \
+    ss -s" \
+    > "logs/lb_stats_$(date +%s).log"
+}
+
+mkdir -p logs
+
+for concurrency in "${CONCURRENCY_LEVELS[@]}"; do
+  collect_lb_stats &
+  RUN_ID=$!
+  run_wrk "${concurrency}"
+  wait "${RUN_ID}"
+  echo ""
+done

pkg/lvs/reconciler.go

@@ -172,18 +172,21 @@ func (r *Reconciler) Cleanup() error {
 	return nil
 }
 
-// reconcileSNAT builds the desired SNAT rules from configs with full_nat enabled
-// and delegates to the SNAT manager for declarative reconciliation.
+// reconcileSNAT builds the desired SNAT and FORWARD rules from configs with
+// full_nat enabled and delegates to the SNAT manager for declarative reconciliation.
+// FORWARD rules are needed because IPVS NAT mode requires packets to traverse
+// the FORWARD chain, which may have a DROP policy (e.g. Docker environments).
 func (r *Reconciler) reconcileSNAT(configs []config.ServiceConfig) error {
-	var desiredRules []snat.SNATRule
+	var desiredSNATRules []snat.SNATRule
+	var desiredForwardRules []snat.ForwardRule
 
 	for _, svcCfg := range configs {
 		if !svcCfg.FullNAT {
 			continue
 		}
 
 		for _, backendCfg := range svcCfg.Backends {
-			// Only create SNAT rules for healthy backends
+			// Only create rules for healthy backends
 			if svcCfg.HealthCheck.IsEnabled() && !r.healthMgr.IsHealthy(backendCfg.Address) {
 				continue
 			}
@@ -202,16 +205,30 @@ func (r *Reconciler) reconcileSNAT(configs []config.ServiceConfig) error {
 				protocol = "tcp"
 			}
 
-			desiredRules = append(desiredRules, snat.SNATRule{
+			desiredSNATRules = append(desiredSNATRules, snat.SNATRule{
 				BackendIP:   backendHost,
 				BackendPort: uint16(backendPort),
 				Protocol:    protocol,
 				SnatIP:      svcCfg.SnatIP,
 			})
+
+			desiredForwardRules = append(desiredForwardRules, snat.ForwardRule{
+				BackendIP:   backendHost,
+				BackendPort: uint16(backendPort),
+				Protocol:    protocol,
+			})
 		}
 	}
 
-	return r.snatMgr.Reconcile(desiredRules)
+	if err := r.snatMgr.Reconcile(desiredSNATRules); err != nil {
+		return fmt.Errorf("snat rules: %w", err)
+	}
+
+	if err := r.snatMgr.ReconcileForward(desiredForwardRules); err != nil {
+		return fmt.Errorf("forward rules: %w", err)
+	}
+
+	return nil
 }
 
 // buildDesiredState converts config services into the desired IPVS state,

pkg/lvs/reconciler_snat_test.go

@@ -50,6 +50,20 @@ func TestReconcile_FullNATGeneratesSNATRules(t *testing.T) {
 	if len(managed) != 2 {
 		t.Fatalf("expected 2 SNAT rules, got %d", len(managed))
 	}
+
+	// Verify FORWARD rules were created via fake manager
+	managedForward := fakeSnatMgr.GetManagedForward()
+	if len(managedForward) != 2 {
+		t.Fatalf("expected 2 FORWARD rules, got %d", len(managedForward))
+	}
+	forwardKey1 := "192.168.1.1:53/udp"
+	if _, exists := managedForward[forwardKey1]; !exists {
+		t.Errorf("expected FORWARD rule %q to exist", forwardKey1)
+	}
+	forwardKey2 := "192.168.1.2:53/udp"
+	if _, exists := managedForward[forwardKey2]; !exists {
+		t.Errorf("expected FORWARD rule %q to exist", forwardKey2)
+	}
 }
 
 func TestReconcile_FullNATDisabledSkipsSNAT(t *testing.T) {
@@ -82,4 +96,10 @@ func TestReconcile_FullNATDisabledSkipsSNAT(t *testing.T) {
 	if len(managed) != 0 {
 		t.Fatalf("expected 0 SNAT rules when full_nat is disabled, got %d", len(managed))
 	}
+
+	// Verify no FORWARD rules were created
+	managedForward := fakeSnatMgr.GetManagedForward()
+	if len(managedForward) != 0 {
+		t.Fatalf("expected 0 FORWARD rules when full_nat is disabled, got %d", len(managedForward))
+	}
 }

pkg/snat/manager_fake.go

@@ -8,19 +8,21 @@ import (
 	"go.uber.org/zap"
 )
 
-// FakeManager provides an in-memory SNAT rule manager for non-Linux systems.
+// FakeManager provides an in-memory SNAT and FORWARD rule manager for non-Linux systems.
 // It simulates iptables behavior for development and testing on macOS.
 type FakeManager struct {
-	managed map[string]SNATRule
-	mu      sync.Mutex
-	logger  *zap.Logger
+	managed        map[string]SNATRule
+	managedForward map[string]ForwardRule
+	mu             sync.Mutex
+	logger         *zap.Logger
 }
 
 // NewManager creates a fake in-memory SNAT Manager for non-Linux systems.
 func NewManager(logger *zap.Logger) (Manager, error) {
 	return &FakeManager{
-		managed: make(map[string]SNATRule),
-		logger:  logger,
+		managed:        make(map[string]SNATRule),
+		managedForward: make(map[string]ForwardRule),
+		logger:         logger,
 	}, nil
 }
 
@@ -55,17 +57,48 @@ func (m *FakeManager) Reconcile(desired []SNATRule) error {
 	return nil
 }
 
-// Cleanup removes all managed SNAT rules from memory.
+// ReconcileForward compares desired FORWARD rules with the currently managed set in memory.
+func (m *FakeManager) ReconcileForward(desired []ForwardRule) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	desiredMap := make(map[string]ForwardRule, len(desired))
+	for _, rule := range desired {
+		desiredMap[rule.Key()] = rule
+	}
+
+	// Remove stale rules
+	for key := range m.managedForward {
+		if _, exists := desiredMap[key]; !exists {
+			delete(m.managedForward, key)
+			m.logger.Debug("fake: deleted FORWARD rule", zap.String("key", key))
+		}
+	}
+
+	// Add missing rules
+	for key, rule := range desiredMap {
+		if _, exists := m.managedForward[key]; exists {
+			continue
+		}
+		m.managedForward[key] = rule
+		m.logger.Debug("fake: added FORWARD rule", zap.String("key", key))
+	}
+
+	return nil
+}
+
+// Cleanup removes all managed SNAT and FORWARD rules from memory.
 func (m *FakeManager) Cleanup() error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
 	m.managed = make(map[string]SNATRule)
-	m.logger.Debug("fake: cleaned up all SNAT rules")
+	m.managedForward = make(map[string]ForwardRule)
+	m.logger.Debug("fake: cleaned up all SNAT and FORWARD rules")
 	return nil
 }
 
-// GetManaged returns a copy of the currently managed rules (for testing).
+// GetManaged returns a copy of the currently managed SNAT rules (for testing).
 func (m *FakeManager) GetManaged() map[string]SNATRule {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -76,3 +109,15 @@ func (m *FakeManager) GetManaged() map[string]SNATRule {
 	}
 	return result
 }
+
+// GetManagedForward returns a copy of the currently managed FORWARD rules (for testing).
+func (m *FakeManager) GetManagedForward() map[string]ForwardRule {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	result := make(map[string]ForwardRule, len(m.managedForward))
+	for k, v := range m.managedForward {
+		result[k] = v
+	}
+	return result
+}