Class: OpenC3::Store

Inherits:
Object show all
Defined in:
lib/openc3/utilities/store_autoload.rb

Direct Known Subclasses

EphemeralStore

Constant Summary collapse

DB_SHARD_CACHE_TIMEOUT =

seconds

60
REDIS_BACKOFF_CAP =

cap/base for the equal-jitter reconnect backoff (seconds)

5.0
REDIS_BACKOFF_BASE = 
0.625
@@db_shard_cache =

DB_Shard cache: { "scope__target_name" => [db_shard_number, Time] }

{}
@@db_shard_cache_mutex = 
Mutex.new
@@instance_mutex =

Mutex used to ensure that only one instance is created

Mutex.new

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(pool_size = 10, db_shard: 0) ⇒ Store

Returns a new instance of Store.



118
119
120
121
122
123
124
 
# File 'lib/openc3/utilities/store_autoload.rb', line 118

def initialize(pool_size = 10, db_shard: 0)
  @redis_username = ENV['OPENC3_REDIS_USERNAME']
  @redis_key = ENV['OPENC3_REDIS_PASSWORD']
  hostname = ENV['OPENC3_REDIS_HOSTNAME'].to_s.gsub("SHARDNUM", db_shard.to_s)
  @redis_url = "redis://#{hostname}:#{ENV.fetch('OPENC3_REDIS_PORT', 6379)}"
  @redis_pool = StoreConnectionPool.new(size: pool_size) { build_redis() }
end

Dynamic Method Handling

This class handles dynamic methods through the method_missing method

#method_missing(message, *args, **kwargs, &block) ⇒ Object

Delegate all unknown methods to redis through the @redis_pool



114
115
116
 
# File 'lib/openc3/utilities/store_autoload.rb', line 114

def method_missing(message, *args, **kwargs, &block)
  @redis_pool.with { |redis| redis.public_send(message, *args, **kwargs, &block) }
end

Instance Attribute Details

#redis_pool ⇒ Object (readonly)

Returns the value of attribute redis_pool.



61
62
63
 
# File 'lib/openc3/utilities/store_autoload.rb', line 61

def redis_pool
  @redis_pool
end

#redis_url ⇒ Object (readonly)

Returns the value of attribute redis_url.



60
61
62
 
# File 'lib/openc3/utilities/store_autoload.rb', line 60

def redis_url
  @redis_url
end

Class Method Details

.db_shard_for_target(target_name, scope: "DEFAULT") ⇒ Object

Look up the db_shard number for a target with a 1-minute cache. Reads directly from Redis db_shard 0 to avoid circular deps with TargetModel. Non-target-specific data (nil target_name) always returns db_shard 0.



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
 
# File 'lib/openc3/utilities/store_autoload.rb', line 66

def self.db_shard_for_target(target_name, scope: "DEFAULT")
  return 0 unless target_name

  cache_key = "#{scope}__#{target_name}"
  now = Time.now

  @@db_shard_cache_mutex.synchronize do
    cached = @@db_shard_cache[cache_key]
    if cached
      db_shard, cached_at = cached
      return db_shard if (now - cached_at) < DB_SHARD_CACHE_TIMEOUT
    end
  end

  begin
    json = Store.instance(db_shard: 0).hget("#{scope}__openc3_targets", target_name)
    db_shard = json ? JSON.parse(json)['db_shard'].to_i : 0
  rescue
    db_shard = 0
  end

  @@db_shard_cache_mutex.synchronize do
    @@db_shard_cache[cache_key] = [db_shard, now]
  end

  db_shard
end

.instance(pool_size = 100, db_shard: 0) ⇒ Object

Get the singleton instance



95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/openc3/utilities/store_autoload.rb', line 95

def self.instance(pool_size = 100, db_shard: 0)
  # Logger.level = Logger::DEBUG
  @instances ||= []
  the_instance = @instances[db_shard]
  return the_instance if the_instance

  @@instance_mutex.synchronize do
    @instances ||= []
    @instances[db_shard] ||= self.new(pool_size, db_shard: db_shard)
    return @instances[db_shard]
  end
end

.method_missing(message, *args, **kwargs, &block) ⇒ Object

Delegate all unknown class methods to delegate to the instance



109
110
111
 
# File 'lib/openc3/utilities/store_autoload.rb', line 109

def self.method_missing(message, *args, **kwargs, &block)
  self.instance.public_send(message, *args, **kwargs, &block)
end

Instance Method Details

#build_redis ⇒ Object 



130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/openc3/utilities/store_autoload.rb', line 130

def build_redis
  # reconnect_attempts retries the connection a few times with equal-jitter
  # backoff so a transient network blip is handled inside the client instead
  # of immediately surfacing a connection error to callers. The jitter
  # de-syncs many clients retrying the same blip to avoid a thundering herd
  # on recovery.
  #
  # This mirrors the Python store's Retry(EqualJitterBackoff(cap: 5, base:
  # 0.625), 3): per-retry backoff tops out at 5s on the final (3rd) retry
  # (~0.6-1.25s, ~1.25-2.5s, ~2.5-5s). redis-rb takes a fixed Array of delays
  # (no per-failure backoff callable), so we sample the jittered delays once
  # per connection here.
  # Connection, read, and write timeouts are left as the default: 1s
  return Redis.new(
    url: @redis_url,
    username: @redis_username,
    password: @redis_key,
    reconnect_attempts: reconnect_backoff_delays()
  )
end

#get_last_offset(topic) ⇒ Object 



197
198
199
200
201
202
203
204
205
206
 
# File 'lib/openc3/utilities/store_autoload.rb', line 197

def get_last_offset(topic)
  @redis_pool.with do |redis|
    result = redis.xrevrange(topic, count: 1)
    if result and result[0] and result[0][0]
      result[0][0]
    else
      "0-0"
    end
  end
end

#get_newest_message(topic) ⇒ Object 



183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'lib/openc3/utilities/store_autoload.rb', line 183

def get_newest_message(topic)
  @redis_pool.with do |redis|
    # Default in xrevrange is range end '+', start '-' which means get all
    # elements from higher ID to lower ID and since we're limiting to 1
    # we get the last element. See https://redis.io/commands/xrevrange.
    result = redis.xrevrange(topic, count: 1)
    if result and result.length > 0
      return result[0]
    else
      return nil
    end
  end
end

#get_oldest_message(topic) ⇒ Object

Stream APIs



172
173
174
175
176
177
178
179
180
181
 
# File 'lib/openc3/utilities/store_autoload.rb', line 172

def get_oldest_message(topic)
  @redis_pool.with do |redis|
    result = redis.xrange(topic, count: 1)
    if result and result.length > 0
      return result[0]
    else
      return nil
    end
  end
end

#read_topics(topics, offsets = nil, timeout_ms = 1000, count = nil) ⇒ Object 



228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
 
# File 'lib/openc3/utilities/store_autoload.rb', line 228

def read_topics(topics, offsets = nil, timeout_ms = 1000, count = nil)
  return {} if topics.empty?
  Thread.current[:topic_offsets] ||= {}
  topic_offsets = Thread.current[:topic_offsets]
  begin
    # Logger.debug "read_topics: #{topics}, #{offsets} pool:#{@redis_pool}"
    @redis_pool.with do |redis|
      offsets = update_topic_offsets(topics) unless offsets
      result = redis.xread(topics, offsets, block: timeout_ms, count: count)
      if result and result.length > 0
        result.each do |topic, messages|
          messages.each do |msg_id, msg_hash|
            topic_offsets[topic] = msg_id
            yield topic, msg_id, msg_hash, redis if block_given?
          end
        end
      end
      # Logger.debug "result:#{result}" if result and result.length > 0
      return result
    end
  rescue Redis::TimeoutError
    return {} # Should return an empty hash not array - xread returns a hash
  end
end

#reconnect_backoff_delays ⇒ Object

Equal-jitter backoff delays for 3 retries, matching Python's EqualJitterBackoff. Each retry's delay is randomized within the upper half of an exponentially-growing ceiling.



154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/openc3/utilities/store_autoload.rb', line 154

def reconnect_backoff_delays
  (1..3).map do |failures|
    # ceiling = exponential growth (base doubles each retry), clamped to cap.
    # For base=0.625, cap=5: failures 1,2,3 -> 1.25, 2.5, 5.0 seconds.
    # temp = half the ceiling: the guaranteed minimum wait for this retry.
    temp = [REDIS_BACKOFF_CAP, REDIS_BACKOFF_BASE * (2 ** failures)].min / 2.0
    # Final delay = fixed half (temp) + random half (rand*temp, in [0, temp)),
    # i.e. a value uniformly in [temp, 2*temp) = [ceiling/2, ceiling).
    # The fixed half keeps a sane floor; the random half de-syncs clients so
    # they don't all reconnect in lockstep (thundering herd) after a blip.
    temp + rand * temp
  end
end

#trim_topic(topic, minid, approximate = true, limit: 0) ⇒ Integer

Trims older entries of the redis stream if needed.

https://www.rubydoc.info/github/redis/redis-rb/Redis:xtrim

Examples:

Without options

store.trim_topic('MANGO__TOPIC', 1000)

With options

store.trim_topic('MANGO__TOPIC', 1000, approximate: 'true', limit: 0)

Parameters:

  • topic (String) —

    the stream key

  • minid (Integer) —

    Id to throw away data up to

  • approximate (Boolean) (defaults to: true) —

    whether to add ~ modifier of maxlen or not

  • limit (Boolean) (defaults to: 0) —

    number of items to return from the call

Returns:

  • (Integer) —

    the number of entries actually deleted



291
292
293
294
295
 
# File 'lib/openc3/utilities/store_autoload.rb', line 291

def trim_topic(topic, minid, approximate = true, limit: 0)
  @redis_pool.with do |redis|
    return redis.xtrim_minid(topic, minid, approximate: approximate, limit: limit)
  end
end

#update_topic_offsets(topics) ⇒ Object 



208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/openc3/utilities/store_autoload.rb', line 208

def update_topic_offsets(topics)
  offsets = []
  topics.each do |topic|
    # Normally we will just be grabbing the topic offset
    # this allows xread to get everything past this point
    Thread.current[:topic_offsets] ||= {}
    topic_offsets = Thread.current[:topic_offsets]
    last_id = topic_offsets[topic]
    if last_id
      offsets << last_id
    else
      # If there is no topic offset this is the first call.
      # Get the last offset ID so we'll start getting everything from now on
      offsets << get_last_offset(topic)
      topic_offsets[topic] = offsets[-1]
    end
  end
  return offsets
end

#write_topic(topic, msg_hash, id = '*', maxlen = nil, approximate = 'true') ⇒ String

Add new entry to the redis stream.

https://www.rubydoc.info/github/redis/redis-rb/Redis:xadd

Examples:

Without options

store.write_topic('MANGO__TOPIC', {'message' => 'something'})

With options

store.write_topic('MANGO__TOPIC', {'message' => 'something'}, id: '0-0', maxlen: 1000, approximate: 'true')

Parameters:

  • topic (String) —

    the stream / topic

  • msg_hash (Hash) —

    one or multiple field-value pairs

  • opts (Hash) —

    a customizable set of options

Returns:



270
271
272
273
274
275
 
# File 'lib/openc3/utilities/store_autoload.rb', line 270

def write_topic(topic, msg_hash, id = '*', maxlen = nil, approximate = 'true')
  id = '*' if id.nil?
  @redis_pool.with do |redis|
    return redis.xadd(topic, msg_hash, id: id, maxlen: maxlen, approximate: approximate)
  end
end