Class: Wurk::ProcessSet

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/wurk/process_set.rb

Overview

Live process list view backed by the processes SET + per-identity hashes. Cluster topology lives here: dashboard process list, total concurrency gauge, RSS roll-up, leader lookup.

Wire-compat is sacred — every Redis call matches Sidekiq OSS exactly. Spec: docs/target/sidekiq-free.md §1 (Redis schema), §19.6.

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(clean_plz = true) ⇒ ProcessSet

processes SET only stores identity strings; the live hash at each identity expires every 60s, so SCARD can lag reality. Constructor opts into a cleanup (rate-limited globally to 1/min) so the size reported below reflects the post-prune state — except when caller explicitly skips it (e.g. inside cleanup itself, or for snapshot reads on hot paths).

Positional Boolean matches Sidekiq's public API — wire-compat sacred.



27
28
29
 
# File 'lib/wurk/process_set.rb', line 27

def initialize(clean_plz = true) # rubocop:disable Style/OptionalBooleanParameter
  cleanup if clean_plz
end

Class Method Details

.[](identity) ⇒ Object

Fetch a single Process by identity. Pipelined SISMEMBER + HMGET so absence (process never registered) and expiry (heartbeat lapsed, info field gone) both return nil.



34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/wurk/process_set.rb', line 34

def self.[](identity)
  exists, fields = Wurk.redis do |conn|
    conn.pipelined do |pipe|
      pipe.call('SISMEMBER', Keys::PROCESSES, identity)
      pipe.call('HMGET', identity, *LOOKUP_FIELDS)
    end
  end
  return nil if exists.to_i.zero? || fields.first.nil?

  build_process(LOOKUP_FIELDS.zip(fields).to_h)
end

Instance Method Details

#cleanup ⇒ Object

SREM identities whose info HASH field has expired. Rate-limited globally to 1/min via SET NX EX so concurrent dashboards / API calls don't dogpile the prune. Returns the number of identities removed (or 0 when the lock was held by someone else).

Spec: docs/target/sidekiq-free.md §31.17.



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/wurk/process_set.rb', line 52

def cleanup
  return 0 unless acquired_cleanup_lock?

  Wurk.redis do |conn|
    procs = conn.call('SMEMBERS', Keys::PROCESSES)
    next 0 if procs.empty?

    heartbeats = conn.pipelined do |pipe|
      procs.each { |key| pipe.call('HGET', key, 'info') }
    end
    to_prune = procs.zip(heartbeats).filter_map { |id, info| id if info.nil? }
    next 0 if to_prune.empty?

    conn.call('SREM', Keys::PROCESSES, *to_prune).to_i
  end
end

#each ⇒ Object

Pipelined identity → HMGET. Skips identities whose info field is gone (heartbeat expired between SMEMBERS and HMGET). Sorted by identity so iteration order is stable for dashboards.



72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/wurk/process_set.rb', line 72

def each
  return enum_for(:each) unless block_given?

  rows = fetch_each_rows
  rows.each do |row|
    attrs = EACH_FIELDS.zip(row).to_h
    next if attrs['info'].nil?

    yield ProcessSet.send(:build_process, attrs)
  end
end

#leader ⇒ Object

Cluster leader identity. Ent-only feature; OSS only reads the key. Memoized per-instance: dashboards re-instantiate ProcessSet per render. ||= with empty-string fallback distinguishes "leader is unset" from "memoization not yet computed".



106
107
108
 
# File 'lib/wurk/process_set.rb', line 106

def leader
  @leader ||= Wurk.redis { |c| c.call('GET', 'dear-leader') } || ''
end

#size ⇒ Object

SCARD over processes. Not pruned — may include identities whose heartbeat has lapsed. Use each for the accurate count.



86
87
88
 
# File 'lib/wurk/process_set.rb', line 86

def size
  Wurk.redis { |conn| conn.call('SCARD', Keys::PROCESSES) }
end

#total_concurrency ⇒ Object

Sum of concurrency across live processes. Iterates each so dead identities are skipped.



92
93
94
 
# File 'lib/wurk/process_set.rb', line 92

def total_concurrency
  sum { |p| p['concurrency'].to_i }
end

#total_rss_in_kb ⇒ Object Also known as: total_rss

Sum of rss (KB) across live processes.



97
98
99
 
# File 'lib/wurk/process_set.rb', line 97

def total_rss_in_kb
  sum { |p| p['rss'].to_i }
end