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
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
|
# frozen_string_literal: true
module Gitlab
module Redis
class MultiStore
include Gitlab::Utils::StrongMemoize
class PipelinedDiffError < StandardError
def initialize(non_default_store_result, default_store_result)
@non_default_store_result = non_default_store_result
@default_store_result = default_store_result
end
def message
"Pipelined command executed on both stores successfully but results differ between them. " \
"Result from the non-default store: #{@non_default_store_result.inspect}. " \
"Result from the default store: #{@default_store_result.inspect}."
end
end
class MethodMissingError < StandardError
def initialize(cmd)
@cmd = cmd
end
def message
"Method missing #{@cmd}. Falling back to execute method on the redis default store in Rails.env.production."
end
end
class NestedReadonlyPipelineError < StandardError
def message
'Nested use of with_readonly_pipeline is detected.'
end
end
attr_reader :primary_store, :secondary_store, :instance_name
FAILED_TO_READ_ERROR_MESSAGE = 'Failed to read from the redis default_store.'
FAILED_TO_WRITE_ERROR_MESSAGE = 'Failed to write to the redis non_default_store.'
FAILED_TO_RUN_PIPELINE = 'Failed to execute pipeline on the redis non_default_store.'
SKIP_LOG_METHOD_MISSING_FOR_COMMANDS = %i[info].freeze
# _client and without_reconnect are Redis::Client methods which may be called through multistore
REDIS_CLIENT_COMMANDS = %i[
_client
without_reconnect
].freeze
PUBSUB_SUBSCRIBE_COMMANDS = %i[
subscribe
unsubscribe
].freeze
READ_COMMANDS = %i[
exists
exists?
get
hexists
hget
hgetall
hlen
hmget
hscan_each
llen
lrange
mapped_hmget
mget
pfcount
pttl
scan
scan_each
scard
sismember
smembers
sscan
sscan_each
strlen
ttl
type
zcard
zcount
zrange
zrangebyscore
zrevrange
zscan_each
zscore
].freeze
WRITE_COMMANDS = %i[
decr
del
eval
expire
flushdb
hdel
hincrby
hset
incr
incrby
mapped_hmset
pfadd
pfmerge
publish
rpush
sadd
sadd?
set
setex
setnx
spop
srem
srem?
unlink
zadd
zpopmin
zrem
zremrangebyrank
zremrangebyscore
memory
].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
# Pipelines are sent to both instances by default since
# they could execute both read and write commands.
#
# But for pipelines that only consists of read commands, this method
# can be used to scope the pipeline and send it only to the default store.
def with_readonly_pipeline
raise NestedReadonlyPipelineError if readonly_pipeline?
Thread.current[:readonly_pipeline] = true
yield
ensure
Thread.current[:readonly_pipeline] = false
end
def readonly_pipeline?
Thread.current[:readonly_pipeline].present?
end
# rubocop:disable GitlabSecurity/PublicSend
(READ_COMMANDS + REDIS_CLIENT_COMMANDS + PUBSUB_SUBSCRIBE_COMMANDS).each 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? && !readonly_pipeline?
pipelined_both(name, *args, **kwargs, &block)
else
send_command(default_store, 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_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 non_default_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
# connection_pool gem calls `#close` method:
#
# https://github.com/mperham/connection_pool/blob/v2.4.1/lib/connection_pool.rb#L63
#
# Let's define it explicitly instead of propagating it to method_missing
def close
if use_primary_and_secondary_stores?
[primary_store, secondary_store].map(&:close).first
else
default_store.close
end
end
# blpop blocks until an element to be popped exist in the list or after a timeout.
def blpop(*args)
result = default_store.blpop(*args)
if !!result && use_primary_and_secondary_stores?
# special case to accommodate Gitlab::JobWaiter as blpop is only used in JobWaiter
# 1s should be sufficient wait time to account for delays between 1st and 2nd lpush
# https://gitlab.com/gitlab-com/gl-infra/scalability/-/issues/2520#note_1630893702
non_default_store.blpop(args.first, timeout: 1)
end
result
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, command_name if Rails.env.test? || Rails.env.development?
log_error(MethodMissingError.new(command_name), 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_from_default(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_from_default(command_name, *args, **kwargs, &block)
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)
raise
end
def write_both(command_name, *args, **kwargs, &block)
result = send_command(default_store, command_name, *args, **kwargs, &block)
# write to the non-default store only if write on default store is successful
begin
send_command(non_default_store, command_name, *args, **kwargs, &block)
rescue StandardError => e
log_error(e, command_name,
multi_store_error_message: FAILED_TO_WRITE_ERROR_MESSAGE)
end
result
end
# Run the entire pipeline on both stores. We assume that `&block` is idempotent.
def pipelined_both(command_name, *args, **kwargs, &block)
result_default = send_command(default_store, command_name, *args, **kwargs, &block)
begin
result_non_default = send_command(non_default_store, command_name, *args, **kwargs, &block)
rescue StandardError => e
log_error(e, command_name, multi_store_error_message: FAILED_TO_RUN_PIPELINE)
end
# Pipelined commands return an array with all results. If they differ, log an error
if result_non_default && result_non_default != result_default
error = PipelinedDiffError.new(result_non_default, result_default)
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_default
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)
# Run wrapped pipeline for each instance individually so that the fan-out is distinct.
# If both primary and secondary are Redis Clusters, the slot-node distribution could
# be different.
#
# We ignore args and kwargs since `pipelined` does not accept arguments
# See https://github.com/redis/redis-rb/blob/v4.8.0/lib/redis.rb#L164
if command_name.to_s == 'pipelined' && redis_instance._client.instance_of?(::Redis::Cluster)
return Gitlab::Redis::CrossSlot::Pipeline.new(redis_instance).pipelined(&block)
end
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)
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
|