Class: PatientHttp::Sidekiq::Stats

Inherits:
Object
  • Object
show all
Includes:
TimeHelper
Defined in:
lib/patient_http/sidekiq/stats.rb

Overview

Tracks processor statistics with local aggregation.

Metrics are accumulated in memory and flushed to a Redis hash on an interval (see stats_flush_interval), so recording a request costs a hash increment instead of a Redis round trip and the shared totals key is not a per-request hot key across the fleet. Deltas are commutative increments, so concurrent flushes from many processes are safe. A flush failure merges the deltas back so they are retried on the next flush; a crashed process loses at most one interval of counters.

Constant Summary collapse

TOTALS_KEY =

Redis key prefixes

"sidekiq:patient_http:totals"
TOTALS_TTL =

TTLs

30 * 24 * 60 * 60
PROCESSOR_METRICS =

Metrics reported for every processor, so that a processor that has only recorded some of them still reports the rest as zero.

{
  "requests" => 0,
  "duration" => 0.0,
  "errors" => 0,
  "max_capacity_exceeded" => 0,
  "max_inflight" => 0
}.freeze
RECORD_MAXIMA_SCRIPT =

Lua script that raises fields to a new high-water mark. A field is only written when the new value is higher, so processes recording their own marks concurrently converge on the highest one.

KEYS = totals key ARGV = alternating field and value

<<~LUA
  for i = 1, #ARGV, 2 do
    local field = ARGV[i]
    local value = tonumber(ARGV[i + 1])
    local current = redis.call('HGET', KEYS[1], field)
    if not current or value > tonumber(current) then
      redis.call('HSET', KEYS[1], field, value)
    end
  end

  return 1
LUA
RECORD_MAXIMA_SHA = 
Digest::SHA1.hexdigest(RECORD_MAXIMA_SCRIPT).freeze

Instance Method Summary collapse

Constructor Details

#initialize(config = nil) ⇒ Stats

Returns a new instance of Stats.



55
56
57
58
59
60
61
62
63
64
 
# File 'lib/patient_http/sidekiq/stats.rb', line 55

def initialize(config = nil)
  @hostname = ::Socket.gethostname.force_encoding("UTF-8").freeze
  @pid = ::Process.pid
  @config = config
  @mutex = Mutex.new
  @pending = Hash.new(0)
  @maxima = Hash.new(0)
  @maxima_changed = false
  @last_flush = monotonic_time
end

Instance Method Details

#flush ⇒ void

This method returns an undefined value.

Flush pending deltas to Redis in a single pipelined write.



138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/patient_http/sidekiq/stats.rb', line 138

def flush
  pending = nil
  maxima = nil
  @mutex.synchronize do
    @last_flush = monotonic_time
    pending, @pending = @pending, Hash.new(0)
    # The high-water marks are not consumed: they are the highest values
    # this process has seen, and they are sent on every flush so that
    # they are restored after the totals are cleared.
    maxima = @maxima.dup
    @maxima_changed = false
  end
  # A processor that is saturated records new marks without completing
  # anything, so a new mark is a reason to flush on its own.
  return if pending.empty? && maxima.empty?

  begin
    # The pipeline is a batch of increments, so it must not be replayed
    # after a connection failure: the server may already have applied it
    # and a replay would double count. A failure merges the deltas back
    # instead, which at worst loses them if the write did land.
    PatientHttp::Sidekiq.redis(retry_on_connection_error: false) do |redis|
      redis.pipelined do |pipeline|
        pending.each do |field, delta|
          if float_field?(field)
            pipeline.hincrbyfloat(TOTALS_KEY, field, delta)
          else
            pipeline.hincrby(TOTALS_KEY, field, delta)
          end
        end
        pipeline.expire(TOTALS_KEY, TOTALS_TTL)
      end
    end
    flush_maxima(maxima)
  rescue => e
    # Put the deltas back so nothing is lost; they are retried on the
    # next flush.
    @mutex.synchronize do
      pending.each { |field, delta| @pending[field] += delta }
      @maxima_changed = true if maxima.any?
    end
    handle_error(e)
  end
end

#flush_if_due ⇒ void

This method returns an undefined value.

Flush when the configured interval has elapsed since the last flush. Called periodically by the task monitor thread.



187
188
189
190
191
192
 
# File 'lib/patient_http/sidekiq/stats.rb', line 187

def flush_if_due
  due = @mutex.synchronize do
    (@pending.any? || @maxima_changed) && (monotonic_time - @last_flush >= flush_interval)
  end
  flush if due
end

#get_totals ⇒ Hash

Get running totals

Returns:

  • (Hash) —

    hash with requests, duration, errors, max_capacity_exceeded, http_status_counts



197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/patient_http/sidekiq/stats.rb', line 197

def get_totals
  # Flush first so this process's own recorded events are visible.
  # Other processes' unflushed deltas are stale by at most their flush
  # interval.
  flush

  PatientHttp::Sidekiq.redis do |redis|
    stats = redis.hgetall(TOTALS_KEY)

    # Extract HTTP status counts, error type counts, and per-processor counts
    http_status_counts = {}
    error_type_counts = {}
    processor_counts = {}
    stats.each do |key, value|
      if key.start_with?("http_status:")
        status = key.sub("http_status:", "").to_i
        http_status_counts[status] = value.to_i
      elsif key.start_with?("errors:") && key != "errors"
        error_type = key.sub("errors:", "")
        error_type_counts[error_type] = value.to_i
      elsif key.start_with?("processor:")
        _, name, metric = key.split(":", 3)
        next unless name && metric

        counts = (processor_counts[name] ||= PROCESSOR_METRICS.dup)
        counts[metric] = (metric == "duration") ? value.to_f.round(6) : value.to_i
      end
    end

    totals = {
      "requests" => (stats["requests"] || 0).to_i,
      "duration" => (stats["duration"] || 0).to_f.round(6),
      "errors" => (stats["errors"] || 0).to_i,
      "max_capacity_exceeded" => (stats["max_capacity_exceeded"] || 0).to_i,
      "http_status_counts" => http_status_counts.sort.to_h,
      "error_type_counts" => error_type_counts.sort.to_h
    }
    totals["processors"] = processor_counts.sort.to_h if processor_counts.any?
    totals
  end
end

#record_capacity_exceeded(processor_name: nil) ⇒ void

This method returns an undefined value.

Record a that a request was refused because the max capacity of the Processor was reached.

Parameters:

  • processor_name (String, Symbol, nil) (defaults to: nil) —

    name of the processor that refused the request; when given, per-processor fields are recorded as well



127
128
129
130
131
132
133
 
# File 'lib/patient_http/sidekiq/stats.rb', line 127

def record_capacity_exceeded(processor_name: nil)
  processor = processor_field_prefix(processor_name)
  record do |pending|
    pending["max_capacity_exceeded"] += 1
    pending["#{processor}max_capacity_exceeded"] += 1 if processor
  end
end

#record_error(error_type, processor_name: nil) ⇒ void

This method returns an undefined value.

Record a request error

Parameters:

  • error_type (String) —

    the type of error that occurred

  • processor_name (String, Symbol, nil) (defaults to: nil) —

    name of the processor that ran the request; when given, per-processor fields are recorded as well



92
93
94
95
96
97
98
99
 
# File 'lib/patient_http/sidekiq/stats.rb', line 92

def record_error(error_type, processor_name: nil)
  processor = processor_field_prefix(processor_name)
  record do |pending|
    pending["errors"] += 1
    pending["errors:#{error_type}"] += 1
    pending["#{processor}errors"] += 1 if processor
  end
end

#record_inflight_peak(count, processor_name:) ⇒ void

This method returns an undefined value.

Record the number of requests a processor had in flight, keeping the highest value seen. The count only rises when a request is handed to a processor, so recording it at that moment captures every high-water mark exactly.

Parameters:

  • count (Integer) —

    the number of requests in flight

  • processor_name (String, Symbol, nil) —

    name of the processor



109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/patient_http/sidekiq/stats.rb', line 109

def record_inflight_peak(count, processor_name:)
  processor = processor_field_prefix(processor_name)
  return unless processor

  field = "#{processor}max_inflight"
  @mutex.synchronize do
    next unless count > @maxima[field]

    @maxima[field] = count
    @maxima_changed = true
  end
end

#record_request(status, duration, processor_name: nil) ⇒ void

This method returns an undefined value.

Record a completed request.

Parameters:

  • status (Integer, nil) —

    HTTP response status code

  • duration (Float) —

    request duration in seconds

  • processor_name (String, Symbol, nil) (defaults to: nil) —

    name of the processor that ran the request; when given, per-processor fields are recorded as well



73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/patient_http/sidekiq/stats.rb', line 73

def record_request(status, duration, processor_name: nil)
  processor = processor_field_prefix(processor_name)
  record do |pending|
    pending["requests"] += 1
    pending["duration"] += duration.to_f
    pending["http_status:#{status}"] += 1 if status && status >= 100 && status < 600
    if processor
      pending["#{processor}requests"] += 1
      pending["#{processor}duration"] += duration.to_f
    end
  end
end

#reset! ⇒ void

This method returns an undefined value.

Reset all stats (useful for testing)



242
243
244
245
246
247
248
249
250
251
252
 
# File 'lib/patient_http/sidekiq/stats.rb', line 242

def reset!
  @mutex.synchronize do
    @pending = Hash.new(0)
    @maxima = Hash.new(0)
    @maxima_changed = false
    @last_flush = monotonic_time
  end
  PatientHttp::Sidekiq.redis do |redis|
    redis.del(TOTALS_KEY)
  end
end