From 6700f156e89346332aa90e2d814ba8f635d24846 Mon Sep 17 00:00:00 2001 From: "M. Mert Yildiran" Date: Thu, 2 Sep 2021 18:39:08 +0300 Subject: [PATCH] Permanently resolve the memory exhaustion in AMQP Introduce; - `MEMORY_PROFILING_DUMP_PATH` - `MEMORY_PROFILING_TIME_INTERVAL` environment variables and make `startMemoryProfiler` method more parameterized. --- .gitignore | 3 ++ tap/extensions/amqp/main.go | 12 ++++++-- tap/extensions/amqp/read.go | 7 ++++- tap/extensions/amqp/spec091.go | 53 +++++++++++++++++----------------- tap/extensions/amqp/types.go | 5 ++-- tap/passive_tapper.go | 27 ++++++++++++----- tap/settings.go | 2 ++ 7 files changed, 71 insertions(+), 38 deletions(-) diff --git a/.gitignore b/.gitignore index 59674fecc..4d5c55b7e 100644 --- a/.gitignore +++ b/.gitignore @@ -26,3 +26,6 @@ build # Environment variables .env + +# pprof +pprof/* diff --git a/tap/extensions/amqp/main.go b/tap/extensions/amqp/main.go index 19b5cd913..7aca76377 100644 --- a/tap/extensions/amqp/main.go +++ b/tap/extensions/amqp/main.go @@ -83,8 +83,12 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co // We must read until we see an EOF... very important! return errors.New("AMQP EOF") } else if err != nil { - // TODO: Causes ignoring some methods. Return only in case of a certain error. But what? - return err + switch err.(type) { + case *Error: + if err.(*Error).Code == MaxHeaderFrameSizeError { + return err + } + } } switch f := frame.(type) { @@ -101,6 +105,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co case *BasicDeliver: eventBasicDeliver.Properties = header.Properties default: + frame = nil } case *BodyFrame: @@ -115,6 +120,8 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co eventBasicDeliver.Body = f.Body emitAMQP(*eventBasicDeliver, amqpRequest, basicMethodMap[60], connectionInfo, emitter) default: + body = nil + frame = nil } case *MethodFrame: @@ -200,6 +207,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co emitAMQP(*eventConnectionClose, amqpRequest, connectionMethodMap[50], connectionInfo, emitter) default: + frame = nil } diff --git a/tap/extensions/amqp/read.go b/tap/extensions/amqp/read.go index e31bb7198..c3ebac7f8 100644 --- a/tap/extensions/amqp/read.go +++ b/tap/extensions/amqp/read.go @@ -54,7 +54,7 @@ func (r *AmqpReader) ReadFrame() (frame frame, err error) { channel := binary.BigEndian.Uint16(scratch[1:3]) size := binary.BigEndian.Uint32(scratch[3:7]) - if size > 1000000 { + if size > 1000000*128 { return nil, ErrMaxSize } @@ -352,6 +352,10 @@ func (r *AmqpReader) parseHeaderFrame(channel uint16, size uint32) (frame frame, return } + if hf.Size > 512 { + return nil, ErrMaxHeaderFrameSize + } + var flags uint16 if err = binary.Read(r.R, binary.BigEndian, &flags); err != nil { @@ -439,6 +443,7 @@ func (r *AmqpReader) parseBodyFrame(channel uint16, size uint32) (frame frame, e } if _, err = io.ReadFull(r.R, bf.Body); err != nil { + bf.Body = nil return nil, err } diff --git a/tap/extensions/amqp/spec091.go b/tap/extensions/amqp/spec091.go index ea8448da4..71c501f69 100644 --- a/tap/extensions/amqp/spec091.go +++ b/tap/extensions/amqp/spec091.go @@ -19,32 +19,33 @@ import ( // ErrCredentials. The text of the error is likely more interesting than // these constants. const ( - frameMethod = 1 - frameHeader = 2 - frameBody = 3 - frameHeartbeat = 8 - frameMinSize = 4096 - frameEnd = 206 - replySuccess = 200 - ContentTooLarge = 311 - NoRoute = 312 - NoConsumers = 313 - ConnectionForced = 320 - InvalidPath = 402 - AccessRefused = 403 - NotFound = 404 - ResourceLocked = 405 - PreconditionFailed = 406 - FrameError = 501 - SyntaxError = 502 - CommandInvalid = 503 - ChannelError = 504 - UnexpectedFrame = 505 - ResourceError = 506 - NotAllowed = 530 - NotImplemented = 540 - InternalError = 541 - MaxSizeError = 551 + frameMethod = 1 + frameHeader = 2 + frameBody = 3 + frameHeartbeat = 8 + frameMinSize = 4096 + frameEnd = 206 + replySuccess = 200 + ContentTooLarge = 311 + NoRoute = 312 + NoConsumers = 313 + ConnectionForced = 320 + InvalidPath = 402 + AccessRefused = 403 + NotFound = 404 + ResourceLocked = 405 + PreconditionFailed = 406 + FrameError = 501 + SyntaxError = 502 + CommandInvalid = 503 + ChannelError = 504 + UnexpectedFrame = 505 + ResourceError = 506 + NotAllowed = 530 + NotImplemented = 540 + InternalError = 541 + MaxSizeError = 551 + MaxHeaderFrameSizeError = 552 ) func isSoftExceptionCode(code int) bool { diff --git a/tap/extensions/amqp/types.go b/tap/extensions/amqp/types.go index 78a1cf3b8..6c070b79a 100644 --- a/tap/extensions/amqp/types.go +++ b/tap/extensions/amqp/types.go @@ -60,8 +60,9 @@ var ( // ErrFieldType is returned when writing a message containing a Go type unsupported by AMQP. ErrFieldType = &Error{Code: SyntaxError, Reason: "unsupported table field type"} - // ErrClosed is returned when the channel or connection is not open - ErrMaxSize = &Error{Code: MaxSizeError, Reason: "an AMQP message cannot be bigger than 1MB"} + ErrMaxSize = &Error{Code: MaxSizeError, Reason: "an AMQP message cannot be bigger than 128MB"} + + ErrMaxHeaderFrameSize = &Error{Code: MaxHeaderFrameSizeError, Reason: "an AMQP header cannot be bigger than 512 bytes"} ) // Error captures the code and reason a channel or connection has been closed diff --git a/tap/passive_tapper.go b/tap/passive_tapper.go index 58695ed70..c4f2b0a22 100644 --- a/tap/passive_tapper.go +++ b/tap/passive_tapper.go @@ -18,6 +18,7 @@ import ( "os/signal" "runtime" "runtime/pprof" + "strconv" "strings" "sync" "time" @@ -168,11 +169,23 @@ func StartPassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem, } func startMemoryProfiler() { - dirname := "/app/pprof" - rlog.Info("Profiling is on, results will be written to %s", dirname) + dumpPath := "/app/pprof" + envDumpPath := os.Getenv(MemoryProfilingDumpPath) + if envDumpPath != "" { + dumpPath = envDumpPath + } + timeInterval := 60 + envTimeInterval := os.Getenv(MemoryProfilingTimeIntervalSeconds) + if envTimeInterval != "" { + if i, err := strconv.Atoi(envTimeInterval); err == nil { + timeInterval = i + } + } + + rlog.Info("Profiling is on, results will be written to %s", dumpPath) go func() { - if _, err := os.Stat(dirname); os.IsNotExist(err) { - if err := os.Mkdir(dirname, 0777); err != nil { + if _, err := os.Stat(dumpPath); os.IsNotExist(err) { + if err := os.Mkdir(dumpPath, 0777); err != nil { log.Fatal("could not create directory for profile: ", err) } } @@ -180,9 +193,9 @@ func startMemoryProfiler() { for true { t := time.Now() - filename := fmt.Sprintf("%s/%s__mem.prof", dirname, t.Format("15_04_05")) + filename := fmt.Sprintf("%s/%s__mem.prof", dumpPath, t.Format("15_04_05")) - rlog.Info("Writing memory profile to %s\n", filename) + rlog.Infof("Writing memory profile to %s\n", filename) f, err := os.Create(filename) if err != nil { @@ -193,7 +206,7 @@ func startMemoryProfiler() { log.Fatal("could not write memory profile: ", err) } _ = f.Close() - time.Sleep(time.Minute) + time.Sleep(time.Second * time.Duration(timeInterval)) } }() } diff --git a/tap/settings.go b/tap/settings.go index 96f12b2db..439e82e3f 100644 --- a/tap/settings.go +++ b/tap/settings.go @@ -7,6 +7,8 @@ import ( const ( MemoryProfilingEnabledEnvVarName = "MEMORY_PROFILING_ENABLED" + MemoryProfilingDumpPath = "MEMORY_PROFILING_DUMP_PATH" + MemoryProfilingTimeIntervalSeconds = "MEMORY_PROFILING_TIME_INTERVAL" MaxBufferedPagesTotalEnvVarName = "MAX_BUFFERED_PAGES_TOTAL" MaxBufferedPagesPerConnectionEnvVarName = "MAX_BUFFERED_PAGES_PER_CONNECTION" MaxBufferedPagesTotalDefaultValue = 5000