Improve netflow zone/region discovery (#614)

Author: RomanMelnyk113

api/v1/kube/kube_api.pb.go

@@ -18,6 +18,9 @@ message GetClusterInfoRequest {}
 message GetClusterInfoResponse {
   repeated string pods_cidr = 1;
   repeated string service_cidr = 2;
+  repeated string node_cidr = 4;
+  repeated string vpc_cidr = 5;
+  // other_cidr is a list of known cloud services CIDRs
   repeated string other_cidr = 3;
 }
 

api/v1/kube/kube_api.proto

@@ -27,7 +27,8 @@ import (
 type clusterInfo struct {
 	podCidr     []netip.Prefix
 	serviceCidr []netip.Prefix
-	otherCidr   []netip.Prefix
+	nodeCidr    []netip.Prefix
+	vpcCidr     []netip.Prefix
 	clusterCidr []netip.Prefix
 }
 
@@ -62,12 +63,27 @@ func (c *Controller) getClusterInfo(ctx context.Context) (*clusterInfo, error) {
 			res.serviceCidr = append(res.serviceCidr, subnet)
 			res.clusterCidr = append(res.clusterCidr, subnet)
 		}
+		for _, cidr := range resp.NodeCidr {
+			subnet, err := netip.ParsePrefix(cidr)
+			if err != nil {
+				return nil, fmt.Errorf("parsing node cidr: %w", err)
+			}
+			res.nodeCidr = append(res.nodeCidr, subnet)
+			res.clusterCidr = append(res.clusterCidr, subnet)
+		}
+		for _, cidr := range resp.VpcCidr {
+			subnet, err := netip.ParsePrefix(cidr)
+			if err != nil {
+				return nil, fmt.Errorf("parsing vpc cidr: %w", err)
+			}
+			res.vpcCidr = append(res.vpcCidr, subnet)
+			res.clusterCidr = append(res.clusterCidr, subnet)
+		}
 		for _, cidr := range resp.OtherCidr {
 			subnet, err := netip.ParsePrefix(cidr)
 			if err != nil {
 				return nil, fmt.Errorf("parsing other cidr: %w", err)
 			}
-			res.otherCidr = append(res.otherCidr, subnet)
 			res.clusterCidr = append(res.clusterCidr, subnet)
 		}
 		return &res, nil
@@ -78,6 +94,9 @@ func (c *Controller) runNetflowPipeline(ctx context.Context) error {
 	c.log.Info("running netflow pipeline")
 	defer c.log.Info("netflow pipeline done")
 
+	// FIXME: we fetch cluster networks ranges only once when agent pod starts
+	// which could lead to a problem that some CIDRs are not yet indexed in controller pod
+	// and some flows can be missed.
 	if c.cfg.Netflow.CheckClusterNetworkRanges {
 		clusterInfoCtx, clusterInfoCancel := context.WithTimeout(ctx, time.Second*60)
 		defer clusterInfoCancel()
@@ -87,7 +106,10 @@ func (c *Controller) runNetflowPipeline(ctx context.Context) error {
 		}
 		c.clusterInfo = clusterInfo
 		if clusterInfo != nil {
-			c.log.Infof("fetched cluster info, pod_cidr=%s, service_cidr=%s", clusterInfo.podCidr, clusterInfo.serviceCidr)
+			c.log.Infof(
+				"fetched cluster info, pod_cidr=%s, service_cidr=%s, node_cidr=%s, vpc_cidr=%s",
+				clusterInfo.podCidr, clusterInfo.serviceCidr, clusterInfo.nodeCidr, clusterInfo.vpcCidr,
+			)
 		}
 	}
 

cmd/agent/daemon/pipeline/netflow_pipeline.go

@@ -15,6 +15,7 @@ type VPCMetadataConfig struct {
 	Type              string        `json:"type"`
 	NetworkName       string        `json:"networkName"`
 	RefreshInterval   time.Duration `json:"refreshInterval"`
+	CacheSize         uint32        `json:"CacheSize"`
 	CredentialsFile   string        `json:"credentialsFile"`
 	GCPProjectID      string        `json:"gcpProjectID"`
 	AWSAccountID      string        `json:"awsAccountID"`
@@ -59,7 +60,7 @@ func (c *VPCMetadataController) Run(ctx context.Context) error {
 
 	c.log.Infof("cloud provider %s initialized successfully", provider.Type())
 
-	vpcIndex := kube.NewVPCIndex(c.log, c.cfg.RefreshInterval)
+	vpcIndex := kube.NewVPCIndex(c.log, c.cfg.RefreshInterval, c.cfg.CacheSize)
 
 	if err := c.fetchInitialMetadata(ctx, provider, vpcIndex); err != nil {
 		c.log.Errorf("failed to fetch initial VPC metadata: %v", err)

cmd/controller/controllers/vpc_metadata_controller.go

@@ -89,21 +89,20 @@ func NewClient(
 		kvisorControllerContainerName: "controller",
 		client:                        client,
 		index:                         NewIndex(),
-		// TODO: set default
-		vpcIndex:  nil,
-		version:   version,
-		ipInfoTTL: 30 * time.Second,
+		vpcIndex:                      NewVPCIndex(log, 0, 0),
+		version:                       version,
+		ipInfoTTL:                     30 * time.Second,
 	}
 }
 
 // SetVPCIndex sets the VPC index for enriching external IPs with VPC metadata.
-func (i *Client) SetVPCIndex(vpcIndex *VPCIndex) {
-	i.vpcIndex = vpcIndex
+func (c *Client) SetVPCIndex(vpcIndex *VPCIndex) {
+	c.vpcIndex = vpcIndex
 }
 
 // GetVPCIndex returns the VPC index if available.
-func (i *Client) GetVPCIndex() *VPCIndex {
-	return i.vpcIndex
+func (c *Client) GetVPCIndex() *VPCIndex {
+	return c.vpcIndex
 }
 
 func (c *Client) RegisterHandlers(factory informers.SharedInformerFactory) {
@@ -173,6 +172,11 @@ func (c *Client) runCleanup() {
 	if deleted > 0 {
 		c.log.Debugf("ips index cleanup done, removed %d ips", deleted)
 	}
+
+	deletedCidrs := c.index.nodesCIDRIndex.Cleanup(c.ipInfoTTL)
+	if deletedCidrs > 0 {
+		c.log.Debugf("nodes cidr index cleanup done, removed %d cidrs", deletedCidrs)
+	}
 }
 
 func (c *Client) eventHandler() cache.ResourceEventHandler {
@@ -259,22 +263,30 @@ func (c *Client) GetIPInfo(ip netip.Addr) (IPInfo, bool) {
 	c.mu.RLock()
 	defer c.mu.RUnlock()
 
-	val, found := c.index.ipsDetails.find(ip)
-	return val, found
-}
-
-func (c *Client) GetIPsInfo(ips []netip.Addr) []IPInfo {
-	c.mu.RLock()
-	defer c.mu.RUnlock()
+	// step 1: check IPs from kube client first
+	// all known pods/services/endpoints/nodes
+	val, kubeIPFound := c.index.ipsDetails.find(ip)
+	if kubeIPFound && val.zone != "" {
+		return val, true
+	}
+	// step 2: check IPs from nodes pod CIDRs
+	// in case when IP is not found within known k8s resources
+	// i.e. CNI bridge gateway IP
+	cidrInfo, nodeCIDRFound := c.index.nodesCIDRIndex.Lookup(ip)
+	if nodeCIDRFound && cidrInfo.Metadata != "" {
+		val.Node = c.index.nodesByName[cidrInfo.Metadata]
+		return val, true
+	}
 
-	var res []IPInfo
-	for _, ip := range ips {
-		val, found := c.index.ipsDetails.find(ip)
-		if found {
-			res = append(res, val)
-		}
+	// step 3: check IPs from VPC index
+	if vpcInfo, vpcCIDRFound := c.vpcIndex.LookupIP(ip); vpcCIDRFound {
+		val.zone = vpcInfo.Zone
+		val.region = vpcInfo.Region
+		val.cloudDomain = vpcInfo.CloudDomain
+		return val, true
 	}
-	return res
+
+	return val, kubeIPFound || nodeCIDRFound
 }
 
 func (c *Client) GetPodInfo(uid string) (*PodInfo, bool) {
@@ -317,6 +329,7 @@ func (c *Client) GetOwnerUID(obj Object) string {
 type ClusterInfo struct {
 	PodCidr     []string
 	ServiceCidr []string
+	NodeCidr    []string
 }
 
 func (c *Client) GetClusterInfo(ctx context.Context) (*ClusterInfo, error) {
@@ -327,11 +340,27 @@ func (c *Client) GetClusterInfo(ctx context.Context) (*ClusterInfo, error) {
 	}
 
 	var res ClusterInfo
+	// Find nodes cidr.
+	// This is neeeded when `netflow-check-cluster-network-ranges` flag enabled
+	// to allow to enrich destination flows with node cidr
+	nodeCidrMap := make(map[string]struct{})
+	for _, node := range c.index.nodesByName {
+		nodeCidr, err := getNodeCidrFromNodeSpec(node)
+		if err != nil {
+			continue
+		}
+		_, found := nodeCidrMap[nodeCidr]
+		if nodeCidr != "" && !found {
+			res.NodeCidr = append(res.NodeCidr, nodeCidr)
+			nodeCidrMap[nodeCidr] = struct{}{}
+		}
+	}
+
 	// Try to find pods cidr from nodes.
 	for _, node := range c.index.nodesByName {
 		podCidr, err := getPodCidrFromNodeSpec(node)
 		if err != nil {
-			return nil, err
+			continue
 		}
 		if len(podCidr) > 0 {
 			res.PodCidr = podCidr
@@ -373,6 +402,22 @@ func (c *Client) GetClusterInfo(ctx context.Context) (*ClusterInfo, error) {
 	return &res, nil
 }
 
+func getPodCidrsFromNodeSpec(node *corev1.Node) ([]netip.Prefix, error) {
+	nodeCidrs := node.Spec.PodCIDRs
+	if len(nodeCidrs) == 0 && node.Spec.PodCIDR != "" {
+		nodeCidrs = []string{node.Spec.PodCIDR}
+	}
+	var podCidrs []netip.Prefix
+	for _, cidr := range nodeCidrs {
+		subnet, err := netip.ParsePrefix(cidr)
+		if err != nil {
+			return nil, fmt.Errorf("parsing pod cidr: %w", err)
+		}
+		podCidrs = append(podCidrs, subnet)
+	}
+	return podCidrs, nil
+}
+
 func getPodCidrFromNodeSpec(node *corev1.Node) ([]string, error) {
 	nodeCidrs := node.Spec.PodCIDRs
 	if len(nodeCidrs) == 0 && node.Spec.PodCIDR != "" {
@@ -407,6 +452,19 @@ func getPodCidrFromPod(pod *corev1.Pod) ([]string, error) {
 	return podCidr, nil
 }
 
+func getNodeCidrFromNodeSpec(node *corev1.Node) (string, error) {
+	for _, address := range node.Status.Addresses {
+		if address.Type == corev1.NodeInternalIP {
+			cidr, err := parseIPToCidr(address.Address)
+			if err != nil {
+				return "", err
+			}
+			return cidr, nil
+		}
+	}
+	return "", nil
+}
+
 func getServiceCidr(ctx context.Context, client kubernetes.Interface, namespace string, ipDetails ipsDetails) ([]string, error) {
 	res, err := getServiceCidrFromServiceCreation(ctx, client, namespace)
 	if err == nil {