apm_tracing.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698
  1. package tracing
  2. import (
  3. "context"
  4. "fmt"
  5. klog "github.com/sirupsen/logrus"
  6. "os"
  7. "strconv"
  8. "strings"
  9. "sync"
  10. "sync/atomic"
  11. "time"
  12. "github.com/coroot/coroot-node-agent/ebpftracer"
  13. "github.com/coroot/coroot-node-agent/ebpftracer/l7"
  14. "github.com/coroot/coroot-node-agent/utils"
  15. . "github.com/coroot/coroot-node-agent/utils/modelse"
  16. "go.opentelemetry.io/otel/attribute"
  17. "go.opentelemetry.io/otel/codes"
  18. semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
  19. "go.opentelemetry.io/otel/trace"
  20. "inet.af/netaddr"
  21. )
  22. /**
  23. * Trace
  24. */
  25. func (t *Trace) setContext(ctx context.Context) {
  26. t.lock.Lock()
  27. defer t.lock.Unlock()
  28. t.ctx = ctx
  29. }
  30. func (t *Trace) setSpan(span trace.Span) {
  31. t.lock.Lock()
  32. defer t.lock.Unlock()
  33. t.span = span
  34. }
  35. type TimeMap struct {
  36. Time uint64
  37. Type int
  38. Map *ebpftracer.StackFunEvent
  39. }
  40. //func (t *Trace) buildFun() {
  41. // mapSlice := []TimeMap{}
  42. // for i, v := range t.stack {
  43. // timeStartMap := TimeMap{}
  44. // if v.StackEvent.Location == 0 {
  45. // timeStartMap = TimeMap{
  46. // Time: v.StackEvent.TimeNsStart,
  47. // Type: 0,
  48. // Map: &t.stack[i],
  49. // }
  50. // } else {
  51. // timeStartMap = TimeMap{
  52. // Time: v.StackEvent.TimeNsEnd,
  53. // Type: 1,
  54. // Map: &t.stack[i],
  55. // }
  56. // }
  57. // mapSlice = append(mapSlice, timeStartMap)
  58. // }
  59. // sort.Slice(mapSlice, func(i, j int) bool {
  60. // return mapSlice[i].Time < mapSlice[j].Time
  61. // })
  62. //
  63. // funStack := []TimeMap{}
  64. //
  65. // currentfunNum := 1
  66. //
  67. // // for k, v := range mapSlice {
  68. // // fmt.Println("---SliceSliceindex", k, "value", v.Time, v.Type, v.Map.Uprobe.Funcname, v.Map.StackEvent.Nid)
  69. // // }
  70. //
  71. // mapSliceLen := len(mapSlice)
  72. // for k, v := range mapSlice {
  73. // // fmt.Println("SliceSliceindex", k, "value", v.Time, v.Type, v.Map.Uprobe.Funcname, v.Map.StackEvent.Nid)
  74. // if v.Type == 0 {
  75. // // 函数入口
  76. // funStack = append(funStack, v)
  77. // } else if v.Type == 1 {
  78. // // 函数出口
  79. // len := len(funStack)
  80. // if len < 1 {
  81. // fmt.Printf("buildFun ErrorError return before enter: %v\n", v)
  82. // continue
  83. // }
  84. // currnt := funStack[len-1]
  85. // if currnt.Map.StackEvent.Location != 0 {
  86. // fmt.Printf("currnt StackEvent ErrorError is not enter: %v\n", v)
  87. // continue
  88. // }
  89. // if k < mapSliceLen-1 && len >= 2 {
  90. // nextfun := mapSlice[k+1]
  91. // preCurrnt := funStack[len-2]
  92. // // // 处理 .netcore 多次 returun
  93. // // 下一个事件是 return 并且函数名跟当前事件是一样的,且上一个函数不是当前函数
  94. // if nextfun.Map.StackEvent.Location == 1 && nextfun.Map.Uprobe.Funcname == currnt.Map.Uprobe.Funcname && preCurrnt.Map.Uprobe.Funcname != currnt.Map.Uprobe.Funcname {
  95. // currentfunNum++
  96. // continue
  97. // }
  98. // }
  99. // funStack = funStack[:len-1]
  100. // duration := v.Map.StackEvent.TimeNsEnd - currnt.Map.StackEvent.TimeNsStart
  101. // t.FuncTraceQuery(currnt.Map.Uprobe.Funcname, time.Duration(duration), currnt.Map.StackEvent.TimeNsStart, v.Map.StackEvent.TimeNsEnd, currentfunNum)
  102. // currentfunNum = 1
  103. // }
  104. // }
  105. //}
  106. func (t *Trace) startReady() {
  107. t.lock.Lock()
  108. defer t.lock.Unlock()
  109. t.startEventReady = true
  110. }
  111. func (t *Trace) endReadyEvent(needCount uint32) {
  112. t.lock.Lock()
  113. defer t.lock.Unlock()
  114. t.endEventReady = true
  115. t.needEventCount = needCount
  116. }
  117. func (t *Trace) AllEventReady(traceID uint64) bool {
  118. klog.Debugf("[AllEventReady (current/need)|start|end|traceid](%d/%d)%v|%v|%d", *t.currenEventCount, t.needEventCount, t.startEventReady, t.endEventReady, traceID)
  119. return t.startEventReady && t.endEventReady && *t.currenEventCount >= t.needEventCount
  120. }
  121. func (t *Trace) TraceStartEvent(method, path, sn string, sport uint16, status l7.Status, addr netaddr.IPPort, pid uint32, appInfo AppInfo, container_id string) {
  122. t.span.SetAttributes(semconv.HTTPURL(fmt.Sprintf("http://%s:%d%s", sn, sport, path)),
  123. semconv.HTTPMethod(method),
  124. attribute.String("http.uri", path))
  125. if status > 399 {
  126. t.span.SetStatus(codes.Error, "")
  127. }
  128. //host := fmt.Sprintf("%s:%s", sn, sport)
  129. t.destination = addr
  130. t.commonAttrs = []attribute.KeyValue{
  131. semconv.NetPeerName(sn),
  132. semconv.NetPeerPort(int(sport)),
  133. // buildAppMapFromEvent
  134. attribute.Int("server.code_type", appInfo.CodeType.Int()),
  135. attribute.String("server.app_name", appInfo.AppName),
  136. attribute.String("server.service_name", appInfo.ServiceType),
  137. attribute.Int64("server.app_id", appInfo.AppIdHash.IntVal),
  138. attribute.Int64("server.agent_id", appInfo.AgentId),
  139. attribute.Int64("server.instance_id", appInfo.InstanceIdHash.IntVal),
  140. attribute.Int("server.pid", int(pid)),
  141. attribute.String("server.container_id", container_id),
  142. }
  143. t.span.SetAttributes(t.commonAttrs...)
  144. t.startReady()
  145. }
  146. // set context span
  147. func (t *Trace) CreateRootSpan(traceId uint64) {
  148. traceIdStr := strconv.Itoa(int(traceId))
  149. ctx, span := tracer(t.containerId).Start(context.Background(), traceIdStr, trace.WithSpanKind(trace.SpanKindClient))
  150. t.setContext(ctx)
  151. t.setSpan(span)
  152. }
  153. // Deprecated: use TraceStartEvent instead.
  154. func (t *Trace) TraceStart(method, path string, status l7.Status, duration time.Duration) {
  155. if t == nil || method == "" {
  156. return
  157. }
  158. t.createParentSpan("APPLICATION", duration, status >= 400,
  159. semconv.HTTPURL(fmt.Sprintf("http://%s%s", t.destination.String(), path)),
  160. semconv.HTTPMethod(method),
  161. //semconv.HTTPStatusCode(int(status)),
  162. attribute.String("http.uri", path),
  163. )
  164. }
  165. // Deprecated: use TraceEndEvent instead.
  166. func (t *Trace) TraceEnd(r *l7.RequestData) {
  167. if t == nil {
  168. return
  169. }
  170. t.span.SetAttributes(
  171. semconv.HTTPStatusCode(int(r.Status)),
  172. attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
  173. )
  174. CalledId, err := strconv.ParseInt(r.ParentSpanContext.CalledId, 10, 64)
  175. if err == nil && CalledId != 0 {
  176. t.span.SetAttributes(attribute.Int64("server.called_id", CalledId))
  177. }
  178. InstanceIdFrom, err := strconv.ParseInt(r.ParentSpanContext.InstanceIdFrom, 10, 64)
  179. if err == nil && InstanceIdFrom != 0 {
  180. t.span.SetAttributes(attribute.Int64("server.instance_id_from", InstanceIdFrom))
  181. }
  182. AppIdFrom, err := strconv.ParseInt(r.ParentSpanContext.AppIdFrom, 10, 64)
  183. if err == nil && AppIdFrom != 0 {
  184. t.span.SetAttributes(attribute.Int64("server.app_id_from", AppIdFrom))
  185. }
  186. if r.ParentSpanContext.SpanIdFrom != "0000000000000000" {
  187. t.span.SetAttributes(attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
  188. }
  189. // klog.Debugf("r.ParentSpanContext.TypeFrom tpp %s", r.ParentSpanContext.TypeFrom)
  190. if r.ParentSpanContext.TypeFrom == "10" {
  191. t.span.SetAttributes(attribute.String("server.type_from", "SDK"))
  192. t.span.SetAttributes(attribute.String("server.app_id_from", "-1"))
  193. t.span.SetAttributes(attribute.String("server.span_id_from", r.ParentSpanContext.TraceIdFrom))
  194. t.span.SetAttributes(attribute.String("server.instance_id_from", "-1"))
  195. t.span.SetAttributes(attribute.String("server.call_id", "-1"))
  196. }
  197. // for _, v := range t.stack {
  198. // fmt.Printf("TraceEndTraceEndTraceEnd%s\n", v)
  199. // }
  200. //t.buildFun()
  201. t.span.End(trace.WithTimestamp(time.Now()))
  202. }
  203. // 新增结束事件
  204. func (t *Trace) TraceEndEvent(r *l7.RequestData) {
  205. if t == nil {
  206. return
  207. }
  208. var attr []attribute.KeyValue
  209. attr = append(attr,
  210. semconv.HTTPStatusCode(int(r.Status)),
  211. attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
  212. )
  213. //t.span.SetAttributes(
  214. // semconv.HTTPStatusCode(int(r.Status)),
  215. // attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
  216. //)
  217. calledId, err := strconv.ParseInt(r.ParentSpanContext.CalledId, 10, 64)
  218. if err == nil && calledId != 0 {
  219. attr = append(attr, attribute.Int64("server.called_id", calledId))
  220. //t.span.SetAttributes(attribute.Int64("server.called_id", CalledId))
  221. }
  222. instanceIdFrom, err := strconv.ParseInt(r.ParentSpanContext.InstanceIdFrom, 10, 64)
  223. if err == nil && instanceIdFrom != 0 {
  224. attr = append(attr, attribute.Int64("server.instance_id_from", instanceIdFrom))
  225. //t.span.SetAttributes(attribute.Int64("server.instance_id_from", InstanceIdFrom))
  226. }
  227. appIdFrom, err := strconv.ParseInt(r.ParentSpanContext.AppIdFrom, 10, 64)
  228. if err == nil && appIdFrom != 0 {
  229. attr = append(attr, attribute.Int64("server.app_id_from", appIdFrom))
  230. //t.span.SetAttributes(attribute.Int64("server.app_id_from", AppIdFrom))
  231. }
  232. if r.ParentSpanContext.SpanIdFrom != "0000000000000000" {
  233. attr = append(attr, attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
  234. //t.span.SetAttributes(attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
  235. }
  236. // klog.Debugf("r.ParentSpanContext.TypeFrom event tpp %s", r.ParentSpanContext.TypeFrom)
  237. if r.ParentSpanContext.TypeFrom == "10" {
  238. attr = append(attr, attribute.String("server.type_from", "SDK"))
  239. attr = append(attr, attribute.String("server.app_id_from", "-1"))
  240. attr = append(attr, attribute.String("server.span_id_from", r.ParentSpanContext.TraceIdFrom))
  241. attr = append(attr, attribute.String("server.instance_id_from", "-1"))
  242. attr = append(attr, attribute.String("server.call_id", "-1"))
  243. } else if r.ParentSpanContext.TypeFrom == "02" {
  244. attr = append(attr, attribute.String("server.type_from", "Nodejs"))
  245. }
  246. attr = append(attr, attribute.String("server.src_addr", r.DAddr.String()))
  247. attr = append(attr, attribute.String("server.dst_addr", r.SAddr.String()))
  248. klog.Debugf("TraceEndEvent SAddr.String %s", r.SAddr.String())
  249. klog.Debugf("TraceEndEvent DAddr.String %s", r.DAddr.String())
  250. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  251. t.span.SetAttributes(attr...)
  252. t.endReadyEvent(r.EventCount)
  253. }
  254. func (t *Trace) appendTimestamp(attr *[]attribute.KeyValue, s, e uint64, d int64) {
  255. *attr = append(*attr,
  256. attribute.Int64("time.start_at", utils.KtimeToTimestamp(s)),
  257. attribute.Int64("time.end_at", utils.KtimeToTimestamp(e)),
  258. attribute.Int64("time.duration", d),
  259. )
  260. }
  261. func (t *Trace) createParentSpan(name string, duration time.Duration, error bool, attrs ...attribute.KeyValue) {
  262. end := time.Now()
  263. start := end.Add(-duration)
  264. ctx, span := tracer(t.containerId).Start(context.Background(), name, trace.WithTimestamp(start), trace.WithSpanKind(trace.SpanKindClient))
  265. span.SetAttributes(attrs...)
  266. span.SetAttributes(t.commonAttrs...)
  267. if error {
  268. span.SetStatus(codes.Error, "")
  269. }
  270. t.setContext(ctx)
  271. t.setSpan(span)
  272. }
  273. func (t *Trace) SendEvent() {
  274. t.span.End()
  275. }
  276. func (t *Trace) GetSpan() trace.Span {
  277. return t.span
  278. }
  279. func (t *Trace) createTraceEvent(name string, eventType int, l7Type int, attrs ...attribute.KeyValue) {
  280. t.span.AddEventApm(name, eventType, l7Type, trace.WithAttributes(attrs...))
  281. //atomic.AddUint32(t.currenEventCount, 1)
  282. }
  283. func (t *Trace) addEvent() {
  284. atomic.AddUint32(t.currenEventCount, 1)
  285. }
  286. func (t *Trace) createTraceSpan(name string, duration time.Duration, error bool, attrs ...attribute.KeyValue) {
  287. end := time.Now()
  288. start := end.Add(-duration)
  289. //fmt.Println("createTraceSpan:", t.ctx)
  290. _, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(start), trace.WithSpanKind(trace.SpanKindClient))
  291. span.SetAttributes(t.commonAttrs...)
  292. span.SetAttributes(attrs...)
  293. if error {
  294. span.SetStatus(codes.Error, "")
  295. }
  296. span.End(trace.WithTimestamp(end))
  297. }
  298. func (t *Trace) MysqlTraceQuery(query string, error bool, duration time.Duration, destination netaddr.IPPort) {
  299. if t == nil || query == "" {
  300. return
  301. }
  302. t.createTraceSpan(l7.ProtocolMysql.String(), duration, error,
  303. semconv.DBSystemMySQL,
  304. semconv.DBStatement(query),
  305. semconv.NetPeerName(destination.IP().String()),
  306. semconv.NetPeerPort(int(destination.Port())),
  307. )
  308. }
  309. // 内置允许操作(始终放行)
  310. var builtinOps = map[string]struct{}{
  311. "SELECT": {},
  312. "INSERT": {},
  313. "UPDATE": {},
  314. "DELETE": {},
  315. "WITH": {},
  316. }
  317. var (
  318. allowedOps map[string]struct{}
  319. initOnce sync.Once
  320. )
  321. // 解析并缓存环境变量中的前缀
  322. func initAllowedOps() {
  323. allowedOps = make(map[string]struct{})
  324. env := os.Getenv("ALLOWED_SQL_PREFIXES")
  325. for _, s := range strings.Split(env, ",") {
  326. if trimmed := strings.ToUpper(strings.TrimSpace(s)); trimmed != "" {
  327. allowedOps[trimmed] = struct{}{}
  328. }
  329. }
  330. }
  331. func isCURDOperation(q string) bool {
  332. if len(q) < 6 {
  333. return false
  334. }
  335. // 提取前缀(考虑最长10个字符)
  336. prefix := strings.ToUpper(strings.TrimSpace(q))
  337. if len(prefix) > 10 {
  338. prefix = prefix[:10]
  339. }
  340. // 提取第一个单词作为操作关键字
  341. spaceIdx := strings.IndexAny(prefix, " \t\n")
  342. if spaceIdx > 0 {
  343. prefix = prefix[:spaceIdx]
  344. }
  345. // 先查固定允许的操作
  346. if _, ok := builtinOps[prefix]; ok {
  347. return true
  348. }
  349. // 加载环境变量控制的操作(只执行一次)
  350. initOnce.Do(initAllowedOps)
  351. // 查额外允许操作
  352. _, ok := allowedOps[prefix]
  353. return ok
  354. }
  355. func (t *Trace) SQLTraceQueryEvent(l7Type l7.Protocol, semconvVal attribute.KeyValue, query string, r *l7.RequestData, destination netaddr.IPPort) {
  356. //fmt.Println("query", query)
  357. if t == nil {
  358. return
  359. }
  360. t.addEvent()
  361. // 只保留增删改查基本操作
  362. if !isCURDOperation(query) {
  363. return
  364. }
  365. var attr []attribute.KeyValue
  366. attr = append(attr,
  367. semconvVal,
  368. semconv.DBStatement(query),
  369. semconv.NetPeerName(destination.IP().String()),
  370. semconv.NetPeerPort(int(destination.Port())),
  371. attribute.String("sql.src_addr", r.ComponentSAddr.String()),
  372. attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
  373. )
  374. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  375. attr = append(attr,
  376. attribute.Bool("sql.exception", r.Status.Error()),
  377. )
  378. if r.Status.Error() {
  379. attr = append(attr,
  380. attribute.String("sql.exception_msg", r.ErrorMsg),
  381. )
  382. }
  383. t.createTraceEvent(l7Type.String(), ebpftracer.EventTypeL7Request.Int(), l7Type.Int(), attr...)
  384. }
  385. //func (t *Trace) MysqlTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
  386. // fmt.Println("query", query)
  387. // if t == nil {
  388. // return
  389. // }
  390. //
  391. // var attr []attribute.KeyValue
  392. // attr = append(attr,
  393. // semconv.DBSystemMySQL,
  394. // semconv.DBStatement(query),
  395. // semconv.NetPeerName(destination.IP().String()),
  396. // semconv.NetPeerPort(int(destination.Port())),
  397. // attribute.String("sql.src_addr", r.ComponentSAddr.String()),
  398. // attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
  399. // )
  400. // t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  401. //
  402. // attr = append(attr,
  403. // attribute.Bool("sql.exception", r.Status.Error()),
  404. // )
  405. // t.createTraceEvent(l7.ProtocolMysql.String(), int(ebpftracer.EventTypeL7Request), int(l7.ProtocolMysql), attr...)
  406. //}
  407. //func (t *Trace) PostGreSqlTraceQuery(query string, error bool, duration time.Duration, destination netaddr.IPPort) {
  408. // if t == nil || query == "" {
  409. // return
  410. // }
  411. // t.createTraceSpan(l7.ProtocolPostgres.String(), duration, error,
  412. // semconv.DBSystemPostgreSQL,
  413. // semconv.DBStatement(query),
  414. // semconv.NetPeerName(destination.IP().String()),
  415. // semconv.NetPeerPort(int(destination.Port())),
  416. // )
  417. //}
  418. //func (t *Trace) PostGreSqlTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
  419. // fmt.Println("query", query)
  420. // if t == nil {
  421. // return
  422. // }
  423. //
  424. // var attr []attribute.KeyValue
  425. // attr = append(attr,
  426. // semconv.DBSystemPostgreSQL,
  427. // semconv.DBStatement(query),
  428. // semconv.NetPeerName(destination.IP().String()),
  429. // semconv.NetPeerPort(int(destination.Port())),
  430. // attribute.String("sql.src_addr", r.ComponentSAddr.String()),
  431. // attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
  432. // )
  433. // t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  434. // attr = append(attr,
  435. // attribute.Bool("sql.exception", r.Status.Error()),
  436. // )
  437. // t.createTraceEvent(l7.ProtocolPostgres.String(), int(ebpftracer.EventTypeL7Request), int(l7.ProtocolPostgres), attr...)
  438. //}
  439. //func (t *Trace) DmTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
  440. // if t == nil || query == "" {
  441. // return
  442. // }
  443. // l7Type := int(l7.ProtocolDM)
  444. // var attr []attribute.KeyValue
  445. // attr = append(attr,
  446. // semconv.DBSystemDaMengDB,
  447. // semconv.DBStatement(query),
  448. // semconv.NetPeerName(destination.IP().String()),
  449. // semconv.NetPeerPort(int(destination.Port())),
  450. // attribute.String("sql.src_addr", r.ComponentSAddr.String()),
  451. // attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
  452. // )
  453. // t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  454. //
  455. // attr = append(attr,
  456. // attribute.Bool("sql.exception", r.Status.Error()),
  457. // )
  458. // t.createTraceEvent(l7.ProtocolHTTP.String(), int(ebpftracer.EventTypeL7Request), l7Type, attr...)
  459. //}
  460. //func (t *Trace) RedisTraceQuery(cmd, args string, error bool, duration time.Duration) {
  461. // if t == nil || cmd == "" {
  462. // return
  463. // }
  464. // statement := cmd
  465. // if args != "" {
  466. // statement += " " + args
  467. // }
  468. // t.createTraceSpan(l7.ProtocolRedis.String(), duration, error,
  469. // semconv.DBSystemRedis,
  470. // semconv.DBOperation(cmd),
  471. // semconv.DBStatement(statement),
  472. // )
  473. //}
  474. func (t *Trace) RedisTraceQueryEvent(cmd, args string, r *l7.RequestData, destination netaddr.IPPort) {
  475. if t == nil {
  476. return
  477. }
  478. t.addEvent()
  479. if cmd == "" {
  480. return
  481. }
  482. statement := cmd
  483. if args != "" {
  484. statement += " " + args
  485. }
  486. var attr []attribute.KeyValue
  487. attr = append(attr,
  488. semconv.DBSystemRedis,
  489. semconv.DBOperation(cmd),
  490. semconv.DBStatement(statement),
  491. semconv.NetPeerName(destination.IP().String()),
  492. semconv.NetPeerPort(int(destination.Port())),
  493. attribute.String("nosql.src_addr", r.ComponentSAddr.String()),
  494. attribute.String("nosql.destination_addr", r.ComponentDAddr.String()),
  495. )
  496. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  497. t.createTraceEvent(l7.ProtocolRedis.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolRedis.Int(), attr...)
  498. }
  499. func (t *Trace) MongoTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
  500. if t == nil {
  501. return
  502. }
  503. t.addEvent()
  504. if query == "" {
  505. return
  506. }
  507. var attr []attribute.KeyValue
  508. attr = append(attr,
  509. semconv.DBSystemMongoDB,
  510. semconv.DBStatement(query),
  511. semconv.NetPeerName(destination.IP().String()),
  512. semconv.NetPeerPort(int(destination.Port())),
  513. attribute.String("nosql.src_addr", r.ComponentSAddr.String()),
  514. attribute.String("nosql.destination_addr", r.ComponentDAddr.String()),
  515. )
  516. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  517. t.createTraceEvent(l7.ProtocolMongo.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolMongo.Int(), attr...)
  518. }
  519. func (t *Trace) DNSTraceQueryEvent(r *l7.RequestData, _type string, fqdn string, ttl uint32, ips []netaddr.IP) {
  520. if t == nil {
  521. return
  522. }
  523. t.addEvent()
  524. var attr []attribute.KeyValue
  525. var ipList []string
  526. for _, ip := range ips {
  527. ipList = append(ipList, ip.String())
  528. }
  529. ip := strings.Join(ipList, ",")
  530. attr = append(attr,
  531. attribute.String("dns.type", _type),
  532. attribute.String("dns.fqdn", fqdn),
  533. attribute.Int("dns.ttl", int(ttl)),
  534. attribute.String("dns.ips", ip),
  535. )
  536. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  537. t.createTraceEvent(l7.ProtocolDNS.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolDNS.Int(), attr...)
  538. }
  539. func (t *Trace) HttpTraceRequest(method, path, ip string, port uint16, r *l7.RequestData) {
  540. if t == nil || method == "" {
  541. return
  542. }
  543. assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
  544. if err != nil {
  545. assumedAppID = 0
  546. }
  547. status := r.Status
  548. duration := r.Duration
  549. t.createTraceSpan(l7.ProtocolHTTP.String(), duration, status >= 400,
  550. semconv.HTTPURL(fmt.Sprintf("http://%s%s", t.destination.String(), path)),
  551. semconv.HTTPMethod(method),
  552. semconv.HTTPStatusCode(int(status)),
  553. attribute.String("http.uri", path),
  554. attribute.String("http.ip", ip),
  555. attribute.Int64("http.assumed_app_id", assumedAppID),
  556. attribute.String("http.span_id", r.SpanId),
  557. attribute.Int("http.port", int(port)),
  558. )
  559. }
  560. // 新增事件处理
  561. func (t *Trace) HttpTraceRequestEvent(method, path, ip string, port uint16, r *l7.RequestData) {
  562. if t == nil {
  563. return
  564. }
  565. t.addEvent()
  566. if method == "" {
  567. return
  568. }
  569. assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
  570. if err != nil {
  571. assumedAppID = 0
  572. }
  573. status := r.Status
  574. var attr []attribute.KeyValue
  575. attr = append(attr,
  576. semconv.HTTPURL(fmt.Sprintf("http://%s:%d%s", ip, port, path)),
  577. semconv.HTTPMethod(method),
  578. semconv.HTTPStatusCode(int(status)),
  579. attribute.Bool("http.status_error", status > 399),
  580. attribute.String("http.uri", path),
  581. attribute.String("http.ip", ip),
  582. attribute.Int64("http.assumed_app_id", assumedAppID),
  583. attribute.String("http.span_id", r.SpanId),
  584. attribute.Int("http.port", int(port)),
  585. attribute.String("http.src_addr", r.ComponentSAddr.String()),
  586. attribute.String("http.destination_addr", r.ComponentDAddr.String()),
  587. attribute.Bool("http.is_tls", r.IsTls),
  588. )
  589. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  590. t.createTraceEvent(l7.ProtocolHTTP.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolHTTP.Int(), attr...)
  591. }
  592. func (t *Trace) FuncTraceQuery(funcname string, duration time.Duration, start uint64, end uint64) {
  593. if t == nil {
  594. return
  595. }
  596. t.addEvent()
  597. if funcname == "" {
  598. return
  599. }
  600. t.createTraceSpanNoTime2(funcname, duration, false, start, end)
  601. //t.createTraceSpanNoTime2(funcname, duration, false, start, end, attribute.Int("num", num))
  602. }
  603. func (t *Trace) createTraceSpanNoTime(name string, duration time.Duration, error bool, start uint64, end uint64, attrs ...attribute.KeyValue) {
  604. // end := time.Now()
  605. // start := end.Add(-duration)
  606. startTime := time.Unix(0, int64(start))
  607. endTime := time.Unix(0, int64(end))
  608. //fmt.Println("createTraceSpan:", t.ctx)
  609. _, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(startTime), trace.WithSpanKind(trace.SpanKindClient))
  610. span.SetAttributes(t.commonAttrs...)
  611. span.SetAttributes(attrs...)
  612. if error {
  613. span.SetStatus(codes.Error, "")
  614. }
  615. span.End(trace.WithTimestamp(endTime))
  616. }
  617. func (t *Trace) createTraceSpanNoTime2(name string, duration time.Duration, error bool, start uint64, end uint64) {
  618. // end := time.Now()
  619. // start := end.Add(-duration)
  620. //startTime := time.Unix(0, int64(start))
  621. //endTime := time.Unix(0, int64(end))
  622. ////fmt.Println("createTraceSpan:", t.ctx)
  623. //_, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(startTime), trace.WithSpanKind(trace.SpanKindClient))
  624. //span.SetAttributes(t.commonAttrs...)
  625. //span.SetAttributes(attrs...)
  626. //if error {
  627. // span.SetStatus(codes.Error, "")
  628. //}
  629. //span.End(trace.WithTimestamp(endTime))
  630. //attrs = append([]attribute.KeyValue{
  631. // attribute.Int64("startAt", int64(start)),
  632. // attribute.Int64("endAt", int64(end)),
  633. //})
  634. var attr []attribute.KeyValue
  635. t.appendTimestamp(&attr, start, end, int64(end-start))
  636. t.createTraceEvent(name, ebpftracer.EventTypeFunEnt.Int(), 0, attr...)
  637. }