Class: Async::Background::Metrics

Inherits:
Object
  • Object
show all
Defined in:
lib/async/background/metrics.rb

Constant Summary collapse

SCHEMA_FIELDS = 
{
  total_runs: :u64,
  total_successes: :u64,
  total_failures: :u64,
  total_timeouts: :u64,
  total_skips: :u64,
  active_jobs: :u32,
  last_run_at: :u64,
  last_duration_ms: :u32
}.freeze
EMPTY_HANDLES = 
{}.freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(worker_index:, total_workers:, shm_path: self.class.default_shm_path) ⇒ Metrics

Returns a new instance of Metrics.



23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/async/background/metrics.rb', line 23

def initialize(worker_index:, total_workers:, shm_path: self.class.default_shm_path)
  @enabled = false
  @registry = nil
  @metric_handles = EMPTY_HANDLES
  @shm_path = shm_path
  @unavailable_reason = nil

  validate_worker!(worker_index, total_workers)
  initialize_registry!(worker_index, total_workers, shm_path)
rescue LoadError => error
  mark_unavailable!(error)
end

Instance Attribute Details

#registry ⇒ Object (readonly)

Returns the value of attribute registry.



21
22
23
 
# File 'lib/async/background/metrics.rb', line 21

def registry
  @registry
end

#shm_path ⇒ Object (readonly)

Returns the value of attribute shm_path.



21
22
23
 
# File 'lib/async/background/metrics.rb', line 21

def shm_path
  @shm_path
end

#unavailable_reason ⇒ Object (readonly)

Returns the value of attribute unavailable_reason.



21
22
23
 
# File 'lib/async/background/metrics.rb', line 21

def unavailable_reason
  @unavailable_reason
end

Class Method Details

.available? ⇒ Boolean

Returns:

  • (Boolean) 


79
80
81
82
83
84
 
# File 'lib/async/background/metrics.rb', line 79

def available?
  load_utilization!
  true
rescue LoadError
  false
end

.default_shm_path ⇒ Object 



113
114
115
 
# File 'lib/async/background/metrics.rb', line 113

def default_shm_path
  ENV.fetch('ASYNC_BACKGROUND_METRICS_PATH') { File.join(Dir.tmpdir, 'async-background.shm') }
end

.load_utilization! ⇒ Object 



86
87
88
 
# File 'lib/async/background/metrics.rb', line 86

def load_utilization!
  require 'async/utilization'
end

.read_all(total_workers:, path: default_shm_path) ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/async/background/metrics.rb', line 95

def read_all(total_workers:, path: default_shm_path)
  validate_total_workers!(total_workers)
  return [] unless available? && File.file?(path)

  layout = schema
  segment = segment_size
  required_size = segment * total_workers

  File.open(path, 'rb') do |file|
    return [] if file.size < required_size

    buffer = IO::Buffer.map(file, required_size, 0, IO::Buffer::READONLY)
    decode_workers(buffer, layout, segment, total_workers)
  end
rescue Errno::ENOENT
  []
end

.schema ⇒ Object 



90
91
92
93
 
# File 'lib/async/background/metrics.rb', line 90

def schema
  load_utilization!
  ::Async::Utilization::Schema.build(SCHEMA_FIELDS)
end

.segment_size ⇒ Object 



117
118
119
 
# File 'lib/async/background/metrics.rb', line 117

def segment_size
  SCHEMA_FIELDS.sum { |_, type| IO::Buffer.size_of(type) }
end

Instance Method Details

#enabled? ⇒ Boolean

Returns:

  • (Boolean) 


36
 
# File 'lib/async/background/metrics.rb', line 36

def enabled? = @enabled

#job_failed(_entry, _error) ⇒ Object 



58
59
60
 
# File 'lib/async/background/metrics.rb', line 58

def job_failed(_entry, _error)
  increment(:total_failures) if enabled?
end

#job_finished(entry, duration) ⇒ Object 



53
54
55
56
 
# File 'lib/async/background/metrics.rb', line 53

def job_finished(entry, duration)
  job_succeeded(entry, duration)
  job_stopped(entry)
end

#job_skipped(_entry) ⇒ Object 



70
71
72
 
# File 'lib/async/background/metrics.rb', line 70

def job_skipped(_entry)
  increment(:total_skips) if enabled?
end

#job_started(_entry) ⇒ Object 



38
39
40
41
42
43
44
 
# File 'lib/async/background/metrics.rb', line 38

def job_started(_entry)
  return unless enabled?

  increment(:total_runs)
  increment(:active_jobs)
  set(:last_run_at, Process.clock_gettime(Process::CLOCK_REALTIME).to_i)
end

#job_stopped(_entry) ⇒ Object 



66
67
68
 
# File 'lib/async/background/metrics.rb', line 66

def job_stopped(_entry)
  decrement(:active_jobs) if enabled?
end

#job_succeeded(_entry, duration) ⇒ Object 



46
47
48
49
50
51
 
# File 'lib/async/background/metrics.rb', line 46

def job_succeeded(_entry, duration)
  return unless enabled?

  increment(:total_successes)
  set(:last_duration_ms, duration_to_milliseconds(duration))
end

#job_timed_out(_entry) ⇒ Object 



62
63
64
 
# File 'lib/async/background/metrics.rb', line 62

def job_timed_out(_entry)
  increment(:total_timeouts) if enabled?
end

#values ⇒ Object 



74
75
76
 
# File 'lib/async/background/metrics.rb', line 74

def values
  enabled? ? registry.values : {}
end