github/kubeshark
1
0
Fork 0
mirror of https://github.com/kubeshark/kubeshark.git synced 2025-09-07 05:22:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add unit tests for Kafka dissector (#807)

Browse Source 
* Add unit tests for Kafka dissector

* Return `io.EOF` if request or response header size is zero

* Sort the slice in `representMapAsTable`

* Remove the dead code

* Remove more dead code

* Remove more dead code

* Fix `dissector.Analyze` call

Co-authored-by: gadotroee <55343099+gadotroee@users.noreply.github.com>
This commit is contained in:
M. Mert Yıldıran
2022-02-16 12:18:33 +03:00
committed by GitHub
parent e8d2b7eb3c
commit c67675c138
25 changed files with 357 additions and 2383 deletions
Download Patch File Download Diff File Expand all files Collapse all files

137
tap/extensions/kafka/protocol.go
View File

@@ -3,7 +3,6 @@ package kafka
import (
"fmt"
"io"
"net"
"reflect"
"strconv"
"strings"
@@ -27,29 +26,20 @@ func (k ApiKey) String() string {
return strconv.Itoa(int(k))
}
func (k ApiKey) MinVersion() int16 { return k.apiType().minVersion() }
func (k ApiKey) MaxVersion() int16 { return k.apiType().maxVersion() }
func (k ApiKey) SelectVersion(minVersion, maxVersion int16) int16 {
min := k.MinVersion()
max := k.MaxVersion()
switch {
case min > maxVersion:
return min
case max < maxVersion:
return max
default:
return maxVersion
}
}
func (k ApiKey) apiType() apiType {
if i := int(k); i >= 0 && i < len(apiTypes) {
return apiTypes[i]
}
return apiType{}
}
const (
// v0 = 0
v1 = 1
v2 = 2
v3 = 3
v4 = 4
v5 = 5
v6 = 6
v7 = 7
v8 = 8
v9 = 9
v10 = 10
v11 = 11
)
const (
Produce ApiKey = 0
@@ -164,48 +154,6 @@ type messageType struct {
flexible bool
gotype reflect.Type
decode decodeFunc
encode encodeFunc
}
type apiType struct {
requests []messageType
responses []messageType
}
func (t apiType) minVersion() int16 {
if len(t.requests) == 0 {
return 0
}
return t.requests[0].version
}
func (t apiType) maxVersion() int16 {
if len(t.requests) == 0 {
return 0
}
return t.requests[len(t.requests)-1].version
}
var apiTypes [numApis]apiType
// Register is automatically called by sub-packages are imported to install a
// new pair of request/response message types.
func Register(req, res Message) {
k1 := req.ApiKey()
k2 := res.ApiKey()
if k1 != k2 {
panic(fmt.Sprintf("[%T/%T]: request and response API keys mismatch: %d != %d", req, res, k1, k2))
}
apiTypes[k1] = apiType{
requests: typesOf(req),
responses: typesOf(res),
}
}
func typesOf(v interface{}) []messageType {
return makeTypes(reflect.TypeOf(v).Elem())
}
func makeTypes(t reflect.Type) []messageType {
@@ -241,7 +189,6 @@ func makeTypes(t reflect.Type) []messageType {
gotype: t,
flexible: flexible,
decode: decodeFuncOf(t, v, flexible, structTag{}),
encode: encodeFuncOf(t, v, flexible, structTag{}),
})
}
@@ -378,31 +325,6 @@ type Broker struct {
Rack string
}
func (b Broker) String() string {
return net.JoinHostPort(b.Host, itoa(b.Port))
}
func (b Broker) Format(w fmt.State, v rune) {
switch v {
case 'd':
_, _ = io.WriteString(w, itoa(b.ID))
case 's':
_, _ = io.WriteString(w, b.String())
case 'v':
_, _ = io.WriteString(w, itoa(b.ID))
_, _ = io.WriteString(w, " ")
_, _ = io.WriteString(w, b.String())
if b.Rack != "" {
_, _ = io.WriteString(w, " ")
_, _ = io.WriteString(w, b.Rack)
}
}
}
func itoa(i int32) string {
return strconv.Itoa(int(i))
}
type Topic struct {
Name string
Error int16
@@ -418,14 +340,6 @@ type Partition struct {
Offline []int32
}
// BrokerMessage is an extension of the Message interface implemented by some
// request types to customize the broker assignment logic.
type BrokerMessage interface {
// Given a representation of the kafka cluster state as argument, returns
// the broker that the message should be routed to.
Broker(Cluster) (Broker, error)
}
// GroupMessage is an extension of the Message interface implemented by some
// request types to inform the program that they should be routed to a group
// coordinator.
@@ -443,16 +357,6 @@ type PreparedMessage interface {
Prepare(apiVersion int16)
}
// Splitter is an interface implemented by messages that can be split into
// multiple requests and have their results merged back by a Merger.
type Splitter interface {
// For a given cluster layout, returns the list of messages constructed
// from the receiver for each requests that should be sent to the cluster.
// The second return value is a Merger which can be used to merge back the
// results of each request into a single message (or an error).
Split(Cluster) ([]Message, Merger, error)
}
// Merger is an interface implemented by messages which can merge multiple
// results into one response.
type Merger interface {
@@ -461,16 +365,3 @@ type Merger interface {
// values, other types should trigger a panic.
Merge(messages []Message, results []interface{}) (Message, error)
}
// Result converts r to a Message or and error, or panics if r could be be
// converted to these types.
func Result(r interface{}) (Message, error) {
switch v := r.(type) {
case Message:
return v, nil
case error:
return nil, v
default:
panic(fmt.Errorf("BUG: result must be a message or an error but not %T", v))
}
}