stop tapping self tapper traffic (#1083)

* stop tapping self tapper traffic

* run go mod tidy

* allow to explicitly ignore ports

* remove unused code

* remove shared from tap + go mod tidy

* move ignroe ports to tapper

* rename TapperPacketsCount to IgnoredPacketsCount

* don't check null - go is smart

* remove nil check

acceptanceTests/cypress/integration/tests/UiTest.js

@@ -193,58 +193,56 @@ function checkFilter(filterDetails) {
     const entriesForDeeperCheck = 5;
 
     it(`checking the filter: ${filter}`, function () {
-        waitForFetch50AndPause();
+        cy.get('.w-tc-editor-text').clear();
+        // applying the filter with alt+enter or with the button
+        cy.get('.w-tc-editor-text').type(`${filter}${applyByCtrlEnter ? '{ctrl+enter}' : ''}`);
+        cy.get('.w-tc-editor').should('have.attr', 'style').and('include', Cypress.env('greenFilterColor'));
+        if (!applyByCtrlEnter)
+            cy.get('[type="submit"]').click();
 
-        cy.get('#total-entries').should('not.have.text', '0').then(number => {
-            const totalEntries = number.text();
+        waitForFetch();
+        pauseStream();
 
-            cy.get(`#list [id^=entry]`).last().then(elem => {
-                const element = elem[0];
-                const entryId = getEntryId(element.id);
-                // checks the hover on the last entry (the only one in DOM at the beginning)
-                leftOnHoverCheck(entryId, leftSidePath, filter);
+        cy.get(`#list [id^=entry]`).last().then(elem => {
+            const element = elem[0];
+            const entryId = getEntryId(element.id);
 
-                cy.get('.w-tc-editor-text').clear();
-                // applying the filter with alt+enter or with the button
-                cy.get('.w-tc-editor-text').type(`${filter}${applyByCtrlEnter ? '{ctrl+enter}' : ''}`);
-                cy.get('.w-tc-editor').should('have.attr', 'style').and('include', Cypress.env('greenFilterColor'));
-                if (!applyByCtrlEnter)
-                    cy.get('[type="submit"]').click();
+            // only one entry in DOM after filtering, checking all checks on it
+            leftTextCheck(entryId, leftSidePath, leftSideExpectedText);
+            leftOnHoverCheck(entryId, leftSidePath, filter);
 
-                waitForFetch50AndPause();
+            rightTextCheck(rightSidePath, rightSideExpectedText);
+            rightOnHoverCheck(rightSidePath, filter);
+            checkRightSideResponseBody();
+        });
 
-                // only one entry in DOM after filtering, checking all checks on it
-                leftTextCheck(entryId, leftSidePath, leftSideExpectedText);
-                leftOnHoverCheck(entryId, leftSidePath, filter);
+        resizeToHugeMizu();
 
-                rightTextCheck(rightSidePath, rightSideExpectedText);
-                rightOnHoverCheck(rightSidePath, filter);
-                checkRightSideResponseBody();
-            });
+        // checking only 'leftTextCheck' on all entries because the rest of the checks require more time
+        cy.get(`#list [id^=entry]`).each(elem => {
+            const element = elem[0];
+            let entryId = getEntryId(element.id);
+            leftTextCheck(entryId, leftSidePath, leftSideExpectedText);
+        });
 
-            resizeToHugeMizu();
+        // making the other 3 checks on the first X entries (longer time for each check)
+        deeperCheck(leftSidePath, rightSidePath, filter, rightSideExpectedText, entriesForDeeperCheck);
 
-            // checking only 'leftTextCheck' on all entries because the rest of the checks require more time
-            cy.get(`#list [id^=entry]`).each(elem => {
-                const element = elem[0];
-                let entryId = getEntryId(element.id);
-                leftTextCheck(entryId, leftSidePath, leftSideExpectedText);
-            });
-
-            // making the other 3 checks on the first X entries (longer time for each check)
-            deeperCheck(leftSidePath, rightSidePath, filter, rightSideExpectedText, entriesForDeeperCheck);
-
-            // reloading then waiting for the entries number to load
-            resizeToNormalMizu();
-            cy.reload();
-            cy.get('#total-entries', {timeout: refreshWaitTimeout}).should('have.text', totalEntries);
-        })
+        // reloading then waiting for the entries number to load
+        resizeToNormalMizu();
+        cy.reload();
+        waitForFetch();
+        pauseStream();
     });
 }
 
-function waitForFetch50AndPause() {
-    // wait half a second and pause the stream to preserve the DOM
-    cy.wait(500);
+function waitForFetch() {
+    cy.get('#entries-length', {timeout: refreshWaitTimeout}).should((el) => {
+        expect(parseInt(el.text().trim(), 10)).to.be.greaterThan(20);
+    });
+}
+
+function pauseStream() {
     cy.get('#pause-icon').click();
     cy.get('#pause-icon').should('not.be.visible');
 }

agent/main.go

@@ -152,7 +152,9 @@ func runInTapperMode() {
 	}
 
 	hostMode := os.Getenv(shared.HostModeEnvVar) == "1"
-	tapOpts := &tap.TapOpts{HostMode: hostMode}
+	tapOpts := &tap.TapOpts{
+		HostMode:     hostMode,
+	}
 
 	filteredOutputItemsChannel := make(chan *tapApi.OutputChannelItem)
 

tap/api/api.go

@@ -104,11 +104,6 @@ type OutputChannelItem struct {
 	Namespace      string
 }
 
-type ProtoIdentifier struct {
-	Protocol       *Protocol
-	IsClosedOthers bool
-}
-
 type ReadProgress struct {
 	readBytes   int
 	lastCurrent int
@@ -123,6 +118,11 @@ func (p *ReadProgress) Current() (n int) {
 	return p.lastCurrent
 }
 
+func (p *ReadProgress) Reset() {
+	p.readBytes = 0
+	p.lastCurrent = 0
+}
+
 type Dissector interface {
 	Register(*Extension)
 	Ping()
@@ -419,13 +419,12 @@ type TcpReader interface {
 	GetCaptureTime() time.Time
 	GetEmitter() Emitter
 	GetIsClosed() bool
-	GetExtension() *Extension
 }
 
 type TcpStream interface {
 	SetProtocol(protocol *Protocol)
 	GetOrigin() Capture
-	GetProtoIdentifier() *ProtoIdentifier
+	GetProtocol() *Protocol
 	GetReqResMatchers() []RequestResponseMatcher
 	GetIsTapTarget() bool
 	GetIsClosed() bool

tap/api/stats_tracker.go

@@ -10,6 +10,7 @@ type AppStats struct {
 	ProcessedBytes              uint64    `json:"processedBytes"`
 	PacketsCount                uint64    `json:"packetsCount"`
 	TcpPacketsCount             uint64    `json:"tcpPacketsCount"`
+	IgnoredPacketsCount         uint64    `json:"ignoredPacketsCount"`
 	ReassembledTcpPayloadsCount uint64    `json:"reassembledTcpPayloadsCount"`
 	TlsConnectionsCount         uint64    `json:"tlsConnectionsCount"`
 	MatchedPairs                uint64    `json:"matchedPairs"`
@@ -33,6 +34,10 @@ func (as *AppStats) IncTcpPacketsCount() {
 	atomic.AddUint64(&as.TcpPacketsCount, 1)
 }
 
+func (as *AppStats) IncIgnoredPacketsCount() {
+	atomic.AddUint64(&as.IgnoredPacketsCount, 1)
+}
+
 func (as *AppStats) IncReassembledTcpPayloadsCount() {
 	atomic.AddUint64(&as.ReassembledTcpPayloadsCount, 1)
 }
@@ -55,6 +60,7 @@ func (as *AppStats) DumpStats() *AppStats {
 	currentAppStats.ProcessedBytes = resetUint64(&as.ProcessedBytes)
 	currentAppStats.PacketsCount = resetUint64(&as.PacketsCount)
 	currentAppStats.TcpPacketsCount = resetUint64(&as.TcpPacketsCount)
+	currentAppStats.IgnoredPacketsCount = resetUint64(&as.IgnoredPacketsCount)
 	currentAppStats.ReassembledTcpPayloadsCount = resetUint64(&as.ReassembledTcpPayloadsCount)
 	currentAppStats.TlsConnectionsCount = resetUint64(&as.TlsConnectionsCount)
 	currentAppStats.MatchedPairs = resetUint64(&as.MatchedPairs)

tap/extensions/amqp/main.go

@@ -75,14 +75,14 @@ func (d dissecting) Dissect(b *bufio.Reader, reader api.TcpReader, options *api.
 	var lastMethodFrameMessage Message
 
 	for {
-		if reader.GetParent().GetProtoIdentifier().Protocol != nil && reader.GetParent().GetProtoIdentifier().Protocol != &protocol {
+		if reader.GetParent().GetProtocol() != nil && reader.GetParent().GetProtocol() != &protocol {
 			return errors.New("Identified by another protocol")
 		}
 
 		frame, err := r.ReadFrame()
 		if err == io.EOF {
 			// We must read until we see an EOF... very important!
-			return nil
+			return err
 		}
 
 		switch f := frame.(type) {

tap/extensions/amqp/tcp_reader_mock_test.go

@@ -78,7 +78,3 @@ func (reader *tcpReader) GetEmitter() api.Emitter {
 func (reader *tcpReader) GetIsClosed() bool {
 	return reader.isClosed
 }
-
-func (reader *tcpReader) GetExtension() *api.Extension {
-	return reader.extension
-}

tap/extensions/amqp/tcp_stream_mock_test.go

@@ -7,18 +7,17 @@ import (
 )
 
 type tcpStream struct {
-	isClosed        bool
-	protoIdentifier *api.ProtoIdentifier
-	isTapTarget     bool
-	origin          api.Capture
-	reqResMatchers  []api.RequestResponseMatcher
+	isClosed       bool
+	protocol       *api.Protocol
+	isTapTarget    bool
+	origin         api.Capture
+	reqResMatchers []api.RequestResponseMatcher
 	sync.Mutex
 }
 
 func NewTcpStream(capture api.Capture) api.TcpStream {
 	return &tcpStream{
-		origin:          capture,
-		protoIdentifier: &api.ProtoIdentifier{},
+		origin: capture,
 	}
 }
 
@@ -28,8 +27,8 @@ func (t *tcpStream) GetOrigin() api.Capture {
 	return t.origin
 }
 
-func (t *tcpStream) GetProtoIdentifier() *api.ProtoIdentifier {
-	return t.protoIdentifier
+func (t *tcpStream) GetProtocol() *api.Protocol {
+	return t.protocol
 }
 
 func (t *tcpStream) GetReqResMatchers() []api.RequestResponseMatcher {

tap/extensions/http/main.go

@@ -144,7 +144,7 @@ func (d dissecting) Dissect(b *bufio.Reader, reader api.TcpReader, options *api.
 			http2Assembler = createHTTP2Assembler(b)
 		}
 
-		if reader.GetParent().GetProtoIdentifier().Protocol != nil && reader.GetParent().GetProtoIdentifier().Protocol != &http11protocol {
+		if reader.GetParent().GetProtocol() != nil && reader.GetParent().GetProtocol() != &http11protocol {
 			return errors.New("Identified by another protocol")
 		}
 
@@ -200,7 +200,7 @@ func (d dissecting) Dissect(b *bufio.Reader, reader api.TcpReader, options *api.
 		}
 	}
 
-	if reader.GetParent().GetProtoIdentifier().Protocol == nil {
+	if reader.GetParent().GetProtocol() == nil {
 		return err
 	}
 

tap/extensions/http/tcp_reader_mock_test.go

@@ -78,7 +78,3 @@ func (reader *tcpReader) GetEmitter() api.Emitter {
 func (reader *tcpReader) GetIsClosed() bool {
 	return reader.isClosed
 }
-
-func (reader *tcpReader) GetExtension() *api.Extension {
-	return reader.extension
-}

tap/extensions/http/tcp_stream_mock_test.go

@@ -7,18 +7,17 @@ import (
 )
 
 type tcpStream struct {
-	isClosed        bool
-	protoIdentifier *api.ProtoIdentifier
-	isTapTarget     bool
-	origin          api.Capture
-	reqResMatchers  []api.RequestResponseMatcher
+	isClosed       bool
+	protocol       *api.Protocol
+	isTapTarget    bool
+	origin         api.Capture
+	reqResMatchers []api.RequestResponseMatcher
 	sync.Mutex
 }
 
 func NewTcpStream(capture api.Capture) api.TcpStream {
 	return &tcpStream{
-		origin:          capture,
-		protoIdentifier: &api.ProtoIdentifier{},
+		origin: capture,
 	}
 }
 
@@ -28,8 +27,8 @@ func (t *tcpStream) GetOrigin() api.Capture {
 	return t.origin
 }
 
-func (t *tcpStream) GetProtoIdentifier() *api.ProtoIdentifier {
-	return t.protoIdentifier
+func (t *tcpStream) GetProtocol() *api.Protocol {
+	return t.protocol
 }
 
 func (t *tcpStream) GetReqResMatchers() []api.RequestResponseMatcher {

tap/extensions/kafka/main.go

@@ -38,7 +38,7 @@ func (d dissecting) Ping() {
 func (d dissecting) Dissect(b *bufio.Reader, reader api.TcpReader, options *api.TrafficFilteringOptions) error {
 	reqResMatcher := reader.GetReqResMatcher().(*requestResponseMatcher)
 	for {
-		if reader.GetParent().GetProtoIdentifier().Protocol != nil && reader.GetParent().GetProtoIdentifier().Protocol != &_protocol {
+		if reader.GetParent().GetProtocol() != nil && reader.GetParent().GetProtocol() != &_protocol {
 			return errors.New("Identified by another protocol")
 		}
 

tap/extensions/kafka/tcp_reader_mock_test.go

@@ -78,7 +78,3 @@ func (reader *tcpReader) GetEmitter() api.Emitter {
 func (reader *tcpReader) GetIsClosed() bool {
 	return reader.isClosed
 }
-
-func (reader *tcpReader) GetExtension() *api.Extension {
-	return reader.extension
-}

tap/extensions/kafka/tcp_stream_mock_test.go

@@ -7,18 +7,17 @@ import (
 )
 
 type tcpStream struct {
-	isClosed        bool
-	protoIdentifier *api.ProtoIdentifier
-	isTapTarget     bool
-	origin          api.Capture
-	reqResMatchers  []api.RequestResponseMatcher
+	isClosed       bool
+	protocol       *api.Protocol
+	isTapTarget    bool
+	origin         api.Capture
+	reqResMatchers []api.RequestResponseMatcher
 	sync.Mutex
 }
 
 func NewTcpStream(capture api.Capture) api.TcpStream {
 	return &tcpStream{
-		origin:          capture,
-		protoIdentifier: &api.ProtoIdentifier{},
+		origin: capture,
 	}
 }
 
@@ -28,8 +27,8 @@ func (t *tcpStream) GetOrigin() api.Capture {
 	return t.origin
 }
 
-func (t *tcpStream) GetProtoIdentifier() *api.ProtoIdentifier {
-	return t.protoIdentifier
+func (t *tcpStream) GetProtocol() *api.Protocol {
+	return t.protocol
 }
 
 func (t *tcpStream) GetReqResMatchers() []api.RequestResponseMatcher {

tap/extensions/redis/tcp_reader_mock_test.go

@@ -78,7 +78,3 @@ func (reader *tcpReader) GetEmitter() api.Emitter {
 func (reader *tcpReader) GetIsClosed() bool {
 	return reader.isClosed
 }
-
-func (reader *tcpReader) GetExtension() *api.Extension {
-	return reader.extension
-}

tap/extensions/redis/tcp_stream_mock_test.go

@@ -7,18 +7,17 @@ import (
 )
 
 type tcpStream struct {
-	isClosed        bool
-	protoIdentifier *api.ProtoIdentifier
-	isTapTarget     bool
-	origin          api.Capture
-	reqResMatchers  []api.RequestResponseMatcher
+	isClosed       bool
+	protocol       *api.Protocol
+	isTapTarget    bool
+	origin         api.Capture
+	reqResMatchers []api.RequestResponseMatcher
 	sync.Mutex
 }
 
 func NewTcpStream(capture api.Capture) api.TcpStream {
 	return &tcpStream{
-		origin:          capture,
-		protoIdentifier: &api.ProtoIdentifier{},
+		origin: capture,
 	}
 }
 
@@ -28,8 +27,8 @@ func (t *tcpStream) GetOrigin() api.Capture {
 	return t.origin
 }
 
-func (t *tcpStream) GetProtoIdentifier() *api.ProtoIdentifier {
-	return t.protoIdentifier
+func (t *tcpStream) GetProtocol() *api.Protocol {
+	return t.protocol
 }
 
 func (t *tcpStream) GetReqResMatchers() []api.RequestResponseMatcher {

tap/go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/cilium/ebpf v0.8.0
 	github.com/go-errors/errors v1.4.2
 	github.com/google/gopacket v1.1.19
+	github.com/hashicorp/golang-lru v0.5.4
 	github.com/up9inc/mizu/logger v0.0.0
 	github.com/up9inc/mizu/tap/api v0.0.0
 	github.com/vishvananda/netns v0.0.0-20211101163701-50045581ed74
@@ -18,7 +19,6 @@ require (
 	github.com/google/go-cmp v0.5.7 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/martian v2.1.0+incompatible // indirect
-	github.com/hashicorp/golang-lru v0.5.4 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
@@ -33,7 +33,6 @@ require (
 	k8s.io/utils v0.0.0-20220127004650-9b3446523e65 // indirect
 	sigs.k8s.io/json v0.0.0-20211208200746-9f7c6b3444d2 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.1 // indirect
-	sigs.k8s.io/yaml v1.3.0 // indirect
 )
 
 replace github.com/up9inc/mizu/logger v0.0.0 => ../logger

tap/go.sum

@@ -276,6 +276,5 @@ sigs.k8s.io/json v0.0.0-20211208200746-9f7c6b3444d2/go.mod h1:B+TnT182UBxE84DiCz
 sigs.k8s.io/structured-merge-diff/v4 v4.0.2/go.mod h1:bJZC9H9iH24zzfZ/41RGcq60oK1F7G282QMXDPYydCw=
 sigs.k8s.io/structured-merge-diff/v4 v4.2.1 h1:bKCqE9GvQ5tiVHn5rfn1r+yao3aLQEaLzkkmAkf+A6Y=
 sigs.k8s.io/structured-merge-diff/v4 v4.2.1/go.mod h1:j/nl6xW8vLS49O8YvXW1ocPhZawJtm+Yrr7PPRQ0Vg4=
+sigs.k8s.io/yaml v1.2.0 h1:kr/MCeFWJWTwyaHoR9c8EjH9OumOmoF9YGiZd7lFm/Q=
 sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
-sigs.k8s.io/yaml v1.3.0 h1:a2VclLzOGrwOHDiV8EfBGhvjHvP46CtW5j6POvhYGGo=
-sigs.k8s.io/yaml v1.3.0/go.mod h1:GeOyir5tyXNByN85N/dRIT9es5UQNerPYEKK56eTBm8=

tap/passive_tapper.go

@@ -16,6 +16,7 @@ import (
 	"runtime"
 	"strings"
 	"time"
+	"strconv"
 
 	"github.com/up9inc/mizu/logger"
 	"github.com/up9inc/mizu/tap/api"
@@ -41,6 +42,7 @@ var debug = flag.Bool("debug", false, "Display debug information")
 var quiet = flag.Bool("quiet", false, "Be quiet regarding errors")
 var hexdumppkt = flag.Bool("dumppkt", false, "Dump packet as hex")
 var procfs = flag.String("procfs", "/proc", "The procfs directory, used when mapping host volumes into a container")
+var ignoredPorts = flag.String("ignore-ports", "", "A comma separated list of ports to ignore")
 
 // capture
 var iface = flag.String("i", "en0", "Interface to read packets from")
@@ -55,7 +57,8 @@ var tls = flag.Bool("tls", false, "Enable TLS tapper")
 var memprofile = flag.String("memprofile", "", "Write memory profile")
 
 type TapOpts struct {
-	HostMode bool
+	HostMode         bool
+	IgnoredPorts     []uint16
 }
 
 var extensions []*api.Extension                     // global
@@ -193,6 +196,8 @@ func initializePassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelI
 		logger.Log.Fatal(err)
 	}
 
+	opts.IgnoredPorts = append(opts.IgnoredPorts, buildIgnoredPortsList(*ignoredPorts)...)
+
 	assembler := NewTcpAssembler(outputItems, streamsMap, opts)
 
 	return assembler
@@ -267,3 +272,19 @@ func startTlsTapper(extension *api.Extension, outputItems chan *api.OutputChanne
 
 	return &tls
 }
+
+func buildIgnoredPortsList(ignoredPorts string) []uint16 {
+	tmp := strings.Split(ignoredPorts, ",")
+	result := make([]uint16, len(tmp))
+
+	for i, raw := range tmp {
+		v, err := strconv.Atoi(raw)
+		if err != nil {
+			continue
+		}
+
+		result[i] = uint16(v)
+	}
+
+	return result
+}

tap/tcp_assembler.go

@@ -23,6 +23,7 @@ type tcpAssembler struct {
 	streamPool     *reassembly.StreamPool
 	streamFactory  *tcpStreamFactory
 	assemblerMutex sync.Mutex
+	ignoredPorts   []uint16
 }
 
 // Context
@@ -48,8 +49,8 @@ func NewTcpAssembler(outputItems chan *api.OutputChannelItem, streamsMap api.Tcp
 
 	maxBufferedPagesTotal := GetMaxBufferedPagesPerConnection()
 	maxBufferedPagesPerConnection := GetMaxBufferedPagesTotal()
-	logger.Log.Infof("Assembler options: maxBufferedPagesTotal=%d, maxBufferedPagesPerConnection=%d",
-		maxBufferedPagesTotal, maxBufferedPagesPerConnection)
+	logger.Log.Infof("Assembler options: maxBufferedPagesTotal=%d, maxBufferedPagesPerConnection=%d, opts=%v",
+		maxBufferedPagesTotal, maxBufferedPagesPerConnection, opts)
 	assembler.AssemblerOptions.MaxBufferedPagesTotal = maxBufferedPagesTotal
 	assembler.AssemblerOptions.MaxBufferedPagesPerConnection = maxBufferedPagesPerConnection
 
@@ -57,6 +58,7 @@ func NewTcpAssembler(outputItems chan *api.OutputChannelItem, streamsMap api.Tcp
 		Assembler:     assembler,
 		streamPool:    streamPool,
 		streamFactory: streamFactory,
+		ignoredPorts:  opts.IgnoredPorts,
 	}
 }
 
@@ -83,14 +85,18 @@ func (a *tcpAssembler) processPackets(dumpPacket bool, packets <-chan source.Tcp
 			diagnose.AppStats.IncTcpPacketsCount()
 			tcp := tcp.(*layers.TCP)
 
-			c := context{
-				CaptureInfo: packet.Metadata().CaptureInfo,
-				Origin:      packetInfo.Source.Origin,
+			if a.shouldIgnorePort(uint16(tcp.DstPort)) {
+				diagnose.AppStats.IncIgnoredPacketsCount()
+			} else {
+				c := context{
+					CaptureInfo: packet.Metadata().CaptureInfo,
+					Origin:      packetInfo.Source.Origin,
+				}
+				diagnose.InternalStats.Totalsz += len(tcp.Payload)
+				a.assemblerMutex.Lock()
+				a.AssembleWithContext(packet.NetworkLayer().NetworkFlow(), tcp, &c)
+				a.assemblerMutex.Unlock()
 			}
-			diagnose.InternalStats.Totalsz += len(tcp.Payload)
-			a.assemblerMutex.Lock()
-			a.AssembleWithContext(packet.NetworkLayer().NetworkFlow(), tcp, &c)
-			a.assemblerMutex.Unlock()
 		}
 
 		done := *maxcount > 0 && int64(diagnose.AppStats.PacketsCount) >= *maxcount
@@ -132,3 +138,13 @@ func (a *tcpAssembler) waitAndDump() {
 	logger.Log.Debugf("%s", a.Dump())
 	a.assemblerMutex.Unlock()
 }
+
+func (a *tcpAssembler) shouldIgnorePort(port uint16) bool {
+	for _, p := range a.ignoredPorts {
+		if port == p {
+			return true
+		}
+	}
+
+	return false
+}

tap/tcp_reader.go

@@ -3,11 +3,9 @@ package tap
 import (
 	"bufio"
 	"io"
-	"io/ioutil"
 	"sync"
 	"time"
 
-	"github.com/up9inc/mizu/logger"
 	"github.com/up9inc/mizu/tap/api"
 )
 
@@ -17,50 +15,48 @@ import (
  * Implements io.Reader interface (Read)
  */
 type tcpReader struct {
-	ident         string
-	tcpID         *api.TcpID
-	isClosed      bool
-	isClient      bool
-	isOutgoing    bool
-	msgQueue      chan api.TcpReaderDataMsg // Channel of captured reassembled tcp payload
-	data          []byte
-	progress      *api.ReadProgress
-	captureTime   time.Time
-	parent        *tcpStream
-	packetsSeen   uint
-	extension     *api.Extension
-	emitter       api.Emitter
-	counterPair   *api.CounterPair
-	reqResMatcher api.RequestResponseMatcher
+	ident           string
+	tcpID           *api.TcpID
+	isClosed        bool
+	isClient        bool
+	isOutgoing      bool
+	msgQueue        chan api.TcpReaderDataMsg // Channel of captured reassembled tcp payload
+	msgBuffer       []api.TcpReaderDataMsg
+	msgBufferMaster []api.TcpReaderDataMsg
+	data            []byte
+	progress        *api.ReadProgress
+	captureTime     time.Time
+	parent          *tcpStream
+	emitter         api.Emitter
+	counterPair     *api.CounterPair
+	reqResMatcher   api.RequestResponseMatcher
 	sync.Mutex
 }
 
-func NewTcpReader(msgQueue chan api.TcpReaderDataMsg, progress *api.ReadProgress, ident string, tcpId *api.TcpID, captureTime time.Time, parent *tcpStream, isClient bool, isOutgoing bool, extension *api.Extension, emitter api.Emitter, counterPair *api.CounterPair, reqResMatcher api.RequestResponseMatcher) *tcpReader {
+func NewTcpReader(ident string, tcpId *api.TcpID, parent *tcpStream, isClient bool, isOutgoing bool, emitter api.Emitter) *tcpReader {
 	return &tcpReader{
-		msgQueue:      msgQueue,
-		progress:      progress,
-		ident:         ident,
-		tcpID:         tcpId,
-		captureTime:   captureTime,
-		parent:        parent,
-		isClient:      isClient,
-		isOutgoing:    isOutgoing,
-		extension:     extension,
-		emitter:       emitter,
-		counterPair:   counterPair,
-		reqResMatcher: reqResMatcher,
+		msgQueue:   make(chan api.TcpReaderDataMsg),
+		progress:   &api.ReadProgress{},
+		ident:      ident,
+		tcpID:      tcpId,
+		parent:     parent,
+		isClient:   isClient,
+		isOutgoing: isOutgoing,
+		emitter:    emitter,
 	}
 }
 
 func (reader *tcpReader) run(options *api.TrafficFilteringOptions, wg *sync.WaitGroup) {
 	defer wg.Done()
-	b := bufio.NewReader(reader)
-	err := reader.extension.Dissector.Dissect(b, reader, options)
-	if err != nil {
-		_, err = io.Copy(ioutil.Discard, reader)
-		if err != nil {
-			logger.Log.Errorf("%v", err)
+	for i, extension := range extensions {
+		reader.reqResMatcher = reader.parent.reqResMatchers[i]
+		reader.counterPair = reader.parent.counterPairs[i]
+		b := bufio.NewReader(reader)
+		extension.Dissector.Dissect(b, reader, options) //nolint
+		if reader.isProtocolIdentified() {
+			break
 		}
+		reader.rewind()
 	}
 }
 
@@ -81,21 +77,60 @@ func (reader *tcpReader) sendMsgIfNotClosed(msg api.TcpReaderDataMsg) {
 	reader.Unlock()
 }
 
+func (reader *tcpReader) isProtocolIdentified() bool {
+	return reader.parent.protocol != nil
+}
+
+func (reader *tcpReader) rewind() {
+	// Reset the data
+	reader.data = make([]byte, 0)
+
+	// Reset msgBuffer from the master record
+	reader.parent.Lock()
+	reader.msgBuffer = make([]api.TcpReaderDataMsg, len(reader.msgBufferMaster))
+	copy(reader.msgBuffer, reader.msgBufferMaster)
+	reader.parent.Unlock()
+
+	// Reset the read progress
+	reader.progress.Reset()
+}
+
+func (reader *tcpReader) populateData(msg api.TcpReaderDataMsg) {
+	reader.data = msg.GetBytes()
+	reader.captureTime = msg.GetTimestamp()
+}
+
 func (reader *tcpReader) Read(p []byte) (int, error) {
 	var msg api.TcpReaderDataMsg
 
+	for len(reader.msgBuffer) > 0 && len(reader.data) == 0 {
+		// Pop first message
+		if len(reader.msgBuffer) > 1 {
+			msg, reader.msgBuffer = reader.msgBuffer[0], reader.msgBuffer[1:]
+		} else {
+			msg = reader.msgBuffer[0]
+			reader.msgBuffer = make([]api.TcpReaderDataMsg, 0)
+		}
+
+		// Get the bytes
+		reader.populateData(msg)
+	}
+
 	ok := true
 	for ok && len(reader.data) == 0 {
 		msg, ok = <-reader.msgQueue
 		if msg != nil {
-			reader.data = msg.GetBytes()
-			reader.captureTime = msg.GetTimestamp()
-		}
+			reader.populateData(msg)
 
-		if len(reader.data) > 0 {
-			reader.packetsSeen += 1
+			if !reader.isProtocolIdentified() {
+				reader.msgBufferMaster = append(
+					reader.msgBufferMaster,
+					msg,
+				)
+			}
 		}
 	}
+
 	if !ok || len(reader.data) == 0 {
 		return 0, io.EOF
 	}
@@ -142,7 +177,3 @@ func (reader *tcpReader) GetEmitter() api.Emitter {
 func (reader *tcpReader) GetIsClosed() bool {
 	return reader.isClosed
 }
-
-func (reader *tcpReader) GetExtension() *api.Extension {
-	return reader.extension
-}

tap/tcp_reassembly_stream.go

@@ -6,7 +6,6 @@ import (
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers" // pulls in all layers decoders
 	"github.com/google/gopacket/reassembly"
-	"github.com/up9inc/mizu/tap/api"
 	"github.com/up9inc/mizu/tap/diagnose"
 )
 
@@ -16,10 +15,10 @@ type tcpReassemblyStream struct {
 	fsmerr     bool
 	optchecker reassembly.TCPOptionCheck
 	isDNS      bool
-	tcpStream  api.TcpStream
+	tcpStream  *tcpStream
 }
 
-func NewTcpReassemblyStream(ident string, tcp *layers.TCP, fsmOptions reassembly.TCPSimpleFSMOptions, stream api.TcpStream) reassembly.Stream {
+func NewTcpReassemblyStream(ident string, tcp *layers.TCP, fsmOptions reassembly.TCPSimpleFSMOptions, stream *tcpStream) reassembly.Stream {
 	return &tcpReassemblyStream{
 		ident:      ident,
 		tcpState:   reassembly.NewTCPSimpleFSM(fsmOptions),
@@ -139,17 +138,10 @@ func (t *tcpReassemblyStream) ReassembledSG(sg reassembly.ScatterGather, ac reas
 			// This channel is read by an tcpReader object
 			diagnose.AppStats.IncReassembledTcpPayloadsCount()
 			timestamp := ac.GetCaptureInfo().Timestamp
-			stream := t.tcpStream.(*tcpStream)
 			if dir == reassembly.TCPDirClientToServer {
-				for i := range stream.getClients() {
-					reader := stream.getClient(i)
-					reader.sendMsgIfNotClosed(NewTcpReaderDataMsg(data, timestamp))
-				}
+				t.tcpStream.client.sendMsgIfNotClosed(NewTcpReaderDataMsg(data, timestamp))
 			} else {
-				for i := range stream.getServers() {
-					reader := stream.getServer(i)
-					reader.sendMsgIfNotClosed(NewTcpReaderDataMsg(data, timestamp))
-				}
+				t.tcpStream.server.sendMsgIfNotClosed(NewTcpReaderDataMsg(data, timestamp))
 			}
 		}
 	}
@@ -157,7 +149,7 @@ func (t *tcpReassemblyStream) ReassembledSG(sg reassembly.ScatterGather, ac reas
 
 func (t *tcpReassemblyStream) ReassemblyComplete(ac reassembly.AssemblerContext) bool {
 	if t.tcpStream.GetIsTapTarget() && !t.tcpStream.GetIsClosed() {
-		t.tcpStream.(*tcpStream).close()
+		t.tcpStream.close()
 	}
 	// do not remove the connection to allow last ACK
 	return false

tap/tcp_stream.go

@@ -13,25 +13,26 @@ import (
  * In our implementation, we pass information from ReassembledSG to the TcpReader through a shared channel.
  */
 type tcpStream struct {
-	id              int64
-	isClosed        bool
-	protoIdentifier *api.ProtoIdentifier
-	isTapTarget     bool
-	clients         []*tcpReader
-	servers         []*tcpReader
-	origin          api.Capture
-	reqResMatchers  []api.RequestResponseMatcher
-	createdAt       time.Time
-	streamsMap      api.TcpStreamMap
+	id             int64
+	isClosed       bool
+	protocol       *api.Protocol
+	isTapTarget    bool
+	client         *tcpReader
+	server         *tcpReader
+	origin         api.Capture
+	counterPairs   []*api.CounterPair
+	reqResMatchers []api.RequestResponseMatcher
+	createdAt      time.Time
+	streamsMap     api.TcpStreamMap
 	sync.Mutex
 }
 
 func NewTcpStream(isTapTarget bool, streamsMap api.TcpStreamMap, capture api.Capture) *tcpStream {
 	return &tcpStream{
-		isTapTarget:     isTapTarget,
-		protoIdentifier: &api.ProtoIdentifier{},
-		streamsMap:      streamsMap,
-		origin:          capture,
+		isTapTarget: isTapTarget,
+		streamsMap:  streamsMap,
+		origin:      capture,
+		createdAt:   time.Now(),
 	}
 }
 
@@ -55,38 +56,12 @@ func (t *tcpStream) close() {
 
 	t.streamsMap.Delete(t.id)
 
-	for i := range t.clients {
-		reader := t.clients[i]
-		reader.close()
-	}
-	for i := range t.servers {
-		reader := t.servers[i]
-		reader.close()
-	}
+	t.client.close()
+	t.server.close()
 }
 
-func (t *tcpStream) addClient(reader *tcpReader) {
-	t.clients = append(t.clients, reader)
-}
-
-func (t *tcpStream) addServer(reader *tcpReader) {
-	t.servers = append(t.servers, reader)
-}
-
-func (t *tcpStream) getClients() []*tcpReader {
-	return t.clients
-}
-
-func (t *tcpStream) getServers() []*tcpReader {
-	return t.servers
-}
-
-func (t *tcpStream) getClient(index int) *tcpReader {
-	return t.clients[index]
-}
-
-func (t *tcpStream) getServer(index int) *tcpReader {
-	return t.servers[index]
+func (t *tcpStream) addCounterPair(counterPair *api.CounterPair) {
+	t.counterPairs = append(t.counterPairs, counterPair)
 }
 
 func (t *tcpStream) addReqResMatcher(reqResMatcher api.RequestResponseMatcher) {
@@ -94,37 +69,21 @@ func (t *tcpStream) addReqResMatcher(reqResMatcher api.RequestResponseMatcher) {
 }
 
 func (t *tcpStream) SetProtocol(protocol *api.Protocol) {
+	t.protocol = protocol
+
+	// Clean the buffers
 	t.Lock()
-	defer t.Unlock()
-
-	if t.protoIdentifier.IsClosedOthers {
-		return
-	}
-
-	t.protoIdentifier.Protocol = protocol
-
-	for i := range t.clients {
-		reader := t.clients[i]
-		if reader.GetExtension().Protocol != t.protoIdentifier.Protocol {
-			reader.close()
-		}
-	}
-	for i := range t.servers {
-		reader := t.servers[i]
-		if reader.GetExtension().Protocol != t.protoIdentifier.Protocol {
-			reader.close()
-		}
-	}
-
-	t.protoIdentifier.IsClosedOthers = true
+	t.client.msgBufferMaster = make([]api.TcpReaderDataMsg, 0)
+	t.server.msgBufferMaster = make([]api.TcpReaderDataMsg, 0)
+	t.Unlock()
 }
 
 func (t *tcpStream) GetOrigin() api.Capture {
 	return t.origin
 }
 
-func (t *tcpStream) GetProtoIdentifier() *api.ProtoIdentifier {
-	return t.protoIdentifier
+func (t *tcpStream) GetProtocol() *api.Protocol {
+	return t.protocol
 }
 
 func (t *tcpStream) GetReqResMatchers() []api.RequestResponseMatcher {

tap/tcp_stream_factory.go

@@ -3,7 +3,6 @@ package tap
 import (
 	"fmt"
 	"sync"
-	"time"
 
 	"github.com/up9inc/mizu/logger"
 	"github.com/up9inc/mizu/tap/api"
@@ -62,62 +61,50 @@ func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcpLayer *lay
 	reassemblyStream := NewTcpReassemblyStream(fmt.Sprintf("%s:%s", net, transport), tcpLayer, fsmOptions, stream)
 	if stream.GetIsTapTarget() {
 		stream.setId(factory.streamsMap.NextId())
-		for i, extension := range extensions {
-			reqResMatcher := extension.Dissector.NewResponseRequestMatcher()
-			stream.addReqResMatcher(reqResMatcher)
+		for _, extension := range extensions {
 			counterPair := &api.CounterPair{
 				Request:  0,
 				Response: 0,
 			}
-			stream.addClient(
-				NewTcpReader(
-					make(chan api.TcpReaderDataMsg),
-					&api.ReadProgress{},
-					fmt.Sprintf("%s %s", net, transport),
-					&api.TcpID{
-						SrcIP:   srcIp,
-						DstIP:   dstIp,
-						SrcPort: srcPort,
-						DstPort: dstPort,
-					},
-					time.Time{},
-					stream,
-					true,
-					props.isOutgoing,
-					extension,
-					factory.emitter,
-					counterPair,
-					reqResMatcher,
-				),
-			)
-			stream.addServer(
-				NewTcpReader(
-					make(chan api.TcpReaderDataMsg),
-					&api.ReadProgress{},
-					fmt.Sprintf("%s %s", net, transport),
-					&api.TcpID{
-						SrcIP:   net.Dst().String(),
-						DstIP:   net.Src().String(),
-						SrcPort: transport.Dst().String(),
-						DstPort: transport.Src().String(),
-					},
-					time.Time{},
-					stream,
-					false,
-					props.isOutgoing,
-					extension,
-					factory.emitter,
-					counterPair,
-					reqResMatcher,
-				),
-			)
+			stream.addCounterPair(counterPair)
 
-			factory.streamsMap.Store(stream.getId(), stream)
-
-			factory.wg.Add(2)
-			go stream.getClient(i).run(filteringOptions, &factory.wg)
-			go stream.getServer(i).run(filteringOptions, &factory.wg)
+			reqResMatcher := extension.Dissector.NewResponseRequestMatcher()
+			stream.addReqResMatcher(reqResMatcher)
 		}
+
+		stream.client = NewTcpReader(
+			fmt.Sprintf("%s %s", net, transport),
+			&api.TcpID{
+				SrcIP:   srcIp,
+				DstIP:   dstIp,
+				SrcPort: srcPort,
+				DstPort: dstPort,
+			},
+			stream,
+			true,
+			props.isOutgoing,
+			factory.emitter,
+		)
+
+		stream.server = NewTcpReader(
+			fmt.Sprintf("%s %s", net, transport),
+			&api.TcpID{
+				SrcIP:   net.Dst().String(),
+				DstIP:   net.Src().String(),
+				SrcPort: transport.Dst().String(),
+				DstPort: transport.Src().String(),
+			},
+			stream,
+			false,
+			props.isOutgoing,
+			factory.emitter,
+		)
+
+		factory.streamsMap.Store(stream.getId(), stream)
+
+		factory.wg.Add(2)
+		go stream.client.run(filteringOptions, &factory.wg)
+		go stream.server.run(filteringOptions, &factory.wg)
 	}
 	return reassemblyStream
 }

tap/tcp_streams_map.go

@@ -57,7 +57,7 @@ func (streamMap *tcpStreamMap) CloseTimedoutTcpStreamChannels() {
 				return true
 			}
 
-			if stream.protoIdentifier.Protocol == nil {
+			if stream.protocol == nil {
 				if !stream.isClosed && time.Now().After(stream.createdAt.Add(tcpStreamChannelTimeoutMs)) {
 					stream.close()
 					diagnose.AppStats.IncDroppedTcpStreams()

tap/tlstapper/tls_poller.go

@@ -188,8 +188,7 @@ func (p *tlsPoller) startNewTlsReader(chunk *tlsChunk, address *addressPair, key
 	}
 
 	stream := &tlsStream{
-		reader:          reader,
-		protoIdentifier: &api.ProtoIdentifier{},
+		reader: reader,
 	}
 	streamsMap.Store(streamsMap.NextId(), stream)
 

tap/tlstapper/tls_reader.go

@@ -94,7 +94,3 @@ func (r *tlsReader) GetEmitter() api.Emitter {
 func (r *tlsReader) GetIsClosed() bool {
 	return false
 }
-
-func (r *tlsReader) GetExtension() *api.Extension {
-	return r.extension
-}

tap/tlstapper/tls_stream.go

@@ -3,20 +3,20 @@ package tlstapper
 import "github.com/up9inc/mizu/tap/api"
 
 type tlsStream struct {
-	reader          *tlsReader
-	protoIdentifier *api.ProtoIdentifier
+	reader   *tlsReader
+	protocol *api.Protocol
 }
 
 func (t *tlsStream) GetOrigin() api.Capture {
 	return api.Ebpf
 }
 
-func (t *tlsStream) GetProtoIdentifier() *api.ProtoIdentifier {
-	return t.protoIdentifier
+func (t *tlsStream) GetProtocol() *api.Protocol {
+	return t.protocol
 }
 
 func (t *tlsStream) SetProtocol(protocol *api.Protocol) {
-	t.protoIdentifier.Protocol = protocol
+	t.protocol = protocol
 }
 
 func (t *tlsStream) GetReqResMatchers() []api.RequestResponseMatcher {

ui-common/src/components/UI/StatusBar.tsx

@@ -22,8 +22,8 @@ export const StatusBar: React.FC<StatusBarProps> = ({isDemoBannerView, disabled}
     const {uniqueNamespaces, amountOfPods, amountOfTappedPods, amountOfUntappedPods} = useRecoilValue(tappingStatusDetails);
     return <div style={{opacity: disabled ? 0.4 : 1}} className={`${isDemoBannerView ? `${style.banner}` : ''} ${style.statusBar} ${(expandedBar && !disabled ? `${style.expandedStatusBar}` : "")}`} onMouseOver={() => setExpandedBar(true)} onMouseLeave={() => setExpandedBar(false)} data-cy="expandedStatusBar">
         <div className={style.podsCount}>
-        {tappingStatus.some(pod => !pod.isTapped) && <img src={warningIcon} alt="warning"/>}
-          {disabled &&  <Tooltip title={"Tapping status is not updated when streaming is paused"} isSimple><img src={warningIcon} alt="warning"/></Tooltip>}
+          {!disabled && tappingStatus.some(pod => !pod.isTapped) && <img src={warningIcon} alt="warning"/>}
+          {disabled && <Tooltip title={"Tapping status is not updated when streaming is paused"} isSimple><img src={warningIcon} alt="warning"/></Tooltip>}
             <span className={style.podsCountText} data-cy="podsCountText">
                 {`Tapping ${amountOfUntappedPods > 0 ? amountOfTappedPods + " / " + amountOfPods : amountOfPods} ${pluralize('pod', amountOfPods)} in ${pluralize('namespace', uniqueNamespaces.length)} ${uniqueNamespaces.join(", ")}`}
             </span>