自社のデータ収集システムが驚くほどコンパクトになった話

# はじめに
こんにちは。オークファンのエンジニアで、今年の10月からマネージャー(仮)になった@tmotです。マネージメントという新しいフィールドにも挑戦し始めているこの頃です。

マネージメントとは関係ないですが、自社のデータ収集システムの改修に結構苦労したので、今回はその話をしたいと思います。

# バッチに問題発生
今年の初めあたりから、弊社で長らく使われていたデータ収集システムが、ときどきOutOfMemoryErrorで失敗するようになってしまいました。
弊社サービスで提供しているデータの根幹となるバッチであり、このエラーを放置していると弊社サービスの大部分に影響を及ぼすため、早急な対応が求められました。

## バッチの詳細
データを収集し、Elasticsearchに保存するバッチです。データをなるべく漏れなく取得するため、1日2回実行しています。
AWS EC2インスタンス(`i3.4xlarge`)上で稼働しており、実行環境はJava11です。

# 暫定対応
まずはバッチを継続して動かしておくことが最優先なので、以下の暫定対応を施しました。
- 割り当てるJVMメモリ量を`96G`から`108G`に引き上げ
    - EC2インスタンスのメモリ量が122GiB(≒130GB)だったので、さほど大きく上げることは出来なさそうでした。
- 並列で処理している箇所の並列数を削減
    - メモリ使用量の抑制はできますが、バッチ稼働時間が延びます。本バッチは12時間以内には終わらせる必要があるため、あまり極端な削減はできません。

今までの経験では上記の対応で収まることがほとんどだったのですが、今回はそうはいかず、OutOfMemoryErrorを封殺することができませんでした。
結局、十分なメモリ量を確保するためにEC2インスタンスを`r6g.8xlarge`(メモリ 256GiB)まで引き上げ、バッチプログラムに割り当てるメモリ量を`192G`にすることで、ひとまず解消しました。

# 調査
今までよりも取得対象となるデータが増えてしまったのが原因だと思われますが、そもそもメモリを100GBも食い潰すような作りになっているのがおかしいのではないかということで、プログラムの中身を調査し始めました。

## コードの整理
コードの内容を整理し、メモリを無駄に使っていそうな箇所を探りました。
このバッチは取得データの一定の分類ごとに並列でスレッドを作り、その分類に属するデータを取得してElasticsearchに保存しています。
取得した情報を格納するMapを各スレッドに持っており、件数が多い分類の場合はページネーションを駆使して何回かに分けてMapに書き込み、全て取得したタイミングでElasticsearchへの書き込み処理へと送るようになっていました。
この挙動のため、件数の多いスレッドが複数走っていると、各スレッド内のMapが肥大化していき、大幅にメモリを食ってしまう可能性があると推測しました。

```plantuml
@startuml
title 分類ごとのデータ取得処理

participant "Tasklet" as tasklet
participant "スレッドのExecutor" as executor
entity "取得済みデータ保持Map" as itemMap
participant "保存処理の管理クラス" as itemManager
database "弊社の\nElasticsearch" as cm2
database "外部API" as yahooApi

activate tasklet

tasklet -> executor: スレッドにデータ取得処理を渡す
activate executor

create itemMap
executor -> itemMap: データを保持するMapを作成
activate itemMap
itemMap --> executor

loop 全件取得するまで15000件ずつページネーションしてループ

 executor -> yahooApi: 外部APIから情報を取得
 activate yahooApi
 yahooApi --> executor: 情報を返却
 deactivate yahooApi
 executor -> itemMap: Mapに登録されていない情報を登録
 itemMap --> executor
end loop

executor -> itemManager: マップに登録されたデータを保存処理に送る
activate itemManager
itemMap -> itemManager
deactivate itemMap
ref over itemManager, cm2
Elasticsearchへの保存処理(別スレッドで実行されている)
end ref
itemManager --> executor
deactivate itemManager

executor --> tasklet: 取得件数を返却
deactivate executor

@enduml
```

## IntelliJ Profilerによる監視
続いて、プログラム実行中にメモリが多く使われている箇所を解析してみることにしました。

開発で使用しているIDEのIntelliJには、Javaプログラム実施時にProfilerをアタッチする機能があり、挙動を解析することができます。実行オプションの`Run 'Application名' with 'IntelliJ Profiler'`でバッチを走らせました。

適当なタイミングでスナップショットを取って内容を確認してみると、`Biggest Objects`の項目の上位には、データ取得メソッド(`fetchItemLists()`)内のLinkedHashMapと、Elasticsearchへの保存メソッド(`saveItems()`)のLinkedListが並んでいました。`fetchItemLists()`で取得した全件のMapを一括で`saveItems()`に送るため、Mapのサイズが`saveItems()`のListのサイズにも影響を及ぼします。

![スクリーンショット 2023-12-12 11.49.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/536819/07c40986-cd44-ab72-967c-006a9a720e5c.png)

予想通り、Mapの肥大化がメモリへの影響が大きいことが確認できました。

# 改修
スレッドごとに保持しているMapについて、全件取得してからやっとElasticsearchに書き込み処理に送るのではなく、一定件数取得ごとにElasticsearchへの書き込み処理に送り、Mapをまっさらにするよう修正しました。
これでMapの肥大化を防ぐことができます。

Mapは同じデータを重複して書き込まないようにする役割もありましたが、15000件ごとにMapをclearすると以前のループで取得済みのデータがわからなくなります。それを防ぐため、ループごとにデータのIDだけ持ち越しできるように、新たにSetを準備しました。このSetは重複検知のためにStringのIDのみ保持するので、データの中身を全てオブジェクトとして保持するよりもメモリを食わずに済むはずです。

```plantuml
@startuml
title 分類ごとのデータ取得処理 (改)

participant "Tasklet" as tasklet
participant "スレッドのExecutor" as executor
entity "取得済みID保持Set" as itemSet
entity "データ保持Map" as itemMap
participant "保存処理の管理クラス" as itemManager
database "弊社の\nElasticsearch" as cm2
database "外部API" as yahooApi

activate tasklet

tasklet -> executor: スレッドにデータ取得処理を渡す
activate executor


create itemSet
executor -> itemSet: **すでに取得されたデータのIDを保持するSetを作成**
activate itemSet
itemSet --> executor
create itemMap
executor -> itemMap: データを保持するMapを作成
activate itemMap
itemMap --> executor

loop 全件取得するまで15000件ずつページネーションしてループ
 executor -> yahooApi: 外部APIから情報を取得
 activate yahooApi
 yahooApi --> executor: 情報を返却
 deactivate yahooApi
 executor -> itemSet: すでに取得されているIDを検索
 itemSet --> executor
 executor -> itemMap: Setにまだ登録されていない情報を登録
 itemMap --> executor
 executor -> itemSet: Mapに登録したデータのIDを登録
 itemSet --> executor

 opt Mapに15000件以上登録された場合に実行
  executor -> itemManager: **Mapに登録されたデータを保存処理に送る**
  activate itemManager
  itemMap -> itemManager
  ref over itemManager, cm2
   Elasticsearchへの保存処理(別スレッドで実行されている)
  end ref
  itemManager --> executor
  deactivate itemManager
  executor -> itemMap: **Mapをclear**
  itemMap --> executor
 end opt
end loop

deactivate itemSet
executor -> itemManager: ループ終了後、Mapに残ったデータを保存処理に送る
activate itemManager
itemMap -> itemManager
deactivate itemMap
ref over itemManager, cm2
 Elasticsearchへの保存処理(別スレッドで実行されている)
end ref
itemManager --> executor
deactivate itemManager
executor --> tasklet: 取得件数を返却
deactivate executor

@enduml
```

# 改修後の確認

## 動作確認
改修によって挙動がおかしくなっていないかを検証しました。

ステージング環境上で改修前後のバッチプログラムをそれぞれ実行し、取得件数が同程度で、抜け漏れが増えていなさそうであることを確認しました。

## メモリ使用量の確認
改修前後でピーク時のメモリ使用量が変わっているかを確認しました。

最初はIntelliJのProfilerで試そうとしたのですが、ピークとなるタイミングでスナップショットを撮るのが難しかったので断念しました。
バッチの進捗に張り付いていなくてもピークのメモリ使用量を確認したかったので、GCログから確認するという手段に落ち着きました。

バッチ実行時に`-Xlog:gc*:file={gcログの出力先}`のJVMオプションをつけてgcログを出力するようにし、`GCeasy`の[REST API](https://blog.gceasy.io/2016/06/18/garbage-collection-log-analysis-api/)を利用してサマリーをJSON形式にまとめました。

特定の分類に絞ってバッチを実行した結果が以下の通りです。

- 改修前
``` json
{
    "isProblem": false,
    "jvmHeapSize": {
        "youngGen": {
            "allocatedSize": "5.55 gb",
            "peakSize": "5.17 gb"
        },
        "oldGen": {
            "allocatedSize": "10.22 gb",
            "peakSize": "11.64 gb"
        },
        "metaSpace": {
            "allocatedSize": "85.25 mb",
            "peakSize": "83.55 mb"
        },
        "total": {
            "allocatedSize": "15.77 gb",
            "peakSize": "14.98 gb"
        }
    },
    "gcStatistics": {
        "totalCreatedBytes": "1.35 tb",
        "totalPromotedSize": "79.77 gb",
        "measurementDuration": "6 hrs 4 min 45 sec",
        "avgAllocationRate": "64.52 mb/sec",
        "avgPromotionRate": "3.73 mb/sec",
        "minorGCCount": "3041",
        "minorGCTotalTime": "4 min 50 sec 192 ms",
        "minorGCAvgTime": "95.4 ms",
        "minorGCAvgTimeStdDeviation": "361 ms",
        "minorGCMinTIme": "0",
        "minorGCMaxTime": "3 sec 200 ms",
        "minorGCReclaimedSize": "3.86 tb",
        "minorGCIntervalAvgTime": "7 sec 185 ms",
        "fullGCCount": "0"
    },
    ...
}
```

- 改修後
```json
{
    "isProblem": false,
    "jvmHeapSize": {
        "youngGen": {
            "allocatedSize": "4.36 gb",
            "peakSize": "4.17 gb"
        },
        "oldGen": {
            "allocatedSize": "4.69 gb",
            "peakSize": "5.09 gb"
        },
        "metaSpace": {
            "allocatedSize": "85.75 mb",
            "peakSize": "83.58 mb"
        },
        "total": {
            "allocatedSize": "9.05 gb",
            "peakSize": "8.26 gb"
        }
    },
    "gcStatistics": {
        "totalCreatedBytes": "2.19 tb",
        "totalPromotedSize": "69.08 gb",
        "measurementDuration": "5 hrs 9 min 53 sec",
        "avgAllocationRate": "123.51 mb/sec",
        "avgPromotionRate": "3.8 mb/sec",
        "minorGCCount": "2933",
        "minorGCTotalTime": "2 min 19 sec 515 ms",
        "minorGCAvgTime": "47.6 ms",
        "minorGCAvgTimeStdDeviation": "62.3 ms",
        "minorGCMinTIme": "0",
        "minorGCMaxTime": "406 ms",
        "minorGCReclaimedSize": "3.56 tb",
        "minorGCIntervalAvgTime": "6 sec 315 ms",
        "fullGCCount": "0"
    },
    ...
}
```

改修後の結果の方が、`jvmHeapSize`が抑えられているようです。

# リリース
データの整合性とメモリ使用量の削減が確認できたので、ようやくリリースです。

## バッチプログラムのリリース
まずは以前立ち上げた`r6g.8xlarge`のEC2インスタンスに改修後プログラムを配置し、１週間程度様子見をしました。

## 割り当てJVMメモリの削減
続いて、EC2インスタンスは変更せず、バッチプログラムに割り当てるJVMメモリ量を削減していきました。

以下の順序で、１週間ずつ様子を見ながら徐々に削減していきました。

- 192G (リリース直後)
- 160G
- 128G
- 108G
- 96G
- 80G
- 64G
- 48G
- 32G
- 24G

最終的に、24Gでも問題なく動くことが確認できました！

以前OutOfMemoryErrorが起きてしまったときの割り当て量(108G)の4分の一以下です。改修の効果は大きかったですね。

## EC2インスタンスのダウングレード
最後に、`r6g.8xlarge`まで上げてしまったEC2インスタンスを、よりコンパクトなものへと入れ替えていきます。

### インスタンスの選定
入れ替え先のインスタンスを選定しました。メモリ割り当ては24GBなので、インスタンスのメモリは32GiBで十分だと判断しました。

メモリ32GiBのものでも、vCPUなどのスペックを考慮していくつか選択肢を洗い出しました。

|  | コスト [USD/h] | vCPU | メモリ [GiB] | ストレージ | ネットワーク帯域幅 [Gbps] | EBS 帯域幅 [Gbps] | 備考 |
| -- | -- | -- | -- | -- | -- | -- | -- |
| **`r6g.8xlarge`** | **1.9456** | **32** | **256** | **EBSのみ** | **12** | **9** | **現在のインスタンス。<br>Graviton 2系。** |
| `c6g.4xlarge` | 0.6848 | 16 | 32 | EBSのみ | 最大10 | 4.75 | Graviton 2系。<br>コンピューティング最適化。 |
| `c7g.4xlarge` | 0.7277 | 16 | 32 | EBSのみ | 最大15 | 最大10 | Graviton 3系。<br>c6gよりもネットワーク帯域が強化。 |
| `m6g.2xlarge` | 0.396 | 8 | 32 | EBSのみ | 最大10 | 4.75 | Graviton 2系。<br>一般用途向け。 |
| `m7g.2xlarge` | 0.4216 | 8 | 32 | EBSのみ | 最大15 | 最大10 | Graviton 3系。<br>m6gよりもネットワーク帯域が強化。 |
| `r6g.xlarge` | 0.2432 | 4 | 32 | EBSのみ | 最大10 | 4.75 | Graviton 2系。<br>メモリ最適化。 |
| `r7g.xlarge` | 0.2584 | 4 | 32 | EBSのみ | 最大12.5 | 最大10 | Graviton 3系。<br>r6gよりもネットワーク帯域が強化。 |

Graviton 3系はGraviton 2系に比べてネットワーク帯域が強化されているので、可能であればGraviton 3系の方が望ましいでしょう。

### インスタンスの入れ替え実験
先に挙げたインスタンスタイプについて、以下の制約を考慮しながらダウングレードしていきました。
- 実行時間が12時間以内に収まる必要がある。
- Graviton 3系のインスタンスで動かしたことがないので、まずGraviton 3系で動作することを確認する。

```plantuml
@startuml
start
:c7g.4xlargeで実験;
if (Graviton 3系でも動作する) then (yes)
 if (実行時間が12時間以内に収まっている) then (yes)
  :m7g.2xlargeで実験;
  if (実行時間が12時間以内に収まっている) then (yes)
   :r7g.xlargeで実験;
   if (実行時間が12時間以内に収まっている) then (yes)
    :r7g.xlargeに決定;
    stop
   else (no)
    :m7g.2xlargeに決定;
    stop
   endif
  else (no)
   :c7g.4xlargeに決定;
   stop
  endif
 else (no)
  :インスタンスタイプ再検討;
  stop
 endif
else (no)
 :c6g.4xlargeで実験;
 if (実行時間が12時間以内に収まっている) then (yes)
  :m6g.2xlargeで実験;
  if (実行時間が12時間以内に収まっている) then (yes)
   :r6g.xlargeで実験;
   if (実行時間が12時間以内に収まっている) then (yes)
    :r6g.xlargeに決定;
    stop
   else (no)
    :m6g.2xlargeに決定;
    stop
   endif
  else (no)
   :c6g.4xlargeに決定;
   stop
  endif
 else (no)
  :インスタンスタイプ再検討;
  stop
 endif
endif
  
@enduml
```

こちらも１週間ずつ様子見をしながら入れ替えていき、最終的に`r7g.xlarge`までダウングレードすることができました。
`r6g.8xlarge`の8分の一程度のコストに抑えられます！問題が発生する前の`i3.4xlarge`と比べても、6分の一程度ですね。

# おわりに
少々乱暴な暫定対応によって一時的にインスタンスコストは上がってしまいましたが、最終的にかなりコストを抑えることができてよかったです。調査や検証にかなり時間をかけたので、達成感もひとしおです。
昔作られたシステムには、「効率がよくないがとりあえず動く」ので放置されていた処理などもあるので、その辺にも手を入れられると良いなと思います。

gcログサマリーの`jvmHeapSize`以外の部分についてはあまり解析できていないので、それ以外の部分に何か気になる点や問題など発見された場合はご教示いただければ幸いです！