sql_exporter × Prometheus で実現する Solid Queue の滞留監視と可視化 [DeNA インフラ SRE]

はじめに

IT 基盤部第二グループの無津呂です。

私たちのチームがインフラ運用を担っているプロダクトにおいて、ジョブキューシステムを Solid Queue に移行しています。

キューの安定運用のためにはスループット可視化や滞留監視が不可欠です。

本記事では、sql_exporter と Prometheus / Grafana を用いたキューのスループット可視化および滞留監視の仕組みについてご紹介します。

Solid Queue の構成と監視の目的

本題に入る前に、Solid Queue の構成と監視の目的について整理します。

我々の環境では、アプリケーションサーバ (app) やバッチ処理 (batch) からキューにジョブを投入し、常駐プロセス (worker) がキューからジョブを取り出して処理しています。 Solid Queue のバックエンドとして MySQL を使用しており、ジョブの実体はデータベース上のレコードです。

ジョブは実行待ち (ready) や実行中 (claimed)、失敗 (failed) などのステータスを持ち、ステータスに応じたテーブル群を遷移する仕組みになっています。（ステータスは他にも、指定時刻になるまで待機する scheduled、同時実行数の制御によってブロックされている blocked、定期実行用の recurring などが存在します）

Solid Queue の構成

このような仕組みにおいて「キューの滞留」とは、ジョブが何らかの要因で worker に消費されず、ステータスが変化しないままテーブルに残り続けている異常な状態を指します。

キューの滞留イメージ

ジョブの処理遅延はエンドユーザの体験を損ねたり、サービス全体の詰まりにつながるため、滞留監視は重要です。また、各ステータスのジョブの数やスループットを常に可視化しておくことで、異常にいち早く気付けるだけでなく、worker の最適なキャパシティを判断するための情報としても活用できます。

こうした背景から、キューの滞留監視および可視化を実現する監視システムの構築に取り組みました。次章では、具体的な構築要件について整理します。

監視要件と技術選定

監視システムを構築するにあたり、以下の 2 つの要件を重視しました。

アプリケーションから独立した監視システム
Solid Queue のキュー監視では、Yabeda などの Ruby 向け計装ライブラリを用いてプロセス内からメトリクスを出力する手法が一般的です。しかし、この手法では worker プロセス自体がダウン・フリーズした際にメトリクスの出力も途絶えてしまい、肝心の滞留を正しく検知できないリスクがあります。そのため、worker の死活にかかわらずキューの状態を監視し続けられるよう、アプリケーション層には一切手を入れない構成としました。
既存の監視基盤 (Prometheus / Grafana) の活用
すでに本環境で運用実績のある Prometheus と Grafana をベースにすることで、新たなツールの運用コストを増やさず、アラート設定やダッシュボード構築を行えるようにしました。

これらを踏まえた最適なアプローチとして、アプリケーション (worker) を介さず、データベースから直接キューの状態を読み取り、Prometheus に渡す方式に決定しました。

データベースからキューの状態を読み取るためのツールとしては、任意の SQL クエリの実行結果を Prometheus 形式のメトリクスとして公開できる sql_exporter を採用しました。

次章では、構築した監視システムの全体像を紹介します。

監視システムの全体像

データベース (MySQL) に対し、sql_exporter から直接クエリを発行してメトリクスを抽出し、Prometheus で収集します。そして Grafana でキューの状態を可視化しつつ、Prometheus でキューの滞留監視を行います。

監視システムの全体像

以下、sql_exporter によるメトリクス化、Prometheus と Grafana による収集・可視化、Prometheus による滞留監視の方法を順に説明します。

Step 1: sql_exporter によるメトリクス化

収集メトリクスと SQL クエリの設計

キューの状態を把握するために、3 つのメトリクスを設計しました。

各メトリクスの目的とデータベースからの抽出方法を説明します。

(1) 各ステータスのジョブ数

現在のキューの状況を可視化するために、ステータスごとのジョブ数を知る必要があります。

Solid Queue はステータスごとにテーブルが分かれているため、以下のような SQL で各テーブルのレコード数をカウントすることでジョブ数を取得できます。より正確には、キュー名・ジョブ優先度・ステータスごとにジョブ数を集計していますが、これは後述する滞留監視用のメトリクスと同じ粒度に揃えるためです。

SELECT queue_name, priority, 'scheduled' AS status, COUNT(*) AS job_count FROM solid_queue_scheduled_executions GROUP BY queue_name, priority
UNION ALL
SELECT queue_name, priority, 'blocked' AS status, COUNT(*) AS job_count FROM solid_queue_blocked_executions GROUP BY queue_name, priority
UNION ALL
SELECT queue_name, priority, 'ready' AS status, COUNT(*) AS job_count FROM solid_queue_ready_executions GROUP BY queue_name, priority
UNION ALL
SELECT j.queue_name, j.priority, 'claimed' AS status, COUNT(*) AS job_count FROM solid_queue_claimed_executions ce JOIN solid_queue_jobs j ON ce.job_id = j.id GROUP BY j.queue_name, j.priority
UNION ALL
SELECT j.queue_name, j.priority, 'failed' AS status, COUNT(*) AS job_count FROM solid_queue_failed_executions fe JOIN solid_queue_jobs j ON fe.job_id = j.id GROUP BY j.queue_name, j.priority
UNION ALL
SELECT j.queue_name, j.priority, 'recurring' AS status, COUNT(*) AS job_count FROM solid_queue_recurring_executions re JOIN solid_queue_jobs j ON re.job_id = j.id GROUP BY j.queue_name, j.priority

上記のクエリの結果を、sql_exporter によって以下のような Prometheus 形式のメトリクスに変換します。

solid_queue_job_count{queue_name="default",priority="10",status="ready"} 5
solid_queue_job_count{queue_name="default",priority="20",status="claimed"} 2
solid_queue_job_count{queue_name="default",priority="30",status="failed"} 1
...

このメトリクスは瞬間的な個数しか示さないため過度な信頼は禁物ですが、キューの大まかな状況を把握するのには使えます。

(2) 最古ジョブのタイムスタンプ

今回の主目的である滞留監視のためには、各ステータスにおいて「もっとも古くから待たされているジョブが、どれくらいの時間滞留しているか」を計測する必要があります。さらに、各ジョブには処理の優先度を割り当てており、優先度ごとに滞留監視の閾値を変える運用にしたいため、優先度ごとに滞留時間を把握する必要もあります。

ここでは、キュー名・ジョブ優先度・ステータスごとの「最古ジョブのタイムスタンプ（Unix 時間）」を取得するメトリクスを設計しました。（滞留時間を直接メトリクス化することも可能ですが、滞留時間を表現する際に Prometheus 側で計算した方がより自然なため、タイムスタンプを取得する方式にしています）

SQL でタイムスタンプを抽出する際は、各ステータスの特性に合わせて以下のようにカラムを使い分けます。

ready / claimed / failed / blocked : created_at（ジョブが各状態に遷移・作成された時刻）
scheduled : scheduled_at（ジョブが実行されるべき予定時刻）
recurring : run_at（定期タスクが実行されるべき予定時刻）

SELECT queue_name, priority, 'scheduled' AS status, UNIX_TIMESTAMP(MIN(scheduled_at)) AS oldest_ts FROM solid_queue_scheduled_executions GROUP BY queue_name, priority
UNION ALL
SELECT queue_name, priority, 'blocked' AS status, UNIX_TIMESTAMP(MIN(created_at)) AS oldest_ts FROM solid_queue_blocked_executions GROUP BY queue_name, priority
UNION ALL
SELECT queue_name, priority, 'ready' AS status, UNIX_TIMESTAMP(MIN(created_at)) AS oldest_ts FROM solid_queue_ready_executions GROUP BY queue_name, priority
UNION ALL
SELECT j.queue_name, j.priority, 'claimed' AS status, UNIX_TIMESTAMP(MIN(ce.created_at)) AS oldest_ts FROM solid_queue_claimed_executions ce JOIN solid_queue_jobs j ON ce.job_id = j.id GROUP BY j.queue_name, j.priority
UNION ALL
SELECT j.queue_name, j.priority, 'failed' AS status, UNIX_TIMESTAMP(MIN(fe.created_at)) AS oldest_ts FROM solid_queue_failed_executions fe JOIN solid_queue_jobs j ON fe.job_id = j.id GROUP BY j.queue_name, j.priority
UNION ALL
SELECT j.queue_name, j.priority, 'recurring' AS status, UNIX_TIMESTAMP(MIN(re.run_at)) AS oldest_ts FROM solid_queue_recurring_executions re JOIN solid_queue_jobs j ON re.job_id = j.id GROUP BY j.queue_name, j.priority

上記のクエリの結果を、sql_exporter によって以下のような Prometheus 形式のメトリクスに変換します。

solid_queue_oldest_job_seconds{queue_name="default",priority="10",status="ready"} 1782864000
solid_queue_oldest_job_seconds{queue_name="default",priority="20",status="claimed"} 1782864010
solid_queue_oldest_job_seconds{queue_name="default",priority="30",status="failed"} 1782864060
...

(3) スループット

キューのスループット（単位時間あたりの enqueue / dequeue されたジョブ数）を可視化するためのメトリクスも必要です。

我々の運用では完了ジョブを即物理削除しているため、「完了したジョブ数」を直接カウントして dequeue 速度を割り出すことは困難でした。 enqueue 速度についても、同様の理由から算出が難しいことがわかりました。

そこで視点を変え、MySQL の performance_schema に着目しました。 performance_schema の table_io_waits_summary_by_table にはテーブルごとの累積 INSERT / DELETE 回数が統計データとして格納されています。これを取得することで、キューや優先度別の細分化はできないものの、各ステータス（テーブル）単位での大まかなスループットを把握できるようにしました。

SELECT OBJECT_NAME AS `table`, COUNT_INSERT AS `insert`, COUNT_DELETE AS `delete`
FROM performance_schema.table_io_waits_summary_by_table
WHERE OBJECT_SCHEMA = DATABASE()

上記のクエリの結果を、sql_exporter によって以下のような Prometheus 形式のメトリクスに変換します。

solid_queue_global_io_counts{table="solid_queue_ready_executions",operation="insert"} 1000
solid_queue_global_io_counts{table="solid_queue_ready_executions",operation="delete"} 800
solid_queue_global_io_counts{table="solid_queue_claimed_executions",operation="insert"} 800
...

sql_exporter の設定ファイル

設計した 3 つのメトリクスを、以下のように sql_exporter の設定ファイルに落とし込みます。

jobs で接続先データベースを設定し、collectors 内の metrics で先ほど解説した各クエリとラベルのマッピングを定義しています。

global:
  scrape_timeout: 10s
  scrape_timeout_offset: 500ms
  max_connections: 3
  max_idle_connections: 3

jobs:
  - job_name: solid_queue
    collectors: [ solid_queue ]
    static_configs:
      - targets:
          solid_queue_dev: 'mysql://user:${PASSWORD}@host:3306/solid_queue_dev'

collectors:
  - collector_name: solid_queue
    metrics:
      - metric_name: solid_queue_job_count
        type: gauge
        help: 'Current number of jobs'
        key_labels: [queue_name, priority, status]
        values: [job_count]
        query: |
          SELECT queue_name, priority, 'scheduled' AS status, COUNT(*) AS job_count FROM solid_queue_scheduled_executions GROUP BY queue_name, priority
          UNION ALL
          SELECT queue_name, priority, 'blocked' AS status, COUNT(*) AS job_count FROM solid_queue_blocked_executions GROUP BY queue_name, priority
          UNION ALL
          SELECT queue_name, priority, 'ready' AS status, COUNT(*) AS job_count FROM solid_queue_ready_executions GROUP BY queue_name, priority
          UNION ALL
          SELECT j.queue_name, j.priority, 'claimed' AS status, COUNT(*) AS job_count FROM solid_queue_claimed_executions ce JOIN solid_queue_jobs j ON ce.job_id = j.id GROUP BY j.queue_name, j.priority
          UNION ALL
          SELECT j.queue_name, j.priority, 'failed' AS status, COUNT(*) AS job_count FROM solid_queue_failed_executions fe JOIN solid_queue_jobs j ON fe.job_id = j.id GROUP BY j.queue_name, j.priority
          UNION ALL
          SELECT j.queue_name, j.priority, 'recurring' AS status, COUNT(*) AS job_count FROM solid_queue_recurring_executions re JOIN solid_queue_jobs j ON re.job_id = j.id GROUP BY j.queue_name, j.priority

      - metric_name: solid_queue_oldest_job_seconds
        type: gauge
        help: 'Unix timestamp of the oldest unfinished job'
        key_labels: [queue_name, priority, status]
        values: [oldest_ts]
        query: |
          SELECT queue_name, priority, 'scheduled' AS status, UNIX_TIMESTAMP(MIN(scheduled_at)) AS oldest_ts FROM solid_queue_scheduled_executions GROUP BY queue_name, priority
          UNION ALL
          SELECT queue_name, priority, 'blocked' AS status, UNIX_TIMESTAMP(MIN(created_at)) AS oldest_ts FROM solid_queue_blocked_executions GROUP BY queue_name, priority
          UNION ALL
          SELECT queue_name, priority, 'ready' AS status, UNIX_TIMESTAMP(MIN(created_at)) AS oldest_ts FROM solid_queue_ready_executions GROUP BY queue_name, priority
          UNION ALL
          SELECT j.queue_name, j.priority, 'claimed' AS status, UNIX_TIMESTAMP(MIN(ce.created_at)) AS oldest_ts FROM solid_queue_claimed_executions ce JOIN solid_queue_jobs j ON ce.job_id = j.id GROUP BY j.queue_name, j.priority
          UNION ALL
          SELECT j.queue_name, j.priority, 'failed' AS status, UNIX_TIMESTAMP(MIN(fe.created_at)) AS oldest_ts FROM solid_queue_failed_executions fe JOIN solid_queue_jobs j ON fe.job_id = j.id GROUP BY j.queue_name, j.priority
          UNION ALL
          SELECT j.queue_name, j.priority, 'recurring' AS status, UNIX_TIMESTAMP(MIN(re.run_at)) AS oldest_ts FROM solid_queue_recurring_executions re JOIN solid_queue_jobs j ON re.job_id = j.id GROUP BY j.queue_name, j.priority

      - metric_name: solid_queue_global_io_counts
        type: counter
        help: 'Total number of table I/O operations'
        key_labels: [table]
        value_label: operation
        values: [insert, delete]
        query: |
          SELECT OBJECT_NAME AS `table`, COUNT_INSERT AS `insert`, COUNT_DELETE AS `delete`
          FROM performance_schema.table_io_waits_summary_by_table
          WHERE OBJECT_SCHEMA = DATABASE()

Step 2: Prometheus と Grafana による収集・可視化

Prometheus 側では、sql_exporter の /metrics エンドポイントを scrape する設定を追加しています。

Grafana では以下のようなダッシュボードでキューの状態を可視化しています。

ジョブ数と滞留時間のグラフ

default という名前のキューについて、ジョブの個数（左）と滞留時間（右）をそれぞれ表示しています。

ジョブ数と滞留時間のグラフ

スループットのグラフ

solid_queue_global_io_counts メトリクスから得られる各テーブルの INSERT / DELETE 累積回数を、PromQL の rate() 関数で秒間の INSERT / DELETE 平均回数に変換し、近似的なスループットとして表示しています。

スループットのグラフ

enqueue / dequeue 速度を知りたいときは、主に以下に着目するのが良いでしょう。

enqueue 速度: ready テーブルの INSERT 速度
dequeue 速度（処理開始時点）: claimed テーブルの INSERT 速度、または ready テーブルの DELETE 速度
dequeue 速度（ジョブ完了時点）: jobs テーブル（全ジョブが共通で入っているテーブル）の DELETE 速度

なお、claimed テーブルのジョブは、対応する jobs テーブル内のジョブの削除に伴ってカスケード削除されるようになっています。そのため claimed テーブルの delete 速度は表面上はゼロに見え、dequeue 速度の指標としては利用できません。

Step 3: Prometheus による滞留監視

滞留監視の閾値

移行前のキューシステムでは、ジョブの種類ごとに個別の閾値を決めて滞留監視を行っていました。個別の閾値管理は細かな調整が効く反面、十数個あるジョブの種類数だけ閾値を管理せねばならず、多少の運用負荷がかかるという課題がありました。

Solid Queue では個別の閾値管理を廃止し、ジョブの優先度に基づく汎用的な閾値運用にしました。ジョブの優先度は High / Normal / Low の 3 段階で、優先度とステータスに応じて滞留監視の閾値を以下のように設定します。（表内の数値はあくまで我々の環境・要件に合わせた一例です）

ステータス	High (`priority=10`)	Normal (`priority=20`)	Low (`priority=30`)
blocked / ready / claimed	1 分	15 分	6 時間
scheduled	1 分	1 分	1 分
failed / recurring	10 分	10 分	10 分

アラートルール実装

Prometheus のアラートルールでは、PromQL の time() 関数と solid_queue_oldest_job_seconds メトリクスの差分を用いてジョブの滞留時間（秒）を評価します。

たとえば、ステータス ready, 優先度 High（閾値 1 分）についてのルールはこのように書きます。

rules:
  - alert: SolidQueueReadyJobWait_PriorityHigh
    expr: |
      time() - solid_queue_oldest_job_seconds{status="ready", priority="10"}
      > 60
    for: 1m
    labels:
      severity: critical