apm_tracing.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669
  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) {
  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. }
  141. t.span.SetAttributes(t.commonAttrs...)
  142. t.startReady()
  143. }
  144. // set context span
  145. func (t *Trace) CreateRootSpan(traceId uint64) {
  146. traceIdStr := strconv.Itoa(int(traceId))
  147. ctx, span := tracer(t.containerId).Start(context.Background(), traceIdStr, trace.WithSpanKind(trace.SpanKindClient))
  148. t.setContext(ctx)
  149. t.setSpan(span)
  150. }
  151. func (t *Trace) TraceStart(method, path string, status l7.Status, duration time.Duration) {
  152. if t == nil || method == "" {
  153. return
  154. }
  155. t.createParentSpan("APPLICATION", duration, status >= 400,
  156. semconv.HTTPURL(fmt.Sprintf("http://%s%s", t.destination.String(), path)),
  157. semconv.HTTPMethod(method),
  158. //semconv.HTTPStatusCode(int(status)),
  159. attribute.String("http.uri", path),
  160. )
  161. }
  162. func (t *Trace) TraceEnd(r *l7.RequestData) {
  163. if t == nil {
  164. return
  165. }
  166. t.span.SetAttributes(
  167. semconv.HTTPStatusCode(int(r.Status)),
  168. attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
  169. )
  170. CalledId, err := strconv.ParseInt(r.ParentSpanContext.CalledId, 10, 64)
  171. if err == nil && CalledId != 0 {
  172. t.span.SetAttributes(attribute.Int64("server.called_id", CalledId))
  173. }
  174. InstanceIdFrom, err := strconv.ParseInt(r.ParentSpanContext.InstanceIdFrom, 10, 64)
  175. if err == nil && InstanceIdFrom != 0 {
  176. t.span.SetAttributes(attribute.Int64("server.instance_id_from", InstanceIdFrom))
  177. }
  178. AppIdFrom, err := strconv.ParseInt(r.ParentSpanContext.AppIdFrom, 10, 64)
  179. if err == nil && AppIdFrom != 0 {
  180. t.span.SetAttributes(attribute.Int64("server.app_id_from", AppIdFrom))
  181. }
  182. if r.ParentSpanContext.SpanIdFrom != "0000000000000000" {
  183. t.span.SetAttributes(attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
  184. }
  185. // for _, v := range t.stack {
  186. // fmt.Printf("TraceEndTraceEndTraceEnd%s\n", v)
  187. // }
  188. //t.buildFun()
  189. t.span.End(trace.WithTimestamp(time.Now()))
  190. }
  191. // 新增结束事件
  192. func (t *Trace) TraceEndEvent(r *l7.RequestData) {
  193. if t == nil {
  194. return
  195. }
  196. var attr []attribute.KeyValue
  197. attr = append(attr,
  198. semconv.HTTPStatusCode(int(r.Status)),
  199. attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
  200. )
  201. //t.span.SetAttributes(
  202. // semconv.HTTPStatusCode(int(r.Status)),
  203. // attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
  204. //)
  205. calledId, err := strconv.ParseInt(r.ParentSpanContext.CalledId, 10, 64)
  206. if err == nil && calledId != 0 {
  207. attr = append(attr, attribute.Int64("server.called_id", calledId))
  208. //t.span.SetAttributes(attribute.Int64("server.called_id", CalledId))
  209. }
  210. instanceIdFrom, err := strconv.ParseInt(r.ParentSpanContext.InstanceIdFrom, 10, 64)
  211. if err == nil && instanceIdFrom != 0 {
  212. attr = append(attr, attribute.Int64("server.instance_id_from", instanceIdFrom))
  213. //t.span.SetAttributes(attribute.Int64("server.instance_id_from", InstanceIdFrom))
  214. }
  215. appIdFrom, err := strconv.ParseInt(r.ParentSpanContext.AppIdFrom, 10, 64)
  216. if err == nil && appIdFrom != 0 {
  217. attr = append(attr, attribute.Int64("server.app_id_from", appIdFrom))
  218. //t.span.SetAttributes(attribute.Int64("server.app_id_from", AppIdFrom))
  219. }
  220. if r.ParentSpanContext.SpanIdFrom != "0000000000000000" {
  221. attr = append(attr, attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
  222. //t.span.SetAttributes(attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
  223. }
  224. attr = append(attr, attribute.String("server.src_addr", r.DAddr.String()))
  225. attr = append(attr, attribute.String("server.dst_addr", r.SAddr.String()))
  226. klog.Debugf("TraceEndEvent SAddr.String %s", r.SAddr.String())
  227. klog.Debugf("TraceEndEvent DAddr.String %s", r.DAddr.String())
  228. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  229. t.span.SetAttributes(attr...)
  230. t.endReadyEvent(r.EventCount)
  231. }
  232. func (t *Trace) appendTimestamp(attr *[]attribute.KeyValue, s, e uint64, d int64) {
  233. *attr = append(*attr,
  234. attribute.Int64("time.start_at", utils.KtimeToTimestamp(s)),
  235. attribute.Int64("time.end_at", utils.KtimeToTimestamp(e)),
  236. attribute.Int64("time.duration", d),
  237. )
  238. }
  239. func (t *Trace) createParentSpan(name string, duration time.Duration, error bool, attrs ...attribute.KeyValue) {
  240. end := time.Now()
  241. start := end.Add(-duration)
  242. ctx, span := tracer(t.containerId).Start(context.Background(), name, trace.WithTimestamp(start), trace.WithSpanKind(trace.SpanKindClient))
  243. span.SetAttributes(attrs...)
  244. span.SetAttributes(t.commonAttrs...)
  245. if error {
  246. span.SetStatus(codes.Error, "")
  247. }
  248. t.setContext(ctx)
  249. t.setSpan(span)
  250. }
  251. func (t *Trace) SendEvent() {
  252. t.span.End()
  253. }
  254. func (t *Trace) GetSpan() trace.Span {
  255. return t.span
  256. }
  257. func (t *Trace) createTraceEvent(name string, eventType int, l7Type int, attrs ...attribute.KeyValue) {
  258. t.span.AddEventApm(name, eventType, l7Type, trace.WithAttributes(attrs...))
  259. //atomic.AddUint32(t.currenEventCount, 1)
  260. }
  261. func (t *Trace) addEvent() {
  262. atomic.AddUint32(t.currenEventCount, 1)
  263. }
  264. func (t *Trace) createTraceSpan(name string, duration time.Duration, error bool, attrs ...attribute.KeyValue) {
  265. end := time.Now()
  266. start := end.Add(-duration)
  267. //fmt.Println("createTraceSpan:", t.ctx)
  268. _, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(start), trace.WithSpanKind(trace.SpanKindClient))
  269. span.SetAttributes(t.commonAttrs...)
  270. span.SetAttributes(attrs...)
  271. if error {
  272. span.SetStatus(codes.Error, "")
  273. }
  274. span.End(trace.WithTimestamp(end))
  275. }
  276. func (t *Trace) MysqlTraceQuery(query string, error bool, duration time.Duration, destination netaddr.IPPort) {
  277. if t == nil || query == "" {
  278. return
  279. }
  280. t.createTraceSpan(l7.ProtocolMysql.String(), duration, error,
  281. semconv.DBSystemMySQL,
  282. semconv.DBStatement(query),
  283. semconv.NetPeerName(destination.IP().String()),
  284. semconv.NetPeerPort(int(destination.Port())),
  285. )
  286. }
  287. // 内置允许操作(始终放行)
  288. var builtinOps = map[string]struct{}{
  289. "SELECT": {},
  290. "INSERT": {},
  291. "UPDATE": {},
  292. "DELETE": {},
  293. "WITH": {},
  294. }
  295. var (
  296. allowedOps map[string]struct{}
  297. initOnce sync.Once
  298. )
  299. // 解析并缓存环境变量中的前缀
  300. func initAllowedOps() {
  301. allowedOps = make(map[string]struct{})
  302. env := os.Getenv("ALLOWED_SQL_PREFIXES")
  303. for _, s := range strings.Split(env, ",") {
  304. if trimmed := strings.ToUpper(strings.TrimSpace(s)); trimmed != "" {
  305. allowedOps[trimmed] = struct{}{}
  306. }
  307. }
  308. }
  309. func isCURDOperation(q string) bool {
  310. if len(q) < 6 {
  311. return false
  312. }
  313. // 提取前缀(考虑最长10个字符)
  314. prefix := strings.ToUpper(strings.TrimSpace(q))
  315. if len(prefix) > 10 {
  316. prefix = prefix[:10]
  317. }
  318. // 提取第一个单词作为操作关键字
  319. spaceIdx := strings.IndexAny(prefix, " \t\n")
  320. if spaceIdx > 0 {
  321. prefix = prefix[:spaceIdx]
  322. }
  323. // 先查固定允许的操作
  324. if _, ok := builtinOps[prefix]; ok {
  325. return true
  326. }
  327. // 加载环境变量控制的操作(只执行一次)
  328. initOnce.Do(initAllowedOps)
  329. // 查额外允许操作
  330. _, ok := allowedOps[prefix]
  331. return ok
  332. }
  333. func (t *Trace) SQLTraceQueryEvent(l7Type l7.Protocol, semconvVal attribute.KeyValue, query string, r *l7.RequestData, destination netaddr.IPPort) {
  334. //fmt.Println("query", query)
  335. if t == nil {
  336. return
  337. }
  338. t.addEvent()
  339. // 只保留增删改查基本操作
  340. if !isCURDOperation(query) {
  341. return
  342. }
  343. var attr []attribute.KeyValue
  344. attr = append(attr,
  345. semconvVal,
  346. semconv.DBStatement(query),
  347. semconv.NetPeerName(destination.IP().String()),
  348. semconv.NetPeerPort(int(destination.Port())),
  349. attribute.String("sql.src_addr", r.ComponentSAddr.String()),
  350. attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
  351. )
  352. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  353. attr = append(attr,
  354. attribute.Bool("sql.exception", r.Status.Error()),
  355. )
  356. t.createTraceEvent(l7Type.String(), ebpftracer.EventTypeL7Request.Int(), l7Type.Int(), attr...)
  357. }
  358. //func (t *Trace) MysqlTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
  359. // fmt.Println("query", query)
  360. // if t == nil {
  361. // return
  362. // }
  363. //
  364. // var attr []attribute.KeyValue
  365. // attr = append(attr,
  366. // semconv.DBSystemMySQL,
  367. // semconv.DBStatement(query),
  368. // semconv.NetPeerName(destination.IP().String()),
  369. // semconv.NetPeerPort(int(destination.Port())),
  370. // attribute.String("sql.src_addr", r.ComponentSAddr.String()),
  371. // attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
  372. // )
  373. // t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  374. //
  375. // attr = append(attr,
  376. // attribute.Bool("sql.exception", r.Status.Error()),
  377. // )
  378. // t.createTraceEvent(l7.ProtocolMysql.String(), int(ebpftracer.EventTypeL7Request), int(l7.ProtocolMysql), attr...)
  379. //}
  380. //func (t *Trace) PostGreSqlTraceQuery(query string, error bool, duration time.Duration, destination netaddr.IPPort) {
  381. // if t == nil || query == "" {
  382. // return
  383. // }
  384. // t.createTraceSpan(l7.ProtocolPostgres.String(), duration, error,
  385. // semconv.DBSystemPostgreSQL,
  386. // semconv.DBStatement(query),
  387. // semconv.NetPeerName(destination.IP().String()),
  388. // semconv.NetPeerPort(int(destination.Port())),
  389. // )
  390. //}
  391. //func (t *Trace) PostGreSqlTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
  392. // fmt.Println("query", query)
  393. // if t == nil {
  394. // return
  395. // }
  396. //
  397. // var attr []attribute.KeyValue
  398. // attr = append(attr,
  399. // semconv.DBSystemPostgreSQL,
  400. // semconv.DBStatement(query),
  401. // semconv.NetPeerName(destination.IP().String()),
  402. // semconv.NetPeerPort(int(destination.Port())),
  403. // attribute.String("sql.src_addr", r.ComponentSAddr.String()),
  404. // attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
  405. // )
  406. // t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  407. // attr = append(attr,
  408. // attribute.Bool("sql.exception", r.Status.Error()),
  409. // )
  410. // t.createTraceEvent(l7.ProtocolPostgres.String(), int(ebpftracer.EventTypeL7Request), int(l7.ProtocolPostgres), attr...)
  411. //}
  412. //func (t *Trace) DmTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
  413. // if t == nil || query == "" {
  414. // return
  415. // }
  416. // l7Type := int(l7.ProtocolDM)
  417. // var attr []attribute.KeyValue
  418. // attr = append(attr,
  419. // semconv.DBSystemDaMengDB,
  420. // semconv.DBStatement(query),
  421. // semconv.NetPeerName(destination.IP().String()),
  422. // semconv.NetPeerPort(int(destination.Port())),
  423. // attribute.String("sql.src_addr", r.ComponentSAddr.String()),
  424. // attribute.String("sql.destination_addr", r.ComponentDAddr.String()),
  425. // )
  426. // t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  427. //
  428. // attr = append(attr,
  429. // attribute.Bool("sql.exception", r.Status.Error()),
  430. // )
  431. // t.createTraceEvent(l7.ProtocolHTTP.String(), int(ebpftracer.EventTypeL7Request), l7Type, attr...)
  432. //}
  433. //func (t *Trace) RedisTraceQuery(cmd, args string, error bool, duration time.Duration) {
  434. // if t == nil || cmd == "" {
  435. // return
  436. // }
  437. // statement := cmd
  438. // if args != "" {
  439. // statement += " " + args
  440. // }
  441. // t.createTraceSpan(l7.ProtocolRedis.String(), duration, error,
  442. // semconv.DBSystemRedis,
  443. // semconv.DBOperation(cmd),
  444. // semconv.DBStatement(statement),
  445. // )
  446. //}
  447. func (t *Trace) RedisTraceQueryEvent(cmd, args string, r *l7.RequestData, destination netaddr.IPPort) {
  448. if t == nil {
  449. return
  450. }
  451. t.addEvent()
  452. if cmd == "" {
  453. return
  454. }
  455. statement := cmd
  456. if args != "" {
  457. statement += " " + args
  458. }
  459. var attr []attribute.KeyValue
  460. attr = append(attr,
  461. semconv.DBSystemRedis,
  462. semconv.DBOperation(cmd),
  463. semconv.DBStatement(statement),
  464. semconv.NetPeerName(destination.IP().String()),
  465. semconv.NetPeerPort(int(destination.Port())),
  466. attribute.String("nosql.src_addr", r.ComponentSAddr.String()),
  467. attribute.String("nosql.destination_addr", r.ComponentDAddr.String()),
  468. )
  469. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  470. t.createTraceEvent(l7.ProtocolRedis.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolRedis.Int(), attr...)
  471. }
  472. func (t *Trace) MongoTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
  473. if t == nil {
  474. return
  475. }
  476. t.addEvent()
  477. if query == "" {
  478. return
  479. }
  480. var attr []attribute.KeyValue
  481. attr = append(attr,
  482. semconv.DBSystemRedis,
  483. semconv.DBStatement(query),
  484. semconv.NetPeerName(destination.IP().String()),
  485. semconv.NetPeerPort(int(destination.Port())),
  486. attribute.String("nosql.src_addr", r.ComponentSAddr.String()),
  487. attribute.String("nosql.destination_addr", r.ComponentDAddr.String()),
  488. )
  489. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  490. t.createTraceEvent(l7.ProtocolMongo.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolMongo.Int(), attr...)
  491. }
  492. func (t *Trace) DNSTraceQueryEvent(r *l7.RequestData, _type string, fqdn string, ttl uint32, ips []netaddr.IP) {
  493. if t == nil {
  494. return
  495. }
  496. t.addEvent()
  497. var attr []attribute.KeyValue
  498. var ipList []string
  499. for _, ip := range ips {
  500. ipList = append(ipList, ip.String())
  501. }
  502. ip := strings.Join(ipList, ",")
  503. attr = append(attr,
  504. attribute.String("dns.type", _type),
  505. attribute.String("dns.fqdn", fqdn),
  506. attribute.Int("dns.ttl", int(ttl)),
  507. attribute.String("dns.ips", ip),
  508. )
  509. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  510. t.createTraceEvent(l7.ProtocolDNS.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolDNS.Int(), attr...)
  511. }
  512. func (t *Trace) HttpTraceRequest(method, path, ip string, port uint16, r *l7.RequestData) {
  513. if t == nil || method == "" {
  514. return
  515. }
  516. assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
  517. if err != nil {
  518. assumedAppID = 0
  519. }
  520. status := r.Status
  521. duration := r.Duration
  522. t.createTraceSpan(l7.ProtocolHTTP.String(), duration, status >= 400,
  523. semconv.HTTPURL(fmt.Sprintf("http://%s%s", t.destination.String(), path)),
  524. semconv.HTTPMethod(method),
  525. semconv.HTTPStatusCode(int(status)),
  526. attribute.String("http.uri", path),
  527. attribute.String("http.ip", ip),
  528. attribute.Int64("http.assumed_app_id", assumedAppID),
  529. attribute.String("http.span_id", r.SpanId),
  530. attribute.Int("http.port", int(port)),
  531. )
  532. }
  533. // 新增事件处理
  534. func (t *Trace) HttpTraceRequestEvent(method, path, ip string, port uint16, r *l7.RequestData) {
  535. if t == nil {
  536. return
  537. }
  538. t.addEvent()
  539. if method == "" {
  540. return
  541. }
  542. assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
  543. if err != nil {
  544. assumedAppID = 0
  545. }
  546. status := r.Status
  547. var attr []attribute.KeyValue
  548. attr = append(attr,
  549. semconv.HTTPURL(fmt.Sprintf("http://%s:%d%s", ip, port, path)),
  550. semconv.HTTPMethod(method),
  551. semconv.HTTPStatusCode(int(status)),
  552. attribute.Bool("http.status_error", status > 399),
  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. attribute.String("http.src_addr", r.ComponentSAddr.String()),
  559. attribute.String("http.destination_addr", r.ComponentDAddr.String()),
  560. )
  561. t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
  562. t.createTraceEvent(l7.ProtocolHTTP.String(), ebpftracer.EventTypeL7Request.Int(), l7.ProtocolHTTP.Int(), attr...)
  563. }
  564. func (t *Trace) FuncTraceQuery(funcname string, duration time.Duration, start uint64, end uint64) {
  565. if t == nil {
  566. return
  567. }
  568. t.addEvent()
  569. if funcname == "" {
  570. return
  571. }
  572. t.createTraceSpanNoTime2(funcname, duration, false, start, end)
  573. //t.createTraceSpanNoTime2(funcname, duration, false, start, end, attribute.Int("num", num))
  574. }
  575. func (t *Trace) createTraceSpanNoTime(name string, duration time.Duration, error bool, start uint64, end uint64, attrs ...attribute.KeyValue) {
  576. // end := time.Now()
  577. // start := end.Add(-duration)
  578. startTime := time.Unix(0, int64(start))
  579. endTime := time.Unix(0, int64(end))
  580. //fmt.Println("createTraceSpan:", t.ctx)
  581. _, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(startTime), trace.WithSpanKind(trace.SpanKindClient))
  582. span.SetAttributes(t.commonAttrs...)
  583. span.SetAttributes(attrs...)
  584. if error {
  585. span.SetStatus(codes.Error, "")
  586. }
  587. span.End(trace.WithTimestamp(endTime))
  588. }
  589. func (t *Trace) createTraceSpanNoTime2(name string, duration time.Duration, error bool, start uint64, end uint64) {
  590. // end := time.Now()
  591. // start := end.Add(-duration)
  592. //startTime := time.Unix(0, int64(start))
  593. //endTime := time.Unix(0, int64(end))
  594. ////fmt.Println("createTraceSpan:", t.ctx)
  595. //_, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(startTime), trace.WithSpanKind(trace.SpanKindClient))
  596. //span.SetAttributes(t.commonAttrs...)
  597. //span.SetAttributes(attrs...)
  598. //if error {
  599. // span.SetStatus(codes.Error, "")
  600. //}
  601. //span.End(trace.WithTimestamp(endTime))
  602. //attrs = append([]attribute.KeyValue{
  603. // attribute.Int64("startAt", int64(start)),
  604. // attribute.Int64("endAt", int64(end)),
  605. //})
  606. var attr []attribute.KeyValue
  607. t.appendTimestamp(&attr, start, end, int64(end-start))
  608. t.createTraceEvent(name, ebpftracer.EventTypeFunEnt.Int(), 0, attr...)
  609. }