Class: Aerospike::Cluster
- Inherits:
-
Object
- Object
- Aerospike::Cluster
show all
- Defined in:
- lib/aerospike/cluster.rb,
lib/aerospike/cluster/find_node.rb,
lib/aerospike/cluster/create_connection.rb,
lib/aerospike/cluster/find_nodes_to_remove.rb
Defined Under Namespace
Modules: CreateConnection, FindNode, FindNodesToRemove
Instance Attribute Summary collapse
Instance Method Summary
collapse
-
#add_alias(host, node) ⇒ Object
-
#add_aliases(node) ⇒ Object
-
#add_cluster_config_change_listener(listener) ⇒ Object
-
#add_nodes(nodes_to_add) ⇒ Object
-
#add_nodes_copy(nodes_to_add) ⇒ Object
-
#add_seeds(hosts) ⇒ Object
-
#batch_read_node(partition, replica_policy) ⇒ Object
Returns a node on the cluster for read operations.
-
#change_password(user, password) ⇒ Object
-
#close ⇒ Object
Closes all cached connections to the cluster nodes and stops the tend thread.
-
#connect ⇒ Object
-
#connected? ⇒ Boolean
-
#create_connection(host) ⇒ Object
-
#create_node(nv) ⇒ Object
-
#credentials_given? ⇒ Boolean
-
#find_alias(aliass) ⇒ Object
-
#find_node_by_name(node_name) ⇒ Object
-
#find_node_in_partition_map(filter) ⇒ Object
-
#find_node_name(list, name) ⇒ Object
Finds a node by name in a list of nodes.
-
#find_nodes_to_remove(refresh_count) ⇒ Object
-
#get_node_by_name(node_name) ⇒ Object
Find a node by name and returns an error if not found.
-
#get_node_for_key(replica_policy, key) ⇒ Object
-
#initialize(policy, hosts) ⇒ Cluster
constructor
A new instance of Cluster.
-
#initialize_tls_host_names(hosts) ⇒ Object
-
#inspect ⇒ Object
-
#launch_tend_thread ⇒ Object
-
#log_tend_stats(nodes) ⇒ Object
-
#master_node(partition) ⇒ Object
Returns a node on the cluster for read operations.
-
#master_proles_node(partition) ⇒ Object
Returns a node on the cluster for read operations.
-
#node_exists(search, node_list) ⇒ Object
-
#node_partitions(node, namespace) ⇒ Object
Returns partitions pertaining to a node.
-
#nodes ⇒ Object
Returns a list of all nodes in the cluster.
-
#notify_cluster_config_changed ⇒ Object
-
#partitions ⇒ Object
-
#rack_node(partition, seq) ⇒ Object
Returns a node on the cluster.
-
#random_node ⇒ Object
Returns a random node on the cluster.
-
#read_node(partition, replica_policy, seq) ⇒ Object
Returns a node on the cluster for read operations.
-
#refresh_nodes ⇒ Object
Refresh status of all nodes in cluster.
-
#remove_alias(aliass) ⇒ Object
-
#remove_cluster_config_change_listener(listener) ⇒ Object
-
#remove_nodes(nodes_to_remove) ⇒ Object
-
#remove_nodes_copy(nodes_to_remove) ⇒ Object
-
#request_info(policy, *commands) ⇒ Object
-
#request_node_info(node, policy, *commands) ⇒ Object
-
#reset_session_info ⇒ Object
-
#seed_nodes ⇒ Object
-
#seeds ⇒ Object
-
#sequence_node(partition, seq) ⇒ Object
Returns a random node on the cluster.
-
#session_valid? ⇒ Boolean
-
#set_nodes(nodes) ⇒ Object
-
#set_partitions(part_map) ⇒ Object
-
#supports_feature?(feature) ⇒ Boolean
-
#supports_peers_protocol? ⇒ Boolean
-
#tend ⇒ Object
Check health of all nodes in cluster.
-
#tls_enabled? ⇒ Boolean
-
#update_cluster_features ⇒ Object
-
#update_partitions(parser) ⇒ Object
-
#wait_till_stablized ⇒ Object
Constructor Details
#initialize(policy, hosts) ⇒ Cluster
Returns a new instance of Cluster.
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
|
# File 'lib/aerospike/cluster.rb', line 29
def initialize(policy, hosts)
@client_policy = policy
@cluster_seeds = hosts
@fail_if_not_connected = policy.fail_if_not_connected
@connection_queue_size = policy.connection_queue_size
@connection_timeout = policy.timeout
@tend_interval = policy.tend_interval
@cluster_name = policy.cluster_name
@tls_options = policy.tls
@rack_aware = policy.rack_aware
@rack_id = policy.rack_id
@replica_index = Atomic.new(0)
@aliases = {}
@cluster_nodes = []
@partition_write_map = {}
@node_index = Atomic.new(0)
@features = Atomic.new(Set.new)
@closed = Atomic.new(true)
@mutex = Mutex.new
@cluster_config_change_listeners = Atomic.new([])
@old_node_count = 0
if policy.requires_authentication
@user = policy.user
@password = LoginCommand.hash_password(policy.password)
end
initialize_tls_host_names(hosts) if tls_enabled?
if policy.min_connections_per_node > policy.max_connections_per_node
raise Aerospike::Exceptions::Aerospike.new(Aerospike::ResultCode::PARAMETER_ERROR, "Invalid policy configuration: Minimum connections per node cannot be greater than maximum connections per node.")
end
end
|
Instance Attribute Details
#aliases ⇒ Object
Returns the value of attribute aliases.
26
27
28
|
# File 'lib/aerospike/cluster.rb', line 26
def aliases
@aliases
end
|
#client_policy ⇒ Object
Returns the value of attribute client_policy.
26
27
28
|
# File 'lib/aerospike/cluster.rb', line 26
def client_policy
@client_policy
end
|
#cluster_id ⇒ Object
Returns the value of attribute cluster_id.
26
27
28
|
# File 'lib/aerospike/cluster.rb', line 26
def cluster_id
@cluster_id
end
|
#cluster_name ⇒ Object
Returns the value of attribute cluster_name.
26
27
28
|
# File 'lib/aerospike/cluster.rb', line 26
def cluster_name
@cluster_name
end
|
#connection_queue_size ⇒ Object
Returns the value of attribute connection_queue_size.
26
27
28
|
# File 'lib/aerospike/cluster.rb', line 26
def connection_queue_size
@connection_queue_size
end
|
#connection_timeout ⇒ Object
Returns the value of attribute connection_timeout.
26
27
28
|
# File 'lib/aerospike/cluster.rb', line 26
def connection_timeout
@connection_timeout
end
|
#features ⇒ Object
Returns the value of attribute features.
26
27
28
|
# File 'lib/aerospike/cluster.rb', line 26
def features
@features
end
|
#password ⇒ Object
Returns the value of attribute password.
26
27
28
|
# File 'lib/aerospike/cluster.rb', line 26
def password
@password
end
|
#rack_aware ⇒ Object
Returns the value of attribute rack_aware.
27
28
29
|
# File 'lib/aerospike/cluster.rb', line 27
def rack_aware
@rack_aware
end
|
#rack_id ⇒ Object
Returns the value of attribute rack_id.
27
28
29
|
# File 'lib/aerospike/cluster.rb', line 27
def rack_id
@rack_id
end
|
#session_expiration ⇒ Object
Returns the value of attribute session_expiration.
27
28
29
|
# File 'lib/aerospike/cluster.rb', line 27
def session_expiration
@session_expiration
end
|
#session_token ⇒ Object
Returns the value of attribute session_token.
27
28
29
|
# File 'lib/aerospike/cluster.rb', line 27
def session_token
@session_token
end
|
#tls_options ⇒ Object
Returns the value of attribute tls_options.
26
27
28
|
# File 'lib/aerospike/cluster.rb', line 26
def tls_options
@tls_options
end
|
#user ⇒ Object
Returns the value of attribute user.
26
27
28
|
# File 'lib/aerospike/cluster.rb', line 26
def user
@user
end
|
Instance Method Details
#add_alias(host, node) ⇒ Object
569
570
571
572
573
574
575
|
# File 'lib/aerospike/cluster.rb', line 569
def add_alias(host, node)
if host && node
@mutex.synchronize do
@aliases[host] = node
end
end
end
|
#add_aliases(node) ⇒ Object
619
620
621
622
623
624
625
|
# File 'lib/aerospike/cluster.rb', line 619
def add_aliases(node)
node.aliases.each do |aliass|
@aliases[aliass] = node
end
end
|
#add_cluster_config_change_listener(listener) ⇒ Object
349
350
351
352
353
|
# File 'lib/aerospike/cluster.rb', line 349
def add_cluster_config_change_listener(listener)
@cluster_config_change_listeners.update do |listeners|
listeners.push(listener)
end
end
|
#add_nodes(nodes_to_add) ⇒ Object
610
611
612
613
614
615
616
617
|
# File 'lib/aerospike/cluster.rb', line 610
def add_nodes(nodes_to_add)
nodes_to_add.each do |node|
add_aliases(node)
end
add_nodes_copy(nodes_to_add)
end
|
#add_nodes_copy(nodes_to_add) ⇒ Object
627
628
629
630
631
|
# File 'lib/aerospike/cluster.rb', line 627
def add_nodes_copy(nodes_to_add)
@mutex.synchronize do
@cluster_nodes.concat(nodes_to_add)
end
end
|
#add_seeds(hosts) ⇒ Object
102
103
104
105
106
|
# File 'lib/aerospike/cluster.rb', line 102
def add_seeds(hosts)
@mutex.synchronize do
@cluster_seeds.concat(hosts)
end
end
|
#batch_read_node(partition, replica_policy) ⇒ Object
Returns a node on the cluster for read operations
#change_password(user, password) ⇒ Object
344
345
346
347
|
# File 'lib/aerospike/cluster.rb', line 344
def change_password(user, password)
@password = password if @user == user
end
|
#close ⇒ Object
Closes all cached connections to the cluster nodes and stops the tend thread
301
302
303
304
305
306
307
308
|
# File 'lib/aerospike/cluster.rb', line 301
def close
return if @closed.value
@closed.value = true
@tend_thread.kill
nodes.each(&:close)
end
|
#connected? ⇒ Boolean
114
115
116
117
118
|
# File 'lib/aerospike/cluster.rb', line 114
def connected?
node_array = nodes
(node_array.length > 0) && !@closed.value
end
|
#create_connection(host) ⇒ Object
#create_node(nv) ⇒ Object
585
586
587
588
589
|
# File 'lib/aerospike/cluster.rb', line 585
def create_node(nv)
node = ::Aerospike::Node.new(self, nv)
node.fill_connection_pool_up_to(@client_policy.min_connections_per_node)
node
end
|
#credentials_given? ⇒ Boolean
79
80
81
|
# File 'lib/aerospike/cluster.rb', line 79
def credentials_given?
!(@user.nil? || @user.empty?)
end
|
#find_alias(aliass) ⇒ Object
310
311
312
313
314
|
# File 'lib/aerospike/cluster.rb', line 310
def find_alias(aliass)
@mutex.synchronize do
@aliases[aliass]
end
end
|
#find_node_by_name(node_name) ⇒ Object
695
696
697
|
# File 'lib/aerospike/cluster.rb', line 695
def find_node_by_name(node_name)
nodes.detect{|node| node.name == node_name }
end
|
#find_node_in_partition_map(filter) ⇒ Object
599
600
601
602
603
604
605
606
607
608
|
# File 'lib/aerospike/cluster.rb', line 599
def find_node_in_partition_map(filter)
partitions_list = partitions
partitions_list.values.each do |replica_array|
replica_array.get.each do |node_array|
return true if node_array.value.any? { |node| node == filter }
end
end
false
end
|
#find_node_name(list, name) ⇒ Object
Finds a node by name in a list of nodes
565
566
567
|
# File 'lib/aerospike/cluster.rb', line 565
def find_node_name(list, name)
list.any? { |node| node.name == name }
end
|
#find_nodes_to_remove(refresh_count) ⇒ Object
595
596
597
|
# File 'lib/aerospike/cluster.rb', line 595
def find_nodes_to_remove(refresh_count)
FindNodesToRemove.(self, refresh_count)
end
|
#get_node_by_name(node_name) ⇒ Object
Find a node by name and returns an error if not found
292
293
294
295
296
297
298
|
# File 'lib/aerospike/cluster.rb', line 292
def get_node_by_name(node_name)
node = find_node_by_name(node_name)
raise Aerospike::Exceptions::InvalidNode unless node
node
end
|
#get_node_for_key(replica_policy, key) ⇒ Object
239
240
241
242
|
# File 'lib/aerospike/cluster.rb', line 239
def get_node_for_key(replica_policy, key)
partition = Partition.new_by_key(key)
batch_read_node(partition, replica_policy)
end
|
#initialize_tls_host_names(hosts) ⇒ Object
96
97
98
99
100
|
# File 'lib/aerospike/cluster.rb', line 96
def initialize_tls_host_names(hosts)
hosts.each do |host|
host.tls_name ||= cluster_id.nil? ? host.name : cluster_id
end
end
|
#inspect ⇒ Object
361
362
363
|
# File 'lib/aerospike/cluster.rb', line 361
def inspect
"#<Aerospike::Cluster @cluster_nodes=#{@cluster_nodes}>"
end
|
#launch_tend_thread ⇒ Object
365
366
367
368
369
370
371
372
373
374
375
376
377
378
|
# File 'lib/aerospike/cluster.rb', line 365
def launch_tend_thread
@tend_thread = Thread.new do
Thread.current.abort_on_exception = false
loop do
begin
tend
sleep(@tend_interval / 1000.0)
rescue => e
Aerospike.logger.error("Exception occured during tend: #{e}")
Aerospike.logger.debug { e.backtrace.join("\n") }
end
end
end
end
|
#log_tend_stats(nodes) ⇒ Object
454
455
456
457
458
459
|
# File 'lib/aerospike/cluster.rb', line 454
def log_tend_stats(nodes)
diff = nodes.size - @old_node_count
action = "#{diff.abs} #{diff.abs == 1 ? "node has" : "nodes have"} #{diff > 0 ? "joined" : "left"} the cluster."
Aerospike.logger.info("Tend finished. #{action} Old node count: #{@old_node_count}, New node count: #{nodes.size}")
@old_node_count = nodes.size
end
|
#master_node(partition) ⇒ Object
Returns a node on the cluster for read operations
155
156
157
158
159
160
161
162
163
164
165
166
167
|
# File 'lib/aerospike/cluster.rb', line 155
def master_node(partition)
partition_map = partitions
replica_array = partition_map[partition.namespace]
raise Aerospike::Exceptions::InvalidNamespace("namespace not found in the partition map") if !replica_array
node_array = (replica_array.get)[0]
raise Aerospike::Exceptions::InvalidNamespace("namespace not found in the partition map") if !node_array
node = (node_array.get)[partition.partition_id]
raise Aerospike::Exceptions::InvalidNode if !node || !node.active?
node
end
|
#master_proles_node(partition) ⇒ Object
Returns a node on the cluster for read operations
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
|
# File 'lib/aerospike/cluster.rb', line 202
def master_proles_node(partition)
partition_map = partitions
replica_array = partition_map[partition.namespace]
raise Aerospike::Exceptions::InvalidNamespace("namespace not found in the partition map") if !replica_array
replica_array = replica_array.get
node = nil
for replica in replica_array
idx = (@replica_index.update{|v| v.succ} % replica_array.size).abs
node = (replica_array[idx].get)[partition.partition_id]
return node if node && node.active?
end
raise Aerospike::Exceptions::InvalidNode
end
|
#node_exists(search, node_list) ⇒ Object
691
692
693
|
# File 'lib/aerospike/cluster.rb', line 691
def node_exists(search, node_list)
node_list.any? {|node| node == search }
end
|
#node_partitions(node, namespace) ⇒ Object
Returns partitions pertaining to a node
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
|
# File 'lib/aerospike/cluster.rb', line 245
def node_partitions(node, namespace)
res = []
partition_map = partitions
replica_array = partition_map[namespace]
raise Aerospike::Exceptions::InvalidNamespace("namespace not found in the partition map") if !replica_array
node_array = (replica_array.get)[0]
raise Aerospike::Exceptions::InvalidNamespace("namespace not found in the partition map") if !node_array
pid = 0
for tnode in node_array.get
res << pid if node == tnode
pid+=1
end
res
end
|
#nodes ⇒ Object
Returns a list of all nodes in the cluster
284
285
286
287
288
289
|
# File 'lib/aerospike/cluster.rb', line 284
def nodes
@mutex.synchronize do
@cluster_nodes.dup
end
end
|
#notify_cluster_config_changed ⇒ Object
503
504
505
506
507
508
|
# File 'lib/aerospike/cluster.rb', line 503
def notify_cluster_config_changed
listeners = @cluster_config_change_listeners.get
listeners.each do |listener|
listener.send(:cluster_config_changed, self)
end
end
|
#partitions ⇒ Object
516
517
518
519
520
521
522
523
|
# File 'lib/aerospike/cluster.rb', line 516
def partitions
res = nil
@mutex.synchronize do
res = @partition_write_map
end
res
end
|
#rack_node(partition, seq) ⇒ Object
Returns a node on the cluster
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
|
# File 'lib/aerospike/cluster.rb', line 170
def rack_node(partition, seq)
partition_map = partitions
replica_array = partition_map[partition.namespace]
raise Aerospike::Exceptions::InvalidNamespace("namespace not found in the partition map") if !replica_array
replica_array = replica_array.get
is_retry = seq.value > -1
node = nil
fallback = nil
for i in 1..replica_array.length
idx = (seq.update{|v| v.succ} % replica_array.size).abs
node = (replica_array[idx].get)[partition.partition_id]
next if !node
fallback = node
return fallback if is_retry && fallback && i == replica_array.length
return node if node && node.active? && node.has_rack(partition.namespace, @rack_id)
end
return fallback if fallback
raise Aerospike::Exceptions::InvalidNode
end
|
#random_node ⇒ Object
Returns a random node on the cluster
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
|
# File 'lib/aerospike/cluster.rb', line 266
def random_node
node_array = nodes
length = node_array.length
i = 0
while i < length
idx = (@node_index.update{ |v| v.succ } % node_array.length).abs
node = node_array[idx]
return node if node.active?
i = i.succ
end
raise Aerospike::Exceptions::InvalidNode
end
|
#read_node(partition, replica_policy, seq) ⇒ Object
Returns a node on the cluster for read operations
#refresh_nodes ⇒ Object
Refresh status of all nodes in cluster. Adds new nodes and/or removes unhealty ones
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
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
|
# File 'lib/aerospike/cluster.rb', line 395
def refresh_nodes
cluster_config_changed = false
nodes = self.nodes
if nodes.empty?
seed_nodes
cluster_config_changed = true
nodes = self.nodes
end
peers = Peers.new
nodes.each do |node|
node.refresh_reset
end
peers.use_peers = supports_peers_protocol?
nodes.each do |node|
node.refresh_info(peers)
end
if peers.generation_changed?
peers.reset_refresh_count!
nodes.each do |node|
node.refresh_peers(peers)
end
end
nodes.each do |node|
node.refresh_partitions(peers) if node.partition_generation.changed?
node.refresh_racks if node.rebalance_generation.changed?
end
if peers.generation_changed? || !peers.use_peers?
nodes_to_remove = find_nodes_to_remove(peers.refresh_count)
if nodes_to_remove.any?
remove_nodes(nodes_to_remove)
cluster_config_changed = true
end
end
if peers.nodes.any?
add_nodes(peers.nodes.values)
cluster_config_changed = true
end
cluster_config_changed
end
|
#remove_alias(aliass) ⇒ Object
577
578
579
580
581
582
583
|
# File 'lib/aerospike/cluster.rb', line 577
def remove_alias(aliass)
if aliass
@mutex.synchronize do
@aliases.delete(aliass)
end
end
end
|
#remove_cluster_config_change_listener(listener) ⇒ Object
355
356
357
358
359
|
# File 'lib/aerospike/cluster.rb', line 355
def remove_cluster_config_change_listener(listener)
@cluster_config_change_listeners.update do |listeners|
listeners.delete(listener)
end
end
|
#remove_nodes(nodes_to_remove) ⇒ Object
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
|
# File 'lib/aerospike/cluster.rb', line 633
def remove_nodes(nodes_to_remove)
nodes_to_remove.each do |node|
node.aliases.each do |aliass|
Aerospike.logger.debug("Removing alias #{aliass}")
remove_alias(aliass)
end
node.close
end
remove_nodes_copy(nodes_to_remove)
end
|
#remove_nodes_copy(nodes_to_remove) ⇒ Object
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
|
# File 'lib/aerospike/cluster.rb', line 661
def remove_nodes_copy(nodes_to_remove)
nodes_list = nodes
node_array = []
count = 0
nodes_list.each do |node|
if node_exists(node, nodes_to_remove)
Aerospike.logger.info("Removed node `#{node}`")
else
node_array[count] = node
count += 1
end
end
if count < node_array.length
Aerospike.logger.warn("Node remove mismatch. Expected #{node_array.length}, Received #{count}")
node_array = node_array.dup[0..count-1]
end
set_nodes(node_array)
end
|
#request_info(policy, *commands) ⇒ Object
321
322
323
324
325
326
327
|
# File 'lib/aerospike/cluster.rb', line 321
def request_info(policy, *commands)
node = random_node
conn = node.get_connection(policy.timeout)
Info.request(conn, *commands).tap do
node.put_connection(conn)
end
end
|
#request_node_info(node, policy, *commands) ⇒ Object
329
330
331
332
333
334
|
# File 'lib/aerospike/cluster.rb', line 329
def request_node_info(node, policy, *commands)
conn = node.get_connection(policy.timeout)
Info.request(conn, *commands).tap do
node.put_connection(conn)
end
end
|
#reset_session_info ⇒ Object
87
88
89
90
|
# File 'lib/aerospike/cluster.rb', line 87
def reset_session_info
@session_token = nil
@session_expiration = nil
end
|
#seed_nodes ⇒ Object
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
|
# File 'lib/aerospike/cluster.rb', line 525
def seed_nodes
seed_array = seeds
Aerospike.logger.info("Seeding the cluster. Seeds count: #{seed_array.length}")
list = []
seed_array.each do |seed|
begin
seed_node_validator = NodeValidator.new(self, seed, @connection_timeout, @cluster_name, tls_options)
rescue => e
Aerospike.logger.error("Seed #{seed} failed: #{e}\n#{e.backtrace.join("\n")}")
next
end
nv = nil
seed_node_validator.aliases.each do |aliass|
if aliass == seed
nv = seed_node_validator
else
begin
nv = NodeValidator.new(self, aliass, @connection_timeout, @cluster_name, tls_options)
rescue => e
Aerospike.logger.error("Seed #{seed} failed: #{e}")
next
end
end
next if find_node_name(list, nv.name)
node = create_node(nv)
add_aliases(node)
list << node
end
end
add_nodes_copy(list) if list.length > 0
end
|
#seeds ⇒ Object
108
109
110
111
112
|
# File 'lib/aerospike/cluster.rb', line 108
def seeds
@mutex.synchronize do
@cluster_seeds.dup
end
end
|
#sequence_node(partition, seq) ⇒ Object
Returns a random node on the cluster
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
|
# File 'lib/aerospike/cluster.rb', line 221
def sequence_node(partition, seq)
partition_map = partitions
replica_array = partition_map[partition.namespace]
raise Aerospike::Exceptions::InvalidNamespace("namespace not found in the partition map") if !replica_array
replica_array = replica_array.get
node = nil
for replica in replica_array
idx = (seq.update{|v| v.succ} % replica_array.size).abs
node = (replica_array[idx].get)[partition.partition_id]
return node if node && node.active?
end
raise Aerospike::Exceptions::InvalidNode
end
|
#session_valid? ⇒ Boolean
83
84
85
|
# File 'lib/aerospike/cluster.rb', line 83
def session_valid?
@session_token && @session_expiration && @session_expiration.to_i < Time.now.to_i
end
|
#set_nodes(nodes) ⇒ Object
654
655
656
657
658
659
|
# File 'lib/aerospike/cluster.rb', line 654
def set_nodes(nodes)
@mutex.synchronize do
@cluster_nodes = nodes
end
end
|
#set_partitions(part_map) ⇒ Object
510
511
512
513
514
|
# File 'lib/aerospike/cluster.rb', line 510
def set_partitions(part_map)
@mutex.synchronize do
@partition_write_map = part_map
end
end
|
#supports_feature?(feature) ⇒ Boolean
336
337
338
|
# File 'lib/aerospike/cluster.rb', line 336
def supports_feature?(feature)
@features.get.include?(feature.to_s)
end
|
#supports_peers_protocol? ⇒ Boolean
340
341
342
|
# File 'lib/aerospike/cluster.rb', line 340
def supports_peers_protocol?
nodes.all? { |node| node.supports_feature?(Aerospike::Features::PEERS) }
end
|
#tend ⇒ Object
Check health of all nodes in cluster
381
382
383
384
385
386
387
388
389
390
391
|
# File 'lib/aerospike/cluster.rb', line 381
def tend
was_changed = refresh_nodes
return unless was_changed
update_cluster_features
notify_cluster_config_changed
log_tend_stats(nodes)
end
|
#tls_enabled? ⇒ Boolean
92
93
94
|
# File 'lib/aerospike/cluster.rb', line 92
def tls_enabled?
!tls_options.nil? && tls_options[:enable] != false
end
|
#update_cluster_features ⇒ Object
495
496
497
498
499
500
501
|
# File 'lib/aerospike/cluster.rb', line 495
def update_cluster_features
@features.update do
node_features = nodes.map(&:features)
node_features.reduce(&:intersection) || Set.new
end
end
|
#update_partitions(parser) ⇒ Object
316
317
318
319
|
# File 'lib/aerospike/cluster.rb', line 316
def update_partitions(parser)
nmap = parser.update_partitions(partitions)
set_partitions(nmap) if nmap
end
|
#wait_till_stablized ⇒ Object
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
|
# File 'lib/aerospike/cluster.rb', line 461
def wait_till_stablized
count = -1
done = false
thr = Thread.new do
loop do
tend
break if count == nodes.length
break if done
sleep(0.001)
count = nodes.length
end
end
thr.join(@connection_timeout * 10)
done = true
sleep(0.001)
thr.kill if thr.alive?
@closed.value = false if @cluster_nodes.length > 0
end
|