apm_exporter.go 33 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145
  1. package otlptrace
  2. import (
  3. "crypto/md5"
  4. "encoding/json"
  5. "fmt"
  6. . "github.com/coroot/coroot-node-agent/ebpftracer"
  7. "github.com/coroot/coroot-node-agent/ebpftracer/l7"
  8. "github.com/coroot/coroot-node-agent/utils"
  9. klog "github.com/sirupsen/logrus"
  10. "math"
  11. "net/url"
  12. "sort"
  13. "strconv"
  14. "sync"
  15. "go.opentelemetry.io/otel/exporters/otlp/otlptrace/internal/tracetransform"
  16. tracesdk "go.opentelemetry.io/otel/sdk/trace"
  17. tracepb "go.opentelemetry.io/proto/otlp/trace/v1"
  18. )
  19. const (
  20. APP_SERVICE_TYPE = "APPLICATION"
  21. SQL_SERVICE_TYPE = "SQL"
  22. NOSQL_SERVICE_TYPE = "NOSQL"
  23. HTTP_SERVICE_TYPE = "HTTP"
  24. NET_SERVICE_TYPE = "L7_NET"
  25. )
  26. const (
  27. GO_SERVICE_NAME = "GO"
  28. MYSQL_SERVICE_NAME = "MYSQL"
  29. DM_SERVICE_NAME = "DM"
  30. REDIS_SERVICE_NAME = "REDIS"
  31. MONGO_SERVICE_NAME = "MONGODB"
  32. HTTP_SERVICE_NAME = "HTTPCLIENT"
  33. POSTGRESQL_SERVICE_NAME = "POSTGRESQL"
  34. )
  35. type apmTraceSpan tracesdk.ReadOnlySpan
  36. // GO:0:10154813500555812:5450531005555981:5610250100539899:ee022542c3940f1b:1001025098564810:888ceb3df1bdbe2c:110
  37. type RootDataT struct {
  38. AccountId int `json:"account_id"`
  39. AgentId int64 `json:"agent_id"`
  40. AgentVersion string `json:"agent_version"`
  41. AppId int64 `json:"app_id"`
  42. AppIdFrom int64 `json:"app_id_from"` // from header app_id
  43. AppName string `json:"app_name"`
  44. CalledId int64 `json:"called_id"` // from header assumed_app_id
  45. ClientIp string `json:"client_ip"`
  46. CollTime uint64 `json:"coll_time"`
  47. Cpu int `json:"cpu"`
  48. Custom string `json:"custom"`
  49. HostId int64 `json:"host_id"`
  50. HostName string `json:"host_name"`
  51. HttpCode int64 `json:"http_code"`
  52. HttpMethod string `json:"http_method"`
  53. InstanceId int64 `json:"instance_id"`
  54. InstanceIdFrom int64 `json:"instance_id_from"` // from header instance_id
  55. LocalPort int64 `json:"local_port"`
  56. Maps []MapInfoT `json:"maps"`
  57. MemU int `json:"mem_u"`
  58. MemUP int `json:"mem_u_p"`
  59. OperType string `json:"oper_type"`
  60. Parameters []ParamStruct `json:"parameters,omitempty"`
  61. ParentTaskName string `json:"parent_task_name"`
  62. Period int `json:"period"`
  63. RespTime uint64 `json:"resp_time"`
  64. Sampling int `json:"sampling"`
  65. ServiceName string `json:"service_name"`
  66. ServiceType string `json:"service_type"`
  67. Sip string `json:"sip"`
  68. Sn string `json:"sn"`
  69. SpanIdFrom string `json:"span_id_from"` // from header span_id
  70. Sport int64 `json:"sport"`
  71. TId int `json:"t_id"`
  72. TName string `json:"t_name"`
  73. TraceId string `json:"trace_id"` // from header trace_id
  74. TransIds []interface{} `json:"trans_ids"`
  75. TypeFrom string `json:"type_from"`
  76. Uri string `json:"uri"`
  77. UserDir string `json:"user_dir"`
  78. VipIds []interface{} `json:"vip_ids"`
  79. SrcAddr string `json:"src_addr"`
  80. DestinationAddr string `json:"destination_addr"`
  81. // op 新增字段
  82. Pid uint32 `json:"pid"`
  83. ContainerID string `json:"container_id"`
  84. Sys string `json:"sys"`
  85. SystemUUID string `json:"system_uuid"`
  86. }
  87. // ParamStruct 定义目标结构
  88. type ParamStruct struct {
  89. Name string `json:"name"`
  90. Values []string `json:"values"`
  91. }
  92. type MapInfoT struct {
  93. Dbn string `json:"dbn,omitempty"`
  94. Exception int `json:"exception,omitempty"`
  95. ExceptionMsg string `json:"exception_msg,omitempty"`
  96. ExceptionStack string `json:"exception_stack,omitempty"`
  97. Ip string `json:"ip,omitempty"`
  98. Level int `json:"level"`
  99. MethodDesc string `json:"method_desc,omitempty"`
  100. MethodName string `json:"method_name"`
  101. Nid int `json:"nid"`
  102. OperType string `json:"oper_type,omitempty"`
  103. Pid int `json:"pid"`
  104. Port int64 `json:"port,omitempty"`
  105. Ps []string `json:"ps,omitempty"`
  106. PureTime uint64 `json:"pure_time"`
  107. ServiceName string `json:"service_name"`
  108. ServiceType string `json:"service_type"`
  109. StartTime uint64 `json:"-"`
  110. EndTime uint64 `json:"-"`
  111. StartTimeMs uint64 `json:"start_time"`
  112. EndTimeMs uint64 `json:"end_time"`
  113. WallTime uint64 `json:"wall_time"`
  114. Schema string `json:"schema,omitempty"`
  115. AssumedAppId int64 `json:"assumed_app_id,omitempty"`
  116. Uri string `json:"uri,omitempty"`
  117. SpanId string `json:"span_id,omitempty"`
  118. SrcAddr string `json:"src_addr,omitempty"`
  119. DestinationAddr string `json:"destination_addr,omitempty"`
  120. }
  121. type TraceMapT struct {
  122. RootData RootDataT
  123. Index int
  124. lock *sync.RWMutex
  125. TheEnd bool
  126. }
  127. var TraceRootMap map[string]*TraceMapT
  128. func init() {
  129. TraceRootMap = make(map[string]*TraceMapT)
  130. //go func() {
  131. // for {
  132. // //fmt.Println(G_sdl)
  133. // time.Sleep(5 * time.Second)
  134. // }
  135. //}()
  136. }
  137. var G_sdl int
  138. func tracetransformData(sdl []tracesdk.ReadOnlySpan) map[int][]RootDataT {
  139. //G_sdl += len(sdl)
  140. if len(sdl) == 0 {
  141. return nil
  142. }
  143. // 多次请求 sdl
  144. sendDataMap := make(map[int][]RootDataT)
  145. //sendData := []RootDataT{}
  146. for _, sd := range sdl {
  147. if sd == nil {
  148. continue
  149. }
  150. //traceId := sd.SpanContext().TraceID().String()
  151. //fmt.Println("------event_num---- "+sd.Name(), "--->", len(sd.Events())) // 一次请求完整数据
  152. // 构建map *RootDataT
  153. var rootData RootDataT
  154. rootData = initRootDataFromEvent()
  155. // build http入口 MapInfoT
  156. code_type := buildAppMapFromEvent(&rootData, sd)
  157. // 构建maps
  158. for _, event := range sd.Events() {
  159. //aaa, _ := json.Marshal(event)
  160. //fmt.Println("event.info", string(aaa))
  161. mNode := buildMapNodeFromEvent(event)
  162. switch EventType(event.EventType) {
  163. // stack
  164. case EventTypeFunEnt:
  165. // l7 event
  166. case EventTypeL7Request:
  167. switch l7.Protocol(event.ProtocolType) {
  168. case l7.ProtocolHTTP:
  169. buildHttpMapFromEvent(&mNode, event)
  170. case l7.ProtocolDNS:
  171. buildDNSMapEvent(&mNode, event)
  172. case l7.ProtocolMysql, l7.ProtocolMariaDB, l7.ProtocolPostgres, l7.ProtocolDM:
  173. buildSQLMapEvent(&mNode, event)
  174. case l7.ProtocolRedis:
  175. buildRedisMapEvent(&mNode, event)
  176. case l7.ProtocolMongo:
  177. buildMongoMapEvent(&mNode, event)
  178. }
  179. }
  180. rootData.Maps = append(rootData.Maps, mNode)
  181. //fmt.Println(event.Name)
  182. //buildAndAssemblyMapFromEvent(event, rootData)
  183. }
  184. buildLevelFromEvent(&rootData)
  185. sendDataMap[code_type] = append(sendDataMap[code_type], rootData)
  186. //a, _ := json.Marshal(rootData)
  187. //fmt.Println(string(a))
  188. //sendData = append(sendData, rootData)
  189. //if _, ok := TraceRootMap[traceId]; !ok {
  190. //TraceRootMap[traceId] = &TraceMapT{RootData: initRootData(traceId), Index: 1}
  191. //}
  192. //TraceRootMap[traceId].Index++
  193. //buildAndAssemblyMap(sd, TraceRootMap[traceId])
  194. }
  195. // 发送完整数据 | 大量长耗时请求会增加内存占用
  196. //sendData := []RootDataT{}
  197. //for traceId, v := range TraceRootMap {
  198. // if v.TheEnd {
  199. // buildLevel(v)
  200. // sendData = append(sendData, v.RootData)
  201. // delete(TraceRootMap, traceId)
  202. // //fmt.Println("the end!")
  203. // } else {
  204. // //fmt.Println("not end!")
  205. // }
  206. //}
  207. //Transform the categorized map into a slice
  208. data, _ := json.Marshal(sendDataMap)
  209. klog.Debug(string(data))
  210. //fmt.Println(len(sendData))
  211. //fmt.Println("sdl len:", len(sdl))
  212. return sendDataMap
  213. }
  214. type TimeMap struct {
  215. Time uint64
  216. Type int
  217. Map *MapInfoT
  218. }
  219. //type TraceMapT struct {
  220. // RootData RootDataT
  221. // Index int
  222. // lock *sync.RWMutex
  223. // TheEnd bool
  224. //}
  225. //func buildLevel(sdl *TraceMapT) {
  226. // nidMap := make(map[int]*MapInfoT)
  227. //
  228. // mapSlice := []TimeMap{}
  229. //
  230. // for i, v := range sdl.RootData.Maps {
  231. // if v.ServiceType == "APPLICATION" {
  232. // continue
  233. // }
  234. // nidMap[v.Nid] = &sdl.RootData.Maps[i]
  235. // timeStartMap := TimeMap{
  236. // Time: v.StartTime,
  237. // Type: 0,
  238. // Map: &sdl.RootData.Maps[i],
  239. // }
  240. // mapSlice = append(mapSlice, timeStartMap)
  241. // timeEndMap := TimeMap{
  242. // Time: v.EndTime,
  243. // Type: 1,
  244. // Map: &sdl.RootData.Maps[i],
  245. // }
  246. // mapSlice = append(mapSlice, timeEndMap)
  247. // }
  248. // sort.Slice(mapSlice, func(i, j int) bool {
  249. // return mapSlice[i].Time < mapSlice[j].Time
  250. // })
  251. //
  252. // funStack := []TimeMap{}
  253. //
  254. // currentNid := 1
  255. // Nid := 2
  256. // level := 2
  257. //
  258. // for _, v := range mapSlice {
  259. // // fmt.Println("SliceSliceindex", k, "value", v.Time, v.Type, v.Map.MethodName, v.Map.Nid)
  260. // if v.Type == 0 {
  261. // // 函数入口
  262. // funStack = append(funStack, v)
  263. // v.Map.Pid = currentNid
  264. // v.Map.Level = level
  265. // v.Map.Nid = Nid
  266. // currentNid = Nid
  267. // level += 1
  268. // Nid += 1
  269. // } else if v.Type == 1 {
  270. // // 函数出口
  271. // len := len(funStack)
  272. // funStack = funStack[:len-1]
  273. // if (len - 2) < 0 {
  274. // currentNid = 1
  275. // } else {
  276. // currentNid = funStack[len-2].Map.Nid
  277. // }
  278. //
  279. // level -= 1
  280. // }
  281. // }
  282. //}
  283. func buildLevelFromEvent(sdl *RootDataT) {
  284. nidMap := make(map[int]*MapInfoT)
  285. mapSlice := []TimeMap{}
  286. for i, v := range sdl.Maps {
  287. if v.ServiceType == "APPLICATION" {
  288. continue
  289. }
  290. nidMap[v.Nid] = &sdl.Maps[i]
  291. timeStartMap := TimeMap{
  292. Time: v.StartTime,
  293. Type: 0,
  294. Map: &sdl.Maps[i],
  295. }
  296. mapSlice = append(mapSlice, timeStartMap)
  297. timeEndMap := TimeMap{
  298. Time: v.EndTime,
  299. Type: 1,
  300. Map: &sdl.Maps[i],
  301. }
  302. mapSlice = append(mapSlice, timeEndMap)
  303. }
  304. sort.Slice(mapSlice, func(i, j int) bool {
  305. return mapSlice[i].Time < mapSlice[j].Time
  306. })
  307. funStack := []TimeMap{}
  308. currentNid := 1
  309. Nid := 2
  310. level := 2
  311. for _, v := range mapSlice {
  312. //klog.Debugln("SliceSliceindex", k, "value", v.Time, v.Type, v.Map.MethodName, v.Map.Nid)
  313. if v.Type == 0 {
  314. // 函数入口
  315. funStack = append(funStack, v)
  316. v.Map.Pid = currentNid
  317. v.Map.Level = level
  318. v.Map.Nid = Nid
  319. currentNid = Nid
  320. level += 1
  321. Nid += 1
  322. } else if v.Type == 1 {
  323. // 函数出口
  324. len := len(funStack)
  325. funStack = funStack[:len-1]
  326. if (len - 2) < 0 {
  327. currentNid = 1
  328. } else {
  329. currentNid = funStack[len-2].Map.Nid
  330. }
  331. level -= 1
  332. }
  333. }
  334. }
  335. //func initRootData(traceId string) RootDataT {
  336. // data := RootDataT{
  337. // AccountId: 110,
  338. // AgentId: 1011005252979954, // TODO 更新 基于 ip:port + process_name + exe路径生成
  339. // AgentVersion: "2.1.0",
  340. // AppId: 5410049101545798, // TODO 更新 基于appname生成
  341. // AppIdFrom: -1,
  342. // AppName: "eBPF-agent", // TODO 更新 ip:port || process_name
  343. // CalledId: -1,
  344. // ClientIp: "",
  345. // CollTime: 0,
  346. // Cpu: 0,
  347. // Custom: "",
  348. // HostId: 10154813500555812,
  349. // HostName: "localhost",
  350. // HttpCode: 0,
  351. // HttpMethod: "",
  352. // InstanceId: 1005051101515357, // TODO 更新 基于ip:port
  353. // InstanceIdFrom: -1,
  354. // Maps: []MapInfoT{},
  355. // MemU: 0,
  356. // MemUP: 0,
  357. // OperType: "",
  358. // Parameters: []interface{}{},
  359. // ParentTaskName: 0,
  360. // Period: -1,
  361. // RespTime: 0,
  362. // Sampling: 0,
  363. // ServiceName: "GO",
  364. // ServiceType: APP_SERVICE_TYPE,
  365. // Sip: "",
  366. // Sn: "",
  367. // SpanIdFrom: "",
  368. // Sport: 0,
  369. // TId: -1,
  370. // TName: "",
  371. // TraceId: traceId,
  372. // TransIds: []interface{}{},
  373. // TypeFrom: "",
  374. // Uri: "",
  375. // UserDir: 0,
  376. // VipIds: []interface{}{},
  377. // }
  378. // return data
  379. //}
  380. func initRootDataFromEvent() RootDataT {
  381. hostID := utils.GetHostID()
  382. accountID := utils.GetAccountID()
  383. sysTag := utils.GetSysTag()
  384. systemUUID := utils.GetSystemUUID()
  385. data := RootDataT{
  386. // todo AccountId
  387. AccountId: accountID,
  388. AgentId: 0, // 基于 ip:port + process_name + exe路径生成
  389. AgentVersion: "2.1.0",
  390. AppId: 0, // 基于appname生成
  391. AppIdFrom: -1,
  392. AppName: "eBPF-agent", // server配置
  393. CalledId: -1,
  394. ClientIp: "",
  395. CollTime: 0,
  396. Cpu: 0,
  397. Custom: "",
  398. HostId: hostID,
  399. HostName: "localhost",
  400. HttpCode: 0,
  401. HttpMethod: "",
  402. InstanceId: 0, // 基于ip:port
  403. InstanceIdFrom: -1,
  404. Maps: []MapInfoT{},
  405. MemU: 0,
  406. MemUP: 0,
  407. OperType: "",
  408. Parameters: []ParamStruct{},
  409. ParentTaskName: "",
  410. Period: -1,
  411. RespTime: 0,
  412. Sampling: 0,
  413. ServiceName: "",
  414. ServiceType: APP_SERVICE_TYPE,
  415. Sip: "",
  416. Sn: "",
  417. SpanIdFrom: "",
  418. Sport: 0,
  419. TId: -1,
  420. TName: "",
  421. TraceId: "",
  422. TransIds: []interface{}{},
  423. TypeFrom: "",
  424. Uri: "",
  425. UserDir: "",
  426. VipIds: []interface{}{},
  427. SrcAddr: "",
  428. DestinationAddr: "",
  429. Sys: sysTag,
  430. SystemUUID: systemUUID,
  431. }
  432. return data
  433. }
  434. func initRootDataJava() RootDataT {
  435. data := RootDataT{
  436. AccountId: 110,
  437. AgentId: 3934815089541000, // TODO 更新 基于 ip:port + process_name + exe路径生成
  438. AgentVersion: "2.21.0",
  439. AppId: 3365853273187618, // TODO 更新 基于appname生成
  440. AppIdFrom: -1,
  441. AppName: "eBPF-javaApplication", // TODO 更新 ip:port || process_name
  442. CalledId: -1,
  443. ClientIp: "",
  444. CollTime: 0,
  445. Cpu: 0,
  446. Custom: "",
  447. HostId: 2315065183171055,
  448. HostName: "localhost",
  449. HttpCode: 0,
  450. HttpMethod: "",
  451. InstanceId: 1128864082033413, // TODO 更新 基于ip:port
  452. InstanceIdFrom: -1,
  453. Maps: []MapInfoT{},
  454. MemU: 0,
  455. MemUP: 0,
  456. OperType: "",
  457. Parameters: []ParamStruct{},
  458. ParentTaskName: "",
  459. Period: -1,
  460. RespTime: 0,
  461. Sampling: 0,
  462. ServiceName: "TOMCAT",
  463. ServiceType: APP_SERVICE_TYPE,
  464. Sip: "",
  465. Sn: "",
  466. SpanIdFrom: "",
  467. Sport: 0,
  468. TId: -1,
  469. TName: "",
  470. TraceId: "",
  471. TransIds: []interface{}{},
  472. TypeFrom: "",
  473. Uri: "",
  474. UserDir: "",
  475. VipIds: []interface{}{},
  476. }
  477. return data
  478. }
  479. //func initMapNode(spanSd *tracepb.Span) (MapInfoT, string) {
  480. // mNode := MapInfoT{
  481. // Exception: 0,
  482. // ExceptionMsg: "",
  483. // ExceptionStack: "",
  484. // Ip: "",
  485. // Level: 2,
  486. // Pid: 1,
  487. // Port: 0,
  488. // Ps: []string{},
  489. // ServiceName: "",
  490. // ServiceType: "",
  491. // WallTime: 0,
  492. // }
  493. // mNode.MethodName = spanSd.Name
  494. // mNode.PureTime = (spanSd.EndTimeUnixNano - spanSd.StartTimeUnixNano) / 1e3
  495. // mNode.WallTime = mNode.PureTime
  496. // mNode.StartTime = spanSd.StartTimeUnixNano
  497. // mNode.EndTime = spanSd.EndTimeUnixNano
  498. //
  499. // for _, attr := range spanSd.GetAttributes() {
  500. // fmt.Println(attr.Key, ":", attr.Value.GetValue())
  501. //
  502. // switch attr.Key {
  503. // case "nid":
  504. // mNode.Nid = int(attr.Value.GetIntValue())
  505. // case "pid":
  506. // mNode.Pid = int(attr.Value.GetIntValue())
  507. // case "level":
  508. // mNode.Level = int(attr.Value.GetIntValue())
  509. // }
  510. // }
  511. //
  512. // return mNode, spanSd.Name
  513. //}
  514. func buildMapNodeFromEvent(event tracesdk.Event) MapInfoT {
  515. mNode := MapInfoT{
  516. Exception: 0,
  517. ExceptionMsg: "",
  518. ExceptionStack: "",
  519. Ip: "",
  520. Level: 2,
  521. Pid: 1,
  522. Port: 0,
  523. Ps: []string{},
  524. ServiceName: "",
  525. ServiceType: "",
  526. WallTime: 0,
  527. }
  528. mNode.MethodName = event.Name
  529. //mNode.PureTime = (event.EndTimeUnixNano - event.StartTimeUnixNano) / 1e3
  530. //mNode.WallTime = mNode.PureTime
  531. //mNode.StartTime = spanSd.StartTimeUnixNano
  532. //mNode.EndTime = spanSd.EndTimeUnixNano
  533. for _, attr := range event.Attributes {
  534. //fmt.Println(event.Name, "--->buildMapNodeFromEvent--->", attr.Key, ":", attr.Value.AsInterface())
  535. switch attr.Key {
  536. case "nid":
  537. mNode.Nid = int(attr.Value.AsInt64())
  538. case "pid":
  539. mNode.Pid = int(attr.Value.AsInt64())
  540. case "level":
  541. mNode.Level = int(attr.Value.AsInt64())
  542. case "time.start_at":
  543. mNode.StartTime, mNode.StartTimeMs = cleanNsTime(attr.Value.AsInt64())
  544. case "time.end_at":
  545. mNode.EndTime, mNode.EndTimeMs = cleanNsTime(attr.Value.AsInt64())
  546. case "time.duration":
  547. //mNode.PureTime = uint64(attr.Value.AsInt64()) / 1e3
  548. mNode.WallTime = uint64(attr.Value.AsInt64()) / 1e3
  549. }
  550. }
  551. return mNode
  552. }
  553. func parseURIToParams(input string) (string, []ParamStruct, error) {
  554. // 解析输入 URI
  555. parsedURL, err := url.Parse(input)
  556. if err != nil {
  557. return "", nil, fmt.Errorf("failed to parse URI: %w", err)
  558. }
  559. // 提取查询参数
  560. queryParams := parsedURL.Query()
  561. // 转换为目标结构
  562. var params []ParamStruct
  563. for key, values := range queryParams {
  564. params = append(params, ParamStruct{
  565. Name: key,
  566. Values: values,
  567. })
  568. }
  569. return parsedURL.Path, params, nil
  570. }
  571. // 构建拼装
  572. //func buildAndAssemblyMap(sd apmTraceSpan, traceRoot *TraceMapT) MapInfoT {
  573. // mNode, mapType := initMapNode(span(sd))
  574. // switch mapType {
  575. // case "APPLICATION":
  576. // buildAppMap(&mNode, traceRoot, sd)
  577. // traceRoot.TheEnd = true
  578. // case "HTTP":
  579. // buildHttpMap(&mNode, sd)
  580. // case "Mysql":
  581. // buildMysqlMap(&mNode, sd)
  582. // case "Redis":
  583. // buildRedisMap(&mNode, sd)
  584. // }
  585. // if mapType != "" {
  586. // mNode.Nid = traceRoot.Index
  587. // traceRoot.RootData.Maps = append(traceRoot.RootData.Maps, mNode)
  588. // }
  589. // return mNode
  590. //}
  591. //func buildAndAssemblyMapFromEvent(event tracesdk.Event, traceRoot *RootDataT) MapInfoT {
  592. // mNode := buildMapNodeFromEvent(event)
  593. // switch mapType {
  594. // case "HTTP":
  595. // buildHttpMapFromEvent(mNode, event)
  596. // //case "Mysql":
  597. // // buildMysqlMap(mNode, sd)
  598. // //case "Redis":
  599. // // buildRedisMap(mNode, sd)
  600. // }
  601. // if mapType != "" {
  602. // //mNode.Nid = traceRoot.Index
  603. // traceRoot.Maps = append(traceRoot.Maps, mNode)
  604. // }
  605. // return mNode
  606. //}
  607. //func buildAppMap(mNode *MapInfoT, traceRoot *TraceMapT, sd apmTraceSpan) {
  608. // mNode.ServiceName = GO_SERVICE_NAME
  609. // mNode.ServiceType = APP_SERVICE_TYPE
  610. // mNode.MethodName = "net/http.(*Transport).roundTrip()"
  611. // mNode.Level = 1
  612. // mNode.Pid = 0
  613. // mNode.Nid = 1
  614. // // 构建root节点
  615. // traceRoot.RootData.RespTime = mNode.PureTime
  616. // traceRoot.RootData.CollTime = mNode.StartTime
  617. // traceRoot.Index = 1
  618. // for _, attr := range sd.Attributes() {
  619. // fmt.Println(attr.Key, ":", attr.Value.AsInterface())
  620. // switch attr.Key {
  621. // case "http.uri":
  622. // traceRoot.RootData.Uri = attr.Value.AsString()
  623. // case "http.method":
  624. // traceRoot.RootData.HttpMethod = attr.Value.AsString()
  625. // case "http.status_code":
  626. // traceRoot.RootData.HttpCode = attr.Value.AsInt64()
  627. // case "net.peer.name":
  628. // traceRoot.RootData.ClientIp = attr.Value.AsString()
  629. // traceRoot.RootData.Sip = attr.Value.AsString()
  630. // traceRoot.RootData.Sn = attr.Value.AsString()
  631. // case "net.peer.port":
  632. // traceRoot.RootData.Sport = attr.Value.AsInt64()
  633. // traceRoot.RootData.LocalPort = attr.Value.AsInt64()
  634. // case "server.trace_id_from":
  635. // traceRoot.RootData.TraceId = attr.Value.AsString()
  636. // case "server.called_id":
  637. // traceRoot.RootData.CalledId = attr.Value.AsInt64()
  638. // case "server.instance_id_from":
  639. // traceRoot.RootData.InstanceIdFrom = attr.Value.AsInt64()
  640. // case "server.app_id_from":
  641. // traceRoot.RootData.AppIdFrom = attr.Value.AsInt64()
  642. // case "server.span_id_from":
  643. // traceRoot.RootData.SpanIdFrom = attr.Value.AsString()
  644. // case "server.type_from":
  645. // traceRoot.RootData.TypeFrom = attr.Value.AsString()
  646. // }
  647. // }
  648. //
  649. //}
  650. func buildAppMapFromEvent(traceRoot *RootDataT, sd apmTraceSpan) int {
  651. mNode := MapInfoT{
  652. Exception: 0,
  653. ExceptionMsg: "",
  654. ExceptionStack: "",
  655. Ip: "",
  656. Level: 1,
  657. Pid: 1,
  658. Port: 0,
  659. Ps: []string{},
  660. ServiceName: "",
  661. ServiceType: "",
  662. WallTime: 0,
  663. }
  664. mNode.ServiceName = GO_SERVICE_NAME
  665. mNode.ServiceType = APP_SERVICE_TYPE
  666. mNode.MethodName = "Kernel Endpoint()"
  667. mNode.Level = 1
  668. mNode.Pid = 0
  669. mNode.Nid = 1
  670. var code_type int64
  671. // 构建root节点
  672. //traceRoot.RespTime = mNode.PureTimex
  673. //traceRoot.CollTime = mNode.StartTime
  674. for _, attr := range sd.Attributes() {
  675. klog.Debugln("Appmap:", attr.Key, ":", attr.Value.AsInterface())
  676. switch attr.Key {
  677. case "http.uri":
  678. traceRoot.Uri, traceRoot.Parameters, _ = parseURIToParams(attr.Value.AsString())
  679. case "http.method":
  680. traceRoot.HttpMethod = attr.Value.AsString()
  681. case "http.status_code":
  682. traceRoot.HttpCode = attr.Value.AsInt64()
  683. case "net.peer.name":
  684. // TODO 修改 ClientIp sip获取方式
  685. traceRoot.ClientIp = attr.Value.AsString()
  686. traceRoot.Sip = attr.Value.AsString()
  687. traceRoot.Sn = attr.Value.AsString()
  688. case "net.peer.port":
  689. traceRoot.Sport = attr.Value.AsInt64()
  690. traceRoot.LocalPort = attr.Value.AsInt64()
  691. case "server.trace_id_from":
  692. traceRoot.TraceId = attr.Value.AsString()
  693. case "server.called_id":
  694. traceRoot.CalledId = attr.Value.AsInt64()
  695. case "server.instance_id_from":
  696. traceRoot.InstanceIdFrom = attr.Value.AsInt64()
  697. case "server.app_id_from":
  698. traceRoot.AppIdFrom = attr.Value.AsInt64()
  699. case "server.span_id_from":
  700. traceRoot.SpanIdFrom = attr.Value.AsString()
  701. case "server.type_from":
  702. traceRoot.TypeFrom = attr.Value.AsString()
  703. case "time.start_at":
  704. mNode.StartTime, mNode.StartTimeMs = cleanNsTime(attr.Value.AsInt64())
  705. traceRoot.CollTime = mNode.StartTimeMs
  706. case "time.end_at":
  707. mNode.EndTime, mNode.EndTimeMs = cleanNsTime(attr.Value.AsInt64())
  708. case "time.duration":
  709. traceRoot.RespTime = uint64(attr.Value.AsInt64()) / 1e3
  710. mNode.PureTime = traceRoot.RespTime
  711. mNode.WallTime = uint64(attr.Value.AsInt64()) / 1e3
  712. case "server.code_type":
  713. code_type = attr.Value.AsInt64()
  714. case "server.app_name":
  715. traceRoot.AppName = attr.Value.AsString()
  716. case "server.service_name":
  717. traceRoot.ServiceName = attr.Value.AsString()
  718. mNode.ServiceName = attr.Value.AsString()
  719. case "server.app_id":
  720. traceRoot.AppId = attr.Value.AsInt64()
  721. case "server.agent_id":
  722. traceRoot.AgentId = attr.Value.AsInt64()
  723. case "server.instance_id":
  724. traceRoot.InstanceId = attr.Value.AsInt64()
  725. case "server.src_addr":
  726. traceRoot.SrcAddr = attr.Value.AsString()
  727. case "server.dst_addr":
  728. traceRoot.DestinationAddr = attr.Value.AsString()
  729. case "server.pid":
  730. traceRoot.Pid = uint32(attr.Value.AsInt64())
  731. case "server.container_id":
  732. traceRoot.ContainerID = attr.Value.AsString()
  733. }
  734. }
  735. traceRoot.Maps = append(traceRoot.Maps, mNode)
  736. return int(code_type)
  737. }
  738. //func buildHttpMap(mNode *MapInfoT, sd apmTraceSpan) {
  739. // mNode.ServiceName = HTTP_SERVICE_NAME
  740. // mNode.ServiceType = HTTP_SERVICE_TYPE
  741. // mNode.Schema = "http"
  742. // mNode.MethodName = "net/http.serverHandler.ServeHTTP()"
  743. // var descAddr string
  744. // for _, attr := range sd.Attributes() {
  745. // //fmt.Println(attr.Key, ":", attr.Value.AsInterface())
  746. // switch attr.Key {
  747. // case "http.ip":
  748. // mNode.Ip = attr.Value.AsString()
  749. // descAddr += mNode.Ip
  750. // case "http.port":
  751. // mNode.Port = attr.Value.AsInt64()
  752. // descAddr += ":" + attr.Value.AsString()
  753. // case "http.uri":
  754. // mNode.Uri = attr.Value.AsString()
  755. // case "http.assumed_app_id":
  756. // mNode.AssumedAppId = attr.Value.AsInt64()
  757. // case "http.span_id":
  758. // mNode.SpanId = attr.Value.AsString()
  759. // }
  760. // }
  761. // //mNode.AssumedAppId = Md5ToInt64(descAddr, 16)
  762. //}
  763. func buildHttpMapFromEvent(mNode *MapInfoT, event tracesdk.Event) {
  764. mNode.ServiceName = HTTP_SERVICE_NAME
  765. mNode.ServiceType = HTTP_SERVICE_TYPE
  766. mNode.Schema = "http"
  767. //mNode.MethodName = "HTTP"
  768. //var descAddr string
  769. var method string
  770. for _, attr := range event.Attributes {
  771. klog.Debugln("HTTP--->", attr.Key, ":", attr.Value.AsInterface())
  772. switch attr.Key {
  773. case "http.ip":
  774. mNode.Ip = attr.Value.AsString()
  775. //descAddr += mNode.Ip
  776. case "http.port":
  777. mNode.Port = attr.Value.AsInt64()
  778. case "http.method":
  779. //mNode.MethodName += " " + attr.Value.AsString()
  780. method = attr.Value.AsString()
  781. //descAddr += ":" + attr.Value.AsString()
  782. case "http.uri":
  783. mNode.Uri = attr.Value.AsString()
  784. //mNode.MethodName += " " + attr.Value.AsString()
  785. case "http.assumed_app_id":
  786. mNode.AssumedAppId = attr.Value.AsInt64()
  787. case "http.span_id":
  788. mNode.SpanId = attr.Value.AsString()
  789. case "time.start_at":
  790. mNode.StartTime = uint64(attr.Value.AsInt64())
  791. case "time.end_at":
  792. mNode.EndTime = uint64(attr.Value.AsInt64())
  793. case "time.duration":
  794. //mNode.PureTime = uint64(attr.Value.AsInt64()) / 1e3
  795. mNode.WallTime = uint64(attr.Value.AsInt64()) / 1e3
  796. case "http.src_addr":
  797. mNode.SrcAddr = attr.Value.AsString()
  798. case "http.destination_addr":
  799. mNode.DestinationAddr = attr.Value.AsString()
  800. }
  801. }
  802. mNode.MethodName = fmt.Sprintf("%s %s %s:%d%s", "HTTP", method, mNode.Ip, mNode.Port, mNode.Uri)
  803. //mNode.AssumedAppId = Md5ToInt64(descAddr, 16)
  804. }
  805. //func buildMysqlMap(mNode *MapInfoT, sd apmTraceSpan) {
  806. // mNode.Dbn = "unknown"
  807. // mNode.ServiceName = MYSQL_SERVICE_NAME
  808. // mNode.ServiceType = SQL_SERVICE_TYPE
  809. // mNode.MethodName = "database/sql.Query()"
  810. // for _, attr := range sd.Attributes() {
  811. // //fmt.Println(attr.Key, ":", attr.Value.AsInterface())
  812. // switch attr.Key {
  813. // case "net.peer.name":
  814. // mNode.Ip = attr.Value.AsString()
  815. // case "net.peer.port":
  816. // mNode.Port = attr.Value.AsInt64()
  817. // case "db.statement":
  818. // query := attr.Value.AsString()
  819. // mNode.Ps = []string{query}
  820. // words := strings.Fields(query)
  821. // if len(words) > 0 {
  822. // mNode.OperType = strings.ToUpper(words[0])
  823. // }
  824. // }
  825. // }
  826. //}
  827. func buildSQLMapEvent(mNode *MapInfoT, event tracesdk.Event) {
  828. mNode.Dbn = "-"
  829. mNode.ServiceName = l7.Protocol(event.ProtocolType).ServiceNameString()
  830. mNode.ServiceType = SQL_SERVICE_TYPE
  831. //mNode.MethodName = "database/sql.Query()"
  832. for _, attr := range event.Attributes {
  833. //fmt.Println(attr.Key, ":", attr.Value.AsInterface())
  834. switch attr.Key {
  835. case "net.peer.name":
  836. mNode.Ip = attr.Value.AsString()
  837. case "net.peer.port":
  838. mNode.Port = attr.Value.AsInt64()
  839. case "db.statement":
  840. query := attr.Value.AsString()
  841. mNode.MethodName = query
  842. mNode.Ps = []string{query}
  843. //words := strings.Fields(query)
  844. //if len(words) > 0 {
  845. // mNode.OperType = strings.ToUpper(words[0])
  846. //}
  847. case "sql.exception":
  848. if attr.Value.AsBool() {
  849. mNode.Exception = 1
  850. } else {
  851. mNode.Exception = 0
  852. }
  853. case "sql.exception_msg":
  854. mNode.ExceptionMsg = attr.Value.AsString()
  855. case "sql.src_addr":
  856. mNode.SrcAddr = attr.Value.AsString()
  857. case "sql.destination_addr":
  858. mNode.DestinationAddr = attr.Value.AsString()
  859. }
  860. }
  861. }
  862. func buildDNSMapEvent(mNode *MapInfoT, event tracesdk.Event) {
  863. mNode.ServiceName = l7.Protocol(event.ProtocolType).ServiceNameString()
  864. mNode.ServiceType = NET_SERVICE_TYPE
  865. var _type string
  866. var fqdn string
  867. var ips string
  868. var ttl int64
  869. for _, attr := range event.Attributes {
  870. switch attr.Key {
  871. case "dns.type":
  872. _type = attr.Value.AsString()
  873. case "dns.fqdn":
  874. fqdn = attr.Value.AsString()
  875. case "dns.ttl":
  876. ttl = attr.Value.AsInt64()
  877. case "dns.ips":
  878. if attr.Value.AsString() != "" {
  879. ips = "Addr: " + attr.Value.AsString()
  880. }
  881. }
  882. }
  883. mNode.MethodName = fmt.Sprintf("DNS Name: %s Type: %s TTL: %d %s", fqdn, _type, ttl, ips)
  884. }
  885. func buildPostGreSqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
  886. mNode.Dbn = "-"
  887. mNode.ServiceName = POSTGRESQL_SERVICE_NAME
  888. mNode.ServiceType = SQL_SERVICE_TYPE
  889. //mNode.MethodName = "database/sql.Query()"
  890. for _, attr := range event.Attributes {
  891. //fmt.Println(attr.Key, ":", attr.Value.AsInterface())
  892. switch attr.Key {
  893. case "net.peer.name":
  894. mNode.Ip = attr.Value.AsString()
  895. case "net.peer.port":
  896. mNode.Port = attr.Value.AsInt64()
  897. case "db.statement":
  898. query := attr.Value.AsString()
  899. mNode.Ps = []string{query}
  900. //words := strings.Fields(query)
  901. //if len(words) > 0 {
  902. // mNode.OperType = strings.ToUpper(words[0])
  903. //}
  904. case "sql.exception":
  905. if attr.Value.AsBool() {
  906. mNode.Exception = 1
  907. } else {
  908. mNode.Exception = 0
  909. }
  910. case "sql.src_addr":
  911. mNode.SrcAddr = attr.Value.AsString()
  912. case "sql.destination_addr":
  913. mNode.DestinationAddr = attr.Value.AsString()
  914. }
  915. }
  916. }
  917. func buildMysqlMapEvent(mNode *MapInfoT, event tracesdk.Event) {
  918. mNode.Dbn = "-"
  919. mNode.ServiceName = MYSQL_SERVICE_NAME
  920. mNode.ServiceType = SQL_SERVICE_TYPE
  921. //mNode.MethodName = "database/sql.Query()"
  922. for _, attr := range event.Attributes {
  923. //fmt.Println(attr.Key, ":", attr.Value.AsInterface())
  924. switch attr.Key {
  925. case "net.peer.name":
  926. mNode.Ip = attr.Value.AsString()
  927. case "net.peer.port":
  928. mNode.Port = attr.Value.AsInt64()
  929. case "db.statement":
  930. query := attr.Value.AsString()
  931. mNode.MethodName = query
  932. mNode.Ps = []string{query}
  933. //words := strings.Fields(query)
  934. //if len(words) > 0 {
  935. // mNode.OperType = strings.ToUpper(words[0])
  936. //}
  937. case "sql.exception":
  938. if attr.Value.AsBool() {
  939. mNode.Exception = 1
  940. } else {
  941. mNode.Exception = 0
  942. }
  943. case "sql.src_addr":
  944. mNode.SrcAddr = attr.Value.AsString()
  945. case "sql.destination_addr":
  946. mNode.DestinationAddr = attr.Value.AsString()
  947. }
  948. }
  949. }
  950. func buildDMMapEvent(mNode *MapInfoT, event tracesdk.Event) {
  951. mNode.Dbn = "TEST"
  952. mNode.ServiceName = DM_SERVICE_NAME
  953. mNode.ServiceType = SQL_SERVICE_TYPE
  954. mNode.MethodName = "database/sql.Query()"
  955. for _, attr := range event.Attributes {
  956. //fmt.Println(attr.Key, ":", attr.Value.AsInterface())
  957. switch attr.Key {
  958. case "net.peer.name":
  959. mNode.Ip = attr.Value.AsString()
  960. case "net.peer.port":
  961. mNode.Port = attr.Value.AsInt64()
  962. case "db.statement":
  963. query := attr.Value.AsString()
  964. mNode.Ps = []string{query}
  965. //words := strings.Fields(query)
  966. //if len(words) > 0 {
  967. // mNode.OperType = strings.ToUpper(words[0])
  968. //}
  969. case "sql.exception":
  970. if attr.Value.AsBool() {
  971. mNode.Exception = 1
  972. } else {
  973. mNode.Exception = 0
  974. }
  975. case "sql.src_addr":
  976. mNode.SrcAddr = attr.Value.AsString()
  977. case "sql.destination_addr":
  978. mNode.DestinationAddr = attr.Value.AsString()
  979. }
  980. }
  981. }
  982. func buildRedisMapEvent(mNode *MapInfoT, event tracesdk.Event) {
  983. mNode.ServiceName = REDIS_SERVICE_NAME
  984. mNode.ServiceType = NOSQL_SERVICE_TYPE
  985. //mNode.MethodName = span(sd).Name + " query"
  986. //mNode.MethodName = "redis.Do()"
  987. for _, attr := range event.Attributes {
  988. //fmt.Println(attr.Key, ":", attr.Value.AsInterface())
  989. switch attr.Key {
  990. case "net.peer.name":
  991. mNode.Ip = attr.Value.AsString()
  992. case "net.peer.port":
  993. mNode.Port = attr.Value.AsInt64()
  994. case "db.statement":
  995. query := attr.Value.AsString()
  996. mNode.MethodName = query
  997. mNode.Ps = []string{query}
  998. //words := strings.Fields(query)
  999. //if len(words) > 0 {
  1000. // mNode.OperType = strings.ToUpper(words[0])
  1001. //}
  1002. case "nosql.src_addr":
  1003. mNode.SrcAddr = attr.Value.AsString()
  1004. case "nosql.destination_addr":
  1005. mNode.DestinationAddr = attr.Value.AsString()
  1006. }
  1007. }
  1008. }
  1009. func buildMongoMapEvent(mNode *MapInfoT, event tracesdk.Event) {
  1010. mNode.ServiceName = MONGO_SERVICE_NAME
  1011. mNode.ServiceType = NOSQL_SERVICE_TYPE
  1012. for _, attr := range event.Attributes {
  1013. switch attr.Key {
  1014. case "net.peer.name":
  1015. mNode.Ip = attr.Value.AsString()
  1016. case "net.peer.port":
  1017. mNode.Port = attr.Value.AsInt64()
  1018. case "db.statement":
  1019. query := attr.Value.AsString()
  1020. mNode.MethodName = query
  1021. mNode.Ps = []string{query}
  1022. case "nosql.src_addr":
  1023. mNode.SrcAddr = attr.Value.AsString()
  1024. case "nosql.destination_addr":
  1025. mNode.DestinationAddr = attr.Value.AsString()
  1026. }
  1027. }
  1028. }
  1029. func isEnter(_type string) bool {
  1030. if _type == "APPLICATION" {
  1031. return true
  1032. }
  1033. return false
  1034. }
  1035. func span(sd apmTraceSpan) *tracepb.Span {
  1036. if sd == nil {
  1037. return nil
  1038. }
  1039. tid := sd.SpanContext().TraceID()
  1040. sid := sd.SpanContext().SpanID()
  1041. s := &tracepb.Span{
  1042. TraceId: tid[:],
  1043. SpanId: sid[:],
  1044. TraceState: sd.SpanContext().TraceState().String(),
  1045. //Status: status(sd.Status().Code, sd.Status().Description),
  1046. StartTimeUnixNano: uint64(sd.StartTime().UnixNano()),
  1047. EndTimeUnixNano: uint64(sd.EndTime().UnixNano()),
  1048. //Links: links(sd.Links()),
  1049. //Kind: spanKind(sd.SpanKind()),
  1050. Name: sd.Name(),
  1051. Attributes: tracetransform.KeyValues(sd.Attributes()),
  1052. //Events: spanEvents(sd.Events()),
  1053. DroppedAttributesCount: uint32(sd.DroppedAttributes()),
  1054. DroppedEventsCount: uint32(sd.DroppedEvents()),
  1055. DroppedLinksCount: uint32(sd.DroppedLinks()),
  1056. }
  1057. if psid := sd.Parent().SpanID(); psid.IsValid() {
  1058. s.ParentSpanId = psid[:]
  1059. }
  1060. return s
  1061. }
  1062. func Md5ToInt64(strParam string, Len int) int64 {
  1063. sign := md5.Sum([]byte(strParam))
  1064. signStr := fmt.Sprintf("%x", sign)
  1065. charArr := []rune(signStr)
  1066. var intStr string
  1067. for _, value := range charArr {
  1068. intStr += strconv.Itoa(int(value))
  1069. }
  1070. intStr = intStr[:Len]
  1071. int64Data, err := strconv.ParseInt(intStr, 10, 64)
  1072. if err != nil {
  1073. return 0
  1074. }
  1075. return int64Data
  1076. }
  1077. // ns,ms
  1078. func cleanNsTime(time int64) (uint64, uint64) {
  1079. return uint64(time), uint64(math.Round(float64(time) / 1e6))
  1080. }