mirror of
https://github.com/kubeshark/kubeshark.git
synced 2026-05-08 02:08:24 +00:00
bfa834e840
* Spawn only two Goroutines per TCP stream * Fix the linter error * Use `isProtocolIdentified` method instead * Fix the `Read` method of `tcpReader` * Remove unnecessary `append` * Copy to buffer only a message is received * Remove `exhaustBuffer` field and add `rewind` function * Rename `buffer` field to `pastData` * Update tap/tcp_reader.go Co-authored-by: Nimrod Gilboa Markevich <59927337+nimrod-up9@users.noreply.github.com> * Use `copy` instead of assignment * No lint * #run_acceptance_tests * Fix `rewind` #run_acceptance_tests * Fix the buffering algorithm #run_acceptance_tests * Add `TODO` * Fix the problems in AMQP and Kafka #run_acceptance_tests * Use `*bytes.Buffer` instead of `[]api.TcpReaderDataMsg` #run_acceptance_tests * Have a single `*bytes.Buffer` * Revert "Have a single `*bytes.Buffer`" This reverts commitfad96a288a. * Revert "Use `*bytes.Buffer` instead of `[]api.TcpReaderDataMsg` #run_acceptance_tests" This reverts commit0fc70bffe2. * Fix the early timing out issue #run_acceptance_tests * Remove `NewBytes()` method * Update the `NewTcpReader` method signature #run_acceptance_tests * #run_acceptance_tests * #run_acceptance_tests * #run_acceptance_tests Co-authored-by: Nimrod Gilboa Markevich <59927337+nimrod-up9@users.noreply.github.com>
98 lines
2.1 KiB
Go
98 lines
2.1 KiB
Go
package tap
|
|
|
|
import (
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/up9inc/mizu/tap/api"
|
|
)
|
|
|
|
/* It's a connection (bidirectional)
|
|
* Implements gopacket.reassembly.Stream interface (Accept, ReassembledSG, ReassemblyComplete)
|
|
* ReassembledSG gets called when new reassembled data is ready (i.e. bytes in order, no duplicates, complete)
|
|
* In our implementation, we pass information from ReassembledSG to the TcpReader through a shared channel.
|
|
*/
|
|
type tcpStream struct {
|
|
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,
|
|
streamsMap: streamsMap,
|
|
origin: capture,
|
|
createdAt: time.Now(),
|
|
}
|
|
}
|
|
|
|
func (t *tcpStream) getId() int64 {
|
|
return t.id
|
|
}
|
|
|
|
func (t *tcpStream) setId(id int64) {
|
|
t.id = id
|
|
}
|
|
|
|
func (t *tcpStream) close() {
|
|
t.Lock()
|
|
defer t.Unlock()
|
|
|
|
if t.isClosed {
|
|
return
|
|
}
|
|
|
|
t.isClosed = true
|
|
|
|
t.streamsMap.Delete(t.id)
|
|
|
|
t.client.close()
|
|
t.server.close()
|
|
}
|
|
|
|
func (t *tcpStream) addCounterPair(counterPair *api.CounterPair) {
|
|
t.counterPairs = append(t.counterPairs, counterPair)
|
|
}
|
|
|
|
func (t *tcpStream) addReqResMatcher(reqResMatcher api.RequestResponseMatcher) {
|
|
t.reqResMatchers = append(t.reqResMatchers, reqResMatcher)
|
|
}
|
|
|
|
func (t *tcpStream) SetProtocol(protocol *api.Protocol) {
|
|
t.protocol = protocol
|
|
|
|
// Clean the buffers
|
|
t.client.msgBufferMaster = make([]api.TcpReaderDataMsg, 0)
|
|
t.server.msgBufferMaster = make([]api.TcpReaderDataMsg, 0)
|
|
}
|
|
|
|
func (t *tcpStream) GetOrigin() api.Capture {
|
|
return t.origin
|
|
}
|
|
|
|
func (t *tcpStream) GetProtocol() *api.Protocol {
|
|
return t.protocol
|
|
}
|
|
|
|
func (t *tcpStream) GetReqResMatchers() []api.RequestResponseMatcher {
|
|
return t.reqResMatchers
|
|
}
|
|
|
|
func (t *tcpStream) GetIsTapTarget() bool {
|
|
return t.isTapTarget
|
|
}
|
|
|
|
func (t *tcpStream) GetIsClosed() bool {
|
|
return t.isClosed
|
|
}
|