feat: bcs-monitor支持泳道特性

bcs-services/bcs-monitor/go.mod

@@ -76,6 +76,7 @@ require (
 	github.com/golang-migrate/migrate/v4 v4.17.0 // indirect
 	github.com/google/s2a-go v0.1.7 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
+	github.com/grpc-ecosystem/grpc-gateway v1.16.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
@@ -278,12 +279,13 @@ require (
 )
 
 replace (
+	github.com/Tencent/bk-bcs/bcs-common => github.com/LidolLxf/bk-bcs/bcs-common v0.0.0-20250120075906-fd5988c50f3f
 	github.com/go-resty/resty/v2 => github.com/ifooth/resty/v2 v2.0.0-20230223083514-3015979960de
 	// from github.com/thanos-io/thanos
 	github.com/prometheus/client_golang => github.com/prometheus/client_golang v1.14.0
 	github.com/prometheus/common => github.com/prometheus/common v0.34.0
 	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.20250102092401-8457964e42cb
+	github.com/thanos-io/thanos => github.com/LidolLxf/thanos v0.26.1-0.20250120075609-1676363bc0cc
 	github.com/vimeo/galaxycache => github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e
 	// from github.com/thanos-io/thanos
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp => go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.44.0

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

@@ -17,6 +17,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/Tencent/bk-bcs/bcs-common/pkg/header"
+
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-monitor/pkg/component"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-monitor/pkg/config"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-monitor/pkg/storage"
@@ -59,7 +61,8 @@ func GetProject(ctx context.Context, bcsConf *config.BCSConf, projectIDOrCode st
 	url := fmt.Sprintf("%s/bcsapi/v4/bcsproject/v1/projects/%s", bcsConf.Host, projectIDOrCode)
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
-		SetHeader("X-Project-Username", ""). // bcs_project 要求有这个header
+		SetHeaders(header.GetLaneIDByCtx(ctx)). // 泳道特性
+		SetHeader("X-Project-Username", "").    // bcs_project 要求有这个header
 		SetAuthToken(bcsConf.Token).
 		Get(url)
 

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

@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/Tencent/bk-bcs/bcs-common/pkg/header"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 
@@ -71,6 +72,7 @@ func ListLogCollectorsWithPath(ctx context.Context, clusterID, spaceUID string,
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		SetQueryParam("bcs_cluster_id", clusterID).
 		SetQueryParam("space_uid", spaceUID).
 		Get(url)
@@ -107,6 +109,7 @@ func CreateLogCollectors(ctx context.Context, req *CreateBCSCollectorReq) (*Crea
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		SetBody(req).
 		Post(url)
 
@@ -143,6 +146,7 @@ func UpdateLogCollectors(ctx context.Context, ruleID int, req *UpdateBCSCollecto
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		SetBody(req).
 		Post(url)
 
@@ -178,6 +182,7 @@ func DeleteLogCollectors(ctx context.Context, ruleID int) error {
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		Delete(url)
 
 	if err != nil {
@@ -212,6 +217,7 @@ func RetryLogCollectors(ctx context.Context, ruleID int, username string) error
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		Post(url)
 
 	if err != nil {
@@ -246,6 +252,7 @@ func StartLogCollectors(ctx context.Context, ruleID int, username string) error
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		Post(url)
 
 	if err != nil {
@@ -280,6 +287,7 @@ func StopLogCollectors(ctx context.Context, ruleID int, username string) error {
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		Post(url)
 
 	if err != nil {
@@ -314,6 +322,7 @@ func HasLog(ctx context.Context, indexSetID int) (bool, error) {
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		SetBody(map[string]interface{}{"index_set_id": indexSetID}).
 		Post(url)
 
@@ -349,6 +358,7 @@ func GetStorageClusters(ctx context.Context, spaceUID string) ([]GetStorageClust
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		SetQueryParam("space_uid", spaceUID).
 		Get(url)
 
@@ -390,6 +400,7 @@ func SwitchStorage(ctx context.Context, spaceUID, bcsClusterID string, storageCl
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		SetBody(body).
 		Post(url)
 
@@ -425,6 +436,7 @@ func GetBcsCollectorStorage(ctx context.Context, spaceUID, clusterID string) (in
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
 		SetHeader("X-Bkapi-Authorization", authInfo).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		SetQueryParam("space_uid", spaceUID).
 		SetQueryParam("bcs_cluster_id", clusterID).
 		Get(url)

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

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/Tencent/bk-bcs/bcs-common/common/blog"
+	"github.com/Tencent/bk-bcs/bcs-common/pkg/header"
 	"github.com/chonla/format"
 	"github.com/pkg/errors"
 	"github.com/prometheus/prometheus/prompb"
@@ -179,6 +180,7 @@ func QueryByPromQLRaw(ctx context.Context, rawURL, bkBizID string, start, end, s
 		SetBody(body).
 		SetHeader("X-Bkapi-Authorization", authInfo).
 		SetHeader("X-Bk-Scope-Space-Uid", fmt.Sprintf("bkcc__%s", bkBizID)). // 支持空间参数
+		SetHeaders(header.GetLaneIDByCtx(ctx)).                              // 泳道特性
 		Post(url)
 
 	if err != nil {
@@ -329,6 +331,7 @@ func GetMetricsList(ctx context.Context, host, clusterID, bizID string) ([]Metri
 	url := fmt.Sprintf("%s/query_bcs_metrics", host)
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
+		SetHeaders(header.GetLaneIDByCtx(ctx)).
 		SetHeader("X-Bkapi-Authorization", authInfo).
 		SetQueryParam("cluster_ids", clusterID).
 		SetQueryString(fmt.Sprintf("bk_biz_ids=0&bk_biz_ids=%s", bizID)).

bcs-services/bcs-monitor/pkg/component/k8sclient/pod_raw.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/Tencent/bk-bcs/bcs-common/pkg/header"
 	"github.com/pkg/errors"
 
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-monitor/pkg/component"
@@ -82,6 +83,7 @@ func GetPod(ctx context.Context, clusterID, namespace, podname string) (*Workloa
 
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
+		SetHeaders(header.GetLaneIDByCtx(ctx)). // 泳道特性
 		SetAuthToken(config.G.BCS.Token).
 		Get(url)
 
@@ -114,6 +116,7 @@ func GetNamespaces(ctx context.Context, clusterID string) ([]string, error) {
 	url := fmt.Sprintf("%s/clusters/%s/api/v1/namespaces", config.G.BCS.Host, clusterID)
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
+		SetHeaders(header.GetLaneIDByCtx(ctx)). // 泳道特性
 		SetAuthToken(config.G.BCS.Token).
 		Get(url)
 

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

@@ -22,6 +22,7 @@ import (
 	"strings"
 	"time"
 
+	pkgheader "github.com/Tencent/bk-bcs/bcs-common/pkg/header"
 	"github.com/pkg/errors"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/promql/parser"
@@ -84,6 +85,7 @@ func QueryInstant(ctx context.Context, rawURL string, header http.Header, promql
 	}
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
+		SetHeaders(pkgheader.GetLaneIDByCtx(ctx)). // 泳道特性
 		SetFormData(data).
 		SetHeaderMultiValues(header).
 		Post(rawURL)
@@ -118,6 +120,7 @@ func QueryRange(ctx context.Context, rawURL string, header http.Header, promql s
 
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
+		SetHeaders(pkgheader.GetLaneIDByCtx(ctx)). // 泳道特性
 		SetFormData(data).
 		SetHeaderMultiValues(header).
 		Post(rawURL)
@@ -222,6 +225,7 @@ func QueryLabels(ctx context.Context, rawURL string, header http.Header,
 
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
+		SetHeaders(pkgheader.GetLaneIDByCtx(ctx)). // 泳道特性
 		SetQueryParams(query).
 		SetHeaderMultiValues(header).
 		Get(rawURL)
@@ -263,6 +267,7 @@ func QueryLabelValues(ctx context.Context, rawURL string, header http.Header,
 
 	resp, err := component.GetClient().R().
 		SetContext(ctx).
+		SetHeaders(pkgheader.GetLaneIDByCtx(ctx)). // 泳道特性
 		SetQueryParams(query).
 		SetHeaderMultiValues(header).
 		Get(rawURL)

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

@@ -18,6 +18,7 @@ import (
 	"encoding/json"
 	"time"
 
+	"github.com/Tencent/bk-bcs/bcs-common/pkg/header"
 	"github.com/pkg/errors"
 	"github.com/prometheus/prometheus/prompb"
 
@@ -166,6 +167,7 @@ func (c *SuanLiClient) QueryInfo(ctx context.Context, metricName string, tagFilt
 		SetHeader("projectname", c.config.ProjectName).
 		SetHeader("token", c.config.Token).
 		SetHeader("Content-Type", "application/json").
+		SetHeaders(header.GetLaneIDByCtx(ctx)). // 泳道特性
 		SetBody(jsonData).
 		Post(rawURL)
 
@@ -213,6 +215,7 @@ func (c *SuanLiClient) QuerySecondInfo(ctx context.Context, metricName string, t
 		SetHeader("projectname", c.config.ProjectName).
 		SetHeader("token", c.config.Token).
 		SetHeader("Content-Type", "application/json").
+		SetHeaders(header.GetLaneIDByCtx(ctx)). // 泳道特性
 		SetBody(jsonData).
 		Post(rawURL)
 

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

@@ -18,6 +18,7 @@ import (
 	"net/http"
 
 	"github.com/Tencent/bk-bcs/bcs-common/common/blog"
+	"github.com/Tencent/bk-bcs/bcs-common/pkg/header"
 	"github.com/pkg/errors"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/promql/parser"
@@ -100,6 +101,9 @@ func (t *TenantAuthMiddleware) NewHandler(handlerName string, handler http.Handl
 		ctx = store.WithRequestIDValue(ctx, requestID)
 		// Traceparent 透传给grpc
 		ctx = metadata.AppendToOutgoingContext(ctx, "Traceparent", r.Header.Get("traceparent"))
+		// X-Lane 透传给grpc
+		laneKey, laneValue := header.GetLaneIDByHeader(r.Header)
+		ctx = store.WithLaneValue(ctx, laneKey, laneValue)
 		r = r.WithContext(ctx)
 		handleFunc(w, r)
 	}

bcs-services/bcs-monitor/pkg/rest/middleware/project.go

@@ -14,6 +14,7 @@ package middleware
 
 import (
 	"github.com/Tencent/bk-bcs/bcs-common/common/blog"
+	"github.com/Tencent/bk-bcs/bcs-common/pkg/header"
 	"github.com/gin-gonic/gin"
 
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-monitor/pkg/component/bcs"
@@ -35,7 +36,10 @@ func ProjectParse() gin.HandlerFunc {
 		if len(restContext.ProjectCode) != 0 {
 			projectIDOrCode = restContext.ProjectCode
 		}
-		project, err := bcs.GetProject(c.Request.Context(), config.G.BCS, projectIDOrCode)
+
+		ctx := header.WithLaneIdCtx(c.Request.Context(), c.Request.Header)
+		c.Request = c.Request.WithContext(ctx)
+		project, err := bcs.GetProject(ctx, config.G.BCS, projectIDOrCode)
 		if err != nil {
 			blog.Errorf("get project error for project %s, error: %s", projectIDOrCode, err.Error())
 			rest.AbortWithBadRequestError(restContext, err)

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

@@ -21,6 +21,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/Tencent/bk-bcs/bcs-common/pkg/header"
+
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-monitor/pkg/api/metrics/query"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-monitor/pkg/component"
 	"github.com/Tencent/bk-bcs/bcs-services/bcs-monitor/pkg/config"
@@ -50,7 +52,8 @@ func main() {
 			defer wg.Done()
 			for {
 				atomic.AddInt64(&count, 1)
-				resp, err := component.GetClient().R().SetContext(ctx).SetAuthToken(config.G.BCS.Token).Get(rawURL)
+				resp, err := component.GetClient().R().SetContext(ctx).SetHeaders(header.GetLaneIDByCtx(ctx)).
+					SetAuthToken(config.G.BCS.Token).Get(rawURL)
 				if err != nil || resp.StatusCode() != http.StatusOK {
 					atomic.AddInt64(&errCount, 1)
 					atomic.AddInt64(&respErrCount, 1)