From a847d8fc9b1ebb4af5ae9f17f419d647115c1c2b Mon Sep 17 00:00:00 2001 From: Mayank Pande Date: Wed, 3 Apr 2024 19:19:11 +0530 Subject: [PATCH] feat: added addition span attributes --- containers/container.go | 31 ++++++++++++++++++++----------- containers/l7.go | 7 +------ ebpftracer/l7/http.go | 21 ++++++++++++++++----- tracing/tracing.go | 18 ++++++++++++++---- 4 files changed, 51 insertions(+), 26 deletions(-) diff --git a/containers/container.go b/containers/container.go index a4ae3ba5..9139fd44 100644 --- a/containers/container.go +++ b/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 { diff --git a/containers/l7.go b/containers/l7.go index 6153616c..f4782eea 100644 --- a/containers/l7.go +++ b/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] diff --git a/ebpftracer/l7/http.go b/ebpftracer/l7/http.go index eed35dcf..1ffa1156 100644 --- a/ebpftracer/l7/http.go +++ b/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 { diff --git a/tracing/tracing.go b/tracing/tracing.go index 386e113b..17fd132b 100644 --- a/tracing/tracing.go +++ b/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), ) }