1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
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
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
|
# frozen_string_literal: true
module Gitlab
module Redis
class MultiStore
include Gitlab::Utils::StrongMemoize
class ReadFromPrimaryError < StandardError
def message
'Value not found on the redis primary store. Read from the redis secondary store successful.'
end
end
class PipelinedDiffError < StandardError
def initialize(result_primary, result_secondary)
@result_primary = result_primary
@result_secondary = result_secondary
end
def message
"Pipelined command executed on both stores successfully but results differ between them. " \
"Result from the primary: #{@result_primary.inspect}. " \
"Result from the secondary: #{@result_secondary.inspect}."
end
end
class MethodMissingError < StandardError
def message
'Method missing. Falling back to execute method on the redis default store in Rails.env.production.'
end
end
attr_reader :primary_store, :secondary_store, :instance_name
FAILED_TO_READ_ERROR_MESSAGE = 'Failed to read from the redis primary_store.'
FAILED_TO_WRITE_ERROR_MESSAGE = 'Failed to write to the redis primary_store.'
FAILED_TO_RUN_PIPELINE = 'Failed to execute pipeline on the redis primary_store.'
SKIP_LOG_METHOD_MISSING_FOR_COMMANDS = %i[info].freeze
# For ENUMERATOR_CACHE_HIT_VALIDATOR and READ_CACHE_HIT_VALIDATOR,
# we define procs to validate cache hit. The only other acceptable value is nil,
# in the case of errors being raised.
#
# If a command has no empty response, set ->(val) { true }
#
# Ref: https://www.rubydoc.info/github/redis/redis-rb/Redis/Commands
#
READ_CACHE_HIT_VALIDATOR = {
exists: ->(val) { val != 0 },
exists?: ->(val) { val },
get: ->(val) { !val.nil? },
hexists: ->(val) { val },
hget: ->(val) { !val.nil? },
hgetall: ->(val) { val.is_a?(Hash) && !val.empty? },
hlen: ->(val) { val != 0 },
hmget: ->(val) { val.is_a?(Array) && !val.compact.empty? },
hscan_each: ->(val) { val.is_a?(Enumerator) && !val.first.nil? },
mapped_hmget: ->(val) { val.is_a?(Hash) && !val.compact.empty? },
mget: ->(val) { val.is_a?(Array) && !val.compact.empty? },
scan_each: ->(val) { val.is_a?(Enumerator) && !val.first.nil? },
scard: ->(val) { val != 0 },
sismember: ->(val) { val },
smembers: ->(val) { val.is_a?(Array) && !val.empty? },
sscan: ->(val) { val != ['0', []] },
sscan_each: ->(val) { val.is_a?(Enumerator) && !val.first.nil? },
ttl: ->(val) { val != 0 && val != -2 }, # ttl returns -2 if the key does not exist. See https://redis.io/commands/ttl/
zscan_each: ->(val) { val.is_a?(Enumerator) && !val.first.nil? }
}.freeze
WRITE_COMMANDS = %i[
del
eval
expire
flushdb
hdel
hset
incr
incrby
mapped_hmset
rpush
sadd
set
setex
setnx
srem
unlink
].freeze
PIPELINED_COMMANDS = %i[
pipelined
multi
].freeze
# To transition between two Redis store, `primary_store` should be the target store,
# and `secondary_store` should be the current store. Transition is controlled with feature flags:
#
# - At the default state, all read and write operations are executed in the secondary instance.
# - Turning use_primary_and_secondary_stores_for_<instance_name> on: The store writes to both instances.
# The read commands are executed in primary, but fallback to secondary.
# Other commands are executed in the the default instance (Secondary).
# - Turning use_primary_store_as_default_for_<instance_name> on: The behavior is the same as above,
# but other commands are executed in the primary now.
# - Turning use_primary_and_secondary_stores_for_<instance_name> off: commands are executed in the primary store.
def initialize(primary_store, secondary_store, instance_name)
@primary_store = primary_store
@secondary_store = secondary_store
@instance_name = instance_name
validate_stores!
end
# rubocop:disable GitlabSecurity/PublicSend
READ_CACHE_HIT_VALIDATOR.each_key do |name|
define_method(name) do |*args, **kwargs, &block|
if use_primary_and_secondary_stores?
read_command(name, *args, **kwargs, &block)
else
default_store.send(name, *args, **kwargs, &block)
end
end
end
WRITE_COMMANDS.each do |name|
define_method(name) do |*args, **kwargs, &block|
if use_primary_and_secondary_stores?
write_command(name, *args, **kwargs, &block)
else
default_store.send(name, *args, **kwargs, &block)
end
end
end
PIPELINED_COMMANDS.each do |name|
define_method(name) do |*args, **kwargs, &block|
if use_primary_and_secondary_stores?
pipelined_both(name, *args, **kwargs, &block)
else
default_store.send(name, *args, **kwargs, &block)
end
end
end
def method_missing(...)
return @instance.send(...) if @instance
log_method_missing(...)
default_store.send(...)
end
# rubocop:enable GitlabSecurity/PublicSend
def respond_to_missing?(command_name, include_private = false)
true
end
# This is needed because of Redis::Rack::Connection is requiring Redis::Store
# https://github.com/redis-store/redis-rack/blob/a833086ba494083b6a384a1a4e58b36573a9165d/lib/redis/rack/connection.rb#L15
# Done similarly in https://github.com/lsegal/yard/blob/main/lib/yard/templates/template.rb#L122
def is_a?(klass)
return true if klass == default_store.class
super(klass)
end
alias_method :kind_of?, :is_a?
def to_s
use_primary_and_secondary_stores? ? primary_store.to_s : default_store.to_s
end
def use_primary_and_secondary_stores?
feature_table_exists? &&
Feature.enabled?("use_primary_and_secondary_stores_for_#{instance_name.underscore}") && # rubocop:disable Cop/FeatureFlagUsage
!same_redis_store?
end
def use_primary_store_as_default?
feature_table_exists? &&
Feature.enabled?("use_primary_store_as_default_for_#{instance_name.underscore}") && # rubocop:disable Cop/FeatureFlagUsage
!same_redis_store?
end
def increment_pipelined_command_error_count(command_name)
@pipelined_command_error ||= Gitlab::Metrics.counter(:gitlab_redis_multi_store_pipelined_diff_error_total,
'Redis MultiStore pipelined command diff between stores')
@pipelined_command_error.increment(command: command_name, instance_name: instance_name)
end
def increment_read_fallback_count(command_name)
@read_fallback_counter ||= Gitlab::Metrics.counter(:gitlab_redis_multi_store_read_fallback_total,
'Client side Redis MultiStore reading fallback')
@read_fallback_counter.increment(command: command_name, instance_name: instance_name)
end
def increment_method_missing_count(command_name)
@method_missing_counter ||= Gitlab::Metrics.counter(:gitlab_redis_multi_store_method_missing_total,
'Client side Redis MultiStore method missing')
@method_missing_counter.increment(command: command_name, instance_name: instance_name)
end
def log_error(exception, command_name, extra = {})
Gitlab::ErrorTracking.log_exception(
exception,
extra.merge(command_name: command_name, instance_name: instance_name))
end
def default_store
use_primary_store_as_default? ? primary_store : secondary_store
end
def fallback_store
use_primary_store_as_default? ? secondary_store : primary_store
end
def ping(message = nil)
if use_primary_and_secondary_stores?
# Both stores have to response success for the ping to be considered success.
# We assume both stores cannot return different responses (only both "PONG" or both echo the message).
# If either store is not reachable, an Error will be raised anyway thus taking any response works.
[primary_store, secondary_store].map { |store| store.ping(message) }.first
else
default_store.ping(message)
end
end
private
# @return [Boolean]
def feature_table_exists?
# Use table_exists? (which uses ActiveRecord's schema cache) instead of Feature.feature_flags_available?
# as the latter runs a ';' SQL query which causes a connection to be checked out.
Feature::FlipperFeature.table_exists?
rescue StandardError
false
end
def log_method_missing(command_name, *_args)
return if SKIP_LOG_METHOD_MISSING_FOR_COMMANDS.include?(command_name)
raise MethodMissingError if Rails.env.test? || Rails.env.development?
log_error(MethodMissingError.new, command_name)
increment_method_missing_count(command_name)
end
def read_command(command_name, *args, **kwargs, &block)
if @instance
send_command(@instance, command_name, *args, **kwargs, &block)
else
read_one_with_fallback(command_name, *args, **kwargs, &block)
end
end
def write_command(command_name, *args, **kwargs, &block)
if @instance
send_command(@instance, command_name, *args, **kwargs, &block)
else
write_both(command_name, *args, **kwargs, &block)
end
end
def read_one_with_fallback(command_name, *args, **kwargs, &block)
begin
value = send_command(default_store, command_name, *args, **kwargs, &block)
rescue StandardError => e
log_error(e, command_name,
multi_store_error_message: FAILED_TO_READ_ERROR_MESSAGE)
end
return value if block.nil? && cache_hit?(command_name, value)
fallback_read(command_name, *args, **kwargs, &block)
end
def cache_hit?(command, value)
validator = READ_CACHE_HIT_VALIDATOR[command]
return false unless validator
!value.nil? && validator.call(value)
end
def fallback_read(command_name, *args, **kwargs, &block)
value = send_command(fallback_store, command_name, *args, **kwargs, &block)
if value
log_error(ReadFromPrimaryError.new, command_name)
increment_read_fallback_count(command_name)
end
value
end
def write_both(command_name, *args, **kwargs, &block)
begin
send_command(primary_store, command_name, *args, **kwargs, &block)
rescue StandardError => e
log_error(e, command_name,
multi_store_error_message: FAILED_TO_WRITE_ERROR_MESSAGE)
end
send_command(secondary_store, command_name, *args, **kwargs, &block)
end
# Run the entire pipeline on both stores. We assume that `&block` is idempotent.
def pipelined_both(command_name, *args, **kwargs, &block)
begin
result_primary = send_command(primary_store, command_name, *args, **kwargs, &block)
rescue StandardError => e
log_error(e, command_name, multi_store_error_message: FAILED_TO_RUN_PIPELINE)
end
result_secondary = send_command(secondary_store, command_name, *args, **kwargs, &block)
# Pipelined commands return an array with all results. If they differ, log an error
if result_primary && result_primary != result_secondary
error = PipelinedDiffError.new(result_primary, result_secondary)
error.set_backtrace(Thread.current.backtrace[1..]) # Manually set backtrace, since the error is not `raise`d
log_error(error, command_name)
increment_pipelined_command_error_count(command_name)
end
result_secondary
end
def same_redis_store?
strong_memoize(:same_redis_store) do
# <Redis client v4.7.1 for unix:///path_to/redis/redis.socket/5>"
primary_store.inspect == secondary_store.inspect
end
end
# rubocop:disable GitlabSecurity/PublicSend
def send_command(redis_instance, command_name, *args, **kwargs, &block)
if block
# Make sure that block is wrapped and executed only on the redis instance that is executing the block
redis_instance.send(command_name, *args, **kwargs) do |*params|
with_instance(redis_instance, *params, &block)
end
else
redis_instance.send(command_name, *args, **kwargs)
end
end
# rubocop:enable GitlabSecurity/PublicSend
def with_instance(instance, *params)
@instance = instance
yield(*params)
ensure
@instance = nil
end
def redis_store?(store)
store.is_a?(::Redis) || store.is_a?(::Redis::Namespace)
end
def validate_stores!
raise ArgumentError, 'primary_store is required' unless primary_store
raise ArgumentError, 'secondary_store is required' unless secondary_store
raise ArgumentError, 'instance_name is required' unless instance_name
raise ArgumentError, 'invalid primary_store' unless redis_store?(primary_store)
raise ArgumentError, 'invalid secondary_store' unless redis_store?(secondary_store)
end
end
end
end
|