Merge remote-tracking branch 'origin/master'

bcs-services/bcs-monitor/go.mod

@@ -210,6 +210,6 @@ require (
 
 replace (
 	github.com/prometheus/prometheus => github.com/prometheus/prometheus v1.8.2-0.20220308163432-03831554a519
-	github.com/thanos-io/thanos => github.com/ifooth/thanos v0.26.1-0.20221027023320-022adb0b704c
+	github.com/thanos-io/thanos => github.com/ifooth/thanos v0.26.1-0.20221117075138-a986a4c5a32f
 	github.com/vimeo/galaxycache => github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e
 )

bcs-services/bcs-monitor/go.sum

@@ -1005,8 +1005,8 @@ github.com/iancoleman/strcase v0.2.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20210905161508-09a460cdf81d/go.mod h1:aYm2/VgdVmcIU8iMfdMvDMsRAQjcfZSKFby6HOFvi/w=
-github.com/ifooth/thanos v0.26.1-0.20221027023320-022adb0b704c h1:eaWXMtFYBATqtJFws4XYC8SqlGQGMkUnam+wOcNDLEk=
-github.com/ifooth/thanos v0.26.1-0.20221027023320-022adb0b704c/go.mod h1:yE4GgQ4dRxDScBTSQGwSFJ/yzPR1q2Szu9iGCLmcFXA=
+github.com/ifooth/thanos v0.26.1-0.20221117075138-a986a4c5a32f h1:DWFJUSN/WSM+Yv6Yj+2kVgPGvq3EYrcwbSI4YRCL23U=
+github.com/ifooth/thanos v0.26.1-0.20221117075138-a986a4c5a32f/go.mod h1:yE4GgQ4dRxDScBTSQGwSFJ/yzPR1q2Szu9iGCLmcFXA=
 github.com/imdario/mergo v0.3.4/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/imdario/mergo v0.3.8/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=

bcs-services/bcs-monitor/pkg/component/client.go

@@ -21,6 +21,7 @@ import (
 	"net/http"
 	"net/url"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/dustin/go-humanize"
@@ -40,6 +41,8 @@ var (
 	maskKeys = map[string]struct{}{
 		"bk_app_secret": {},
 	}
+	clientOnce   sync.Once
+	globalClient *resty.Client
 )
 
 // restyReqToCurl curl 格式的请求日志
@@ -119,14 +122,18 @@ func restyBeforeRequestHook(c *resty.Client, r *http.Request) error {
 
 // GetClient : 新建Client, 设置公共参数，每次新建，cookies不复用
 func GetClient() *resty.Client {
-	client := resty.New().SetTimeout(timeout)
-	client = client.SetDebug(false) // 更多详情, 可以开启为 true
-	client.SetDebugBodyLimit(1024)
-	client.OnAfterResponse(restyAfterResponseHook)
-	client.SetPreRequestHook(restyBeforeRequestHook)
-	client.OnError(restyErrHook)
-	client.SetTLSClientConfig(&tls.Config{InsecureSkipVerify: true})
-	return client
+	if globalClient == nil {
+		clientOnce.Do(func() {
+			globalClient = resty.New().SetTimeout(timeout)
+			globalClient = globalClient.SetDebug(false) // 更多详情, 可以开启为 true
+			globalClient.SetDebugBodyLimit(1024)
+			globalClient.OnAfterResponse(restyAfterResponseHook)
+			globalClient.SetPreRequestHook(restyBeforeRequestHook)
+			globalClient.OnError(restyErrHook)
+			globalClient.SetTLSClientConfig(&tls.Config{InsecureSkipVerify: true})
+		})
+	}
+	return globalClient
 }
 
 // BKResult 蓝鲸返回规范的结构体

bcs-services/bcs-monitor/pkg/query/discovery_client.go

@@ -81,6 +81,7 @@ func NewDiscoveryClient(ctx context.Context, reg *prometheus.Registry, tracer op
 	dialOpts = append(dialOpts,
 		grpc.WithStreamInterceptor(grpc_retry.StreamClientInterceptor(opts...)),
 		grpc.WithUnaryInterceptor(grpc_retry.UnaryClientInterceptor(opts...)),
+		grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`), // 负载均衡
 	)
 
 	if err != nil {

bcs-services/bcs-monitor/pkg/query/middleware.go

@@ -73,11 +73,13 @@ func (t *tenantAuthMiddleware) NewHandler(handlerName string, handler http.Handl
 			return
 		}
 
+		scopeClusteID := r.Header.Get("X-Scope-ClusterId")
 		requestID := tracing.RequestIDValue(r, true)
 		logger.Infow("handle request",
 			"request_id", requestID,
 			"handler_name", handlerName,
 			"label_matchers", fmt.Sprintf("%s", labelMatchers),
+			"X-Scope-ClusterId", scopeClusteID,
 			"req", fmt.Sprintf("%s %s", r.Method, r.URL),
 		)
 
@@ -85,6 +87,7 @@ func (t *tenantAuthMiddleware) NewHandler(handlerName string, handler http.Handl
 		w.Header().Set(store.RequestIdHeaderKey(), requestID)
 
 		ctx := store.WithLabelMatchValue(r.Context(), labelMatchers)
+		ctx = store.WithScopeClusterIDValue(ctx, scopeClusteID)
 		ctx = store.WithRequestIDValue(ctx, requestID)
 		r = r.WithContext(ctx)
 		handleFunc(w, r)

bcs-services/bcs-monitor/pkg/storegw/bk_monitor/bk_monitor.go

@@ -143,18 +143,24 @@ func (s *BKMonitorStore) Series(r *storepb.SeriesRequest, srv storepb.Store_Seri
 	ctx := srv.Context()
 	klog.InfoS(clientutil.DumpPromQL(r), "request_id", store.RequestIDValue(ctx), "minTime", r.MinTime, "maxTime", r.MaxTime, "step", r.QueryHints.StepMillis)
 
-	if r.Step < 60 {
-		r.Step = 60
+	var step int64
+	if r.QueryHints.StepMillis > 0 {
+		step = r.QueryHints.StepMillis / 1000
+	}
+
+	// 最小步长
+	if step < clientutil.MinStepSeconds {
+		step = clientutil.MinStepSeconds
 	}
 
 	// 毫秒转换为秒
 	start := time.UnixMilli(r.MinTime).Unix()
 	end := time.UnixMilli(r.MaxTime).Unix()
 
-	// series 数据, 这里只查询最近1分钟
+	// series 数据, 这里只查询最近 SeriesStepDeltaSeconds
 	if r.SkipChunks {
 		end = time.Now().Unix()
-		start = end - 60
+		start = end - clientutil.SeriesStepDeltaSeconds
 	}
 
 	metricName, err := clientutil.GetLabelMatchValue("__name__", r.Matchers)
@@ -176,9 +182,14 @@ func (s *BKMonitorStore) Series(r *storepb.SeriesRequest, srv storepb.Store_Seri
 		return err
 	}
 
-	if clusterId == "" {
+	scopeClusterID := store.ClusterIDValue(ctx)
+	if clusterId == "" && scopeClusterID == "" {
 		return nil
-		// return errors.New("cluster_id is required")
+	}
+
+	// 优先使用 clusterID
+	if scopeClusterID != "" {
+		clusterId = scopeClusterID
 	}
 
 	newMatchers := make([]storepb.LabelMatcher, 0, len(r.Matchers))
@@ -206,7 +217,7 @@ func (s *BKMonitorStore) Series(r *storepb.SeriesRequest, srv storepb.Store_Seri
 		return err
 	}
 
-	promSeriesSet, err := bkmonitor_client.QueryByPromQL(srv.Context(), s.config.URL, cluster.BKBizID, start, end, r.Step,
+	promSeriesSet, err := bkmonitor_client.QueryByPromQL(srv.Context(), s.config.URL, cluster.BKBizID, start, end, step,
 		newMatchers)
 	if err != nil {
 		return err

bcs-services/bcs-monitor/pkg/storegw/clientutil/constants.go

+/*
+ * Tencent is pleased to support the open source community by making Blueking Container Service available.
+ * Copyright (C) 2019 THL A29 Limited, a Tencent company. All rights reserved.
+ * Licensed under the MIT License (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ * http://opensource.org/licenses/MIT
+ * 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 clientutil
+
+const (
+	// MinStepSeconds 最小步长, 单位秒
+	MinStepSeconds = 30
+	// SeriesStepDelta 查询 Series 的回溯步长, 单位秒
+	SeriesStepDeltaSeconds = 60 * 5
+)

bcs-services/bcs-monitor/test/bcs_monitor_query/main.go

@@ -16,6 +16,8 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-monitor/pkg/component/promclient"
@@ -35,13 +37,23 @@ func main() {
 		errCount int64
 	)
 
-	for {
-		count += 1
-		result, err := promclient.QueryInstant(ctx, rawURL, header, promql, time.Now())
-		if err != nil || len(result.Warnings) > 0 {
-			errCount += 1
-		}
-		fmt.Println("count", count, "errCount", errCount)
-		time.Sleep(time.Millisecond * 100)
+	wg := &sync.WaitGroup{}
+
+	c := 10
+	for i := 0; i < c; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for {
+				atomic.AddInt64(&count, 1)
+				result, err := promclient.QueryInstant(ctx, rawURL, header, promql, time.Now())
+				if err != nil || len(result.Warnings) > 0 {
+					atomic.AddInt64(&errCount, 1)
+				}
+				fmt.Println("count", atomic.LoadInt64(&count), "errCount", atomic.LoadInt64(&errCount))
+				time.Sleep(time.Millisecond * 100)
+			}
+		}()
 	}
+	wg.Wait()
 }