AppSignalのメトリクスを活用してキューワーカーを効率的にスケーリングする方法

多くのWebアプリケーションは、バックグラウンドキューを導入することで大きなメリットを得られます。バックグラウンドジョブは、エラーが発生しやすい処理や時間のかかるサイドタスクを捌くために広く使われており、その用途はメール送信からキャッシュの更新、さらには中核となるビジネスロジックの実行まで多岐にわたります。

バックグラウンドキューイングシステムが処理すべきジョブ数を増やしていくにつれ、それらを処理するワーカーのプールも合わせてスケールさせる必要があります。ジョブのエンキュー率が変動する環境では、処理速度を維持するためにワーカー数を増やすことが重要なポイントになります。さらに、キューのスループットが低い時間帯にワーカーを減らせば、大幅なコスト削減にもつながります。

残念ながら、多くのキューイングバックエンドには、ワーカーをオン/オフ切り替えするスケーリングロジックが備わっていません。しかし、シンプルな計算とパフォーマンスデータを組み合わせれば、キューで待機している仕事量に基づいて最適なワーカー数を導き出すことができます。

キューイングの経験則（QROT）

ジョブがキューワーカーの処理速度よりも速いペースでエンキューされると、キューの深さは増加し続け、各ジョブがキュー内で過ごす時間も伸びていきます。一般的に、各ジョブの待ち時間（キュー内での滞在時間）は、0秒から許容できる上限まで、可能な限り短くしたいものです。

目標とする待ち時間を満たすために必要なワーカー数を見積もるには、「キューイングの経験則」（Queueing Rule of Thumb：QROT）を利用できます。QROTは通常、ジョブのキューを捌くために必要なサーバー数を表す不等式として示されますが、次のような形式で書くこともできます。

workers = (number_of_jobs * avg_service_time_per_job) / time_to_finish_queue

つまり、たとえば「30秒」という目標時間内にキューを処理し切るために必要なワーカー数を求めたい場合、必要な情報はキュー内のジョブ数（キューのサイズ）と、各ジョブの平均実行時間だけです。例を挙げると、7,500件のジョブがキューにあり、各ジョブの実行に平均0.3秒かかる場合、75個のワーカーがあれば30秒でこのキューを処理し終えられます。

パフォーマンスメトリクスへのアクセス

キュー内のジョブの平均サービス時間を見積もるには、ジョブクラスごとのパフォーマンスメトリクスにアクセスできる必要があります。幸い、AppSignalは主要なキューイングバックエンドのパフォーマンスデータを標準機能として記録しており、ジョブが実行されるたびにメトリクスを収集します。

近日公開予定のAppSignal GraphQL APIを使えば、直近24時間における各ジョブタイプの平均実行時間を取得できます。このAPIはまだ完全には公開されていませんが、現在はAppSignalのパフォーマンスグラフなどのデータ表示に利用されています。GraphQL APIは自己文書化されるように設計されているため、GraphiQLのようなツールを使ってAPIをイントロスペクションすれば、どのようなデータオブジェクトが公開されているかを調べられます。

GraphQLクエリの構築手順自体は本稿の範囲外ですが、以下は定番のHTTPクライアントライブラリ「Faraday」を使ってAppSignal GraphQL APIに接続し、基本的なメトリクス集計を照会するRubyクラスの例です。

require 'json'
require 'faraday'
 
class AppsignalClient
  BASE_URL = 'https://appsignal.com/'
  DEFAULT_APP_ID = ENV['APPSIGNAL_APP_ID']
  DEFAULT_TOKEN = ENV['APPSIGNAL_API_TOKEN']
  # GraphQL query to fetch the "mean" metric for the selected app.
  METRICS_QUERY = <<~GRAPHQL.freeze
    query($appId: String!, $query: [MetricAggregation!]!, $timeframe: TimeframeEnum!) {
      app(id: $appId) {
        metrics {
          list(timeframe: $timeframe, query: $query) {
            start
            end
            rows {
              fields {
                key
                value
              }
            }
          }
        }
      }
    }
  GRAPHQL
 
  def initialize(app_id: DEFAULT_APP_ID, client_secret: DEFAULT_TOKEN)
    @app_id = app_id
    @client_secret = client_secret
  end
 
  # Fetch the average duration for a job class's perform action
  # Default timeframe is last 24 hours
  def average_job_duration(job_class, timeframe: 'R24H')
    response =
      connection.post(
        'graphql',
        JSON.dump(
          query: METRICS_QUERY,
          variables: {
            appId: @app_id,
            timeframe: timeframe,
            query: [
              name: 'transaction_duration',
              headerType: 'legacy',
              tags: [
                { key: 'namespace', value: 'background' },
                { key: 'action', value: "#{job_class.name}#perform" },
              ],
              fields: [{ field: 'MEAN', aggregate: 'AVG' }],
            ],
          }
        )
      )
    data = JSON.parse(response.body, symbolize_names: true)
    rows = data.dig(:data, :app, :metrics, :list, :rows)
    # There may be no metrics in the selected timeframe
    return 0.0 if rows.empty?
 
    rows.first[:fields].first[:value]
  end
 
  private
 
  def connection
    @connection ||= Faraday.new(
      url: BASE_URL,
      params: { token: @client_secret },
      headers: { 'Content-Type' => 'application/json' },
      request: { timeout: 10 }
    ) do |faraday|
      faraday.response :raise_error
      faraday.adapter Faraday.default_adapter
    end
  end
end

このクラスを使えば、指定したActiveJobクラスの平均実行時間をミリ秒単位で取得できます。

AppsignalClient.new.average_job_duration(MyMailerJob)
# => 233.1

デフォルトでは、直近24時間分のデータに基づくジョブの平均トランザクション時間を取得します。ジョブの実行頻度がそれよりはるかに高い場合は、集計期間を短縮して、最近の実行結果により大きな重みを置くとよいでしょう。たとえば、1時間に数百回実行されるジョブがあるなら、timeframeを1時間（R1H）に変更することで、「今まさに実行した場合」にかかる時間をより正確に見積もれます。

なお、このパフォーマンスデータはサーバー利用率のデータとは別物である点に注意してください。このデータが示すのは、各ジョブに必要な作業を実際に完了するまでの所要時間です。ワーカーのスケーリングにおいては、CPU使用率のような外部指標よりもはるかに有用な情報となります。

キューのイントロスペクション

次に、処理すべきジョブを特定するために、キューそのものを調査（イントロスペクション）する必要があります。Rubyの代表的なキューイングバックエンドのひとつがResqueで、ActiveJobともスムーズに統合できます。Resqueでは特定のキューにエンキュー済みのジョブへアクセスできるため、前述のAppsignalClientクラスを使って、ジョブのクラスに基づいた実行時間を見積もることが可能です。

require 'resque'
 
class ResqueEstimator
  def initialize(queue: 'default')
    @queue = queue
    @cache = {}
    @appsignal_client = AppsignalClient.new
  end
 
  def enqueued_duration_estimate
    Resque.data_store.everything_in_queue(queue).map do |job|
      estimate_job_duration decode_activejob_args(job)
    end.sum
  end
 
  def estimate_job_duration(job)
    @cache[job['job_class']] ||= @appsignal_client
                                 .average_job_duration job['job_class']
  end
 
  private
 
  # ActiveJob-specific method for parsing job arguments
  # for ActiveJob+Resque integration
  def decode_activejob_args(job)
    decoded_job = job
    decoded_job = Resque.decode(job) if job.is_a? String
    decoded_job['args'].first
  end
end

このクラスの使い方は非常にシンプルです。

ResqueEstimator.new(queue: 'my_queue').enqueued_duration_estimate
# => 23000 (ms)

estimate_job_durationメソッドでは、ジョブの実行時間をシンプルにメモ化している点に注目してください。これはAppSignal APIへの重複した呼び出しを避けるためです。キューには同じクラスのジョブが多数含まれているのが普通なので、クラスごとの実行時間は1回だけ見積もればよく、これによりオーバーヘッドを大幅に抑えられます。

パフォーマンスデータを使ったスケーリング

ここまでの要素をすべて組み合わせると、キューの中身に基づいてワーカーをスケールアップ／スケールダウンするために、最新のパフォーマンスデータを活用できるようになります。いつでもキュー内のジョブを確認し、目標の制限時間内に処理を完了するために必要なワーカー数を見積もれるわけです。

まず、目標とするキュー待ち時間の上限（どのジョブもキュー内で待機できる最大時間。例：30秒）を決める必要があります。あわせて、ワーカー数の最小値と最大値も指定しましょう。キューがしばらく空だった後に最初のジョブが投入されたときに備えて、少なくとも1つのワーカーを常に稼働させておくのが有効です。同時に最大ワーカー数も設定しましょう。ワーカーが多すぎると、データベース接続数やサーバー利用コストが必要以上に膨らんでしまうためです。

このロジックを担うクラスを作成しましょう。その内容は、基本的に前述のキューイングの経験則（QROT）の実装そのものです。

class ResqueWorkerScaler
  def initialize(queue: 'default', workers_range: 1..100, desired_wait_ms: 300_000)
    @queue = queue
    @workers_range = workers_range
    @desired_wait_ms = desired_wait_ms
    @estimator = ResqueEstimator.new(queue: @queue)
  end
 
  def desired_workers
    total_time_ms = @estimator.enqueued_duration_estimate
    workers_required = [(total_time_ms / desired_wait_ms).ceil, workers_range.last].min
    [workers_required, workers_range.first].max
  end
 
  def scale
    # using platform-specific scaling interface, scale to desired_workers
  end
end

需要に応じてスケールアップ／ダウンできるよう、ワーキングは定期的に実行したいところです。ResqueWorkerScalerクラスを呼び出してワーカーをスケールさせるRakeタスクを作成できます。

# inside lib/tasks/resque_workers.rake
 
namespaace :resque_workers do
  desc 'Scale worker pool based on enqueued jobs'
  task :scale, [:queue] => [:environment] do |_t, args|
    queue = args[:queue] || 'default'
    ResqueWorkerScaler.new(queue: queue).scale
  end
end

そして、cronジョブを設定して、このRakeタスクを定期的に実行します。

*/5 * * * * /path/to/our/rake resque_workers:scale
# scale a non-default queue:
*/5 * * * * /path/to/our/rake resque_workers:scale['my_queue']

ここで、スケーリングタスクを5分ごとに実行するよう設定している点に注目してください。新しく起動したワーカーがオンラインになりジョブの処理を開始するまでには、ある程度の時間がかかります。コードベースの規模や依存gemの数にもよりますが、おそらく10〜40秒程度でしょう。そのため、1分ごとにスケーリングしようとすると、前回の変更が反映される前に再びスケールアップ／ダウンしてしまう可能性が高くなります。

アプリのキュー使用量が1日の中の時間帯によってのみ変動するのであれば、Rakeタスクを1時間ごとに実行するだけで十分でしょう。一方、1時間以内でもキューのサイズが変動する場合は、上記の例のように5分間隔など、より短い周期でキューを調査する必要があります。

次のステップ

実際のパフォーマンスデータを使ってインフラをスケールさせるこの種の仕組みは、需要の変化に素早く反応でき、多様な使用状況にも強靭に対応できます。特にバックグラウンド処理の環境では、メモリ使用量やロードアベレージといったホストメトリクスはほとんど変動しないため、パフォーマンスメトリクスに基づくスケーリングの方がはるかに適切です。

代替的なキュースケーリングの実装として、キュー全体を調査する代わりにジョブごとの平均待ち時間を測定する方法もあります。しかしこの指標は、キューの中身やサイズが急激に変化する場面では実態を正しく反映しないことがあります。大量のジョブが一斉にエンキューされたり、ジョブの実行時間が大きくばらついたりするような、変動の激しい負荷を扱うシステムでは、キューのイントロスペクションの方がはるかに迅速に反応し、確実に正確な結果をもたらします。

一方で、キューイントロスペクション方式には考慮すべき限界もあります。キューが非常に大きい場合、すべてのジョブを1件ずつ調べて実行時間を見積もるのは非現実的なほど遅くなります。そのようなケースでは、まずジョブ総数を取得したうえで、キューからランダムな代表サンプルを抜き出し、そのサンプルから平均実行時間を算出する方が賢明です。また、ジョブクラスにまだパフォーマンスデータが存在しない場合は、数回実行されて記録が蓄積されるまでの間、想定される実行時間を代わりに使う必要があります。

さらに、上記の仕組みはいくつかの工夫によって大きく改善できます。各見積もり処理は互いに独立して冪等なので、ジョブクラスごとの実行時間の見積もりを並列実行することを検討してみましょう。また、合計サービス時間の推定精度を高めるために、ワーカーが現在実行中のジョブもキュー調査の対象に含めるとよいでしょう。複数のキューを持つバックグラウンド処理アーキテクチャでは、キューの優先度に応じてキューごとに目標待ち時間を割り当て、それぞれ適切にワーカーをスケールさせることも可能です。

キューイングシステムは、プロジェクト内の変動の激しい作業を数多く集約しがちです。キューのジョブ実行に関するパフォーマンスデータがあれば、そうした作業すべてを迅速かつ効率的に処理できるよう、リソースを効果的にスケールできます。

それでは、快適なスケーリングライフを！

P.S. Ruby Magicの記事を公開と同時にお読みになりたい方は、ぜひRuby Magicニュースレターを購読してください。記事を見逃すことはもうありません！