Class: Aerospike::Cluster

Inherits:
Object
  • Object
show all
Defined in:
lib/aerospike/cluster.rb,
lib/aerospike/cluster/find_node.rb,
lib/aerospike/cluster/create_connection.rb

Defined Under Namespace

Modules: CreateConnection, FindNode

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(policy, hosts) ⇒ Cluster

Returns a new instance of Cluster.



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
 
# File 'lib/aerospike/cluster.rb', line 32

def initialize(policy, hosts)
  @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
  @ssl_options = policy.ssl_options

  @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

  # setup auth info for cluster
  if policy.requires_authentication
    @user = policy.user
    @password = AdminCommand.hash_password(policy.password)
  end

  initialize_tls_host_names(hosts) if tls_enabled?
end

Instance Attribute Details

#aliases ⇒ Object (readonly)

Returns the value of attribute aliases.



29
30
31
 
# File 'lib/aerospike/cluster.rb', line 29

def aliases
  @aliases
end

#cluster_id ⇒ Object (readonly)

Returns the value of attribute cluster_id.



29
30
31
 
# File 'lib/aerospike/cluster.rb', line 29

def cluster_id
  @cluster_id
end

#cluster_name ⇒ Object (readonly)

Returns the value of attribute cluster_name.



30
31
32
 
# File 'lib/aerospike/cluster.rb', line 30

def cluster_name
  @cluster_name
end

#connection_queue_size ⇒ Object (readonly)

Returns the value of attribute connection_queue_size.



27
28
29
 
# File 'lib/aerospike/cluster.rb', line 27

def connection_queue_size
  @connection_queue_size
end

#connection_timeout ⇒ Object (readonly)

Returns the value of attribute connection_timeout.



27
28
29
 
# File 'lib/aerospike/cluster.rb', line 27

def connection_timeout
  @connection_timeout
end

#features ⇒ Object (readonly)

Returns the value of attribute features.



28
29
30
 
# File 'lib/aerospike/cluster.rb', line 28

def features
  @features
end

#password ⇒ Object (readonly)

Returns the value of attribute password.



27
28
29
 
# File 'lib/aerospike/cluster.rb', line 27

def password
  @password
end

#ssl_options ⇒ Object (readonly)

Returns the value of attribute ssl_options.



28
29
30
 
# File 'lib/aerospike/cluster.rb', line 28

def ssl_options
  @ssl_options
end

#user ⇒ Object (readonly)

Returns the value of attribute user.



27
28
29
 
# File 'lib/aerospike/cluster.rb', line 27

def user
  @user
end

Instance Method Details

#add_alias(host, node) ⇒ Object 



409
410
411
412
413
414
415
 
# File 'lib/aerospike/cluster.rb', line 409

def add_alias(host, node)
  if host && node
    @mutex.synchronize do
      @aliases[host] = node
    end
  end
end

#add_aliases(node) ⇒ Object 



494
495
496
497
498
499
500
 
# File 'lib/aerospike/cluster.rb', line 494

def add_aliases(node)
  # Add node's aliases to global alias set.
  # Aliases are only used in tend thread, so synchronization is not necessary.
  node.aliases.each do |aliass|
    @aliases[aliass] = node
  end
end

#add_cluster_config_change_listener(listener) ⇒ Object 



197
198
199
200
201
 
# File 'lib/aerospike/cluster.rb', line 197

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 



485
486
487
488
489
490
491
492
 
# File 'lib/aerospike/cluster.rb', line 485

def add_nodes(nodes_to_add)
  # Add all nodes at once to avoid copying entire array multiple times.
  nodes_to_add.each do |node|
    add_aliases(node)
  end

  add_nodes_copy(nodes_to_add)
end

#add_nodes_copy(nodes_to_add) ⇒ Object 



502
503
504
505
506
 
# File 'lib/aerospike/cluster.rb', line 502

def add_nodes_copy(nodes_to_add)
  @mutex.synchronize do
    @cluster_nodes.concat(nodes_to_add)
  end
end

#add_seeds(hosts) ⇒ Object 



87
88
89
90
91
 
# File 'lib/aerospike/cluster.rb', line 87

def add_seeds(hosts)
  @mutex.synchronize do
    @cluster_seeds.concat(hosts)
  end
end

#change_password(user, password) ⇒ Object 



192
193
194
195
 
# File 'lib/aerospike/cluster.rb', line 192

def change_password(user, password)
  # change password ONLY if the user is the same
  @password = password if @user == user
end

#close ⇒ Object

Closes all cached connections to the cluster nodes and stops the tend thread



153
154
155
156
157
158
159
160
 
# File 'lib/aerospike/cluster.rb', line 153

def close
  return if @closed.value
  # send close signal to maintenance channel
  @closed.value = true
  @tend_thread.kill

  nodes.each(&:close)
end

#connect ⇒ Object 



61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/aerospike/cluster.rb', line 61

def connect
  wait_till_stablized

  if @fail_if_not_connected && !connected?
    raise Aerospike::Exceptions::Aerospike, Aerospike::ResultCode::SERVER_NOT_AVAILABLE
  end

  launch_tend_thread

  Aerospike.logger.info('New cluster initialized and ready to be used...')
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


99
100
101
102
103
 
# File 'lib/aerospike/cluster.rb', line 99

def connected?
  # Must copy array reference for copy on write semantics to work.
  node_array = nodes
  (node_array.length > 0) && !@closed.value
end

#create_node(nv) ⇒ Object 



425
426
427
 
# File 'lib/aerospike/cluster.rb', line 425

def create_node(nv)
  ::Aerospike::Node.new(self, nv)
end

#credentials_given? ⇒ Boolean

Returns:

  • (Boolean) 


73
74
75
 
# File 'lib/aerospike/cluster.rb', line 73

def credentials_given?
  !(@user.nil? || @user.empty?)
end

#find_alias(aliass) ⇒ Object 



162
163
164
165
166
 
# File 'lib/aerospike/cluster.rb', line 162

def find_alias(aliass)
  @mutex.synchronize do
    @aliases[aliass]
  end
end

#find_node_by_name(node_name) ⇒ Object 



570
571
572
 
# File 'lib/aerospike/cluster.rb', line 570

def find_node_by_name(node_name)
  nodes.detect{|node| node.name == node_name }
end

#find_node_in_partition_map(filter) ⇒ Object 



476
477
478
479
480
481
482
483
 
# File 'lib/aerospike/cluster.rb', line 476

def find_node_in_partition_map(filter)
  partitions_list = partitions

  partitions_list.values.each do |node_array|
    return true if node_array.value.any? { |node| node == filter }
  end
  false
end

#find_node_name(list, name) ⇒ Object

Finds a node by name in a list of nodes



405
406
407
 
# File 'lib/aerospike/cluster.rb', line 405

def find_node_name(list, name)
  list.any? { |node| node.name == name }
end

#find_nodes_to_remove(refresh_count) ⇒ Object 



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
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
 
# File 'lib/aerospike/cluster.rb', line 429

def find_nodes_to_remove(refresh_count)
  node_list = nodes

  remove_list = []

  node_list.each do |node|
    if !node.active?
      # Inactive nodes must be removed.
      remove_list << node
      next
    end

    case node_list.length
    when 1
      # Single node clusters rely solely on node health.
      remove_list << node if node.unhealthy?

    when 2
      # Two node clusters require at least one successful refresh before removing.
      if refresh_count == 2 && node.reference_count.value == 0 && !node.responded?
        # Node is not referenced nor did it respond.
        remove_list << node
      end

    else
      # Multi-node clusters require two successful node refreshes before removing.
      if refresh_count >= 2 && node.reference_count.value == 0
        # Node is not referenced by other nodes.
        # Check if node responded to info request.
        if node.responded?
          # Node is alive, but not referenced by other nodes.  Check if mapped.
          unless find_node_in_partition_map(node)
            # Node doesn't have any partitions mapped to it.
            # There is not point in keeping it in the cluster.
            remove_list << node
          end
        else
          # Node not responding. Remove it.
          remove_list << node
        end
      end
    end
  end

  remove_list
end

#get_node(partition) ⇒ Object 



105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/aerospike/cluster.rb', line 105

def get_node(partition)
  # Must copy hashmap reference for copy on write semantics to work.
  nmap = partitions
  if node_array = nmap[partition.namespace]
    node = node_array.value[partition.partition_id]

    return node if node && node.active?
  end

  random_node
end

#get_node_by_name(node_name) ⇒ Object

Find a node by name and returns an error if not found

Raises:



144
145
146
147
148
149
150
 
# File 'lib/aerospike/cluster.rb', line 144

def get_node_by_name(node_name)
  node = find_node_by_name(node_name)

  raise Aerospike::Exceptions::InvalidNode unless node

  node
end

#initialize_tls_host_names(hosts) ⇒ Object 



81
82
83
84
85
 
# File 'lib/aerospike/cluster.rb', line 81

def initialize_tls_host_names(hosts)
  hosts.each do |host|
    host.tls_name ||= cluster_id.nil? ? host.name : cluster_id
  end
end

#inspect ⇒ Object 



209
210
211
 
# File 'lib/aerospike/cluster.rb', line 209

def inspect
  "#<Aerospike::Cluster @cluster_nodes=#{@cluster_nodes}>"
end

#launch_tend_thread ⇒ Object 



213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/aerospike/cluster.rb', line 213

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 



298
299
300
301
302
303
 
# File 'lib/aerospike/cluster.rb', line 298

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

#node_exists(search, node_list) ⇒ Object 



566
567
568
 
# File 'lib/aerospike/cluster.rb', line 566

def node_exists(search, node_list)
  node_list.any? {|node| node == search }
end

#nodes ⇒ Object

Returns a list of all nodes in the cluster



136
137
138
139
140
141
 
# File 'lib/aerospike/cluster.rb', line 136

def nodes
  @mutex.synchronize do
    # Must copy array reference for copy on write semantics to work.
    @cluster_nodes.dup
  end
end

#notify_cluster_config_changed ⇒ Object 



343
344
345
346
347
348
 
# File 'lib/aerospike/cluster.rb', line 343

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 



356
357
358
359
360
361
362
363
 
# File 'lib/aerospike/cluster.rb', line 356

def partitions
  res = nil
  @mutex.synchronize do
    res = @partition_write_map
  end

  res
end

#random_node ⇒ Object

Returns a random node on the cluster

Raises:



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/aerospike/cluster.rb', line 118

def random_node
  # Must copy array reference for copy on write semantics to work.
  node_array = nodes
  length = node_array.length
  i = 0
  while i < length
    # Must handle concurrency with other non-tending threads, so node_index is consistent.
    index = (@node_index.update{ |v| v+1 } % node_array.length).abs
    node = node_array[index]

    return node if node.active?

    i = i.succ
  end
  raise Aerospike::Exceptions::InvalidNode
end

#refresh_nodes ⇒ Object

Refresh status of all nodes in cluster. Adds new nodes and/or removes unhealty ones



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
 
# File 'lib/aerospike/cluster.rb', line 243

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

  # Clear node reference count
  nodes.each do |node|
    node.refresh_reset
  end

  peers.use_peers = supports_peers_protocol?

  # refresh all known nodes
  nodes.each do |node|
    node.refresh_info(peers)
  end

  # refresh peers when necessary
  if peers.generation_changed?
    # Refresh peers for all nodes that responded the first time even if only
    # one node's peers changed.
    nodes.each do |node|
      node.refresh_peers(peers)
    end
  end

  nodes.each do |node|
    node.refresh_partitions(peers) if node.partition_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

  # Add any new nodes from peer refresh
  if peers.nodes.any?
    # peers.nodes is a Hash. Pass only values, ie. the array of nodes
    add_nodes(peers.nodes.values)
    cluster_config_changed = true
  end

  cluster_config_changed
end

#remove_alias(aliass) ⇒ Object 



417
418
419
420
421
422
423
 
# File 'lib/aerospike/cluster.rb', line 417

def remove_alias(aliass)
  if aliass
    @mutex.synchronize do
      @aliases.delete(aliass)
    end
  end
end

#remove_cluster_config_change_listener(listener) ⇒ Object 



203
204
205
206
207
 
# File 'lib/aerospike/cluster.rb', line 203

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 



508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
 
# File 'lib/aerospike/cluster.rb', line 508

def remove_nodes(nodes_to_remove)
  # There is no need to delete nodes from partition_write_map because the nodes
  # have already been set to inactive. Further connection requests will result
  # in an exception and a different node will be tried.

  # Cleanup node resources.
  nodes_to_remove.each do |node|
    # Remove node's aliases from cluster alias set.
    # Aliases are only used in tend thread, so synchronization is not necessary.
    node.aliases.each do |aliass|
      Aerospike.logger.debug("Removing alias #{aliass}")
      remove_alias(aliass)
    end

    node.close
  end

  # Remove all nodes at once to avoid copying entire array multiple times.
  remove_nodes_copy(nodes_to_remove)
end

#remove_nodes_copy(nodes_to_remove) ⇒ Object 



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
 
# File 'lib/aerospike/cluster.rb', line 536

def remove_nodes_copy(nodes_to_remove)
  # Create temporary nodes array.
  # Since nodes are only marked for deletion using node references in the nodes array,
  # and the tend thread is the only thread modifying nodes, we are guaranteed that nodes
  # in nodes_to_remove exist.  Therefore, we know the final array size.
  nodes_list = nodes
  node_array = []
  count = 0

  # Add nodes that are not in remove list.
  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

  # Do sanity check to make sure assumptions are correct.
  if count < node_array.length
    Aerospike.logger.warn("Node remove mismatch. Expected #{node_array.length}, Received #{count}")

    # Resize array.
    node_array = node_array.dup[0..count-1]
  end

  set_nodes(node_array)
end

#request_info(policy, *commands) ⇒ Object 



176
177
178
179
180
181
182
 
# File 'lib/aerospike/cluster.rb', line 176

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

#seed_nodes ⇒ Object 



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
 
# File 'lib/aerospike/cluster.rb', line 365

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, ssl_options)
    rescue => e
      Aerospike.logger.error("Seed #{seed} failed: #{e}\n#{e.backtrace.join("\n")}")
      next
    end

    nv = nil
    # Seed host may have multiple aliases in the case of round-robin dns configurations.
    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, ssl_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 



93
94
95
96
97
 
# File 'lib/aerospike/cluster.rb', line 93

def seeds
  @mutex.synchronize do
    @cluster_seeds.dup
  end
end

#set_nodes(nodes) ⇒ Object 



529
530
531
532
533
534
 
# File 'lib/aerospike/cluster.rb', line 529

def set_nodes(nodes)
  @mutex.synchronize do
    # Replace nodes with copy.
    @cluster_nodes = nodes
  end
end

#set_partitions(part_map) ⇒ Object 



350
351
352
353
354
 
# File 'lib/aerospike/cluster.rb', line 350

def set_partitions(part_map)
  @mutex.synchronize do
    @partition_write_map = part_map
  end
end

#supports_feature?(feature) ⇒ Boolean

Returns:

  • (Boolean) 


184
185
186
 
# File 'lib/aerospike/cluster.rb', line 184

def supports_feature?(feature)
  @features.get.include?(feature.to_s)
end

#supports_peers_protocol? ⇒ Boolean

Returns:

  • (Boolean) 


188
189
190
 
# File 'lib/aerospike/cluster.rb', line 188

def supports_peers_protocol?
  nodes.all? { |node| node.supports_feature?('peers') }
end

#tend ⇒ Object

Check health of all nodes in cluster



229
230
231
232
233
234
235
236
237
238
239
 
# File 'lib/aerospike/cluster.rb', line 229

def tend
  was_changed = refresh_nodes

  return unless was_changed

  update_cluster_features
  notify_cluster_config_changed
  # only log the tend finish IF the number of nodes has been changed.
  # This prevents spamming the log on every tend interval
  log_tend_stats(nodes)
end

#tls_enabled? ⇒ Boolean

Returns:

  • (Boolean) 


77
78
79
 
# File 'lib/aerospike/cluster.rb', line 77

def tls_enabled?
  !ssl_options.nil? && ssl_options[:enable] != false
end

#update_cluster_features ⇒ Object 



335
336
337
338
339
340
341
 
# File 'lib/aerospike/cluster.rb', line 335

def update_cluster_features
  # Cluster supports features that are supported by all nodes
  @features.update do
    node_features = nodes.map(&:features)
    node_features.reduce(&:intersection) || Set.new
  end
end

#update_partitions(tokens, node) ⇒ Object 



168
169
170
171
172
173
174
 
# File 'lib/aerospike/cluster.rb', line 168

def update_partitions(tokens, node)
  nmap = tokens.update_partition(partitions, node)
  # update partition write map
  set_partitions(nmap) if nmap

  Aerospike.logger.info("Partitions for node #{node.name} updated")
end

#wait_till_stablized ⇒ Object 



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
 
# File 'lib/aerospike/cluster.rb', line 305

def wait_till_stablized
  count = -1

  # will run until the cluster is stablized
  thr = Thread.new do
    loop do
      tend

      # Check to see if cluster has changed since the last Tend.
      # If not, assume cluster has stabilized and return.
      break if count == nodes.length

      sleep(0.001) # sleep for a miliseconds

      count = nodes.length
    end
  end

  # wait for the thread to finish or timeout
  begin
    Timeout.timeout(@connection_timeout) do
      thr.join
    end
  rescue Timeout::Error
    thr.kill if thr.alive?
  end

  @closed.value = false if @cluster_nodes.length > 0
end