feat: bkbase apigw 认证信息修改 --story=117623329 (#352)

pkg/bk-monitor-worker/internal/metadata/task/space.go

@@ -210,7 +210,7 @@ func PushAndPublishSpaceRouterInfo(ctx context.Context, t *t.Task) error {
 		logger.Errorf("PushAndPublishSpaceRouterInfo task error, push table detail error: %s")
 	}
 
-	if err := pusher.PushEsTableIdDetail([]string{}, true); err!= nil {
+	if err := pusher.PushEsTableIdDetail([]string{}, true); err != nil {
 		return err
 	}
 

pkg/unify-query/curl/mock.go

@@ -12,9 +12,9 @@ package curl
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"io"
 	"net/http"
-	"net/url"
 
 	"github.com/pkg/errors"
 
@@ -36,6 +36,40 @@ type TestCurl struct {
 	Params []byte
 }
 
+func (c *TestCurl) WithDecoder(decoder func(ctx context.Context, reader io.Reader, resp interface{}) (int, error)) {
+	//TODO implement me
+	panic("implement me")
+}
+
+func (c *TestCurl) hashOption(opt Options) string {
+	s := opt.UrlPath + string(opt.Body)
+	return s
+}
+
+func (c *TestCurl) Request(ctx context.Context, method string, opt Options, res interface{}) (int, error) {
+	c.log.Infof(ctx, "http %s: %s", method, opt.UrlPath)
+
+	c.Url = opt.UrlPath
+	c.Params = opt.Body
+
+	var (
+		err error
+		out string
+		ok  bool
+	)
+
+	hashKey := c.hashOption(opt)
+	if out, ok = c.data[hashKey]; ok {
+		err = json.Unmarshal([]byte(out), res)
+	} else {
+		err = errors.New("mock data is not exists: " + hashKey)
+	}
+
+	return len(out), err
+}
+
+var _ Curl = &TestCurl{}
+
 func (c *TestCurl) resp(body string) *http.Response {
 	return &http.Response{
 		Status:        "200 OK",
@@ -48,16 +82,3 @@ func (c *TestCurl) resp(body string) *http.Response {
 		Header:        make(http.Header, 0),
 	}
 }
-
-func (c *TestCurl) Request(ctx context.Context, method string, opt Options) (*http.Response, error) {
-	c.log.Infof(ctx, "http %s: %s", method, opt.UrlPath)
-
-	c.Url = opt.UrlPath
-	c.Params = opt.Body
-
-	if res, ok := c.data[opt.UrlPath]; ok {
-		return c.resp(res), nil
-	} else {
-		return nil, errors.New("mock data is not exists: " + url.QueryEscape(opt.UrlPath))
-	}
-}

pkg/unify-query/mock/mock.go

@@ -12,7 +12,6 @@ package mock
 import (
 	"context"
 	"fmt"
-	"os"
 	"sync"
 
 	goRedis "github.com/go-redis/redis/v8"
@@ -34,7 +33,8 @@ var (
 
 func Init() {
 	once.Do(func() {
-		config.CustomConfigFilePath = os.Getenv("UNIFY-QUERY-CONFIG-FILE-PATH")
+		config.CustomConfigFilePath = `../../dist/local/unify-query.yaml`
+		config.InitConfig()
 		log.InitTestLogger()
 
 		metadata.InitMetadata()

pkg/unify-query/tsdb/bksql/client.go

@@ -50,11 +50,9 @@ func (c *Client) curl(ctx context.Context, method, url, sql string, res *Result,
 	}
 	params := &Params{
 		BkdataAuthenticationMethod: c.BkdataAuthenticationMethod,
-		BkUsername:                 c.BkUsername,
 		BkAppCode:                  c.BkAppCode,
 		PreferStorage:              c.PreferStorage,
 		BkdataDataToken:            c.BkdataDataToken,
-		BkAppSecret:                c.BkAppSecret,
 	}
 
 	if sql != "" {
@@ -76,7 +74,8 @@ func (c *Client) curl(ctx context.Context, method, url, sql string, res *Result,
 			UrlPath: url,
 			Body:    body,
 			Headers: map[string]string{
-				ContentType: c.ContentType,
+				ContentType:   c.ContentType,
+				Authorization: c.authorization(),
 			},
 		},
 		res,
@@ -97,6 +96,16 @@ func (c *Client) curl(ctx context.Context, method, url, sql string, res *Result,
 	return nil
 }
 
+func (c *Client) authorization() string {
+	auth := fmt.Sprintf(
+		`{"bk_username": "%s", "bk_app_code": "%s", "bk_app_secret": "%s"}`,
+		c.BkUsername,
+		c.BkAppCode,
+		c.BkAppSecret,
+	)
+	return auth
+}
+
 func (c *Client) QuerySync(ctx context.Context, sql string, span *trace.Span) *Result {
 	data := &QuerySyncResultData{}
 	res := c.response(data)

pkg/unify-query/tsdb/bksql/client_test.go

@@ -11,22 +11,25 @@ package bksql
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/unify-query/curl"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/unify-query/log"
 	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/unify-query/mock"
+	"github.com/TencentBlueKing/bkmonitor-datalink/pkg/unify-query/trace"
 )
 
 var (
-	url    string
-	code   string
-	token  string
-	secret string
+	address string
+	code    string
+	token   string
+	secret  string
 
 	client *Client
 
@@ -35,13 +38,13 @@ var (
 
 func mockClient() *Client {
 	once.Do(func() {
-		url = "http://127.0.0.1"
-		code = "code"
-		secret = "secret"
-		token = "token"
+		address = viper.GetString("mock.bk_sql.address")
+		code = viper.GetString("mock.bk_sql.code")
+		secret = viper.GetString("mock.bk_sql.secret")
+		token = viper.GetString("mock.bk_sql.token")
 
 		client = &Client{
-			Address:                    url,
+			Address:                    address,
 			BkdataAuthenticationMethod: "token",
 			BkUsername:                 BkUserName,
 			BkAppCode:                  code,
@@ -59,42 +62,105 @@ func mockClient() *Client {
 	return client
 }
 
-func TestClient_QueryAsync(t *testing.T) {
+func TestClient_QuerySync(t *testing.T) {
 	ctx := context.Background()
 
 	mock.Init()
 	mockClient()
 
-	res := client.QueryAsync(ctx, `SELECT * FROM 132_hander_opmon_avg WHERE dtEventTimeStamp >= 1700745780000 AND dtEventTimeStamp < 1700746080000 LIMIT 10`, nil)
+	ctx, span := trace.NewSpan(ctx, "test_client")
 
-	assert.Equal(t, res.Code, StatusOK)
-	d, ok := res.Data.(*QueryAsyncData)
+	end := time.Now()
+	start := end.Add(time.Minute * -5)
+
+	res := client.QuerySync(
+		ctx,
+		fmt.Sprintf(
+			`SELECT * FROM 132_hander_opmon_avg WHERE dtEventTimeStamp >= %d AND dtEventTimeStamp < %d LIMIT 10`,
+			start.UnixMilli(),
+			end.UnixMilli(),
+		),
+		span,
+	)
+
+	assert.Equal(t, StatusOK, res.Code)
+	d, ok := res.Data.(*QuerySyncResultData)
 	assert.True(t, ok)
 
 	if d != nil {
-		assert.NotEmpty(t, d.QueryId)
+		assert.NotEmpty(t, d.List)
 		log.Infof(ctx, "%+v", d)
 	}
 }
 
-func TestClient_QueryAsyncState(t *testing.T) {
+func TestClient_QueryAsync(t *testing.T) {
 	ctx := context.Background()
 
 	mock.Init()
 	mockClient()
 
-	res := client.QueryAsyncState(ctx, "BK912760164455546880", nil)
+	ctx, span := trace.NewSpan(ctx, "test_client")
 
-	assert.Equal(t, res.Code, StatusOK)
-	d, ok := res.Data.(*QueryAsyncStateData)
+	end := time.Now()
+	start := end.Add(time.Minute * -5)
+
+	res := client.QueryAsync(
+		ctx,
+		fmt.Sprintf(
+			`SELECT * FROM 132_hander_opmon_avg WHERE dtEventTimeStamp >= %d AND dtEventTimeStamp < %d LIMIT 10`,
+			start.UnixMilli(),
+			end.UnixMilli(),
+		),
+		span,
+	)
+
+	assert.Equal(t, StatusOK, res.Code)
+	d, ok := res.Data.(*QueryAsyncData)
 	assert.True(t, ok)
 
 	if d != nil {
-		assert.NotEmpty(t, d.State)
+		assert.NotEmpty(t, d.QueryId)
 		log.Infof(ctx, "%+v", d)
 	}
 }
 
+func TestClient_QueryAsyncState(t *testing.T) {
+	ctx := context.Background()
+
+	mock.Init()
+	mockClient()
+
+	ctx, span := trace.NewSpan(ctx, "test_client")
+
+	end := time.Now()
+	start := end.Add(time.Minute * -5)
+
+	res := client.QueryAsync(
+		ctx,
+		fmt.Sprintf(
+			`SELECT * FROM 132_hander_opmon_avg WHERE dtEventTimeStamp >= %d AND dtEventTimeStamp < %d LIMIT 10`,
+			start.UnixMilli(),
+			end.UnixMilli(),
+		),
+		span,
+	)
+
+	assert.Equal(t, StatusOK, res.Code)
+	data := res.Data.(*QueryAsyncData)
+	if data != nil {
+		r := client.QueryAsyncState(ctx, data.QueryId, span)
+
+		assert.Equal(t, r.Code, StatusOK)
+		d, ok := r.Data.(*QueryAsyncStateData)
+		assert.True(t, ok)
+
+		if d != nil {
+			assert.NotEmpty(t, d.State)
+			log.Infof(ctx, "%+v", d)
+		}
+	}
+}
+
 func TestClient_QueryAsyncResult(t *testing.T) {
 	ctx := context.Background()
 

pkg/unify-query/tsdb/bksql/struct.go

@@ -33,7 +33,8 @@ const (
 	RUNNING  = "running"
 	FINISHED = "finished"
 
-	ContentType = "Content-Type"
+	ContentType   = "Content-Type"
+	Authorization = "X-Bkapi-Authorization"
 )
 
 var (
@@ -51,11 +52,9 @@ var (
 type Params struct {
 	SQL                        string `json:"sql"`
 	BkdataAuthenticationMethod string `json:"bkdata_authentication_method"`
-	BkUsername                 string `json:"bk_username"`
 	BkAppCode                  string `json:"bk_app_code"`
 	PreferStorage              string `json:"prefer_storage"`
 	BkdataDataToken            string `json:"bkdata_data_token"`
-	BkAppSecret                string `json:"bk_app_secret"`
 }
 
 type Result struct {

pkg/unify-query/tsdb/victoriaMetrics/instance.go

@@ -34,7 +34,8 @@ const (
 	BkUserName    = "admin"
 	PreferStorage = "vm"
 
-	ContentType = "Content-Type"
+	ContentType   = "Content-Type"
+	Authorization = "X-Bkapi-Authorization"
 
 	APISeries      = "series"
 	APILabelNames  = "labels"
@@ -74,6 +75,16 @@ type Instance struct {
 
 var _ tsdb.Instance = (*Instance)(nil)
 
+func (i *Instance) authorization() string {
+	auth := fmt.Sprintf(
+		`{"bk_username": "%s", "bk_app_code": "%s", "bk_app_secret": "%s"}`,
+		BkUserName,
+		i.Code,
+		i.Secret,
+	)
+	return auth
+}
+
 func (i *Instance) vectorFormat(ctx context.Context, resp *VmResponse, span *trace.Span) (promql.Vector, error) {
 	if !resp.Result {
 		return nil, fmt.Errorf(
@@ -309,11 +320,9 @@ func (i *Instance) vmQuery(
 	params := &Params{
 		SQL:                        sql,
 		BkdataAuthenticationMethod: i.AuthenticationMethod,
-		BkUsername:                 BkUserName,
 		BkAppCode:                  i.Code,
 		PreferStorage:              PreferStorage,
 		BkdataDataToken:            i.Token,
-		BkAppSecret:                i.Secret,
 	}
 	body, err := json.Marshal(params)
 	if err != nil {
@@ -341,7 +350,8 @@ func (i *Instance) vmQuery(
 			UrlPath: address,
 			Body:    body,
 			Headers: map[string]string{
-				ContentType: i.ContentType,
+				ContentType:   i.ContentType,
+				Authorization: i.authorization(),
 			},
 		},
 		data,