fix issue polarismesh#102

chuntaojun · Oct 25, 2022 · 43091a5 · 43091a5
1 parent ac9bd6b
commit 43091a5
Show file tree

Hide file tree

Showing 7 changed files with 50 additions and 43 deletions.
diff --git a/examples/circuitbreaker/consumer/polaris.yaml b/examples/circuitbreaker/consumer/polaris.yaml
@@ -1,7 +1,7 @@
 global:
   serverConnector:
     addresses:
-      - 172.18.0.1:8091
+      - 127.0.0.1:8091
   statReporter:
     enable: true
     chain:

diff --git a/examples/circuitbreaker/provider/polaris.yaml b/examples/circuitbreaker/provider/polaris.yaml
@@ -1,7 +1,7 @@
 global:
   serverConnector:
     addresses:
-      - 172.18.0.1:8091
+      - 127.0.0.1:8091
   statReporter:
     enable: true
     chain:

diff --git a/examples/quickstart/provider/main.go b/examples/quickstart/provider/main.go
@@ -27,7 +27,6 @@ import (
 	"os/signal"
 	"strings"
 	"syscall"
-	"time"
 
 	"github.com/polarismesh/polaris-go"
 )
@@ -121,38 +120,6 @@ func (svr *PolarisProvider) deregisterService() {
 	log.Printf("deregister successfully.")
 }
 
-func (svr *PolarisProvider) doHeartbeat() {
-	log.Printf("start to invoke heartbeat operation")
-	ticker := time.NewTicker(time.Duration(5 * time.Second))
-	for range ticker.C {
-		if !svr.isShutdown {
-			heartbeatRequest := &polaris.InstanceHeartbeatRequest{}
-			heartbeatRequest.Namespace = namespace
-			heartbeatRequest.Service = service
-			heartbeatRequest.Host = svr.host
-			heartbeatRequest.Port = svr.port
-			heartbeatRequest.ServiceToken = token
-			svr.provider.Heartbeat(heartbeatRequest)
-		}
-	}
-}
-
-func (svr *PolarisProvider) doHeartbeat() {
-	log.Printf("start to invoke heartbeat operation")
-	ticker := time.NewTicker(time.Duration(5 * time.Second))
-	for range ticker.C {
-		if !svr.isShutdown {
-			heartbeatRequest := &polaris.InstanceHeartbeatRequest{}
-			heartbeatRequest.Namespace = namespace
-			heartbeatRequest.Service = service
-			heartbeatRequest.Host = svr.host
-			heartbeatRequest.Port = svr.port
-			heartbeatRequest.ServiceToken = token
-			svr.provider.Heartbeat(heartbeatRequest)
-		}
-	}
-}
-
 func (svr *PolarisProvider) runMainLoop() {
 	ch := make(chan os.Signal, 1)
 	signal.Notify(ch, []os.Signal{

diff --git a/pkg/model/service.go b/pkg/model/service.go
@@ -772,12 +772,18 @@ type CircuitBreakGauge struct {
 	EmptyInstanceGauge
 	ChangeInstance Instance
 	Method         string
-	CBStatus       CircuitBreakerStatus
+	PreCBStatus    CircuitBreakerStatus
+	NextCBStatus   CircuitBreakerStatus
 }
 
-// GetCircuitBreakerStatus 获取变化前的熔断状态
-func (cbg *CircuitBreakGauge) GetCircuitBreakerStatus() CircuitBreakerStatus {
-	return cbg.CBStatus
+// GetPreCircuitBreakerStatus 获取变化前的熔断状态
+func (cbg *CircuitBreakGauge) GetPreCircuitBreakerStatus() CircuitBreakerStatus {
+	return cbg.PreCBStatus
+}
+
+// GetNextCircuitBreakerStatus 获取变化后的熔断状态
+func (cbg *CircuitBreakGauge) GetNextCircuitBreakerStatus() CircuitBreakerStatus {
+	return cbg.NextCBStatus
 }
 
 // GetCalledInstance 获取状态发生改变的实例

diff --git a/pkg/model/stat.go b/pkg/model/stat.go
@@ -64,6 +64,31 @@ const (
 	RouteStat
 )
 
+func DescMetricType(t MetricType) string {
+	switch t {
+	case SDKAPIStat:
+		return "SDKAPIStat"
+	case ServiceStat:
+		return "ServiceStat"
+	case InstanceStat:
+		return "InstanceStat"
+	case SDKCfgStat:
+		return "SDKCfgStat"
+	case CircuitBreakStat:
+		return "CircuitBreakStat"
+	case PluginAPIStat:
+		return "PluginAPIStat"
+	case LoadBalanceStat:
+		return "LoadBalanceStat"
+	case RateLimitStat:
+		return "RateLimitStat"
+	case RouteStat:
+		return "RouteStat"
+	default:
+		return "Unknown"
+	}
+}
+
 var metricTypes = HashSet{}
 
 // ValidMetircType 检测是不是合法的统计类型.

diff --git a/plugin/localregistry/inmemory/inmemory.go b/plugin/localregistry/inmemory/inmemory.go
@@ -575,7 +575,7 @@ func (g *LocalCache) UpdateInstances(svcUpdateReq *localregistry.ServiceUpdateRe
 					cbStatusUpdated = false
 				}
 				err := g.engine.SyncReportStat(model.CircuitBreakStat,
-					&model.CircuitBreakGauge{ChangeInstance: updateInstance, CBStatus: preCBStatus})
+					&model.CircuitBreakGauge{ChangeInstance: updateInstance, PreCBStatus: preCBStatus, NextCBStatus: nextCBStatus})
 				if err != nil {
 					log.GetBaseLogger().Errorf("fail to report circuitbreak change, error %v", err)
 				}

diff --git a/plugin/statreporter/prometheus/prometheus_handler.go b/plugin/statreporter/prometheus/prometheus_handler.go
@@ -24,6 +24,7 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"go.uber.org/zap"
 
 	"github.com/polarismesh/polaris-go/pkg/log"
 	"github.com/polarismesh/polaris-go/pkg/model"
@@ -122,6 +123,13 @@ func (p *PrometheusHandler) registerMetrics() error {
 
 // ReportStat 上报采集指标到 prometheus，这里只针对部分 model.InstanceGauge 的实现做处理
 func (p *PrometheusHandler) ReportStat(metricsType model.MetricType, metricsVal model.InstanceGauge) error {
+	defer func() {
+		if err := recover(); err != nil {
+			log.GetBaseLogger().Errorf("[StatReporter][Prometheus] do report panic", zap.String("metric-type", model.DescMetricType(metricsType)),
+				zap.Any("metricsVal", metricsVal))
+		}
+	}()
+
 	switch metricsType {
 	case model.ServiceStat:
 		val, ok := metricsVal.(*model.ServiceCallResult)
@@ -225,19 +233,20 @@ func (p *PrometheusHandler) handleCircuitBreakGauge(metricsType model.MetricType
 
 	open := p.metricVecCaches[MetricsNameCircuitBreakerOpen].(*prometheus.CounterVec)
 
-	status := val.GetCircuitBreakerStatus().GetStatus()
+	// 计算完之后的熔断状态
+	status := val.GetNextCircuitBreakerStatus().GetStatus()
 	if status == model.Open {
 		open.With(labels).Inc()
 	} else {
-		open.With(labels).Add(-1)
+		open.With(labels).Desc()
 	}
 
 	halfOpen := p.metricVecCaches[MetricsNameCircuitBreakerHalfOpen].(*prometheus.CounterVec)
 
 	if status == model.HalfOpen {
 		halfOpen.With(labels).Inc()
 	} else {
-		halfOpen.With(labels).Add(-1)
+		halfOpen.With(labels).Desc()
 	}
 }