github/kubeshark
1
0
Fork 0
mirror of https://github.com/kubeshark/kubeshark.git synced 2025-06-21 22:08:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add an indicator for the eBPF sourced entries (#886)

Browse Source 
* Define `Capture` type and expect it as an argument in `Dissect` method

* Add an indicator for the  eBPF sourced entries

* Fix the Go lint error

* Fix the logic in the UI

* Update the expected JSONs

* Add TODOs

* Add `UndefinedCapture` constant

* Define `CaptureTypes` enum
This commit is contained in:
M. Mert Yıldıran 2022-03-16 23:32:09 -07:00 committed by GitHub
parent 237002ef29
commit 5455220a3a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
24 changed files with 103 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
tap/api/api.go
View File

@ -48,6 +48,16 @@ type Extension struct {
Dissector Dissector Dissector Dissector
} }
type Capture string
const (
UndefinedCapture Capture = ""
Pcap Capture = "pcap"
Envoy Capture = "envoy"
Linkerd Capture = "linkerd"
Ebpf Capture = "ebpf"
)
type ConnectionInfo struct { type ConnectionInfo struct {
ClientIP string ClientIP string
ClientPort string ClientPort string
@ -84,6 +94,7 @@ type RequestResponsePair struct {
// `Protocol` is modified in the later stages of data propagation. Therefore it's not a pointer. // `Protocol` is modified in the later stages of data propagation. Therefore it's not a pointer.
type OutputChannelItem struct { type OutputChannelItem struct {
Protocol Protocol Protocol Protocol
Capture Capture
Timestamp int64 Timestamp int64
ConnectionInfo *ConnectionInfo ConnectionInfo *ConnectionInfo
Pair *RequestResponsePair Pair *RequestResponsePair
@ -102,7 +113,7 @@ type SuperIdentifier struct {
type Dissector interface { type Dissector interface {
Register(*Extension) Register(*Extension)
Ping() Ping()
Dissect(b *bufio.Reader, isClient bool, tcpID *TcpID, counterPair *CounterPair, superTimer *SuperTimer, superIdentifier *SuperIdentifier, emitter Emitter, options *TrafficFilteringOptions, reqResMatcher RequestResponseMatcher) error Dissect(b *bufio.Reader, capture Capture, isClient bool, tcpID *TcpID, counterPair *CounterPair, superTimer *SuperTimer, superIdentifier *SuperIdentifier, emitter Emitter, options *TrafficFilteringOptions, reqResMatcher RequestResponseMatcher) error
Analyze(item *OutputChannelItem, resolvedSource string, resolvedDestination string, namespace string) *Entry Analyze(item *OutputChannelItem, resolvedSource string, resolvedDestination string, namespace string) *Entry
Summarize(entry *Entry) *BaseEntry Summarize(entry *Entry) *BaseEntry
Represent(request map[string]interface{}, response map[string]interface{}) (object []byte, bodySize int64, err error) Represent(request map[string]interface{}, response map[string]interface{}) (object []byte, bodySize int64, err error)
@ -132,6 +143,7 @@ func (e *Emitting) Emit(item *OutputChannelItem) {
type Entry struct { type Entry struct {
Id uint `json:"id"` Id uint `json:"id"`
Protocol Protocol `json:"proto"` Protocol Protocol `json:"proto"`
Capture Capture `json:"capture"`
Source *TCP `json:"src"` Source *TCP `json:"src"`
Destination *TCP `json:"dst"` Destination *TCP `json:"dst"`
Namespace string `json:"namespace,omitempty"` Namespace string `json:"namespace,omitempty"`
@ -162,6 +174,7 @@ type EntryWrapper struct {
type BaseEntry struct { type BaseEntry struct {
Id uint `json:"id"` Id uint `json:"id"`
Protocol Protocol `json:"proto,omitempty"` Protocol Protocol `json:"proto,omitempty"`
Capture Capture `json:"capture"`
Summary string `json:"summary,omitempty"` Summary string `json:"summary,omitempty"`
SummaryQuery string `json:"summaryQuery,omitempty"` SummaryQuery string `json:"summaryQuery,omitempty"`
Status int `json:"status"` Status int `json:"status"`

2
tap/extensions/amqp/Makefile
View File

@ -13,4 +13,4 @@ test-pull-bin:
test-pull-expect: test-pull-expect:
@mkdir -p expect @mkdir -p expect
@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5' -m cp -r gs://static.up9.io/mizu/test-pcap/expect3/amqp/\* expect @[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5' -m cp -r gs://static.up9.io/mizu/test-pcap/expect4/amqp/\* expect

3
tap/extensions/amqp/helpers.go
View File

@ -94,7 +94,7 @@ type AMQPWrapper struct {
Details interface{} `json:"details"` Details interface{} `json:"details"`
} }
func emitAMQP(event interface{}, _type string, method string, connectionInfo *api.ConnectionInfo, captureTime time.Time, emitter api.Emitter) { func emitAMQP(event interface{}, _type string, method string, connectionInfo *api.ConnectionInfo, captureTime time.Time, emitter api.Emitter, capture api.Capture) {
request := &api.GenericMessage{ request := &api.GenericMessage{
IsRequest: true, IsRequest: true,
CaptureTime: captureTime, CaptureTime: captureTime,
@ -108,6 +108,7 @@ func emitAMQP(event interface{}, _type string, method string, connectionInfo *ap
} }
item := &api.OutputChannelItem{ item := &api.OutputChannelItem{
Protocol: protocol, Protocol: protocol,
Capture: capture,
Timestamp: captureTime.UnixNano() / int64(time.Millisecond), Timestamp: captureTime.UnixNano() / int64(time.Millisecond),
ConnectionInfo: connectionInfo, ConnectionInfo: connectionInfo,
Pair: &api.RequestResponsePair{ Pair: &api.RequestResponsePair{

20
tap/extensions/amqp/main.go
View File

@ -39,7 +39,7 @@ func (d dissecting) Ping() {
const amqpRequest string = "amqp_request" const amqpRequest string = "amqp_request"
func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions, _reqResMatcher api.RequestResponseMatcher) error { func (d dissecting) Dissect(b *bufio.Reader, capture api.Capture, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions, _reqResMatcher api.RequestResponseMatcher) error {
r := AmqpReader{b} r := AmqpReader{b}
var remaining int var remaining int
@ -113,11 +113,11 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
case *BasicPublish: case *BasicPublish:
eventBasicPublish.Body = f.Body eventBasicPublish.Body = f.Body
superIdentifier.Protocol = &protocol superIdentifier.Protocol = &protocol
emitAMQP(*eventBasicPublish, amqpRequest, basicMethodMap[40], connectionInfo, superTimer.CaptureTime, emitter) emitAMQP(*eventBasicPublish, amqpRequest, basicMethodMap[40], connectionInfo, superTimer.CaptureTime, emitter, capture)
case *BasicDeliver: case *BasicDeliver:
eventBasicDeliver.Body = f.Body eventBasicDeliver.Body = f.Body
superIdentifier.Protocol = &protocol superIdentifier.Protocol = &protocol
emitAMQP(*eventBasicDeliver, amqpRequest, basicMethodMap[60], connectionInfo, superTimer.CaptureTime, emitter) emitAMQP(*eventBasicDeliver, amqpRequest, basicMethodMap[60], connectionInfo, superTimer.CaptureTime, emitter, capture)
} }
case *MethodFrame: case *MethodFrame:
@ -138,7 +138,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
Arguments: m.Arguments, Arguments: m.Arguments,
} }
superIdentifier.Protocol = &protocol superIdentifier.Protocol = &protocol
emitAMQP(*eventQueueBind, amqpRequest, queueMethodMap[20], connectionInfo, superTimer.CaptureTime, emitter) emitAMQP(*eventQueueBind, amqpRequest, queueMethodMap[20], connectionInfo, superTimer.CaptureTime, emitter, capture)
case *BasicConsume: case *BasicConsume:
eventBasicConsume := &BasicConsume{ eventBasicConsume := &BasicConsume{
@ -151,7 +151,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
Arguments: m.Arguments, Arguments: m.Arguments,
} }
superIdentifier.Protocol = &protocol superIdentifier.Protocol = &protocol
emitAMQP(*eventBasicConsume, amqpRequest, basicMethodMap[20], connectionInfo, superTimer.CaptureTime, emitter) emitAMQP(*eventBasicConsume, amqpRequest, basicMethodMap[20], connectionInfo, superTimer.CaptureTime, emitter, capture)
case *BasicDeliver: case *BasicDeliver:
eventBasicDeliver.ConsumerTag = m.ConsumerTag eventBasicDeliver.ConsumerTag = m.ConsumerTag
@ -171,7 +171,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
Arguments: m.Arguments, Arguments: m.Arguments,
} }
superIdentifier.Protocol = &protocol superIdentifier.Protocol = &protocol
emitAMQP(*eventQueueDeclare, amqpRequest, queueMethodMap[10], connectionInfo, superTimer.CaptureTime, emitter) emitAMQP(*eventQueueDeclare, amqpRequest, queueMethodMap[10], connectionInfo, superTimer.CaptureTime, emitter, capture)
case *ExchangeDeclare: case *ExchangeDeclare:
eventExchangeDeclare := &ExchangeDeclare{ eventExchangeDeclare := &ExchangeDeclare{
@ -185,7 +185,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
Arguments: m.Arguments, Arguments: m.Arguments,
} }
superIdentifier.Protocol = &protocol superIdentifier.Protocol = &protocol
emitAMQP(*eventExchangeDeclare, amqpRequest, exchangeMethodMap[10], connectionInfo, superTimer.CaptureTime, emitter) emitAMQP(*eventExchangeDeclare, amqpRequest, exchangeMethodMap[10], connectionInfo, superTimer.CaptureTime, emitter, capture)
case *ConnectionStart: case *ConnectionStart:
eventConnectionStart := &ConnectionStart{ eventConnectionStart := &ConnectionStart{
@ -196,7 +196,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
Locales: m.Locales, Locales: m.Locales,
} }
superIdentifier.Protocol = &protocol superIdentifier.Protocol = &protocol
emitAMQP(*eventConnectionStart, amqpRequest, connectionMethodMap[10], connectionInfo, superTimer.CaptureTime, emitter) emitAMQP(*eventConnectionStart, amqpRequest, connectionMethodMap[10], connectionInfo, superTimer.CaptureTime, emitter, capture)
case *ConnectionClose: case *ConnectionClose:
eventConnectionClose := &ConnectionClose{ eventConnectionClose := &ConnectionClose{
@ -206,7 +206,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
MethodId: m.MethodId, MethodId: m.MethodId,
} }
superIdentifier.Protocol = &protocol superIdentifier.Protocol = &protocol
emitAMQP(*eventConnectionClose, amqpRequest, connectionMethodMap[50], connectionInfo, superTimer.CaptureTime, emitter) emitAMQP(*eventConnectionClose, amqpRequest, connectionMethodMap[50], connectionInfo, superTimer.CaptureTime, emitter, capture)
} }
default: default:
@ -222,6 +222,7 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
reqDetails["method"] = request["method"] reqDetails["method"] = request["method"]
return &api.Entry{ return &api.Entry{
Protocol: protocol, Protocol: protocol,
Capture: item.Capture,
Source: &api.TCP{ Source: &api.TCP{
Name: resolvedSource, Name: resolvedSource,
IP: item.ConnectionInfo.ClientIP, IP: item.ConnectionInfo.ClientIP,
@ -283,6 +284,7 @@ func (d dissecting) Summarize(entry *api.Entry) *api.BaseEntry {
return &api.BaseEntry{ return &api.BaseEntry{
Id: entry.Id, Id: entry.Id,
Protocol: entry.Protocol, Protocol: entry.Protocol,
Capture: entry.Capture,
Summary: summary, Summary: summary,
SummaryQuery: summaryQuery, SummaryQuery: summaryQuery,
Status: 0, Status: 0,

4
tap/extensions/amqp/main_test.go
View File

@ -122,7 +122,7 @@ func TestDissect(t *testing.T) {
DstPort: "2", DstPort: "2",
} }
reqResMatcher := dissector.NewResponseRequestMatcher() reqResMatcher := dissector.NewResponseRequestMatcher()
err = dissector.Dissect(bufferClient, true, tcpIDClient, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher) err = dissector.Dissect(bufferClient, api.Pcap, true, tcpIDClient, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher)
if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
panic(err) panic(err)
} }
@ -140,7 +140,7 @@ func TestDissect(t *testing.T) {
SrcPort: "2", SrcPort: "2",
DstPort: "1", DstPort: "1",
} }
err = dissector.Dissect(bufferServer, false, tcpIDServer, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher) err = dissector.Dissect(bufferServer, api.Pcap, false, tcpIDServer, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher)
if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
panic(err) panic(err)
} }

2
tap/extensions/http/Makefile
View File

@ -13,4 +13,4 @@ test-pull-bin:
test-pull-expect: test-pull-expect:
@mkdir -p expect @mkdir -p expect
@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5' -m cp -r gs://static.up9.io/mizu/test-pcap/expect3/http/\* expect @[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5' -m cp -r gs://static.up9.io/mizu/test-pcap/expect4/http/\* expect

9
tap/extensions/http/handlers.go
View File

@ -47,7 +47,7 @@ func replaceForwardedFor(item *api.OutputChannelItem) {
item.ConnectionInfo.ClientPort = "" item.ConnectionInfo.ClientPort = ""
} }
func handleHTTP2Stream(http2Assembler *Http2Assembler, tcpID *api.TcpID, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions, reqResMatcher *requestResponseMatcher) error { func handleHTTP2Stream(http2Assembler *Http2Assembler, capture api.Capture, tcpID *api.TcpID, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions, reqResMatcher *requestResponseMatcher) error {
streamID, messageHTTP1, isGrpc, err := http2Assembler.readMessage() streamID, messageHTTP1, isGrpc, err := http2Assembler.readMessage()
if err != nil { if err != nil {
return err return err
@ -104,13 +104,14 @@ func handleHTTP2Stream(http2Assembler *Http2Assembler, tcpID *api.TcpID, superTi
} else { } else {
item.Protocol = http2Protocol item.Protocol = http2Protocol
} }
item.Capture = capture
filterAndEmit(item, emitter, options) filterAndEmit(item, emitter, options)
} }
return nil return nil
} }
func handleHTTP1ClientStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions, reqResMatcher *requestResponseMatcher) (switchingProtocolsHTTP2 bool, req *http.Request, err error) { func handleHTTP1ClientStream(b *bufio.Reader, capture api.Capture, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions, reqResMatcher *requestResponseMatcher) (switchingProtocolsHTTP2 bool, req *http.Request, err error) {
req, err = http.ReadRequest(b) req, err = http.ReadRequest(b)
if err != nil { if err != nil {
return return
@ -147,12 +148,13 @@ func handleHTTP1ClientStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api
ServerPort: tcpID.DstPort, ServerPort: tcpID.DstPort,
IsOutgoing: true, IsOutgoing: true,
} }
item.Capture = capture
filterAndEmit(item, emitter, options) filterAndEmit(item, emitter, options)
} }
return return
} }
func handleHTTP1ServerStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions, reqResMatcher *requestResponseMatcher) (switchingProtocolsHTTP2 bool, err error) { func handleHTTP1ServerStream(b *bufio.Reader, capture api.Capture, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, options *api.TrafficFilteringOptions, reqResMatcher *requestResponseMatcher) (switchingProtocolsHTTP2 bool, err error) {
var res *http.Response var res *http.Response
res, err = http.ReadResponse(b, nil) res, err = http.ReadResponse(b, nil)
if err != nil { if err != nil {
@ -190,6 +192,7 @@ func handleHTTP1ServerStream(b *bufio.Reader, tcpID *api.TcpID, counterPair *api
ServerPort: tcpID.SrcPort, ServerPort: tcpID.SrcPort,
IsOutgoing: false, IsOutgoing: false,
} }
item.Capture = capture
filterAndEmit(item, emitter, options) filterAndEmit(item, emitter, options)
} }
return return

11
tap/extensions/http/main.go
View File

@ -86,7 +86,7 @@ func (d dissecting) Ping() {
log.Printf("pong %s", http11protocol.Name) log.Printf("pong %s", http11protocol.Name)
} }
func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions, _reqResMatcher api.RequestResponseMatcher) error { func (d dissecting) Dissect(b *bufio.Reader, capture api.Capture, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions, _reqResMatcher api.RequestResponseMatcher) error {
reqResMatcher := _reqResMatcher.(*requestResponseMatcher) reqResMatcher := _reqResMatcher.(*requestResponseMatcher)
var err error var err error
@ -121,7 +121,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
} }
if isHTTP2 { if isHTTP2 {
err = handleHTTP2Stream(http2Assembler, tcpID, superTimer, emitter, options, reqResMatcher) err = handleHTTP2Stream(http2Assembler, capture, tcpID, superTimer, emitter, options, reqResMatcher)
if err == io.EOF || err == io.ErrUnexpectedEOF { if err == io.EOF || err == io.ErrUnexpectedEOF {
break break
} else if err != nil { } else if err != nil {
@ -130,7 +130,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
superIdentifier.Protocol = &http11protocol superIdentifier.Protocol = &http11protocol
} else if isClient { } else if isClient {
var req *http.Request var req *http.Request
switchingProtocolsHTTP2, req, err = handleHTTP1ClientStream(b, tcpID, counterPair, superTimer, emitter, options, reqResMatcher) switchingProtocolsHTTP2, req, err = handleHTTP1ClientStream(b, capture, tcpID, counterPair, superTimer, emitter, options, reqResMatcher)
if err == io.EOF || err == io.ErrUnexpectedEOF { if err == io.EOF || err == io.ErrUnexpectedEOF {
break break
} else if err != nil { } else if err != nil {
@ -157,11 +157,12 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
ServerPort: tcpID.DstPort, ServerPort: tcpID.DstPort,
IsOutgoing: true, IsOutgoing: true,
} }
item.Capture = capture
filterAndEmit(item, emitter, options) filterAndEmit(item, emitter, options)
} }
} }
} else { } else {
switchingProtocolsHTTP2, err = handleHTTP1ServerStream(b, tcpID, counterPair, superTimer, emitter, options, reqResMatcher) switchingProtocolsHTTP2, err = handleHTTP1ServerStream(b, capture, tcpID, counterPair, superTimer, emitter, options, reqResMatcher)
if err == io.EOF || err == io.ErrUnexpectedEOF { if err == io.EOF || err == io.ErrUnexpectedEOF {
break break
} else if err != nil { } else if err != nil {
@ -259,6 +260,7 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
httpPair, _ := json.Marshal(item.Pair) httpPair, _ := json.Marshal(item.Pair)
return &api.Entry{ return &api.Entry{
Protocol: item.Protocol, Protocol: item.Protocol,
Capture: item.Capture,
Source: &api.TCP{ Source: &api.TCP{
Name: resolvedSource, Name: resolvedSource,
IP: item.ConnectionInfo.ClientIP, IP: item.ConnectionInfo.ClientIP,
@ -291,6 +293,7 @@ func (d dissecting) Summarize(entry *api.Entry) *api.BaseEntry {
return &api.BaseEntry{ return &api.BaseEntry{
Id: entry.Id, Id: entry.Id,
Protocol: entry.Protocol, Protocol: entry.Protocol,
Capture: entry.Capture,
Summary: summary, Summary: summary,
SummaryQuery: summaryQuery, SummaryQuery: summaryQuery,
Status: status, Status: status,

4
tap/extensions/http/main_test.go
View File

@ -124,7 +124,7 @@ func TestDissect(t *testing.T) {
DstPort: "2", DstPort: "2",
} }
reqResMatcher := dissector.NewResponseRequestMatcher() reqResMatcher := dissector.NewResponseRequestMatcher()
err = dissector.Dissect(bufferClient, true, tcpIDClient, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher) err = dissector.Dissect(bufferClient, api.Pcap, true, tcpIDClient, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher)
if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
panic(err) panic(err)
} }
@ -142,7 +142,7 @@ func TestDissect(t *testing.T) {
SrcPort: "2", SrcPort: "2",
DstPort: "1", DstPort: "1",
} }
err = dissector.Dissect(bufferServer, false, tcpIDServer, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher) err = dissector.Dissect(bufferServer, api.Pcap, false, tcpIDServer, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher)
if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
panic(err) panic(err)
} }

2
tap/extensions/kafka/Makefile
View File

@ -13,4 +13,4 @@ test-pull-bin:
test-pull-expect: test-pull-expect:
@mkdir -p expect @mkdir -p expect
@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5' -m cp -r gs://static.up9.io/mizu/test-pcap/expect3/kafka/\* expect @[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5' -m cp -r gs://static.up9.io/mizu/test-pcap/expect4/kafka/\* expect

6
tap/extensions/kafka/main.go
View File

@ -35,7 +35,7 @@ func (d dissecting) Ping() {
log.Printf("pong %s", _protocol.Name) log.Printf("pong %s", _protocol.Name)
} }
func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions, _reqResMatcher api.RequestResponseMatcher) error { func (d dissecting) Dissect(b *bufio.Reader, capture api.Capture, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions, _reqResMatcher api.RequestResponseMatcher) error {
reqResMatcher := _reqResMatcher.(*requestResponseMatcher) reqResMatcher := _reqResMatcher.(*requestResponseMatcher)
for { for {
if superIdentifier.Protocol != nil && superIdentifier.Protocol != &_protocol { if superIdentifier.Protocol != nil && superIdentifier.Protocol != &_protocol {
@ -49,7 +49,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
} }
superIdentifier.Protocol = &_protocol superIdentifier.Protocol = &_protocol
} else { } else {
err := ReadResponse(b, tcpID, counterPair, superTimer, emitter, reqResMatcher) err := ReadResponse(b, capture, tcpID, counterPair, superTimer, emitter, reqResMatcher)
if err != nil { if err != nil {
return err return err
} }
@ -68,6 +68,7 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
} }
return &api.Entry{ return &api.Entry{
Protocol: _protocol, Protocol: _protocol,
Capture: item.Capture,
Source: &api.TCP{ Source: &api.TCP{
Name: resolvedSource, Name: resolvedSource,
IP: item.ConnectionInfo.ClientIP, IP: item.ConnectionInfo.ClientIP,
@ -190,6 +191,7 @@ func (d dissecting) Summarize(entry *api.Entry) *api.BaseEntry {
return &api.BaseEntry{ return &api.BaseEntry{
Id: entry.Id, Id: entry.Id,
Protocol: entry.Protocol, Protocol: entry.Protocol,
Capture: entry.Capture,
Summary: summary, Summary: summary,
SummaryQuery: summaryQuery, SummaryQuery: summaryQuery,
Status: status, Status: status,

4
tap/extensions/kafka/main_test.go
View File

@ -123,7 +123,7 @@ func TestDissect(t *testing.T) {
} }
reqResMatcher := dissector.NewResponseRequestMatcher() reqResMatcher := dissector.NewResponseRequestMatcher()
reqResMatcher.SetMaxTry(10) reqResMatcher.SetMaxTry(10)
err = dissector.Dissect(bufferClient, true, tcpIDClient, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher) err = dissector.Dissect(bufferClient, api.Pcap, true, tcpIDClient, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher)
if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
log.Println(err) log.Println(err)
} }
@ -141,7 +141,7 @@ func TestDissect(t *testing.T) {
SrcPort: "2", SrcPort: "2",
DstPort: "1", DstPort: "1",
} }
err = dissector.Dissect(bufferServer, false, tcpIDServer, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher) err = dissector.Dissect(bufferServer, api.Pcap, false, tcpIDServer, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher)
if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF { if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
log.Println(err) log.Println(err)
} }

3
tap/extensions/kafka/response.go
View File

@ -16,7 +16,7 @@ type Response struct {
CaptureTime time.Time `json:"captureTime"` CaptureTime time.Time `json:"captureTime"`
} }
func ReadResponse(r io.Reader, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, reqResMatcher *requestResponseMatcher) (err error) { func ReadResponse(r io.Reader, capture api.Capture, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, reqResMatcher *requestResponseMatcher) (err error) {
d := &decoder{reader: r, remain: 4} d := &decoder{reader: r, remain: 4}
size := d.readInt32() size := d.readInt32()
@ -258,6 +258,7 @@ func ReadResponse(r io.Reader, tcpID *api.TcpID, counterPair *api.CounterPair, s
item := &api.OutputChannelItem{ item := &api.OutputChannelItem{
Protocol: _protocol, Protocol: _protocol,
Capture: capture,
Timestamp: reqResPair.Request.CaptureTime.UnixNano() / int64(time.Millisecond), Timestamp: reqResPair.Request.CaptureTime.UnixNano() / int64(time.Millisecond),
ConnectionInfo: connectionInfo, ConnectionInfo: connectionInfo,
Pair: &api.RequestResponsePair{ Pair: &api.RequestResponsePair{

2
tap/extensions/redis/Makefile
View File

@ -13,4 +13,4 @@ test-pull-bin:
test-pull-expect: test-pull-expect:
@mkdir -p expect @mkdir -p expect
@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5' -m cp -r gs://static.up9.io/mizu/test-pcap/expect3/redis/\* expect @[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5' -m cp -r gs://static.up9.io/mizu/test-pcap/expect4/redis/\* expect

6
tap/extensions/redis/handlers.go
View File

@ -6,7 +6,7 @@ import (
"github.com/up9inc/mizu/tap/api" "github.com/up9inc/mizu/tap/api"
) )
func handleClientStream(tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, request *RedisPacket, reqResMatcher *requestResponseMatcher) error { func handleClientStream(capture api.Capture, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, request *RedisPacket, reqResMatcher *requestResponseMatcher) error {
counterPair.Lock() counterPair.Lock()
counterPair.Request++ counterPair.Request++
requestCounter := counterPair.Request requestCounter := counterPair.Request
@ -23,6 +23,7 @@ func handleClientStream(tcpID *api.TcpID, counterPair *api.CounterPair, superTim
item := reqResMatcher.registerRequest(ident, request, superTimer.CaptureTime) item := reqResMatcher.registerRequest(ident, request, superTimer.CaptureTime)
if item != nil { if item != nil {
item.Capture = capture
item.ConnectionInfo = &api.ConnectionInfo{ item.ConnectionInfo = &api.ConnectionInfo{
ClientIP: tcpID.SrcIP, ClientIP: tcpID.SrcIP,
ClientPort: tcpID.SrcPort, ClientPort: tcpID.SrcPort,
@ -35,7 +36,7 @@ func handleClientStream(tcpID *api.TcpID, counterPair *api.CounterPair, superTim
return nil return nil
} }
func handleServerStream(tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, response *RedisPacket, reqResMatcher *requestResponseMatcher) error { func handleServerStream(capture api.Capture, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, emitter api.Emitter, response *RedisPacket, reqResMatcher *requestResponseMatcher) error {
counterPair.Lock() counterPair.Lock()
counterPair.Response++ counterPair.Response++
responseCounter := counterPair.Response responseCounter := counterPair.Response
@ -52,6 +53,7 @@ func handleServerStream(tcpID *api.TcpID, counterPair *api.CounterPair, superTim
item := reqResMatcher.registerResponse(ident, response, superTimer.CaptureTime) item := reqResMatcher.registerResponse(ident, response, superTimer.CaptureTime)
if item != nil { if item != nil {
item.Capture = capture
item.ConnectionInfo = &api.ConnectionInfo{ item.ConnectionInfo = &api.ConnectionInfo{
ClientIP: tcpID.DstIP, ClientIP: tcpID.DstIP,
ClientPort: tcpID.DstPort, ClientPort: tcpID.DstPort,

8
tap/extensions/redis/main.go
View File

@ -34,7 +34,7 @@ func (d dissecting) Ping() {
log.Printf("pong %s", protocol.Name) log.Printf("pong %s", protocol.Name)
} }
func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions, _reqResMatcher api.RequestResponseMatcher) error { func (d dissecting) Dissect(b *bufio.Reader, capture api.Capture, isClient bool, tcpID *api.TcpID, counterPair *api.CounterPair, superTimer *api.SuperTimer, superIdentifier *api.SuperIdentifier, emitter api.Emitter, options *api.TrafficFilteringOptions, _reqResMatcher api.RequestResponseMatcher) error {
reqResMatcher := _reqResMatcher.(*requestResponseMatcher) reqResMatcher := _reqResMatcher.(*requestResponseMatcher)
is := &RedisInputStream{ is := &RedisInputStream{
Reader: b, Reader: b,
@ -48,9 +48,9 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
} }
if isClient { if isClient {
err = handleClientStream(tcpID, counterPair, superTimer, emitter, redisPacket, reqResMatcher) err = handleClientStream(capture, tcpID, counterPair, superTimer, emitter, redisPacket, reqResMatcher)
} else { } else {
err = handleServerStream(tcpID, counterPair, superTimer, emitter, redisPacket, reqResMatcher) err = handleServerStream(capture, tcpID, counterPair, superTimer, emitter, redisPacket, reqResMatcher)
} }
if err != nil { if err != nil {
@ -71,6 +71,7 @@ func (d dissecting) Analyze(item *api.OutputChannelItem, resolvedSource string,
} }
return &api.Entry{ return &api.Entry{
Protocol: protocol, Protocol: protocol,
Capture: item.Capture,
Source: &api.TCP{ Source: &api.TCP{
Name: resolvedSource, Name: resolvedSource,
IP: item.ConnectionInfo.ClientIP, IP: item.ConnectionInfo.ClientIP,
@ -113,6 +114,7 @@ func (d dissecting) Summarize(entry *api.Entry) *api.BaseEntry {
return &api.BaseEntry{ return &api.BaseEntry{
Id: entry.Id, Id: entry.Id,
Protocol: entry.Protocol, Protocol: entry.Protocol,
Capture: entry.Capture,
Summary: summary, Summary: summary,
SummaryQuery: summaryQuery, SummaryQuery: summaryQuery,
Status: status, Status: status,

4
tap/extensions/redis/main_test.go
View File

@ -123,7 +123,7 @@ func TestDissect(t *testing.T) {
DstPort: "2", DstPort: "2",
} }
reqResMatcher := dissector.NewResponseRequestMatcher() reqResMatcher := dissector.NewResponseRequestMatcher()
err = dissector.Dissect(bufferClient, true, tcpIDClient, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher) err = dissector.Dissect(bufferClient, api.Pcap, true, tcpIDClient, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher)
if err != nil && reflect.TypeOf(err) != reflect.TypeOf(&ConnectError{}) && err != io.EOF && err != io.ErrUnexpectedEOF { if err != nil && reflect.TypeOf(err) != reflect.TypeOf(&ConnectError{}) && err != io.EOF && err != io.ErrUnexpectedEOF {
log.Println(err) log.Println(err)
} }
@ -141,7 +141,7 @@ func TestDissect(t *testing.T) {
SrcPort: "2", SrcPort: "2",
DstPort: "1", DstPort: "1",
} }
err = dissector.Dissect(bufferServer, false, tcpIDServer, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher) err = dissector.Dissect(bufferServer, api.Pcap, false, tcpIDServer, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher)
if err != nil && reflect.TypeOf(err) != reflect.TypeOf(&ConnectError{}) && err != io.EOF && err != io.ErrUnexpectedEOF { if err != nil && reflect.TypeOf(err) != reflect.TypeOf(&ConnectError{}) && err != io.EOF && err != io.ErrUnexpectedEOF {
log.Println(err) log.Println(err)
} }

3
tap/tcp_reader.go
View File

@ -95,7 +95,8 @@ func (h *tcpReader) Close() {
func (h *tcpReader) run(wg *sync.WaitGroup) { func (h *tcpReader) run(wg *sync.WaitGroup) {
defer wg.Done() defer wg.Done()
b := bufio.NewReader(h) b := bufio.NewReader(h)
err := h.extension.Dissector.Dissect(b, h.isClient, h.tcpID, h.counterPair, h.superTimer, h.parent.superIdentifier, h.emitter, filteringOptions, h.reqResMatcher) // TODO: Add api.Pcap, api.Envoy and api.Linkerd distinction by refactoring NewPacketSourceManager method
err := h.extension.Dissector.Dissect(b, api.Pcap, h.isClient, h.tcpID, h.counterPair, h.superTimer, h.parent.superIdentifier, h.emitter, filteringOptions, h.reqResMatcher)
if err != nil { if err != nil {
_, err = io.Copy(ioutil.Discard, b) _, err = io.Copy(ioutil.Discard, b)
if err != nil { if err != nil {

2
tap/tlstapper/tls_poller.go
View File

@ -158,7 +158,7 @@ func dissect(extension *api.Extension, reader *tlsReader, isRequest bool, tcpid
emitter api.Emitter, options *api.TrafficFilteringOptions, reqResMatcher api.RequestResponseMatcher) { emitter api.Emitter, options *api.TrafficFilteringOptions, reqResMatcher api.RequestResponseMatcher) {
b := bufio.NewReader(reader) b := bufio.NewReader(reader)
err := extension.Dissector.Dissect(b, isRequest, tcpid, &api.CounterPair{}, err := extension.Dissector.Dissect(b, api.Ebpf, isRequest, tcpid, &api.CounterPair{},
&api.SuperTimer{}, &api.SuperIdentifier{}, emitter, options, reqResMatcher) &api.SuperTimer{}, &api.SuperIdentifier{}, emitter, options, reqResMatcher)
if err != nil { if err != nil {

7
ui-common/src/components/TrafficViewer/EntryListItem/EntryListItem.module.sass
View File

@ -62,6 +62,13 @@
width: 185px width: 185px
text-align: left text-align: left
.capture
margin-top: -60px
.capture img
height: 20px
z-index: 1000
.endpointServiceContainer .endpointServiceContainer
display: flex display: flex
flex-direction: column flex-direction: column

21
ui-common/src/components/TrafficViewer/EntryListItem/EntryListItem.tsx
View File

@ -4,6 +4,7 @@ import SwapHorizIcon from '@material-ui/icons/SwapHoriz';
import styles from './EntryListItem.module.sass'; import styles from './EntryListItem.module.sass';
import StatusCode, {getClassification, StatusCodeClassification} from "../../UI/StatusCode"; import StatusCode, {getClassification, StatusCodeClassification} from "../../UI/StatusCode";
import Protocol, {ProtocolInterface} from "../../UI/Protocol" import Protocol, {ProtocolInterface} from "../../UI/Protocol"
import eBPFLogo from '../assets/ebpf.png';
import {Summary} from "../../UI/Summary"; import {Summary} from "../../UI/Summary";
import Queryable from "../../UI/Queryable"; import Queryable from "../../UI/Queryable";
import ingoingIconSuccess from "assets/ingoing-traffic-success.svg" import ingoingIconSuccess from "assets/ingoing-traffic-success.svg"
@ -24,6 +25,7 @@ interface TCPInterface {
interface Entry { interface Entry {
proto: ProtocolInterface, proto: ProtocolInterface,
capture: string,
method?: string, method?: string,
methodQuery?: string, methodQuery?: string,
summary: string, summary: string,
@ -52,6 +54,14 @@ interface EntryProps {
headingMode: boolean; headingMode: boolean;
} }
enum CaptureTypes {
UndefinedCapture = "",
Pcap = "pcap",
Envoy = "envoy",
Linkerd = "linkerd",
Ebpf = "ebpf",
}
export const EntryItem: React.FC<EntryProps> = ({entry, style, headingMode}) => { export const EntryItem: React.FC<EntryProps> = ({entry, style, headingMode}) => {
const [focusedEntryId, setFocusedEntryId] = useRecoilState(focusedEntryIdAtom); const [focusedEntryId, setFocusedEntryId] = useRecoilState(focusedEntryIdAtom);
@ -154,6 +164,17 @@ export const EntryItem: React.FC<EntryProps> = ({entry, style, headingMode}) =>
protocol={entry.proto} protocol={entry.proto}
horizontal={false} horizontal={false}
/> : null} /> : null}
{/* TODO: Update the code below once we have api.Pcap, api.Envoy and api.Linkerd distinction in the backend */}
{entry.capture === CaptureTypes.Ebpf ? <div className={styles.capture}>
<Queryable
query={`capture == "${entry.capture}"`}
displayIconOnMouseOver={true}
flipped={false}
style={{position: "absolute"}}
>
<img src={eBPFLogo} alt="eBPF"/>
</Queryable>
</div> : null}
{isStatusCodeEnabled && <div> {isStatusCodeEnabled && <div>
<StatusCode statusCode={entry.status} statusQuery={entry.statusQuery}/> <StatusCode statusCode={entry.status} statusQuery={entry.statusQuery}/>
</div>} </div>}

1
ui-common/src/components/UI/Protocol.tsx
View File

@ -46,6 +46,7 @@ const Protocol: React.FC<ProtocolProps> = ({protocol, horizontal}) => {
displayIconOnMouseOver={true} displayIconOnMouseOver={true}
flipped={false} flipped={false}
iconStyle={{marginTop: "52px", marginRight: "10px", zIndex: 1000}} iconStyle={{marginTop: "52px", marginRight: "10px", zIndex: 1000}}
tooltipStyle={{marginTop: "-22px", zIndex: 1001}}
> >
<span <span
className={`${styles.base} ${styles.vertical}`} className={`${styles.base} ${styles.vertical}`}

5
ui-common/src/components/UI/Queryable.tsx
View File

@ -11,11 +11,12 @@ interface Props {
iconStyle?: object, iconStyle?: object,
className?: string, className?: string,
useTooltip?: boolean, useTooltip?: boolean,
tooltipStyle?: object,
displayIconOnMouseOver?: boolean, displayIconOnMouseOver?: boolean,
flipped?: boolean, flipped?: boolean,
} }
const Queryable: React.FC<Props> = ({query, style, iconStyle, className, useTooltip= true, displayIconOnMouseOver = false, flipped = false, children}) => { const Queryable: React.FC<Props> = ({query, style, iconStyle, className, useTooltip = true, tooltipStyle = null, displayIconOnMouseOver = false, flipped = false, children}) => {
const [showAddedNotification, setAdded] = useState(false); const [showAddedNotification, setAdded] = useState(false);
const [showTooltip, setShowTooltip] = useState(false); const [showTooltip, setShowTooltip] = useState(false);
const [queryState, setQuery] = useRecoilState(queryAtom); const [queryState, setQuery] = useRecoilState(queryAtom);
@ -53,7 +54,7 @@ const Queryable: React.FC<Props> = ({query, style, iconStyle, className, useTool
{flipped && addButton} {flipped && addButton}
{children} {children}
{!flipped && addButton} {!flipped && addButton}
{useTooltip && showTooltip && <span className={QueryableStyle.QueryableTooltip}>{query}</span>} {useTooltip && showTooltip && <span className={QueryableStyle.QueryableTooltip} style={tooltipStyle}>{query}</span>}
</div> </div>
); );
}; };

BIN
ui/src/components/assets/ebpf.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 21 KiB