ebpf/tracing/apm_tracing.go

package tracing

import (
	"context"
	"fmt"
	"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"
	"time"
)

/**
 * 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) 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)))
	t.span.End(trace.WithTimestamp(time.Now()))
}

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) 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) 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) HttpTraceRequest(method, path, ip string, port uint16, status l7.Status, duration time.Duration) {
	if t == nil || method == "" {
		return
	}
	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.Int("http.port", int(port)),
	)
}