写入到es

TrackContext.go

@@ -5,15 +5,16 @@ import (
 	"github.com/farseer-go/fs/parse"
 	"github.com/farseer-go/fs/snowflake"
 	"github.com/farseer-go/linkTrace/eumLinkType"
+	"github.com/farseer-go/queue"
 	"time"
 )
 
-type TrackContext struct {
+type TraceContext struct {
 	ParentAppName   string                                 // 上游应用
 	TraceId         int64                                  // 上下文ID
 	StartTs         int64                                  // 调用开始时间戳
 	EndTs           int64                                  // 调用结束时间戳
-	UseTs           int64                                  // 总共使用时间毫秒
+	UseTs           time.Duration                          // 总共使用时间毫秒
 	LinkType        eumLinkType.Enum                       // 状态码
 	Domain          string                                 // 请求域名
 	Path            string                                 // 请求地址
@@ -28,12 +29,12 @@ type TrackContext struct {
 	ExceptionDetail ExceptionDetail                        // 是否执行异常
 }
 
-func NewWebApi(domain string, path string, method string, contentType string, headerDictionary collections.ReadonlyDictionary[string, string], requestBody string, requestIp string) *TrackContext {
+func NewWebApi(domain string, path string, method string, contentType string, headerDictionary collections.ReadonlyDictionary[string, string], requestBody string, requestIp string) *TraceContext {
 	traceId := parse.ToInt64(headerDictionary.GetValue("TraceId"))
 	if traceId == 0 {
 		traceId = snowflake.GenerateId()
 	}
-	return &TrackContext{
+	return &TraceContext{
 		ParentAppName: headerDictionary.GetValue("AppName"),
 		TraceId:       traceId,
 		StartTs:       time.Now().UnixMicro(),
@@ -51,9 +52,14 @@ func NewWebApi(domain string, path string, method string, contentType string, he
 }
 
 // End 结束当前链路
-func (receiver *TrackContext) End() {
+func (receiver *TraceContext) End() {
 	receiver.EndTs = time.Now().UnixMicro()
-	receiver.UseTs = receiver.EndTs - receiver.StartTs
+	receiver.UseTs = time.Duration(receiver.EndTs-receiver.StartTs) * time.Microsecond
+
+	// 启用了链路追踪后，把数据写入到本地队列中
+	if Enable {
+		queue.Push("TraceContext", receiver)
+	}
 }
 
 type LinkTraceDetail struct {

esContext.go

@@ -0,0 +1,15 @@
+package linkTrace
+
+import "github.com/farseer-go/elasticSearch"
+
+var ESContext *esContext
+
+// EsContext 链路追踪上下文
+type esContext struct {
+	TraceContext elasticSearch.IndexSet[TraceContext] `es:"index=linktrace_yyyy_MM;alias=linktrace;shards=1;replicas=0;refresh=3"`
+}
+
+// initEsContext 初始化上下文
+func initEsContext() {
+	ESContext = elasticSearch.NewContext[esContext]("LinkTrace")
+}

global.go

@@ -5,14 +5,14 @@ import (
 )
 
 // CurTraceContext 当前请求的Trace上下文
-var curTraceContext routine.ThreadLocal[*TrackContext]
+var curTraceContext = routine.NewInheritableThreadLocal[*TraceContext]()
 
 // GetCurTrace 获取当前TrackContext
-func GetCurTrace() *TrackContext {
+func GetCurTrace() *TraceContext {
 	return curTraceContext.Get()
 }
 
 // SetCurTrace 设置当前TrackContext
-func SetCurTrace(context *TrackContext) {
+func SetCurTrace(context *TraceContext) {
 	curTraceContext.Set(context)
 }

go.mod

@@ -3,11 +3,22 @@ module github.com/farseer-go/linkTrace
 go 1.19
 
 require (
-	github.com/eiannone/keyboard v0.0.0-20220611211555-0d226195f203
+	github.com/farseer-go/collections v0.8.0
+	github.com/farseer-go/elasticSearch v0.8.0
+	github.com/farseer-go/fs v0.8.0
+	github.com/farseer-go/mapper v0.8.0
+	github.com/farseer-go/queue v0.8.0
 	github.com/timandy/routine v1.1.1
 )
 
 require (
+	github.com/devfeel/mapper v0.7.13 // indirect
+	github.com/eiannone/keyboard v0.0.0-20220611211555-0d226195f203 // indirect
+	github.com/josharian/intern v1.0.0 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
+	github.com/olivere/elastic/v7 v7.0.32 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/stretchr/testify v1.8.4 // indirect
-	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
+	golang.org/x/sys v0.2.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

module.go

@@ -0,0 +1,26 @@
+package linkTrace
+
+import (
+	"github.com/farseer-go/elasticSearch"
+	"github.com/farseer-go/fs/modules"
+	"github.com/farseer-go/queue"
+)
+
+// Enable 是否启用
+var Enable bool
+
+type Module struct {
+}
+
+func (module Module) DependsModule() []modules.FarseerModule {
+	return []modules.FarseerModule{queue.Module{}, elasticSearch.Module{}}
+}
+
+func (module Module) PreInitialize() {
+	Enable = true
+}
+
+func (module Module) PostInitialize() {
+	initEsContext()
+	queue.Subscribe("TraceContext", "SaveTraceContext", 1000, saveTraceContextConsumer)
+}

saveTraceContext.go

@@ -0,0 +1,14 @@
+package linkTrace
+
+import (
+	"github.com/farseer-go/collections"
+	"github.com/farseer-go/fs/flog"
+	"github.com/farseer-go/mapper"
+)
+
+func saveTraceContextConsumer(subscribeName string, lstMessage collections.ListAny, remainingCount int) {
+	lstTraceContext := mapper.ToList[TraceContext](lstMessage)
+	err := ESContext.TraceContext.InsertList(lstTraceContext)
+	_ = flog.Error(err)
+	return
+}