apm_tracing.go 22 KB

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