Repository: kmesh-net/kmesh

Issue:
values is not updated in tcp_conn map
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!
-->

**What happened**:

**What you expected to happen**:

**How to reproduce it (as minimally and precisely as possible)**:

**Anything else we need to know?**:

**Environment**:

- Kmesh version:
- Kmesh mode(kmesh has `Kernel-Native Mode` and `Duel-Engine Mode`):
- Istio version:
- Kernel version:
- Others:


Discussion:
@yp969803: /assign

Solution patch:
diff --git a/pkg/controller/telemetry/metric.go b/pkg/controller/telemetry/metric.go
index d633fb65..98354355 100644
--- a/pkg/controller/telemetry/metric.go
+++ b/pkg/controller/telemetry/metric.go
@@ -516,19 +516,19 @@ func (m *MetricController) Run(ctx context.Context, mapOfTcpInfo *ebpf.Map) {
 				OutputAccesslog(data, tcpConns[data.conSrcDstInfo], accesslog)
 			}
 
-			if data.state == TCP_CLOSTED {
-				delete(tcpConns, data.conSrcDstInfo)
-			}
-
 			m.mutex.Lock()
 			if m.EnableWorkloadMetric.Load() {
-				m.updateWorkloadMetricCache(data, workloadLabels, tcpConns)
+				m.updateWorkloadMetricCache(data, workloadLabels, tcpConns[data.conSrcDstInfo])
 			}
-			m.updateServiceMetricCache(data, serviceLabels, tcpConns)
+			m.updateServiceMetricCache(data, serviceLabels, tcpConns[data.conSrcDstInfo])
 			if m.EnableConnectionMetric.Load() && data.duration > LONG_CONN_METRIC_THRESHOLD {
 				m.updateConnectionMetricCache(data, tcpConns[data.conSrcDstInfo], connectionLabels)
 			}
 			m.mutex.Unlock()
+
+			if data.state == TCP_CLOSTED {
+				delete(tcpConns, data.conSrcDstInfo)
+			}
 		}
 	}
 }
@@ -577,6 +577,7 @@ func buildV4Metric(buf *bytes.Buffer, tcpConns map[connectionSrcDst]connMetric)
 		cm.totalRetrans = connectData.statistics.Retransmits
 		cm.packetLost = connectData.statistics.LostPackets
 		cm.totalReports++
+		tcpConns[data.conSrcDstInfo] = cm
 	} else {
 		tcpConns[data.conSrcDstInfo] = connMetric{
 			receivedBytes: connectData.ReceivedBytes,
@@ -632,6 +633,7 @@ func buildV6Metric(buf *bytes.Buffer, tcpConns map[connectionSrcDst]connMetric)
 		cm.totalRetrans = connectData.statistics.Retransmits
 		cm.packetLost = connectData.statistics.LostPackets
 		cm.totalReports++
+		tcpConns[data.conSrcDstInfo] = cm
 	} else {
 		tcpConns[data.conSrcDstInfo] = connMetric{
 			receivedBytes: connectData.ReceivedBytes,
@@ -840,10 +842,10 @@ func buildPrincipal(workload *workloadapi.Workload) string {
 	return DEFAULT_UNKNOWN
 }
 
-func (m *MetricController) updateWorkloadMetricCache(data requestMetric, labels workloadMetricLabels, tcpConns map[connectionSrcDst]connMetric) {
+func (m *MetricController) updateWorkloadMetricCache(data requestMetric, labels workloadMetricLabels, metric connMetric) {
 	v, ok := m.workloadMetricCache[labels]
 	if ok {
-		if data.state == TCP_ESTABLISHED && tcpConns[data.conSrcDstInfo].totalReports == 1 {
+		if data.state == TCP_ESTABLISHED && metric.totalReports == 1 {
 			v.WorkloadConnOpened = v.WorkloadConnOpened + 1
 		}
 		if data.state == TCP_CLOSTED {
@@ -858,7 +860,7 @@ func (m *MetricController) updateWorkloadMetricCache(data requestMetric, labels
 		v.WorkloadConnPacketLost = v.WorkloadConnPacketLost + float64(data.packetLost)
 	} else {
 		newWorkloadMetricInfo := workloadMetricInfo{}
-		if data.state == TCP_ESTABLISHED && tcpConns[data.conSrcDstInfo].totalReports == 1 {
+		if data.state == TCP_ESTABLISHED && metric.totalReports == 1 {
 			newWorkloadMetricInfo.WorkloadConnOpened = 1
 		}
 		if data.state == TCP_CLOSTED {
@@ -875,10 +877,10 @@ func (m *MetricController) updateWorkloadMetricCache(data requestMetric, labels
 	}
 }
 
-func (m *MetricController) updateServiceMetricCache(data requestMetric, labels serviceMetricLabels, tcpConns map[connectionSrcDst]connMetric) {
+func (m *MetricController) updateServiceMetricCache(data requestMetric, labels serviceMetricLabels, metric connMetric) {
 	v, ok := m.serviceMetricCache[labels]
 	if ok {
-		if data.state == TCP_ESTABLISHED && tcpConns[data.conSrcDstInfo].totalReports == 1 {
+		if data.state == TCP_ESTABLISHED && metric.totalReports == 1 {
 			v.ServiceConnOpened = v.ServiceConnOpened + 1
 		}
 		if data.state == TCP_CLOSTED {
@@ -891,7 +893,7 @@ func (m *MetricController) updateServiceMetricCache(data requestMetric, labels s
 		v.ServiceConnSentBytes = v.ServiceConnSentBytes + float64(data.sentBytes)
 	} else {
 		newServiceMetricInfo := serviceMetricInfo{}
-		if data.state == TCP_ESTABLISHED && tcpConns[data.conSrcDstInfo].totalReports == 1 {
+		if data.state == TCP_ESTABLISHED && metric.totalReports == 1 {
 			newServiceMetricInfo.ServiceConnOpened = 1
 		}
 		if data.state == TCP_CLOSTED {



Test patch:
diff --git a/pkg/controller/telemetry/metric_test.go b/pkg/controller/telemetry/metric_test.go
index 928a78f9..263699a7 100644
--- a/pkg/controller/telemetry/metric_test.go
+++ b/pkg/controller/telemetry/metric_test.go
@@ -17,6 +17,7 @@
 package telemetry
 
 import (
+	"bytes"
 	"context"
 	"net"
 	"reflect"
@@ -306,7 +307,7 @@ func TestBuildMetricsToPrometheus(t *testing.T) {
 				workloadMetricCache: map[workloadMetricLabels]*workloadMetricInfo{},
 				serviceMetricCache:  map[serviceMetricLabels]*serviceMetricInfo{},
 			}
-			m.updateWorkloadMetricCache(tt.args.data, tt.args.labels, tt.args.tcpConns)
+			m.updateWorkloadMetricCache(tt.args.data, tt.args.labels, tt.args.tcpConns[tt.args.data.conSrcDstInfo])
 			assert.Equal(t, m.workloadMetricCache[tt.args.labels].WorkloadConnClosed, tt.want[0])
 			assert.Equal(t, m.workloadMetricCache[tt.args.labels].WorkloadConnOpened, tt.want[1])
 			assert.Equal(t, m.workloadMetricCache[tt.args.labels].WorkloadConnReceivedBytes, tt.want[2])
@@ -440,7 +441,7 @@ func TestBuildServiceMetricsToPrometheus(t *testing.T) {
 				workloadMetricCache: map[workloadMetricLabels]*workloadMetricInfo{},
 				serviceMetricCache:  map[serviceMetricLabels]*serviceMetricInfo{},
 			}
-			m.updateServiceMetricCache(tt.args.data, tt.args.labels, tt.args.tcpConns)
+			m.updateServiceMetricCache(tt.args.data, tt.args.labels, tt.args.tcpConns[tt.args.data.conSrcDstInfo])
 			assert.Equal(t, m.serviceMetricCache[tt.args.labels].ServiceConnClosed, tt.want[0])
 			assert.Equal(t, m.serviceMetricCache[tt.args.labels].ServiceConnOpened, tt.want[1])
 			assert.Equal(t, m.serviceMetricCache[tt.args.labels].ServiceConnReceivedBytes, tt.want[2])
@@ -1677,3 +1678,96 @@ func TestMetricController_updatePrometheusMetric(t *testing.T) {
 		})
 	}
 }
+
+func TestBuildV4Metric(t *testing.T) {
+	buff := bytes.NewBuffer([]byte{10, 244, 1, 13, 10, 244, 1, 12, 34, 208, 144, 31, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+		0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 10, 96, 46, 224, 144, 31, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+		0, 3, 0, 0, 0, 147, 0, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 1, 0, 0, 0, 167, 122, 203, 84, 2, 0, 0, 0, 153, 163,
+		210, 202, 232, 184, 0, 0, 64, 30, 158, 31, 235, 184, 0, 0, 0, 0, 0, 0, 150, 158, 0, 0, 19, 0, 0, 0, 0, 0,
+		0, 0, 0, 0, 0, 0, 0, 0, 0, 0})
+	data := requestMetric{
+		conSrcDstInfo: connectionSrcDst{
+			src:     [4]uint32{218231818, 0, 0, 0},
+			dst:     [4]uint32{201454602, 0, 0, 0},
+			srcPort: 53282,
+			dstPort: 8080,
+		},
+		origDstAddr:    [4]uint32{3761135626, 0, 0, 0},
+		origDstPort:    8080,
+		direction:      2,
+		receivedBytes:  147,
+		sentBytes:      3,
+		state:          1,
+		success:        1,
+		duration:       10012555943,
+		startTime:      203309974725529,
+		lastReportTime: 203319987281472,
+		srtt:           40598,
+		minRtt:         19,
+		totalRetrans:   0,
+		packetLost:     0,
+	}
+
+	tests := []struct {
+		name          string
+		tcpConns      map[connectionSrcDst]connMetric
+		newConnMetric connMetric
+	}{
+		{
+			name:     "v4 metric test, tcpConn is empty",
+			tcpConns: map[connectionSrcDst]connMetric{},
+			newConnMetric: connMetric{
+				receivedBytes: 147,
+				sentBytes:     3,
+				packetLost:    0,
+				totalRetrans:  0,
+				totalReports:  1,
+			},
+		},
+		{
+			name: "v4 metric test, tcpConns in not empty",
+			tcpConns: map[connectionSrcDst]connMetric{
+				data.conSrcDstInfo: {
+					receivedBytes: 1,
+					sentBytes:     1,
+					packetLost:    0,
+					totalRetrans:  0,
+					totalReports:  1,
+				},
+			},
+			newConnMetric: connMetric{
+				receivedBytes: 147,
+				sentBytes:     3,
+				packetLost:    0,
+				totalRetrans:  0,
+				totalReports:  2,
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			copiedBytes := make([]byte, len(buff.Bytes()))
+			copy(copiedBytes, buff.Bytes())
+
+			copiedBuff := bytes.NewBuffer(copiedBytes)
+			prevTcpConns := make(map[connectionSrcDst]connMetric)
+			for k, v := range tt.tcpConns {
+				prevTcpConns[k] = v
+			}
+
+			got, err := buildV4Metric(copiedBuff, tt.tcpConns)
+			if err != nil {
+				t.Errorf("buildV4Metric() error = %v", err)
+			}
+			assert.Equal(t, tt.newConnMetric, tt.tcpConns[data.conSrcDstInfo])
+
+			temp := data
+			temp.sentBytes = temp.sentBytes - prevTcpConns[temp.conSrcDstInfo].sentBytes
+			temp.receivedBytes = temp.receivedBytes - prevTcpConns[temp.conSrcDstInfo].receivedBytes
+			temp.packetLost = temp.packetLost - prevTcpConns[temp.conSrcDstInfo].packetLost
+			temp.totalRetrans = temp.totalRetrans - prevTcpConns[temp.conSrcDstInfo].totalRetrans
+			assert.Equal(t, temp, got)
+		})
+	}
+}
diff --git a/test/e2e/baseline_test.go b/test/e2e/baseline_test.go
index e683565d..1101c5ff 100644
--- a/test/e2e/baseline_test.go
+++ b/test/e2e/baseline_test.go
@@ -849,7 +849,7 @@ func buildL4Query(src, dst echo.Instance) prometheus.Query {
 		"source_cluster":                 "Kubernetes",
 	}
 
-	query.Metric = "kmesh_tcp_connections_opened_total"
+	query.Metric = "kmesh_tcp_sent_bytes_total"
 	query.Labels = labels
 
 	return query