ebpf/pkg/go.opentelemetry.io/otel/exporters/otlp/otlptrace/apm_exporter.go

package otlptrace

import (
	"encoding/json"
	"fmt"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/internal/tracetransform"
	tracesdk "go.opentelemetry.io/otel/sdk/trace"
	tracepb "go.opentelemetry.io/proto/otlp/trace/v1"
	"strings"
)

type RootData struct {
	AccountId      int           `json:"account_id"`
	AgentId        int64         `json:"agent_id"`
	AgentVersion   string        `json:"agent_version"`
	AppId          int64         `json:"app_id"`
	AppIdFrom      int           `json:"app_id_from"`
	AppName        string        `json:"app_name"`
	CalledId       int           `json:"called_id"`
	ClientIp       string        `json:"client_ip"`
	CollTime       uint64        `json:"coll_time"`
	Cpu            int           `json:"cpu"`
	Custom         string        `json:"custom"`
	HostId         int64         `json:"host_id"`
	HostName       string        `json:"host_name"`
	HttpCode       int64         `json:"http_code"`
	HttpMethod     string        `json:"http_method"`
	InstanceId     int64         `json:"instance_id"`
	InstanceIdFrom int           `json:"instance_id_from"`
	LocalPort      int64         `json:"local_port"`
	Maps           []MapInfo     `json:"maps"`
	MemU           int           `json:"mem_u"`
	MemUP          int           `json:"mem_u_p"`
	OperType       string        `json:"oper_type"`
	Parameters     []interface{} `json:"parameters"`
	ParentTaskName int           `json:"parent_task_name"`
	Period         int           `json:"period"`
	RespTime       uint64        `json:"resp_time"`
	Sampling       int           `json:"sampling"`
	ServiceName    string        `json:"service_name"`
	ServiceType    string        `json:"service_type"`
	Sip            string        `json:"sip"`
	Sn             string        `json:"sn"`
	SpanIdFrom     string        `json:"span_id_from"`
	Sport          int64         `json:"sport"`
	TId            int           `json:"t_id"`
	TName          string        `json:"t_name"`
	TraceId        string        `json:"trace_id"`
	TransIds       []interface{} `json:"trans_ids"`
	TypeFrom       string        `json:"type_from"`
	Uri            string        `json:"uri"`
	UserDir        int           `json:"user_dir"`
	VipIds         []interface{} `json:"vip_ids"`
}

type MapInfo struct {
	Dbn            string   `json:"dbn,omitempty"`
	Exception      int      `json:"exception"`
	ExceptionMsg   string   `json:"exception_msg"`
	ExceptionStack string   `json:"exception_stack"`
	Ip             string   `json:"ip,omitempty"`
	Level          int      `json:"level"`
	MethodDesc     string   `json:"method_desc"`
	MethodName     string   `json:"method_name"`
	Nid            int      `json:"nid"`
	OperType       string   `json:"oper_type"`
	Pid            int      `json:"pid"`
	Port           int64    `json:"port,omitempty"`
	Ps             []string `json:"ps,omitempty"`
	PureTime       uint64   `json:"pure_time"`
	ServiceName    string   `json:"service_name"`
	ServiceType    string   `json:"service_type"`
	StartTime      uint64   `json:"start_time"`
	WallTime       uint64   `json:"wall_time"`
}

type TraceMapT struct {
	RootData RootData `json:"root_data"`
	Index    int      `json:"index"`
}

func tracetransformData(sdl []tracesdk.ReadOnlySpan) []RootData {
	if len(sdl) == 0 {
		return nil
	}

	//traceMap := make(map[string][]MapInfo)
	//traceIndexMap := make(map[string]int)

	traceRoot := make(map[string]*TraceMapT)
	sendData := []RootData{}
	for _, sd := range sdl {
		if sd == nil {
			continue
		}
		traceId := sd.SpanContext().TraceID().String()
		if _, ok := traceRoot[traceId]; !ok {
			traceRoot[traceId] = &TraceMapT{RootData: initRootData(traceId), Index: 1}
		}

		traceRoot[traceId].Index++
		initMapInfo(sd, traceRoot[traceId])
	}
	for _, v := range traceRoot {
		sendData = append(sendData, v.RootData)
	}

	// Transform the categorized map into a slice
	aa, _ := json.Marshal(sendData)
	fmt.Println(string(aa))
	fmt.Println(len(sendData))
	return sendData
}

func initRootData(traceId string) RootData {
	data := RootData{
		AccountId:      110,
		AgentId:        1011005252979954,
		AgentVersion:   "2.1.0",
		AppId:          5410049101545798,
		AppIdFrom:      -1,
		AppName:        "eBPF-agent",
		CalledId:       -1,
		ClientIp:       "",
		CollTime:       0,
		Cpu:            0,
		Custom:         "",
		HostId:         10154813500555812,
		HostName:       "localhost",
		HttpCode:       0,
		HttpMethod:     "GET",
		InstanceId:     1005051101515357,
		InstanceIdFrom: -1,
		Maps:           []MapInfo{},
		MemU:           0,
		MemUP:          0,
		OperType:       "",
		Parameters:     []interface{}{},
		ParentTaskName: 0,
		Period:         -1,
		RespTime:       0,
		Sampling:       0,
		ServiceName:    "GO",
		ServiceType:    "APPLICATION",
		Sip:            "",
		Sn:             "",
		SpanIdFrom:     "",
		Sport:          0,
		TId:            -1,
		TName:          "",
		TraceId:        traceId,
		TransIds:       []interface{}{},
		TypeFrom:       "",
		Uri:            "",
		UserDir:        0,
		VipIds:         []interface{}{},
	}
	return data
}

func initMapInfo(sd tracesdk.ReadOnlySpan, traceRoot *TraceMapT) MapInfo {
	mNode := MapInfo{
		Exception:      0,
		ExceptionMsg:   "",
		ExceptionStack: "",
		Ip:             "",
		Level:          2,
		MethodDesc:     "unknown",
		MethodName:     "unknown",
		Nid:            traceRoot.Index,
		OperType:       "",
		Pid:            1,
		Port:           0,
		Ps:             []string{},
		PureTime:       (span(sd).EndTimeUnixNano - span(sd).StartTimeUnixNano) / 1e3,
		ServiceName:    "",
		ServiceType:    "",
		StartTime:      span(sd).StartTimeUnixNano / 1e6,
		WallTime:       0,
	}
	mNode.WallTime = mNode.PureTime
	mapType := span(sd).Name
	switch mapType {
	case "MYSQL":
		mNode.Dbn = "unknown"
		mNode.ServiceName = mapType
		mNode.ServiceType = "SQL"
		mNode.MethodName = "(mysql)"
		mNode.MethodDesc = "(mysql)"
		for _, attr := range sd.Attributes() {
			fmt.Println(attr.Key, ":", attr.Value.AsInterface())
			switch attr.Key {
			case "net.peer.name":
				mNode.Ip = attr.Value.AsString()
			case "net.peer.port":
				mNode.Port = attr.Value.AsInt64()
			case "db.statement":
				query := attr.Value.AsString()
				mNode.Ps = []string{query}
				upperOperType := strings.ToUpper(query[:6])
				if upperOperType == "SELECT" ||
					upperOperType == "UPDATE" ||
					upperOperType == "INSERT" ||
					upperOperType == "DELETE" {
					mNode.OperType = upperOperType
				}
			}
		}
		traceRoot.RootData.Maps = append(traceRoot.RootData.Maps, mNode)
	case "APPLICATION":
		mNode.ServiceName = "GO"
		mNode.ServiceType = "APPLICATION"
		mNode.MethodName = "main"
		mNode.MethodDesc = "main"
		mNode.Level = 1
		mNode.Nid = 1
		mNode.Pid = 0
		// 构建root节点
		traceRoot.RootData.RespTime = mNode.PureTime
		traceRoot.RootData.CollTime = mNode.StartTime
		for _, attr := range sd.Attributes() {
			switch attr.Key {
			case "http.uri":
				traceRoot.RootData.Uri = attr.Value.AsString()
			case "http.method":
				traceRoot.RootData.HttpMethod = attr.Value.AsString()
			case "http.status_code":
				traceRoot.RootData.HttpCode = attr.Value.AsInt64()
			case "net.peer.name":
				traceRoot.RootData.ClientIp = attr.Value.AsString()
				traceRoot.RootData.Sip = attr.Value.AsString()
				traceRoot.RootData.Sn = attr.Value.AsString()
			case "net.peer.port":
				traceRoot.RootData.Sport = attr.Value.AsInt64()
				traceRoot.RootData.LocalPort = attr.Value.AsInt64()
			}
		}
		traceRoot.RootData.Maps = append([]MapInfo{mNode}, traceRoot.RootData.Maps...)
	}
	return mNode
}

func isEnter(_type string) bool {
	if _type == "APPLICATION" {
		return true
	}
	return false
}

func span(sd tracesdk.ReadOnlySpan) *tracepb.Span {
	if sd == nil {
		return nil
	}

	tid := sd.SpanContext().TraceID()
	sid := sd.SpanContext().SpanID()

	s := &tracepb.Span{
		TraceId:    tid[:],
		SpanId:     sid[:],
		TraceState: sd.SpanContext().TraceState().String(),
		//Status:                 status(sd.Status().Code, sd.Status().Description),
		StartTimeUnixNano: uint64(sd.StartTime().UnixNano()),
		EndTimeUnixNano:   uint64(sd.EndTime().UnixNano()),
		//Links:                  links(sd.Links()),
		//Kind:                   spanKind(sd.SpanKind()),
		Name:       sd.Name(),
		Attributes: tracetransform.KeyValues(sd.Attributes()),
		//Events:                 spanEvents(sd.Events()),
		DroppedAttributesCount: uint32(sd.DroppedAttributes()),
		DroppedEventsCount:     uint32(sd.DroppedEvents()),
		DroppedLinksCount:      uint32(sd.DroppedLinks()),
	}

	if psid := sd.Parent().SpanID(); psid.IsValid() {
		s.ParentSpanId = psid[:]
	}

	return s
}