polarismesh · chuntaojun · Dec 28, 2022 · Sep 23, 2022 · Sep 25, 2022 · Oct 19, 2022
diff --git a/.github/workflows/integration-testing-mysql.yml b/.github/workflows/integration-testing-mysql.yml
@@ -143,6 +143,11 @@ jobs:
           chmod +x ./tool/*.sh
           ./tool/start.sh
 
+          sleep 120s
+          ls -alR
+          cat ./log/stdout
+          grep -nr "error" ./log/runtime/*
+
           cd ..
           # 先测试普通的集成测试
           pushd ./test
@@ -152,6 +157,8 @@ jobs:
           ./tool/stop.sh
           cd ..
 
+
+
       # - name: run config-center interface tests
       #   run: |
       #     # 进行配置中心的集成测试

diff --git a/.github/workflows/integration-testing.yml b/.github/workflows/integration-testing.yml
@@ -64,7 +64,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.17.1
+          go-version: 1.19
       # Checkout latest code
       - name: Checkout repo
         uses: actions/checkout@v2
@@ -100,6 +100,11 @@ jobs:
           chmod +x ./tool/*.sh
           ./tool/start.sh
 
+          sleep 120s
+          ls -alR
+          cat ./log/stdout
+          grep -nr "error" ./log/runtime/*
+
           cd ..
           # 先测试普通的集成测试
           pushd ./test
@@ -108,3 +113,4 @@ jobs:
           cd $work_dir
           ./tool/stop.sh
           cd ..
+
diff --git a/apiserver/grpcserver/base.go b/apiserver/grpcserver/base.go
@@ -470,4 +470,8 @@ func (h *connCounterHook) OnRelease(conn net.Conn) {
 	metrics.RemoveSDKClient()
 }
 
-func (h *connCounterHook) OnClose() {}
+func (h *connCounterHook) OnClose() {
+	metrics.ResetDiscoveryClientConn()
+	metrics.ResetConfigurationClientConn()
+	metrics.ResetSDKClient()
+}
diff --git a/apiserver/xdsserverv3/log.go b/apiserver/xdsserverv3/log.go
@@ -21,4 +21,4 @@ import (
 	commonlog "github.com/polarismesh/polaris/common/log"
 )
 
-var log = commonlog.GetScopeOrDefaultByName(commonlog.NamingLoggerName)
+var log = commonlog.GetScopeOrDefaultByName(commonlog.XDSLoggerName)
diff --git a/common/conn/hook/listener.go b/common/conn/hook/listener.go
@@ -21,9 +21,13 @@ import (
 	"net"
 )
 
+// Hook Hook
 type Hook interface {
+	// OnAccept call when net.Conn accept
 	OnAccept(conn net.Conn)
+	// OnRelease call when net.Conn release
 	OnRelease(conn net.Conn)
+	// OnClose call when net.Listener close
 	OnClose()
 }
 
@@ -34,6 +38,7 @@ func NewHookListener(l net.Listener, hooks ...Hook) net.Listener {
 	}
 }
 
+// HookListener net.Listener can add hook
 type HookListener struct {
 	hooks  []Hook
 	target net.Listener

diff --git a/common/metrics/metrics.go b/common/metrics/metrics.go
@@ -19,6 +19,7 @@ package metrics
 
 import (
 	"net/http"
+	"sync/atomic"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -28,8 +29,12 @@ import (
 
 var (
 	registry = prometheus.NewRegistry()
+
+	lastRedisReadFailureReport  atomic.Value
+	lastRedisWriteFailureReport atomic.Value
 )
 
+// GetRegistry 获取 metrics 的 registry
 func GetRegistry() *prometheus.Registry {
 	return registry
 }
@@ -43,6 +48,7 @@ func GetHttpHandler() http.Handler {
 	return promhttp.HandlerFor(registry, promhttp.HandlerOpts{EnableOpenMetrics: true})
 }
 
+// InitMetrics 初始化 metrics 的所有指标
 func InitMetrics() {
 	_ = registry.Register(instanceAsyncRegisCost)
 	_ = registry.Register(instanceRegisTaskExpire)
@@ -54,6 +60,21 @@ func InitMetrics() {
 	_ = registry.Register(discoveryConnTotal)
 	_ = registry.Register(configurationConnTotal)
 	_ = registry.Register(sdkClientTotal)
+
+	go func() {
+		lastRedisReadFailureReport.Store(time.Now())
+		lastRedisWriteFailureReport.Store(time.Now())
+		ticker := time.NewTicker(time.Minute)
+		for range ticker.C {
+			tn := time.Now()
+			if tn.Sub(lastRedisReadFailureReport.Load().(time.Time)) > time.Minute {
+				redisReadFailure.Set(0)
+			}
+			if tn.Sub(lastRedisWriteFailureReport.Load().(time.Time)) > time.Minute {
+				redisWriteFailure.Set(0)
+			}
+		}
+	}()
 }
 
 // ReportInstanceRegisCost Total time to report the short-term registered task of the reporting instance
@@ -68,11 +89,13 @@ func ReportDropInstanceRegisTask() {
 
 // ReportRedisReadFailure report redis exec read operatio failure
 func ReportRedisReadFailure() {
+	lastRedisReadFailureReport.Store(time.Now())
 	redisReadFailure.Inc()
 }
 
 // ReportRedisWriteFailure report redis exec write operatio failure
 func ReportRedisWriteFailure() {
+	lastRedisWriteFailureReport.Store(time.Now())
 	redisWriteFailure.Inc()
 }
 
@@ -81,7 +104,7 @@ func ReportRedisIsDead() {
 	redisAliveStatus.Set(0)
 }
 
-// ReportRedisIsDead report redis alive status is health
+// ReportRedisIsAlive report redis alive status is health
 func ReportRedisIsAlive() {
 	redisAliveStatus.Set(1)
 }

diff --git a/common/metrics/types.go b/common/metrics/types.go
@@ -58,17 +58,17 @@ var (
 
 // redis metrics
 var (
-	redisReadFailure = prometheus.NewCounter(prometheus.CounterOpts{
+	redisReadFailure = prometheus.NewGauge(prometheus.GaugeOpts{
 		Name: "redis_read_failure",
-		Help: "polaris exec redis read operation success",
+		Help: "polaris exec redis read operation failure",
 		ConstLabels: map[string]string{
 			"polaris_server_instance": utils.LocalHost,
 		},
 	})
 
-	redisWriteFailure = prometheus.NewCounter(prometheus.CounterOpts{
+	redisWriteFailure = prometheus.NewGauge(prometheus.GaugeOpts{
 		Name: "redis_write_failure",
-		Help: "polaris exec redis write operation success",
+		Help: "polaris exec redis write operation failure",
 		ConstLabels: map[string]string{
 			"polaris_server_instance": utils.LocalHost,
 		},

diff --git a/common/redispool/redis_pool.go b/common/redispool/redis_pool.go
@@ -222,10 +222,8 @@ func (p *Pool) Del(id string) *Resp {
 
 func (p *Pool) checkRedisDead() error {
 	if atomic.LoadUint32(&p.redisDead) == 1 {
-		metrics.ReportRedisIsDead()
 		return fmt.Errorf("redis %s is dead", p.config.KvAddr)
 	}
-	metrics.ReportRedisIsAlive()
 	return nil
 }
 
@@ -333,10 +331,12 @@ func (p *Pool) checkRedis(wg *sync.WaitGroup) {
 				}
 			}
 			if errCount >= errCountThreshold {
+				metrics.ReportRedisIsDead()
 				if atomic.CompareAndSwapUint32(&p.redisDead, 0, 1) {
 					atomic.StoreInt64(&p.recoverTimeSec, 0)
 				}
 			} else {
+				metrics.ReportRedisIsAlive()
 				if atomic.CompareAndSwapUint32(&p.redisDead, 1, 0) {
 					atomic.StoreInt64(&p.recoverTimeSec, time.Now().Unix())
 				}
@@ -393,15 +393,6 @@ func (p *Pool) handleTaskWithRetries(task *Task) *Resp {
 		}
 		log.Errorf("[RedisPool] fail to handle task %s, retry count %d, err is %v", *task, i, resp.Err)
 	}
-	if resp.Err != nil {
-		switch task.taskType {
-		case Set, Del, Sadd, Srem:
-			metrics.ReportRedisWriteFailure()
-		default:
-			metrics.ReportRedisReadFailure()
-		}
-	}
-
 	return resp
 }
 
@@ -443,6 +434,12 @@ func (p *Pool) afterHandleTask(startTime time.Time, command string, task *Task,
 	code := callResultOk
 	if resp.Err != nil {
 		code = callResultFail
+		switch task.taskType {
+		case Set, Del, Sadd, Srem:
+			metrics.ReportRedisWriteFailure()
+		default:
+			metrics.ReportRedisReadFailure()
+		}
 	}
 	if p.statis != nil {
 		_ = p.statis.AddRedisCall(command, code, costDuration.Nanoseconds())

diff --git a/plugin/cmdb/memory/fetcher.go b/plugin/cmdb/memory/fetcher.go
@@ -0,0 +1,148 @@
+/**
+ * Tencent is pleased to support the open source community by making Polaris available.
+ *
+ * Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
+ *
+ * Licensed under the BSD 3-Clause License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://opensource.org/licenses/BSD-3-Clause
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed
+ * under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+ * CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package memory
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/polarismesh/polaris/common/utils"
+)
+
+const (
+	FetchSuccess    = 200000 // 成功
+	FetchForbiden   = 401000 // 无权限
+	FetchBadRequest = 400001 // 分页参数错误
+	FetchException  = 500000 // 服务端错误
+)
+
+// IPs
+type IPs struct {
+	Hosts   map[string]IP
+	Mask    []IP
+	Backoff *IP
+}
+
+// Fetcher fetcher by get cmdb data
+type Fetcher interface {
+	GetIPs() ([]IPInfo, IPs, error)
+}
+
+type fetcher struct {
+	url   string
+	token string
+}
+
+// GetIPs get all ips from server
+func (f *fetcher) GetIPs() ([]IPInfo, IPs, error) {
+	ret := IPs{
+		Hosts: map[string]IP{},
+		Mask:  make([]IP, 0, 8),
+	}
+
+	values, err := f.getFromRemote()
+		return nil, IPs{}, err
+	}
+
+	for i := range values {
+		item := values[i]
+		if item.Type == Host {
+			data, _ := NewIP(item)
+			ret.Hosts[item.IP] = data
+			continue
+		}
+
+		if item.Type == Mask {
+			data, err := NewIP(item)
+			if err != nil {
+				return nil, ret, err
+			}
+			ret.Mask = append(ret.Mask, data)
+			continue
+		}
+
+		if item.Type == Backoff {
+			data, _ := NewIP(item)
+			ret.Backoff = &data
+		}
+	}
+
+	return values, ret, nil
+}
+
+func (f *fetcher) getFromRemote() ([]IPInfo, error) {
+	requestId := utils.NewUUID()
+
+	total := 0
+	curQuery := 0
+	pageNo := 0
+	first := true
+	values := make([]IPInfo, 0, 1)
+	for {
+		if curQuery >= total && !first {
+			break
+		}
+		pageNo++
+		body, err := json.Marshal(Request{
+			RequestID: requestId,
+			PageNo:    int64(pageNo),
+			PageSize:  100,
+		})
+		if err != nil {
+			return nil, err
+		}
+		hreq, err := http.NewRequest(http.MethodPost, f.url, bytes.NewBuffer(body))
+		if err != nil {
+			return nil, err
+		}
+		hreq.Header.Set("Authorization", f.token)
+		hreq.Header.Set("Content-Type", "application/json")
+		resp, err := http.DefaultClient.Do(hreq)
+		if err != nil {
+			return nil, err
+		}
+
+		defer resp.Body.Close()
+
+		data, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return nil, err
+		}
+
+		queryResp := &Response{}
+		if err := json.Unmarshal(data, queryResp); err != nil {
+			return nil, err
+		}
+		if queryResp.Code != FetchSuccess {
+			return nil, errors.New(queryResp.Info)
+		}
+		if total == 0 {
+			total = int(queryResp.Total)
+			values = make([]IPInfo, 0, total)
+		}
+		curQuery += int(queryResp.Size)
+		values = append(values, queryResp.Data...)
+		first = false
+	}
+
+	return values, nil
+}