ปรับขนาดพนักงานคิวอย่างมีประสิทธิภาพด้วยเมตริก AppSignal

เว็บแอปส่วนใหญ่จะได้รับประโยชน์จากคิวในเบื้องหลัง ซึ่งมักใช้ในการประมวลผลงานด้านที่มีแนวโน้มว่าจะเกิดข้อผิดพลาดหรือใช้เวลานาน งานเบื้องหลังเหล่านี้อาจแตกต่างกันไปตั้งแต่การส่งอีเมล การอัปเดตแคช ไปจนถึงการดำเนินการตามหลักตรรกะทางธุรกิจ

เนื่องจากระบบการจัดคิวในเบื้องหลังจะปรับขนาดจำนวนงานที่จำเป็นต้องดำเนินการ กลุ่มคนงานที่ประมวลผลงานเหล่านั้นก็จำเป็นต้องปรับขนาดเช่นกัน ในกรณีที่อัตราของงานที่เข้าคิวแตกต่างกันไป การปรับจำนวนผู้ปฏิบัติงานในคิวจะกลายเป็นประเด็นสำคัญ ในการรักษาความเร็วในการประมวลผล นอกจากนี้ การลดขนาดผู้ปฏิบัติงานระหว่างปริมาณงานในคิวที่ต่ำสามารถช่วยให้ประหยัดได้มาก!

ขออภัย แบ็กเอนด์ในคิวจำนวนมากไม่ได้มาพร้อมกับตรรกะการปรับขนาดเพื่อเปิดหรือปิดผู้ปฏิบัติงาน แต่เราสามารถใช้ข้อมูลทางคณิตศาสตร์และประสิทธิภาพง่ายๆ เพื่อค้นหาจำนวนผู้ปฏิบัติงานที่เหมาะสมที่สุดโดยพิจารณาจากงานที่รออยู่ในคิว

กฎการเข้าคิว

หากงานถูกจัดคิวในอัตราที่สูงกว่าที่พนักงานคิวดำเนินการ ความลึกของคิวจะเพิ่มขึ้น และเวลาแต่ละงานที่ใช้ในคิวก็จะเพิ่มขึ้นด้วย โดยทั่วไปเราต้องการเวลารอ (ระยะเวลาใน คิว) สำหรับแต่ละงานให้ต่ำที่สุดเท่าที่จะเป็นไปได้ — ตั้งแต่ 0 วินาทีจนถึงขีดจำกัดที่ยอมรับได้ ในการประมาณจำนวนผู้ปฏิบัติงานที่ต้องใช้เพื่อให้ตรงตามเวลารอที่ต้องการ เราสามารถใช้ Queuing Rule of Thumb (QROT) โดยปกติ QROT จะแสดงเป็นความไม่เท่าเทียมกันที่อธิบายจำนวนเซิร์ฟเวอร์ที่จำเป็นสำหรับการให้บริการคิวงาน แต่รูปแบบหนึ่งสามารถเขียนได้ดังนี้:

workers = (number_of_jobs * avg_service_time_per_job) / time_to_finish_queue

ดังนั้น หากเราต้องการหาจำนวนพนักงานที่ต้องใช้บริการคิวของเราในเวลาที่ต้องการ เช่น 30 วินาที เราเพียงแค่ต้องทราบจำนวนงาน (ขนาดของคิว) และเวลาเฉลี่ยที่ใช้ไป ดำเนินการแต่ละงาน ตัวอย่างเช่น หากเรามีคิวงาน 7500 งาน และงานแต่ละงานใช้เวลาเฉลี่ย 0.3 วินาทีในการดำเนินการ เราก็สามารถจบคิวนั้นใน 30 วินาทีด้วยคนงาน 75 คน

การเข้าถึงเมตริกประสิทธิภาพ

ในการประมาณการเวลาให้บริการโดยเฉลี่ยสำหรับงานในคิว เราจำเป็นต้องเข้าถึงตัววัดประสิทธิภาพสำหรับแต่ละคลาสงาน โชคดีที่ AppSignal บันทึกข้อมูลประสิทธิภาพสำหรับแบ็กเอนด์ในคิวทั่วไปที่พร้อมใช้งานทันที บันทึกเมตริกสำหรับแต่ละครั้ง ได้ดำเนินการแล้ว

เราสามารถใช้ AppSignal GraphQL API ที่กำลังจะมีขึ้นเพื่อดูระยะเวลาเฉลี่ยของงานแต่ละประเภทในช่วง 24 ชั่วโมงที่ผ่านมา API นี้ยังไม่เป็นสาธารณะอย่างสมบูรณ์ แม้ว่าปัจจุบันจะใช้สำหรับกราฟประสิทธิภาพของ AppSignal และการแสดงข้อมูลอื่นๆ โชคดีที่ GraphQL API คือ มีวัตถุประสงค์เพื่อจัดทำเอกสารด้วยตนเอง และเราสามารถใช้เครื่องมือเช่น GraphiQL เพื่อตรวจสอบ API และค้นหาว่าข้อมูลดังกล่าวเปิดเผยวัตถุใด

ขั้นตอนการสร้างแบบสอบถาม GraphQL อยู่นอกขอบเขตของโพสต์นี้ แต่ด้านล่างเป็นตัวอย่างคลาส Ruby ที่เชื่อมต่อกับ AppSignal GraphQL API โดยใช้ไลบรารีไคลเอ็นต์ Faraday HTTP ยอดนิยมเพื่อค้นหาการรวมตัววัดพื้นฐาน

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 ชั่วโมงที่ผ่านมาของข้อมูล หากงานของเราถูกดำเนินการบ่อยกว่านั้นมาก เราอาจต้องการย่อหน้าต่างนั้นให้สั้นลง โดยชั่งน้ำหนักการดำเนินการล่าสุดใน เฉลี่ย ตัวอย่างเช่น หากเรามีงานที่ทำหลายร้อยครั้งต่อชั่วโมง เราอาจต้องการเปลี่ยน timeframe ถึงหนึ่งชั่วโมง (R1H ) เพื่อประเมินระยะเวลาของงานดังกล่าวได้ดีขึ้นหากดำเนินการในตอนนี้

โปรดทราบว่าข้อมูลประสิทธิภาพนี้แยกจากข้อมูลการใช้เซิร์ฟเวอร์ของเรา ข้อมูลนี้บอกเราว่าต้องใช้เวลานานแค่ไหนในการทำงานที่จำเป็นสำหรับแต่ละงาน ข้อมูลนี้จะเป็นประโยชน์สำหรับเราในการปรับขนาดผู้ปฏิบัติงานมากกว่าการวัดภายนอก เช่น เมตริกการใช้งาน .

ทบทวนคิว

ต่อไป เราต้องพิจารณาคิวของเราเพื่อกำหนดงานที่จะให้บริการ แบ็กเอนด์คิว 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 เป็นไปได้มากว่าคิวของเราจะมีงานจำนวนมากในคลาสเดียวกัน และเราสามารถลดค่าใช้จ่ายลงได้โดยการประมาณการดำเนินการของแต่ละคลาสเพียงครั้งเดียว

การใช้ข้อมูลประสิทธิภาพเพื่อปรับขนาด

เมื่อนำทั้งหมดนี้มารวมกัน ตอนนี้เราสามารถใช้ข้อมูลประสิทธิภาพล่าสุดของเราเพื่อปรับขนาดผู้ปฏิบัติงานคิวของเราขึ้นหรือลงตามเนื้อหาของคิวของเรา! คุณสามารถดูงานในคิวของเราและรับค่าประมาณของผู้ปฏิบัติงานที่ต้องการได้ทุกเมื่อ เพื่อให้บริการในเวลาที่กำหนด

เราจะต้องตัดสินใจกำหนดเวลาการเข้าคิวที่ต้องการ (จำนวนเวลาสูงสุดที่งานควรรอในคิว) เช่น 30 วินาที เราจะต้องระบุจำนวนผู้ปฏิบัติงานขั้นต่ำและสูงสุดด้วย การรักษาให้พนักงานทำงานอย่างน้อยหนึ่งคนทำงานอยู่ในคิว เพื่อจัดการกับงานแรกที่เข้าคิวหลังจากที่คิวว่างไปสักระยะหนึ่งจะเป็นประโยชน์อย่างยิ่ง เราจะ และต้องการจำนวนผู้ปฏิบัติงานสูงสุดด้วย เพื่อหลีกเลี่ยงการขยายขนาดการเชื่อมต่อฐานข้อมูลของเราและ/หรือค่าใช้จ่ายในการใช้งานเซิร์ฟเวอร์กับผู้ปฏิบัติงานมากเกินไป

เราสามารถสร้างชั้นเรียนเพื่อจัดการกับตรรกะนี้ได้ ซึ่งโดยพื้นฐานแล้วเป็นเพียงการนำกฎการจัดคิวไปปฏิบัติเมื่อก่อนเท่านั้น

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

เราจะต้องการปรับขนาดพนักงานของเราตามช่วงเวลาปกติ เพื่อที่เราจะได้ปรับขนาดขึ้นและลงตามความต้องการ เราสามารถสร้างงาน Rake ที่เรียก ResqueWorkerScaler . ของเราได้ คลาสเพื่อปรับขนาดผู้ปฏิบัติงาน:

# 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 เพื่อรันงาน 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 นาที ผู้ปฏิบัติงานใหม่แต่ละคนจะใช้เวลาพอสมควรในการออนไลน์และเริ่มดำเนินการงาน — มีแนวโน้มว่าจะอยู่ที่ใดก็ได้ตั้งแต่ 10-40 วินาที ขึ้นอยู่กับขนาดของฐานรหัสของเราและจำนวนอัญมณี เราใช้ ดังนั้น หากเราพยายามปรับขนาดพนักงานของเราทุกนาที เราอาจจะเพิ่มหรือลดขนาดอีกครั้งก่อนที่การเปลี่ยนแปลงที่ต้องการจะมีผล หากแอปของเราเห็นเฉพาะการใช้งานคิวที่ผันผวนในช่วงเวลาต่างๆ ของวัน เราก็ทำได้ มีแนวโน้มจะเรียกใช้งาน Rake ของเราทุกชั่วโมง แต่ถ้าขนาดคิวของเราผันผวนภายในหนึ่งชั่วโมง เราจะต้องการพิจารณาคิวของเราเป็นระยะๆ เช่น 5 นาทีข้างต้น

ขั้นตอนต่อไป

ระบบดังกล่าวที่ใช้ข้อมูลประสิทธิภาพจริงเพื่อปรับขนาดโครงสร้างพื้นฐานสามารถตอบสนองต่อความต้องการและยืดหยุ่นต่อการใช้งานที่หลากหลาย โดยเฉพาะอย่างยิ่งในสภาพแวดล้อมเช่นการประมวลผลพื้นหลัง ซึ่งตัวชี้วัดโฮสต์ เช่น การใช้หน่วยความจำและค่าเฉลี่ยโหลดไม่น่าจะแตกต่างกัน โดยใช้ตัวชี้วัดประสิทธิภาพเพื่อ มาตราส่วนเหมาะสมกว่ามาก

การใช้งานการปรับสเกลคิวสำรองสามารถวัดเวลารอเฉลี่ยต่องานแทนที่จะพิจารณาคิวทั้งหมด แต่ตัวชี้วัดนั้นอาจไม่เป็นตัวแทนเมื่อเนื้อหาคิวและขนาดเปลี่ยนแปลงอย่างรวดเร็ว หากระบบของเราประสบกับโหลดที่แปรผันอย่างกว้างขวาง โดยมีงานเข้าคิวจำนวนมากในคราวเดียว หรือเวลาในการดำเนินงานที่แปรผันอย่างกว้างขวาง ดังนั้นการพิจารณาคิวจึงเร็วกว่ามากในการตอบสนองและแก้ไขได้อย่างน่าเชื่อถือ

แต่มีข้อจำกัดบางประการที่ต้องพิจารณาในระบบการพิจารณาคิวของเรา หากคิวมีขนาดใหญ่เพียงพอ การดูงานแต่ละงานสำหรับการประเมินการดำเนินการจะช้ามาก ในกรณีเช่นนี้ หาจำนวนงานทั้งหมดได้ดีกว่า แล้วเลือก สุ่มตัวอย่างตัวแทนสุ่มของงานจากคิวและคำนวณการดำเนินการเฉลี่ยจากตัวอย่างนั้น อีกทางหนึ่งหากคลาสงานยังไม่มีข้อมูลประสิทธิภาพที่เกี่ยวข้อง เราจะต้องใช้เวลาดำเนินการตามสมมติฐานจนกว่าจะมีการดำเนินการและบันทึก ไม่กี่ครั้ง

ระบบที่สรุปไว้ข้างต้นสามารถปรับปรุงได้อย่างมากด้วยการปรับแต่งเล็กน้อย พิจารณาการประมาณเวลาดำเนินการสำหรับแต่ละคลาสงานแบบคู่ขนาน เนื่องจากค่าประมาณแต่ละรายการแยกออกมาและมีค่าเท่ากัน เรายังสามารถอัปเดตการพิจารณาคิวของเราเพื่อรวมงานเหล่านั้นที่กำลังดำเนินการอยู่ในปัจจุบันโดย เพื่อปรับปรุงความแม่นยำของการประมาณเวลาบริการทั้งหมดของเรา สำหรับสถาปัตยกรรมการประมวลผลเบื้องหลังที่มีหลายคิว เราสามารถกำหนดเวลารอที่ต้องการให้แต่ละคิว ตามลำดับความสำคัญของคิว และปรับขนาดผู้ปฏิบัติงานได้อย่างเหมาะสม

ระบบการจัดคิวมีแนวโน้มที่จะรวบรวมงานที่มีความแปรปรวนสูงจำนวนมากในโครงการใดๆ ด้วยข้อมูลประสิทธิภาพในการดำเนินการงานจากคิว เราสามารถปรับขนาดทรัพยากรอย่างมีประสิทธิภาพเพื่อให้บริการทั้งหมดที่ทำงานในลักษณะตอบสนองและมีประสิทธิภาพ

มีความสุขในการปรับขนาด!

ป.ล. หากคุณต้องการอ่านโพสต์ Ruby Magic ทันทีที่ออกจากสื่อ สมัครรับจดหมายข่าว Ruby Magic ของเราและไม่พลาดแม้แต่โพสต์เดียว!