Support protocol-independent NPL annotation

* Add protocol string to new NPLAnnotation set the protocols map as a deprecated value.
* Support Node port for UDP and TCP using the different number for a single Pod on Windows.

issue antrea-io#3826

Signed-off-by: Shuyang Xin <gavinx@vmware.com>

pkg/agent/nodeportlocal/k8s/annotations.go

@@ -39,7 +39,8 @@ type NPLAnnotation struct {
 	PodPort   int      `json:"podPort"`
 	NodeIP    string   `json:"nodeIP"`
 	NodePort  int      `json:"nodePort"`
-	Protocols []string `json:"protocols"`
+	Protocol  string   `json:"protocol"`
+	Protocols []string `json:"protocols"` // deprecated, array with a single member which is equal to the Protocol field
 }
 
 func toJSON(serialize interface{}) string {

pkg/agent/nodeportlocal/k8s/npl_controller.go

@@ -447,7 +447,7 @@ func (c *NPLController) handleAddUpdatePod(key string, obj interface{}) error {
 		}
 	}
 
-	nplAnnotationsRequiredMap := map[int]NPLAnnotation{}
+	nplAnnotationsRequiredMap := map[string]NPLAnnotation{}
 	nplAnnotationsRequired := []NPLAnnotation{}
 
 	hostPorts := make(map[string]int)
@@ -491,7 +491,7 @@ func (c *NPLController) handleAddUpdatePod(key string, obj interface{}) error {
 			return fmt.Errorf("failed to parse port number and protocol from %s for Pod %s: %v", targetPortProto, key, err)
 		}
 		podPorts[targetPortProto] = struct{}{}
-		portData := c.portTable.GetEntry(podIP, port)
+		portData := c.portTable.GetEntry(podIP, port, protocol)
 		if portData != nil && !portData.ProtocolInUse(protocol) {
 			// If the PortTable has an entry for the Pod but does not have an
 			// entry with protocol, we enforce AddRule for the missing Protocol.
@@ -510,14 +510,12 @@ func (c *NPLController) handleAddUpdatePod(key string, obj interface{}) error {
 			nodePort = portData.NodePort
 		}
 
-		if val, ok := nplAnnotationsRequiredMap[nodePort]; ok {
-			val.Protocols = append(val.Protocols, protocol)
-			nplAnnotationsRequiredMap[nodePort] = val
-		} else {
-			nplAnnotationsRequiredMap[nodePort] = NPLAnnotation{
+		if _, ok := nplAnnotationsRequiredMap[portcache.NodePortProtoFormat(nodePort, protocol)]; !ok {
+			nplAnnotationsRequiredMap[portcache.NodePortProtoFormat(nodePort, protocol)] = NPLAnnotation{
 				PodPort:   port,
 				NodeIP:    pod.Status.HostIP,
 				NodePort:  nodePort,
+				Protocol:  protocol,
 				Protocols: []string{protocol},
 			}
 		}
@@ -604,6 +602,7 @@ func (c *NPLController) waitForRulesInitialization() {
 				NodePort:  npl.NodePort,
 				PodPort:   npl.PodPort,
 				PodIP:     pod.Status.PodIP,
+				Protocol:  npl.Protocol,
 				Protocols: npl.Protocols,
 			})
 		}

pkg/agent/nodeportlocal/npl_agent_test.go

@@ -66,7 +66,7 @@ const (
 
 func newPortTable(mockIPTables rules.PodPortRules, mockPortOpener portcache.LocalPortOpener) *portcache.PortTable {
 	return &portcache.PortTable{
-		NodePortTable:    make(map[int]*portcache.NodePortData),
+		NodePortTable:    make(map[string]*portcache.NodePortData),
 		PodEndpointTable: make(map[string]*portcache.NodePortData),
 		StartPort:        defaultStartPort,
 		EndPort:          defaultEndPort,

pkg/agent/nodeportlocal/portcache/port_table.go

@@ -25,26 +25,10 @@ import (
 	"antrea.io/antrea/pkg/agent/nodeportlocal/rules"
 )
 
-var (
-	supportedProtocols = []string{"tcp", "udp"}
-)
-
 // protocolSocketState represents the state of the socket corresponding to a
 // given (Node port, protocol) tuple.
 type protocolSocketState int
 
-const (
-	// stateOpen means that a listening socket has been opened for the
-	// protocol (as a means to reserve the port for this protocol), but no
-	// NPL rule has been installed for it.
-	stateOpen protocolSocketState = iota
-	// stateInUse means that a listening socket has been opened AND a NPL
-	// rule has been installed.
-	stateInUse
-	// stateClosed means that the socket has been closed.
-	stateClosed
-)
-
 type ProtocolSocketData struct {
 	Protocol string
 	State    protocolSocketState
@@ -83,7 +67,7 @@ type LocalPortOpener interface {
 type localPortOpener struct{}
 
 type PortTable struct {
-	NodePortTable    map[int]*NodePortData
+	NodePortTable    map[string]*NodePortData
 	PodEndpointTable map[string]*NodePortData
 	StartPort        int
 	EndPort          int
@@ -95,7 +79,7 @@ type PortTable struct {
 
 func NewPortTable(start, end int) (*PortTable, error) {
 	ptable := PortTable{
-		NodePortTable:    make(map[int]*NodePortData),
+		NodePortTable:    make(map[string]*NodePortData),
 		PodEndpointTable: make(map[string]*NodePortData),
 		StartPort:        start,
 		EndPort:          end,
@@ -112,21 +96,10 @@ func NewPortTable(start, end int) (*PortTable, error) {
 func (pt *PortTable) CleanupAllEntries() {
 	pt.tableLock.Lock()
 	defer pt.tableLock.Unlock()
-	pt.NodePortTable = make(map[int]*NodePortData)
+	pt.NodePortTable = make(map[string]*NodePortData)
 	pt.PodEndpointTable = make(map[string]*NodePortData)
 }
 
-func (pt *PortTable) GetEntry(ip string, port int) *NodePortData {
-	pt.tableLock.RLock()
-	defer pt.tableLock.RUnlock()
-	// Return pointer to copy of data from the PodEndpointTable.
-	if data := pt.getEntryByPodIPPort(ip, port); data != nil {
-		dataCopy := *data
-		return &dataCopy
-	}
-	return nil
-}
-
 func (pt *PortTable) GetDataForPodIP(ip string) []NodePortData {
 	pt.tableLock.RLock()
 	defer pt.tableLock.RUnlock()
@@ -156,6 +129,11 @@ func (pt *PortTable) RuleExists(podIP string, podPort int, protocol string) bool
 	return false
 }
 
+// nodePortProtoFormat formats the nodeport, protocol to string port:protocol.
+func NodePortProtoFormat(nodeport int, protocol string) string {
+	return fmt.Sprintf("%d:%s", nodeport, protocol)
+}
+
 // podIPPortFormat formats the ip, port to string ip:port.
 func podIPPortFormat(ip string, port int) string {
 	return fmt.Sprintf("%s:%d", ip, port)

pkg/agent/nodeportlocal/portcache/port_table_linux.go

@@ -19,16 +19,46 @@ package portcache
 
 import (
 	"fmt"
+	"strconv"
 	"time"
 
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/agent/nodeportlocal/rules"
 )
 
+const (
+	// stateOpen means that a listening socket has been opened for the
+	// protocol (as a means to reserve the port for this protocol), but no
+	// NPL rule has been installed for it.
+	stateOpen protocolSocketState = iota
+	// stateInUse means that a listening socket has been opened AND a NPL
+	// rule has been installed.
+	stateInUse
+	// stateClosed means that the socket has been closed.
+	stateClosed
+)
+
+var (
+	supportedProtocols = []string{"tcp", "udp"}
+)
+
+func (pt *PortTable) GetEntry(ip string, port int, protocol string) *NodePortData {
+	var _ = protocol
+	pt.tableLock.RLock()
+	defer pt.tableLock.RUnlock()
+	// Return pointer to copy of data from the PodEndpointTable.
+	if data := pt.getEntryByPodIPPort(ip, port); data != nil {
+		dataCopy := *data
+		return &dataCopy
+	}
+	return nil
+}
+
 func openSocketsForPort(localPortOpener LocalPortOpener, port int) ([]ProtocolSocketData, error) {
-	// port needs to be available for all supported protocols: we want to use the same port
+	// Port needs to be available for all supported protocols: we want to use the same port
 	// number for all protocols and we don't know at this point which protocols are needed.
+	// This is to preserve the legacy behavior of allocating the same nodePort for all protocols.
 	protocols := make([]ProtocolSocketData, 0, len(supportedProtocols))
 	for _, protocol := range supportedProtocols {
 		socket, err := localPortOpener.OpenLocalPort(port, protocol)
@@ -54,7 +84,7 @@ func (pt *PortTable) getFreePort(podIP string, podPort int) (int, []ProtocolSock
 			// handle wrap around
 			port = port - numPorts
 		}
-		if _, ok := pt.NodePortTable[port]; ok {
+		if _, ok := pt.NodePortTable[strconv.Itoa(port)]; ok {
 			// port is already taken
 			continue
 		}
@@ -172,7 +202,7 @@ func (pt *PortTable) AddRule(podIP string, podPort int, protocol string) (int, e
 
 	protocolSocketData.State = stateInUse
 	if !exists {
-		pt.NodePortTable[nodePort] = npData
+		pt.NodePortTable[strconv.Itoa(nodePort)] = npData
 		pt.PodEndpointTable[podIPPortFormat(podIP, podPort)] = npData
 	}
 	return npData.NodePort, nil
@@ -210,7 +240,7 @@ func (pt *PortTable) DeleteRule(podIP string, podPort int, protocol string) erro
 		if err := data.CloseSockets(); err != nil {
 			return err
 		}
-		delete(pt.NodePortTable, data.NodePort)
+		delete(pt.NodePortTable, strconv.Itoa(data.NodePort))
 		delete(pt.PodEndpointTable, podIPPortFormat(podIP, podPort))
 	}
 	return nil
@@ -231,7 +261,7 @@ func (pt *PortTable) DeleteRulesForPod(podIP string) error {
 			}
 			podEntry.Protocols = podEntry.Protocols[1:]
 		}
-		delete(pt.NodePortTable, podEntry.NodePort)
+		delete(pt.NodePortTable, strconv.Itoa(podEntry.NodePort))
 		delete(pt.PodEndpointTable, podIPPortFormat(podIP, podEntry.PodPort))
 	}
 	return nil
@@ -293,8 +323,8 @@ func (pt *PortTable) RestoreRules(allNPLPorts []rules.PodNodePort, synced chan<-
 			}
 			protocolSocketData.State = stateInUse
 		}
-		pt.NodePortTable[nplPort.NodePort] = npData
-		pt.PodEndpointTable[podIPPortFormat(nplPort.PodIP, nplPort.PodPort)] = pt.NodePortTable[nplPort.NodePort]
+		pt.NodePortTable[strconv.Itoa(nplPort.NodePort)] = npData
+		pt.PodEndpointTable[podIPPortFormat(nplPort.PodIP, nplPort.PodPort)] = pt.NodePortTable[strconv.Itoa(nplPort.NodePort)]
 	}
 	// retry mechanism as iptables-restore can fail if other components (in Antrea or other
 	// software) are accessing iptables.

pkg/agent/nodeportlocal/portcache/port_table_test.go

@@ -35,7 +35,7 @@ const (
 
 func newPortTable(mockIPTables rules.PodPortRules, mockPortOpener LocalPortOpener) *PortTable {
 	return &PortTable{
-		NodePortTable:    make(map[int]*NodePortData),
+		NodePortTable:    make(map[string]*NodePortData),
 		PodEndpointTable: make(map[string]*NodePortData),
 		StartPort:        startPort,
 		EndPort:          endPort,