introduce freecache to avoid golang map OOM (#1039)

Co-authored-by: wangfeng115 <wangfeng115@ke.com>

Using the freecache library to avoid golang map OOM. As follows
https://github.com/golang/go/issues/20135

emitter.go

@@ -6,9 +6,9 @@ import (
 	"io"
 	"log"
 	"sync"
-	"time"
 
 	"github.com/buger/goreplay/byteutils"
+	"github.com/coocood/freecache"
 )
 
 // Emitter represents an abject to manage plugins communication
@@ -76,9 +76,7 @@ func (e *Emitter) Close() {
 func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 	wIndex := 0
 	modifier := NewHTTPModifier(&Settings.ModifierConfig)
-	filteredRequests := make(map[string]int64)
-	filteredRequestsLastCleanTime := time.Now().UnixNano()
-	filteredCount := 0
+	filteredRequests := freecache.NewCache(200 * 1024 * 1024) // 200M
 
 	for {
 		msg, err := src.PluginRead()
@@ -97,7 +95,7 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 				Debug(2, fmt.Sprintf("[EMITTER] Found malformed record %q from %q", msg.Meta, src))
 				continue
 			}
-			requestID := byteutils.SliceToString(meta[1])
+			requestID := meta[1]
 			// start a subroutine only when necessary
 			if Settings.Verbose >= 3 {
 				Debug(3, "[EMITTER] input: ", byteutils.SliceToString(msg.Meta[:len(msg.Meta)-1]), " from: ", src)
@@ -108,16 +106,15 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 					msg.Data = modifier.Rewrite(msg.Data)
 					// If modifier tells to skip request
 					if len(msg.Data) == 0 {
-						filteredRequests[requestID] = time.Now().UnixNano()
-						filteredCount++
+						filteredRequests.Set(requestID, []byte{}, 60) //
 						continue
 					}
 					Debug(3, "[EMITTER] Rewritten input:", requestID, "from:", src)
 
 				} else {
-					if _, ok := filteredRequests[requestID]; ok {
-						delete(filteredRequests, requestID)
-						filteredCount--
+					_, err := filteredRequests.Get(requestID)
+					if err == nil {
+						filteredRequests.Del(requestID)
 						continue
 					}
 				}
@@ -158,20 +155,5 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 				}
 			}
 		}
-
-		// Run GC on each 1000 request
-		if filteredCount > 0 && filteredCount%1000 == 0 {
-			// Clean up filtered requests for which we didn't get a response to filter
-			now := time.Now().UnixNano()
-			if now-filteredRequestsLastCleanTime > int64(60*time.Second) {
-				for k, v := range filteredRequests {
-					if now-v > int64(60*time.Second) {
-						delete(filteredRequests, k)
-						filteredCount--
-					}
-				}
-				filteredRequestsLastCleanTime = time.Now().UnixNano()
-			}
-		}
 	}
 }

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/aws/aws-sdk-go v1.33.2
 	github.com/bitly/go-hostpool v0.1.0 // indirect
 	github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
+	github.com/coocood/freecache v1.2.0
 	github.com/google/gopacket v1.1.20-0.20210429153827-3eaba0894325
 	github.com/klauspost/compress v1.10.10 // indirect
 	github.com/mattbaird/elastigo v0.0.0-20170123220020-2fe47fd29e4b

go.sum

@@ -1,3 +1,4 @@
+github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU=
 github.com/Shopify/sarama v1.26.4 h1:+17TxUq/PJEAfZAll0T7XJjSgQWCpaQSoki/x5yN8o8=
 github.com/Shopify/sarama v1.26.4/go.mod h1:NbSGBSSndYaIhRcBtY9V0U7AyH+x71bG668AuWys/yU=
 github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWsokNbMijUGhmcoBJc=
@@ -10,6 +11,10 @@ github.com/bitly/go-hostpool v0.1.0 h1:XKmsF6k5el6xHG3WPJ8U0Ku/ye7njX7W81Ng7O2io
 github.com/bitly/go-hostpool v0.1.0/go.mod h1:4gOCgp6+NZnVqlKyZ/iBZFTAJKembaVENUpMkpg42fw=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
+github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
+github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
+github.com/coocood/freecache v1.2.0 h1:p8RhjN6Y4DRBIMzdRlm1y+M7h7YJxye3lGW8/VvzCz0=
+github.com/coocood/freecache v1.2.0/go.mod h1:OKrEjkGVoxZhyWAJoeFi5BMLUJm2Tit0kpGkIr7NGYY=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -65,6 +70,7 @@ github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykE
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
+github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=