NETOBSERV-926: Added key to kafka message for connection tracking consistency #107
Conversation
|
@OlivierCazade: This pull request references NETOBSERV-926 which is a valid jira issue. In response to this: Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. |
OlivierCazade
force-pushed
the
kafka-connection-tracking
branch
2 times, most recently
from
bfc7ccf to
915de15
Compare
Codecov Report
@@ Coverage Diff @@
## main #107 +/- ##
==========================================
+ Coverage 41.59% 41.80% +0.20%
==========================================
Files 30 30
Lines 2041 2050 +9
==========================================
+ Hits 849 857 +8
- Misses 1154 1155 +1
Partials 38 38
Flags with carried forward coverage won't be shown. Click here to find out more.
Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here. |
pkg/exporter/kafka_proto.go
Outdated
| @@ -28,6 +29,26 @@ func (kp *KafkaProto) ExportFlows(input <-chan []*flow.Record) { | |||
| } | |||
| } | |||
|
|
|||
| func getIPKey(ip [16]uint8) string { | |||
There was a problem hiding this comment.
nit: should we move this util to record.go where we keep other tools for conversion ?
also I didn't take a close look at net package to see if there is another ways to doing this but I assume u checked ? probably it can be called IPAddrToStriing() or something like that too ?
There was a problem hiding this comment.
I realized that we are transforming from byte array to string and then back to byte array.
I reworked the PR to directly work with byte arrays.
OlivierCazade
force-pushed
the
kafka-connection-tracking
branch
from
915de15 to
eb3f171
Compare
|
@OlivierCazade: This pull request references NETOBSERV-926 which is a valid jira issue. In response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. |
Done, thanks. |
| for k := range record.Id.SrcIp { | ||
| if record.Id.SrcIp[k] < record.Id.DstIp[k] { | ||
| return append(record.Id.SrcIp[:], record.Id.DstIp[:]...) | ||
| } else if record.Id.SrcIp[k] > record.Id.DstIp[k] { |
There was a problem hiding this comment.
maybe just check for > condition only since == and < both appends the same way so we can avoid if else ?
There was a problem hiding this comment.
No, if both bytes are equal we first want to iterate another time. But we return as soon as there is a difference.
There was a problem hiding this comment.
ok Thanks for the explaining it
| @@ -28,6 +28,17 @@ func (kp *KafkaProto) ExportFlows(input <-chan []*flow.Record) { | |||
| } | |||
| } | |||
|
|
|||
| func getFlowKey(record *flow.Record) []byte { | |||
| for k := range record.Id.SrcIp { | |||
There was a problem hiding this comment.
this assumes number of bytes in src and dest are equal .. what if there's an ipv4 and an ipv6 ?
There was a problem hiding this comment.
maybe check len first to return the smallest first (or last) if they aren't equal size?
Also, can you add a comment about why we need to sort this? I fear if we look at the code later we might hav forgotten :-)
There was a problem hiding this comment.
SrcIp and DstIp are not slice but fixed length array:
And more general question, is it possible to have a src ipv4 and a dst ipv6? I would have said no but you made me wonder
There was a problem hiding this comment.
that is probably impossible to have a pkt header "flow" with src and dst from different address family.
/lgtm
|
/lgtm |
|
/approve |
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: OlivierCazade The full list of commands accepted by this bot can be found here. The pull request process is described here
Needs approval from an approver in each of these files:
Approvers can indicate their approval by writing |
…sistency (netobserv#107) * Added key to kafka message for connection tracking consistency * Added comment about why we are sorting IP address for generating kafka key
Add keys to each kafka message. Kafka then ensure that all messages with the same key will be consumed by the same client.
We use src and destination IP as key so all messages from a connection will be consumed by the same transformer so connection tracking metrics will be accurate.