Class: Wurk::Metrics::Query

Inherits:
Object
  • Object
show all
Extended by:
DashboardSeries
Defined in:
lib/wurk/metrics/query.rb

Overview

Read side of the per-class buckets Wurk::Metrics::History writes.

The instance API is Sidekiq's Sidekiq::Metrics::Query (spec docs/target/sidekiq-free.md §20): Query.new(now:).top_jobs / #for_job return a Result whose job_results[klass] is a JobResult with totals / series / hist. Window caps clamp silently, as upstream.

The class-level functions (Metrics::DashboardSeries) back Wurk's dashboard JSON API.

Defined Under Namespace

Classes: JobResult, MarkResult, Result, WindowTooWide

Constant Summary collapse

MAX_MINUTES = 
480
MAX_HOURS =

Class-level dashboard cap: the 10-minute buckets live MID_TERM (30 days), so a 720-hour window is fully backed by data. The instance-side clamp below keeps upstream's 72h cap — Sidekiq parity is pinned by test/parity/metrics_query_test.rb.

720
MAX_QUEUE_SERIES = 
25
ROLLUPS = 
{
  minutely: [60, ->(time) { History.minute_key(time) }],
  hourly: [600, ->(time) { History.ten_minute_key(time) }]
}.freeze
FOR_JOB_METRICS = 
%w[ms p f].freeze

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(pool: nil, now: ::Time.now) ⇒ Query

Returns a new instance of Query.



42
43
44
45
 
# File 'lib/wurk/metrics/query.rb', line 42

def initialize(pool: nil, now: ::Time.now)
  @time = now.utc
  @pool = pool
end

Class Method Details

.bkt_time_s(time, granularity) ⇒ Object 



37
38
39
40
 
# File 'lib/wurk/metrics/query.rb', line 37

def self.bkt_time_s(time, granularity)
  truncation = granularity == :hourly ? 600 : 60
  ::Time.at(time.to_i - (time.to_i % truncation)).utc.iso8601
end

.for_job(klass, minutes: nil, hours: nil, now: ::Time.now) ⇒ Object Originally defined in module DashboardSeries

Dashboard per-class series, oldest→newest, gap-filled: [{at:, p:, f:, ms:}, …]. minutes: gives one row per minute, hours: one per 10-minute bucket.

.history(bucket, window_seconds, now: ::Time.now) ⇒ Object Originally defined in module DashboardSeries

Cluster-total time-series for the dashboard throughput/failures charts, read from the compact buckets written by Wurk::Metrics::Rollup. bucket is '1m'/'5m'/'1h'; window_seconds is clamped to that bucket's retention. Returns [{at:, p:, f:, ms:}, ...] oldest→newest, gap-filled with zeros so a chart has a continuous x-axis.

.queue_history(bucket, window_seconds, queues: nil, now: ::Time.now) ⇒ Object Originally defined in module DashboardSeries

Per-queue size/latency gauge time-series written by Metrics::QueueRollup. Returns one entry per live queue (or the explicit queues: list, capped at MAX_QUEUE_SERIES) — [{name:, points: [{at:, size:, latency:}, …]}, …] — oldest→newest, gap-filled with zeros.

.top_jobs(class_filter: nil, minutes: 60, hours: nil, now: ::Time.now) ⇒ Object Originally defined in module DashboardSeries

Dashboard top-N: [[klass, {p:, f:, ms:}], …] over the window, busiest (p, which counts failures) first. class_filter is a class-name prefix. Sums straight off the buckets rather than through Query#top_jobs, whose per-bucket series a table never shows.

Instance Method Details

#for_job(klass, minutes: nil, hours: nil) ⇒ Object 



58
59
60
61
62
63
64
65
66
67
 
# File 'lib/wurk/metrics/query.rb', line 58

def for_job(klass, minutes: nil, hours: nil)
  result, keys = window(minutes, hours)
  fields = FOR_JOB_METRICS.map { |metric| "#{klass}|#{metric}" }
  job = result.job_results[klass]
  each_bucket(result, pipelined(keys) { |pipe, key| pipe.call('HMGET', key, *fields) }) do |time, values|
    add_job_bucket(job, time, values)
  end
  add_histograms(job, klass, keys.size) if result.granularity == :minutely
  finish(result)
end

#top_jobs(class_filter: nil, minutes: nil, hours: nil) ⇒ Object

class_filter is matched with class_filter.match?(klass), so pass a Regexp, as Sidekiq's Web UI does.



49
50
51
52
53
54
 
# File 'lib/wurk/metrics/query.rb', line 49

def top_jobs(class_filter: nil, minutes: nil, hours: nil)
  result, keys = window(minutes, hours)
  rows = pipelined(keys) { |pipe, key| pipe.call('HGETALL', key) }
  each_bucket(result, rows) { |time, hash| add_bucket(result, time, hash_reply(hash), class_filter) }
  finish(result)
end