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 }}

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/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/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 {