ebpf/tracing/apm_tracing.go

package tracing

import (
	"context"
	"fmt"
	"github.com/coroot/coroot-node-agent/ebpftracer"
	"github.com/coroot/coroot-node-agent/utils"
	"sync/atomic"
	"time"

	"strconv"

	"github.com/coroot/coroot-node-agent/ebpftracer/l7"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	semconv "go.opentelemetry.io/otel/semconv/v1.18.0"
	"go.opentelemetry.io/otel/trace"
	"inet.af/netaddr"
)

/**
 * Trace
 */

func (t *Trace) setContext(ctx context.Context) {
	t.lock.Lock()
	defer t.lock.Unlock()
	t.ctx = ctx
}

func (t *Trace) setSpan(span trace.Span) {
	t.lock.Lock()
	defer t.lock.Unlock()
	t.span = span
}

func (t *Trace) startReady() {
	t.lock.Lock()
	defer t.lock.Unlock()
	t.startEventReady = true
}

func (t *Trace) endReadyEvent(needCount uint32) {
	t.lock.Lock()
	defer t.lock.Unlock()
	t.endEventReady = true
	t.needEventCount = needCount
}

func (t *Trace) AllEventReady(traceID uint64) bool {
	//fmt.Printf("[AllEventReady (current/need)|start|end|traceid](%d/%d)%v|%v|%d\n", *t.currenEventCount, t.needEventCount, t.startEventReady, t.endEventReady, traceID)
	return t.startEventReady && t.endEventReady && *t.currenEventCount >= t.needEventCount
}

func (t *Trace) TraceStartEvent(method, path string, status l7.Status, addr netaddr.IPPort) {
	t.span.SetAttributes(semconv.HTTPURL(fmt.Sprintf("http://%s%s", addr.String(), path)),
		semconv.HTTPMethod(method),
		attribute.String("http.uri", path))
	if status > 399 {
		t.span.SetStatus(codes.Error, "")
	}
	t.destination = addr
	t.commonAttrs = []attribute.KeyValue{
		semconv.NetPeerName(addr.IP().String()),
		semconv.NetPeerPort(int(addr.Port())),
	}
	t.span.SetAttributes(t.commonAttrs...)
	t.startReady()
}

// set context span
func (t *Trace) CreateRootSpan(traceId uint64) {
	traceIdStr := strconv.Itoa(int(traceId))
	ctx, span := tracer(t.containerId).Start(context.Background(), traceIdStr, trace.WithSpanKind(trace.SpanKindClient))
	t.setContext(ctx)
	t.setSpan(span)
}

func (t *Trace) TraceStart(method, path string, status l7.Status, duration time.Duration) {
	if t == nil || method == "" {
		return
	}
	t.createParentSpan("APPLICATION", duration, status >= 400,
		semconv.HTTPURL(fmt.Sprintf("http://%s%s", t.destination.String(), path)),
		semconv.HTTPMethod(method),
		//semconv.HTTPStatusCode(int(status)),
		attribute.String("http.uri", path),
	)
}

func (t *Trace) TraceEnd(r *l7.RequestData) {
	if t == nil {
		return
	}
	t.span.SetAttributes(
		semconv.HTTPStatusCode(int(r.Status)),
		attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
	)

	CalledId, err := strconv.ParseInt(r.ParentSpanContext.CalledId, 10, 64)
	if err == nil && CalledId != 0 {
		t.span.SetAttributes(attribute.Int64("server.called_id", CalledId))
	}

	InstanceIdFrom, err := strconv.ParseInt(r.ParentSpanContext.InstanceIdFrom, 10, 64)
	if err == nil && InstanceIdFrom != 0 {
		t.span.SetAttributes(attribute.Int64("server.instance_id_from", InstanceIdFrom))
	}

	AppIdFrom, err := strconv.ParseInt(r.ParentSpanContext.AppIdFrom, 10, 64)
	if err == nil && AppIdFrom != 0 {
		t.span.SetAttributes(attribute.Int64("server.app_id_from", AppIdFrom))
	}
	if r.ParentSpanContext.SpanIdFrom != "0000000000000000" {
		t.span.SetAttributes(attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
	}
	t.span.End(trace.WithTimestamp(time.Now()))
}

// 新增结束事件
func (t *Trace) TraceEndEvent(r *l7.RequestData) {
	if t == nil {
		return
	}
	var attr []attribute.KeyValue
	attr = append(attr,
		semconv.HTTPStatusCode(int(r.Status)),
		attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
	)
	//t.span.SetAttributes(
	//	semconv.HTTPStatusCode(int(r.Status)),
	//	attribute.String("server.trace_id_from", r.ParentSpanContext.TraceIdFrom),
	//)

	calledId, err := strconv.ParseInt(r.ParentSpanContext.CalledId, 10, 64)
	if err == nil && calledId != 0 {
		attr = append(attr, attribute.Int64("server.called_id", calledId))
		//t.span.SetAttributes(attribute.Int64("server.called_id", CalledId))
	}

	instanceIdFrom, err := strconv.ParseInt(r.ParentSpanContext.InstanceIdFrom, 10, 64)
	if err == nil && instanceIdFrom != 0 {
		attr = append(attr, attribute.Int64("server.instance_id_from", instanceIdFrom))

		//t.span.SetAttributes(attribute.Int64("server.instance_id_from", InstanceIdFrom))
	}

	appIdFrom, err := strconv.ParseInt(r.ParentSpanContext.AppIdFrom, 10, 64)
	if err == nil && appIdFrom != 0 {
		attr = append(attr, attribute.Int64("server.app_id_from", appIdFrom))
		//t.span.SetAttributes(attribute.Int64("server.app_id_from", AppIdFrom))
	}
	if r.ParentSpanContext.SpanIdFrom != "0000000000000000" {
		attr = append(attr, attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
		//t.span.SetAttributes(attribute.String("server.span_id_from", r.ParentSpanContext.SpanIdFrom))
	}

	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
	t.span.SetAttributes(attr...)
	t.endReadyEvent(r.EventCount)
}

func (t *Trace) appendTimestamp(attr *[]attribute.KeyValue, s, e uint64, d int64) {
	*attr = append(*attr,
		attribute.Int64("time.start_at", utils.KtimeToTimestamp(s)),
		attribute.Int64("time.end_at", utils.KtimeToTimestamp(e)),
		attribute.Int64("time.duration", d),
	)
}

func (t *Trace) createParentSpan(name string, duration time.Duration, error bool, attrs ...attribute.KeyValue) {
	end := time.Now()
	start := end.Add(-duration)
	ctx, span := tracer(t.containerId).Start(context.Background(), name, trace.WithTimestamp(start), trace.WithSpanKind(trace.SpanKindClient))
	span.SetAttributes(attrs...)
	span.SetAttributes(t.commonAttrs...)
	if error {
		span.SetStatus(codes.Error, "")
	}
	t.setContext(ctx)
	t.setSpan(span)
}

func (t *Trace) SendEvent() {
	t.span.End()
}

func (t *Trace) GetSpan() trace.Span {
	return t.span
}

func (t *Trace) createTraceEvent(name string, eventType int, l7Type int, attrs ...attribute.KeyValue) {
	t.span.AddEventApm(name, eventType, l7Type, trace.WithAttributes(attrs...))
	atomic.AddUint32(t.currenEventCount, 1)
}

func (t *Trace) createTraceSpan(name string, duration time.Duration, error bool, attrs ...attribute.KeyValue) {
	end := time.Now()
	start := end.Add(-duration)
	//fmt.Println("createTraceSpan:", t.ctx)
	_, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(start), trace.WithSpanKind(trace.SpanKindClient))
	span.SetAttributes(t.commonAttrs...)
	span.SetAttributes(attrs...)
	if error {
		span.SetStatus(codes.Error, "")
	}
	span.End(trace.WithTimestamp(end))
}

func (t *Trace) MysqlTraceQuery(query string, error bool, duration time.Duration, destination netaddr.IPPort) {
	if t == nil || query == "" {
		return
	}
	t.createTraceSpan(l7.ProtocolMysql.String(), duration, error,
		semconv.DBSystemMySQL,
		semconv.DBStatement(query),
		semconv.NetPeerName(destination.IP().String()),
		semconv.NetPeerPort(int(destination.Port())),
	)
}

func (t *Trace) MysqlTraceQueryEvent(query string, r *l7.RequestData, destination netaddr.IPPort) {
	if t == nil || query == "" {
		return
	}

	var attr []attribute.KeyValue
	attr = append(attr,
		semconv.DBSystemMySQL,
		semconv.DBStatement(query),
		semconv.NetPeerName(destination.IP().String()),
		semconv.NetPeerPort(int(destination.Port())),
	)
	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
	t.createTraceEvent(l7.ProtocolHTTP.String(), int(ebpftracer.EventTypeL7Request), int(l7.ProtocolMysql), attr...)
}

func (t *Trace) RedisTraceQuery(cmd, args string, error bool, duration time.Duration) {
	if t == nil || cmd == "" {
		return
	}
	statement := cmd
	if args != "" {
		statement += " " + args
	}
	t.createTraceSpan(l7.ProtocolRedis.String(), duration, error,
		semconv.DBSystemRedis,
		semconv.DBOperation(cmd),
		semconv.DBStatement(statement),
	)
}

func (t *Trace) RedisTraceQueryEvent(cmd, args string, r *l7.RequestData) {
	if t == nil || cmd == "" {
		return
	}
	statement := cmd
	if args != "" {
		statement += " " + args
	}

	var attr []attribute.KeyValue
	attr = append(attr,
		semconv.DBSystemRedis,
		semconv.DBOperation(cmd),
		semconv.DBStatement(statement),
	)
	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
	t.createTraceEvent(l7.ProtocolHTTP.String(), int(ebpftracer.EventTypeL7Request), int(l7.ProtocolRedis), attr...)
}

func (t *Trace) HttpTraceRequest(method, path, ip string, port uint16, r *l7.RequestData) {
	if t == nil || method == "" {
		return
	}
	assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
	if err != nil {
		assumedAppID = 0
	}
	status := r.Status
	duration := r.Duration
	t.createTraceSpan(l7.ProtocolHTTP.String(), duration, status >= 400,
		semconv.HTTPURL(fmt.Sprintf("http://%s%s", t.destination.String(), path)),
		semconv.HTTPMethod(method),
		semconv.HTTPStatusCode(int(status)),
		attribute.String("http.uri", path),
		attribute.String("http.ip", ip),
		attribute.Int64("http.assumed_app_id", assumedAppID),
		attribute.String("http.span_id", r.SpanId),
		attribute.Int("http.port", int(port)),
	)
}

// 新增事件处理
func (t *Trace) HttpTraceRequestEvent(method, path, ip string, port uint16, r *l7.RequestData) {
	if t == nil || method == "" {
		return
	}
	assumedAppID, err := strconv.ParseInt(r.AssumedAppId, 10, 64)
	if err != nil {
		assumedAppID = 0
	}
	status := r.Status
	var attr []attribute.KeyValue
	attr = append(attr,
		semconv.HTTPURL(fmt.Sprintf("http://%s%s", t.destination.String(), path)),
		semconv.HTTPMethod(method),
		semconv.HTTPStatusCode(int(status)),
		attribute.Bool("http.status_error", status > 399),
		attribute.String("http.uri", path),
		attribute.String("http.ip", ip),
		attribute.Int64("http.assumed_app_id", assumedAppID),
		attribute.String("http.span_id", r.SpanId),
		attribute.Int("http.port", int(port)),
	)
	t.appendTimestamp(&attr, r.StartAt, r.EndAt, r.Duration.Nanoseconds())
	t.createTraceEvent(l7.ProtocolHTTP.String(), int(ebpftracer.EventTypeL7Request), int(l7.ProtocolHTTP), attr...)
}

func (t *Trace) FuncTraceQuery(funcname string, duration time.Duration, start uint64, end uint64) {
	if t == nil || funcname == "" {
		return
	}
	t.createTraceSpanNoTime2(funcname, duration, false, start, end)
}

func (t *Trace) createTraceSpanNoTime(name string, duration time.Duration, error bool, start uint64, end uint64, attrs ...attribute.KeyValue) {
	// end := time.Now()
	// start := end.Add(-duration)
	startTime := time.Unix(0, int64(start))
	endTime := time.Unix(0, int64(end))
	//fmt.Println("createTraceSpan:", t.ctx)
	_, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(startTime), trace.WithSpanKind(trace.SpanKindClient))
	span.SetAttributes(t.commonAttrs...)
	span.SetAttributes(attrs...)
	if error {
		span.SetStatus(codes.Error, "")
	}
	span.End(trace.WithTimestamp(endTime))
}

func (t *Trace) createTraceSpanNoTime2(name string, duration time.Duration, error bool, start uint64, end uint64) {
	// end := time.Now()
	// start := end.Add(-duration)
	//startTime := time.Unix(0, int64(start))
	//endTime := time.Unix(0, int64(end))
	////fmt.Println("createTraceSpan:", t.ctx)
	//_, span := tracer(t.containerId).Start(t.ctx, name, trace.WithTimestamp(startTime), trace.WithSpanKind(trace.SpanKindClient))
	//span.SetAttributes(t.commonAttrs...)
	//span.SetAttributes(attrs...)
	//if error {
	//	span.SetStatus(codes.Error, "")
	//}
	//span.End(trace.WithTimestamp(endTime))
	//attrs = append([]attribute.KeyValue{
	//	attribute.Int64("startAt", int64(start)),
	//	attribute.Int64("endAt", int64(end)),
	//})
	var attr []attribute.KeyValue
	t.appendTimestamp(&attr, start, end, int64(end-start))
	t.createTraceEvent(name, int(ebpftracer.EventTypeFunEnt), 0, attr...)
}