fix:es代理等待数监听功能

config.yaml

@@ -14,6 +14,8 @@ elasticSearch:
   waitTime: 5 #单位秒
   complexQueryLen: 200 #查询条件复杂长度
 
+  threshold: 13
+
   reverseProxy:
     timeout: 15
     keepAlive: 60
@@ -33,7 +35,7 @@ elasticSearch:
 
 redis:
   default: # 同步es状态至redis，供其他程序使用
-    address: 192.168.3.11:1712
+    address: 192.168.3.149:1712
 
 logger:
   path: "logs/"               # 日志文件路径。默认为空，表示关闭，仅输出到终端

internal/cmd/cmd.go

@@ -7,6 +7,7 @@ import (
 	"github.com/gogf/gf/v2/frame/g"
 	"github.com/gogf/gf/v2/net/ghttp"
 	"github.com/gogf/gf/v2/os/gcmd"
+	"strings"
 	"time"
 )
 
@@ -19,6 +20,14 @@ var (
 			s := g.Server()
 			s.Use(service.Middleware)
 			s.BindHandler("/*", func(r *ghttp.Request) {
+				if strings.Contains(r.RequestURI, "/favicon.ico") {
+					return
+				}
+				if strings.Contains(r.RequestURI, "/links/number") {
+					data := service.InformationNumber(ctx)
+					r.Response.Write(data)
+					return
+				}
 				queryLevel, now := r.GetCtxVar(consts.QueryLevelKey).Int(), time.Now()
 				rp, err := service.EsProxyManager.GetProxy(r.Context(), queryLevel)
 				r.SetCtxVar(consts.QueryWaitPoolTime, time.Now().Sub(now).Seconds()) //记录等待时长

internal/service/esQuery.go

@@ -52,8 +52,7 @@ func (*esStatusQuery) GetEsStatus(ctx context.Context) (esStatus consts.EsStatus
 		g.Log().Errorf(ctx, "请求结果异常 err:%v", err)
 		return
 	}
-	maxQueue, maxActive := 0, 0
-
+	maxQueue, maxActive, threshold := 0, 0, g.Cfg().MustGet(ctx, "elasticSearch.threshold", 13).Int()
 	for _, val := range rs.Nodes {
 		if val.ThreadPool.Search.Queue > maxQueue {
 			maxQueue = val.ThreadPool.Search.Queue
@@ -61,11 +60,14 @@ func (*esStatusQuery) GetEsStatus(ctx context.Context) (esStatus consts.EsStatus
 		if val.ThreadPool.Search.Active > maxActive {
 			maxActive = val.ThreadPool.Search.Active
 		}
+		if val.ThreadPool.Search.Largest > threshold {
+			threshold = val.ThreadPool.Search.Largest
+		}
 	}
 
-	if maxQueue == 0 && maxActive < 6 {
+	if maxQueue == 0 && maxActive < threshold/2 {
 		esStatus = consts.EsStatus_Free
-	} else if maxQueue < 6 && maxActive < 10 {
+	} else if maxQueue < threshold/2 && maxActive < threshold {
 		esStatus = consts.EsStatus_Busy
 	}
 

internal/service/manager.go

@@ -9,6 +9,8 @@ import (
 	"net/http/httputil"
 	"net/url"
 	"os"
+	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -51,7 +53,6 @@ func initManager(ctx context.Context) *esProxyManager {
 	}
 
 	go emp.UpdateEsStatus(ctx)
-
 	return emp
 }
 
@@ -81,15 +82,19 @@ func (epm *esProxyManager) UpdateEsStatus(ctx context.Context) {
 // GetProxy 获取代理对象
 func (epm *esProxyManager) GetProxy(ctx context.Context, queryLevel int) (*httputil.ReverseProxy, error) {
 	if queryLevel+int(epm.esStatus) > 2 {
+		go RejectedIncrement(ctx) //程序拒绝数
 		return nil, fmt.Errorf("server is busy")
 	}
 	g.Log().Debugf(ctx, "esProxyManager pool %+v", epm.Status())
+	Increment() //增加等待数
+	defer Decrement()
 	select {
 	case <-epm.getPool(queryLevel):
 		c := <-epm.proxyPool
 		epm.proxyPool <- c
 		return c, nil
 	case <-time.After(time.Second * time.Duration(g.Cfg().MustGet(ctx, "elasticSearch.queryState.waitTime", 5).Int())):
+		go OvertimeIncrement(ctx) //超时退出数
 		return nil, fmt.Errorf("wait time out")
 	}
 }
@@ -120,3 +125,47 @@ func (epm *esProxyManager) Status() map[string]interface{} {
 		"complex": len(epm.complexQueryPool),
 	}
 }
+
+// QueueCounter 程序等待数  RejectedCounter程序繁忙拒绝数 OvertimeCounter超时拒绝数
+var (
+	QueueCounter int64
+	sy, sn       sync.Mutex
+)
+
+func InformationNumber(ctx context.Context) map[string]interface{} {
+	rejected_count, _ := g.Redis().Get(ctx, "es_rejected_count")
+	timeout_count, _ := g.Redis().Get(ctx, "es_timeout_count")
+	return map[string]interface{}{
+		"simple":        g.Cfg().MustGet(ctx, "elasticSearch.pool.simple", 10).Int() - len(EsProxyManager.simpleQueryPool),   //简单执行数
+		"aggs":          g.Cfg().MustGet(ctx, "elasticSearch.pool.aggs", 10).Int() - len(EsProxyManager.aggsQueryPool),       //聚合执行数
+		"complex":       g.Cfg().MustGet(ctx, "elasticSearch.pool.complex", 10).Int() - len(EsProxyManager.complexQueryPool), //复杂执行数
+		"queueCount":    QueueCounter,                                                                                        //等待数
+		"rejectedCount": rejected_count.Val(),                                                                                //拒绝数
+		"timeOutCount":  timeout_count.Val(),                                                                                 //超时数
+	}
+}
+
+// RejectedIncrement queryLevel+int(epm.esStatus) > 2拒绝数
+func RejectedIncrement(ctx context.Context) {
+	sy.Lock()
+	defer sy.Unlock()
+	_, _ = g.Redis().Incr(ctx, "es_rejected_count")
+}
+
+// OvertimeIncrement 超时断开数
+func OvertimeIncrement(ctx context.Context) {
+	sn.Lock()
+	defer sn.Unlock()
+	fmt.Println("超时")
+	_, _ = g.Redis().Incr(ctx, "es_overtime_increment")
+}
+
+// Increment 程序等待数
+func Increment() {
+	atomic.AddInt64(&QueueCounter, 1)
+}
+
+// Decrement 获取到连接池 减少程序等待数
+func Decrement() {
+	atomic.AddInt64(&QueueCounter, -1)
+}

internal/service/middle.go

@@ -51,7 +51,7 @@ func getQueryLevel(detail []byte) consts.QueryLevel {
 }
 
 func LogRestrict(now time.Time, r *ghttp.Request, queryLevel consts.QueryLevel, bodyStr string) {
-	if r.RequestURI != "/" && isLoggableRequest(r) { //打印非建立链接的日志
+	if isLoggableRequest(r) { //打印非建立链接的日志
 		//请求时间大于配置时间打印
 		if time.Now().Sub(now).Seconds() > consts.LogTime || consts.LogEquity {
 			g.Log().Infof(r.Context(), "status:%d  time:%fs  req:%s   waitPool:%fs  level:%d  from:%s  query:%s", r.Response.Status, time.Since(now).Seconds(), r.RequestURI, r.GetCtxVar(consts.QueryWaitPoolTime).Float64(), queryLevel, r.GetClientIp(), bodyStr)
@@ -69,7 +69,7 @@ func isLoggableRequest(r *ghttp.Request) bool {
 		return true
 	}
 	// 根据要求判断是否打印日志
-	if !strings.Contains(method, "HEAD") && !regExp.MatchString(path) {
+	if r.RequestURI != "/" && !strings.Contains(method, "HEAD") && !regExp.MatchString(path) {
 		return true
 	}
 	return false