Effortlessly Scale Queue Workers with AppSignal Metrics for Peak Performance

Web apps ส่วนใหญ่ได้ประโยชน์จาก background queue ใช้ประมวลผล error-prone หรือ time-consuming side jobs Background jobs เหล่านี้หลากหลายจากส่ง emails, อัปเดต caches, ไปจนถึง core business logic

เมื่อ background queueing system scale จำนวน jobs ที่ต้องประมวลผล pool ของ workers ที่ประมวลผล jobs ต้อง scale ด้วย ในกรณีที่ rate ของ jobs ที่ enqueued varies การ scale จำนวน queue workers ขึ้นเป็น key aspect รักษา processing speed นอกจากนี้ scale down workers ระหว่าง low queue throughput ให้ significant savings!

น่าเสียดายหลาย queueing backends ไม่มา equipped กับ scaling logic เปิด/ปิด workers แต่เราใช้ simple math และ performance data หา optimal worker count จาก work ที่รอใน queue

Queueing Rule of Thumb (QROT)

ถ้า jobs enqueued rate สูงกว่า processed rate โดย queue workers ความลึกของ queue จะโต และเวลาที่แต่ละ job ใช้ใน queue ก็โต ปกติอยากให้ wait time (เวลาใน queue) ต่ำที่สุด — จาก 0 วินาที ถึง acceptable limit เพื่อประมาณจำนวน workers ที่จำเป็น satisfy desired wait time ใช้ Queueing Rule of Thumb (QROT) ปกติ QROT expressed as inequality อธิบายจำนวน servers ที่จำเป็น service queue of jobs แต่รูปแบบหนึ่งเขียนได้ว่า:

workers = (number_of_jobs * avg_service_time_per_job) / time_to_finish_queue 

ดังนั้น ถ้าอยากรู้จำนวน workers required service queue ใน desired time เช่น 30 seconds ต้องรู้ number of jobs (queue size) และ average time execute แต่ละ job ตัวอย่าง มี queue 7500 jobs แต่ละ job ใช้ average 0.3 seconds สามารถจบ queue ใน 30 seconds ด้วย 75 workers

เข้าถึง Performance Metrics

เพื่อประมาณ average service time สำหรับ jobs ใน queue ต้องเข้าถึง performance metrics สำหรับแต่ละ job class โชคดี AppSignal record performance data สำหรับ common queueing backends out-of-the-box บันทึก metrics ทุกครั้งที่ job executed

ใช้ AppSignal GraphQL API (กำลังจะมาถึง) ดึง average duration ของแต่ละ job type ตลับ 24 ชั่วโมง API นี้ยังไม่ fully public แตใช้สำหรับ Performance graphs และ data displays ของ AppSignal โชคดี GraphQL APIs intended self-documenting ใช้ tool เช่น GraphiQL introspect API หา data objects ที่ expose

Process สร้าง GraphQL query นอก scope ของโพสต์ แต่ด้านล่างคือ example Ruby class connect ไป AppSignal GraphQL API ด้วย Faraday HTTP client library query สำหรับ basic metrics aggregation

require 'json'
require 'faraday'

class AppsignalClient
  BASE_URL = 'https://appsignal.com/'
  DEFAULT_APP_ID = ENV['APPSIGNAL_APP_ID']
  DEFAULT_TOKEN = ENV['APPSIGNAL_API_TOKEN']

  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

  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)
    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

ด้วย class นี้ได้ average job duration สำหรับ ActiveJob class ใดๆ คืนค่าเป็น milliseconds:

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

โดย default เรียก average transaction duration ของ job ตลับ 24 ชั่วโมงล่าสุด ถ้า jobs execute บ่อยกว่านั้น อาจ shorten window ให้น้ำหนัก recent executions มากกว่าใน average เช่น jobs รัน hundreds of times ต่อชั่วโมง เปลี่ยน timeframe เป็นหนึ่งชั่วโมง (R1H) เพื่อ estimate duration ของ job หนึ่งถ้ารันทันที

Note: performance data นี้แยกจาก server utilization data Data นี้บอกว่า actual work ของแต่ละ job ใช้เวลานานแค่ไหน จะ useful สำหรับ scale workers มากกว่า external measurements เช่น utilization metrics

Introspecting the Queue

ต่อไป introspect queue หา jobs ที่ต้อง service Ruby queueing backend ยอดนิยมคือ Resque integrate ดีกับ ActiveJob เข้าถึง enqueued jobs สำหรับ queue ใน Resque แล้วประมาณ execution time สำหรับแต่ละ job จาก class ใช้ AppsignalClient class ข้างต้น

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

  def decode_activejob_args(job)
    decoded_job = job
    decoded_job = Resque.decode(job) if job.is_a? String
    decoded_job['args'].first
  end
end

ใช้ class นี้ง่าย:

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

ใช้ simple memoization ของ job durations ใน estimate_job_duration หลีกเลี่ยง duplicate calls ไป AppSignal API queue ส่วนใหญ่มี jobs class เดียวกันเยอะ ลด overhead โดย estimate execution ของแต่ละ class ครั้งเดียว

ใช้ Performance Data Scale Workers

รวมทั้งหมด ตอนนี้ใช้ recent performance data scale queue workers up หรือ down ตาม content ของ queue! ทุกเมื่อ มอง jobs ใน queue ได้ estimate workers required service ใน desired time limit

ต้อง decide desired queueing time limit (max เวลา job ควร wait ใน queue) เช่น 30 seconds ต้องระบุ minimum และ maximum worker count ช่วยเก็บอย่างน้อยหนึ่ง worker running สำหรับ queue handle first job(s) enqueued หลัง queue ว่างไปพักหนึ่ง ต้องการ maximum worker count หลีกเลี่ยง over-scaling database connections และ/หรือ server utilization costs ด้วย workers มากเกินไป

สร้าง class handle logic นี้ basically implementation ของ 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

อยาก scale workers เป็นระยะสม่ำเสมอ เพื่อ scale up/down ตาม demand สร้าง Rake task เรียก ResqueWorkerScaler scale workers:

# inside lib/tasks/resque_workers.rake
namespace :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 job รัน Rake task นี้เป็นระยะ:

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

ตั้ง scaling task รันทุก 5 นาที worker ใหม่แต่ละตัวใช้เวลา come online และเริ่ม process jobs — 10-40 seconds ขึ้นกับ codebase size และ gems ถ้าพยายาม scale ทุกนาที อาจ scale up/down อีกก่อน changes take effect ถ้าแอปเห็น queue usage fluctuate ต่างเวลาในวัน เรียก Rake task ทุกชั่วโมงได้ แตถ้า queue size fluctuate ภายในชั่วโมง introspect queue บ่อยขึ้น เช่น 5 นาทีข้างต้น

Next Steps

ระบบที่ actual performance data ใช้ scale infrastructure responsive to demand และ resilient to varied usage โดยเฉพาะใน environment เช่น background processing ที่ host metrics เช่น memory usage และ load average ไม่ค่อย vary การใช้ performance metrics scale เหมาะสมกว่า

Alternate queue scaling implementations อาจ measure mean wait time per job แทน introspect full queue แต metric นั้น unrepresentative เมื่อ queue contents และ size เปลี่ยนเร็ว ถ้าระบบมี widely variable load jobs enqueue มากพร้อมกัน หรือ widely variable job execution times queue introspection เร็วกว่า respond และ reliably correct

แต่มี limitations ใน queue introspection system ถ้า queue ใหญ่พอ looking at each job สำหรับ execution estimate จะ prohibitively slow ในกรณีดีกว่า หา total job count แล้ว select random representative sampling ของ jobs จาก queue คำนวณ average execution จาก sample หรือถ้า job class ไม่มี performance data ยัง ต้องใช้ assumed execution time จนกว่า executed และ recorded 몇 ครั้ง

System outlined ข้างต้น improve significantly ด้วย tweaks นิดๆ พิจารณา estimate execution time สำหรับแต่ละ job class แบบ parallel เพราะแต่ละ estimation isolated และ idempotent อัปเดต queue introspection รวม jobs ที่ worker กำลัง execute เพื่อ improve accuracy ของ total service time estimate สำหรับ background processing architecture หลาย queues assign แต่ละ queue desired wait time ตาม queue priority และ scale workers appropriately

Queueing systems tend to collect highly variable work ในโปรเจกต์ ด้วย performance data บน execution ของ jobs จาก queue สามารถ scale resources service ทั้งหมดอย่าง responsive และ efficient

Happy scaling!

P.S. หากอยากอ่าน Ruby Magic posts ทันทีที่เผยแพร่ สมัครสมาชิก Ruby Magic newsletter และไม่พลาดโพสต์เด็ดๆ!