perf(net): reduce status transport and controller memory overhead

cmd/unbounded-net-controller/cluster_status.go

@@ -1061,15 +1061,9 @@ func buildConnectivityMatrix(nodes []*NodeStatusResponse, gatewayPools []Gateway
 
 		siteNodes[site][name] = true
 
-		var allPeers []WireGuardPeerStatus
-
-		for _, p := range n.Peers {
-			if p.PeerType == "site" || p.PeerType == "gateway" {
-				allPeers = append(allPeers, p)
-			}
-		}
-
-		nodePeers[name] = allPeers
+		// Keep the immutable snapshot's slice; filter when reading rather
+		// than copying every peer, including for scopes above the size limit.
+		nodePeers[name] = n.Peers
 
 		for _, p := range n.Peers {
 			if p.PeerType == "gateway" && p.Name != "" && p.SiteName == site {
@@ -1111,6 +1105,10 @@ func buildConnectivityMatrix(nodes []*NodeStatusResponse, gatewayPools []Gateway
 			}
 
 			for _, peer := range nodePeers[srcNode] {
+				if !isConnectivityMatrixPeer(peer) {
+					continue
+				}
+
 				tgtNode := peer.Name
 				if tgtNode == "" || tgtNode == srcNode || !nodeSet[tgtNode] {
 					continue
@@ -1153,6 +1151,10 @@ func buildConnectivityMatrix(nodes []*NodeStatusResponse, gatewayPools []Gateway
 
 			poolNodeSet[name] = true
 			for _, peer := range nodePeers[name] {
+				if !isConnectivityMatrixPeer(peer) {
+					continue
+				}
+
 				peerName := strings.TrimSpace(peer.Name)
 				if peerName == "" {
 					continue
@@ -1165,6 +1167,10 @@ func buildConnectivityMatrix(nodes []*NodeStatusResponse, gatewayPools []Gateway
 
 			for srcNodeName, peers := range nodePeers {
 				for _, peer := range peers {
+					if !isConnectivityMatrixPeer(peer) {
+						continue
+					}
+
 					if strings.TrimSpace(peer.Name) == name {
 						if _, ok := nodeByName[srcNodeName]; ok {
 							poolNodeSet[srcNodeName] = true
@@ -1188,3 +1194,7 @@ func buildConnectivityMatrix(nodes []*NodeStatusResponse, gatewayPools []Gateway
 
 	return result
 }
+
+func isConnectivityMatrixPeer(peer WireGuardPeerStatus) bool {
+	return peer.PeerType == "site" || peer.PeerType == "gateway"
+}

cmd/unbounded-net-controller/cni_status_test.go

@@ -121,7 +121,7 @@ func TestCNIGuardProtoStatusRecovery(t *testing.T) {
 					if source == "ws" || source == "apiserver-ws" {
 						var messageType string
 
-						messageType, ack = handleProtoWSMessage(health, data, source)
+						messageType, ack = handleProtoWSBytes(health, data, source)
 						if messageType != "node_status_ack" {
 							t.Fatalf("unexpected message type %q: %+v", messageType, ack)
 						}

cmd/unbounded-net-controller/gzip_writer_pool.go

@@ -0,0 +1,59 @@
+// Copyright (c) Microsoft Corporation.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"compress/gzip"
+	"io"
+)
+
+const gzipIdleWriterLimit = 4
+
+type gzipWriterPool struct {
+	idle chan *gzip.Writer
+}
+
+func newGzipWriterPool() *gzipWriterPool {
+	return &gzipWriterPool{idle: make(chan *gzip.Writer, gzipIdleWriterLimit)}
+}
+
+func (p *gzipWriterPool) get(output io.Writer) (*gzip.Writer, error) {
+	var writer *gzip.Writer
+
+	select {
+	case writer = <-p.idle:
+	default:
+		var err error
+
+		writer, err = gzip.NewWriterLevel(io.Discard, gzip.BestSpeed)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	writer.Reset(output)
+
+	return writer, nil
+}
+
+func (p *gzipWriterPool) put(writer *gzip.Writer, completed bool) {
+	closed := false
+
+	defer func() {
+		// Detach even if Close panics, and reset before another request can
+		// acquire the writer. Failed or interrupted responses are not pooled.
+		writer.Reset(io.Discard)
+
+		if !completed || !closed {
+			return
+		}
+
+		select {
+		case p.idle <- writer:
+		default:
+		}
+	}()
+
+	closed = writer.Close() == nil
+}