feat: collector 支持 Taf 指标上报 --story=118216976 (#401)

.gitignore

@@ -15,3 +15,4 @@ go.work.sum
 .idea
 .vscode
 /.gtm.yaml
+.python-version

pkg/collector/README.md

@@ -18,16 +18,17 @@ bk-collector 是一个通用的数据接收和清洗框架，依赖配置进行
 
 [receiver](./receiver): 接收层负责接收来个多个组件的上报的数据，目前已实现的组件包括：
 
-| 组件 | Http(Traces) | Http(Metrics) | Http(Logs)| Http(Profiles) | Grpc(Traces) | Grpc(Metrics) | Grpc(Logs) |
-| --- |  -- | --- | --- |----------------|--------------| --- |------------|
-| jaeger | ✅ |  |  |                |              |  |            |
-| otlp | ✅ | ✅ | ✅ |                | ✅            | ✅ | ✅          |
-| skywalking | ✅ |  ✅    |      |                |              | |            |
-| pushgateway(prometheus) | | ✅ (pb+text) | |                |              | |            |
-| remotewrite(prometheus) | | ✅ (pb+text) | |                |              | |            |
-| fta | | ✅ | |                |              | |            |
-| beat | | | ✅ |                |              | |            |
-| pyroscope | | | |        ✅             |              | |            |
+| 组件                      | Http(Traces) | Http(Metrics) | Http(Logs) | Http(Profiles) | Grpc(Traces) | Grpc(Metrics) | Grpc(Logs) | Tars(Metrics) |
+|-------------------------|--------------|---------------|------------|----------------|--------------|---------------|------------|---------------|
+| jaeger                  | ✅            |               |            |                |              |               |            |               |
+| otlp                    | ✅            | ✅             | ✅          |                | ✅            | ✅             | ✅          |               |
+| skywalking              | ✅            | ✅             |            |                |              |               |            |               |
+| pushgateway(prometheus) |              | ✅ (pb+text)   |            |                |              |               |            |               |
+| remotewrite(prometheus) |              | ✅ (pb+text)   |            |                |              |               |            |               |
+| fta                     |              | ✅             |            |                |              |               |            |               |
+| beat                    |              |               | ✅          |                |              |               |            |               |
+| pyroscope               |              |               |            | ✅              |              |               |            |               |
+| tars                    |              |               |            |                |              |               |            | ✅             |
 
 [proxy](./proxy): 接收自定指标和自定义时序数据上报。
 

pkg/collector/VERSION

@@ -1 +1 @@
-0.61.x
+0.62.x

pkg/collector/controller/register.go

@@ -34,5 +34,6 @@ import (
 	_ "github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/receiver/pyroscope"
 	_ "github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/receiver/remotewrite"
 	_ "github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/receiver/skywalking"
+	_ "github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/receiver/tars"
 	_ "github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/receiver/zipkin"
 )

pkg/collector/define/record.go

@@ -10,11 +10,15 @@
 package define
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"strings"
 	"time"
 
+	"github.com/TarsCloud/TarsGo/tars/protocol/res/propertyf"
+	"github.com/TarsCloud/TarsGo/tars/protocol/res/statf"
+	"github.com/TarsCloud/TarsGo/tars/util/current"
 	"github.com/google/pprof/profile"
 	dto "github.com/prometheus/client_model/go"
 	"github.com/prometheus/prometheus/prompb"
@@ -39,6 +43,7 @@ const (
 	SourceProxy       = "proxy"
 	SourceSkywalking  = "skywalking"
 	SourceBeat        = "beat"
+	SourceTars        = "tars"
 
 	KeyToken    = "X-BK-TOKEN"
 	KeyDataID   = "X-BK-DATA-ID"
@@ -66,6 +71,7 @@ const (
 	RecordProxy          RecordType = "proxy"
 	RecordPingserver     RecordType = "pingserver"
 	RecordBeat           RecordType = "beat"
+	RecordTars           RecordType = "tars"
 )
 
 // IntoRecordType 将字符串描述转换为 RecordType 并返回是否为 Derived 类型
@@ -98,6 +104,8 @@ func IntoRecordType(s string) (RecordType, bool) {
 		t = RecordFta
 	case RecordBeat.S():
 		t = RecordBeat
+	case RecordTars.S():
+		t = RecordTars
 	default:
 		t = RecordUndefined
 	}
@@ -114,6 +122,7 @@ const (
 	RequestGrpc    RequestType = "grpc"
 	RequestICMP    RequestType = "icmp"
 	RequestDerived RequestType = "derived"
+	RequestTars    RequestType = "tars"
 )
 
 type RequestClient struct {
@@ -151,6 +160,43 @@ type RemoteWriteData struct {
 	Timeseries []prompb.TimeSeries
 }
 
+const (
+	TarsStatType     = "stat"
+	TarsPropertyType = "property"
+)
+
+type TarsAdapter struct {
+	Name     string
+	Servant  string
+	Endpoint string
+}
+
+type TarsServerConfig struct {
+	App      string
+	Server   string
+	LogPath  string
+	LogLevel string
+	Adapters []TarsAdapter
+}
+
+type TarsData struct {
+	// 标识为 TarsStatType 或者 ProxyEvent
+	Type      string
+	Timestamp int64
+	Data      interface{}
+}
+
+// TarsPropertyData 属性统计数据
+type TarsPropertyData struct {
+	Props map[propertyf.StatPropMsgHead]propertyf.StatPropMsgBody
+}
+
+// TarsStatData 服务指标数据
+type TarsStatData struct {
+	FromClient bool
+	Stats      map[statf.StatMicMsgHead]statf.StatMicMsgBody
+}
+
 type ProxyData struct {
 	DataId      int64       `json:"data_id"`
 	AccessToken string      `json:"access_token"`
@@ -253,7 +299,7 @@ func (t Token) GetDataID(rtype RecordType) int32 {
 	switch rtype {
 	case RecordTraces, RecordTracesDerived:
 		return t.TracesDataId
-	case RecordMetrics, RecordMetricsDerived, RecordPushGateway, RecordRemoteWrite, RecordPingserver, RecordFta:
+	case RecordMetrics, RecordMetricsDerived, RecordPushGateway, RecordRemoteWrite, RecordPingserver, RecordFta, RecordTars:
 		return t.MetricsDataId
 	case RecordLogs, RecordLogsDerived:
 		return t.LogsDataId
@@ -321,6 +367,31 @@ func TokenFromGrpcMetadata(md metadata.MD) string {
 	return ""
 }
 
+// TokenFromTarsCtx 从 Tars ctx（类似 gPRC MetaData）中提取 token
+func TokenFromTarsCtx(ctx context.Context) string {
+	rc, ok := current.GetRequestContext(ctx)
+	if !ok {
+		return ""
+	}
+	token, ok := rc[KeyToken]
+	if !ok {
+		return ""
+	}
+	return token
+}
+
+// TokenFromString 从 {KeyToken}:{token}:value 中提取 token
+func TokenFromString(s string) (string, string) {
+	if !strings.HasPrefix(s, KeyToken) {
+		return s, ""
+	}
+	parts := strings.SplitN(s, ":", 3)
+	if len(parts) != 3 {
+		return s, ""
+	}
+	return parts[2], parts[1]
+}
+
 const (
 	FormatPprof = "pprof"
 	FormatJFR   = "jfr"

pkg/collector/example/.gitignore

@@ -0,0 +1,2 @@
+sidecar.yml
+fixtures/subconfig2.yml

pkg/collector/example/example.yml

@@ -160,6 +160,18 @@ bk-collector:
       middlewares:
         - "maxbytes"
 
+    # Tars Server Config
+    tars_server:
+      # 是否启动 Tars 服务
+      # default: false
+      enabled: true
+      # 传输协议，目前支持 tcp
+      # default: ""
+      transport: "tcp"
+      # 服务监听端点
+      # default: ""
+      endpoint: ":4319"
+
     components:
       jaeger:
         enabled: true
@@ -179,7 +191,8 @@ bk-collector:
         enabled: true
       beat:
         enabled: true
-
+      tars:
+        enabled: true
 
   # =============================== Processor ================================
   # name: 名称规则为 ${processor}[/${id}]，id 字段为可选项
@@ -1091,6 +1104,12 @@ bk-collector:
         - "token_checker/beat"
         - "rate_limiter/token_bucket"
 
+    - name: "tars_pipeline/common"
+      type: "tars"
+      processors:
+        - "token_checker/aes256"
+        - "rate_limiter/token_bucket"
+
   # =============================== Exporter =================================
   exporter:
     slow_send:

pkg/collector/example/tars/go.mod

@@ -0,0 +1,7 @@
+module github.com/TencentBlueKing/bkmonitor-datalink/pkg/collector/example/tars
+
+go 1.21.0
+
+require github.com/TarsCloud/TarsGo v1.4.5
+
+require go.uber.org/automaxprocs v1.5.2 // indirect

pkg/collector/example/tars/go.sum

@@ -0,0 +1,36 @@
+github.com/TarsCloud/TarsGo v1.4.5 h1:AeOILCND6p35Swnu8MRHVLWqcjzJ+M6aKi6UIfsemNM=
+github.com/TarsCloud/TarsGo v1.4.5/go.mod h1:fQITmq34rZnC0bz+KbQcHGdQUijzcmVtowXlic33jSk=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prashantv/gostub v1.1.0 h1:BTyx3RfQjRHnUWaGF9oQos79AlQ5k8WNktv7VGvVH4g=
+github.com/prashantv/gostub v1.1.0/go.mod h1:A5zLQHz7ieHGG7is6LLXLz7I8+3LZzsrV0P1IAHhP5U=
+github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc=
+github.com/rogpeppe/go-internal v1.8.0/go.mod h1:WmiCO8CzOY8rg0OYDC4/i/2WRWAB6poM+XZ2dLUbcbE=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+go.uber.org/automaxprocs v1.5.2 h1:2LxUOGiR3O6tw8ui5sZa2LAaHnsviZdVOUZw4fvbnME=
+go.uber.org/automaxprocs v1.5.2/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=