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
|
# 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 secondary store.'
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
READ_COMMANDS = %i(
get
mget
smembers
scard
).freeze
WRITE_COMMANDS = %i(
set
setnx
setex
sadd
srem
del
flushdb
rpush
).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_COMMANDS.each do |name|
define_method(name) do |*args, &block|
if use_primary_and_secondary_stores?
read_command(name, *args, &block)
else
default_store.send(name, *args, &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_enabled?("use_primary_and_secondary_stores_for")
end
def use_primary_store_as_default?
feature_enabled?("use_primary_store_as_default_for")
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
private
# @return [Boolean]
def feature_enabled?(prefix)
feature_table_exists? &&
Feature.enabled?("#{prefix}_#{instance_name.underscore}") &&
!same_redis_store?
end
# @return [Boolean]
def feature_table_exists?
Feature::FlipperFeature.table_exists?
rescue StandardError
false
end
def default_store
use_primary_store_as_default? ? primary_store : secondary_store
end
def log_method_missing(command_name, *_args)
return if SKIP_LOG_METHOD_MISSING_FOR_COMMANDS.include?(command_name)
log_error(MethodMissingError.new, command_name)
increment_method_missing_count(command_name)
end
def read_command(command_name, *args, &block)
if @instance
send_command(@instance, command_name, *args, &block)
else
read_one_with_fallback(command_name, *args, &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, &block)
begin
value = send_command(primary_store, command_name, *args, &block)
rescue StandardError => e
log_error(e, command_name,
multi_store_error_message: FAILED_TO_READ_ERROR_MESSAGE)
end
value || fallback_read(command_name, *args, &block)
end
def fallback_read(command_name, *args, &block)
value = send_command(secondary_store, command_name, *args, &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
|