🦔
Elasticsearchとrefreshと僕とハリネズミ

2024/12/11に公開
完全に理解したと思っていたら、実はそうじゃなくて痛い目にあったってことありませんか? 僕はよくあります。
例えば、おもちくんの気持ちを完全に理解したと思って、おなかをいじっていると怒られて針が指に刺さることがよくありました。やっぱり完全に理解することは簡単ではないです。油断はよくありませんね。


ごきげんなおもちくん

 はじめにこちらはCADDi プロダクトチーム Advent Calendar 2024の15日目の記事になります。
https://adventar.org/calendars/10554
CADDiで提供しているプロダクトには、図面の検索という重要な機能があります。この検索を支えているのが、みんな大好き Elasticsearch です。Elasticsearch、いいですよね。皆さんも使っていますか？
Elasticsearchは強力で使いやすい反面、運用にはちょっとしたコツが必要になることが多いですよね。そのため、使っている人たちは日々あれこれと悩んでいるのではないでしょうか? 僕自身も悩みは尽きません。最近の悩みはデータのサイズです。
そんなElasticsearchなので、これにまつわる知見はどれだけあっても困ることはありません。小さなことでもぜひ教えていただきたいですし、共有していきたいなーという気持ちです。 そこで、日々大量のドキュメントをインデックスしているCADDiではこんな運用しているよというところをお伝えできたらと思っています。
なお、この内容はフィクションを含むこともありますし、含まないこともあります。

 図面検索ここで言っている図面検索とは大きく2種類あって、図面の図形を使って検索するものと、図面に書かれているキーワードを検索するものがあります。前者は図形をベクトルに変換し、それをベクトル検索します。後者は図面に書かれた文字列をOCR的なもので読み取って、それをキーワードとして検索します。
ただ、図面検索はこれだけでは完結しません。図面そのものだけでなく、多数の関連情報と紐付けて検索したいというニーズがあります。例えば、受発注実績データを活用して「～円以上で受注した図面を検索」といった条件を実現するケースです。
こうした関連データを利用した検索は、RDBMSであれば別のテーブルをJOINすれば簡単に実現できます。これは「Relational（リレーショナル）」なデータベースの得意分野ですから、当然とも言えます。
ただ、Elasticsearchは検索に特化している反面このJOINという操作が少々苦手で、1:nならばパフォーマンスは落ちるものの対応はしています(join field type , nested field type)。しかし、この図面と関連データは n:mの関係(1つの図面が複数の関連データを持ち、関連データも複数の図面を含むことがある)なので、それらの方法は使えません。
この問題を解決するために、図面ドキュメントと関連データドキュメントを結びつけ、1つのドキュメントにまとめて新しいインデックスを作成しています。これにより、n:mの関係を1:nの形に変換します。（この際、関連データは複数の図面にコピーされる形になります。）
この図面と関連データを結びつけるという処理は単純に全ての図面データに対して以下を行います、
関連データの取得

関連データのインデックスを検索し、図面と関連するデータドキュメントを取得
nested documentとして追加

取得した関連データを図面データにnested documentとして追加
新しいインデックスへの登録

結びつけたデータを1つのドキュメントとして新しいインデックスに登録
そうすることで、こんな感じで検索をすると、「"SUS304"というキーワードを含み、1000円以上の発注実績のあるもの」という図面が検索できるようになります。
GET /<index>/_search
{
  "query": {
    "bool": {
      "must": [
        {
          "multi_match": {
            "query": "SUS304",
            "fields": [
              "extractedText",
              "material"
            ]
          }
        }
      ],
      "filter": [
        {
          "nested": {
            "path": "purchaseOrders",
            "query": {
              "bool": {
                "filter": {
                  "range": {
                    "purchaseOrders.price": {
                      "gte": "1000"
                    }
                  }
                }
              }
            }
          }
        }
      ]
    }
  }
}

 関連データ結びつけバッチ処理このデータを結びつける処理はバッチで毎日行われており、インデックス毎に完了したらしたらエイリアスを切り替えて新しいインデックスを使うようにしています。
ざっくり書くとこんな感じです。
for customer of customers {
  // 新しいインデックスを作る
  const created = await client.indices.create({
    index: `${customer}-${timestamp}`.
    mappings: {
      ...mappings
    },
    settings: {
      refresh_interval: -1,  
      number_of_replicas: 0,
    }
  });

  let searchAfter;
  while (true) {
    // batchSize件数の図面ドキュメントを取得する
    // (これはsearchAfterを使って、全ドキュメントを取得します)
    const result = await this.searchDrawings(searchAfter, batchSize);
    searchAfter = result.searchAfter;
    if (result.drawings.length === 0) {
      break;
    }

    const operations = [];
    for drawing of result.drawings {
      // 紐付けキーを使用して、発注実績を取得する
      const poResult = await this.searchPurchaseOrders(drawing.key);
      // 発注実績をnested documentsとして図面に追加
      drawing.purchaseOrders = poResult.purchaseOrders;

      // 紐付けキーを使用して、受注実績を取得する
      const soResult = await this.searchSalesOrders(drawing.key);
      // 受注実績をnested documentsとして図面に追加
      drawing.salesOrders = soResult.salesOrders;
                      :
                      :
      operations.push([
        { index: { "_index": created.index, "_id": drawing.id } },
        drawing
      ]);
    }

    await client.bulk({  
      index: created.index,
      operations
    });
  }

  await client.indices.updateAliases({  
    actions: [  
      { add: { index: result.index, alias: this.getAlias(customer) } },  
      { remove_index: { indices: this.currentIndex(customer) } 
    ]
  });

  client.indices.putSettings({
    index: created.index,
      settings: {
        refresh_interval: null,  
        number_of_replicas: 1,
    }
  });
}
これを見ていただいたら薄々お気づきかも知れませんが、この処理は結構時間がかかります。全てのお客様のインデックスを0から作るので、それなりの時間がかかります。前述しましたが、図形の検索ができるようにするためにベクトル用のフィールドを作っていて、これがHNSWでインデックスされます。このインデックスの作成に時間がかかることもあって、余計に長くなってしまっています。
この処理を少しでも短くしたいので、このバッチ処理ではインデックスを作るときに refresh_interval: -1とnumber_of_replicas: 0というオプションをつけています。
この部分です。
  const created = await client.indices.create({
    index: `${customer}-${timestamp}`.
    mappings: {
      ...mappings
    },
    settings: {
      refresh_interval: -1,  
      number_of_replicas: 0,
    }
  });

 replicaElasticsearchのreplicaとreplica shardのことで、データの高可用性と検索パフォーマンスを向上させるため、データの複製を作る(primaryとreplicaと呼びます)の仕組みです。検索パフォーマンスが上がるのはうれしいのですが、インデックス時はprimaryとreplicaの両方にドキュメントをインデックスする必要があるため、リソースも消費するし、時間もかかります。その結果インデックスのパフォーマンスは結構下がってしまいます。
このバッチ処理では、インデックス中はサービスで使っていないので、可用性はいりません。そのため、インデックスはreplicaなしで作って、サービスで使うようになってから1にすることで、パフォーマンスの低下を抑えています。

 refresh intervalrefresh intervalはElasticsearchのrefreshが実行される間隔を設定しています。ではrefreshとは何? と言う話なのですが、話すと長くなるので大分端折って説明すると、インデックスしたドキュメントを検索できるようにする処理です。ElasticsearchというかLuceneではドキュメントをインデックスするだけでは検索できるようにならず、インデックスを再読み込みする必要があります。ちなみに、この処理はLuceneで言うところのDirectoryReaderの"open"的なAPIに当てはまり、Elasticsearchでrefreshが実行されるとopenIfChanged が呼ばれます。
このrefreshがデフォルトでは1秒に1回実行されるため、Elasticsearchは"near real time"と呼ばれています。
refreshは軽い処理なのですが、多少なりともリソースを使用します。インデックス中のインデックス(インデックスとという言葉はややこしですね)は検索には使用していないので、少しでも負荷を下げるためにこのrefreshを無効しています。そのための設定がrefresh_interval=-1となります。

 アップグレードElasticsearchに限らず、ソフトウェアは常にアップデートされていくため、定期的なアップグレードが欠かせません。特に重要なインフラで使用しているものについては、慎重な計画と実行が求められます。
Elasticsearchでは、「1台のノードを停止 → そのノードをアップグレード → 再起動」 というローリングアップグレードを行うことで、サービスを停止することなくアップグレードを進められます。この仕組みのおかげで、稼働中のサービスに影響を与えずに更新ができ、とても助かっています。
CADDiでも定期的にElasticsearchのアップグレードを行っていますが、1つ難しい点があります。それは、前述のバッチ処理がほぼ1日中実行されていることです。そのため、アップグレードのタイミングがこの処理と重なってしまうことがあります。
Elasticsearchは「サービスを稼働したままアップグレードを行える」という利点がありますが、実際には設定によって問題が発生する場合があります。具体的には、インデックス作成中のインデックスにnumber_of_replicas=0を設定していることが原因です。
この状態でノードが再起動すると、該当するインデックスが一時的に「ステータスRED」となり、インデックス処理が継続できなくなってしまいます。その結果、バッチ処理が失敗してしまうことになります。
ですので、この問題を回避しつつアップグレードを実施するには、replicaを作った状態でインデックスを作成する必要があります。これにより、ノードが再起動しても他のノード上のreplicaが代わりに機能するため、インデックスを継続できるようになります。
そこで、先ほどのコードを以下のように変更します。
  const numberOfReplicas = updateMode ? 1 : 0;
  const created = await client.indices.create({
    index: `${customer}-${timestamp}`.
    mappings: {
      ...mappings
    },
    settings: {
      refresh_interval: -1,  
      number_of_replicas: numberOfReplicas
    }
  });
こんな感じにしてupdateModeを使用したときにはnumber_of_replicas=1にします。こうすることで、常にreplicaが存在する状態でインデックスを行うことになるので、アップグレードでノードが突然再起動したとしても、インデックスを続けられるようになります。すばらしいです。
ついでに、常にnumber_of_replicas=1なので、最後のupdateSettingsも不要になるのでスキップしましょう。
if (!updateMode) {
  client.indices.putSettings({
    index: created.index,
      settings: {
        refresh_interval: null,  
        number_of_replicas: 1,
      }
  });
}
updateMode をバージョンアップ前日に設定しておくことで安心してアップグレードを行うことができます。

 最後にこんな感じでElasticsearchの気持ちを完全に理解し、通常はインデックスを高速化しつつも必要な時には冗長性を持たせることができるようになったので、安心してアップグレードを行えました。皆さんもこんな感じのTipsがあったら教えてください。

 最後ではなかった準備しておいたオプションを設定することで、アップグレードを安全に実施できました。気分良く帰ろうとしていたその時、不穏なメッセージが飛び込んできました。
「受注実績を使った検索ができません!」
最初は「データ登録ミスかな？」と思いましたが、すぐに複数のインデックスで同じ現象が報告され、事態は一気に深刻化しました。
影響範囲が広がり、障害対策としてチームが緊急招集されました。各メンバーが状況確認と原因調査に取りかかり、僕も早速バッチ処理の実行ログを確認することにしました。しかし、ログには不自然な点が見当たりません。むしろ、処理は正常に完了しているように見えます。
そこでElasticsearchのインデックスを確認してみます。Kibanaのコンソールからこんな感じで実行すると
GET _cat/indices/index-<customer ID>*?v&h=index,health,status,docs.count,pri.store.size

index                            health  status  docs.count  pri.store.size
index-<customer ID>-<timestamp>  green   open    8257524     9.1gb
よし、ちゃんとインデックスされてます。

次に検索をしてみます。
GET /index-<customer ID>-<timestamp>/_search
{
  "query": {
    "match_all": {}
  }
}

{
  "took": 0,
  "timed_out": false,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": {
      "value": 0,
      "relation": "eq"
    },
    "max_score": null,
    "hits": []
  }
}
ん?　検索できない...　何かがおかしいかも知れません。

念のため、件数を確認してみます。
GET /index-<customer ID>-<timestamp>/_count

{
  "count": 0,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  }
}
さっきは40万件あったのに件数が0件?　どういうことでしょうか?

他のインデックスも確認してみると、同じような状態になっています。これはよろしくないですね。ジョブがうまくいってなかった可能性がありますが、ログでは問題がなかったし... と思っていると、
「index settingsおかしくないですか？」
その言葉にハッとして、早速インデックスの設定を確認してみました。
GET /index-<customer ID>-<timestamp>/_settings
{
  "index-<customer ID>-<timestamp>": {
    "settings": {
               :
        "refresh_interval": "-1",
        "number_of_replicas": "1",
               :
    }
  }
}
ああ...　この瞬間全てを理解しました。

これは全部のインデックスでこうなっていますね...
賢明な皆さんであればもうお気づきだと思いますが、「refresh_interval=-1はElasticsearchのrefreshを無効にする」と書きました。それが付いているということは、いくらドキュメントインデックスをしても、検索できる状態にはならないのです。
ついでに、常にnumber_of_replicas=1なので、最後のupdateSettingsも不要になるのでスキップしましょう。
はぁ、こんなことをした自分が恨めしい。それはそうなるよねという感じです。number_of_replicasの再設定だけではなく、refresh_intervalの再設定もスキップしてしまっていました...
幸いなことに、対処は簡単です。refreshがされていないならrefreshをすれば良いだけです。
GET /_refresh

{
  "_shards": {
    "total": xxxx,
    "successful": xxxx,
    "failed": 0
  }
}
これでもう一度 _countしてみると
GET /index-<customer ID>-<timestamp>/_count

{
  "count": 325078,
  "_shards": {
    "total": 1,
    "successful": 1,
    "skipped": 0,
    "failed": 0
  }
}
直った!
こうして全インデックスが無事に復旧しました。検索を試してみたところ、期待通りレスポンスも返ってきています。障害対応は無事完了です。
影響範囲ですが、通常の図面の検索は別のインデックスで行われており、問題のあったインデックスは受注実績・発注実績と組み合わせて検索した時のみ利用されるものであったため、被害は最小限にとどめられたのが不幸中の幸いでした(個人の感想です、関係者の皆様ご迷惑をおかけしました)。
このような問題を引き起こしてしまい、申し訳ない気持ちでいっぱいなのですが、ポジティブな面としてはElasticsearchのアップグレードは安全に行えるようになったということで自分を慰めています。
完全に理解することは簡単ではないです。油断はよくありませんね。皆さんもくれぐれも油断しないようにしてください。油断をすると事故が起こります。
https://www.youtube.com/watch?v=27oGrDi-Ce4

 参考までにちなみに、なんで/_cat/indicesで正しい件数を返して/_countで件数を返さないのかと気になりませんか?
まず/_cat/indicesはインデックスのstatsを読んでいて、そのstatsはLuceneのセグメントから読み取られています。これはなんというか物理的な数という感じです。検索ができなくても関係ありません。
一方/_countは内部的にはsearch をしています。そのため、refreshが無効になっているので、検索できずその結果0件になっていたということになります。
落ち着いて考えればそうなるよね、って感じなのですが、障害対応などをしているとなかなか気づかないものです。
さらに直った後も件数が違うのが気になりませんか?
nested documentはこんな形で親のドキュメントの一部になってます。
{
             :
  "nested_field": [
    { "key": "value1"},
    { "key": "value2"},
    { "key": "value3"}
  ]
             :
}
ですが、インデックスされるときは、全て独立したドキュメントとしてインデックスされます。つまり上のドキュメントの場合、親ドキュメントも含めて4つのドキュメントが作られます。検索実行時には子ドキュメントがフィルタされて親ドキュメントしか検索結果に含まないようになっています。そのため、検索を使っている_countでは親ドキュメントの件数しか返ってこないため数が少なく、_cat/indicesではいわゆる物理的な数、つまり全てのドキュメントの数をカウントしているため数が多くなっているというわけです。今回の場合は、_countが約32万件で/_cat/indicesが約820万件なので、1図面あたり平均25個程度の関連データがつけられていることになります。
いろいろ秘密がありますね。
はじめに

図面検索

関連データ結びつけバッチ処理

replica

refresh interval

アップグレード

最後に

最後ではなかった

参考までに

Discussion
