Repository: kmesh-net/kmesh

Issue:
New prometheus-metric for long conn
<!-- Please only use this template for submitting enhancement requests -->

**What would you like to be added**:
fixes: #1211 

This issue is to  be done after the merge of #1249
A new prometheus-metric which will contain connection specific infomation, and updates at regular intervals, for long conn


**Why is this needed**:
Increase observablity

Discussion:

Solution patch:
diff --git a/ctl/monitoring/monitoring.go b/ctl/monitoring/monitoring.go
index 2fbd12b1..7105b595 100644
--- a/ctl/monitoring/monitoring.go
+++ b/ctl/monitoring/monitoring.go
@@ -33,9 +33,18 @@ import (
 )
 
 const (
-	patternAccesslog       = "/accesslog"
-	patternMonitoring      = "/monitoring"
-	patternWorkloadMetrics = "/workload_metrics"
+	patternAccesslog         = "/accesslog"
+	patternMonitoring        = "/monitoring"
+	patternWorkloadMetrics   = "/workload_metrics"
+	patternConnectionMetrics = "/connection_metrics"
+)
+
+// Different types of monitoring
+const (
+	MONITORING = "monitoring"
+	ACCESSLOG  = "accesslog"
+	WORKLOAD   = "workload metrics"
+	CONNECTION = "connection metrics"
 )
 
 var log = logger.NewLoggerScope("kmeshctl/monitoring")
@@ -53,6 +62,9 @@ kmeshctl monitoring <kmesh-daemon-pod> --all enable/disable
 # Enable/Disable workload granularity metrics:
 kmeshctl monitoring <kmesh-daemon-pod> --workloadMetrics enable/disable
 
+# Enable/Disable connection granularity metrics:
+kmeshctl monitoring <kmesh-daemon-pod> --connectionMetrics enable/disable
+
 # If you want to change the monitoring functionality of all kmesh daemons in the cluster
 # Enable/Disable Kmesh's accesslog in each node:
 kmeshctl monitoring --accesslog enable/disable
@@ -60,7 +72,10 @@ kmeshctl monitoring --accesslog enable/disable
 # Enable/Disable workload granularity metrics in each node:
 kmeshctl monitoring --workloadMetrics enable/disable
 
-#Enable/Disable services' and workloads' metrics and accesslog generated from bpf in each node:
+# Enable/Disable connection granularity metrics in each node:
+kmeshctl monitoring --connectionMetrics enable/disable
+
+#Enable/Disable services', workloads' and 'connections' metrics and accesslog generated from bpf in each node:
 kmeshctl monitoring --all enable/disable`,
 		Args: cobra.MaximumNArgs(1),
 		Run: func(cmd *cobra.Command, args []string) {
@@ -69,7 +84,8 @@ kmeshctl monitoring --all enable/disable`,
 	}
 	cmd.Flags().String("accesslog", "", "Control accesslog enable or disable")
 	cmd.Flags().String("all", "", "Control accesslog and services' and workloads' metrics enable or disable together")
-	cmd.Flags().String("workloadMetrics", "", "Control granularity metrics enable or disable")
+	cmd.Flags().String("workloadMetrics", "", "Control workload granularity metrics enable or disable")
+	cmd.Flags().String("connectionMetrics", "", "Control connection granularity metrics enable or disable")
 	return cmd
 }
 
@@ -82,8 +98,9 @@ func ControlMonitoring(cmd *cobra.Command, args []string) {
 	accesslogFlag, _ := cmd.Flags().GetString("accesslog")
 	allFlag, _ := cmd.Flags().GetString("all")
 	workloadMetricsFlag, _ := cmd.Flags().GetString("workloadMetrics")
-	if accesslogFlag == "" && allFlag == "" && workloadMetricsFlag == "" {
-		log.Print("no parameters. Need --accesslog, --workloadMetric or --all")
+	connectionMetricsFlag, _ := cmd.Flags().GetString("connectionMetrics")
+	if accesslogFlag == "" && allFlag == "" && workloadMetricsFlag == "" && connectionMetricsFlag == "" {
+		log.Print("no parameters. Need --accesslog, --workloadMetrics, --connectionMetrics or --all")
 		return
 	}
 
@@ -91,13 +108,16 @@ func ControlMonitoring(cmd *cobra.Command, args []string) {
 	if hasKmeshPod {
 		// Processes triggers for specified kmesh daemon.
 		if allFlag != "" {
-			SetMonitoringPerKmeshDaemon(client, podName, allFlag)
+			SetObservabilityPerKmeshDaemon(client, podName, allFlag, MONITORING, patternMonitoring)
 		}
 		if accesslogFlag != "" {
-			SetAccesslogPerKmeshDaemon(client, podName, accesslogFlag)
+			SetObservabilityPerKmeshDaemon(client, podName, accesslogFlag, ACCESSLOG, patternAccesslog)
 		}
 		if workloadMetricsFlag != "" {
-			SetWorkloadMetricsPerKmeshDaemon(client, podName, workloadMetricsFlag)
+			SetObservabilityPerKmeshDaemon(client, podName, workloadMetricsFlag, WORKLOAD, patternWorkloadMetrics)
+		}
+		if connectionMetricsFlag != "" {
+			SetObservabilityPerKmeshDaemon(client, podName, connectionMetricsFlag, CONNECTION, patternConnectionMetrics)
 		}
 	} else {
 		// Perform operations on all kmesh daemons.
@@ -108,13 +128,16 @@ func ControlMonitoring(cmd *cobra.Command, args []string) {
 		}
 		for _, pod := range podList.Items {
 			if allFlag != "" {
-				SetMonitoringPerKmeshDaemon(client, pod.GetName(), allFlag)
+				SetObservabilityPerKmeshDaemon(client, pod.GetName(), allFlag, MONITORING, patternMonitoring)
 			}
 			if accesslogFlag != "" {
-				SetAccesslogPerKmeshDaemon(client, pod.GetName(), accesslogFlag)
+				SetObservabilityPerKmeshDaemon(client, pod.GetName(), accesslogFlag, ACCESSLOG, patternAccesslog)
 			}
 			if workloadMetricsFlag != "" {
-				SetWorkloadMetricsPerKmeshDaemon(client, pod.GetName(), workloadMetricsFlag)
+				SetObservabilityPerKmeshDaemon(client, pod.GetName(), workloadMetricsFlag, WORKLOAD, patternWorkloadMetrics)
+			}
+			if connectionMetricsFlag != "" {
+				SetObservabilityPerKmeshDaemon(client, pod.GetName(), connectionMetricsFlag, CONNECTION, patternConnectionMetrics)
 			}
 		}
 	}
@@ -130,12 +153,12 @@ func getKmeshDaemonPod(args []string) (string, bool) {
 	return args[0], true
 }
 
-func SetAccesslogPerKmeshDaemon(cli kube.CLIClient, podName, info string) {
-	var accesslogInfo string
+func SetObservabilityPerKmeshDaemon(cli kube.CLIClient, podName, info string, observablityType string, pattern string) {
+	var status string
 	if info == "enable" {
-		accesslogInfo = "true"
+		status = "true"
 	} else if info == "disable" {
-		accesslogInfo = "false"
+		status = "false"
 	} else {
 		log.Errorf("Error: Argument must be 'enable' or 'disable'")
 		os.Exit(1)
@@ -152,7 +175,7 @@ func SetAccesslogPerKmeshDaemon(cli kube.CLIClient, podName, info string) {
 	}
 	defer fw.Close()
 
-	url := fmt.Sprintf("http://%s%s?enable=%s", fw.Address(), patternAccesslog, accesslogInfo)
+	url := fmt.Sprintf("http://%s%s?enable=%s", fw.Address(), pattern, status)
 
 	req, err := http.NewRequest(http.MethodPost, url, nil)
 	if err != nil {
@@ -170,114 +193,19 @@ func SetAccesslogPerKmeshDaemon(cli kube.CLIClient, podName, info string) {
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		bodyBytes, readErr := io.ReadAll(resp.Body)
-		if readErr != nil {
-			log.Errorf("Error reading response body: %v", readErr)
+		if observablityType == MONITORING {
+			log.Errorf("Error: received status code %d", resp.StatusCode)
 			return
 		}
-		bodyString := string(bodyBytes)
-		if resp.StatusCode == http.StatusBadRequest && bytes.Contains(bodyBytes, []byte("Kmesh monitoring is disable, cannot enable accesslog")) {
-			log.Errorf("failed to enable accesslog: %v. Need to start Kmesh's Monitoring. Please run `kmeshctl monitoring -h` for more help.", bodyString)
-			return
-		}
-		log.Errorf("Error: received status code %d", resp.StatusCode)
-		return
-	}
-}
-
-func SetMonitoringPerKmeshDaemon(cli kube.CLIClient, podName, info string) {
-	var monitoringInfo string
-	if info == "enable" {
-		monitoringInfo = "true"
-	} else if info == "disable" {
-		monitoringInfo = "false"
-	} else {
-		log.Errorf("Error: Argument must be 'enable' or 'disable'")
-		os.Exit(1)
-	}
-
-	fw, err := utils.CreateKmeshPortForwarder(cli, podName)
-	if err != nil {
-		log.Errorf("failed to create port forwarder for Kmesh daemon pod %s: %v", podName, err)
-		os.Exit(1)
-	}
-	if err := fw.Start(); err != nil {
-		log.Errorf("failed to start port forwarder for Kmesh daemon pod %s: %v", podName, err)
-		os.Exit(1)
-	}
-	defer fw.Close()
-
-	url := fmt.Sprintf("http://%s%s?enable=%s", fw.Address(), patternMonitoring, monitoringInfo)
-
-	req, err := http.NewRequest(http.MethodPost, url, nil)
-	if err != nil {
-		log.Errorf("Error creating request: %v", err)
-		return
-	}
-
-	req.Header.Set("Content-Type", "application/json")
-	client := &http.Client{}
-	resp, err := client.Do(req)
-	if err != nil {
-		log.Errorf("failed to make HTTP request: %v", err)
-		return
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode != http.StatusOK {
-		log.Errorf("Error: received status code %d", resp.StatusCode)
-		return
-	}
-}
 
-func SetWorkloadMetricsPerKmeshDaemon(cli kube.CLIClient, podName, workloadMetricsInfo string) {
-	var info string
-	if workloadMetricsInfo == "enable" {
-		info = "true"
-	} else if workloadMetricsInfo == "disable" {
-		info = "false"
-	} else {
-		log.Errorf("Error: Argument must be 'enable' or 'disable'")
-		os.Exit(1)
-	}
-
-	fw, err := utils.CreateKmeshPortForwarder(cli, podName)
-	if err != nil {
-		log.Errorf("failed to create port forwarder for Kmesh daemon pod %s: %v", podName, err)
-		os.Exit(1)
-	}
-	if err := fw.Start(); err != nil {
-		log.Errorf("failed to start port forwarder for Kmesh daemon pod %s: %v", podName, err)
-		os.Exit(1)
-	}
-	defer fw.Close()
-
-	url := fmt.Sprintf("http://%s%s?enable=%s", fw.Address(), patternWorkloadMetrics, info)
-
-	req, err := http.NewRequest(http.MethodPost, url, nil)
-	if err != nil {
-		log.Errorf("Error creating request: %v", err)
-		return
-	}
-
-	req.Header.Set("Content-Type", "application/json")
-	client := &http.Client{}
-	resp, err := client.Do(req)
-	if err != nil {
-		log.Errorf("failed to make HTTP request: %v", err)
-		return
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode != http.StatusOK {
 		bodyBytes, readErr := io.ReadAll(resp.Body)
 		if readErr != nil {
 			log.Errorf("Error reading response body: %v", readErr)
 			return
 		}
 		bodyString := string(bodyBytes)
-		if resp.StatusCode == http.StatusBadRequest && bytes.Contains(bodyBytes, []byte("Kmesh monitoring is disable, cannot enable accesslog")) {
-			log.Errorf("failed to enable workload metrics: %v. Need to start Kmesh's Monitoring. Please run `kmeshctl monitoring -h` for more help.", bodyString)
+		if resp.StatusCode == http.StatusBadRequest && bytes.Contains(bodyBytes, []byte(fmt.Sprintf("Kmesh monitoring is disable, cannot enable %s.", observablityType))) {
+			log.Errorf("failed to enable %s: %v. Need to start Kmesh's Monitoring. Please run `kmeshctl monitoring -h` for more help.", observablityType, bodyString)
 			return
 		}
 		log.Errorf("Error: received status code %d", resp.StatusCode)
diff --git a/docs/ctl/kmeshctl_monitoring.md b/docs/ctl/kmeshctl_monitoring.md
index 6a912376..684dbce1 100644
--- a/docs/ctl/kmeshctl_monitoring.md
+++ b/docs/ctl/kmeshctl_monitoring.md
@@ -18,6 +18,9 @@ kmeshctl monitoring <kmesh-daemon-pod> --all enable/disable
 # Enable/Disable workload granularity metrics:
 kmeshctl monitoring <kmesh-daemon-pod> --workloadMetrics enable/disable
 
+# Enable/Disable connection granularity metrics:
+kmeshctl monitoring <kmesh-daemon-pod> --connectionMetrics enable/disable
+
 # If you want to change the monitoring functionality of all kmesh daemons in the cluster
 # Enable/Disable Kmesh's accesslog in each node:
 kmeshctl monitoring --accesslog enable/disable
@@ -25,17 +28,21 @@ kmeshctl monitoring --accesslog enable/disable
 # Enable/Disable workload granularity metrics in each node:
 kmeshctl monitoring --workloadMetrics enable/disable
 
-#Enable/Disable services' and workloads' metrics and accesslog generated from bpf in each node:
+# Enable/Disable connection granularity metrics in each node:
+kmeshctl monitoring --connectionMetrics enable/disable
+
+#Enable/Disable services', workloads' and 'connections' metrics and accesslog generated from bpf in each node:
 kmeshctl monitoring --all enable/disable
 ```
 
 ### Options
 
 ```
-      --accesslog string         Control accesslog enable or disable
-      --all string               Control accesslog and services' and workloads' metrics enable or disable together
-  -h, --help                     help for monitoring
-      --workloadMetrics string   Control granularity metrics enable or disable
+      --accesslog string           Control accesslog enable or disable
+      --all string                 Control accesslog and services' and workloads' metrics enable or disable together
+      --connectionMetrics string   Control connection granularity metrics enable or disable
+  -h, --help                       help for monitoring
+      --workloadMetrics string     Control workload granularity metrics enable or disable
 ```
 
 ### SEE ALSO
diff --git a/docs/proposal/tcp_long_connection_metrics.md b/docs/proposal/tcp_long_connection_metrics.md
index 19d58aec..ec2de64e 100644
--- a/docs/proposal/tcp_long_connection_metrics.md
+++ b/docs/proposal/tcp_long_connection_metrics.md
@@ -168,6 +168,58 @@ static inline void observe_on_data(struct bpf_sock *sk)
 We will update the functions of metric.go for periodic updating the workload and service metrics, also we will create a new metric for long tcp connections.
 
 ![design](./pics/tcp_long_conn_design.png)
+
+#### Exposing long connection prometheus metrics 
+
+We will expose metrics for the connections whose duration exceeds 30 seconds. Not exposing metrics for short connection as it can lead to lot of metrics and they are also not suitable for prometheus metrics because prometheus itself has a scrape interval of maximum 15s, and short-lived connections may start and end between scrapes, resulting in incomplete or misleading data. By focusing only on longer-lived connections, we ensure the metrics are stable, meaningful, and better aligned with Prometheus’s time-series data model.
+
+We can have a another component in future which reports realtime information about connections like cilium hubble. 
+
+Prometheus metrics exposed 
+
+- `kmesh_tcp_connection_sent_bytes_total` : The total number of bytes sent over established TCP connection
+
+- `kmesh_tcp_connection_received_bytes_total` : The total number of bytes received over established TCP connection
+
+- `kmesh_tcp_connection_packet_lost_total` : Total number of packets lost during transmission in a TCP connection
+
+- `kmesh_tcp_connection_retrans_total` : The total number of retransmits over established TCP connection
+
+The above metrics has following labels
+
+```
+		"reporter"
+		"start_time"
+		"source_workload"
+		"source_canonical_service"
+		"source_canonical_revision"
+		"source_workload_namespace"
+		"source_principal"
+		"source_app"
+		"source_version"
+		"source_cluster"
+		"source_address"
+		"destination_address"
+		"destination_pod_address"
+		"destination_pod_namespace"
+		"destination_pod_name"
+		"destination_service"
+		"destination_service_namespace"
+		"destination_service_name"
+		"destination_workload"
+		"destination_canonical_service"
+		"destination_canonical_revision"
+		"destination_workload_namespace"
+		"destination_principal"
+		"destination_app"
+		"destination_version"
+		"destination_cluster"
+		"request_protocol"
+		"response_flags"
+		"connection_security_policy"
+```
+
+
 #### User Stories (Optional)
 
 <!--
diff --git a/pkg/controller/telemetry/metric.go b/pkg/controller/telemetry/metric.go
index fd683632..d633fb65 100644
--- a/pkg/controller/telemetry/metric.go
+++ b/pkg/controller/telemetry/metric.go
@@ -47,6 +47,8 @@ const (
 	metricFlushInterval = 5 * time.Second
 
 	DEFAULT_UNKNOWN = "-"
+
+	LONG_CONN_METRIC_THRESHOLD = uint64(30 * time.Second)
 )
 
 var osStartTime time.Time
@@ -68,14 +70,16 @@ var TCP_STATES = map[uint32]string{
 }
 
 type MetricController struct {
-	EnableAccesslog      atomic.Bool
-	EnableMonitoring     atomic.Bool
-	EnableWorkloadMetric atomic.Bool
-	workloadCache        cache.WorkloadCache
-	serviceCache         cache.ServiceCache
-	workloadMetricCache  map[workloadMetricLabels]*workloadMetricInfo
-	serviceMetricCache   map[serviceMetricLabels]*serviceMetricInfo
-	mutex                sync.RWMutex
+	EnableAccesslog        atomic.Bool
+	EnableMonitoring       atomic.Bool
+	EnableWorkloadMetric   atomic.Bool
+	EnableConnectionMetric atomic.Bool
+	workloadCache          cache.WorkloadCache
+	serviceCache           cache.ServiceCache
+	workloadMetricCache    map[workloadMetricLabels]*workloadMetricInfo
+	serviceMetricCache     map[serviceMetricLabels]*serviceMetricInfo
+	connectionMetricCache  map[connectionMetricLabels]*connectionMetricInfo
+	mutex                  sync.RWMutex
 }
 
 type workloadMetricInfo struct {
@@ -96,6 +100,13 @@ type serviceMetricInfo struct {
 	ServiceConnFailed        float64
 }
 
+type connectionMetricInfo struct {
+	ConnSentBytes     float64
+	ConnReceivedBytes float64
+	ConnTotalRetrans  float64
+	ConnPacketLost    float64
+}
+
 type statistics struct {
 	SentBytes      uint32
 	ReceivedBytes  uint32
@@ -232,6 +243,42 @@ type serviceMetricLabels struct {
 	connectionSecurityPolicy string
 }
 
+type connectionMetricLabels struct {
+	reporter  string
+	startTime string
+
+	sourceWorkload          string
+	sourceCanonicalService  string
+	sourceCanonicalRevision string
+	sourceWorkloadNamespace string
+	sourcePrincipal         string
+	sourceApp               string
+	sourceVersion           string
+	sourceCluster           string
+	sourceAddress           string
+
+	destinationAddress           string
+	destinationPodAddress        string
+	destinationPodNamespace      string
+	destinationPodName           string
+	destinationService           string
+	destinationServiceNamespace  string
+	destinationServiceName       string
+	destinationWorkload          string
+	destinationCanonicalService  string
+	destinationCanonicalRevision string
+	destinationWorkloadNamespace string
+	destinationPrincipal         string
+	destinationApp               string
+	destinationVersion           string
+	destinationCluster           string
+
+	requestProtocol string
+	// TODO: responseFlags is not used for now
+	responseFlags            string
+	connectionSecurityPolicy string
+}
+
 func NewServiceMetricLabel() *serviceMetricLabels {
 	return &serviceMetricLabels{}
 }
@@ -240,6 +287,10 @@ func NewWorkloadMetricLabel() *workloadMetricLabels {
 	return &workloadMetricLabels{}
 }
 
+func NewConnectionMetricLabel() *connectionMetricLabels {
+	return &connectionMetricLabels{}
+}
+
 func (w *workloadMetricLabels) withSource(workload *workloadapi.Workload) *workloadMetricLabels {
 	if workload == nil {
 		return w
@@ -312,16 +363,60 @@ func (s *serviceMetricLabels) withDestination(workload *workloadapi.Workload) *s
 	return s
 }
 
+func (c *connectionMetricLabels) withSource(workload *workloadapi.Workload) *connectionMetricLabels {
+	if workload == nil {
+		return c
+	}
+	c.sourceWorkload = workload.GetWorkloadName()
+	c.sourceCanonicalService = workload.GetCanonicalName()
+	c.sourceCanonicalRevision = workload.GetCanonicalRevision()
+	c.sourceWorkloadNamespace = workload.GetNamespace()
+	c.sourceApp = workload.GetCanonicalName()
+	c.sourceVersion = workload.GetCanonicalRevision()
+	c.sourceCluster = workload.GetClusterId()
+	c.sourcePrincipal = buildPrincipal(workload)
+	return c
+}
+
+func (c *connectionMetricLabels) withDestination(workload *workloadapi.Workload) *connectionMetricLabels {
+	if workload == nil {
+		return c
+	}
+	c.destinationPodNamespace = workload.GetNamespace()
+	c.destinationPodName = workload.GetName()
+	c.destinationWorkload = workload.GetWorkloadName()
+	c.destinationCanonicalService = workload.GetCanonicalName()
+	c.destinationCanonicalRevision = workload.GetCanonicalRevision()
+	c.destinationWorkloadNamespace = workload.GetNamespace()
+	c.destinationApp = workload.GetCanonicalName()
+	c.destinationVersion = workload.GetCanonicalRevision()
+	c.destinationCluster = workload.GetClusterId()
+	c.destinationPrincipal = buildPrincipal(workload)
+	return c
+}
+
+func (c *connectionMetricLabels) withDestinationService(service *workloadapi.Service) *connectionMetricLabels {
+	if service == nil {
+		return c
+	}
+	c.destinationService = service.GetHostname()
+	c.destinationServiceName = service.GetName()
+	c.destinationServiceNamespace = service.GetNamespace()
+	return c
+}
+
 func NewMetric(workloadCache cache.WorkloadCache, serviceCache cache.ServiceCache, enableMonitoring bool) *MetricController {
 	m := &MetricController{
-		workloadCache:       workloadCache,
-		serviceCache:        serviceCache,
-		workloadMetricCache: map[workloadMetricLabels]*workloadMetricInfo{},
-		serviceMetricCache:  map[serviceMetricLabels]*serviceMetricInfo{},
+		workloadCache:         workloadCache,
+		serviceCache:          serviceCache,
+		workloadMetricCache:   map[workloadMetricLabels]*workloadMetricInfo{},
+		serviceMetricCache:    map[serviceMetricLabels]*serviceMetricInfo{},
+		connectionMetricCache: map[connectionMetricLabels]*connectionMetricInfo{},
 	}
 	m.EnableMonitoring.Store(enableMonitoring)
 	m.EnableAccesslog.Store(false)
 	m.EnableWorkloadMetric.Store(false)
+	m.EnableConnectionMetric.Store(false)
 	return m
 }
 
@@ -348,7 +443,7 @@ func (m *MetricController) Run(ctx context.Context, mapOfTcpInfo *ebpf.Map) {
 		}
 	}()
 
-	tcp_conns := make(map[connectionSrcDst]connMetric)
+	tcpConns := make(map[connectionSrcDst]connMetric)
 
 	// Register metrics to Prometheus and start Prometheus server
 	go RunPrometheusClient(ctx)
@@ -390,13 +485,13 @@ func (m *MetricController) Run(ctx context.Context, mapOfTcpInfo *ebpf.Map) {
 			buf := bytes.NewBuffer(originInfo)
 			switch connectType {
 			case constants.MSG_TYPE_IPV4:
-				data, err = buildV4Metric(buf, tcp_conns)
+				data, err = buildV4Metric(buf, tcpConns)
 				if err != nil {
 					log.Errorf("get connectionV4 info failed: %v", err)
 					continue
 				}
 			case constants.MSG_TYPE_IPV6:
-				data, err = buildV6Metric(buf, tcp_conns)
+				data, err = buildV6Metric(buf, tcpConns)
 				if err != nil {
 					log.Errorf("get connectionV6 info failed: %v", err)
 					continue
@@ -412,26 +507,33 @@ func (m *MetricController) Run(ctx context.Context, mapOfTcpInfo *ebpf.Map) {
 				workloadLabels = m.buildWorkloadMetric(&data)
 			}
 
+			connectionLabels := connectionMetricLabels{}
+			if m.EnableConnectionMetric.Load() && data.duration > LONG_CONN_METRIC_THRESHOLD {
+				connectionLabels = m.buildConnectionMetric(&data)
+			}
 			if m.EnableAccesslog.Load() {
 				// accesslogs at start of connection, at interval of 5 sec during connection lifecycle and at close of connection
-				OutputAccesslog(data, tcp_conns[data.conSrcDstInfo], accesslog)
+				OutputAccesslog(data, tcpConns[data.conSrcDstInfo], accesslog)
 			}
 
 			if data.state == TCP_CLOSTED {
-				delete(tcp_conns, data.conSrcDstInfo)
+				delete(tcpConns, data.conSrcDstInfo)
 			}
 
 			m.mutex.Lock()
 			if m.EnableWorkloadMetric.Load() {
-				m.updateWorkloadMetricCache(data, workloadLabels, tcp_conns)
+				m.updateWorkloadMetricCache(data, workloadLabels, tcpConns)
+			}
+			m.updateServiceMetricCache(data, serviceLabels, tcpConns)
+			if m.EnableConnectionMetric.Load() && data.duration > LONG_CONN_METRIC_THRESHOLD {
+				m.updateConnectionMetricCache(data, tcpConns[data.conSrcDstInfo], connectionLabels)
 			}
-			m.updateServiceMetricCache(data, serviceLabels, tcp_conns)
 			m.mutex.Unlock()
 		}
 	}
 }
 
-func buildV4Metric(buf *bytes.Buffer, tcp_conns map[connectionSrcDst]connMetric) (requestMetric, error) {
+func buildV4Metric(buf *bytes.Buffer, tcpConns map[connectionSrcDst]connMetric) (requestMetric, error) {
 	data := requestMetric{}
 	connectData := connectionDataV4{}
 
@@ -456,8 +558,8 @@ func buildV4Metric(buf *bytes.Buffer, tcp_conns map[connectionSrcDst]connMetric)
 		data.origDstPort = connectData.OriginalPort
 	}
 
-	data.sentBytes = connectData.SentBytes - tcp_conns[data.conSrcDstInfo].sentBytes
-	data.receivedBytes = connectData.ReceivedBytes - tcp_conns[data.conSrcDstInfo].receivedBytes
+	data.sentBytes = connectData.SentBytes - tcpConns[data.conSrcDstInfo].sentBytes
+	data.receivedBytes = connectData.ReceivedBytes - tcpConns[data.conSrcDstInfo].receivedBytes
 	data.state = connectData.State
 	data.success = connectData.ConnectSuccess
 	data.duration = connectData.Duration
@@ -465,10 +567,10 @@ func buildV4Metric(buf *bytes.Buffer, tcp_conns map[connectionSrcDst]connMetric)
 	data.lastReportTime = connectData.LastReportTime
 	data.srtt = connectData.statistics.SRttTime
 	data.minRtt = connectData.statistics.RttMin
-	data.totalRetrans = connectData.statistics.Retransmits - tcp_conns[data.conSrcDstInfo].totalRetrans
-	data.packetLost = connectData.statistics.LostPackets - tcp_conns[data.conSrcDstInfo].packetLost
+	data.totalRetrans = connectData.statistics.Retransmits - tcpConns[data.conSrcDstInfo].totalRetrans
+	data.packetLost = connectData.statistics.LostPackets - tcpConns[data.conSrcDstInfo].packetLost
 
-	cm, ok := tcp_conns[data.conSrcDstInfo]
+	cm, ok := tcpConns[data.conSrcDstInfo]
 	if ok {
 		cm.receivedBytes = connectData.ReceivedBytes
 		cm.sentBytes = connectData.SentBytes
@@ -476,7 +578,7 @@ func buildV4Metric(buf *bytes.Buffer, tcp_conns map[connectionSrcDst]connMetric)
 		cm.packetLost = connectData.statistics.LostPackets
 		cm.totalReports++
 	} else {
-		tcp_conns[data.conSrcDstInfo] = connMetric{
+		tcpConns[data.conSrcDstInfo] = connMetric{
 			receivedBytes: connectData.ReceivedBytes,
 			sentBytes:     connectData.SentBytes,
 			totalRetrans:  connectData.statistics.Retransmits,
@@ -488,7 +590,7 @@ func buildV4Metric(buf *bytes.Buffer, tcp_conns map[connectionSrcDst]connMetric)
 	return data, nil
 }
 
-func buildV6Metric(buf *bytes.Buffer, tcp_conns map[connectionSrcDst]connMetric) (requestMetric, error) {
+func buildV6Metric(buf *bytes.Buffer, tcpConns map[connectionSrcDst]connMetric) (requestMetric, error) {
 	data := requestMetric{}
 	connectData := connectionDataV6{}
 	if err := binary.Read(buf, binary.LittleEndian, &connectData); err != nil {
@@ -511,8 +613,8 @@ func buildV6Metric(buf *bytes.Buffer, tcp_conns map[connectionSrcDst]connMetric)
 		data.origDstPort = connectData.OriginalPort
 	}
 
-	data.sentBytes = connectData.SentBytes - tcp_conns[data.conSrcDstInfo].sentBytes
-	data.receivedBytes = connectData.ReceivedBytes - tcp_conns[data.conSrcDstInfo].receivedBytes
+	data.sentBytes = connectData.SentBytes - tcpConns[data.conSrcDstInfo].sentBytes
+	data.receivedBytes = connectData.ReceivedBytes - tcpConns[data.conSrcDstInfo].receivedBytes
 	data.state = connectData.State
 	data.success = connectData.ConnectSuccess
 	data.duration = connectData.Duration
@@ -520,10 +622,10 @@ func buildV6Metric(buf *bytes.Buffer, tcp_conns map[connectionSrcDst]connMetric)
 	data.lastReportTime = connectData.LastReportTime
 	data.srtt = connectData.statistics.SRttTime
 	data.minRtt = connectData.statistics.RttMin
-	data.totalRetrans = connectData.statistics.Retransmits - tcp_conns[data.conSrcDstInfo].totalRetrans
-	data.packetLost = connectData.statistics.LostPackets - tcp_conns[data.conSrcDstInfo].packetLost
+	data.totalRetrans = connectData.statistics.Retransmits - tcpConns[data.conSrcDstInfo].totalRetrans
+	data.packetLost = connectData.statistics.LostPackets - tcpConns[data.conSrcDstInfo].packetLost
 
-	cm, ok := tcp_conns[data.conSrcDstInfo]
+	cm, ok := tcpConns[data.conSrcDstInfo]
 	if ok {
 		cm.receivedBytes = connectData.ReceivedBytes
 		cm.sentBytes = connectData.SentBytes
@@ -531,7 +633,7 @@ func buildV6Metric(buf *bytes.Buffer, tcp_conns map[connectionSrcDst]connMetric)
 		cm.packetLost = connectData.statistics.LostPackets
 		cm.totalReports++
 	} else {
-		tcp_conns[data.conSrcDstInfo] = connMetric{
+		tcpConns[data.conSrcDstInfo] = connMetric{
 			receivedBytes: connectData.ReceivedBytes,
 			sentBytes:     connectData.SentBytes,
 			totalRetrans:  connectData.statistics.Retransmits,
@@ -675,6 +777,52 @@ func (m *MetricController) buildServiceMetric(data *requestMetric) (serviceMetri
 	return *trafficLabels, *accesslog
 }
 
+func (m *MetricController) buildConnectionMetric(data *requestMetric) connectionMetricLabels {
+	var dstAddr, srcAddr, origAddr []byte
+	for i := range data.conSrcDstInfo.dst {
+		dstAddr = binary.LittleEndian.AppendUint32(dstAddr, data.conSrcDstInfo.dst[i])
+		srcAddr = binary.LittleEndian.AppendUint32(srcAddr, data.conSrcDstInfo.src[i])
+		origAddr = binary.LittleEndian.AppendUint32(origAddr, data.origDstAddr[i])
+	}
+
+	dstWorkload, dstIP := m.getWorkloadByAddress(restoreIPv4(dstAddr))
+	srcWorkload, srcIP := m.getWorkloadByAddress(restoreIPv4(srcAddr))
+
+	if srcWorkload == nil {
+		return connectionMetricLabels{}
+	}
+
+	dstService := m.fetchOriginalService(restoreIPv4(origAddr), uint32(data.origDstPort))
+	// if dstService not found, we use the address as hostname for metrics
+	if dstService == nil {
+		dstService = &workloadapi.Service{
+			Hostname: dstIP,
+		}
+	}
+
+	trafficLabels := NewConnectionMetricLabel()
+	trafficLabels.withSource(srcWorkload).withDestination(dstWorkload).withDestinationService(dstService)
+
+	trafficLabels.destinationAddress = dstIP + ":" + fmt.Sprintf("%d", data.conSrcDstInfo.dstPort)
+	trafficLabels.sourceAddress = srcIP + ":" + fmt.Sprintf("%d", data.conSrcDstInfo.srcPort)
+	trafficLabels.destinationPodAddress = dstIP
+	trafficLabels.requestProtocol = "tcp"
+	trafficLabels.connectionSecurityPolicy = "mutual_tls"
+
+	switch data.direction {
+	case constants.INBOUND:
+		trafficLabels.reporter = "destination"
+	case constants.OUTBOUND:
+		trafficLabels.reporter = "source"
+	}
+
+	startTime := calculateUptime(osStartTime, data.startTime)
+	startTimeInfo := fmt.Sprintf("%v", startTime)
+	trafficLabels.startTime = startTimeInfo
+
+	return *trafficLabels
+}
+
 func (m *MetricController) getWorkloadByAddress(address []byte) (*workloadapi.Workload, string) {
 	networkAddr := cache.NetworkAddress{}
 	networkAddr.Address, _ = netip.AddrFromSlice(address)
@@ -692,10 +840,10 @@ func buildPrincipal(workload *workloadapi.Workload) string {
 	return DEFAULT_UNKNOWN
 }
 
-func (m *MetricController) updateWorkloadMetricCache(data requestMetric, labels workloadMetricLabels, tcp_conns map[connectionSrcDst]connMetric) {
+func (m *MetricController) updateWorkloadMetricCache(data requestMetric, labels workloadMetricLabels, tcpConns map[connectionSrcDst]connMetric) {
 	v, ok := m.workloadMetricCache[labels]
 	if ok {
-		if data.state == TCP_ESTABLISHED && tcp_conns[data.conSrcDstInfo].totalReports == 1 {
+		if data.state == TCP_ESTABLISHED && tcpConns[data.conSrcDstInfo].totalReports == 1 {
 			v.WorkloadConnOpened = v.WorkloadConnOpened + 1
 		}
 		if data.state == TCP_CLOSTED {
@@ -710,7 +858,7 @@ func (m *MetricController) updateWorkloadMetricCache(data requestMetric, labels
 		v.WorkloadConnPacketLost = v.WorkloadConnPacketLost + float64(data.packetLost)
 	} else {
 		newWorkloadMetricInfo := workloadMetricInfo{}
-		if data.state == TCP_ESTABLISHED && tcp_conns[data.conSrcDstInfo].totalReports == 1 {
+		if data.state == TCP_ESTABLISHED && tcpConns[data.conSrcDstInfo].totalReports == 1 {
 			newWorkloadMetricInfo.WorkloadConnOpened = 1
 		}
 		if data.state == TCP_CLOSTED {
@@ -727,10 +875,10 @@ func (m *MetricController) updateWorkloadMetricCache(data requestMetric, labels
 	}
 }
 
-func (m *MetricController) updateServiceMetricCache(data requestMetric, labels serviceMetricLabels, tcp_conns map[connectionSrcDst]connMetric) {
+func (m *MetricController) updateServiceMetricCache(data requestMetric, labels serviceMetricLabels, tcpConns map[connectionSrcDst]connMetric) {
 	v, ok := m.serviceMetricCache[labels]
 	if ok {
-		if data.state == TCP_ESTABLISHED && tcp_conns[data.conSrcDstInfo].totalReports == 1 {
+		if data.state == TCP_ESTABLISHED && tcpConns[data.conSrcDstInfo].totalReports == 1 {
 			v.ServiceConnOpened = v.ServiceConnOpened + 1
 		}
 		if data.state == TCP_CLOSTED {
@@ -743,7 +891,7 @@ func (m *MetricController) updateServiceMetricCache(data requestMetric, labels s
 		v.ServiceConnSentBytes = v.ServiceConnSentBytes + float64(data.sentBytes)
 	} else {
 		newServiceMetricInfo := serviceMetricInfo{}
-		if data.state == TCP_ESTABLISHED && tcp_conns[data.conSrcDstInfo].totalReports == 1 {
+		if data.state == TCP_ESTABLISHED && tcpConns[data.conSrcDstInfo].totalReports == 1 {
 			newServiceMetricInfo.ServiceConnOpened = 1
 		}
 		if data.state == TCP_CLOSTED {
@@ -758,12 +906,36 @@ func (m *MetricController) updateServiceMetricCache(data requestMetric, labels s
 	}
 }
 
+func (m *MetricController) updateConnectionMetricCache(data requestMetric, connData connMetric, labels connectionMetricLabels) {
+	v, ok := m.connectionMetricCache[labels]
+	if ok {
+		v.ConnSentBytes = v.ConnSentBytes + float64(data.sentBytes)
+		v.ConnReceivedBytes = v.ConnReceivedBytes + float64(data.receivedBytes)
+		v.ConnPacketLost = v.ConnPacketLost + float64(data.packetLost)
+		v.ConnTotalRetrans = v.ConnTotalRetrans + float64(data.totalRetrans)
+	} else {
+		newConnectionMetricInfo := connectionMetricInfo{}
+		newConnectionMetricInfo.ConnSentBytes = float64(connData.sentBytes)
+		newConnectionMetricInfo.ConnReceivedBytes = float64(connData.receivedBytes)
+		newConnectionMetricInfo.ConnPacketLost = float64(connData.packetLost)
+		newConnectionMetricInfo.ConnTotalRetrans = float64(connData.totalRetrans)
+		m.connectionMetricCache[labels] = &newConnectionMetricInfo
+	}
+	if data.state == TCP_CLOSTED {
+		deleteLock.Lock()
+		deleteConnection = append(deleteConnection, &labels)
+		deleteLock.Unlock()
+	}
+}
+
 func (m *MetricController) updatePrometheusMetric() {
 	m.mutex.Lock()
 	workloadInfoCache := m.workloadMetricCache
 	serviceInfoCache := m.serviceMetricCache
+	connectionInfoCache := m.connectionMetricCache
 	m.workloadMetricCache = map[workloadMetricLabels]*workloadMetricInfo{}
 	m.serviceMetricCache = map[serviceMetricLabels]*serviceMetricInfo{}
+	m.connectionMetricCache = map[connectionMetricLabels]*connectionMetricInfo{}
 	m.mutex.Unlock()
 
 	for k, v := range workloadInfoCache {
@@ -786,6 +958,14 @@ func (m *MetricController) updatePrometheusMetric() {
 		tcpSentBytesInService.With(serviceLabels).Add(v.ServiceConnSentBytes)
 	}
 
+	for k, v := range connectionInfoCache {
+		connectionLabels := struct2map(k)
+		tcpConnectionTotalSendBytes.With(connectionLabels).Add(v.ConnSentBytes)
+		tcpConnectionTotalReceivedBytes.With(connectionLabels).Add(v.ConnReceivedBytes)
+		tcpConnectionTotalPacketLost.With(connectionLabels).Add(v.ConnPacketLost)
+		tcpConnectionTotalRetrans.With(connectionLabels).Add(v.ConnTotalRetrans)
+	}
+
 	// delete metrics
 	deleteLock.Lock()
 	// Creating a copy reduces the amount of time spent adding locks in the programme
@@ -793,6 +973,8 @@ func (m *MetricController) updatePrometheusMetric() {
 	deleteWorkload = nil
 	serviceReplica := deleteService
 	deleteService = nil
+	connReplica := deleteConnection
+	deleteConnection = []*connectionMetricLabels{}
 	deleteLock.Unlock()
 
 	for i := 0; i < len(workloadReplica); i++ {
@@ -801,6 +983,9 @@ func (m *MetricController) updatePrometheusMetric() {
 	for i := 0; i < len(serviceReplica); i++ {
 		deleteServiceMetricInPrometheus(serviceReplica[i])
 	}
+	for i := 0; i < len(connReplica); i++ {
+		deleteConnectionMetricInPrometheus(connReplica[i])
+	}
 }
 
 func struct2map(labels interface{}) map[string]string {
diff --git a/pkg/controller/telemetry/utils.go b/pkg/controller/telemetry/utils.go
index edbc2776..59f06a02 100644
--- a/pkg/controller/telemetry/utils.go
+++ b/pkg/controller/telemetry/utils.go
@@ -34,9 +34,10 @@ var (
 	// ensure not occur matche the same requests as /status/metric panic in unit test
 	mu sync.Mutex
 	// Ensure concurrency security when removing metriclabels from workloads and services.
-	deleteLock     sync.Mutex
-	deleteWorkload = []*workloadapi.Workload{}
-	deleteService  = []string{}
+	deleteLock       sync.Mutex
+	deleteWorkload   = []*workloadapi.Workload{}
+	deleteService    = []string{}
+	deleteConnection = []*connectionMetricLabels{}
 
 	workloadLabels = []string{
 		"reporter",
@@ -90,8 +91,41 @@ var (
 		"connection_security_policy",
 	}
 
+	connectionLabels = []string{
+		"reporter",
+		"start_time",
+		"source_workload",
+		"source_canonical_service",
+		"source_canonical_revision",
+		"source_workload_namespace",
+		"source_principal",
+		"source_app",
+		"source_version",
+		"source_cluster",
+		"source_address",
+		"destination_address",
+		"destination_pod_address",
+		"destination_pod_namespace",
+		"destination_pod_name",
+		"destination_service",
+		"destination_service_namespace",
+		"destination_service_name",
+		"destination_workload",
+		"destination_canonical_service",
+		"destination_canonical_revision",
+		"destination_workload_namespace",
+		"destination_principal",
+		"destination_app",
+		"destination_version",
+		"destination_cluster",
+		"request_protocol",
+		"response_flags",
+		"connection_security_policy",
+	}
+
 	labelsMap = map[string]string{
 		"reporter":                     "reporter",
+		"startTime":                    "start_time",
 		"sourceWorkload":               "source_workload",
 		"sourceCanonicalService":       "source_canonical_service",
 		"sourceCanonicalRevision":      "source_canonical_revision",
@@ -100,6 +134,8 @@ var (
 		"sourceApp":                    "source_app",
 		"sourceVersion":                "source_version",
 		"sourceCluster":                "source_cluster",
+		"sourceAddress":                "source_address",
+		"destinationAddress":           "destination_address",
 		"destinationService":           "destination_service",
 		"destinationServiceNamespace":  "destination_service_namespace",
 		"destinationServiceName":       "destination_service_name",
@@ -209,6 +245,31 @@ var (
 			Help: "The total number of TCP connections failed to a service.",
 		}, serviceLabels)
 
+	// Metrics to track the status of long lived TCP connections (duration > 30s)
+	tcpConnectionTotalSendBytes = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "kmesh_tcp_connection_sent_bytes_total",
+			Help: "The total number of bytes sent over established TCP connection.",
+		}, connectionLabels)
+
+	tcpConnectionTotalReceivedBytes = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "kmesh_tcp_connection_received_bytes_total",
+			Help: "The total number of bytes received over established TCP connection.",
+		}, connectionLabels)
+
+	tcpConnectionTotalPacketLost = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "kmesh_tcp_connection_packet_lost_total",
+			Help: "Total number of packets lost during transmission in a TCP connection.",
+		}, connectionLabels)
+
+	tcpConnectionTotalRetrans = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Name: "kmesh_tcp_connection_retrans_total",
+			Help: "The total number of retransmits over established TCP connection.",
+		}, connectionLabels)
+
 	// New operation metrics
 	bpfProgOpDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -257,6 +318,7 @@ func runPrometheusClient(registry *prometheus.Registry) {
 	defer mu.Unlock()
 	registry.MustRegister(tcpConnectionOpenedInWorkload, tcpConnectionClosedInWorkload, tcpReceivedBytesInWorkload, tcpSentBytesInWorkload, tcpConnectionTotalRetransInWorkload, tcpConnectionPacketLostInWorkload)
 	registry.MustRegister(tcpConnectionOpenedInService, tcpConnectionClosedInService, tcpReceivedBytesInService, tcpSentBytesInService)
+	registry.MustRegister(tcpConnectionTotalSendBytes, tcpConnectionTotalReceivedBytes, tcpConnectionTotalPacketLost, tcpConnectionTotalRetrans)
 	registry.MustRegister(bpfProgOpDuration, bpfProgOpCount)
 	registry.MustRegister(mapEntryCount, mapCountInNode)
 
@@ -322,3 +384,10 @@ func deleteServiceMetricInPrometheus(serviceName string) {
 	_ = tcpReceivedBytesInService.DeletePartialMatch(prometheus.Labels{"destination_service_name": svcHost, "destination_service_namespace": svcNamespace})
 	_ = tcpSentBytesInService.DeletePartialMatch(prometheus.Labels{"destination_service_name": svcHost, "destination_service_namespace": svcNamespace})
 }
+
+func deleteConnectionMetricInPrometheus(connLabels *connectionMetricLabels) {
+	_ = tcpConnectionTotalSendBytes.DeletePartialMatch(prometheus.Labels{"source_address": connLabels.sourceAddress, "destination_address": connLabels.destinationAddress})
+	_ = tcpConnectionTotalReceivedBytes.DeletePartialMatch(prometheus.Labels{"source_address": connLabels.sourceAddress, "destination_address": connLabels.destinationAddress})
+	_ = tcpConnectionTotalPacketLost.DeletePartialMatch(prometheus.Labels{"source_address": connLabels.sourceAddress, "destination_address": connLabels.destinationAddress})
+	_ = tcpConnectionTotalRetrans.DeletePartialMatch(prometheus.Labels{"source_address": connLabels.sourceAddress, "destination_address": connLabels.destinationAddress})
+}
diff --git a/pkg/controller/workload/workload_controller.go b/pkg/controller/workload/workload_controller.go
index 4897bbda..1eb57ec9 100644
--- a/pkg/controller/workload/workload_controller.go
+++ b/pkg/controller/workload/workload_controller.go
@@ -176,3 +176,15 @@ func (c *Controller) GetAccesslogTrigger() bool {
 func (c *Controller) SetWorkloadMetricTrigger(enable bool) {
 	c.MetricController.EnableWorkloadMetric.Store(enable)
 }
+
+func (c *Controller) GetWorklaodMetricTrigger() bool {
+	return c.MetricController.EnableWorkloadMetric.Load()
+}
+
+func (c *Controller) SetConnectionMetricTrigger(enable bool) {
+	c.MetricController.EnableConnectionMetric.Store(enable)
+}
+
+func (c *Controller) GetConnectionMetricTrigger() bool {
+	return c.MetricController.EnableConnectionMetric.Load()
+}
diff --git a/pkg/status/status_server.go b/pkg/status/status_server.go
index 2f5d034d..fc73cb09 100644
--- a/pkg/status/status_server.go
+++ b/pkg/status/status_server.go
@@ -57,6 +57,7 @@ const (
 	patternAccesslog          = "/accesslog"
 	patternMonitoring         = "/monitoring"
 	patternWorkloadMetrics    = "/workload_metrics"
+	patternConnectionMetrics  = "/connection_metrics"
 	patternAuthz              = "/authz"
 
 	bpfLoggerName = "bpf"
@@ -97,6 +98,7 @@ func NewServer(c *controller.XdsClient, configs *options.BootstrapConfigs, loade
 	s.mux.HandleFunc(patternAccesslog, s.accesslogHandler)
 	s.mux.HandleFunc(patternMonitoring, s.monitoringHandler)
 	s.mux.HandleFunc(patternWorkloadMetrics, s.workloadMetricHandler)
+	s.mux.HandleFunc(patternConnectionMetrics, s.connectionMetricHandler)
 	s.mux.HandleFunc(patternAuthz, s.authzHandler)
 
 	// TODO: add dump certificate, authorizationPolicies and services
@@ -265,6 +267,7 @@ func (s *Server) monitoringHandler(w http.ResponseWriter, r *http.Request) {
 	s.xdsClient.WorkloadController.SetMonitoringTrigger(enabled)
 	s.xdsClient.WorkloadController.SetAccesslogTrigger(enabled)
 	s.xdsClient.WorkloadController.SetWorkloadMetricTrigger(enabled)
+	s.xdsClient.WorkloadController.SetConnectionMetricTrigger(enabled)
 	w.WriteHeader(http.StatusOK)
 }
 
@@ -291,6 +294,29 @@ func (s *Server) workloadMetricHandler(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusOK)
 }
 
+func (s *Server) connectionMetricHandler(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
+		return
+	}
+
+	info := r.URL.Query().Get("enable")
+	enabled, err := strconv.ParseBool(info)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		_, _ = w.Write([]byte(fmt.Sprintf("invalid accesslog enable=%s", info)))
+		return
+	}
+
+	if s.loader.GetEnableMonitoring() == constants.DISABLED && enabled {
+		http.Error(w, "Kmesh monitoring is disabled, cannot enable connection metrics.", http.StatusBadRequest)
+		return
+	}
+
+	s.xdsClient.WorkloadController.SetConnectionMetricTrigger(enabled)
+	w.WriteHeader(http.StatusOK)
+}
+
 func (s *Server) authzHandler(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)



Test patch:
diff --git a/pkg/controller/telemetry/metric_test.go b/pkg/controller/telemetry/metric_test.go
index 01045c79..928a78f9 100644
--- a/pkg/controller/telemetry/metric_test.go
+++ b/pkg/controller/telemetry/metric_test.go
@@ -175,9 +175,9 @@ func TestCommonTrafficLabels2map(t *testing.T) {
 
 func TestBuildMetricsToPrometheus(t *testing.T) {
 	type args struct {
-		data      requestMetric
-		labels    workloadMetricLabels
-		tcp_conns map[connectionSrcDst]connMetric
+		data     requestMetric
+		labels   workloadMetricLabels
+		tcpConns map[connectionSrcDst]connMetric
 	}
 	tests := []struct {
 		name string
@@ -185,7 +185,7 @@ func TestBuildMetricsToPrometheus(t *testing.T) {
 		want []float64
 	}{
 		{
-			name: "test build workload metrisc to metricCache",
+			name: "test build workload metrics to metricCache",
 			args: args{
 				data: requestMetric{
 					conSrcDstInfo: connectionSrcDst{
@@ -223,7 +223,7 @@ func TestBuildMetricsToPrometheus(t *testing.T) {
 					responseFlags:                "-",
 					connectionSecurityPolicy:     "mutual_tls",
 				},
-				tcp_conns: map[connectionSrcDst]connMetric{
+				tcpConns: map[connectionSrcDst]connMetric{
 					{
 						src: [4]uint32{183763210, 0, 0, 0},
 						dst: [4]uint32{183762951, 0, 0, 0},
@@ -242,7 +242,7 @@ func TestBuildMetricsToPrometheus(t *testing.T) {
 			},
 		},
 		{
-			name: "test build workload metrisc to metricCache for conn report at intervals",
+			name: "test build workload metrics in metricCache for conn report at intervals",
 			args: args{
 				data: requestMetric{
 					conSrcDstInfo: connectionSrcDst{
@@ -280,7 +280,7 @@ func TestBuildMetricsToPrometheus(t *testing.T) {
 					responseFlags:                "-",
 					connectionSecurityPolicy:     "mutual_tls",
 				},
-				tcp_conns: map[connectionSrcDst]connMetric{
+				tcpConns: map[connectionSrcDst]connMetric{
 					{
 						src: [4]uint32{183763210, 0, 0, 0},
 						dst: [4]uint32{183762951, 0, 0, 0},
@@ -306,7 +306,7 @@ func TestBuildMetricsToPrometheus(t *testing.T) {
 				workloadMetricCache: map[workloadMetricLabels]*workloadMetricInfo{},
 				serviceMetricCache:  map[serviceMetricLabels]*serviceMetricInfo{},
 			}
-			m.updateWorkloadMetricCache(tt.args.data, tt.args.labels, tt.args.tcp_conns)
+			m.updateWorkloadMetricCache(tt.args.data, tt.args.labels, tt.args.tcpConns)
 			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])
@@ -319,9 +319,9 @@ func TestBuildMetricsToPrometheus(t *testing.T) {
 
 func TestBuildServiceMetricsToPrometheus(t *testing.T) {
 	type args struct {
-		data      requestMetric
-		labels    serviceMetricLabels
-		tcp_conns map[connectionSrcDst]connMetric
+		data     requestMetric
+		labels   serviceMetricLabels
+		tcpConns map[connectionSrcDst]connMetric
 	}
 	tests := []struct {
 		name string
@@ -364,7 +364,7 @@ func TestBuildServiceMetricsToPrometheus(t *testing.T) {
 					responseFlags:                "-",
 					connectionSecurityPolicy:     "mutual_tls",
 				},
-				tcp_conns: map[connectionSrcDst]connMetric{
+				tcpConns: map[connectionSrcDst]connMetric{
 					{
 						src: [4]uint32{183763210, 0, 0, 0},
 						dst: [4]uint32{183762951, 0, 0, 0},
@@ -416,7 +416,7 @@ func TestBuildServiceMetricsToPrometheus(t *testing.T) {
 					responseFlags:                "-",
 					connectionSecurityPolicy:     "mutual_tls",
 				},
-				tcp_conns: map[connectionSrcDst]connMetric{
+				tcpConns: map[connectionSrcDst]connMetric{
 					{
 						src: [4]uint32{183763210, 0, 0, 0},
 						dst: [4]uint32{183762951, 0, 0, 0},
@@ -440,7 +440,7 @@ func TestBuildServiceMetricsToPrometheus(t *testing.T) {
 				workloadMetricCache: map[workloadMetricLabels]*workloadMetricInfo{},
 				serviceMetricCache:  map[serviceMetricLabels]*serviceMetricInfo{},
 			}
-			m.updateServiceMetricCache(tt.args.data, tt.args.labels, tt.args.tcp_conns)
+			m.updateServiceMetricCache(tt.args.data, tt.args.labels, tt.args.tcpConns)
 			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])
@@ -449,6 +449,190 @@ func TestBuildServiceMetricsToPrometheus(t *testing.T) {
 	}
 }
 
+func TestBuildConnectionMetricsToPrometheus(t *testing.T) {
+	type args struct {
+		data     requestMetric
+		labels   connectionMetricLabels
+		tcpConns map[connectionSrcDst]connMetric
+	}
+	tests := []struct {
+		id   int32
+		name string
+		args args
+		want []float64
+	}{
+		{
+			id:   int32(1),
+			name: "test build connection metric to metricCache",
+			args: args{
+				data: requestMetric{
+					conSrcDstInfo: connectionSrcDst{
+						src: [4]uint32{183763210, 0, 0, 0},
+						dst: [4]uint32{183762951, 0, 0, 0},
+					},
+					sentBytes:     0x0000003,
+					receivedBytes: 0x0000004,
+					packetLost:    0x0000001,
+					totalRetrans:  0x0000002,
+					state:         TCP_ESTABLISHED,
+				},
+				labels: connectionMetricLabels{
+					reporter:                     "destination",
+					sourceWorkload:               "sleep",
+					sourceCanonicalService:       "sleep",
+					sourceCanonicalRevision:      "latest",
+					sourceWorkloadNamespace:      "ambient-demo",
+					sourcePrincipal:              "spiffe://cluster.local/ns/ambient-demo/sa/sleep",
+					sourceApp:                    "sleep",
+					sourceVersion:                "latest",
+					sourceCluster:                "Kubernetes",
+					destinationPodAddress:        "192.168.20.25",
+					destinationPodNamespace:      "ambient-demo",
+					destinationPodName:           "tcp-echo",
+					destinationWorkload:          "tcp-echo",
+					destinationCanonicalService:  "tcp-echo",
+					destinationCanonicalRevision: "v1",
+					destinationWorkloadNamespace: "ambient-demo",
+					destinationPrincipal:         "spiffe://cluster.local/ns/ambient-demo/sa/default",
+					destinationApp:               "tcp-echo",
+					destinationVersion:           "v1",
+					destinationCluster:           "Kubernetes",
+					requestProtocol:              "tcp",
+					responseFlags:                "-",
+					connectionSecurityPolicy:     "mutual_tls",
+				},
+				tcpConns: map[connectionSrcDst]connMetric{
+					{
+						src: [4]uint32{183763210, 0, 0, 0},
+						dst: [4]uint32{183762951, 0, 0, 0},
+					}: {
+						sentBytes:     0x0000003,
+						receivedBytes: 0x0000004,
+						packetLost:    0x0000001,
+						totalRetrans:  0x0000002,
+						totalReports:  1,
+					},
+				},
+			},
+			want: []float64{
+				3,
+				4,
+				1,
+				2,
+			},
+		},
+		{
+			id:   int32(2),
+			name: "test build workload metrisc to metricCache for conn report at intervals",
+			args: args{
+				data: requestMetric{
+					conSrcDstInfo: connectionSrcDst{
+						src: [4]uint32{183763210, 0, 0, 0},
+						dst: [4]uint32{183762951, 0, 0, 0},
+					},
+					sentBytes:     0x0000003,
+					receivedBytes: 0x0000004,
+					packetLost:    0x0000001,
+					totalRetrans:  0x0000002,
+					state:         TCP_CLOSTED,
+				},
+				labels: connectionMetricLabels{
+					reporter:                     "source",
+					sourceWorkload:               "sleep",
+					sourceCanonicalService:       "sleep",
+					sourceCanonicalRevision:      "latest",
+					sourceWorkloadNamespace:      "ambient-demo",
+					sourcePrincipal:              "spiffe://cluster.local/ns/ambient-demo/sa/sleep",
+					sourceApp:                    "sleep",
+					sourceVersion:                "latest",
+					sourceCluster:                "Kubernetes",
+					destinationPodAddress:        "192.168.20.25",
+					destinationPodNamespace:      "ambient-demo",
+					destinationPodName:           "tcp-echo",
+					destinationWorkload:          "tcp-echo",
+					destinationCanonicalService:  "tcp-echo",
+					destinationCanonicalRevision: "v1",
+					destinationWorkloadNamespace: "ambient-demo",
+					destinationPrincipal:         "spiffe://cluster.local/ns/ambient-demo/sa/default",
+					destinationApp:               "tcp-echo",
+					destinationVersion:           "v1",
+					destinationCluster:           "Kubernetes",
+					requestProtocol:              "tcp",
+					responseFlags:                "-",
+					connectionSecurityPolicy:     "mutual_tls",
+				},
+				tcpConns: map[connectionSrcDst]connMetric{
+					{
+						src: [4]uint32{183763210, 0, 0, 0},
+						dst: [4]uint32{183762951, 0, 0, 0},
+					}: {
+						totalReports: 3,
+					},
+				},
+			},
+			want: []float64{
+				4,
+				5,
+				2,
+				3,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m := MetricController{
+				workloadCache:       cache.NewWorkloadCache(),
+				workloadMetricCache: map[workloadMetricLabels]*workloadMetricInfo{},
+				serviceMetricCache:  map[serviceMetricLabels]*serviceMetricInfo{},
+				connectionMetricCache: map[connectionMetricLabels]*connectionMetricInfo{
+					{
+						reporter:                     "source",
+						sourceWorkload:               "sleep",
+						sourceCanonicalService:       "sleep",
+						sourceCanonicalRevision:      "latest",
+						sourceWorkloadNamespace:      "ambient-demo",
+						sourcePrincipal:              "spiffe://cluster.local/ns/ambient-demo/sa/sleep",
+						sourceApp:                    "sleep",
+						sourceVersion:                "latest",
+						sourceCluster:                "Kubernetes",
+						destinationPodAddress:        "192.168.20.25",
+						destinationPodNamespace:      "ambient-demo",
+						destinationPodName:           "tcp-echo",
+						destinationWorkload:          "tcp-echo",
+						destinationCanonicalService:  "tcp-echo",
+						destinationCanonicalRevision: "v1",
+						destinationWorkloadNamespace: "ambient-demo",
+						destinationPrincipal:         "spiffe://cluster.local/ns/ambient-demo/sa/default",
+						destinationApp:               "tcp-echo",
+						destinationVersion:           "v1",
+						destinationCluster:           "Kubernetes",
+						requestProtocol:              "tcp",
+						responseFlags:                "-",
+						connectionSecurityPolicy:     "mutual_tls",
+					}: {
+						ConnSentBytes:     0x0000001,
+						ConnReceivedBytes: 0x0000001,
+						ConnPacketLost:    0x0000001,
+						ConnTotalRetrans:  0x0000001,
+					},
+				},
+			}
+
+			deleteConnection = []*connectionMetricLabels{}
+			m.updateConnectionMetricCache(tt.args.data, tt.args.tcpConns[tt.args.data.conSrcDstInfo], tt.args.labels)
+			assert.Equal(t, m.connectionMetricCache[tt.args.labels].ConnSentBytes, tt.want[0])
+			assert.Equal(t, m.connectionMetricCache[tt.args.labels].ConnReceivedBytes, tt.want[1])
+			assert.Equal(t, m.connectionMetricCache[tt.args.labels].ConnPacketLost, tt.want[2])
+			assert.Equal(t, m.connectionMetricCache[tt.args.labels].ConnTotalRetrans, tt.want[3])
+			if tt.id == 1 {
+				assert.Equal(t, len(deleteConnection), 0)
+			} else {
+				assert.Equal(t, len(deleteConnection), 1)
+			}
+		})
+	}
+}
+
 func TestMetricGetWorkloadByAddress(t *testing.T) {
 	workload := &workloadapi.Workload{
 		Name: "ut-workload",
@@ -1066,6 +1250,151 @@ func TestBuildServiceMetric(t *testing.T) {
 	}
 }
 
+func TestBuildConnectionMetric(t *testing.T) {
+	serviceCache := cache.NewServiceCache()
+	serviceCache.AddOrUpdateService(&workloadapi.Service{
+		Hostname:  "kmesh.kmesh-system.svc.cluster.local",
+		Namespace: "kmesh-system",
+		Name:      "kmesh",
+		Addresses: []*workloadapi.NetworkAddress{
+			{
+				Address: net.ParseIP("192.168.1.22").To4(),
+			},
+		},
+	})
+	serviceCache.AddOrUpdateService(&workloadapi.Service{
+		Hostname:  "httpbin.default.svc.cluster.local",
+		Namespace: "default",
+		Name:      "httpbin",
+		Addresses: []*workloadapi.NetworkAddress{
+			{
+				Address: net.ParseIP("192.168.1.23").To4(),
+			},
+		},
+	})
+
+	workloadCache := cache.NewWorkloadCache()
+
+	workloadCache.AddOrUpdateWorkload(&workloadapi.Workload{
+		Namespace:         "default",
+		Name:              "sleep",
+		WorkloadName:      "sleep",
+		CanonicalName:     "sleepCanonical",
+		CanonicalRevision: "sleepVersion",
+		ClusterId:         "Kubernetes",
+		TrustDomain:       "cluster.local",
+		ServiceAccount:    "default",
+		Addresses: [][]byte{
+			{10, 19, 25, 33},
+		},
+	})
+
+	// kmesh workload with service attached
+	workloadCache.AddOrUpdateWorkload(&workloadapi.Workload{
+		Namespace:         "kmesh-system",
+		Name:              "kmesh",
+		WorkloadName:      "kmesh-daemon",
+		CanonicalName:     "dstCanonical",
+		CanonicalRevision: "dstVersion",
+		ClusterId:         "Kubernetes",
+		TrustDomain:       "cluster.local",
+		ServiceAccount:    "default",
+		Services: map[string]*workloadapi.PortList{
+			"kmesh-system/kmesh.kmesh-system.svc.cluster.local": {
+				Ports: []*workloadapi.Port{
+					{
+						ServicePort: 80,
+						TargetPort:  8000,
+					},
+				},
+			},
+		},
+		Addresses: [][]byte{
+			{10, 19, 25, 31},
+		},
+	})
+	type args struct {
+		data *requestMetric
+	}
+
+	tests := []struct {
+		name    string
+		args    args
+		want    connectionMetricLabels
+		wantErr bool
+	}{
+		{
+			name: "normal capability test",
+			args: args{
+				data: &requestMetric{
+					conSrcDstInfo: connectionSrcDst{
+						// sleep
+						src: [4]uint32{nets.ConvertIpToUint32("10.19.25.33"), 0, 0, 0},
+						// kmesh-daemon
+						dst:     [4]uint32{nets.ConvertIpToUint32("10.19.25.31"), 0, 0, 0},
+						dstPort: uint16(8000),
+						srcPort: uint16(8000),
+					},
+					// kmesh-daemon
+					origDstAddr:   [4]uint32{nets.ConvertIpToUint32("192.168.1.22"), 0, 0, 0},
+					origDstPort:   uint16(8000),
+					direction:     uint32(2),
+					sentBytes:     uint32(156),
+					receivedBytes: uint32(1024),
+				},
+			},
+			want: connectionMetricLabels{
+				reporter:                "source",
+				startTime:               "",
+				sourceWorkload:          "sleep",
+				sourceCanonicalService:  "sleepCanonical",
+				sourceCanonicalRevision: "sleepVersion",
+				sourceWorkloadNamespace: "default",
+				sourcePrincipal:         "spiffe://cluster.local/ns/default/sa/default",
+				sourceApp:               "sleepCanonical",
+				sourceVersion:           "sleepVersion",
+				sourceCluster:           "Kubernetes",
+				sourceAddress:           "10.19.25.33:8000",
+
+				destinationAddress:           "10.19.25.31:8000",
+				destinationPodAddress:        "10.19.25.31",
+				destinationPodNamespace:      "kmesh-system",
+				destinationPodName:           "kmesh",
+				destinationService:           "kmesh.kmesh-system.svc.cluster.local",
+				destinationServiceNamespace:  "kmesh-system",
+				destinationServiceName:       "kmesh",
+				destinationWorkload:          "kmesh-daemon",
+				destinationCanonicalService:  "dstCanonical",
+				destinationCanonicalRevision: "dstVersion",
+				destinationWorkloadNamespace: "kmesh-system",
+				destinationPrincipal:         "spiffe://cluster.local/ns/kmesh-system/sa/default",
+				destinationApp:               "dstCanonical",
+				destinationVersion:           "dstVersion",
+				destinationCluster:           "Kubernetes",
+
+				requestProtocol: "tcp",
+				responseFlags:   "",
+
+				connectionSecurityPolicy: "mutual_tls",
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			m := MetricController{
+				workloadCache: workloadCache,
+				serviceCache:  serviceCache,
+			}
+			got := m.buildConnectionMetric(tt.args.data)
+			got.startTime = ""
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Metric.buildMetric() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 func TestMetricController_updatePrometheusMetric(t *testing.T) {
 	testworkloadLabel1 := workloadMetricLabels{
 		sourceWorkload:               "kmesh-daemon",
@@ -1164,20 +1493,96 @@ func TestMetricController_updatePrometheusMetric(t *testing.T) {
 		responseFlags:                "-",
 		connectionSecurityPolicy:     "mutual_tls",
 	}
+
+	testConnLabel1 := connectionMetricLabels{
+		reporter:                "source",
+		startTime:               "0001-01-01 00:00:00 +0000 UTC",
+		sourceWorkload:          "sleep",
+		sourceCanonicalService:  "sleepCanonical",
+		sourceCanonicalRevision: "sleepVersion",
+		sourceWorkloadNamespace: "default",
+		sourcePrincipal:         "spiffe://cluster.local/ns/default/sa/default",
+		sourceApp:               "sleepCanonical",
+		sourceVersion:           "sleepVersion",
+		sourceCluster:           "Kubernetes",
+		sourceAddress:           "10.19.25.33:8000",
+
+		destinationAddress:           "10.19.25.31:8000",
+		destinationPodAddress:        "10.19.25.31",
+		destinationPodNamespace:      "kmesh-system",
+		destinationPodName:           "kmesh",
+		destinationService:           "kmesh.kmesh-system.svc.cluster.local",
+		destinationServiceNamespace:  "kmesh-system",
+		destinationServiceName:       "kmesh",
+		destinationWorkload:          "kmesh-daemon",
+		destinationCanonicalService:  "dstCanonical",
+		destinationCanonicalRevision: "dstVersion",
+		destinationWorkloadNamespace: "kmesh-system",
+		destinationPrincipal:         "spiffe://cluster.local/ns/kmesh-system/sa/default",
+		destinationApp:               "dstCanonical",
+		destinationVersion:           "dstVersion",
+		destinationCluster:           "Kubernetes",
+
+		requestProtocol: "tcp",
+		responseFlags:   "",
+
+		connectionSecurityPolicy: "mutual_tls",
+	}
+
+	testConnLabel2 := connectionMetricLabels{
+		reporter:                "destination",
+		startTime:               "0001-01-01 00:00:00 +0000 UTC",
+		sourceWorkload:          "sleep",
+		sourceCanonicalService:  "sleepCanonical",
+		sourceCanonicalRevision: "sleepVersion",
+		sourceWorkloadNamespace: "default",
+		sourcePrincipal:         "spiffe://cluster.local/ns/default/sa/default",
+		sourceApp:               "sleepCanonical",
+		sourceVersion:           "sleepVersion",
+		sourceCluster:           "Kubernetes",
+		sourceAddress:           "10.19.25.33:8000",
+
+		destinationAddress:           "10.19.25.31:8000",
+		destinationPodAddress:        "10.19.25.31",
+		destinationPodNamespace:      "kmesh-system",
+		destinationPodName:           "kmesh",
+		destinationService:           "kmesh.kmesh-system.svc.cluster.local",
+		destinationServiceNamespace:  "kmesh-system",
+		destinationServiceName:       "kmesh",
+		destinationWorkload:          "kmesh-daemon",
+		destinationCanonicalService:  "dstCanonical",
+		destinationCanonicalRevision: "dstVersion",
+		destinationWorkloadNamespace: "kmesh-system",
+		destinationPrincipal:         "spiffe://cluster.local/ns/kmesh-system/sa/default",
+		destinationApp:               "dstCanonical",
+		destinationVersion:           "dstVersion",
+		destinationCluster:           "Kubernetes",
+
+		requestProtocol: "tcp",
+		responseFlags:   "",
+
+		connectionSecurityPolicy: "mutual_tls",
+	}
+
 	workloadPrometheusLabel1 := struct2map(testworkloadLabel1)
 	workloadPrometheusLabel2 := struct2map(testworkloadLabel2)
 	servicePrometheusLabel1 := struct2map(testServiceLabel1)
 	servicePrometheusLabel2 := struct2map(testServiceLabel2)
+	connectionPrometheusLabel1 := struct2map(testConnLabel1)
+	connectionPrometheusLabel2 := struct2map(testConnLabel2)
+
 	tests := []struct {
-		name                  string
-		workloadMetricCache   workloadMetricInfo
-		serviceMetricCache    serviceMetricInfo
-		exportWorkloadMetrics []*prometheus.GaugeVec
-		exportServiceMetrics  []*prometheus.GaugeVec
-		want                  []float64
+		name                    string
+		workloadMetricCache     workloadMetricInfo
+		serviceMetricCache      serviceMetricInfo
+		connectionMetricCache   connectionMetricInfo
+		exportWorkloadMetrics   []*prometheus.GaugeVec
+		exportServiceMetrics    []*prometheus.GaugeVec
+		exportConnectionMetrics []*prometheus.GaugeVec
+		want                    []float64
 	}{
 		{
-			name: "update workload metric in Prometheus",
+			name: "update metric in Prometheus",
 			workloadMetricCache: workloadMetricInfo{
 				WorkloadConnOpened:        1,
 				WorkloadConnClosed:        2,
@@ -1194,6 +1599,12 @@ func TestMetricController_updatePrometheusMetric(t *testing.T) {
 				ServiceConnSentBytes:     9,
 				ServiceConnReceivedBytes: 10,
 			},
+			connectionMetricCache: connectionMetricInfo{
+				ConnSentBytes:     1,
+				ConnReceivedBytes: 2,
+				ConnPacketLost:    3,
+				ConnTotalRetrans:  4,
+			},
 			exportWorkloadMetrics: []*prometheus.GaugeVec{
 				tcpConnectionOpenedInWorkload,
 				tcpConnectionClosedInWorkload,
@@ -1208,6 +1619,12 @@ func TestMetricController_updatePrometheusMetric(t *testing.T) {
 				tcpSentBytesInService,
 				tcpReceivedBytesInService,
 			},
+			exportConnectionMetrics: []*prometheus.GaugeVec{
+				tcpConnectionTotalSendBytes,
+				tcpConnectionTotalReceivedBytes,
+				tcpConnectionTotalPacketLost,
+				tcpConnectionTotalRetrans,
+			},
 			want: []float64{
 				1, 2, 3, 4, 5, 6, 7, 8, 9, 10,
 			},
@@ -1227,6 +1644,10 @@ func TestMetricController_updatePrometheusMetric(t *testing.T) {
 					testServiceLabel1: &tt.serviceMetricCache,
 					testServiceLabel2: &tt.serviceMetricCache,
 				},
+				connectionMetricCache: map[connectionMetricLabels]*connectionMetricInfo{
+					testConnLabel1: &tt.connectionMetricCache,
+					testConnLabel2: &tt.connectionMetricCache,
+				},
 			}
 			m.updatePrometheusMetric()
 			index := 0
@@ -1244,6 +1665,14 @@ func TestMetricController_updatePrometheusMetric(t *testing.T) {
 				assert.Equal(t, tt.want[index], v2)
 				index = index + 1
 			}
+			index = 0
+			for _, metric := range tt.exportConnectionMetrics {
+				v1 := testutil.ToFloat64(metric.With(connectionPrometheusLabel1))
+				assert.Equal(t, tt.want[index], v1)
+				v2 := testutil.ToFloat64(metric.With(connectionPrometheusLabel2))
+				assert.Equal(t, tt.want[index], v2)
+				index = index + 1
+			}
 			cancel()
 		})
 	}
diff --git a/pkg/controller/telemetry/utils_test.go b/pkg/controller/telemetry/utils_test.go
index aef1845e..8e0c5c03 100644
--- a/pkg/controller/telemetry/utils_test.go
+++ b/pkg/controller/telemetry/utils_test.go
@@ -50,6 +50,10 @@ func TestRegisterMetrics(t *testing.T) {
 		tcpReceivedBytesInService,
 		tcpSentBytesInService,
 		tcpConnectionFailedInService,
+		tcpConnectionTotalSendBytes,
+		tcpConnectionTotalReceivedBytes,
+		tcpConnectionTotalPacketLost,
+		tcpConnectionTotalRetrans,
 	}
 
 	workloadLabels := map[string]string{
@@ -104,6 +108,38 @@ func TestRegisterMetrics(t *testing.T) {
 		"connection_security_policy":     "mutual_tls",
 	}
 
+	connectionLabels := map[string]string{
+		"reporter":                       "destination",
+		"start_time":                     "2023-10-01T00:00:00Z",
+		"source_workload":                "sleep",
+		"source_canonical_service":       "sleep",
+		"source_canonical_revision":      "latest",
+		"source_workload_namespace":      "ambient-demo",
+		"source_principal":               "spiffe://cluster.local/ns/ambient-demo/sa/sleep",
+		"source_app":                     "sleep",
+		"source_version":                 "latest",
+		"source_cluster":                 "Kubernetes",
+		"source_address":                 "192.068.10.24:5665",
+		"destination_address":            "192.068.10.25:80",
+		"destination_pod_address":        "192.068.10.25",
+		"destination_pod_namespace":      "ambient-demo",
+		"destination_pod_name":           "httpbin-hjsrf",
+		"destination_service":            "sleep.ambient.svc.cluster.local",
+		"destination_service_namespace":  "ambient-demo",
+		"destination_service_name":       "tcp-echo",
+		"destination_workload":           "tcp-echo",
+		"destination_canonical_service":  "tcp-echo",
+		"destination_canonical_revision": "v1",
+		"destination_workload_namespace": "ambient-demo",
+		"destination_principal":          "spiffe://cluster.local/ns/ambient-demo/sa/default",
+		"destination_app":                "tcp-echo",
+		"destination_version":            "v1",
+		"destination_cluster":            "Kubernetes",
+		"request_protocol":               "tcp",
+		"response_flags":                 "-",
+		"connection_security_policy":     "mutual_tls",
+	}
+
 	tcpConnectionClosedInWorkload.With(workloadLabels).Set(2)
 	tcpConnectionOpenedInWorkload.With(workloadLabels).Set(4)
 	tcpReceivedBytesInWorkload.With(workloadLabels).Set(12.64)
@@ -116,6 +152,11 @@ func TestRegisterMetrics(t *testing.T) {
 	tcpConnectionOpenedInService.With(serviceLabels).Set(16.25)
 	tcpConnectionFailedInService.With(serviceLabels).Set(6.0)
 
+	tcpConnectionTotalSendBytes.With(connectionLabels).Set(12.64)
+	tcpConnectionTotalReceivedBytes.With(connectionLabels).Set(11.45)
+	tcpConnectionTotalPacketLost.With(connectionLabels).Set(1.0)
+	tcpConnectionTotalRetrans.With(connectionLabels).Set(2.0)
+
 	for _, metric := range exportMetrics {
 		if err := prometheus.Register(metric); err != nil {
 			t.Errorf("metric not register")
@@ -287,3 +328,92 @@ func TestDeleteServiceMetric(t *testing.T) {
 	}
 	cancel()
 }
+
+func TestDeleteConnectionMetric(t *testing.T) {
+	registry := prometheus.NewRegistry()
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				runPrometheusClient(registry)
+			}
+		}
+	}()
+
+	exportMetrics := []*prometheus.GaugeVec{
+		tcpConnectionTotalSendBytes,
+		tcpConnectionTotalReceivedBytes,
+		tcpConnectionTotalPacketLost,
+		tcpConnectionTotalRetrans,
+	}
+
+	connectionLabels := map[string]string{
+		"reporter":                       "destination",
+		"start_time":                     "2023-10-01T00:00:00Z",
+		"source_workload":                "sleep",
+		"source_canonical_service":       "sleep",
+		"source_canonical_revision":      "latest",
+		"source_workload_namespace":      "ambient-demo",
+		"source_principal":               "spiffe://cluster.local/ns/ambient-demo/sa/sleep",
+		"source_app":                     "sleep",
+		"source_version":                 "latest",
+		"source_cluster":                 "Kubernetes",
+		"source_address":                 "192.068.10.24:5665",
+		"destination_address":            "192.068.10.25:80",
+		"destination_pod_address":        "192.068.10.25",
+		"destination_pod_namespace":      "ambient-demo",
+		"destination_pod_name":           "httpbin-hjsrf",
+		"destination_service":            "sleep.ambient.svc.cluster.local",
+		"destination_service_namespace":  "ambient-demo",
+		"destination_service_name":       "tcp-echo",
+		"destination_workload":           "tcp-echo",
+		"destination_canonical_service":  "tcp-echo",
+		"destination_canonical_revision": "v1",
+		"destination_workload_namespace": "ambient-demo",
+		"destination_principal":          "spiffe://cluster.local/ns/ambient-demo/sa/default",
+		"destination_app":                "tcp-echo",
+		"destination_version":            "v1",
+		"destination_cluster":            "Kubernetes",
+		"request_protocol":               "tcp",
+		"response_flags":                 "-",
+		"connection_security_policy":     "mutual_tls",
+	}
+	type args struct {
+		connLabels *connectionMetricLabels
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "delete connection",
+			args: args{
+				connLabels: &connectionMetricLabels{
+					sourceAddress:      "192.068.10.24:5665",
+					destinationAddress: "192.068.10.25:80",
+					responseFlags:      "-",
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tcpConnectionTotalSendBytes.With(connectionLabels).Set(12.64)
+			tcpConnectionTotalReceivedBytes.With(connectionLabels).Set(11.45)
+			tcpConnectionTotalPacketLost.With(connectionLabels).Set(1.0)
+			tcpConnectionTotalRetrans.With(connectionLabels).Set(2.0)
+
+			deleteConnectionMetricInPrometheus(tt.args.connLabels)
+
+			for _, metric := range exportMetrics {
+				if err := prometheus.Register(metric); err == nil {
+					t.Errorf("metric not clean up")
+				}
+			}
+		})
+	}
+	cancel()
+}
diff --git a/pkg/status/status_server_test.go b/pkg/status/status_server_test.go
index a32e81f2..eacdf57d 100644
--- a/pkg/status/status_server_test.go
+++ b/pkg/status/status_server_test.go
@@ -480,6 +480,97 @@ func TestServerAccesslogHandler(t *testing.T) {
 	})
 }
 
+func TestServerMetricHandler(t *testing.T) {
+	t.Run("change accesslog, workload metrics and connection metric config info", func(t *testing.T) {
+		config := options.BpfConfig{
+			Mode:        constants.DualEngineMode,
+			BpfFsPath:   "/sys/fs/bpf",
+			Cgroup2Path: "/mnt/kmesh_cgroup2",
+		}
+		cleanup, loader := test.InitBpfMap(t, config)
+		defer cleanup()
+
+		server := &Server{
+			xdsClient: &controller.XdsClient{
+				WorkloadController: &workload.Controller{
+					MetricController: &telemetry.MetricController{},
+				},
+			},
+			loader: loader,
+		}
+		server.xdsClient.WorkloadController.MetricController.EnableWorkloadMetric.Store(true)
+		server.xdsClient.WorkloadController.MetricController.EnableConnectionMetric.Store(true)
+		server.xdsClient.WorkloadController.MetricController.EnableAccesslog.Store(false)
+
+		url := fmt.Sprintf("%s?enable=%s", patternMonitoring, "true")
+		req := httptest.NewRequest(http.MethodPost, url, nil)
+		w := httptest.NewRecorder()
+		server.monitoringHandler(w, req)
+
+		url = fmt.Sprintf("%s?enable=%s", patternAccesslog, "false")
+		req = httptest.NewRequest(http.MethodPost, url, nil)
+		w = httptest.NewRecorder()
+		server.accesslogHandler(w, req)
+
+		assert.Equal(t, server.xdsClient.WorkloadController.GetAccesslogTrigger(), false)
+
+		url = fmt.Sprintf("%s?enable=%s", patternWorkloadMetrics, "false")
+		req = httptest.NewRequest(http.MethodPost, url, nil)
+		w = httptest.NewRecorder()
+		server.workloadMetricHandler(w, req)
+
+		assert.Equal(t, server.xdsClient.WorkloadController.GetWorklaodMetricTrigger(), false)
+
+		url = fmt.Sprintf("%s?enable=%s", patternConnectionMetrics, "false")
+		req = httptest.NewRequest(http.MethodPost, url, nil)
+		w = httptest.NewRecorder()
+		server.connectionMetricHandler(w, req)
+
+		assert.Equal(t, server.xdsClient.WorkloadController.GetConnectionMetricTrigger(), false)
+	})
+
+	t.Run("when monitoring is disable, cannot enable accesslog, workload metrics and connection metrics", func(t *testing.T) {
+		config := options.BpfConfig{
+			Mode:        constants.DualEngineMode,
+			BpfFsPath:   "/sys/fs/bpf",
+			Cgroup2Path: "/mnt/kmesh_cgroup2",
+		}
+		cleanup, loader := test.InitBpfMap(t, config)
+		defer cleanup()
+
+		server := &Server{
+			xdsClient: &controller.XdsClient{
+				WorkloadController: &workload.Controller{
+					MetricController: &telemetry.MetricController{},
+				},
+			},
+			loader: loader,
+		}
+		server.xdsClient.WorkloadController.MetricController.EnableAccesslog.Store(false)
+
+		url := fmt.Sprintf("%s?enable=%s", patternAccesslog, "true")
+		req := httptest.NewRequest(http.MethodPost, url, nil)
+		w := httptest.NewRecorder()
+		server.accesslogHandler(w, req)
+
+		assert.Equal(t, server.xdsClient.WorkloadController.GetAccesslogTrigger(), false)
+
+		url = fmt.Sprintf("%s?enable=%s", patternWorkloadMetrics, "true")
+		req = httptest.NewRequest(http.MethodPost, url, nil)
+		w = httptest.NewRecorder()
+		server.workloadMetricHandler(w, req)
+
+		assert.Equal(t, server.xdsClient.WorkloadController.GetWorklaodMetricTrigger(), false)
+
+		url = fmt.Sprintf("%s?enable=%s", patternConnectionMetrics, "true")
+		req = httptest.NewRequest(http.MethodPost, url, nil)
+		w = httptest.NewRecorder()
+		server.connectionMetricHandler(w, req)
+
+		assert.Equal(t, server.xdsClient.WorkloadController.GetConnectionMetricTrigger(), false)
+	})
+}
+
 func TestServerMonitoringHandler(t *testing.T) {
 	t.Run("change monitoring config info", func(t *testing.T) {
 		config := options.BpfConfig{