NETOBSERV-1408: migrating netobserv-agent to TCx for new kernels (#262)

Signed-off-by: Mohamed Mahmoud <[email protected]>

Author: msherif1234

bpf/flows.c

@@ -214,13 +214,27 @@ static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
 }
 
 SEC("tc_ingress")
-int ingress_flow_parse(struct __sk_buff *skb) {
+int tc_ingress_flow_parse(struct __sk_buff *skb) {
     return flow_monitor(skb, INGRESS);
 }
 
 SEC("tc_egress")
-int egress_flow_parse(struct __sk_buff *skb) {
+int tc_egress_flow_parse(struct __sk_buff *skb) {
     return flow_monitor(skb, EGRESS);
 }
 
+SEC("tcx_ingress")
+int tcx_ingress_flow_parse(struct __sk_buff *skb) {
+    flow_monitor(skb, INGRESS);
+    // return TCX_NEXT to allow existing with other TCX hooks
+    return TCX_NEXT;
+}
+
+SEC("tcx_egress")
+int tcx_egress_flow_parse(struct __sk_buff *skb) {
+    flow_monitor(skb, EGRESS);
+    // return TCX_NEXT to allow existing with other TCX hooks
+    return TCX_NEXT;
+}
+
 char _license[] SEC("license") = "GPL";

bpf/headers/vmlinux_amd64.h

@@ -59957,6 +59957,13 @@ struct netdev_notifier_bonding_info {
 	struct netdev_bonding_info bonding_info;
 };
 
+enum tcx_action_base {
+	TCX_NEXT = -1,
+	TCX_PASS = 0,
+	TCX_DROP = 2,
+	TCX_REDIRECT = 7,
+};
+
 enum qdisc_state_t {
 	__QDISC_STATE_SCHED = 0,
 	__QDISC_STATE_DEACTIVATED = 1,

bpf/headers/vmlinux_arm64.h

@@ -77797,6 +77797,13 @@ struct netdev_notifier_bonding_info {
 	struct netdev_bonding_info bonding_info;
 };
 
+enum tcx_action_base {
+	TCX_NEXT = -1,
+	TCX_PASS = 0,
+	TCX_DROP = 2,
+	TCX_REDIRECT = 7,
+};
+
 enum qdisc_state2_t {
 	__QDISC_STATE2_RUNNING = 0,
 };

bpf/headers/vmlinux_s390.h

@@ -96143,6 +96143,13 @@ struct netdev_notifier_bonding_info {
 	struct netdev_bonding_info bonding_info;
 };
 
+enum tcx_action_base {
+	TCX_NEXT = -1,
+	TCX_PASS = 0,
+	TCX_DROP = 2,
+	TCX_REDIRECT = 7,
+};
+
 typedef int (*bpf_op_t)(struct net_device *, struct netdev_bpf *);
 
 struct dev_kfree_skb_cb {

bpf/pca.h

@@ -106,12 +106,22 @@ static inline int export_packet_payload(struct __sk_buff *skb) {
 }
 
 SEC("tc_pca_ingress")
-int ingress_pca_parse(struct __sk_buff *skb) {
+int tc_ingress_pca_parse(struct __sk_buff *skb) {
     return export_packet_payload(skb);
 }
 
 SEC("tc_pca_egress")
-int egress_pca_parse(struct __sk_buff *skb) {
+int tc_egress_pca_parse(struct __sk_buff *skb) {
+    return export_packet_payload(skb);
+}
+
+SEC("tcx_pca_ingress")
+int tcx_ingress_pca_parse(struct __sk_buff *skb) {
+    return export_packet_payload(skb);
+}
+
+SEC("tcx_pca_egress")
+int tcx_egress_pca_parse(struct __sk_buff *skb) {
     return export_packet_payload(skb);
 }
 

pkg/agent/agent.go

@@ -125,6 +125,7 @@ type Flows struct {
 type ebpfFlowFetcher interface {
 	io.Closer
 	Register(iface ifaces.Interface) error
+	AttachTCX(iface ifaces.Interface) error
 
 	LookupAndDeleteMap(*metrics.Metrics) map[ebpf.BpfFlowId][]ebpf.BpfFlowMetrics
 	DeleteMapsStaleEntries(timeOut time.Duration)
@@ -509,10 +510,14 @@ func (f *Flows) onInterfaceAdded(iface ifaces.Interface) {
 			Debug("interface does not match the allow/exclusion filters. Ignoring")
 		return
 	}
-	alog.WithField("interface", iface).Info("interface detected. Registering flow ebpfFetcher")
-	if err := f.ebpf.Register(iface); err != nil {
+	alog.WithField("interface", iface).Info("interface detected. trying to attach TCX hook")
+	if err := f.ebpf.AttachTCX(iface); err != nil {
 		alog.WithField("interface", iface).WithError(err).
-			Warn("can't register flow ebpfFetcher. Ignoring")
-		return
+			Info("can't attach to TCx hook flow ebpfFetcher. fall back to use legacy TC hook")
+		if err := f.ebpf.Register(iface); err != nil {
+			alog.WithField("interface", iface).WithError(err).
+				Warn("can't register flow ebpfFetcher. Ignoring")
+			return
+		}
 	}
 }

pkg/agent/packets_agent.go

@@ -40,7 +40,7 @@ type Packets struct {
 type ebpfPacketFetcher interface {
 	io.Closer
 	Register(iface ifaces.Interface) error
-
+	AttachTCX(iface ifaces.Interface) error
 	LookupAndDeleteMap(*metrics.Metrics) map[int][]*byte
 	ReadPerf() (perf.Record, error)
 }
@@ -246,10 +246,14 @@ func (p *Packets) onInterfaceAdded(iface ifaces.Interface) {
 			Debug("[PCA]interface does not match the allow/exclusion filters. Ignoring")
 		return
 	}
-	plog.WithField("[PCA]interface", iface).Info("interface detected. Registering packets ebpfFetcher")
-	if err := p.ebpf.Register(iface); err != nil {
+	plog.WithField("interface", iface).Info("interface detected. trying to attach TCX hook")
+	if err := p.ebpf.AttachTCX(iface); err != nil {
 		plog.WithField("[PCA]interface", iface).WithError(err).
-			Warn("can't register packet ebpfFetcher. Ignoring")
-		return
+			Info("can't attach to TCx hook packet ebpfFetcher. fall back to use legacy TC hook")
+		if err := p.ebpf.Register(iface); err != nil {
+			plog.WithField("[PCA]interface", iface).WithError(err).
+				Warn("can't register packet ebpfFetcher. Ignoring")
+			return
+		}
 	}
 }