Increase the OOM risk in exchange of less idle CPU usage (#979)

* Increase the OOM risk in exchange of less idle CPU usage

* Read the interval from an environment variable named `CLOSE_TIMEDOUT_TCP_STREAM_CHANNELS_INTERVAL_MS`

* Log the `getCloseTimedoutTcpChannelsInterval` return value

.github/workflows/inactive-issues-close.yaml

@@ -13,10 +13,11 @@ jobs:
       - uses: actions/stale@v3
         with:
           days-before-issue-stale: 30
-          days-before-issue-close: 14
+          days-before-issue-close: -1
+          exempt-issue-labels: "enhancement"
           stale-issue-label: "stale"
           stale-issue-message: "This issue is stale because it has been open for 30 days with no activity."
-          close-issue-message: "This issue was closed because it has been inactive for 14 days since being marked as stale."
+          close-issue-message: ""
           days-before-pr-stale: -1
           days-before-pr-close: -1
           repo-token: ${{ secrets.GITHUB_TOKEN }}

acceptanceTests/cypress/integration/testHelpers/TrafficHelper.js

@@ -39,13 +39,13 @@ export function verifyMinimumEntries() {
     });
 }
 
-export function leftTextCheck(entryNum, path, expectedText) {
-    cy.get(`#list #entry-${entryNum} ${path}`).invoke('text').should('eq', expectedText);
+export function leftTextCheck(entryId, path, expectedText) {
+    cy.get(`#list #entry-${entryId} ${path}`).invoke('text').should('eq', expectedText);
 }
 
-export function leftOnHoverCheck(entryNum, path, filterName) {
-    cy.get(`#list #entry-${entryNum} ${path}`).trigger('mouseover');
-    cy.get(`#list #entry-${entryNum} [data-cy='QueryableTooltip']`).invoke('text').should('match', new RegExp(filterName));
+export function leftOnHoverCheck(entryId, path, filterName) {
+    cy.get(`#list #entry-${entryId} ${path}`).trigger('mouseover');
+    cy.get(`#list #entry-${entryId} [data-cy='QueryableTooltip']`).invoke('text').should('match', new RegExp(filterName));
 }
 
 export function rightTextCheck(path, expectedText) {
@@ -89,13 +89,13 @@ export function checkFilterByMethod(funcDict) {
                     cy.get('#entries-length').invoke('text').then(len => {
                         resizeIfNeeded(len);
                         listElmWithIdAttr.forEach(entry => {
-                            if (entry?.id && entry.id.match(RegExp(/entry-(\d{2}|\d{1})$/gm))) {
-                                const entryNum = getEntryNumById(entry.id);
+                            if (entry?.id && entry.id.match(RegExp(/entry-(\d{24})$/gm))) {
+                                const entryId = getEntryId(entry.id);
 
-                                leftTextCheck(entryNum, methodDict.pathLeft, methodDict.expectedText);
-                                leftTextCheck(entryNum, protocolDict.pathLeft, protocolDict.expectedTextLeft);
+                                leftTextCheck(entryId, methodDict.pathLeft, methodDict.expectedText);
+                                leftTextCheck(entryId, protocolDict.pathLeft, protocolDict.expectedTextLeft);
                                 if (summaryDict)
-                                    leftTextCheck(entryNum, summaryDict.pathLeft, summaryDict.expectedText);
+                                    leftTextCheck(entryId, summaryDict.pathLeft, summaryDict.expectedText);
 
                                 if (!doneCheckOnFirst) {
                                     deepCheck(funcDict, protocolDict, methodDict, entry);
@@ -111,6 +111,11 @@ export function checkFilterByMethod(funcDict) {
     });
 }
 
+export function getEntryId(id) {
+    // take the second part from the string (entry-<ID>)
+    return id.split('-')[1];
+}
+
 function resizeIfNeeded(entriesLen) {
     if (entriesLen > maxEntriesInDom){
         Cypress.config().viewportHeight === Cypress.env('normalMizuHeight') ?
@@ -119,14 +124,14 @@ function resizeIfNeeded(entriesLen) {
 }
 
 function deepCheck(generalDict, protocolDict, methodDict, entry) {
-    const entryNum = getEntryNumById(entry.id);
+    const entryId = getEntryId(entry.id);
     const {summary, value} = generalDict;
     const summaryDict = getSummaryDict(summary);
 
-    leftOnHoverCheck(entryNum, methodDict.pathLeft, methodDict.expectedOnHover);
-    leftOnHoverCheck(entryNum, protocolDict.pathLeft, protocolDict.expectedOnHover);
+    leftOnHoverCheck(entryId, methodDict.pathLeft, methodDict.expectedOnHover);
+    leftOnHoverCheck(entryId, protocolDict.pathLeft, protocolDict.expectedOnHover);
     if (summaryDict)
-        leftOnHoverCheck(entryNum, summaryDict.pathLeft, summaryDict.expectedOnHover);
+        leftOnHoverCheck(entryId, summaryDict.pathLeft, summaryDict.expectedOnHover);
 
     cy.get(`#${entry.id}`).click();
 
@@ -183,7 +188,3 @@ function getProtocolDict(protocol, protocolText) {
         expectedOnHover: protocol.toLowerCase()
     };
 }
-
-function getEntryNumById (id) {
-    return parseInt(id.split('-')[1]);
-}

acceptanceTests/cypress/integration/tests/IgnoredUserAgents.js

@@ -15,15 +15,11 @@ function checkEntries() {
         checkThatAllEntriesShown();
         resizeToHugeMizu();
 
-        cy.get('#total-entries').then(number => {
-            const numOfEntries = parseInt(number.text());
-            [...Array(numOfEntries).keys()].map(checkEntry);
+        cy.get('#list [id^=entry]').each(entryElement => {
+            entryElement.click();
+            cy.get('#tbody-Headers').should('be.visible');
+            isValueExistsInElement(false, 'Ignored-User-Agent', '#tbody-Headers');
         });
     });
 }
 
-function checkEntry(entryIndex) {
-    cy.get(`#entry-${entryIndex}`).click();
-    cy.get('#tbody-Headers').should('be.visible');
-    isValueExistsInElement(false, 'Ignored-User-Agent', '#tbody-Headers');
-}

acceptanceTests/cypress/integration/tests/UiTest.js

@@ -1,5 +1,6 @@
 import {findLineAndCheck, getExpectedDetailsDict} from "../testHelpers/StatusBarHelper";
 import {
+    getEntryId,
     leftOnHoverCheck,
     leftTextCheck,
     resizeToHugeMizu,
@@ -148,9 +149,6 @@ function checkFilterNoResults(filterName) {
             // the DOM should show 0 entries
             cy.get('#entries-length').should('have.text', '0');
 
-            // going through every potential entry and verifies that it doesn't exist
-            [...Array(parseInt(totalEntries)).keys()].map(shouldNotExist);
-
             cy.get('[title="Fetch old records"]').click();
             cy.get('#noMoreDataTop', {timeout: refreshWaitTimeout}).should('be.visible');
             cy.get('#entries-length').should('have.text', '0'); // after loading all entries there should still be 0 entries
@@ -162,10 +160,6 @@ function checkFilterNoResults(filterName) {
     });
 }
 
-function shouldNotExist(entryNum) {
-    cy.get(`entry-${entryNum}`).should('not.exist');
-}
-
 function checkIllegalFilter(illegalFilterName) {
     it(`should show red search bar with the input: ${illegalFilterName}`, function () {
         cy.reload();
@@ -185,30 +179,44 @@ function checkIllegalFilter(illegalFilterName) {
         });
     });
 }
-function checkFilter(filterDetails){
-    const {name, leftSidePath, rightSidePath, rightSideExpectedText, leftSideExpectedText, applyByEnter} = filterDetails;
+
+function checkFilter(filterDetails) {
+    const {
+        name,
+        leftSidePath,
+        rightSidePath,
+        rightSideExpectedText,
+        leftSideExpectedText,
+        applyByEnter
+    } = filterDetails;
+
     const entriesForDeeperCheck = 5;
 
     it(`checking the filter: ${name}`, function () {
         cy.get('#total-entries').should('not.have.text', '0').then(number => {
             const totalEntries = number.text();
 
-            // checks the hover on the last entry (the only one in DOM at the beginning)
-            leftOnHoverCheck(totalEntries - 1, leftSidePath, name);
+            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, name);
 
-            cy.get('.w-tc-editor-text').clear();
-            // applying the filter with alt+enter or with the button
-            cy.get('.w-tc-editor-text').type(`${name}${applyByEnter ? '{alt+enter}' : ''}`);
-            cy.get('.w-tc-editor').should('have.attr', 'style').and('include', Cypress.env('greenFilterColor'));
-            if (!applyByEnter)
-                cy.get('[type="submit"]').click();
+                cy.get('.w-tc-editor-text').clear();
+                // applying the filter with alt+enter or with the button
+                cy.get('.w-tc-editor-text').type(`${name}${applyByEnter ? '{alt+enter}' : ''}`);
+                cy.get('.w-tc-editor').should('have.attr', 'style').and('include', Cypress.env('greenFilterColor'));
+                if (!applyByEnter)
+                    cy.get('[type="submit"]').click();
 
-            // only one entry in DOM after filtering, checking all checks on it
-            leftTextCheck(totalEntries - 1, leftSidePath, leftSideExpectedText);
-            leftOnHoverCheck(totalEntries - 1, leftSidePath, name);
-            rightTextCheck(rightSidePath, rightSideExpectedText);
-            rightOnHoverCheck(rightSidePath, name);
-            checkRightSideResponseBody();
+                // only one entry in DOM after filtering, checking all checks on it
+                leftTextCheck(entryId, leftSidePath, leftSideExpectedText);
+                leftOnHoverCheck(entryId, leftSidePath, name);
+
+                rightTextCheck(rightSidePath, rightSideExpectedText);
+                rightOnHoverCheck(rightSidePath, name);
+                checkRightSideResponseBody();
+            });
 
             cy.get('[title="Fetch old records"]').click();
             resizeToHugeMizu();
@@ -217,28 +225,33 @@ function checkFilter(filterDetails){
             cy.get('#entries-length', {timeout: refreshWaitTimeout}).should('have.text', totalEntries);
 
             // checking only 'leftTextCheck' on all entries because the rest of the checks require more time
-            [...Array(parseInt(totalEntries)).keys()].forEach(entryNum => {
-                leftTextCheck(entryNum, leftSidePath, leftSideExpectedText);
+            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)
-            deeperChcek(leftSidePath, rightSidePath, name, leftSideExpectedText, rightSideExpectedText, entriesForDeeperCheck);
+            deeperCheck(leftSidePath, rightSidePath, name, leftSideExpectedText, rightSideExpectedText, entriesForDeeperCheck);
 
             // reloading then waiting for the entries number to load
             resizeToNormalMizu();
             cy.reload();
             cy.get('#total-entries', {timeout: refreshWaitTimeout}).should('have.text', totalEntries);
-        });
+        })
     });
 }
 
-function deeperChcek(leftSidePath, rightSidePath, filterName, leftSideExpectedText, rightSideExpectedText, entriesNumToCheck) {
-    [...Array(entriesNumToCheck).keys()].forEach(entryNum => {
-        leftOnHoverCheck(entryNum, leftSidePath, filterName);
+function deeperCheck(leftSidePath, rightSidePath, filterName, leftSideExpectedText, rightSideExpectedText, entriesNumToCheck) {
+    cy.get(`#list [id^=entry]`).each((element, index) => {
+        if (index < entriesNumToCheck) {
+            const entryId = getEntryId(element[0].id);
+            leftOnHoverCheck(entryId, leftSidePath, filterName);
 
-        cy.get(`#list #entry-${entryNum}`).click();
-        rightTextCheck(rightSidePath, rightSideExpectedText);
-        rightOnHoverCheck(rightSidePath, filterName);
+            element.click();
+            rightTextCheck(rightSidePath, rightSideExpectedText);
+            rightOnHoverCheck(rightSidePath, filterName);
+        }
     });
 }
 

shared/kubernetes/mizuTapperSyncer.go

@@ -329,7 +329,7 @@ func (tapperSyncer *MizuTapperSyncer) updateMizuTappers() error {
 			TapperDaemonSetName,
 			tapperSyncer.config.AgentImage,
 			TapperPodName,
-			fmt.Sprintf("%s.%s.svc.cluster.local", ApiServerPodName, tapperSyncer.config.MizuResourcesNamespace),
+			fmt.Sprintf("%s.%s.svc", ApiServerPodName, tapperSyncer.config.MizuResourcesNamespace),
 			nodeNames,
 			serviceAccountName,
 			tapperSyncer.config.TapperResources,

tap/extensions/amqp/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-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/expect6/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/expect7/amqp/\* expect

tap/extensions/http/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-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/expect6/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/expect7/http/\* expect

tap/extensions/kafka/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-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/expect6/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/expect7/kafka/\* expect

tap/extensions/redis/Makefile

@@ -13,4 +13,4 @@ test-pull-bin:
 
 test-pull-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/expect6/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/expect7/redis/\* expect

tap/settings.go

@@ -13,6 +13,7 @@ const (
 	MaxBufferedPagesTotalEnvVarName           = "MAX_BUFFERED_PAGES_TOTAL"
 	MaxBufferedPagesPerConnectionEnvVarName   = "MAX_BUFFERED_PAGES_PER_CONNECTION"
 	TcpStreamChannelTimeoutMsEnvVarName       = "TCP_STREAM_CHANNEL_TIMEOUT_MS"
+	CloseTimedoutTcpChannelsIntervalMsEnvVar  = "CLOSE_TIMEDOUT_TCP_STREAM_CHANNELS_INTERVAL_MS"
 	MaxBufferedPagesTotalDefaultValue         = 5000
 	MaxBufferedPagesPerConnectionDefaultValue = 5000
 	TcpStreamChannelTimeoutMsDefaultValue     = 10000

tap/source/netns_packet_source.go

@@ -5,11 +5,12 @@ import (
 	"runtime"
 
 	"github.com/up9inc/mizu/shared/logger"
+	"github.com/up9inc/mizu/tap/api"
 	"github.com/vishvananda/netns"
 )
 
-func newNetnsPacketSource(procfs string, pid string,
-	interfaceName string, behaviour TcpPacketSourceBehaviour) (*tcpPacketSource, error) {
+func newNetnsPacketSource(procfs string, pid string, interfaceName string,
+	behaviour TcpPacketSourceBehaviour, origin api.Capture) (*tcpPacketSource, error) {
 	nsh, err := netns.GetFromPath(fmt.Sprintf("%s/%s/ns/net", procfs, pid))
 
 	if err != nil {
@@ -17,7 +18,7 @@ func newNetnsPacketSource(procfs string, pid string,
 		return nil, err
 	}
 
-	src, err := newPacketSourceFromNetnsHandle(pid, nsh, interfaceName, behaviour)
+	src, err := newPacketSourceFromNetnsHandle(pid, nsh, interfaceName, behaviour, origin)
 
 	if err != nil {
 		logger.Log.Errorf("Error starting netns packet source for %s - %w", pid, err)
@@ -28,7 +29,7 @@ func newNetnsPacketSource(procfs string, pid string,
 }
 
 func newPacketSourceFromNetnsHandle(pid string, nsh netns.NsHandle, interfaceName string,
-	behaviour TcpPacketSourceBehaviour) (*tcpPacketSource, error) {
+	behaviour TcpPacketSourceBehaviour, origin api.Capture) (*tcpPacketSource, error) {
 
 	done := make(chan *tcpPacketSource)
 	errors := make(chan error)
@@ -57,7 +58,7 @@ func newPacketSourceFromNetnsHandle(pid string, nsh netns.NsHandle, interfaceNam
 		}
 
 		name := fmt.Sprintf("netns-%s-%s", pid, interfaceName)
-		src, err := newTcpPacketSource(name, "", interfaceName, behaviour)
+		src, err := newTcpPacketSource(name, "", interfaceName, behaviour, origin)
 
 		if err != nil {
 			logger.Log.Errorf("Error listening to PID %s - %w", pid, err)

tap/source/packet_source_manager.go

@@ -5,6 +5,7 @@ import (
 	"strings"
 
 	"github.com/up9inc/mizu/shared/logger"
+	"github.com/up9inc/mizu/tap/api"
 	v1 "k8s.io/api/core/v1"
 )
 
@@ -37,10 +38,10 @@ func NewPacketSourceManager(procfs string, filename string, interfaceName string
 	}
 
 	sourceManager.config = PacketSourceManagerConfig{
-		mtls: mtls,
-		procfs: procfs,
+		mtls:          mtls,
+		procfs:        procfs,
 		interfaceName: interfaceName,
-		behaviour: behaviour,
+		behaviour:     behaviour,
 	}
 
 	go hostSource.readPackets(ipdefrag, packets)
@@ -56,7 +57,7 @@ func newHostPacketSource(filename string, interfaceName string,
 		name = fmt.Sprintf("file-%s", filename)
 	}
 
-	source, err := newTcpPacketSource(name, filename, interfaceName, behaviour)
+	source, err := newTcpPacketSource(name, filename, interfaceName, behaviour, api.Pcap)
 	if err != nil {
 		return nil, err
 	}
@@ -85,9 +86,9 @@ func (m *PacketSourceManager) updateMtlsPods(procfs string, pods []v1.Pod,
 		}
 	}
 
-	for pid := range relevantPids {
+	for pid, origin := range relevantPids {
 		if _, ok := m.sources[pid]; !ok {
-			source, err := newNetnsPacketSource(procfs, pid, interfaceName, behaviour)
+			source, err := newNetnsPacketSource(procfs, pid, interfaceName, behaviour, origin)
 
 			if err == nil {
 				go source.readPackets(ipdefrag, packets)
@@ -97,15 +98,15 @@ func (m *PacketSourceManager) updateMtlsPods(procfs string, pods []v1.Pod,
 	}
 }
 
-func (m *PacketSourceManager) getRelevantPids(procfs string, pods []v1.Pod) map[string]bool {
-	relevantPids := make(map[string]bool)
-	relevantPids[hostSourcePid] = true
+func (m *PacketSourceManager) getRelevantPids(procfs string, pods []v1.Pod) map[string]api.Capture {
+	relevantPids := make(map[string]api.Capture)
+	relevantPids[hostSourcePid] = api.Pcap
 
 	if envoyPids, err := discoverRelevantEnvoyPids(procfs, pods); err != nil {
 		logger.Log.Warningf("Unable to discover envoy pids - %w", err)
 	} else {
 		for _, pid := range envoyPids {
-			relevantPids[pid] = true
+			relevantPids[pid] = api.Envoy
 		}
 	}
 
@@ -113,7 +114,7 @@ func (m *PacketSourceManager) getRelevantPids(procfs string, pods []v1.Pod) map[
 		logger.Log.Warningf("Unable to discover linkerd pids - %w", err)
 	} else {
 		for _, pid := range linkerdPids {
-			relevantPids[pid] = true
+			relevantPids[pid] = api.Linkerd
 		}
 	}
 

tap/source/tcp_packet_source.go

@@ -10,6 +10,7 @@ import (
 	"github.com/google/gopacket/layers"
 	"github.com/google/gopacket/pcap"
 	"github.com/up9inc/mizu/shared/logger"
+	"github.com/up9inc/mizu/tap/api"
 	"github.com/up9inc/mizu/tap/diagnose"
 )
 
@@ -19,6 +20,7 @@ type tcpPacketSource struct {
 	defragger *ip4defrag.IPv4Defragmenter
 	Behaviour *TcpPacketSourceBehaviour
 	name      string
+	Origin    api.Capture
 }
 
 type TcpPacketSourceBehaviour struct {
@@ -36,13 +38,14 @@ type TcpPacketInfo struct {
 }
 
 func newTcpPacketSource(name, filename string, interfaceName string,
-	behaviour TcpPacketSourceBehaviour) (*tcpPacketSource, error) {
+	behaviour TcpPacketSourceBehaviour, origin api.Capture) (*tcpPacketSource, error) {
 	var err error
 
 	result := &tcpPacketSource{
 		name:      name,
 		defragger: ip4defrag.NewIPv4Defragmenter(),
 		Behaviour: &behaviour,
+		Origin:    origin,
 	}
 
 	if filename != "" {

tap/tcp_assembler.go

@@ -29,6 +29,7 @@ type tcpAssembler struct {
 // The assembler context
 type context struct {
 	CaptureInfo gopacket.CaptureInfo
+	Origin      api.Capture
 }
 
 func (c *context) GetCaptureInfo() gopacket.CaptureInfo {
@@ -87,8 +88,10 @@ func (a *tcpAssembler) processPackets(dumpPacket bool, packets <-chan source.Tcp
 					logger.Log.Fatalf("Failed to set network layer for checksum: %s", err)
 				}
 			}
+
 			c := context{
 				CaptureInfo: packet.Metadata().CaptureInfo,
+				Origin:      packetInfo.Source.Origin,
 			}
 			diagnose.InternalStats.Totalsz += len(tcp.Payload)
 			a.assemblerMutex.Lock()

tap/tcp_reader.go

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

tap/tcp_stream.go

@@ -29,6 +29,7 @@ type tcpStream struct {
 	clients         []tcpReader
 	servers         []tcpReader
 	ident           string
+	origin          api.Capture
 	sync.Mutex
 	streamsMap *tcpStreamMap
 }

tap/tcp_stream_factory.go

@@ -78,6 +78,7 @@ func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.T
 		optchecker:      reassembly.NewTCPOptionCheck(),
 		superIdentifier: &api.SuperIdentifier{},
 		streamsMap:      factory.streamsMap,
+		origin:          getPacketOrigin(ac),
 	}
 	if stream.isTapTarget {
 		stream.id = factory.streamsMap.nextId()
@@ -182,6 +183,17 @@ func (factory *tcpStreamFactory) shouldNotifyOnOutboundLink(dstIP string, dstPor
 	return true
 }
 
+func getPacketOrigin(ac reassembly.AssemblerContext) api.Capture {
+	c, ok := ac.(*context)
+
+	if !ok {
+		// If ac is not our context, fallback to Pcap
+		return api.Pcap
+	}
+
+	return c.Origin
+}
+
 type streamProps struct {
 	isTapTarget bool
 	isOutgoing  bool

tap/tcp_streams_map.go

@@ -1,8 +1,10 @@
 package tap
 
 import (
+	"os"
 	"runtime"
 	_debug "runtime/debug"
+	"strconv"
 	"sync"
 	"time"
 
@@ -34,10 +36,35 @@ func (streamMap *tcpStreamMap) nextId() int64 {
 	return streamMap.streamId
 }
 
+func (streamMap *tcpStreamMap) getCloseTimedoutTcpChannelsInterval() time.Duration {
+	defaultDuration := 1000 * time.Millisecond
+	rangeMin := 10
+	rangeMax := 10000
+	closeTimedoutTcpChannelsIntervalMsStr := os.Getenv(CloseTimedoutTcpChannelsIntervalMsEnvVar)
+	if closeTimedoutTcpChannelsIntervalMsStr == "" {
+		return defaultDuration
+	} else {
+		closeTimedoutTcpChannelsIntervalMs, err := strconv.Atoi(closeTimedoutTcpChannelsIntervalMsStr)
+		if err != nil {
+			logger.Log.Warningf("Error parsing environment variable %s: %v\n", CloseTimedoutTcpChannelsIntervalMsEnvVar, err)
+			return defaultDuration
+		} else {
+			if closeTimedoutTcpChannelsIntervalMs < rangeMin || closeTimedoutTcpChannelsIntervalMs > rangeMax {
+				logger.Log.Warningf("The value of environment variable %s is not in acceptable range: %d - %d\n", CloseTimedoutTcpChannelsIntervalMsEnvVar, rangeMin, rangeMax)
+				return defaultDuration
+			} else {
+				return time.Duration(closeTimedoutTcpChannelsIntervalMs) * time.Millisecond
+			}
+		}
+	}
+}
+
 func (streamMap *tcpStreamMap) closeTimedoutTcpStreamChannels() {
 	tcpStreamChannelTimeout := GetTcpChannelTimeoutMs()
+	closeTimedoutTcpChannelsIntervalMs := streamMap.getCloseTimedoutTcpChannelsInterval()
+	logger.Log.Infof("Using %d ms as the close timedout TCP stream channels interval", closeTimedoutTcpChannelsIntervalMs/time.Millisecond)
 	for {
-		time.Sleep(10 * time.Millisecond)
+		time.Sleep(closeTimedoutTcpChannelsIntervalMs)
 		_debug.FreeOSMemory()
 		streamMap.streams.Range(func(key interface{}, value interface{}) bool {
 			streamWrapper := value.(*tcpStreamWrapper)
@@ -47,7 +74,7 @@ func (streamMap *tcpStreamMap) closeTimedoutTcpStreamChannels() {
 					stream.Close()
 					diagnose.AppStats.IncDroppedTcpStreams()
 					logger.Log.Debugf("Dropped an unidentified TCP stream because of timeout. Total dropped: %d Total Goroutines: %d Timeout (ms): %d",
-						diagnose.AppStats.DroppedTcpStreams, runtime.NumGoroutine(), tcpStreamChannelTimeout/1000000)
+						diagnose.AppStats.DroppedTcpStreams, runtime.NumGoroutine(), tcpStreamChannelTimeout/time.Millisecond)
 				}
 			} else {
 				if !stream.superIdentifier.IsClosedOthers {