NETOBSERV-1965: Add UDNs merger logic to merge udns accross nodes for the same flow

pkg/model/fields/fields.go

@@ -35,6 +35,7 @@ const (
 	DstZone                = Dst + Zone
 	Cluster                = "K8S_ClusterName"
 	UDN                    = "UDN"
+	Udns                   = "Udns"
 	Layer                  = "K8S_FlowLayer"
 	Packets                = "Packets"
 	Proto                  = "Proto"

web/locales/en/plugin__netobserv-plugin.json

@@ -42,6 +42,7 @@
   "The flow contains packets with various flags: ": "The flow contains packets with various flags: ",
   "ICMP type provided but protocol is {{proto}}": "ICMP type provided but protocol is {{proto}}",
   "ICMP code provided but protocol is {{proto}}": "ICMP code provided but protocol is {{proto}}",
+  "None": "None",
   "Invalid data provided. Check JSON for details.": "Invalid data provided. Check JSON for details.",
   "dropped": "dropped",
   "dropped by": "dropped by",
@@ -156,7 +157,6 @@
   "M": "M",
   "S": "S",
   "XS": "XS",
-  "None": "None",
   "Step {{index}}/{{count}}": "Step {{index}}/{{count}}",
   "Step {{index}}/{{count}}_plural": "Step {{index}}/{{count}}",
   "Previous tip": "Previous tip",

web/src/api/ipfix.ts

@@ -107,6 +107,10 @@ export interface Fields {
   SrcK8S_Zone?: string;
   /** Destination zone */
   DstK8S_Zone?: string;
+  /** Source network name (e.g. secondary networks or UDN) */
+  SrcK8S_NetworkName?: string;
+  /** Destination network name (e.g. secondary networks or UDN) */
+  DstK8S_NetworkName?: string;
   /** Cluster name */
   K8S_ClusterName?: string;
   /** L4 protocol */
@@ -115,6 +119,8 @@ export interface Fields {
   Interfaces?: string[];
   /** Flow direction array from the network interface observation point */
   IfDirections?: IfDirection[];
+  /** UDNs labels array */
+  Udns?: string[];
   /** Network Events */
   NetworkEvents?: string[];
   /** Logical OR combination of unique TCP flags comprised in the flow, as per RFC-9293, with additional custom flags to represent the following per-packet combinations: SYN+ACK (0x100), FIN+ACK (0x200) and RST+ACK (0x400). */

web/src/components/drawer/record/record-field.tsx

@@ -423,6 +423,16 @@ export const RecordField: React.FC<RecordFieldProps> = ({
         }
         return singleContainer(simpleTextWithTooltip(String(value)));
       }
+      case ColumnsId.udns: {
+        if (Array.isArray(value)) {
+          return nthContainer(
+            value.map(iName => simpleTextWithTooltip(iName !== '' ? String(iName) : t('None'))),
+            true,
+            false
+          );
+        }
+        return singleContainer(simpleTextWithTooltip(String(value)));
+      }
       case ColumnsId.flowdirints: {
         if (
           flow.fields.Interfaces &&

web/src/utils/__tests__/flows.spec.ts

@@ -76,7 +76,8 @@ describe('mergeFlowReporters', () => {
           SrcAddr: '10.0.0.1',
           DstAddr: '10.0.0.2',
           IfDirections: [IfDirection.Ingress, IfDirection.Egress],
-          Interfaces: ['eth0', 'abcd']
+          Interfaces: ['eth0', 'abcd'],
+          Udns: ['udn1', 'udn2']
         } as Fields,
         labels: { FlowDirection: FlowDirection.Ingress }
       },
@@ -86,7 +87,8 @@ describe('mergeFlowReporters', () => {
           SrcAddr: '10.0.0.1',
           DstAddr: '10.0.0.2',
           IfDirections: [IfDirection.Ingress],
-          Interfaces: ['genev']
+          Interfaces: ['genev'],
+          Udns: ['udn3']
         } as Fields,
         labels: { FlowDirection: FlowDirection.Egress }
       }
@@ -99,7 +101,8 @@ describe('mergeFlowReporters', () => {
         SrcAddr: '10.0.0.1',
         DstAddr: '10.0.0.2',
         IfDirections: [IfDirection.Ingress, IfDirection.Egress, IfDirection.Ingress],
-        Interfaces: ['eth0', 'abcd', 'genev']
+        Interfaces: ['eth0', 'abcd', 'genev'],
+        Udns: ['udn1', 'udn2', 'udn3']
       } as Fields,
       labels: { FlowDirection: FlowDirection.Ingress }
     });

web/src/utils/columns.ts

@@ -79,6 +79,7 @@ export enum ColumnsId {
   interfaces = 'Interfaces',
   ifdirs = 'IfDirections',
   flowdirints = 'FlowDirInts',
+  udns = 'Udns',
   recordtype = 'RecordType',
   bytesab = 'Bytes_AB',
   bytesba = 'Bytes_BA',

web/src/utils/flows.ts

@@ -1,6 +1,6 @@
 import * as _ from 'lodash';
 import { IfDirection, Record } from '../api/ipfix';
-import { get5Tuple } from './ids';
+import { get7Tuple } from './ids';
 
 const electMostRelevant = (flowsFor5Tuples: Record[]): Record => {
   // Get most relevant record, in priority with Dns or Drops info
@@ -35,24 +35,47 @@ const getInvolvedInterfaces = (flowsFor5Tuples: Record[]): { ifnames: string[];
   return { ifnames, ifdirs };
 };
 
+const getInvolvedUdns = (flowsFor5Tuples: Record[]): string[] => {
+  const cache = new Set();
+  const udns: string[] = [];
+  flowsFor5Tuples.forEach(f => {
+    if (f.fields.Udns) {
+      _.zip(f.fields.Udns).forEach(([udn]) => {
+        const key = `${udn}`;
+        if (!cache.has(key)) {
+          cache.add(key);
+          udns.push(udn!);
+        }
+      });
+    }
+  });
+  return udns;
+};
+
 export const mergeFlowReporters = (flows: Record[]): Record[] => {
   // The purpose of this function is to determine if, for a given 5 tuple, we'll look at INGRESS, EGRESS or INNER reporter
   // The assumption is that INGRESS alone, EGRESS alone or INNER alone always provide a complete visiblity
   // Favor whichever contains pktDrop and/or DNS responses
-  const grouped = _.groupBy(flows, get5Tuple);
+  const grouped = _.groupBy(flows, get7Tuple);
   const filtersIndex = _.mapValues(grouped, (records: Record[]) => electMostRelevant(records));
   const involvedInterfaces = _.mapValues(grouped, (records: Record[]) => getInvolvedInterfaces(records));
-  // Filter and inject other interfaces in elected flows
-  // An assumption is made that interfaces involved for a 5 tuples will keep being involved in the whole flows sequence
+  const involvedUdns = _.mapValues(grouped, (records: Record[]) => getInvolvedUdns(records));
+  // Filter and inject other interfaces and udns in elected flows
+  // An assumption is made that interfaces and udns involved for a 5 tuples will keep being involved in the whole flows sequence
   // If that assumption proves wrong, we may refine by looking at time overlaps between flows
   return flows
-    .filter((r: Record) => r.labels.FlowDirection === filtersIndex[get5Tuple(r)].labels.FlowDirection)
+    .filter((r: Record) => r.labels.FlowDirection === filtersIndex[get7Tuple(r)].labels.FlowDirection)
     .map(r => {
-      const interfaces = involvedInterfaces[get5Tuple(r)];
+      const key = get7Tuple(r);
+      const interfaces = involvedInterfaces[key];
       if (interfaces) {
         r.fields.Interfaces = interfaces.ifnames;
         r.fields.IfDirections = interfaces.ifdirs;
       }
+      const udns = involvedUdns[key];
+      if (udns) {
+        r.fields.Udns = udns;
+      }
       return r;
     });
 };

web/src/utils/ids.ts

@@ -45,8 +45,8 @@ export const getPeerId = (fields: Partial<TopologyMetricPeer>): string => {
   return parts.length > 0 ? parts.join(',') : idUnknown;
 };
 
-export const get5Tuple = (r: Record): string => {
-  return `${r.fields.SrcAddr}:${r.fields.SrcPort || 'x'}→${r.fields.DstAddr}:${r.fields.DstPort || 'x'}@${
-    r.fields.Proto
-  }`;
+export const get7Tuple = (r: Record): string => {
+  return `${r.fields.SrcAddr}:${r.fields.SrcPort || 'x'}:${r.fields.SrcK8S_NetworkName || 'x'}→${r.fields.DstAddr}:${
+    r.fields.DstPort || 'x'
+  }:${r.fields.DstK8S_NetworkName || 'x'}@${r.fields.Proto}`;
 };