Class: Stoplight::Infrastructure::Redis::DataStore

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/stoplight/infrastructure/redis/data_store.rb,
lib/stoplight/infrastructure/redis/data_store/scripting.rb,
lib/stoplight/infrastructure/redis/data_store/recovery_lock_store.rb,
lib/stoplight/infrastructure/redis/data_store/recovery_lock_token.rb

Overview

Errors

All errors are stored in the sorted set where keys are serialized errors and values (Redis uses “score” term) contain integer representations of the time when an error happened.

This data structure enables us to query errors that happened within a specific period. We use this feature to support window_size option.

To avoid uncontrolled memory consumption, we keep at most config.threshold number of errors happened within last config.window_size seconds (by default infinity).

steep:ignore:start

See Also:

  • Base

Defined Under Namespace

Classes: RecoveryLockStore, RecoveryLockToken, Scripting

Constant Summary collapse

METRICS_RETENTION_TIME =

1 day

60 * 60 * 24
KEY_SEPARATOR = 
":"
KEY_PREFIX = 
%w[stoplight v5].join(KEY_SEPARATOR)

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redis:, recovery_lock_store:, scripting:, clock:, warn_on_clock_skew: true) ⇒ DataStore

Returns a new instance of DataStore.

Parameters:



106
107
108
109
110
111
112
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 106

def initialize(redis:, recovery_lock_store:, scripting:, clock:, warn_on_clock_skew: true)
  @clock = clock
  @warn_on_clock_skew = warn_on_clock_skew
  @redis = redis
  @recovery_lock_store = recovery_lock_store
  @scripting = scripting
end

Class Method Details

.bucket_key(light_name, metric:, time:) ⇒ String

Generates a Redis key for a specific metric and time.

Parameters:

  • light_name (String) —

    The name of the light.

  • metric (String) —

    The metric type (e.g., “errors”).

  • time (Time, Numeric) —

    The time for which to generate the key.

Returns:

  • (String) —

    The generated Redis key.



65
66
67
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 65

def bucket_key(light_name, metric:, time:)
  key("metrics", light_name, metric, (time.to_i / bucket_size) * bucket_size)
end

.buckets_for_window(light_name, metric:, window_end:, window_size:) ⇒ Array<String>

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Retrieves the list of Redis bucket keys required to cover a specific time window.

Parameters:

  • light_name (String) —

    The name of the light (used as part of the Redis key).

  • metric (String) —

    The metric type (e.g., “errors”).

  • window_end (Time, Numeric) —

    The end time of the window (can be a Time object or a numeric timestamp).

  • window_size (Numeric) —

    The size of the time window in seconds.

Returns:

  • (Array<String>) —

    A list of Redis keys for the buckets that cover the time window.



44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 44

def buckets_for_window(light_name, metric:, window_end:, window_size:)
  window_end_ts = window_end.to_i
  window_start_ts = window_end_ts - [window_size, METRICS_RETENTION_TIME].compact.min.to_i

  # Find bucket timestamps that contain any part of the window
  start_bucket = (window_start_ts / bucket_size) * bucket_size

  # End bucket is the last bucket that contains data within our window
  end_bucket = ((window_end_ts - 1) / bucket_size) * bucket_size

  (start_bucket..end_bucket).step(bucket_size).map do |bucket_start|
    bucket_key(light_name, metric: metric, time: bucket_start)
  end
end

.key(*pieces) ⇒ String

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Generates a Redis key by joining the prefix with the provided pieces.

Parameters:

  • pieces (Array<String, Integer>) —

    Parts of the key to be joined.

Returns:

  • (String) —

    The generated Redis key.



30
31
32
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 30

def key(*pieces)
  [KEY_PREFIX, *pieces].join(KEY_SEPARATOR)
end

Instance Method Details

#acquire_recovery_lock(config) ⇒ Stoplight::Infrastructure::Redis::DataStore::RecoveryLockToken?

Parameters:

Returns:



328
329
330
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 328

def acquire_recovery_lock(config)
  recovery_lock_store.acquire_lock(config.name)
end

#clear_metrics(config) ⇒ Object 



210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 210

def clear_metrics(config)
  if config.window_size
    window_end_ts = clock.current_time.to_i
    @redis.with do |client|
      client.multi do |tx|
        tx.unlink(
          *failure_bucket_keys(config, window_end: window_end_ts),
          *success_bucket_keys(config, window_end: window_end_ts)
        )
        tx.hdel(metadata_key(config), "last_success_at", "last_error_json", "consecutive_errors", "consecutive_successes")
      end
    end
  else
    @redis.with do |client|
      client.hdel(metadata_key(config), "last_success_at", "last_error_json", "consecutive_errors", "consecutive_successes")
    end
  end
end

#clear_recovery_metrics(config) ⇒ Object 



229
230
231
232
233
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 229

def clear_recovery_metrics(config)
  @redis.with do |client|
    client.del(recovery_metrics_key(config))
  end
end

#delete_light(config) ⇒ void

This method returns an undefined value.

Removes all traces of a light from Redis metadata (metrics will expire by TTL).

Parameters:



392
393
394
395
396
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 392

def delete_light(config)
  @redis.then do |client|
    client.del(metadata_key(config), recovery_metrics_key(config))
  end
end

#get_metrics(config) ⇒ Stoplight::Domain::MetricsSnapshot

Parameters:

Returns:



130
131
132
133
134
135
136
137
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
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 130

def get_metrics(config)
  config.name

  window_end_ts = clock.current_time.to_f
  window_start_ts = window_end_ts - config.window_size.to_i

  if config.window_size
    failure_keys = failure_bucket_keys(config, window_end: window_end_ts)
    success_keys = success_bucket_keys(config, window_end: window_end_ts)
  else
    failure_keys = []
    success_keys = []
  end

  successes, errors, last_success_at, last_error_json, consecutive_errors, consecutive_successes = scripting.call(
    :get_metrics,
    args: [
      failure_keys.count,
      window_start_ts,
      window_end_ts,
      "last_success_at", "last_error_json", "consecutive_errors", "consecutive_successes"
    ],
    keys: [
      metadata_key(config),
      *success_keys,
      *failure_keys
    ]
  )
  consecutive_errors = config.window_size ? [consecutive_errors.to_i, errors].min : consecutive_errors.to_i
  consecutive_successes = config.window_size ? [consecutive_successes.to_i, successes].min : consecutive_successes.to_i

  Domain::MetricsSnapshot.new(
    successes: (successes if config.window_size),
    errors: (errors if config.window_size),
    consecutive_errors:,
    consecutive_successes:,
    last_error: deserialize_failure(last_error_json),
    last_success_at: (clock.at(last_success_at.to_f) if last_success_at)
  )
end

#get_recovery_metrics(config) ⇒ Stoplight::Domain::MetricsSnapshot

Parameters:

Returns:



173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 173

def get_recovery_metrics(config)
  last_success_at, last_error_json, consecutive_errors, consecutive_successes = @redis.with do |client|
    client.hmget(
      recovery_metrics_key(config),
      "last_success_at", "last_error_json", "consecutive_errors", "consecutive_successes"
    )
  end

  Domain::MetricsSnapshot.new(
    successes: nil, errors: nil,
    consecutive_errors: consecutive_errors.to_i,
    consecutive_successes: consecutive_successes.to_i,
    last_error: deserialize_failure(last_error_json),
    last_success_at: (clock.at(last_success_at.to_f) if last_success_at)
  )
end

#get_state_snapshot(config) ⇒ Stoplight::Domain::StateSnapshot

Returns:



191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 191

def get_state_snapshot(config)
  detect_clock_skew

  breached_at_raw, locked_state, recovery_scheduled_after_raw, recovery_started_at_raw = @redis.with do |client|
    client.hmget(metadata_key(config), :breached_at, :locked_state, :recovery_scheduled_after, :recovery_started_at)
  end
  breached_at = breached_at_raw&.to_f
  recovery_scheduled_after = recovery_scheduled_after_raw&.to_f
  recovery_started_at = recovery_started_at_raw&.to_f

  Domain::StateSnapshot.new(
    breached_at: (clock.at(breached_at) if breached_at),
    locked_state: locked_state || State::UNLOCKED,
    recovery_scheduled_after: (clock.at(recovery_scheduled_after) if recovery_scheduled_after),
    recovery_started_at: (clock.at(recovery_started_at) if recovery_started_at),
    time: clock.current_time
  )
end

#inspect ⇒ Object 



304
305
306
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 304

def inspect
  "#<#{self.class.name} redis=#{@redis.inspect}>"
end

#names ⇒ Object 



114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 114

def names
  metadata, recovery_metrics = @redis.then do |client|
    [
      [key("metadata", "*"), /^#{key("metadata", "")}/],
      [key("recovery_metrics", "*"), /^#{key("recovery_metrics", "")}/]
    ].map do |(pattern, prefix_regex)|
      client.scan_each(match: pattern).to_a.map do |key|
        key.sub(prefix_regex, "")
      end
    end
  end
  metadata + recovery_metrics
end

#record_failure(config, exception) ⇒ void

This method returns an undefined value.

Parameters:



238
239
240
241
242
243
244
245
246
247
248
249
250
251
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 238

def record_failure(config, exception)
  current_time = clock.current_time
  current_ts = clock.current_time.to_f
  failure = Domain::Failure.from_error(exception, time: current_time)

  scripting.call(
    :record_failure,
    args: [current_ts, SecureRandom.hex(12), serialize_failure(failure), metrics_ttl, metadata_ttl],
    keys: [
      metadata_key(config),
      config.window_size && errors_key(config, time: current_ts)
    ].compact
  )
end

#record_recovery_probe_failure(config, exception) ⇒ void

This method returns an undefined value.

Records a failed recovery probe for a specific light configuration.

Parameters:



271
272
273
274
275
276
277
278
279
280
281
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 271

def record_recovery_probe_failure(config, exception)
  current_time = clock.current_time
  current_ts = clock.current_time.to_f
  failure = Domain::Failure.from_error(exception, time: current_time)

  scripting.call(
    :record_recovery_probe_failure,
    args: [current_ts, serialize_failure(failure)],
    keys: [recovery_metrics_key(config)]
  )
end

#record_recovery_probe_success(config) ⇒ void

This method returns an undefined value.

Records a successful recovery probe for a specific light configuration.

Parameters:



287
288
289
290
291
292
293
294
295
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 287

def record_recovery_probe_success(config)
  current_ts = clock.current_time.to_f

  scripting.call(
    :record_recovery_probe_success,
    args: [current_ts],
    keys: [recovery_metrics_key(config)]
  )
end

#record_success(config, request_id: SecureRandom.hex(12)) ⇒ Object 



253
254
255
256
257
258
259
260
261
262
263
264
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 253

def record_success(config, request_id: SecureRandom.hex(12))
  current_ts = clock.current_time.to_f

  scripting.call(
    :record_success,
    args: [current_ts, request_id, metrics_ttl, metadata_ttl],
    keys: [
      metadata_key(config),
      config.window_size && successes_key(config, time: current_ts)
    ].compact
  )
end

#release_recovery_lock(lock) ⇒ void

This method returns an undefined value.

Parameters:



334
335
336
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 334

def release_recovery_lock(lock)
  recovery_lock_store.release_lock(lock)
end

#set_state(config, state) ⇒ Object 



297
298
299
300
301
302
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 297

def set_state(config, state)
  @redis.then do |client|
    client.hset(metadata_key(config), "locked_state", state)
  end
  state
end

#transition_to_color(config, color) ⇒ Boolean

Combined method that performs the state transition based on color

Parameters:

  • config (Stoplight::Domain::Config) —

    The light configuration

  • color (String) —

    The color to transition to (“green”, “yellow”, or “red”)

Returns:

  • (Boolean) —

    true if this is the first instance to detect this transition



313
314
315
316
317
318
319
320
321
322
323
324
 
# File 'lib/stoplight/infrastructure/redis/data_store.rb', line 313

def transition_to_color(config, color)
  case color
  when Color::GREEN
    transition_to_green(config)
  when Color::YELLOW
    transition_to_yellow(config)
  when Color::RED
    transition_to_red(config)
  else
    raise ArgumentError, "Invalid color: #{color}"
  end
end