support ip-in-ip mode network interface

erda-project · Aug 22, 2024 · c83dbca · c83dbca
1 parent fab1a85
commit c83dbca
Show file tree

Hide file tree

Showing 10 changed files with 124 additions and 88 deletions.
diff --git a/ebpf/include/protocol.h b/ebpf/include/protocol.h
@@ -42,6 +42,9 @@ enum eth_ip_type {
 
 #define MYSQL_ERROR_MESSAGE_MAX_SIZE 10
 
+#define IP_IN_IP_L3_PROTO 2654
+#define IP_IN_IP_OUTER_HEADER_SIZE 20
+
 struct rpc_package_t {
     __u32 rpc_type; // 4
 	__u32 phase; // 8
@@ -505,10 +508,16 @@ static __always_inline bool is_http2_preface(const char* buf, __u32 buf_size) {
 
 __maybe_unused static __always_inline __u64 read_conn_tuple_skb(struct __sk_buff *skb, skb_info_t *info, conn_tuple_t *tup) {
     bpf_memset(info, 0, sizeof(skb_info_t));
-    info->data_off = ETH_HLEN;
+//    info->data_off = ETH_HLEN;
 
     __u16 l3_proto = load_half(skb, offsetof(struct ethhdr, h_proto));
-    info->data_end = ETH_HLEN;
+//    info->data_end = ETH_HLEN;
+    if (l3_proto == IP_IN_IP_L3_PROTO) {
+        l3_proto = ETH_P_IP;
+    } else {
+        info->data_off = ETH_HLEN;
+        info->data_end = ETH_HLEN;
+    }
     __u8 l4_proto = 0;
     switch (l3_proto) {
     case ETH_P_IP:

diff --git a/ebpf/plugins/http/protocols/http/http.h b/ebpf/plugins/http/protocols/http/http.h
@@ -202,9 +202,9 @@ static __always_inline void read_http_info(struct __sk_buff *skb, conn_tuple_t *
     // Logical processing based on the phase.
     switch (phase) {
         case HTTP_REQUEST: {
-            if (bpf_map_lookup_elem(&filter_map, &conn_key.srcIP) == NULL) {
-                return;
-            }
+//            if (bpf_map_lookup_elem(&filter_map, &conn_key.srcIP) == NULL) {
+//                return;
+//            }
 
             if (method == HTTP_METHOD_UNKNOWN) {
                 return;

diff --git a/ebpf/plugins/netfilter/main.c b/ebpf/plugins/netfilter/main.c
@@ -83,9 +83,9 @@ int kretprobe_nf_nat_setup_info(uint ret) {
         return 0;
     }
     // ignore same ip
-//    if (originTuple.dst.u3.ip == replyTuple.src.u3.ip) {
-//        return 0;
-//    }
+    if (originTuple.dst.u3.ip == replyTuple.src.u3.ip) {
+        return 0;
+    }
     struct nf_tuple conn_ev = {
         .sport = bpf_ntohs(replyTuple.src.u.tcp.port),
         .dport = bpf_ntohs(replyTuple.dst.u.tcp.port),

diff --git a/ebpf/plugins/rpc/main.c b/ebpf/plugins/rpc/main.c
@@ -127,11 +127,11 @@ int rpc__filter_package(struct __sk_buff *skb)
         pkg.pid = pid_info->pid;
     }
     if (pkg.phase == P_REQUEST) {
-        __u32 ip;
-        ip = __get_target_ip();
-        if (ip != 0 && ip != pkg.srcIP) {
-            return 0;
-        }
+//        __u32 ip;
+//        ip = __get_target_ip();
+//        if (ip != 0 && ip != pkg.srcIP) {
+//            return 0;
+//        }
         sock_key req_conn = {0};
         req_conn.srcIP = pkg.srcIP;
         req_conn.dstIP = pkg.dstIP;

diff --git a/pkg/plugins/kprobe/kprobe.go b/pkg/plugins/kprobe/kprobe.go
@@ -31,7 +31,7 @@ type provider struct {
 func (p *provider) Init(ctx servicehub.Context) error {
 	p.kprobeController = controller.NewController()
 	p.netLinks = make(map[int]NeighLink)
-	neighs, err := getAllVethes()
+	neighs, err := GetAllVethes()
 	if err != nil {
 		return err
 	}

diff --git a/pkg/plugins/kprobe/kprobesysctl/ebpf_kprobe_sysctl.go b/pkg/plugins/kprobe/kprobesysctl/ebpf_kprobe_sysctl.go
@@ -284,23 +284,24 @@ func makeServiceNodeMetric(pod *corev1.Pod) *metric.Metric {
 		Name:        "application_service_node",
 		Timestamp:   time.Now().UnixNano(),
 		Tags: map[string]string{
-			"_meta":               "true",
-			"_metric_scope":       "micro_service",
-			"_metric_scope_id":    pod.Annotations["msp.erda.cloud/terminus_key"],
-			"application_id":      pod.Labels["DICE_DEPLOYMENT_ID"],
-			"application_name":    pod.Labels["DICE_APPLICATION_NAME"],
-			"cluster_name":        pod.Labels["DICE_CLUSTER_NAME"],
-			"env_id":              pod.Annotations["msp.erda.cloud/terminus_key"],
-			"host_ip":             pod.Status.HostIP,
-			"host":                pod.Spec.NodeName,
-			"instance_id":         string(pod.UID),
-			"org_id":              pod.Labels["DICE_ORG_ID"],
-			"org_name":            pod.Labels["DICE_ORG_NAME"],
-			"project_id":          pod.Labels["DICE_PROJECT_ID"],
-			"project_name":        pod.Labels["DICE_PROJECT_NAME"],
-			"runtime_id":          pod.Labels["DICE_RUNTIME_ID"],
-			"runtime_name":        pod.Annotations["msp.erda.cloud/runtime_name"],
-			"service_id":          pod.Labels["DICE_SERVICE"],
+			"_meta":            "true",
+			"_metric_scope":    "micro_service",
+			"_metric_scope_id": pod.Annotations["msp.erda.cloud/terminus_key"],
+			"application_id":   pod.Labels["DICE_DEPLOYMENT_ID"],
+			"application_name": pod.Labels["DICE_APPLICATION_NAME"],
+			"cluster_name":     pod.Labels["DICE_CLUSTER_NAME"],
+			"env_id":           pod.Annotations["msp.erda.cloud/terminus_key"],
+			"host_ip":          pod.Status.HostIP,
+			"host":             pod.Spec.NodeName,
+			"instance_id":      string(pod.UID),
+			"org_id":           pod.Labels["DICE_ORG_ID"],
+			"org_name":         pod.Labels["DICE_ORG_NAME"],
+			"project_id":       pod.Labels["DICE_PROJECT_ID"],
+			"project_name":     pod.Labels["DICE_PROJECT_NAME"],
+			"runtime_id":       pod.Labels["DICE_RUNTIME_ID"],
+			"runtime_name":     pod.Annotations["msp.erda.cloud/runtime_name"],
+			//"service_id":          pod.Labels["DICE_SERVICE"],
+			"service_id":          fmt.Sprintf("%s_%s_%s", pod.Annotations["msp.erda.cloud/application_id"], pod.Annotations["msp.erda.cloud/runtime_name"], pod.Annotations["msp.erda.cloud/service_name"]),
 			"service_instance_id": string(pod.UID),
 			"service_ip":          pod.Status.PodIP,
 			"service_name":        pod.Labels["DICE_SERVICE_NAME"],

diff --git a/pkg/plugins/kprobe/netlink_manager.go b/pkg/plugins/kprobe/netlink_manager.go
@@ -2,7 +2,6 @@ package kprobe
 
 import (
 	"github.com/vishvananda/netlink"
-	"golang.org/x/sys/unix"
 	"net"
 )
 
@@ -23,68 +22,89 @@ type NeighLinkEvent struct {
 	NeighLink
 }
 
-func getAllVethes() ([]NeighLink, error) {
+// GetAllVethes get all bridge veth like docker0, cni0 instead of vethxxx, in this way, we can mount fewer socket network cards.
+func GetAllVethes() ([]NeighLink, error) {
 	links, err := netlink.LinkList()
 	if err != nil {
 		return nil, err
 	}
-	targetLinks := make([]netlink.Link, 0)
+	//targetLinks := make([]netlink.Link, 0)
+	ipipLinks := make([]NeighLink, 0)
+	bridgeLinks := make([]NeighLink, 0)
 	for _, link := range links {
-		if link.Type() == "bridge" {
-			targetLinks = append(targetLinks, link)
-		}
-		if link.Type() == "veth" {
-			targetLinks = append(targetLinks, link)
-		}
-	}
-	ifs, err := net.Interfaces()
-	if err != nil {
-		return nil, err
-	}
-	ans := make([]NeighLink, 0)
-	for _, l := range targetLinks {
-		neighs, err := netlink.NeighList(l.Attrs().Index, unix.AF_INET)
-		if err != nil {
-			return nil, err
-		}
-		// veth bind neigh ip, like vethvethfb8d1967, vethXXX and so on
-		if len(neighs) == 1 && l.Type() == "veth" {
-			ans = append(ans, NeighLink{
-				Neigh: neighs[0],
-				Link:  l,
+		if link.Type() == "ipip" {
+			ipipLinks = append(ipipLinks, NeighLink{
+				Neigh: netlink.Neigh{
+					IP: net.IP{0, 0, 0, 0},
+				},
+				Link: link,
 			})
-			continue
+			break
 		}
-		// bridge bind veth neigh ips, like calie9c9ef7ca49, caliXXX binded to docker0
-		for _, neigh := range neighs {
-			for _, iface := range ifs {
-				link, err := netlink.LinkByName(iface.Name)
-				if err != nil {
-					continue
-				}
-				if link.Type() == "veth" {
-					neighBr, errBr := netlink.NeighList(link.Attrs().Index, int(unix.AF_BRIDGE))
-					if errBr != nil {
-						continue
-					}
-					for _, neighB := range neighBr {
-						if neighB.HardwareAddr.String() == neigh.HardwareAddr.String() {
-							ans = append(ans, NeighLink{
-								Neigh: neigh,
-								Link:  link,
-							})
-							break
-						}
-					}
-				}
-			}
+		if link.Type() == "bridge" {
+			bridgeLinks = append(bridgeLinks, NeighLink{
+				Neigh: netlink.Neigh{
+					IP: net.IP{0, 0, 0, 0},
+				},
+				Link: link,
+			})
 		}
+		//if link.Type() == "veth" {
+		//	targetLinks = append(targetLinks, link)
+		//}
+	}
+	if len(ipipLinks) > 0 {
+		return ipipLinks, nil
 	}
-	return ans, nil
+	//ifs, err := net.Interfaces()
+	//if err != nil {
+	//	return nil, err
+	//}
+	//ans := make([]NeighLink, 0)
+	//for _, l := range targetLinks {
+	//	neighs, err := netlink.NeighList(l.Attrs().Index, unix.AF_INET)
+	//	if err != nil {
+	//		return nil, err
+	//	}
+	//	// veth bind neigh ip, like vethvethfb8d1967, vethXXX and so on
+	//	if len(neighs) == 1 && l.Type() == "bridge" {
+	//		ans = append(ans, NeighLink{
+	//			Neigh: neighs[0],
+	//			Link:  l,
+	//		})
+	//		continue
+	//	}
+	//	fmt.Printf("neighs: %+v\n", neighs)
+	//	// bridge bind veth neigh ips, like calie9c9ef7ca49, caliXXX binded to docker0
+	//	for _, neigh := range neighs {
+	//		for _, iface := range ifs {
+	//			link, err := netlink.LinkByName(iface.Name)
+	//			if err != nil {
+	//				continue
+	//			}
+	//			if link.Type() == "bridge" {
+	//				neighBr, errBr := netlink.NeighList(link.Attrs().Index, int(unix.AF_BRIDGE))
+	//				if errBr != nil {
+	//					continue
+	//				}
+	//				for _, neighB := range neighBr {
+	//					if neighB.HardwareAddr.String() == neigh.HardwareAddr.String() {
+	//						ans = append(ans, NeighLink{
+	//							Neigh: neigh,
+	//							Link:  link,
+	//						})
+	//						break
+	//					}
+	//				}
+	//			}
+	//		}
+	//	}
+	//}
+	return bridgeLinks, nil
 }
 
 func (p *provider) getVethesDiff() (added []NeighLink, removed []NeighLink, err error) {
-	neighs, err := getAllVethes()
+	neighs, err := GetAllVethes()
 	if err != nil {
 		return nil, nil, err
 	}

diff --git a/pkg/plugins/protocols/http/meta/provider.go b/pkg/plugins/protocols/http/meta/provider.go
@@ -91,7 +91,8 @@ func (p *provider) Convert(m *ebpf.Metric) *metric.Metric {
 		output.Tags["source_runtime_name"] = sourcePod.Annotations["msp.erda.cloud/runtime_name"]
 		//output.Tags["source_service_id"] = fmt.Sprintf("%s_%s_%s",
 		//	sourcePod.Labels["DICE_APPLICATION_ID"], sourcePod.Annotations["msp.erda.cloud/runtime_name"], sourcePod.Labels["DICE_SERVICE_NAME"])
-		output.Tags["source_service_id"] = sourcePod.Annotations["msp.erda.cloud/service_name"]
+		//output.Tags["source_service_id"] = sourcePod.Annotations["msp.erda.cloud/service_name"]
+		output.Tags["source_service_id"] = fmt.Sprintf("%s_%s_%s", sourcePod.Annotations["msp.erda.cloud/application_id"], sourcePod.Annotations["msp.erda.cloud/runtime_name"], sourcePod.Annotations["msp.erda.cloud/service_name"])
 		output.Tags["source_service_instance_id"] = string(sourcePod.UID)
 		output.Tags["source_service_name"] = sourcePod.Annotations["msp.erda.cloud/service_name"]
 		output.Tags["source_terminus_key"] = sourcePod.Annotations["msp.erda.cloud/terminus_key"]
@@ -146,7 +147,8 @@ func (p *provider) Convert(m *ebpf.Metric) *metric.Metric {
 		output.Tags["target_runtime_name"] = t.Annotations["msp.erda.cloud/runtime_name"]
 		//output.Tags["target_service_id"] = fmt.Sprintf("%s_%s_%s",
 		//	t.Labels["DICE_APPLICATION_ID"], t.Annotations["msp.erda.cloud/runtime_name"], t.Labels["DICE_SERVICE_NAME"])
-		output.Tags["target_service_id"] = t.Annotations["msp.erda.cloud/service_name"]
+		//output.Tags["target_service_id"] = t.Annotations["msp.erda.cloud/service_name"]
+		output.Tags["target_service_id"] = fmt.Sprintf("%s_%s_%s", t.Annotations["msp.erda.cloud/application_id"], t.Annotations["msp.erda.cloud/runtime_name"], t.Annotations["msp.erda.cloud/service_name"])
 		output.Tags["target_service_instance_id"] = string(t.UID)
 		output.Tags["target_service_name"] = t.Annotations["msp.erda.cloud/service_name"]
 		output.Tags["target_terminus_key"] = t.Annotations["msp.erda.cloud/terminus_key"]

diff --git a/pkg/plugins/protocols/kafka/kafka.go b/pkg/plugins/protocols/kafka/kafka.go
@@ -147,7 +147,8 @@ func (p *provider) convert2Metric(ev Event) *metric.Metric {
 		m.Tags["source_runtime_name"] = sourcePod.Annotations["msp.erda.cloud/runtime_name"]
 		//output.Tags["source_service_id"] = fmt.Sprintf("%s_%s_%s",
 		//	sourcePod.Labels["DICE_APPLICATION_ID"], sourcePod.Annotations["msp.erda.cloud/runtime_name"], sourcePod.Labels["DICE_SERVICE_NAME"])
-		m.Tags["source_service_id"] = sourcePod.Annotations["msp.erda.cloud/service_name"]
+		//m.Tags["source_service_id"] = sourcePod.Annotations["msp.erda.cloud/service_name"]
+		m.Tags["source_service_id"] = fmt.Sprintf("%s_%s_%s", sourcePod.Annotations["msp.erda.cloud/application_id"], sourcePod.Annotations["msp.erda.cloud/runtime_name"], sourcePod.Annotations["msp.erda.cloud/service_name"])
 		m.Tags["source_service_instance_id"] = string(sourcePod.UID)
 		m.Tags["source_service_name"] = sourcePod.Annotations["msp.erda.cloud/service_name"]
 		m.Tags["source_terminus_key"] = sourcePod.Annotations["msp.erda.cloud/terminus_key"]
@@ -192,7 +193,8 @@ func (p *provider) convert2Metric(ev Event) *metric.Metric {
 		m.Tags["target_runtime_name"] = t.Annotations["msp.erda.cloud/runtime_name"]
 		//output.Tags["target_service_id"] = fmt.Sprintf("%s_%s_%s",
 		//	t.Labels["DICE_APPLICATION_ID"], t.Annotations["msp.erda.cloud/runtime_name"], t.Labels["DICE_SERVICE_NAME"])
-		m.Tags["target_service_id"] = t.Annotations["msp.erda.cloud/service_name"]
+		//m.Tags["target_service_id"] = t.Annotations["msp.erda.cloud/service_name"]
+		m.Tags["target_service_id"] = fmt.Sprintf("%s_%s_%s", t.Annotations["msp.erda.cloud/application_id"], t.Annotations["msp.erda.cloud/runtime_name"], t.Annotations["msp.erda.cloud/service_name"])
 		m.Tags["target_service_instance_id"] = string(t.UID)
 		m.Tags["target_service_name"] = t.Annotations["msp.erda.cloud/service_name"]
 		m.Tags["target_terminus_key"] = t.Annotations["msp.erda.cloud/terminus_key"]

diff --git a/pkg/plugins/protocols/rpc/rpc.go b/pkg/plugins/protocols/rpc/rpc.go
@@ -226,7 +226,8 @@ func (p *provider) convertRpc2Metric(m *rpcebpf.Metric) metric.Metric {
 		res.Tags["source_runtime_id"] = sourcePod.Labels["DICE_RUNTIME_ID"]
 		res.Tags["source_runtime_name"] = sourcePod.Annotations["msp.erda.cloud/runtime_name"]
 		//res.Tags["source_service_id"] = fmt.Sprintf("%s_%s_%s", sourcePod.Labels["DICE_APPLICATION_ID"], sourcePod.Annotations["msp.erda.cloud/runtime_name"], sourcePod.Labels["DICE_SERVICE_NAME"])
-		res.Tags["source_service_id"] = sourcePod.Annotations["msp.erda.cloud/service_name"]
+		//res.Tags["source_service_id"] = sourcePod.Annotations["msp.erda.cloud/service_name"]
+		res.Tags["source_service_id"] = fmt.Sprintf("%s_%s_%s", sourcePod.Annotations["msp.erda.cloud/application_id"], sourcePod.Annotations["msp.erda.cloud/runtime_name"], sourcePod.Annotations["msp.erda.cloud/service_name"])
 		res.Tags["source_service_name"] = sourcePod.Annotations["msp.erda.cloud/service_name"]
 		res.Tags["source_workspace"] = sourcePod.Annotations["msp.erda.cloud/workspace"]
 		res.Tags["source_terminus_key"] = sourcePod.Annotations["msp.erda.cloud/terminus_key"]
@@ -253,7 +254,8 @@ func (p *provider) convertRpc2Metric(m *rpcebpf.Metric) metric.Metric {
 		res.Tags["target_runtime_id"] = targetPod.Labels["DICE_RUNTIME_ID"]
 		res.Tags["target_runtime_name"] = targetPod.Annotations["msp.erda.cloud/runtime_name"]
 		//res.Tags["target_service_id"] = fmt.Sprintf("%s_%s_%s", targetPod.Labels["DICE_APPLICATION_ID"], targetPod.Annotations["msp.erda.cloud/runtime_name"], targetPod.Labels["DICE_SERVICE_NAME"])
-		res.Tags["target_service_id"] = targetPod.Annotations["msp.erda.cloud/service_name"]
+		//res.Tags["target_service_id"] = targetPod.Annotations["msp.erda.cloud/service_name"]
+		res.Tags["target_service_id"] = fmt.Sprintf("%s_%s_%s", targetPod.Annotations["msp.erda.cloud/application_id"], targetPod.Annotations["msp.erda.cloud/runtime_name"], targetPod.Annotations["msp.erda.cloud/service_name"])
 		res.Tags["target_service_instance_id"] = string(targetPod.UID)
 		res.Tags["target_service_name"] = targetPod.Annotations["msp.erda.cloud/service_name"]
 		res.Tags["target_terminus_key"] = targetPod.Annotations["msp.erda.cloud/terminus_key"]