High Memory Usage in Kubernetes (1)

Addressing Challenges During Large File Uploads

8 min readNov 2, 2024

背景

最近接手公司四五年的舊專案踩到一個 K8S 的雷想記錄一下，先描述一下背景，我們的服務整套服務架在客戶的 K8S 內，服務內有隻 API 會進行上傳加密壓縮檔並做後續處理。

整個 API 的流程大概如下，由於該 API 是以非同步進行，所以在第一次解壓縮後的流程都是以非同步 Job 的方式進行。

Service 收到 File 利用 Golang package 進行解密解壓縮 -> 
解壓縮檔裡面還會有一份壓縮檔，會利用 7Zip 進行第二次的解壓縮 -> 
最後會取得一份大檔，在透過 Linux 的 Split 指令切分為小檔案 -> 
依序對切分完的檔案做後續的處理

壓縮檔的大小約 3GB，進行完兩次解壓縮後的做後大小約 10 GB

在客戶環境發現 API 使用時 Pod 的 WSS 會飆到約 10 GB

測試環境與工具

由於給地端的服務測試環境沒有真正的 K8S 所以開一台 EC2 利用 Kind 建立 K8S 進行測試

檔案：壓縮檔約 700MB 解開後約 3.xGB
fluentBit：利用 node exporter 抓取 instance metrics
prometheus：蒐集 node exporter & go process metrics
pprof：分析與定位資源消耗位置

由於 pprof 只能抓到觀測當下的指標，因此比較難評估抓取的時間點，所以採用程式碼埋點的方式進行，總共埋了四個位置。

Controller 收到 Request 的時間點
Controller 準備進行 response 時
非同步 Job 未進行任何動作時
非同步 Job 結束時

尚未修改

go process metrics

可以觀察到服務在高峰時本身使用了約 2GB 的記憶體

pprof

pprof 觀測結果

middleware request_log 使用到了 io.ReadAll 佔用大量記憶體
heap2 與 heap3 結果相同
heap4 並無觀察到有大量記憶體佔用的情況
根據上述兩點推斷，在非同步 Job 執行時並無明顯的記憶體資源佔用

檢查後發現先前在記錄 request body 時因為沒有過濾掉檔案上傳的部分，因此會在記憶體複製一份一模一樣的 body，導致 middleware 佔用與壓縮檔相同大小的記憶體。

修改 Middleware 後

go process metrics

高峰下降到了 1.37 GB 左右

pprof

pprof 觀測結果

request_log 佔用大量記憶體情況改善
heap2 與 heap3 在 ZipCryptoDecryptor 有佔用大量記憶體情況
heap4 並無觀察到有大量記憶體佔用的情況

ZipCryptoDecryptor

由於 Golang 官方的 Zip Package 本身沒有支援解密的功能，因此使用了 https://github.com/yeka/zip 的套件進行解密解壓縮

爬了這個 source code 後發現 ZipCryptoDecryptor 會建立一個與檔案大小一樣的 slice 當作暫存，因此會消耗大量的記憶體空間

func ZipCryptoDecryptor(r *io.SectionReader, password []byte) (*io.SectionReader, error) {
 z := NewZipCrypto(password)
 b := make([]byte, r.Size())

 r.Read(b)

 m := z.Decrypt(b)
 return io.NewSectionReader(bytes.NewReader(m), 12, int64(len(m))), nil
}

修改後改用 chunk 的方式進行解密，並將結果透過 io.Pipe 的方式回傳，不使用 memory 當作暫存，詳細實作方法可參考我的 github

type OffsetReader struct {
 r      io.Reader
 offset int64
 read   int64
}

func NewOffsetReader(r io.Reader, offset int64) *OffsetReader {
 return &OffsetReader{
  r:      r,
  offset: offset,
 }
}

func (o *OffsetReader) Read(p []byte) (n int, err error) {
 if o.read < o.offset {
  skip := o.offset - o.read
  var discarded int64
  for skip > 0 {
   discarded, err = io.CopyN(io.Discard, o.r, skip)
   if err != nil {
    return 0, err
   }
   o.read += discarded
   skip -= discarded
  }
 }

 n, err = o.r.Read(p)
 if err != nil {
  o.read += int64(n)
 }
 return n, err
}

func ZipCryptoDecryptor(r *io.SectionReader, password []byte) (io.Reader, error) {
 z := NewZipCrypto(password)
 pr, pw := io.Pipe()

 go func() {
  defer pw.Close()
  err := z.DecryptStream(pw, r)
  if err != nil {
   pw.CloseWithError(err)
  }
 }()

 or := NewOffsetReader(pr, 12)

 return or, nil
}

func (z *ZipCrypto) DecryptStream(w io.Writer, r io.Reader) error {
 buffer := make([]byte, 4096) // 4KB Buffer size
 for {
  n, err := r.Read(buffer)
  if n > 0 {
   //Decrypt the chunk
   decryptedChunk := z.Decrypt(buffer[:n])

   _, writerErr := w.Write(decryptedChunk)
   if writerErr != nil {
    return writerErr
   }
  }
  if err == io.EOF {
   break
  } else if err != nil {
   return err
  }

 }
 return nil
}