feat: created tcp_long_conn e2e test

Signed-off-by: Yash Patel <yp969803@gmail.com>

Author: yp969803

bpf/kmesh/bpf2go/dualengine/kmeshcgroupskb_bpfeb.go

@@ -9,7 +9,7 @@
 
 #define LONG_CONN_THRESHOLD_TIME (5 * 1000000000ULL) // 5s
 
-volatile __u32 enable_monitoring = 0;
+volatile __u32 enable_monitoring = 1;
 
 static inline bool is_monitoring_enable()
 {

bpf/kmesh/bpf2go/dualengine/kmeshcgroupskb_bpfel.go

@@ -5,6 +5,7 @@
 #define __KMESH_BPF_ACCESS_LOG_H__
 
 #include "bpf_common.h"
+#include "config.h"
 
 // direction
 enum {

bpf/kmesh/bpf2go/dualengine/kmeshcgroupskbcompat_bpfeb.go

@@ -8,6 +8,7 @@
 #include "bpf_log.h"
 #include "bpf_common.h"
 #include "probe.h"
+#include "config.h"
 
 SEC("cgroup_skb/ingress")
 int cgroup_skb_ingress_prog(struct __sk_buff *skb)

bpf/kmesh/bpf2go/dualengine/kmeshcgroupskbcompat_bpfel.go

@@ -793,7 +793,7 @@ func TestL4Telemetry(t *testing.T) {
 						To:      localDst,
 					}
 
-					query := buildL4Query(localSrc, localDst)
+					query := buildL4Query(localSrc, localDst, "kmesh_tcp_connections_opened_total")
 					stc.Logf("prometheus query: %#v", query)
 					err := retry.Until(func() bool {
 						stc.Logf("sending call from %q to %q", deployName(localSrc), localDst.Config().Service)
@@ -819,7 +819,50 @@ func TestL4Telemetry(t *testing.T) {
 	})
 }
 
-func buildL4Query(src, dst echo.Instance) prometheus.Query {
+func TestLongConnL4Telemetry(t *testing.T) {
+	framework.NewTest(t).Run(func(tc framework.TestContext) {
+		for _, src := range apps.EnrolledToKmesh {
+			for _, dst := range apps.EnrolledToKmesh {
+				tc.NewSubTestf("from %q to %q", src.Config().Service, dst.Config().Service).Run(func(stc framework.TestContext) {
+					localDst := dst
+					localSrc := src
+					opt := echo.CallOptions{
+						Port:                    echo.Port{Name: "tcp"},
+						Scheme:                  scheme.TCP,
+						Count:                   10,
+						Timeout:                 time.Second,
+						Check:                   check.OK(),
+						To:                      localDst,
+						NewConnectionPerRequest: false,
+					}
+
+					query := buildL4Query(localSrc, localDst, "kmesh_tcp_sent_bytes_total")
+					stc.Logf("prometheus query: %#v", query)
+					err := retry.Until(func() bool {
+						stc.Logf("sending call from %q to %q", deployName(localSrc), localDst.Config().Service)
+						localSrc.CallOrFail(stc, opt)
+						reqs, err := prom.QuerySum(localSrc.Config().Cluster, query)
+						if err != nil {
+							stc.Logf("could not query for traffic from %q to %q: %v", deployName(localSrc), localDst.Config().Service, err)
+							return false
+						}
+						if reqs == 0.0 {
+							stc.Logf("found zero-valued sum for traffic from %q to %q: %v", deployName(localSrc), localDst.Config().Service, err)
+							return false
+						}
+						return true
+					}, retry.Timeout(15*time.Second), retry.BackoffDelay(1*time.Second))
+					if err != nil {
+						PromDiff(t, prom, localSrc.Config().Cluster, query)
+						stc.Errorf("could not validate L4 telemetry for %q to %q: %v", deployName(localSrc), localDst.Config().Service, err)
+					}
+				})
+			}
+		}
+	})
+}
+
+func buildL4Query(src, dst echo.Instance, metricName string) prometheus.Query {
 	query := prometheus.Query{}
 
 	srcns := src.NamespaceName()
@@ -849,7 +892,7 @@ func buildL4Query(src, dst echo.Instance) prometheus.Query {
 		"source_cluster":                 "Kubernetes",
 	}
 
-	query.Metric = "kmesh_tcp_connections_opened_total"
+	query.Metric = metricName
 	query.Labels = labels
 
 	return query
@@ -1150,7 +1193,6 @@ spec:
 							}
 						})
 				})
-
 			}
 		})
 }

bpf/kmesh/probes/probe.h

@@ -299,3 +299,4 @@ if [[ -n "${CLEANUP_REGISTRY}" ]]; then
 fi
 
 rm -rf "${TMP}"
+