NETOBSERV-1996: in-kernel de-duplication

- Remove interface from the flow key; instead, use it as flow value. The first interface+dir seen for a given flow is the one taken into account for counters. Other interfaces+dirs are stored in a separate map for this flow. This algorithm is more or less the deduper algo that we had in userspace. - Remove user-space deduper - Adapt user-space model for the new interfaces+directions array provided directly from ebpf structs - Remove "decorator" (which was doing the interface naming). This is just for simplification. This enrichment is now done in a more straightforward way, when creating the Record objects try optimizing alignment
netobserv · Dec 9, 2024 · 93f3d03 · 93f3d03
1 parent f8c24aa
commit 93f3d03
Show file tree

Hide file tree

Showing 40 changed files with 603 additions and 1,181 deletions.
diff --git a/bpf/flows.c b/bpf/flows.c
@@ -72,6 +72,20 @@ static inline void update_dns(additional_metrics *extra_metrics, pkt_info *pkt,
     }
 }
 
+static inline void add_observed_intf(additional_metrics *value, u32 if_index, u8 direction) {
+    if (value->nb_observed_intf < MAX_OBSERVED_INTERFACES) {
+        for (u8 i = 0; i < value->nb_observed_intf; i++) {
+            if (value->observed_intf[i].if_index == if_index &&
+                value->observed_intf[i].direction == direction) {
+                return;
+            }
+        }
+        value->observed_intf[value->nb_observed_intf].if_index = if_index;
+        value->observed_intf[value->nb_observed_intf].direction = direction;
+        value->nb_observed_intf++;
+    }
+}
+
 static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
     // If sampling is defined, will only parse 1 out of "sampling" flows
     if (sampling > 1 && (bpf_get_prandom_u32() % sampling) != 0) {
@@ -100,12 +114,8 @@ static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
         return TC_ACT_OK;
     }
 
-    //Set extra fields
-    id.if_index = skb->ifindex;
-    id.direction = direction;
-
     // check if this packet need to be filtered if filtering feature is enabled
-    bool skip = check_and_do_flow_filtering(&id, pkt.flags, 0, eth_protocol);
+    bool skip = check_and_do_flow_filtering(&id, pkt.flags, 0, eth_protocol, direction);
     if (skip) {
         return TC_ACT_OK;
     }
@@ -116,10 +126,25 @@ static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
     }
     flow_metrics *aggregate_flow = (flow_metrics *)bpf_map_lookup_elem(&aggregated_flows, &id);
     if (aggregate_flow != NULL) {
-        update_existing_flow(aggregate_flow, &pkt, len);
+        if (aggregate_flow->if_index_first_seen == skb->ifindex) {
+            update_existing_flow(aggregate_flow, &pkt, len);
+        } else {
+            // Only add info that we've seen this interface
+            additional_metrics *extra_metrics =
+                (additional_metrics *)bpf_map_lookup_elem(&additional_flow_metrics, &id);
+            if (extra_metrics != NULL) {
+                add_observed_intf(extra_metrics, skb->ifindex, direction);
+            } else {
+                additional_metrics new_metrics = {};
+                add_observed_intf(&new_metrics, skb->ifindex, direction);
+                bpf_map_update_elem(&additional_flow_metrics, &id, &new_metrics, BPF_NOEXIST);
+            }
+        }
     } else {
         // Key does not exist in the map, and will need to create a new entry.
         flow_metrics new_flow = {
+            .if_index_first_seen = skb->ifindex,
+            .direction_first_seen = direction,
             .packets = 1,
             .bytes = len,
             .eth_protocol = eth_protocol,
@@ -172,9 +197,6 @@ static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
 
     // Update additional metrics (per-CPU map)
     if (pkt.dns_id != 0 || dns_errno != 0) {
-        // hack on id will be removed with dedup-in-kernel work
-        id.direction = 0;
-        id.if_index = 0;
         additional_metrics *extra_metrics =
             (additional_metrics *)bpf_map_lookup_elem(&additional_flow_metrics, &id);
         if (extra_metrics != NULL) {

diff --git a/bpf/flows_filter.h b/bpf/flows_filter.h
@@ -33,7 +33,7 @@ static __always_inline int is_equal_ip(u8 *ip1, u8 *ip2, u8 len) {
 
 static __always_inline int do_flow_filter_lookup(flow_id *id, struct filter_key_t *key,
                                                  filter_action *action, u8 len, u8 offset,
-                                                 u16 flags, u32 drop_reason) {
+                                                 u16 flags, u32 drop_reason, u8 direction) {
     int result = 0;
 
     struct filter_value_t *rule = (struct filter_value_t *)bpf_map_lookup_elem(&filter_map, key);
@@ -157,7 +157,7 @@ static __always_inline int do_flow_filter_lookup(flow_id *id, struct filter_key_
 
         if (!is_zero_ip(rule->ip, len)) {
             // for Ingress side we can filter using dstIP and for Egress side we can filter using srcIP
-            if (id->direction == INGRESS) {
+            if (direction == INGRESS) {
                 if (is_equal_ip(rule->ip, id->dst_ip + offset, len)) {
                     BPF_PRINTK("dstIP matched\n");
                     result++;
@@ -177,7 +177,7 @@ static __always_inline int do_flow_filter_lookup(flow_id *id, struct filter_key_
         }
 
         if (rule->direction != MAX_DIRECTION) {
-            if (rule->direction == id->direction) {
+            if (rule->direction == direction) {
                 BPF_PRINTK("direction matched\n");
                 result++;
             } else {
@@ -233,7 +233,7 @@ static __always_inline int flow_filter_setup_lookup_key(flow_id *id, struct filt
  * check if the flow match filter rule and return >= 1 if the flow is to be dropped
  */
 static __always_inline int is_flow_filtered(flow_id *id, filter_action *action, u16 flags,
-                                            u32 drop_reason, u16 eth_protocol) {
+                                            u32 drop_reason, u16 eth_protocol, u8 direction) {
     struct filter_key_t key;
     u8 len, offset;
     int result = 0;
@@ -247,7 +247,7 @@ static __always_inline int is_flow_filtered(flow_id *id, filter_action *action,
         return result;
     }
 
-    result = do_flow_filter_lookup(id, &key, action, len, offset, flags, drop_reason);
+    result = do_flow_filter_lookup(id, &key, action, len, offset, flags, drop_reason, direction);
     // we have a match so return
     if (result > 0) {
         return result;
@@ -259,7 +259,7 @@ static __always_inline int is_flow_filtered(flow_id *id, filter_action *action,
         return result;
     }
 
-    return do_flow_filter_lookup(id, &key, action, len, offset, flags, drop_reason);
+    return do_flow_filter_lookup(id, &key, action, len, offset, flags, drop_reason, direction);
 }
 
 #endif //__FLOWS_FILTER_H__
diff --git a/bpf/network_events_monitoring.h b/bpf/network_events_monitoring.h
@@ -93,17 +93,14 @@ static inline int trace_network_events(struct sk_buff *skb, struct rh_psample_me
     }
 
     // check if this packet need to be filtered if filtering feature is enabled
-    bool skip = check_and_do_flow_filtering(&id, flags, 0, eth_protocol);
+    bool skip = check_and_do_flow_filtering(&id, flags, 0, eth_protocol, 0);
     if (skip) {
         return 0;
     }
 
-    for (direction dir = INGRESS; dir < MAX_DIRECTION; dir++) {
-        id.direction = dir;
-        ret = lookup_and_update_existing_flow_network_events(&id, md_len, user_cookie);
-        if (ret == 0) {
-            return ret;
-        }
+    ret = lookup_and_update_existing_flow_network_events(&id, md_len, user_cookie);
+    if (ret == 0) {
+        return ret;
     }
 
     // there is no matching flows so lets create new one and add the network event metadata

diff --git a/bpf/pca.h b/bpf/pca.h
@@ -53,12 +53,8 @@ static inline bool validate_pca_filter(struct __sk_buff *skb, direction dir) {
         return false;
     }
 
-    //Set extra fields
-    id.if_index = skb->ifindex;
-    id.direction = dir;
-
     // check if this packet need to be filtered if filtering feature is enabled
-    bool skip = check_and_do_flow_filtering(&id, pkt.flags, 0, eth_protocol);
+    bool skip = check_and_do_flow_filtering(&id, pkt.flags, 0, eth_protocol, dir);
     if (skip) {
         return false;
     }

diff --git a/bpf/pkt_drops.h b/bpf/pkt_drops.h
@@ -63,7 +63,7 @@ static inline int trace_pkt_drop(void *ctx, u8 state, struct sk_buff *skb,
     }
 
     // check if this packet need to be filtered if filtering feature is enabled
-    bool skip = check_and_do_flow_filtering(&id, flags, reason, eth_protocol);
+    bool skip = check_and_do_flow_filtering(&id, flags, reason, eth_protocol, 0);
     if (skip) {
         return 0;
     }

diff --git a/bpf/rtt_tracker.h b/bpf/rtt_tracker.h
@@ -58,7 +58,7 @@ static inline int calculate_flow_rtt_tcp(struct sock *sk, struct sk_buff *skb) {
     rtt *= 1000u;
 
     // check if this packet need to be filtered if filtering feature is enabled
-    bool skip = check_and_do_flow_filtering(&id, flags, 0, eth_protocol);
+    bool skip = check_and_do_flow_filtering(&id, flags, 0, eth_protocol, 0);
     if (skip) {
         return 0;
     }

diff --git a/bpf/types.h b/bpf/types.h
@@ -66,6 +66,7 @@ typedef __u64 u64;
 #define MAX_FILTER_ENTRIES 1 // we have only one global filter
 #define MAX_EVENT_MD 8
 #define MAX_NETWORK_EVENTS 4
+#define MAX_OBSERVED_INTERFACES 4
 
 // according to field 61 in https://www.iana.org/assignments/ipfix/ipfix.xhtml
 typedef enum direction_t {
@@ -80,19 +81,22 @@ const enum direction_t *unused1 __attribute__((unused));
 const u8 ip4in6[] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff};
 
 typedef struct flow_metrics_t {
-    struct bpf_spin_lock lock;
-    u16 eth_protocol;
-    // L2 data link layer
-    u8 src_mac[ETH_ALEN];
-    u8 dst_mac[ETH_ALEN];
-    u32 packets;
-    u64 bytes;
     // Flow start and end times as monotomic timestamps in nanoseconds
     // as output from bpf_ktime_get_ns()
     u64 start_mono_time_ts;
     u64 end_mono_time_ts;
+    u64 bytes;
+    u32 packets;
+    u16 eth_protocol;
     // TCP Flags from https://www.ietf.org/rfc/rfc793.txt
     u16 flags;
+    // L2 data link layer
+    u8 src_mac[ETH_ALEN];
+    u8 dst_mac[ETH_ALEN];
+    // OS interface index
+    u32 if_index_first_seen;
+    struct bpf_spin_lock lock;
+    u8 direction_first_seen;
     // The positive errno of a failed map insertion that caused a flow
     // to be sent via ringbuffer.
     // 0 otherwise
@@ -106,35 +110,36 @@ const struct flow_metrics_t *unused2 __attribute__((unused));
 
 typedef struct additional_metrics_t {
     struct dns_record_t {
+        u64 latency;
         u16 id;
         u16 flags;
-        u64 latency;
         u8 errno;
     } dns_record;
     struct pkt_drops_t {
-        u32 packets;
         u64 bytes;
+        u32 packets;
+        u32 latest_drop_cause;
         u16 latest_flags;
         u8 latest_state;
-        u32 latest_drop_cause;
     } pkt_drops;
     u64 flow_rtt;
-    u8 network_events_idx;
     u8 network_events[MAX_NETWORK_EVENTS][MAX_EVENT_MD];
+    struct observed_intf_t {
+        u8 direction;
+        u32 if_index;
+    } __attribute__((packed)) observed_intf[MAX_OBSERVED_INTERFACES];
+    u8 network_events_idx;
+    u8 nb_observed_intf;
 } additional_metrics;
 
 // Force emitting enums/structs into the ELF
 const struct additional_metrics_t *unused3 __attribute__((unused));
-
-// Force emitting enums/structs into the ELF
 const struct dns_record_t *unused4 __attribute__((unused));
-
-// Force emitting enums/structs into the ELF
 const struct pkt_drops_t *unused5 __attribute__((unused));
+const struct observed_intf_t *unused9 __attribute__((unused));
 
 // Attributes that uniquely identify a flow
 typedef struct flow_id_t {
-    u8 direction;
     // L3 network layer
     // IPv4 addresses are encoded as IPv6 addresses with prefix ::ffff/96
     // as described in https://datatracker.ietf.org/doc/html/rfc4038#section-4.2
@@ -147,8 +152,6 @@ typedef struct flow_id_t {
     // ICMP protocol
     u8 icmp_type;
     u8 icmp_code;
-    // OS interface index
-    u32 if_index;
 } flow_id;
 
 // Force emitting enums/structs into the ELF

diff --git a/bpf/utils.h b/bpf/utils.h
@@ -178,11 +178,11 @@ static inline int fill_ethhdr(struct ethhdr *eth, void *data_end, pkt_info *pkt,
  * check if flow filter is enabled and if we need to continue processing the packet or not
  */
 static inline bool check_and_do_flow_filtering(flow_id *id, u16 flags, u32 drop_reason,
-                                               u16 eth_protocol) {
+                                               u16 eth_protocol, u8 direction) {
     // check if this packet need to be filtered if filtering feature is enabled
     if (enable_flows_filtering || enable_pca) {
         filter_action action = ACCEPT;
-        if (is_flow_filtered(id, &action, flags, drop_reason, eth_protocol) != 0 &&
+        if (is_flow_filtered(id, &action, flags, drop_reason, eth_protocol, direction) != 0 &&
             action != MAX_FILTER_ACTIONS) {
             // we have matching rules follow through the actions to decide if we should accept or reject the flow
             // and update global counter for both cases

diff --git a/cmd/netobserv-ebpf-agent.go b/cmd/netobserv-ebpf-agent.go
@@ -51,9 +51,6 @@ func main() {
 				Error("PProf HTTP listener stopped working")
 		}()
 	}
-	if config.DeduperFCExpiry == 0 {
-		config.DeduperFCExpiry = 2 * config.CacheActiveTimeout
-	}
 
 	logrus.WithField("configuration", fmt.Sprintf("%#v", config)).Debugf("configuration loaded")
 

diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go
@@ -118,13 +118,8 @@ type Flows struct {
 	rbTracer  *flow.RingBufTracer
 	accounter *flow.Accounter
 	limiter   *flow.CapacityLimiter
-	deduper   node.MiddleFunc[[]*model.Record, []*model.Record]
 	exporter  node.TerminalFunc[[]*model.Record]
 
-	// elements used to decorate flows with extra information
-	interfaceNamer flow.InterfaceNamer
-	agentIP        net.IP
-
 	status        Status
 	promoServer   *http.Server
 	sampleDecoder *ovnobserv.SampleDecoder
@@ -275,6 +270,8 @@ func flowsAgent(cfg *Config, m *metrics.Metrics,
 		}
 		return iface
 	}
+	model.SetGlobals(agentIP, interfaceNamer)
+
 	var promoServer *http.Server
 	if cfg.MetricsEnable {
 		promoServer = promo.InitializePrometheus(m.Settings)
@@ -287,26 +284,19 @@ func flowsAgent(cfg *Config, m *metrics.Metrics,
 	rbTracer := flow.NewRingBufTracer(fetcher, mapTracer, cfg.CacheActiveTimeout, m)
 	accounter := flow.NewAccounter(cfg.CacheMaxFlows, cfg.CacheActiveTimeout, time.Now, monotime.Now, m)
 	limiter := flow.NewCapacityLimiter(m)
-	var deduper node.MiddleFunc[[]*model.Record, []*model.Record]
-	if cfg.Deduper == DeduperFirstCome {
-		deduper = flow.Dedupe(cfg.DeduperFCExpiry, cfg.DeduperJustMark, cfg.DeduperMerge, interfaceNamer, m)
-	}
 
 	return &Flows{
-		ebpf:           fetcher,
-		exporter:       exporter,
-		interfaces:     registerer,
-		filter:         filter,
-		cfg:            cfg,
-		mapTracer:      mapTracer,
-		rbTracer:       rbTracer,
-		accounter:      accounter,
-		limiter:        limiter,
-		deduper:        deduper,
-		agentIP:        agentIP,
-		interfaceNamer: interfaceNamer,
-		promoServer:    promoServer,
-		sampleDecoder:  s,
+		ebpf:          fetcher,
+		exporter:      exporter,
+		interfaces:    registerer,
+		filter:        filter,
+		cfg:           cfg,
+		mapTracer:     mapTracer,
+		rbTracer:      rbTracer,
+		accounter:     accounter,
+		limiter:       limiter,
+		promoServer:   promoServer,
+		sampleDecoder: s,
 	}, nil
 }
 
@@ -501,9 +491,6 @@ func (f *Flows) buildAndStartPipeline(ctx context.Context) (*node.Terminal[[]*mo
 	limiter := node.AsMiddle(f.limiter.Limit,
 		node.ChannelBufferLen(f.cfg.BuffersLength))
 
-	decorator := node.AsMiddle(flow.Decorate(f.agentIP, f.interfaceNamer),
-		node.ChannelBufferLen(f.cfg.BuffersLength))
-
 	ebl := f.cfg.ExporterBufferLength
 	if ebl == 0 {
 		ebl = f.cfg.BuffersLength
@@ -514,17 +501,9 @@ func (f *Flows) buildAndStartPipeline(ctx context.Context) (*node.Terminal[[]*mo
 
 	rbTracer.SendsTo(accounter)
 
-	if f.deduper != nil {
-		deduper := node.AsMiddle(f.deduper, node.ChannelBufferLen(f.cfg.BuffersLength))
-		mapTracer.SendsTo(deduper)
-		accounter.SendsTo(deduper)
-		deduper.SendsTo(limiter)
-	} else {
-		mapTracer.SendsTo(limiter)
-		accounter.SendsTo(limiter)
-	}
-	limiter.SendsTo(decorator)
-	decorator.SendsTo(export)
+	mapTracer.SendsTo(limiter)
+	accounter.SendsTo(limiter)
+	limiter.SendsTo(export)
 
 	alog.Debug("starting graph")
 	mapTracer.Start()