Hầu hết web apps hưởng lợi từ background queue, thường dùng xử lý error-prone hoặc time-consuming side jobs. Background jobs có thể từ sending emails, updating caches, đến performing core business logic.
Khi bất kỳ background queueing system scale số jobs cần xử lý, pool of workers xử lý jobs đó cũng cần scale. Trong cases rate of jobs enqueued varies, scaling số queue workers lên trở thành key aspect duy trì processing speed. Ngoài ra, scaling down workers khi queue throughput thấp cung cấp significant savings!
Thật tiếc, nhiều queueing backends không equipped với scaling logic để bật/tắt workers. Nhưng ta dùng simple math và performance data tìm optimal worker count dựa trên work waiting trong queue.
Queueing Rule of Thumb
Nếu jobs enqueued rate cao hơn rate processed bởi queue workers, depth của queue sẽ grow và time mỗi job wait trong queue cũng grow. Tổng quát, muốn wait time (thời gian trong queue) cho mỗi job càng thấp càng tốt — từ 0 giây đến acceptable limit. Để estimate số workers required satisfy desired wait time, dùng Queueing Rule of Thumb (QROT). Thường QROT biểu diễn as inequality mô tả số servers required service queue of jobs, nhưng một form viết:
workers = (number_of_jobs * avg_service_time_per_job) / time_to_finish_queue Vì vậy, nếu muốn figure out số workers required service queue trong desired time, nói 30 giây, chỉ cần biết số jobs (queue size) và average time execute mỗi job. Ví dụ, queue 7500 jobs, mỗi job trung bình 0.3 giây execute, finish queue trong 30 giây với 75 workers.
Truy Cập Performance Metrics
Để estimate average service time cho jobs trong queue, cần access performance metrics cho mỗi job class. May mắn, AppSignal record performance data cho common queueing backends out-of-the-box, recording metrics mỗi lần job executed.
Ta dùng upcoming AppSignal GraphQL API lấy average duration của mỗi job type trong 24h qua. API này chưa fully public, dù đang dùng cho Performance graphs và data displays của AppSignal. May mắn, GraphQL APIs intended self-documenting, và dùng tool như GraphiQL introspect API tìm data objects nó expose.
Process building GraphQL query ngoài scope bài này, nhưng dưới là example Ruby class connect đến AppSignal GraphQL API dùng popular 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']
# 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
endVới class này, lấy average job duration cho ActiveJob class, trả về milliseconds:
AppsignalClient.new.average_job_duration(MyMailerJob) # => 233.1Mặc định, call average transaction duration của job trong 24h data. Nếu job(s) executed nhiều hơn, muốn shorten window, weigh recent executions hơn. Ví dụ, jobs chạy hundreds of times an hour, muốn change timeframe thành one hour (R1H) để better estimate duration job nếu executed ngay bây giờ.
Lưu ý performance data này separate từ server utilization data. Data này cho biết actual time để làm work required cho mỗi job. Hữu ích hơn scaling workers hơn external measurements như utilization metrics.
Introspecting Queue
Tiếp, cần introspect queue determine jobs to be serviced. Common Ruby queueing backend là Resque, integrates nicely với ActiveJob. Có thể access enqueued jobs cho given queue trong Resque và estimate execution time cho mỗi job based on class, dùng AppsignalClient class từ trên.
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
endDùng class này đơn giản:
ResqueEstimator.new(queue: 'my_queue').enqueued_duration_estimate # => 23000 (ms)Chú ý dùng simple memoization của job durations trong estimate_job_duration để tránh duplicate calls AppSignal API. Hầu như queue chứa nhiều jobs cùng class và reduce overhead bằng cách estimate execution mỗi class một lần.
Dùng Performance Data Để Scale
Kết hợp tất cả, giờ dùng recent performance data scale queue workers up hoặc down based on queue content! Bất kỳ lúc nào, look at jobs in queue và get estimate workers required service nó trong desired time limit.
Cần decide desired queueing time limit (max time job wait in queue), ví dụ 30 seconds. Cũng specify minimum và maximum worker count. Hữu ích keep ít nhất one worker running cho queue, handle first job(s) enqueued sau queue empty một lúc. Cũng muốn maximum worker count, tránh over-scaling database connections và/hoặc server utilization costs với quá nhiều workers.
Tạo class handle logic này, basically implementation của Queueing Rule of Thumb từ trước.
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
endMuốn scale workers on regular interval để scale up/down based on demand. Tạo Rake task gọi 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
endSetup cron job chạy Rake task này regular interval:
*/5 * * * * /path/to/our/rake resque_workers:scale # scale a non-default queue:
*/5 * * * * /path/to/our/rake resque_workers:scale['my_queue'] Chú ý set scaling task chạy every 5 minutes. Mỗi new worker mất time come online và begin processing jobs — likely 10-40 seconds, depending on codebase size và số gems. Vì vậy, nếu scale workers every minute, likely scale up/down again trước khi desired changes take effect. Nếu app chỉ thấy queue usage fluctuate different times of day, có thể call Rake task hourly. Nhưng nếu queue size fluctuate within hour, muốn introspect queue frequent interval, như 5 phút.
Next Steps
Hệ thống dùng actual performance data scale infrastructure có thể very responsive to demand và resilient to varied usage. Đặc biệt trong environment như background processing, host metrics như memory usage và load average unlikely vary, dùng performance metrics scale much more appropriate.
Alternate queue scaling implementations có thể measure mean wait time per job thay vì introspect full queue, nhưng metric đó unrepresentative khi queue contents và size change rapidly. Nếu system experiences widely variable load, lots of jobs enqueueing at once, hoặc widely variable job execution times, queue introspection much faster respond và reliably correct.
Nhưng có limitations trong queue introspection system. Nếu queue sufficiently large, look at each job execution estimate prohibitively slow. Trong cases đó, better find total job count, select random representative sampling jobs từ queue và calculate average execution từ sample. Alternately, nếu job class chưa performance data, cần assumed execution time cho đến khi executed và recorded vài lần.
Hệ thống outlined above cũng improved significantly với few tweaks. Cân nhắc estimate execution time cho mỗi job class in parallel, vì mỗi estimation isolated và idempotent. Cũng update queue introspection include jobs currently being executed by worker để improve accuracy total service time estimate. Cho background processing architecture với multiple queues, assign mỗi queue desired wait time based on queue priority, scale workers appropriately.
Queueing systems tend to collect highly variable work trong project. Với performance data trên execution của jobs từ queue, có thể effectively scale resources service all that work responsive, efficient manner.
Happy scaling!
P.S. Nếu muốn đọc bài Ruby Magic ngay khi xuất bản, đăng ký newsletter Ruby Magic để không bỏ lỡ!