Class: CloudCrowd::NodeRecord

Inherits:

ActiveRecord::Base

• Object
• ActiveRecord::Base
• CloudCrowd::NodeRecord

show all

Defined in:

lib/cloud_crowd/models/node_record.rb

Overview

Constant Summary

PORT =

Extract the port number from the host id.

/:(\d+)\Z/

Class Method Summary (collapse)

• + (Object) check_in(params, request)

  Register a Node with the central server.

Instance Method Summary (collapse)

• - (Object) actions

  What Actions is this Node able to run?.

• - (Boolean) busy?

  Is this Node too busy for more work? Determined by number of workers, or the Node's load average, as configured in config.yml.

• - (Object) display_status

  The printable status of the Node.

• - (Object) node

  Keep a RestClient::Resource handy for contacting the Node, including HTTP authentication, if configured.

• - (Object) redistribute_work_units private

  When a Node exits, release its WorkUnits and redistribute them to others.

• - (Object) release_work_units

  Release all of this Node's WorkUnits for other nodes to take.

• - (Object) send_work_unit(unit)

  Dispatch a WorkUnit to this node.

• - (Object) to_json(opts = {})

  The JSON representation of a NodeRecord includes its worker_pids.

• - (Object) url

  The URL at which this Node may be reached.

• - (Object) worker_pids

  A list of the process ids of the workers currently being run by the Node.

26
27
28
29
30
31
32
33
34
35
36

# File 'lib/cloud_crowd/models/node_record.rb', line 26

def self.check_in(params, request)
  attrs = {
    :ip_address       => request.ip,
    :port             => params[:host].match(PORT)[1].to_i,
    :busy             => params[:busy],
    :tag              => params[:tag],
    :max_workers      => params[:max_workers],
    :enabled_actions  => params[:enabled_actions]
  }
  self.find_or_create_by_host(params[:host]).update_attributes!(attrs)
end

56
57
58

# File 'lib/cloud_crowd/models/node_record.rb', line 56

def actions
  @actions ||= enabled_actions.split(',')
end

62
63
64

# File 'lib/cloud_crowd/models/node_record.rb', line 62

def busy?
  busy || (max_workers && work_units.count >= max_workers)
end

79
80
81

# File 'lib/cloud_crowd/models/node_record.rb', line 79

def display_status
  busy? ? 'busy' : 'available'
end

74
75
76

# File 'lib/cloud_crowd/models/node_record.rb', line 74

def node
  @node ||= RestClient::Resource.new(url, CloudCrowd.client_options)
end

107
108
109
110

# File 'lib/cloud_crowd/models/node_record.rb', line 107

def redistribute_work_units
  release_work_units
  Thread.new { WorkUnit.distribute_to_nodes }
end

89
90
91

# File 'lib/cloud_crowd/models/node_record.rb', line 89

def release_work_units
  WorkUnit.update_all('node_record_id = null, worker_pid = null', "node_record_id = #{id}")
end

43
44
45
46
47
48
49
50
51
52
53

# File 'lib/cloud_crowd/models/node_record.rb', line 43

def send_work_unit(unit)
  result = node['/work'].post(:work_unit => unit.to_json)
  unit.assign_to(self, JSON.parse(result.body)['pid'])
  touch && true
rescue RestClient::RequestFailed => e
  raise e unless e.http_code == 503 && e.http_body == Node::OVERLOADED_MESSAGE
  update_attribute(:busy, true) && false
rescue RestClient::Exception, Errno::ECONNREFUSED, Timeout::Error, RestClient::RequestTimeout, Errno::ECONNRESET
  # Couldn't post to node, assume it's gone away.
  destroy && false
end

94
95
96
97
98
99
100

# File 'lib/cloud_crowd/models/node_record.rb', line 94

def to_json(opts={})
  { 'host'    => host,
    'workers' => worker_pids,
    'status'  => display_status,
    'tag'     => tag
  }.to_json
end

68
69
70

# File 'lib/cloud_crowd/models/node_record.rb', line 68

def url
  @url ||= "http://#{host}"
end

84
85
86

# File 'lib/cloud_crowd/models/node_record.rb', line 84

def worker_pids
  work_units.all(:select => 'worker_pid').map(&:worker_pid)
end