proxy: Find CRD proxy ports by name instead of by type

Find CRD proxy port by name instead of type. This is needed for enabling
CEC CRD defined listeners to be used in CNPs. Prior to this CRD proxy
ports did not use this code path, which is only called from endpoint
policy updates, so there was no need to find CRD proxy ports by name.

Signed-off-by: Jarno Rajahalme <jarno@isovalent.com>

pkg/policy/l4.go

@@ -410,6 +410,11 @@ func (l4 *L4Filter) GetPort() uint16 {
 	return uint16(l4.Port)
 }
 
+// GetListener returns the optional listener name.
+func (l4 *L4Filter) GetListener() string {
+	return ""
+}
+
 // ToMapState converts filter into a MapState with two possible values:
 //
 //	Entry with ProxyPort = 0: No proxy redirection is needed for this key
@@ -1182,4 +1187,5 @@ type ProxyPolicy interface {
 	GetL7Parser() L7ParserType
 	GetIngress() bool
 	GetPort() uint16
+	GetListener() string
 }

pkg/policy/visibility.go

@@ -211,3 +211,8 @@ func (v *VisibilityMetadata) GetIngress() bool {
 func (v *VisibilityMetadata) GetPort() uint16 {
 	return v.Port
 }
+
+// GetListener returns the optional listener name.
+func (l4 *VisibilityMetadata) GetListener() string {
+	return ""
+}

pkg/proxy/proxy.go

@@ -283,7 +283,7 @@ func (p *Proxy) AckProxyPort(ctx context.Context, name string) error {
 // to keep the use count up-to-date.
 // Must be called with proxyPortsMutex held!
 func (p *Proxy) ackProxyPort(ctx context.Context, name string, pp *ProxyPort) error {
-	scopedLog := log.WithField("proxy port name", name)
+	scopedLog := log.WithField(fieldProxyRedirectID, name)
 
 	// Datapath rules are added only after we know the proxy configuration
 	// with the actual port number has succeeded. Deletion of the rules
@@ -303,6 +303,7 @@ func (p *Proxy) ackProxyPort(ctx context.Context, name string, pp *ProxyPort) er
 		pp.rulesPort = pp.proxyPort
 	}
 	pp.nRedirects++
+	scopedLog.Debugf("AckProxyPort: acked proxy port %d (%v)", pp.proxyPort, *pp)
 	return nil
 }
 
@@ -320,6 +321,8 @@ func (p *Proxy) releaseProxyPort(name string) error {
 			return fmt.Errorf("Can't release proxy port: proxy %s on %d has a static listener", name, pp.proxyPort)
 		}
 
+		log.WithField(fieldProxyRedirectID, name).Debugf("Delayed release of proxy port %d", pp.proxyPort)
+
 		// Allow the port to be reallocated for other use if needed.
 		allocatedPorts[pp.proxyPort] = false
 		pp.proxyPort = 0
@@ -329,24 +332,30 @@ func (p *Proxy) releaseProxyPort(name string) error {
 		// Leave the datapath rules behind on the hope that they get reused later.
 		// This becomes possible when we are able to keep the proxy listeners
 		// configured also when there are no redirects.
-		log.WithField(fieldProxyRedirectID, name).Debugf("Delayed release of proxy port %d", pp.proxyPort)
 	}
 
 	return nil
 }
 
-// findProxyPortByType returns a ProxyPort matching the given type and direction, if found.
-// This is mainly useful for getting a statically defined proxy port. Dynamic types
-// may have multiple instances and any one of them can be returned.
-// Must be called with proxyPortsMutex held!
-func findProxyPortByType(l7Type ProxyType, ingress bool) (string, *ProxyPort) {
+// findProxyPortByType returns a ProxyPort matching the given type, listener name, and direction, if
+// found.  Must be called with proxyPortsMutex held!
+func findProxyPortByType(l7Type ProxyType, listener string, ingress bool) (string, *ProxyPort) {
 	portType := l7Type
 	switch l7Type {
-	case ProxyTypeDNS, ProxyTypeHTTP, ProxyTypeCRD:
+	case ProxyTypeCRD:
+		// CRD proxy ports are dynamically created, look up by name
+		if pp, ok := proxyPorts[listener]; ok && pp.proxyType == ProxyTypeCRD {
+			return listener, pp
+		}
+		log.Debugf("findProxyPortByType: can not find crd listener %s from %v", listener, proxyPorts)
+		return "", nil
+	case ProxyTypeDNS, ProxyTypeHTTP:
+		// Look up by the given type
 	default:
 		// "Unknown" parsers are assumed to be Proxylib (TCP) parsers, which
 		// is registered with an empty string.
-		// This works also for explicit TCP and TLS parser types.
+		// This works also for explicit TCP and TLS parser types, which are backed by the
+		// TCP Proxy filter chain.
 		portType = ProxyTypeAny
 	}
 	// proxyPorts is small enough to not bother indexing it.
@@ -358,12 +367,12 @@ func findProxyPortByType(l7Type ProxyType, ingress bool) (string, *ProxyPort) {
 	return "", nil
 }
 
-func proxyTypeNotFoundError(proxyType ProxyType, ingress bool) error {
+func proxyTypeNotFoundError(proxyType ProxyType, listener string, ingress bool) error {
 	dir := "egress"
 	if ingress {
 		dir = "ingress"
 	}
-	return fmt.Errorf("unrecognized %s proxy type: %s", dir, proxyType)
+	return fmt.Errorf("unrecognized %s proxy type for %s: %s", dir, listener, proxyType)
 }
 
 func proxyNotFoundError(name string) error {
@@ -415,6 +424,8 @@ func (p *Proxy) AllocateProxyPort(name string, ingress bool) (uint16, error) {
 	proxyPorts[name] = pp
 	pp.reservePort() // marks port as reserved, 'pp' as configured
 
+	log.WithField(fieldProxyRedirectID, name).Debugf("AllocateProxyPort: allocated proxy port %d (%v)", pp.proxyPort, *pp)
+
 	return pp.proxyPort, nil
 }
 
@@ -530,9 +541,9 @@ func (p *Proxy) CreateOrUpdateRedirect(ctx context.Context, l4 policy.ProxyPolic
 
 	proxyPortsMutex.Lock()
 	defer proxyPortsMutex.Unlock()
-	ppName, pp := findProxyPortByType(ProxyType(l4.GetL7Parser()), l4.GetIngress())
+	ppName, pp := findProxyPortByType(ProxyType(l4.GetL7Parser()), l4.GetListener(), l4.GetIngress())
 	if pp == nil {
-		err = proxyTypeNotFoundError(ProxyType(l4.GetL7Parser()), l4.GetIngress())
+		err = proxyTypeNotFoundError(ProxyType(l4.GetL7Parser()), l4.GetListener(), l4.GetIngress())
 		revertFunc()
 		return 0, err, nil, nil
 	}
@@ -617,7 +628,7 @@ func (p *Proxy) CreateOrUpdateRedirect(ctx context.Context, l4 policy.ProxyPolic
 	}
 
 	// an error occurred, and we have no more retries
-	scopedLog.WithError(err).Error("Unable to create ", l4.GetL7Parser(), " proxy")
+	scopedLog.WithError(err).Errorf("Unable to create %s proxy %s", l4.GetL7Parser(), l4.GetListener())
 	revertFunc() // Ignore errors while reverting. This is best-effort.
 	return 0, err, nil, nil
 }

pkg/proxy/proxy_test.go

@@ -48,7 +48,7 @@ func (s *ProxySuite) TestPortAllocator(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(port1a, Equals, port1)
 
-	name, pp := findProxyPortByType(ProxyTypeCRD, false)
+	name, pp := findProxyPortByType(ProxyTypeCRD, "listener1", false)
 	c.Assert(name, Equals, "listener1")
 	c.Assert(pp.proxyType, Equals, ProxyTypeCRD)
 	c.Assert(pp.proxyPort, Equals, port)