Fix wrong kafka protocol detection #994
Conversation
| @@ -1,5 +1,5 @@ | |||
| # Build the autoinstrumenter binary | |||
| FROM golang:1.22 as builder | |||
| FROM golang:1.22 AS builder | |||
There was a problem hiding this comment.
New docker version complains about using different cases for FROM and AS.
| @@ -101,7 +101,7 @@ var MisclassifiedEvents = make(chan MisclassifiedEvent) | |||
|
|
|||
| func ptlog() *slog.Logger { return slog.With("component", "ebpf.ProcessTracer") } | |||
|
|
|||
| func ReadHTTPRequestTraceAsSpan(record *ringbuf.Record, filter ServiceFilter) (request.Span, bool, error) { | |||
| func ReadBPFTraceAsSpan(record *ringbuf.Record, filter ServiceFilter) (request.Span, bool, error) { | |||
There was a problem hiding this comment.
Renamed this since it's no longer just HTTP :).
| @@ -45,6 +45,7 @@ func (k Operation) String() string { | |||
| } | |||
|
|
|||
| const KafkaMinLength = 14 | |||
| const KafkaMaxPayload = 20 * 1024 * 1024 // 20 MB max, 1MB is default for most Kafka installations | |||
There was a problem hiding this comment.
Check for invalid packets that have the payload something ridiculous.
| @@ -294,14 +307,21 @@ func getTopicOffsetFromFetchOperation(header *Header) int { | |||
| if header.APIVersion >= 3 { | |||
| offset += 4 // max_bytes | |||
| if header.APIVersion >= 4 { | |||
| if origOffset+offset >= len(pkt) { | |||
There was a problem hiding this comment.
Validate that the isolation makes sense (0 or 1)
| @@ -38,7 +38,7 @@ func isRedisOp(buf []uint8) bool { | |||
| return isRedisError(buf[1:]) | |||
| case ':', '$', '*': | |||
| return crlfTerminatedMatch(buf[1:], func(c uint8) bool { | |||
| return (c >= '0' && c <= '9') | |||
| return (c >= '0' && c <= '9') || c == '-' | |||
There was a problem hiding this comment.
Result could be '-1', so we need to allow '-' in the list of characters.
| @@ -40,7 +40,7 @@ func ReadTCPRequestIntoSpan(record *ringbuf.Record, filter ServiceFilter) (reque | |||
| switch { | |||
| case validSQL(op, table): | |||
| return TCPToSQLToSpan(&event, op, table, sql), false, nil | |||
| case isRedis(b) && isRedis(event.Rbuf[:]): | |||
| case isRedis(b) && isRedis(event.Rbuf[:rl]): | |||
There was a problem hiding this comment.
This missing bounds was the main bug.
| @@ -40,7 +40,8 @@ func (k *Metadata) initServiceIPInformer(informerFactory informers.SharedInforme | |||
| return nil, fmt.Errorf("was expecting a Service. Got: %T", i) | |||
| } | |||
| if svc.Spec.ClusterIP == corev1.ClusterIPNone { | |||
| k.log.Warn("Service doesn't have any ClusterIP. Beyla won't decorate their flows", | |||
| // this will be normal for headless services | |||
There was a problem hiding this comment.
Changed this to debug because the message keeps appearing for headless services.
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #994 +/- ##
==========================================
+ Coverage 80.59% 80.79% +0.19%
==========================================
Files 134 134
Lines 10688 10734 +46
==========================================
+ Hits 8614 8672 +58
+ Misses 1565 1554 -11
+ Partials 509 508 -1
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. |
There were a few bugs here, but the major one was that we didn't limit the return buffer of the TCP event to the return buffer length, so there was a mix of redis and kafka buffer, which somehow still parsed successfully as Kafka.