AppController/djinn.rb
# Queries ZooKeeper to see if our local copy of @nodes is out of date and
# should be regenerated with up to date data from ZooKeeper. If data on
# our node has changed, this starts or stops the necessary roles.
Queries ZooKeeper to see if our local copy of @nodes is out of date and
should be regenerated with up to date data from ZooKeeper. If data on
our node has changed, this starts or stops the necessary roles.
Queries ZooKeeper to see if our local copy of @nodes is out of date and should be regenerated with up to date data from ZooKeeper . If data on our node has changed this starts or stops the necessary roles .
def update_local_nodes()
    begin
      ZKInterface.lock_and_run {
        zk_ips_info = ZKInterface.get_ip_info()
        if zk_ips_info["last_updated"] <= @last_updated
          return false
        else
          Djinn.log_info("Updating data from ZK. Our timestamp, " +
            "#{@last_updated}, was older than the ZK timestamp, " +
            "#{zk_ips_info['last_updated']}")
        end
        all_ips = zk_ips_info["ips"]
        new_nodes = []
        all_ips.each { |ip|
          new_nodes << DjinnJobData.new(ZKInterface.get_job_data_for_ip(ip),
            @options['keyname'])
        }
        old_roles = my_node.jobs
        @state_change_lock.synchronize {
          @nodes = new_nodes.uniq
        }
        find_me_in_locations()
        new_roles = my_node.jobs
        Djinn.log_info("My new nodes are [#{@nodes.join(', ')}], and my new " +
          "node is #{my_node}")
        @done_loading = false
        ZKInterface.set_done_loading(my_node.public_ip, false)
        roles_to_start = new_roles - old_roles
        unless roles_to_start.empty?
          Djinn.log_info("Need to start [#{roles_to_start.join(', ')}] " +
            "roles on this node")
          roles_to_start.each { |role|
            Djinn.log_info("Starting role #{role}")
            send("start_#{role}".to_sym)
          }
        end
        roles_to_stop = old_roles - new_roles
        unless roles_to_stop.empty?
          Djinn.log_info("Need to stop [#{roles_to_stop.join(', ')}] " +
            "roles on this node")
          roles_to_stop.each { |role|
            send("stop_#{role}".to_sym)
          }
        end
        ZKInterface.set_done_loading(my_node.public_ip, true)
        @done_loading = true
        @last_updated = zk_ips_info['last_updated']
      }
    rescue => e
      Djinn.log_warn("(update_local_node) saw exception #{e.message}")
      return false
    end
    return true
  end
def update_local_nodes ( ) begin ZKInterface . lock_and_run { zk_ips_info = ZKInterface . get_ip_info ( ) if zk_ips_info [ "last_updated" ] <= @last_updated return false else Djinn . log_info ( "Updating data from ZK. Our timestamp, " + "#{@last_updated}, was older than the ZK timestamp, " + "#{zk_ips_info['last_updated']}" ) end all_ips = zk_ips_info [ "ips" ] new_nodes = [ ] all_ips . each { | ip | new_nodes << DjinnJobData . new ( ZKInterface . get_job_data_for_ip ( ip ) , @options [ 'keyname' ] ) } old_roles = my_node . jobs @state_change_lock . synchronize { @nodes = new_nodes . uniq } find_me_in_locations ( ) new_roles = my_node . jobs Djinn . log_info ( "My new nodes are [#{@nodes.join(', ')}], and my new " + "node is #{my_node}" ) @done_loading = false ZKInterface . set_done_loading ( my_node . public_ip , false ) roles_to_start = new_roles - old_roles unless roles_to_start . empty? Djinn . log_info ( "Need to start [#{roles_to_start.join(', ')}] " + "roles on this node" ) roles_to_start . each { | role | Djinn . log_info ( "Starting role #{role}" ) send ( "start_#{role}" . to_sym ) } end roles_to_stop = old_roles - new_roles unless roles_to_stop . empty? Djinn . log_info ( "Need to stop [#{roles_to_stop.join(', ')}] " + "roles on this node" ) roles_to_stop . each { | role | send ( "stop_#{role}" . to_sym ) } end ZKInterface . set_done_loading ( my_node . public_ip , true ) @done_loading = true @last_updated = zk_ips_info [ 'last_updated' ] } rescue => e Djinn . log_warn ( "(update_local_node) saw exception #{e.message}" ) return false end return true end
Queries ZooKeeper to see if our local copy of @nodes is out of date and
should be regenerated with up to date data from ZooKeeper.
Queries ZooKeeper to see if our local copy of @nodes is out of date and should be regenerated with up to date data from ZooKeeper .
# See if the ZooKeeper data is newer than ours - if not, don't # update anything and return. # Since we're about to possibly load and unload roles, set done_loading # for our node to false, so that other nodes don't erroneously send us # additional roles to do while we're in this state where lots of side # effects are happening. # And now that we're done loading/unloading roles, set done_loading for # our node back to true.