perf(pool): parallelize network slot cleanup on shutdown-Conch-AtomGit

已合并

perf(pool): parallelize network slot cleanup on shutdown #171

jing-rui创建于 8月18日从jing-rui/Conch： dev 合入到 openeuler/Conch： dev

perf(pool): parallelize network slot cleanup on shutdown #171

已合并

jing-rui创建于 8月18日

从jing-rui/Conch dev合入到openeuler/Conch dev

共 4 个文件变更+70-35

MMakefile+7-20

	@@ -84,27 +84,13 @@ gen-proto-py: ## Generate Python protobuf and Connect client code
84	@python3 -c "import pathlib; pathlib.Path('$(CONCH_PY_PROTO_DIR)/__init__.py').write_text('# Auto-generated by Makefile gen-proto-py\nfrom . import agent_pb2, agent_connect\n')"	84	@python3 -c "import pathlib; pathlib.Path('$(CONCH_PY_PROTO_DIR)/__init__.py').write_text('# Auto-generated by Makefile gen-proto-py\nfrom . import agent_pb2, agent_connect\n')"
85	@echo "Python proto code generated to $(CONCH_PY_PROTO_DIR)"	85	@echo "Python proto code generated to $(CONCH_PY_PROTO_DIR)"
86		86
87	-build: ## Build all Conch binaries	87	+build:
88	- @mkdir -p $(BIN_DIR)	88	+ mkdir -p $(BIN_DIR)
89	- @set -e; for cmd in $(CMDS); do \	89	+ CGO_ENABLED=0 GOOS=linux GOARCH=$$(go env GOARCH) \
90	- echo "building $$cmd..."; \	90	+ $(GOBUILD) -mod=readonly -tags "netgo,osusergo" \
91	- $(GOBUILD) -ldflags "$(VERSION_LDFLAGS)" -o "$(BIN_DIR)/$$cmd" "./cmd/$$cmd"; \	91	+ -ldflags '-s -w -extldflags "-static" $(VERSION_LDFLAGS)' \
92	- done	92	+ -o $(BIN_DIR)/ ./cmd/...
93		93
94	-build-offline:
95	- @echo "syncing dependency manifests from image backup..."
96	- @cp /go/go.mod.backup ./go.mod
97	- @cp /go/go.sum.backup ./go.sum
98	- @echo "building binaries using image cache..."
99	- @mkdir -p $(BIN_DIR)
100	- @for cmd in $(CMDS); do \
101	- echo "building static cmd/$$cmd..."; \
102	- CGO_ENABLED=0 GOOS=linux GOARCH=$$(go env GOARCH) \
103	- $(GOBUILD) -mod=readonly -tags "netgo,osusergo" \
104	- -ldflags '-s -w -extldflags "-static" $(VERSION_LDFLAGS)' \
105	- -o $(BIN_DIR)/$$cmd ./cmd/$$cmd; \
106	- done
107	- @git checkout go.mod go.sum 2>/dev/null \|\| true
108		94
109	build-%: ## Build specific binary (e.g., make build-conchd)	95	build-%: ## Build specific binary (e.g., make build-conchd)
110	@echo "building cmd/$*..."	96	@echo "building cmd/$*..."
	@@ -166,3 +152,4 @@ cleancode: ## Clean code (remove trailing spaces/CR)
166	fi	152	fi
167	@$(CLEANSCRIPT) > /dev/null 2>&1	153	@$(CLEANSCRIPT) > /dev/null 2>&1
168	@echo "code cleaning completed"	154	@echo "code cleaning completed"
		155	+

Minternal/netstack/guest_tap.go+14-2

	@@ -23,6 +23,7 @@ import (
23	"fmt"	23	"fmt"
24	"net"	24	"net"
25	"os"	25	"os"
		26	+ "strings"
26		27
27	"github.com/coreos/go-iptables/iptables"	28	"github.com/coreos/go-iptables/iptables"
28	"github.com/vishvananda/netlink"	29	"github.com/vishvananda/netlink"
	@@ -79,8 +80,8 @@ func configureGuestTapNetwork(slot *Slot, cniIP string) error {
79	return fmt.Errorf("error setting lo device up: %w", err)	80	return fmt.Errorf("error setting lo device up: %w", err)
80	}	81	}
81		82
82	- if err := os.WriteFile(ipv4ForwardingSysctlPath, []byte("1\n"), 0o644); err != nil {	83	+ if err := ensureIPv4Forwarding(); err != nil {
83	- return fmt.Errorf("error enabling ipv4 forwarding via %s: %w", ipv4ForwardingSysctlPath, err)	84	+ return err
84	}	85	}
85		86
86	tables, err := iptables.New()	87	tables, err := iptables.New()
	@@ -97,6 +98,17 @@ func configureGuestTapNetwork(slot *Slot, cniIP string) error {
97	return nil	98	return nil
98	}	99	}
99		100
		101	+func ensureIPv4Forwarding() error {
		102	+ data, err := os.ReadFile(ipv4ForwardingSysctlPath)
		103	+ if err == nil && strings.TrimSpace(string(data)) == "1" {
		104	+ return nil
		105	+ }
		106	+ if err := os.WriteFile(ipv4ForwardingSysctlPath, []byte("1\n"), 0o644); err != nil {
		107	+ return fmt.Errorf("enable IPv4 forwarding via %s: %w", ipv4ForwardingSysctlPath, err)
		108	+ }
		109	+ return nil
		110	+}
		111	+
100	func removeGuestTapNetwork(slot *Slot, cniIP string) error {	112	func removeGuestTapNetwork(slot *Slot, cniIP string) error {
101	var errs []error	113	var errs []error
102		114

Minternal/netstack/netns.go+6-6

	@@ -27,9 +27,11 @@ import (
27	"github.com/vishvananda/netlink"	27	"github.com/vishvananda/netlink"
28	"github.com/vishvananda/netns"	28	"github.com/vishvananda/netns"
29	"golang.org/x/sys/unix"	29	"golang.org/x/sys/unix"
		30	+
		31	+ "github.com/openeuler/Conch/pkg/ulog"
30	)	32	)
31		33
32	-func configureIPv4OnlyCurrentNetworkNamespace() error {	34	+func tryDisableIPv6() {
33	for _, setting := range [][2]string{	35	for _, setting := range [][2]string{
34	{"/proc/sys/net/ipv6/conf/all/accept_ra", "0\n"},	36	{"/proc/sys/net/ipv6/conf/all/accept_ra", "0\n"},
35	{"/proc/sys/net/ipv6/conf/all/autoconf", "0\n"},	37	{"/proc/sys/net/ipv6/conf/all/autoconf", "0\n"},
	@@ -39,10 +41,10 @@ func configureIPv4OnlyCurrentNetworkNamespace() error {
39	{"/proc/sys/net/ipv6/conf/default/disable_ipv6", "1\n"},	41	{"/proc/sys/net/ipv6/conf/default/disable_ipv6", "1\n"},
40	} {	42	} {
41	if err := os.WriteFile(setting[0], []byte(setting[1]), 0o644); err != nil {	43	if err := os.WriteFile(setting[0], []byte(setting[1]), 0o644); err != nil {
42	- return fmt.Errorf("set IPv4-only sysctl %s: %w", setting[0], err)	44	+ ulog.GetLogger().Warn("could not configure IPv4-only network namespace",
		45	+ ulog.F("sysctl", setting[0]), ulog.F("error", err))
43	}	46	}
44	}	47	}
45	- return nil
46	}	48	}
47		49
48	func createNetworkNamespace(slot *Slot) (retErr error) {	50	func createNetworkNamespace(slot *Slot) (retErr error) {
	@@ -98,9 +100,7 @@ func createNetworkNamespace(slot *Slot) (retErr error) {
98	return fmt.Errorf("bind mount network namespace at %s: %w", netnsPath, err)	100	return fmt.Errorf("bind mount network namespace at %s: %w", netnsPath, err)
99	}	101	}
100	mounted = true	102	mounted = true
101	- if err := configureIPv4OnlyCurrentNetworkNamespace(); err != nil {	103	+ tryDisableIPv6()
102	- return fmt.Errorf("configure IPv4-only network namespace: %w", err)
103	- }
104		104
105	return nil	105	return nil
106	}	106	}

Minternal/netstack/pool.go+43-7

	@@ -41,6 +41,11 @@ const (
41	prefillCreateAttempts = 2	41	prefillCreateAttempts = 2
42	populateRetryMinDelay = time.Second	42	populateRetryMinDelay = time.Second
43	populateRetryMaxDelay = 30 * time.Second	43	populateRetryMaxDelay = 30 * time.Second
		44	+ // cleanupWorkers caps concurrent teardown of buffered network slots
		45	+ // during shutdown. CNI/netns removal is I/O-bound, so a fixed cap
		46	+ // independent of the host CPU count keeps the host network stack from
		47	+ // being overwhelmed on pools with many slots.
		48	+ cleanupWorkers = 16
44	)	49	)
45		50
46	var (	51	var (
	@@ -209,6 +214,7 @@ func (p *Pool) Start(ctx context.Context) error {
209	// Close stops the population loop, drains and closes the warm queue, and makes	214	// Close stops the population loop, drains and closes the warm queue, and makes
210	// a best-effort attempt to tear down every buffered slot.	215	// a best-effort attempt to tear down every buffered slot.
211	func (p *Pool) Close() {	216	func (p *Pool) Close() {
		217	+ defer ulog.TraceCost(ulog.TraceStart(), "", "netstack.Pool.Close()")
212	if p == nil {	218	if p == nil {
213	return	219	return
214	}	220	}
	@@ -217,16 +223,45 @@ func (p *Pool) Close() {
217	<-p.populateDone	223	<-p.populateDone
218	}	224	}
219	if p.warmSlots != nil {	225	if p.warmSlots != nil {
		226	+ // Detach all buffered slots before starting cleanup so the queue is no
		227	+ // longer usable while teardown runs concurrently.
		228	+ var slots []*Slot
220	for {	229	for {
221	slot, err := p.warmSlots.Pop()	230	slot, err := p.warmSlots.Pop()
222	if err != nil {	231	if err != nil {
223	break	232	break
224	}	233	}
225	- if err := p.destroyNetworkSlot(context.Background(), slot); err != nil {	234	+ slots = append(slots, slot)
226	- ulog.GetLogger().Warn("failed to clean up warm network slot during shutdown", ulog.F("slot_id", slot.ID()), ulog.F("error", err))
227	- }
228	}	235	}
229	p.warmSlots.Close()	236	p.warmSlots.Close()
		237	+
		238	+ if len(slots) > 0 {
		239	+ workers := cleanupWorkers
		240	+ if len(slots) < workers {
		241	+ workers = len(slots)
		242	+ }
		243	+ // Semaphore-bounded goroutines: slots are already materialized in
		244	+ // a slice, so a channel-based job queue would just re-buffer them.
		245	+ // The semaphore caps concurrency without that extra hop.
		246	+ sem := make(chan struct{}, workers)
		247	+ var wg sync.WaitGroup
		248	+ for _, slot := range slots {
		249	+ wg.Add(1)
		250	+ sem <- struct{}{}
		251	+ go func(s *Slot) {
		252	+ defer wg.Done()
		253	+ defer func() { <-sem }()
		254	+ if err := p.destroyNetworkSlot(context.Background(), s); err != nil {
		255	+ ulog.GetLogger().Warn(
		256	+ "failed to clean up warm network slot during shutdown",
		257	+ ulog.F("slot_id", s.ID()),
		258	+ ulog.F("error", err),
		259	+ )
		260	+ }
		261	+ }(slot)
		262	+ }
		263	+ wg.Wait()
		264	+ }
230	}	265	}
231	if p.cniManager != nil {	266	if p.cniManager != nil {
232	if err := removeHostForwardingRules(p.cniManager.bridgeName, p.hostInterface); err != nil {	267	if err := removeHostForwardingRules(p.cniManager.bridgeName, p.hostInterface); err != nil {
	@@ -582,18 +617,19 @@ func (p Pool) provisionSlotNetwork(ctx context.Context, slot Slot) error {
582	return err	617	return err
583	}	618	}
584	if len(addresses) != 0 {	619	if len(addresses) != 0 {
585	- return fmt.Errorf("network namespace has IPv6 address %s", addresses[0].String())	620	+ ulog.GetLogger().Warn("network namespace has IPv6 address", ulog.F("address", addresses[0].String()))
586	}	621	}
587	routes, err := netlink.RouteListFiltered(netlink.FAMILY_V6, &netlink.Route{Table: unix.RT_TABLE_UNSPEC}, netlink.RT_FILTER_TABLE)	622	routes, err := netlink.RouteListFiltered(netlink.FAMILY_V6, &netlink.Route{Table: unix.RT_TABLE_UNSPEC}, netlink.RT_FILTER_TABLE)
588	if err != nil {	623	if err != nil {
589	return err	624	return err
590	}	625	}
591	if len(routes) != 0 {	626	if len(routes) != 0 {
592	- return fmt.Errorf("network namespace has IPv6 route %+v", routes[0])	627	+ ulog.GetLogger().Warn("network namespace has IPv6 route", ulog.F("route", routes[0]))
593	}	628	}
594	- return configureIPv4OnlyCurrentNetworkNamespace()	629	+ tryDisableIPv6()
		630	+ return nil
595	}); err != nil {	631	}); err != nil {
596	- return fmt.Errorf("enforce IPv4-only network namespace: %w", err)	632	+ return fmt.Errorf("check network namespace: %w", err)
597	}	633	}
598		634
599	return nil	635	return nil