feat: added addition span attributes

containers/container.go

@@ -24,8 +24,9 @@ import (
 )
 
 var (
-	gcInterval  = 10 * time.Minute
-	pingTimeout = 300 * time.Millisecond
+	gcInterval       = 10 * time.Minute
+	pingTimeout      = 300 * time.Millisecond
+	payloadThreshold = 1024 * 2
 )
 
 type ContainerID string
@@ -75,7 +76,7 @@ type ActiveConnection struct {
 	ActualDest         netaddr.IPPort
 	srcWorkload        common.Workload
 	dstWorkload        common.Workload
-	actialDestWorkload common.Workload
+	actualDestWorkload common.Workload
 	Pid                uint32
 	Fd                 uint64
 	Timestamp          uint64
@@ -536,9 +537,9 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst netaddr.IPP
 	} else {
 		srcWorkload := c.ip_resolver.ResolveIP(src.IP().String())
 		dstWorkload := c.ip_resolver.ResolveIP(dst.IP().String())
-		actialDestWorkload := c.ip_resolver.ResolveActualIP(actualDst.IP().String())
+		actualDestWorkload := c.ip_resolver.ResolveActualIP(actualDst.IP().String())
 		c.connectsSuccessful[AddrPair{src: dst, dst: *actualDst, srcWorkload: srcWorkload,
-			dstWorkload: dstWorkload, actualDestWorkload: actialDestWorkload}]++
+			dstWorkload: dstWorkload, actualDestWorkload: actualDestWorkload}]++
 		connection := &ActiveConnection{
 			Dest:               dst,
 			ActualDest:         *actualDst,
@@ -547,9 +548,9 @@ func (c *Container) onConnectionOpen(pid uint32, fd uint64, src, dst netaddr.IPP
 			Timestamp:          timestamp,
 			srcWorkload:        srcWorkload,
 			dstWorkload:        dstWorkload,
-			actialDestWorkload: actialDestWorkload,
+			actualDestWorkload: actualDestWorkload,
 		}
-		c.connectionsActive[AddrPair{src: src, dst: dst, srcWorkload: srcWorkload, dstWorkload: dstWorkload, actualDestWorkload: actialDestWorkload}] = connection
+		c.connectionsActive[AddrPair{src: src, dst: dst, srcWorkload: srcWorkload, dstWorkload: dstWorkload, actualDestWorkload: actualDestWorkload}] = connection
 		c.connectionsByPidFd[PidFd{Pid: pid, Fd: fd}] = connection
 	}
 	c.connectLastAttempt[dst] = time.Now()
@@ -608,12 +609,16 @@ func (c *Container) onL7Request(pid uint32, fd uint64, timestamp uint64, r *l7.R
 		return
 	}
 
-	stats := c.l7Stats.get(r.Protocol, conn.Dest, conn.ActualDest, r, conn.srcWorkload, conn.dstWorkload, conn.actialDestWorkload)
-	trace := tracing.NewTrace(string(c.id), conn.ActualDest)
+	stats := c.l7Stats.get(r.Protocol, conn.Dest, conn.ActualDest, r, conn.srcWorkload, conn.dstWorkload, conn.actualDestWorkload)
+	trace := tracing.NewTrace(string(c.id), conn.ActualDest, conn.srcWorkload, conn.dstWorkload, conn.actualDestWorkload)
 	switch r.Protocol {
 	case l7.ProtocolHTTP:
 		stats.observe(r.Status.Http(), "", r.Duration)
-		method, path, payload := l7.ParseHttpAndRest(r.Payload)
+		method, path := l7.ParseHttp(r.Payload)
+		payload := ""
+		if int(r.PayloadSize) < 2048 {
+			payload = string(r.Payload)
+		}
 		trace.HttpRequest(method, path, r.Status, r.Duration, r.PayloadSize, payload)
 	case l7.ProtocolHTTP2:
 		if conn.http2Parser == nil {
@@ -622,7 +627,11 @@ func (c *Container) onL7Request(pid uint32, fd uint64, timestamp uint64, r *l7.R
 		requests := conn.http2Parser.Parse(r.Method, r.Payload, uint64(r.Duration))
 		for _, req := range requests {
 			stats.observe(req.Status.Http(), "", req.Duration)
-			trace.Http2Request(req.Method, req.Path, req.Scheme, req.Status, req.Duration)
+			payload := ""
+			if int(r.PayloadSize) < payloadThreshold {
+				payload = string(r.Payload)
+			}
+			trace.Http2Request(req.Method, req.Path, req.Scheme, req.Status, req.Duration, string(payload))
 		}
 	case l7.ProtocolPostgres:
 		if r.Method != l7.MethodStatementClose {

containers/l7.go

@@ -1,7 +1,6 @@
 package containers
 
 import (
-	"log"
 	"time"
 
 	"github.com/coroot/coroot-node-agent/common"
@@ -69,11 +68,7 @@ func (s L7Stats) get(protocol l7.Protocol, destination, actualDestination netadd
 		case l7.ProtocolRabbitmq, l7.ProtocolNats:
 			labels = append(labels, "method")
 		case l7.ProtocolHTTP:
-			method, path, payload := l7.ParseHttpAndRest(r.Payload)
-			if r.Status.Http() == "400" || r.Status.Http() == "500" {
-				log.Printf("Captured failed request actual body %s, converted body %s, status %s", string(r.Payload), payload, r.Status.Http())
-				constLabels["payload"] = payload
-			}
+			method, path := l7.ParseHttp(r.Payload)
 			constLabels["path"] = path
 			constLabels["method"] = method
 			hOpts := L7Latency[protocol]

ebpftracer/l7/http.go

@@ -23,19 +23,30 @@ func ParseHttp(payload []byte) (string, string) {
 	return string(method), string(uri)
 }
 
-func ParseHttpAndRest(payload []byte) (string, string, string) {
-	method, rest, ok := bytes.Cut(payload, space)
+func ParseHttpAndRest(payload []byte) (string, string, string, string) {
+	data := string(payload)
+	data = strings.ReplaceAll(data, "\\n\\n", "\n\n")
+	split := strings.Split(data, "\n\n")
+	rest := []byte(split[0])
+	d := split[1]
+	method, rest, ok := bytes.Cut(rest, space)
 	if !ok {
-		return "", "", ""
+		return "", "", "", ""
 	}
 	if !isHttpMethod(string(method)) {
-		return "", "", ""
+		return "", "", "", ""
 	}
 	uri, rest, ok := bytes.Cut(rest, space)
 	if !ok {
 		uri = append(uri, []byte("...")...)
 	}
-	return string(method), string(uri), SanitizeString(string(rest))
+
+	_, headers, ok := bytes.Cut(rest, []byte{'\n'})
+	if !ok {
+		uri = append(uri, []byte("...")...)
+	}
+	log.Printf("data %s", headers)
+	return string(method), string(uri), string(headers), string(d)
 }
 
 func SanitizeString(input string) string {

tracing/tracing.go

@@ -76,13 +76,22 @@ type Trace struct {
 	commonAttrs []attribute.KeyValue
 }
 
-func NewTrace(containerId string, destination netaddr.IPPort) *Trace {
+func NewTrace(containerId string, destination netaddr.IPPort, srcWorkload common.Workload, dstWorkload common.Workload, actualDstWorkload common.Workload) *Trace {
 	if tracer == nil {
 		return nil
 	}
 	return &Trace{containerId: containerId, destination: destination, commonAttrs: []attribute.KeyValue{
 		semconv.NetPeerName(destination.IP().String()),
 		semconv.NetPeerPort(int(destination.Port())),
+		attribute.Key("destination.workload_name").String(dstWorkload.Name),
+		attribute.Key("destination.workload_namepace").String(dstWorkload.Namespace),
+		attribute.Key("destination.workload_kind").String(dstWorkload.Kind),
+		attribute.Key("source.workload_name").String(srcWorkload.Name),
+		attribute.Key("source.workload_namepace").String(srcWorkload.Namespace),
+		attribute.Key("source.workload_kind").String(srcWorkload.Kind),
+		attribute.Key("destination.name").String(actualDstWorkload.Name),
+		attribute.Key("destination.namepace").String(actualDstWorkload.Namespace),
+		attribute.Key("destination.kind").String(actualDstWorkload.Kind),
 	}}
 }
 
@@ -98,7 +107,7 @@ func (t *Trace) createSpan(name string, duration time.Duration, error bool, attr
 	span.End(trace.WithTimestamp(end))
 }
 
-func (t *Trace) HttpRequest(method, path string, status l7.Status, duration time.Duration, requestSize uint64, requestBody string) {
+func (t *Trace) HttpRequest(method, path string, status l7.Status, duration time.Duration, requestSize uint64, payload string) {
 	if t == nil || method == "" {
 		return
 	}
@@ -107,11 +116,11 @@ func (t *Trace) HttpRequest(method, path string, status l7.Status, duration time
 		semconv.HTTPMethod(method),
 		semconv.HTTPStatusCode(int(status)),
 		semconv.HTTPRequestContentLength(int(requestSize)),
-		attribute.Key("http.request_body").String(requestBody),
+		attribute.Key("http.request_payload").String(payload),
 	)
 }
 
-func (t *Trace) Http2Request(method, path, scheme string, status l7.Status, duration time.Duration) {
+func (t *Trace) Http2Request(method, path, scheme string, status l7.Status, duration time.Duration, payload string) {
 	if t == nil {
 		return
 	}
@@ -128,6 +137,7 @@ func (t *Trace) Http2Request(method, path, scheme string, status l7.Status, dura
 		semconv.HTTPURL(fmt.Sprintf("%s://%s%s", scheme, t.destination.String(), path)),
 		semconv.HTTPMethod(method),
 		semconv.HTTPStatusCode(int(status)),
+		attribute.Key("http.request_payload").String(payload),
 	)
 }