github/kubeshark
1
0
Fork 0
mirror of https://github.com/kubeshark/kubeshark.git synced 2025-07-29 23:41:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

adding the cleaner again (why we removed it?).

Browse Source 
add TODO: on the extension loop .
This commit is contained in:
Roee Gadot 2021-08-23 11:48:00 +03:00
parent efde8ae359
commit 20a2caf3d9
4 changed files with 18 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cli/kubernetes/provider.go
View File

@ -575,10 +575,10 @@ func (provider *Provider) CreateConfigMap(ctx context.Context, namespace string,
} }
func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespace string, daemonSetName string, podImage string, tapperPodName string, apiServerPodIp string, nodeToTappedPodIPMap map[string][]string, serviceAccountName string, tapOutgoing bool, resources configStructs.Resources, imagePullPolicy core.PullPolicy) error { func (provider *Provider) ApplyMizuTapperDaemonSet(ctx context.Context, namespace string, daemonSetName string, podImage string, tapperPodName string, apiServerPodIp string, nodeToTappedPodIPMap map[string][]string, serviceAccountName string, tapOutgoing bool, resources configStructs.Resources, imagePullPolicy core.PullPolicy) error {
logger.Log.Debugf("Applying %d tapper deamonsets, ns: %s, daemonSetName: %s, podImage: %s, tapperPodName: %s", len(nodeToTappedPodIPMap), namespace, daemonSetName, podImage, tapperPodName) logger.Log.Debugf("Applying %d tapper daemon sets, ns: %s, daemonSetName: %s, podImage: %s, tapperPodName: %s", len(nodeToTappedPodIPMap), namespace, daemonSetName, podImage, tapperPodName)
if len(nodeToTappedPodIPMap) == 0 { if len(nodeToTappedPodIPMap) == 0 {
return fmt.Errorf("Daemon set %s must tap at least 1 pod", daemonSetName) return fmt.Errorf("daemon set %s must tap at least 1 pod", daemonSetName)
} }
nodeToTappedPodIPMapJsonStr, err := json.Marshal(nodeToTappedPodIPMap) nodeToTappedPodIPMapJsonStr, err := json.Marshal(nodeToTappedPodIPMap)

23
tap/cleaner.go
View File

@ -1,6 +1,8 @@
package tap package tap
import ( import (
"github.com/google/gopacket/reassembly"
"github.com/romana/rlog"
"sync" "sync"
"time" "time"
) )
@ -12,6 +14,7 @@ type CleanerStats struct {
} }
type Cleaner struct { type Cleaner struct {
assembler *reassembly.Assembler
assemblerMutex *sync.Mutex assemblerMutex *sync.Mutex
cleanPeriod time.Duration cleanPeriod time.Duration
connectionTimeout time.Duration connectionTimeout time.Duration
@ -20,18 +23,18 @@ type Cleaner struct {
} }
func (cl *Cleaner) clean() { func (cl *Cleaner) clean() {
// startCleanTime := time.Now() startCleanTime := time.Now()
// cl.assemblerMutex.Lock() cl.assemblerMutex.Lock()
// rlog.Debugf("Assembler Stats before cleaning %s", cl.assembler.Dump()) rlog.Debugf("Assembler Stats before cleaning %s", cl.assembler.Dump())
// flushed, closed := cl.assembler.FlushCloseOlderThan(startCleanTime.Add(-cl.connectionTimeout)) flushed, closed := cl.assembler.FlushCloseOlderThan(startCleanTime.Add(-cl.connectionTimeout))
// cl.assemblerMutex.Unlock() cl.assemblerMutex.Unlock()
// cl.statsMutex.Lock() cl.statsMutex.Lock()
// rlog.Debugf("Assembler Stats after cleaning %s", cl.assembler.Dump()) rlog.Debugf("Assembler Stats after cleaning %s", cl.assembler.Dump())
// cl.stats.flushed += flushed cl.stats.flushed += flushed
// cl.stats.closed += closed cl.stats.closed += closed
// cl.statsMutex.Unlock() cl.statsMutex.Unlock()
} }
func (cl *Cleaner) start() { func (cl *Cleaner) start() {

1
tap/passive_tapper.go
View File

@ -341,6 +341,7 @@ func startPassiveTapper(outputItems chan *api.OutputChannelItem, allExtensionPor
staleConnectionTimeout := time.Second * time.Duration(*staleTimeoutSeconds) staleConnectionTimeout := time.Second * time.Duration(*staleTimeoutSeconds)
cleaner := Cleaner{ cleaner := Cleaner{
assembler: assembler,
assemblerMutex: &assemblerMutex, assemblerMutex: &assemblerMutex,
cleanPeriod: cleanPeriod, cleanPeriod: cleanPeriod,
connectionTimeout: staleConnectionTimeout, connectionTimeout: staleConnectionTimeout,

2
tap/tcp_reader.go
View File

@ -107,6 +107,8 @@ func (h *tcpReader) run(wg *sync.WaitGroup) {
port = h.tcpID.SrcPort port = h.tcpID.SrcPort
} }
b := bufio.NewReader(h) b := bufio.NewReader(h)
// TODO: maybe check for kafka and amqp and when it is not one of those pass it to the HTTP?
// because it will check for the ports that we checked in the "isTapTarget"
for _, extension := range extensions { for _, extension := range extensions {
if containsPort(extension.Protocol.Ports, port) { if containsPort(extension.Protocol.Ports, port) {
extension.Dissector.Dissect(b, h.isClient, h.tcpID, h.Emitter) extension.Dissector.Dissect(b, h.isClient, h.tcpID, h.Emitter)