Support service destination in traceflow

build/yamls/antrea-eks.yml

@@ -211,6 +211,9 @@ spec:
               - required:
                 - pod
                 - namespace
+              - required:
+                - service
+                - namespace
               - required:
                 - ip
               properties:
@@ -221,6 +224,8 @@ spec:
                   type: string
                 pod:
                   type: string
+                service:
+                  type: string
               type: object
             packet:
               properties:

build/yamls/antrea-gke.yml

@@ -211,6 +211,9 @@ spec:
               - required:
                 - pod
                 - namespace
+              - required:
+                - service
+                - namespace
               - required:
                 - ip
               properties:
@@ -221,6 +224,8 @@ spec:
                   type: string
                 pod:
                   type: string
+                service:
+                  type: string
               type: object
             packet:
               properties:

build/yamls/antrea-ipsec.yml

@@ -211,6 +211,9 @@ spec:
               - required:
                 - pod
                 - namespace
+              - required:
+                - service
+                - namespace
               - required:
                 - ip
               properties:
@@ -221,6 +224,8 @@ spec:
                   type: string
                 pod:
                   type: string
+                service:
+                  type: string
               type: object
             packet:
               properties:

build/yamls/antrea.yml

@@ -211,6 +211,9 @@ spec:
               - required:
                 - pod
                 - namespace
+              - required:
+                - service
+                - namespace
               - required:
                 - ip
               properties:
@@ -221,6 +224,8 @@ spec:
                   type: string
                 pod:
                   type: string
+                service:
+                  type: string
               type: object
             packet:
               properties:

build/yamls/base/crds.yml

@@ -104,13 +104,16 @@ spec:
               properties:
                 pod:
                   type: string
+                service:
+                  type: string
                 namespace:
                   type: string
                 ip:
                   type: string
                   format: ipv4
               oneOf:
                 - required: ["pod", "namespace"]
+                - required: ["service", "namespace"]
                 - required: ["ip"]
             packet:
               type: object

cmd/antrea-agent/agent.go

@@ -138,6 +138,7 @@ func run(o *Options) error {
 	if features.DefaultFeatureGate.Enabled(features.Traceflow) {
 		traceflowController = traceflow.NewTraceflowController(
 			k8sClient,
+			informerFactory,
 			crdClient,
 			traceflowInformer,
 			ofClient,

cmd/antrea-controller/controller.go

@@ -93,7 +93,7 @@ func run(o *Options) error {
 
 	var traceflowController *traceflow.Controller
 	if features.DefaultFeatureGate.Enabled(features.Traceflow) {
-		traceflowController = traceflow.NewTraceflowController(crdClient, traceflowInformer)
+		traceflowController = traceflow.NewTraceflowController(crdClient, podInformer, traceflowInformer)
 	}
 
 	apiServerConfig, err := createAPIServerConfig(o.config.ClientConnection.Kubeconfig,

pkg/agent/controller/traceflow/packetin.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/contiv/libOpenflow/openflow13"
+	"github.com/contiv/libOpenflow/protocol"
 	"github.com/contiv/ofnet/ofctrl"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/util/retry"
@@ -100,6 +101,27 @@ func (c *Controller) parsePacketIn(pktIn *ofctrl.PacketIn) (*opsv1alpha1.Tracefl
 		obs = append(obs, *ob)
 	}
 
+	// Collect service DNAT.
+	if pktIn.Data.Ethertype == 0x800 {
+		ipPacket, ok := pktIn.Data.Data.(*protocol.IPv4)
+		if !ok {
+			return nil, nil, errors.New("invalid traceflow IPv4 packet")
+		}
+		ctNwDst, err := getInfoInCtNwDstField(matchers)
+		if err != nil {
+			return nil, nil, err
+		}
+		ipDst := ipPacket.NWDst.String()
+		if ctNwDst != "" && ipDst != ctNwDst {
+			ob := &opsv1alpha1.Observation{
+				Component:       opsv1alpha1.LB,
+				Action:          opsv1alpha1.Forwarded,
+				TranslatedDstIP: ipDst,
+			}
+			obs = append(obs, *ob)
+		}
+	}
+
 	// Collect egress conjunctionID and get NetworkPolicy from cache.
 	if match = getMatchRegField(matchers, uint32(openflow.EgressReg)); match != nil {
 		egressInfo, err := getInfoInReg(match, nil)
@@ -194,3 +216,15 @@ func getInfoInTunnelDst(regMatch *ofctrl.MatchField) (string, error) {
 	}
 	return regValue.String(), nil
 }
+
+func getInfoInCtNwDstField(matchers *ofctrl.Matchers) (string, error) {
+	match := matchers.GetMatchByName("NXM_NX_CT_NW_DST")
+	if match == nil {
+		return "", nil
+	}
+	regValue, ok := match.GetValue().(net.IP)
+	if !ok {
+		return "", errors.New("packet-in conntrack IP destination value cannot be retrieved from metadata")
+	}
+	return regValue.String(), nil
+}

pkg/agent/controller/traceflow/traceflow_controller.go

@@ -26,7 +26,9 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/informers"
 	clientset "k8s.io/client-go/kubernetes"
+	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog"
@@ -38,6 +40,7 @@ import (
 	clientsetversioned "github.com/vmware-tanzu/antrea/pkg/client/clientset/versioned"
 	opsinformers "github.com/vmware-tanzu/antrea/pkg/client/informers/externalversions/ops/v1alpha1"
 	opslisters "github.com/vmware-tanzu/antrea/pkg/client/listers/ops/v1alpha1"
+	"github.com/vmware-tanzu/antrea/pkg/features"
 	"github.com/vmware-tanzu/antrea/pkg/ovs/ovsconfig"
 )
 
@@ -65,6 +68,8 @@ const (
 // the switch for traceflow request.
 type Controller struct {
 	kubeClient             clientset.Interface
+	serviceLister          corelisters.ServiceLister
+	serviceListerSynced    cache.InformerSynced
 	traceflowClient        clientsetversioned.Interface
 	traceflowInformer      opsinformers.TraceflowInformer
 	traceflowLister        opslisters.TraceflowLister
@@ -85,6 +90,7 @@ type Controller struct {
 // events.
 func NewTraceflowController(
 	kubeClient clientset.Interface,
+	informerFactory informers.SharedInformerFactory,
 	traceflowClient clientsetversioned.Interface,
 	traceflowInformer opsinformers.TraceflowInformer,
 	client openflow.Client,
@@ -118,6 +124,11 @@ func NewTraceflowController(
 	)
 	// Register packetInHandler
 	c.ofClient.RegisterPacketInHandler("traceflow", c)
+	// Add serviceLister if AntreaProxy enabled
+	if features.DefaultFeatureGate.Enabled(features.AntreaProxy) {
+		c.serviceLister = informerFactory.Core().V1().Services().Lister()
+		c.serviceListerSynced = informerFactory.Core().V1().Services().Informer().HasSynced
+	}
 	return c
 }
 
@@ -135,6 +146,12 @@ func (c *Controller) Run(stopCh <-chan struct{}) {
 	defer klog.Infof("Shutting down %s", controllerName)
 
 	klog.Infof("Waiting for caches to sync for %s", controllerName)
+	if features.DefaultFeatureGate.Enabled(features.AntreaProxy) {
+		if !cache.WaitForCacheSync(stopCh, c.serviceListerSynced) {
+			klog.Errorf("Unable to sync service cache for %s", controllerName)
+			return
+		}
+	}
 	if !cache.WaitForCacheSync(stopCh, c.traceflowListerSynced) {
 		klog.Errorf("Unable to sync caches for %s", controllerName)
 		return
@@ -244,17 +261,21 @@ func (c *Controller) syncTraceflow(traceflowName string) error {
 // startTraceflow deploys OVS flow entries for Traceflow and inject packet if current Node
 // is Sender Node.
 func (c *Controller) startTraceflow(tf *opsv1alpha1.Traceflow) error {
-	// Deploy flow entries for traceflow
-	klog.V(2).Infof("Deploy flow entries for Traceflow %s", tf.Name)
-	err := c.ofClient.InstallTraceflowFlows(tf.Status.DataplaneTag)
+	err := c.validateTraceflow(tf)
 	defer func() {
 		if err != nil {
-			c.errorTraceflowCRD(tf, fmt.Sprintf("Node: %s, error: %+v", tf.Name, err))
+			c.errorTraceflowCRD(tf, fmt.Sprintf("Node: %s, error: %+v", c.nodeConfig.Name, err))
 		}
 	}()
 	if err != nil {
 		return err
 	}
+	// Deploy flow entries for traceflow
+	klog.V(2).Infof("Deploy flow entries for Traceflow %s", tf.Name)
+	err = c.ofClient.InstallTraceflowFlows(tf.Status.DataplaneTag)
+	if err != nil {
+		return err
+	}
 
 	// TODO: let controller compute the source Node, and the source Node can just return an error,
 	//  if fails to find the Pod.
@@ -268,6 +289,13 @@ func (c *Controller) startTraceflow(tf *opsv1alpha1.Traceflow) error {
 	return err
 }
 
+func (c *Controller) validateTraceflow(tf *opsv1alpha1.Traceflow) error {
+	if tf.Spec.Destination.Service != "" && !features.DefaultFeatureGate.Enabled(features.AntreaProxy) {
+		return errors.New("service destination requires AntreaProxy feature enabled")
+	}
+	return nil
+}
+
 func (c *Controller) injectPacket(tf *opsv1alpha1.Traceflow) error {
 	podInterfaces := c.interfaceStore.GetContainerInterfacesByPod(tf.Spec.Source.Pod, tf.Spec.Source.Namespace)
 	// Update Traceflow phase to Running.
@@ -285,7 +313,7 @@ func (c *Controller) injectPacket(tf *opsv1alpha1.Traceflow) error {
 		if hasInterface {
 			dstMAC = dstPodInterface.MAC.String()
 		}
-	} else {
+	} else if tf.Spec.Destination.Pod != "" {
 		dstPodInterfaces := c.interfaceStore.GetContainerInterfacesByPod(tf.Spec.Destination.Pod, tf.Spec.Destination.Namespace)
 		if len(dstPodInterfaces) > 0 {
 			dstMAC = dstPodInterfaces[0].MAC.String()
@@ -299,10 +327,18 @@ func (c *Controller) injectPacket(tf *opsv1alpha1.Traceflow) error {
 			dstIP = dstPod.Status.PodIP
 			dstNodeIP = dstPod.Status.HostIP
 		}
+	} else if tf.Spec.Destination.Service != "" {
+		dstSvc, err := c.serviceLister.Services(tf.Spec.Destination.Namespace).Get(tf.Spec.Destination.Service)
+		if err != nil {
+			return err
+		}
+		dstIP = dstSvc.Spec.ClusterIP
 	}
-	if dstNodeIP != "" {
+	// Check encap status if Node of destination Pod is remote or destination is IP/Service
+	if dstNodeIP != "" || tf.Spec.Destination.Pod == "" {
 		peerIP := net.ParseIP(dstNodeIP)
-		if c.networkConfig.TunnelType == ovsconfig.GeneveTunnel && peerIP != nil && c.networkConfig.TrafficEncapMode.NeedsEncapToPeer(peerIP, c.nodeConfig.NodeIPAddr) {
+		if c.networkConfig.TunnelType == ovsconfig.GeneveTunnel && (tf.Spec.Destination.Pod == "" || c.networkConfig.TrafficEncapMode.NeedsEncapToPeer(peerIP, c.nodeConfig.NodeIPAddr)) {
+			// Destination is IP/Service or packet will be encapsulated to remote Node
 			// Wait a small period for other Nodes.
 			time.Sleep(time.Duration(injectPacketDelay) * time.Second)
 		} else {

pkg/agent/openflow/pipeline.go

@@ -513,12 +513,19 @@ func (c *client) connectionTrackFlows(category cookie.Category) []binding.Flow {
 // avoid unexpected packet drop in Traceflow.
 func (c *client) traceflowConnectionTrackFlows(dataplaneTag uint8, category cookie.Category) binding.Flow {
 	connectionTrackStateTable := c.pipeline[conntrackStateTable]
-	return connectionTrackStateTable.BuildFlow(priorityNormal+2).
+	flowBuilder := connectionTrackStateTable.BuildFlow(priorityLow+2).
 		MatchRegRange(int(TraceflowReg), uint32(dataplaneTag), OfTraceflowMarkRange).
 		SetHardTimeout(300).
-		Action().ResubmitToTable(connectionTrackStateTable.GetNext()).
-		Cookie(c.cookieAllocator.Request(category).Raw()).
-		Done()
+		Cookie(c.cookieAllocator.Request(category).Raw())
+	if c.enableProxy {
+		flowBuilder = flowBuilder.
+			Action().ResubmitToTable(sessionAffinityTable).
+			Action().ResubmitToTable(serviceLBTable)
+	} else {
+		flowBuilder = flowBuilder.
+			Action().ResubmitToTable(connectionTrackStateTable.GetNext())
+	}
+	return flowBuilder.Done()
 }
 
 // reEntranceBypassCTFlow generates flow that bypass CT for traffic re-entering host network space.