apm_tracing.go 22 KB

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