[skywalking-satellite] branch main updated: fix active connection gauge not correct (#96)

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

liuhan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-satellite.git


The following commit(s) were added to refs/heads/main by this push:
     new 3443fcc  fix active connection gauge not correct (#96)
3443fcc is described below

commit 3443fcc6dc41988f792f95050df959abf25cbf8f
Author: mrproliu <74...@qq.com>
AuthorDate: Sat Dec 18 13:47:16 2021 +0800

    fix active connection gauge not correct (#96)
---
 internal/satellite/telemetry/metricservice/server.go |  5 -----
 plugins/server/grpc/server_connections.go            | 15 +++++++++++----
 2 files changed, 11 insertions(+), 9 deletions(-)

diff --git a/internal/satellite/telemetry/metricservice/server.go b/internal/satellite/telemetry/metricservice/server.go
index de4bd01..8b24846 100644
--- a/internal/satellite/telemetry/metricservice/server.go
+++ b/internal/satellite/telemetry/metricservice/server.go
@@ -20,7 +20,6 @@ package metricservice
 import (
 	"context"
 	"fmt"
-	"io"
 	"sync"
 	"time"
 
@@ -116,10 +115,6 @@ func (s *Server) sendMetrics() error {
 	appender.metrics[0].Service = s.service
 	appender.metrics[0].ServiceInstance = s.instance
 	if err := s.reportStream.Send(&v3.MeterDataCollection{MeterData: appender.metrics}); err != nil {
-		if err != io.EOF {
-			return fmt.Errorf("could send metrics: %v", err)
-		}
-
 		if openErr := s.openBatchStream(); openErr != nil {
 			log.Logger.Warnf("detect send message error and reopen stream failure: %v", openErr)
 		}
diff --git a/plugins/server/grpc/server_connections.go b/plugins/server/grpc/server_connections.go
index cd02775..5372bed 100644
--- a/plugins/server/grpc/server_connections.go
+++ b/plugins/server/grpc/server_connections.go
@@ -19,6 +19,7 @@ package grpc
 
 import (
 	"net"
+	"sync"
 
 	"github.com/apache/skywalking-satellite/internal/pkg/log"
 )
@@ -53,8 +54,7 @@ func (c *ConnectionManager) Accept() (net.Conn, error) {
 			conn.RemoteAddr(), conn.LocalAddr(), c.acceptLimiter.CurrentCPU, c.acceptLimiter.ActiveConnection)
 		return nil, &outOfLimit{}
 	}
-
-	return &ConnectionWrapper{conn, c}, nil
+	return &ConnectionWrapper{Conn: conn, manager: c}, nil
 }
 
 func (c *ConnectionManager) notifyCloseConnection() {
@@ -63,14 +63,21 @@ func (c *ConnectionManager) notifyCloseConnection() {
 
 type ConnectionWrapper struct {
 	net.Conn
-	manager *ConnectionManager
+	manager   *ConnectionManager
+	closeOnce sync.Once
 }
 
 func (c *ConnectionWrapper) Close() error {
-	defer c.manager.notifyCloseConnection()
+	defer c.CloseNotify()
 	return c.Conn.Close()
 }
 
+func (c *ConnectionWrapper) CloseNotify() {
+	c.closeOnce.Do(func() {
+		c.manager.notifyCloseConnection()
+	})
+}
+
 type outOfLimit struct {
 	error
 }