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, is_write: false) ⇒ 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
573
574
575
576
577
578
579
|
# File 'lib/aerospike/cluster.rb', line 573
def add_alias(host, node)
if host && node
@mutex.synchronize do
@aliases[host] = node
end
end
end
|
#add_aliases(node) ⇒ Object
623
624
625
626
627
628
629
|
# File 'lib/aerospike/cluster.rb', line 623
def add_aliases(node)
node.aliases.each do |aliass|
@aliases[aliass] = node
end
end
|
#add_cluster_config_change_listener(listener) ⇒ Object
353
354
355
356
357
|
# File 'lib/aerospike/cluster.rb', line 353
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
614
615
616
617
618
619
620
621
|
# File 'lib/aerospike/cluster.rb', line 614
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
631
632
633
634
635
|
# File 'lib/aerospike/cluster.rb', line 631
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
348
349
350
351
|
# File 'lib/aerospike/cluster.rb', line 348
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
305
306
307
308
309
310
311
312
|
# File 'lib/aerospike/cluster.rb', line 305
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
589
590
591
592
593
|
# File 'lib/aerospike/cluster.rb', line 589
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
314
315
316
317
318
|
# File 'lib/aerospike/cluster.rb', line 314
def find_alias(aliass)
@mutex.synchronize do
@aliases[aliass]
end
end
|
#find_node_by_name(node_name) ⇒ Object
699
700
701
|
# File 'lib/aerospike/cluster.rb', line 699
def find_node_by_name(node_name)
nodes.detect { |node| node.name == node_name }
end
|
#find_node_in_partition_map(filter) ⇒ Object
603
604
605
606
607
608
609
610
611
612
|
# File 'lib/aerospike/cluster.rb', line 603
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
569
570
571
|
# File 'lib/aerospike/cluster.rb', line 569
def find_node_name(list, name)
list.any? { |node| node.name == name }
end
|
#find_nodes_to_remove(refresh_count) ⇒ Object
599
600
601
|
# File 'lib/aerospike/cluster.rb', line 599
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
296
297
298
299
300
301
302
|
# File 'lib/aerospike/cluster.rb', line 296
def get_node_by_name(node_name)
node = find_node_by_name(node_name)
raise Aerospike::Exceptions::InvalidNode.new("node `#{node_name}` not found") unless node
node
end
|
#get_node_for_key(replica_policy, key, is_write: false) ⇒ Object
239
240
241
242
243
244
245
246
|
# File 'lib/aerospike/cluster.rb', line 239
def get_node_for_key(replica_policy, key, is_write: false)
partition = Partition.new_by_key(key)
if is_write
master_node(partition)
else
batch_read_node(partition, replica_policy)
end
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
365
366
367
|
# File 'lib/aerospike/cluster.rb', line 365
def inspect
"#<Aerospike::Cluster @cluster_nodes=#{@cluster_nodes}>"
end
|
#launch_tend_thread ⇒ Object
369
370
371
372
373
374
375
376
377
378
379
380
381
382
|
# File 'lib/aerospike/cluster.rb', line 369
def launch_tend_thread
@tend_thread = Thread.new do
Thread.current.abort_on_exception = false
loop do
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
|
#log_tend_stats(nodes) ⇒ Object
458
459
460
461
462
463
|
# File 'lib/aerospike/cluster.rb', line 458
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.new("namespace not found in the partition map") unless replica_array
node_array = replica_array.get[0]
raise Aerospike::Exceptions::InvalidNamespace.new("namespace not found in the partition map") unless node_array
node = node_array.get[partition.partition_id]
raise Aerospike::Exceptions::InvalidNode.new("no active node found") 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.new("namespace not found in the partition map") unless 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.new("no active node found")
end
|
#node_exists(search, node_list) ⇒ Object
695
696
697
|
# File 'lib/aerospike/cluster.rb', line 695
def node_exists(search, node_list)
node_list.any? { |node| node == search }
end
|
#node_partitions(node, namespace) ⇒ Object
Returns partitions pertaining to a node
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
|
# File 'lib/aerospike/cluster.rb', line 249
def node_partitions(node, namespace)
res = []
partition_map = partitions
replica_array = partition_map[namespace]
raise Aerospike::Exceptions::InvalidNamespace.new("namespace not found in the partition map") unless replica_array
node_array = replica_array.get[0]
raise Aerospike::Exceptions::InvalidNamespace.new("namespace not found in the partition map") unless 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
288
289
290
291
292
293
|
# File 'lib/aerospike/cluster.rb', line 288
def nodes
@mutex.synchronize do
@cluster_nodes.dup
end
end
|
#notify_cluster_config_changed ⇒ Object
507
508
509
510
511
512
|
# File 'lib/aerospike/cluster.rb', line 507
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
520
521
522
523
524
525
526
527
|
# File 'lib/aerospike/cluster.rb', line 520
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.new("namespace not found in the partition map") unless 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 unless 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.new("no active node found")
end
|
#random_node ⇒ Object
Returns a random node on the cluster
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
|
# File 'lib/aerospike/cluster.rb', line 270
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.new("no active node found")
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
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
453
454
455
456
|
# File 'lib/aerospike/cluster.rb', line 399
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
581
582
583
584
585
586
587
|
# File 'lib/aerospike/cluster.rb', line 581
def remove_alias(aliass)
if aliass
@mutex.synchronize do
@aliases.delete(aliass)
end
end
end
|
#remove_cluster_config_change_listener(listener) ⇒ Object
359
360
361
362
363
|
# File 'lib/aerospike/cluster.rb', line 359
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
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
|
# File 'lib/aerospike/cluster.rb', line 637
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
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
690
691
692
693
|
# File 'lib/aerospike/cluster.rb', line 665
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
325
326
327
328
329
330
331
|
# File 'lib/aerospike/cluster.rb', line 325
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
333
334
335
336
337
338
|
# File 'lib/aerospike/cluster.rb', line 333
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
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
563
564
565
566
|
# File 'lib/aerospike/cluster.rb', line 529
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.new("namespace not found in the partition map") unless 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.new("node active node found")
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
658
659
660
661
662
663
|
# File 'lib/aerospike/cluster.rb', line 658
def set_nodes(nodes)
@mutex.synchronize do
@cluster_nodes = nodes
end
end
|
#set_partitions(part_map) ⇒ Object
514
515
516
517
518
|
# File 'lib/aerospike/cluster.rb', line 514
def set_partitions(part_map)
@mutex.synchronize do
@partition_write_map = part_map
end
end
|
#supports_feature?(feature) ⇒ Boolean
340
341
342
|
# File 'lib/aerospike/cluster.rb', line 340
def supports_feature?(feature)
@features.get.include?(feature.to_s)
end
|
#supports_peers_protocol? ⇒ Boolean
344
345
346
|
# File 'lib/aerospike/cluster.rb', line 344
def supports_peers_protocol?
nodes.all? { |node| node.supports_feature?(Aerospike::Features::PEERS) }
end
|
#tend ⇒ Object
Check health of all nodes in cluster
385
386
387
388
389
390
391
392
393
394
395
|
# File 'lib/aerospike/cluster.rb', line 385
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
499
500
501
502
503
504
505
|
# File 'lib/aerospike/cluster.rb', line 499
def update_cluster_features
@features.update do
node_features = nodes.map(&:features)
node_features.reduce(&:intersection) || Set.new
end
end
|
#update_partitions(parser) ⇒ Object
320
321
322
323
|
# File 'lib/aerospike/cluster.rb', line 320
def update_partitions(parser)
nmap = parser.update_partitions(partitions)
set_partitions(nmap) if nmap
end
|
#wait_till_stablized ⇒ Object
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
494
495
496
497
|
# File 'lib/aerospike/cluster.rb', line 465
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
|