Merge lp:~hegde-shashank/appscale/trunk into lp:~cgb-cs/appscale/main-cgb-research

[Shashank Hegde]

Adding fault tolerance to AppScale

Unmerged revisions

801. By root <root@appscale-image0> on 2012-02-28

Deleting commented code

800. By root <root@appscale-image0> on 2012-02-28

refactored the code

799. By root <root@appscale-image0> on 2012-02-23

Neptune manager migration works!

798. By root <root@appscale-image0> on 2012-02-22

Saving neptune jobs and nodes to zookeeper

797. By root <root@appscale-image0> on 2012-02-15

Fixed neptune and saving the neptune data to zookeeper

796. By root <root@appscale-image0> on 2012-02-11

Storing nodes into zookeeper

795. By root <root@appscale-image0> on 2012-02-05

Added app manager failover

794. By root <root@appscale-image0> on 2012-01-28

Added leader election for status_monitor

793. By root <root@appscale-image0> on 2012-01-26

Added role watchers and callbacks

792. By root <root@appscale-image0> on 2012-01-25

Reconnecting to zookeeper if required

=== modified file 'AppController/djinn.rb'
--- AppController/djinn.rb	2012-02-20 22:36:25 +0000
+++ AppController/djinn.rb	2012-02-28 20:21:24 +0000
@@ -263,6 +263,14 @@
     @registered_with_sisyphus = false
   end
 
+  LOG_DEBUG = "debug"
+  LOG_TRACE = "trace"
+  LOG_INFO  = "info"
+  LOG_ERROR = "error!"
+
+  ROOT_APP_PATH = "/apps"
+  ROLE_PATH = "/roles"
+
   def done(secret)
     if valid_secret?(secret)
       return @done_loading
@@ -272,9 +280,16 @@
   end
 
   def kill(secret)
+<<<<<<< TREE
     if !valid_secret?(secret)
       return BAD_SECRET_MSG
     end
+=======
+    return BAD_SECRET_MSG unless valid_secret?(secret)
+
+    unregister_roles
+
+>>>>>>> MERGE-SOURCE
     @kill_sig_received = true
     
     if is_hybrid_cloud? 
@@ -294,8 +309,15 @@
       # turned on since that was the state they started in
 
       stop_ejabberd if my_node.is_login?
+<<<<<<< TREE
       stop_sisyphus if my_node.is_appengine?
       Repo.stop if my_node.is_shadow? or my_node.is_appengine?
+=======
+
+      if my_node.is_status_monitor? or my_node.is_appengine?
+        Repo.stop
+      end
+>>>>>>> MERGE-SOURCE
 
       jobs_to_run = my_node.jobs
       commands = {
@@ -492,7 +514,7 @@
     # error messages and have the tools poll it 
     Thread.new {
       # Tell other nodes to shutdown this application
-      if @app_names.include?(app_name) and !my_node.is_appengine?
+      if @app_names.include?(app_name) and my_node.is_app_manager?
         @nodes.each { |node|
           next if node.private_ip == my_node.private_ip
           if node.is_appengine? or node.is_login?
@@ -533,7 +555,7 @@
         HAProxy.remove_app(app_name)
         Nginx.reload
         Collectd.restart
-        ZKInterface.remove_app_entry(app_name)
+        zk_remove_app_entry(app_name)
         # TODO God does not shut down the application, so do it here for 
         # A temp fix.
         `ps -ef | grep dev_appserver | grep #{app_name} | grep -v grep | grep cookie_secret | awk '{print $2}' | xargs kill -9`
@@ -602,7 +624,15 @@
       parse_creds
       change_job
     end
-    
+
+    register_roles
+    setup_watchers
+
+    keyname = @creds["keyname"]
+
+    # Initially the status monitor konws everything about all the nodes.
+    save_nodes_to_zookeeper if my_node.is_status_monitor?
+ 
     while !@kill_sig_received do
       @state = "Done starting up AppScale, now in heartbeat mode"
       write_database_info
@@ -611,22 +641,39 @@
       update_api_status
       send_logs_to_sisyphus
 
-      if my_node.is_shadow?
+      @nodes.each { |n|
+        Djinn.log(LOG_DEBUG, "NODE from memory:", "#{n}")
+      }
+
+      # TODO: neptune manager might spawn/ delete nodes according to the neptune jobs.
+      # So should we get the nodes from zookeeper in every loop? Or should we listen for changes?
+      # @nodes must be updated on all nodes. so all nodes must listen/get the data from zookeeper
+      # Getting the data from zookeeper in each loop now. Optimize this later.
+      @nodes = zk_get_nodes(keyname)
+
+      neptune_nodes = get_neptune_nodes(@nodes)
+
+      if my_node.is_status_monitor?
         @nodes.each { |node|
           get_status(node)
 
           if node.should_destroy?
             Djinn.log_debug("Heartbeat - destroying node [#{node}]")
             @nodes.delete(node)
-            @neptune_nodes.delete(node)
+            zk_delete_node(keyname,node)
             infrastructure = @creds["infrastructure"]
             HelperFunctions.terminate_vms([node], infrastructure)
             FileUtils.rm_f("/etc/appscale/status-#{node.private_ip}.json")
           end
         }
+      
         Djinn.log_debug("Finished contacting all other nodes")
+      else
+        Djinn.log_debug("No need to heartbeat, we aren't the status monitor")
+      end
 
-        @neptune_nodes.each { |node|
+      if my_node.is_neptune_manager?
+        neptune_nodes.each { |node|
           Djinn.log_debug("Currently examining node [#{node}]")
           if node.should_extend?
             Djinn.log_debug("Extending time for node [#{node}]")
@@ -634,19 +681,26 @@
           elsif node.should_destroy?
             Djinn.log_debug("Time is up for node [#{node}] - destroying it")
             @nodes.delete(node)
-            @neptune_nodes.delete(node)
+            zk_delete_node(keyname,node)
             infrastructure = @creds["infrastructure"]
             HelperFunctions.terminate_vms([node], infrastructure)
             FileUtils.rm_f("/etc/appscale/status-#{node.private_ip}.json")
           end
         }
-      else
-        Djinn.log_debug("No need to heartbeat, we aren't the shadow")
+      
+        save_neptune_data_to_zookeeper
       end
 
       # TODO: consider only calling this if new apps are found
       start_appengine
+<<<<<<< TREE
       Kernel.sleep(20)
+=======
+      
+      #TODO: Check if the @nodes have changed. If so, update zookeeper
+ 
+      sleep(20)
+>>>>>>> MERGE-SOURCE
     end
   end
 
@@ -674,34 +728,39 @@
     end
 
     if File.exists?(location)
-      ZKInterface.add_app_entry(appname, my_node.serialize, location)
-      result = "success"
+      zk_add_app_entry(appname, my_node, location)
+      result = "success! #{appname} : #{location}"
     else
       result = "The #{appname} app was not found at #{location}."
     end
 
-    Djinn.log_debug(result)
+    Djinn.log(Djinn::LOG_TRACE, "done_uploading", result)
     return result
   end
 
   def is_app_running(appname, secret)
+<<<<<<< TREE
     if !valid_secret?(secret)
       return BAD_SECRET_MSG
     end
 
     hosters = ZKInterface.get_app_hosters(appname)
+=======
+    return BAD_SECRET_MSG unless valid_secret?(secret)
+    hosters = zk_get_app_hosters(appname)
+>>>>>>> MERGE-SOURCE
     hosters_w_appengine = []
     hosters.each { |node|
       hosters_w_appengine << node if node.is_appengine?
     }
  
     app_running = !hosters_w_appengine.empty?
-    Djinn.log_debug("Is app #{appname} running? #{app_running}")
     return app_running
   end
 
 
   def add_role(new_role, secret)
+<<<<<<< TREE
     if !valid_secret?(secret)
       return BAD_SECRET_MSG
     end
@@ -722,22 +781,47 @@
           send("start_#{role}".to_sym)
         end
       }
+=======
+    return BAD_SECRET_MSG unless valid_secret?(secret)
+    my_node.add_roles(new_role)
+    start_roles = new_role.split(":")
+    start_roles.each { |role|
+      Djinn.log(Djinn::LOG_TRACE,"add_role","Calling start_#{role}")
+      send("start_#{role}".to_sym)
+>>>>>>> MERGE-SOURCE
     }
-
+<<<<<<< TREE
+
+=======
+
+    # add the node as a role actor
+    zk_add_role(my_node, new_role)
+	
+>>>>>>> MERGE-SOURCE
     return "OK"
   end
 
   def remove_role(old_role, secret)
+<<<<<<< TREE
     if !valid_secret?(secret)
       return BAD_SECRET_MSG
     end
 
     my_node.remove_roles(old_role)
+=======
+    Djinn.log(Djinn::LOG_TRACE,"remove_role","Removing role #{old_role}")
+    return BAD_SECRET_MSG unless valid_secret?(secret)
+>>>>>>> MERGE-SOURCE
     stop_roles = old_role.split(":")
     stop_roles.each { |role|
       Djinn.log_debug("Removing and stopping role #{role}")
       send("stop_#{role}".to_sym)
     }
+    my_node.remove_roles(old_role)
+    
+    # remove the node as a role actor
+    zk_remove_role(my_node, old_role)
+
     return "OK"
   end
 
@@ -750,6 +834,7 @@
   # Important: Definitely do not log within the following three methods, as
   # it would cause an infinite loop.
   def self.log_debug(msg)
+<<<<<<< TREE
     time = Time.now
     self.log_to_stdout(time, msg)
     self.log_to_buffer(time, msg)
@@ -777,6 +862,45 @@
   # Logs and runs the given command, which is assumed to be trusted and thus
   # needs no filtering on our part. Obviously this should not be executed by
   # anything that the user could inject input into.
+=======
+      # TODO: reduce the copypasta here
+      puts "(Appscale) [#{Time.now}] #{msg}"
+      STDOUT.flush # TODO: examine performance impact of this
+      if @lock.nil?
+        begin
+          Syslog.open("appscale") { |s| s.debug(msg) }
+        rescue RuntimeError
+        end
+      else
+        @lock.synchronize {
+          begin
+            Syslog.open("appscale") { |s| s.debug(msg) }
+          rescue RuntimeError
+          end
+        }
+      end
+  end
+
+  def self.log(level, tag, msg)
+      # TODO: reduce the copypasta here
+      puts "(Appscale) [#{Time.now}] [#{level}] [#{tag}] #{msg}"
+      STDOUT.flush # TODO: examine performance impact of this
+      if @lock.nil?
+        begin
+          Syslog.open("appscale") { |s| s.debug(msg) }
+        rescue RuntimeError
+        end
+      else
+        @lock.synchronize {
+          begin
+            Syslog.open("appscale") { |s| s.debug(msg) }
+          rescue RuntimeError
+          end
+        }
+      end
+  end
+  
+>>>>>>> MERGE-SOURCE
   def self.log_run(command)
     Djinn.log_debug(command)
     Djinn.log_debug(`#{command}`)
@@ -819,6 +943,35 @@
     return djinn_loc_array
   end
     
+<<<<<<< TREE
+=======
+  def initialize()
+    @job = "none"
+    @nodes = []
+    @creds = {}
+    @app_names = []
+    @apps_loaded = []
+    @kill_sig_received = false
+    @my_index = nil
+    @@secret = HelperFunctions.get_secret
+    @done_loading = false
+    @port = 8080
+    @haproxy = 10000
+    @userappserver_public_ip = "not-up-yet"
+    @userappserver_private_ip = "not-up-yet"
+    @state = "AppController just started"
+    @total_boxes = 0
+    @num_appengines = 3
+    @restored = false
+    @neptune_jobs = {}
+    @neptune_nodes = []
+    @lock = Monitor.new
+    @api_status = {}
+
+    @watcher_callbacks = {}
+  end
+
+>>>>>>> MERGE-SOURCE
   def get_login
     @nodes.each { |node|
       return node if node.is_login?
@@ -835,6 +988,30 @@
     abort("No shadow nodes found.")
   end
 
+  def get_status_monitor
+    @nodes.each { |node|
+      return node if node.is_status_monitor?
+    }
+    return nil
+    #abort("No status monitor found.")
+  end
+
+  def get_app_manager
+    @nodes.each { |node|
+      return node if node.is_app_manager?
+    }
+    return nil
+    #abort("No app manager found.")
+  end
+
+  def get_neptune_manager
+    @nodes.each { |node|
+      return node if node.is_neptune_manager?
+    }
+    return nil
+    #abort("No neptune manager found.")
+  end
+
   def get_db_master
     @nodes.each { |node|
       return node if node.is_db_master?
@@ -1062,6 +1239,7 @@
     Djinn.log_debug("Restoring neptune data!")
     jobs_info = (File.open(file_to_load) { |f| f.read }).chomp
     jobs = []
+<<<<<<< TREE
 
     json_data = JSON.load(jobs_info)
     return if json_data.nil?
@@ -1074,6 +1252,27 @@
 
       if @neptune_jobs[job_name].nil?
         @neptune_jobs[job_name] = [this_job]
+=======
+    jobs_info.split("\n").each { |job|
+      info = job.split("::")
+      name = info[0]
+      num_nodes = Integer(info[1])
+
+      begin
+        start_time = Time._load(info[2])
+        end_time = Time._load(info[3])
+      rescue TypeError
+        start_time = Time.now
+        end_time = Time.now
+      end
+
+      job_id = info[6]
+
+      this_job = NeptuneJobData.new(name, num_nodes, start_time, end_time, job_id)
+
+      if @neptune_jobs[name].nil?
+        @neptune_jobs[name] = [this_job]
+>>>>>>> MERGE-SOURCE
       else
         @neptune_jobs[job_name] = [this_job]
       end
@@ -1384,8 +1583,14 @@
       @total_boxes = Integer(@creds['min_images'])
     end
 
+<<<<<<< TREE
     Djinn.log_debug("Pre-loop: #{@nodes.join('\n')}")
+=======
+    Djinn.log_debug("pre-loop: #{@nodes.join('\n')}")
+    # On the head node @nodes.size = 1. It contains only the head node at this point. Other nodes aren't populated untill the call to spawn_and_setup_appengine.
+>>>>>>> MERGE-SOURCE
     if @nodes.size == 1 and @total_boxes > 1
+      #spawn_and_setup_appengine parses the ips from creds, converts it to nodes and add them to the node array.
       spawn_and_setup_appengine
       loop {
         Djinn.log_debug("Looping: #{@nodes.join('\n')}")
@@ -1480,6 +1685,7 @@
       @state = "Starting up SOAP Server and PBServer"
       start_pbserver
       start_soap_server
+<<<<<<< TREE
       HelperFunctions.sleep_until_port_is_open(HelperFunctions.local_ip, UserAppClient::SERVER_PORT)
     end
 
@@ -1508,8 +1714,112 @@
     end
 
     # appengine is started elsewhere
-  end
-
+=======
+      HelperFunctions.sleep_until_port_is_open(HelperFunctions.local_ip, UA_SERVER_PORT)
+    end
+
+   start_blobstore_server if my_node.is_appengine?
+
+   # for neptune jobs, start a place where they can save output to
+   # also, since repo does health checks on the app engine apis, start it up there too
+
+   repo_ip = get_status_monitor.public_ip
+   repo_private_ip = get_status_monitor.private_ip
+   repo_ip = my_node.public_ip if my_node.is_appengine?
+   repo_private_ip = my_node.private_ip if my_node.is_appengine?
+   Repo.init(repo_ip, repo_private_ip,  @@secret)
+
+   if my_node.is_status_monitor? or my_node.is_appengine?
+     Repo.start(get_login.public_ip, @userappserver_private_ip)
+   end
+
+   # appengine is started elsewhere
+>>>>>>> MERGE-SOURCE
+  end
+
+  def start_status_monitor
+    Djinn.log(Djinn::LOG_TRACE, "start_status_monitor", "Starting status monitor")
+    my_node.add_roles(DjinnJobData::ROLE_STATUS_MTR)
+
+    # Get the @nodes from zookeeper
+    Djinn.log(Djinn::LOG_TRACE, "start_status_monitor", "Getting nodes from zookeeper")
+    keyname = @creds['keyname']
+    @nodes = zk_get_nodes(keyname)
+
+    zk_add_role(my_node,DjinnJobData::ROLE_STATUS_MTR)
+    zk_save_node(keyname,my_node)
+  end
+
+  def stop_status_monitor
+    Djinn.log(Djinn::LOG_TRACE,self.class,"Starting status monitor")
+
+    if !my_node.is_status_monitor?
+      Djinn.log(Djinn::LOG_ERROR,"stop_status_monitor", "I am not the status monitor")
+      return
+    end
+
+    my_node.remove_roles(DjinnJobData::ROLE_STATUS_MTR)
+    zk_remove_role(my_node,DjinnJobData::ROLE_STATUS_MTR)
+    keyname = @creds['keyname']
+    zk_save_node(keyname,my_node)
+  end
+  
+  def start_neptune_manager
+    Djinn.log(LOG_TRACE, self.class, "Starting neptune manager")
+    my_node.add_roles(DjinnJobData::ROLE_NEPTUNE_MGR)
+    zk_add_role(my_node,DjinnJobData::ROLE_NEPTUNE_MGR)
+    keyname = @creds['keyname']
+    zk_save_node(keyname,my_node)
+    Djinn.log(Djinn::LOG_TRACE, "start_neptune_manager", "Node saved, Getting neptune data from zookeeper")
+
+    # Get the neptune nodes and neptune jobs from zookeeper
+    get_neptune_data_from_zookeeper
+   
+    Djinn.log(Djinn::LOG_TRACE, "start_neptune_manager", "Completing old neptune jobs") 
+    # Wait for old jobs to complete
+    complete_old_jobs(keyname)
+
+  end
+
+  def stop_neptune_manager
+    Djinn.log(Djinn::LOG_TRACE, "stop_neptune_manager", "Stopping neptune manager")
+
+    if !my_node.is_neptune_manager?
+      Djinn.log(Djinn::LOG_ERROR, "stop_neptune_manager", "I am not the neptune manager.")
+      return
+    end
+
+    my_node.remove_roles(DjinnJobData::ROLE_NEPTUNE_MGR)
+    zk_remove_role(my_node,DjinnJobData::ROLE_NEPTUNE_MGR)
+    Djinn.log(Djinn::LOG_TRACE, "stop_neptune_manager", "Removed role from Zookeeper")
+
+    keyname = @creds['keyname']
+    zk_save_node(keyname,my_node)
+    Djinn.log(Djinn::LOG_TRACE, "stop_neptune_manager", "neptune manager stopper")
+  end
+
+  def start_app_manager
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Adding the role of app manager")
+    my_node.add_roles(DjinnJobData::ROLE_APP_MGR)
+    zk_add_role(my_node,DjinnJobData::ROLE_APP_MGR)
+    keyname = @creds['keyname']
+    zk_save_node(keyname,my_node)
+  end
+
+  def stop_app_manager
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Stopping app manager")
+
+    if !my_node.is_app_manager?
+      Djinn.log(Djinn::LOG_ERROR, "stop_app_manager", "I am not the app manager")
+      return
+    end
+
+    my_node.remove_roles(DjinnJobData::ROLE_APP_MGR)
+    zk_remove_role(my_node,DjinnJobData::ROLE_APP_MGR)
+    keyname = @creds['keyname']
+    zk_save_node(keyname,my_node)
+  end
+  
   def start_blobstore_server
     db_local_ip = @userappserver_private_ip
     my_ip = my_node.public_ip
@@ -1978,6 +2288,7 @@
     start_cmd = "/usr/bin/memcached -d -m 32 -p 11211 -u root"
     stop_cmd = "pkill memcached"
     GodInterface.start(:memcached, start_cmd, stop_cmd, [11211])
+  
   end
 
   def stop_memcached()
@@ -2025,6 +2336,7 @@
       rescue SocketError
       end
     end
+
   end
 
   # TODO: this function should use hadoop_helper
@@ -2058,7 +2370,7 @@
   # TODO: this function should use hadoop_helper
   def start_hadoop_org()
     i = my_node
-    return unless i.is_shadow? # change this later to db_master
+    return unless i.is_db_master?
     hadoop_home = File.expand_path("#{APPSCALE_HOME}/AppDB/hadoop-0.20.2/")
     Djinn.log_run("#{hadoop_home}/bin/hadoop namenode -format 2>&1")
     Djinn.log_run("#{hadoop_home}/bin/start-dfs.sh 2>&1")
@@ -2123,9 +2435,14 @@
       app_language = app_data.scan(/language:(\w+)/).flatten.to_s
             
       # TODO: merge these 
-      shadow = get_shadow
-      shadow_ip = shadow.private_ip
-      ssh_key = shadow.ssh_key
+      appmanager = get_app_manager
+      if appmanager == nil
+        Djinn.log(Djinn::LOG_ERROR,"start_app_engine","App manager not found")
+        return
+      end
+
+      appmanager_ip = appmanager.private_ip
+      ssh_key = appmanager.ssh_key
       app_dir = "/var/apps/#{app}/app"
       app_path = "#{app_dir}/#{app}.tar.gz"
       FileUtils.mkdir_p(app_dir)
@@ -2134,14 +2451,21 @@
       HelperFunctions.setup_app(app)
 
        
-      if my_node.is_shadow?
+      if my_node.is_app_manager?
+        Djinn.log(Djinn::LOG_DEBUG, "start_appengine", "app_manager - updating cron and starting xmpp")
         CronHelper.update_cron(my_public, app_language, app)
         start_xmpp_for_app(app, app_language)
       end
 
       if my_node.is_appengine?
+<<<<<<< TREE
         app_number = @nginx_port - 8080
         start_port = HelperFunctions::APP_START_PORT
+=======
+        Djinn.log(Djinn::LOG_DEBUG, "start_appengine", "appengine - starting app") 
+        app_number = @port - 8080
+        start_port = 20000
+>>>>>>> MERGE-SOURCE
         static_handlers = HelperFunctions.parse_static_data(app)
         proxy_port = HAProxy.app_listen_port(app_number)
         login_ip = get_login.public_ip
@@ -2210,13 +2534,14 @@
         done_uploading(app, app_path, @@secret)
       end
 
-      Monitoring.restart if my_node.is_shadow?
+      Monitoring.restart if my_node.is_status_monitor?
 
       if @app_names.include?("none")
         @apps_loaded = @apps_loaded - ["none"]
         @app_names = @app_names - ["none"]
       end
-        
+      
+      Djinn.log_debug("Loading #{app}")
       @apps_loaded << app
     }
 
@@ -2272,7 +2597,7 @@
 
     nodes_with_app = []
     loop {
-      nodes_with_app = ZKInterface.get_app_hosters(appname)
+      nodes_with_app = zk_get_app_hosters(appname)
       break unless nodes_with_app.empty?
       Djinn.log_debug("No nodes currently have a copy of app #{appname}, waiting...")
       Kernel.sleep(5)
@@ -2310,7 +2635,8 @@
     # for app named baz, this translates to baz@login_ip
 
     login_ip = get_login.public_ip
-    login_uac = UserAppClient.new(login_ip, @@secret)
+    Djinn.log_debug("Login ip: #{login_ip}")
+    login_uac = UserAppClient.new(@userappserver_private_ip, @@secret)
     xmpp_user = "#{app}@#{login_ip}"
     xmpp_pass = HelperFunctions.encrypt_password(xmpp_user, @@secret)
     login_uac.commit_new_user(xmpp_user, xmpp_pass, "app")
@@ -2407,4 +2733,339 @@
     Djinn.log_debug("Stopping Sisyphus")
     stop_app("sisyphus", @@secret)
   end
+
+
+  # Saves the node's role to zookeeper.
+  # Interested nodes can set a watch on these role paths for changes
+  def register_roles
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Registering roles with zookeeper")
+
+    me = my_node # save the node to avoid unnecessary calls to my_node
+    jobs = me.jobs
+    jobs.each { |job|
+      zk_add_role(me,job)
+    }
+  
+  end
+
+  # Removes all the node's roles from zookeeper
+  def unregister_roles
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Unregistering roles with zookeeper")
+    
+    me = my_node # save the node to avoid unnecessary calls to my_node
+    jobs = me.jobs
+    jobs.each { |job|
+      zk_add_role(me,job)
+    }
+  
+  end
+
+  def setup_watchers
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Setting up watchers...")
+
+    watch_role(DjinnJobData::ROLE_STATUS_MTR)
+    watch_role(DjinnJobData::ROLE_APP_MGR)
+    watch_role(DjinnJobData::ROLE_NEPTUNE_MGR)
+
+  end
+
+  def clear_votes(aRole)
+    # Wait for some time so that everyone can cast a vote
+    # This is not required for leader election, but for clearing the votes.
+    # If the votes are cleared immediately, then a slow node can win,
+    # resulting in two leaders
+    sleep(10)
+    ZKInterface.clear_votes(aRole)
+  end
+
+  def watch_role(role)
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Watching role #{role}")
+    if @watcher_callbacks[role].nil?
+      Djinn.log(Djinn::LOG_TRACE, "watch_role", "Creating a callback for #{role}")
+      @watcher_callbacks[role] = Zookeeper::WatcherCallback.new {
+        role_change_notification(@watcher_callbacks[role].context)
+      }
+    end
+    zk_get_actors(role,@watcher_callbacks[role])
+  end
+
+public
+  def role_change_notification(context)
+    path = context[:path]
+    # path returned would be of the form /roles/<role-name>
+    ar = path.split('/')
+    role = ar[-1]
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Received a role change notification for #{role}")
+
+    new_actors = zk_get_actors(role, @watcher_callbacks[role])
+    Djinn.log(Djinn::LOG_DEBUG,"role_change_notification","actors of #{role}: #{new_actors}")
+
+    if (role == DjinnJobData::ROLE_STATUS_MTR) or (role == DjinnJobData::ROLE_APP_MGR) or (role == DjinnJobData::ROLE_NEPTUNE_MGR)
+      Djinn.log(Djinn::LOG_DEBUG,"role_change_notification","should elect new leader")
+      if new_actors.size == 0
+        elect_new_actor(role)
+      end
+    end
+    # Nodes might have changed after leader election. So get the latest nodes' state from zk
+    keyname = @creds["keyname"]
+    @my_index = nil
+    @nodes = zk_get_nodes(keyname)
+  end
+
+private
+  def elect_new_actor(role)
+    Djinn.log(Djinn::LOG_TRACE,"elect_new_actor","Electing a new actor for #{role}")
+    my_ip = my_node.private_ip
+    winner = ZKInterface.elect_leader(role,my_ip)
+    Djinn.log(Djinn::LOG_TRACE,"elect_new_actor","Winner of #{role} election is #{winner}")
+    if my_ip == winner
+      action = "start_#{role}".to_sym
+      send(action)
+    else
+      # Some other node was the winner
+      # Update the node array
+      @nodes.each { |n|
+        if n.public_ip == winner
+          n.add_roles(role)
+        end
+      }
+    end
+    ZKInterface.complete_election(role)
+  end
+
+  def save_nodes_to_zookeeper
+    Djinn.log(LOG_TRACE, self.class, "Saving nodes to zookeeper")
+  
+    keyname = @creds["keyname"]
+    @nodes.each { |node|
+      zk_save_node(keyname, node)
+    }
+  end
+
+  def save_neptune_data_to_zookeeper
+    Djinn.log(LOG_TRACE, self.class, "Saving neptune data to zookeeper")
+  
+    keyname = @creds["keyname"]
+    
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Neptune jobs: #{@neptune_jobs}")
+    job_names = @neptune_jobs.keys
+    job_names.each { |nj_name|
+      jobs = @neptune_jobs[nj_name]
+      Djinn.log(Djinn::LOG_DEBUG, self.class, "Saving job #{jobs.class} #{jobs.class.name} #{jobs}")
+      # jobs is again an array of jobs with the same name
+      jobs.each{ |j|
+        Djinn.log(Djinn::LOG_DEBUG, self.class, "Saving job #{j.class} #{j.class.name} #{j}")
+        zk_neptune_save_job(keyname,j.job_id,j)
+      }
+    }
+  end
+
+  def get_neptune_data_from_zookeeper
+    Djinn.log(LOG_TRACE, self.class, "Getting neptune data from zookeeper")
+    keyname = @creds["keyname"]
+    @neptune_jobs  = zk_neptune_get_jobs(keyname)
+  end
+
+  def get_neptune_nodes(nodes)
+    Djinn.log(LOG_TRACE,self.class,"Filtering neptune nodes from all nodes")
+    neptune_nodes = []
+    nodes.each { |node|
+      neptune_nodes << node if node.neptune_worker
+    }
+    return neptune_nodes
+  end
+
+  ###############################################
+  ##       Zookeeper helper functions          ##
+  ###############################################
+
+  # Saves the application name in zookeeper
+  # Path: 
+  #   ROOT_APP_PATH/app_name/private_ip => serialized node data
+  #   ROOT_APP_PATH/app_name/private_ip/app_file => location of the app tar file
+  #
+  def zk_add_app_entry(appname, node, location)
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Adding an app entry for #{appname}")
+
+    appname_path = ROOT_APP_PATH + "/#{appname}"
+    ZKInterface.ensure_path(appname_path)
+
+    node_path = appname_path + "/#{node.private_ip}"
+
+    # Save the node data in the zknode
+    ZKInterface.set(node_path, node.serialize, ZKInterface::EPHEMERAL)
+    # Save the location of the file
+
+    file_location = node_path + "/app_file"
+    ZKInterface.set(file_location, location, ZKInterface::NOT_EPHEMERAL)
+  end
+
+  # Removes the application name from zookeeper
+  def zk_remove_app_entry(appname)
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Removing app entry for #{appname}")
+
+    appname_path = ROOT_APP_PATH + "/#{appname}"
+    ZKInterface.delete(appname_path)
+  end
+
+  # Gets the nodes that are serving the given application
+  # Path: 
+  #   ROOT_APP_PATH/app_name/private_ip => serialized node data
+  #   ROOT_APP_PATH/app_name/private_ip/app_file => location of the app tar file
+  def zk_get_app_hosters(appname)
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Getting app hosters")
+
+    appname_path = ROOT_APP_PATH + "/#{appname}"
+    app_hoster_ips = []
+    app_hoster_ips = ZKInterface.get_children(appname_path)
+    hoster_nodes = []
+    app_hoster_ips.each { |ip|
+      node_data = ZKInterface.get(appname_path + "/#{ip}")
+      hoster_nodes << DjinnJobData.deserialize(node_data)
+    }
+    return hoster_nodes
+  end
+
+
+  # Add the node's role to zookeeper.
+  def zk_add_role(node, role)
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Adding the role #{role} to #{node.private_ip}")
+
+    path = ROLE_PATH + "/#{role}"
+    ZKInterface.ensure_path(path)
+
+    path = ROLE_PATH + "/#{role}/#{node.private_ip}"
+    ZKInterface.set(path, DEFAULT_DATA, ZKInterface::EPHEMERAL)
+  end
+
+  # Remove the node's role from zookeeper.
+  def zk_remove_role(node, role)
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Removing the role #{role} to #{node.private_ip}")
+
+    path = ROLE_PATH + "/#{role}/#{node.private_ip}"
+    ZKInterface.delete(path)
+  end
+
+  # Gets the actors of the given role
+  def zk_get_actors(role, callback = nil)
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Getting actors for #{role}")
+
+    path = ROLE_PATH + "/#{role}"
+    actors = ZKInterface.get_children(path, callback)
+    return actors
+  end
+
+  # Save the node into zookeeper
+  def zk_save_node(keyname, node)
+
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Saving node to zookeeper")
+    
+    private_ip  = node.private_ip
+    public_ip   = node.public_ip
+    instance_id = node.instance_id
+    cloud       = node.cloud
+    ssh_key     = node.ssh_key
+    c_time      = node.creation_time
+    d_time      = node.destruction_time
+    jobs        = node.jobs.join(":")
+    n_worker    = node.neptune_worker
+
+    if c_time == nil
+      c_time = Time.mktime(EPOCH_YEAR)
+    end
+    c_time = c_time._dump
+
+    if d_time == nil
+      d_time = Time.mktime(NEVER_DESTROY)
+    end
+    d_time = d_time._dump
+
+    Djinn.log(Djinn::LOG_DEBUG, "zk_save_node", "#{private_ip}, #{public_ip}, #{instance_id}, #{cloud}, #{ssh_key}, #{c_time}, #{d_time}, #{jobs}, #{n_worker}")
+
+    path = "/#{keyname}/nodes"
+    ZKInterface.ensure_path(path)
+
+    path = "/#{keyname}/nodes/#{node.private_ip}"
+    ZKInterface.set(path,node.public_ip,ZKInterface::NOT_EPHEMERAL)
+
+    ZKInterface.set(path+"/instance_id", instance_id,ZKInterface::NOT_EPHEMERAL)
+    ZKInterface.set(path+"/cloud", cloud,ZKInterface::NOT_EPHEMERAL)
+    ZKInterface.set(path+"/ssh_key", ssh_key,ZKInterface::NOT_EPHEMERAL)
+    ZKInterface.set(path+"/c_time", c_time.to_s,ZKInterface::NOT_EPHEMERAL)
+    ZKInterface.set(path+"/d_time", d_time.to_s,ZKInterface::NOT_EPHEMERAL)
+    ZKInterface.set(path+"/jobs", jobs,ZKInterface::NOT_EPHEMERAL)
+    ZKInterface.set(path+"/n_worker", n_worker.to_s,ZKInterface::NOT_EPHEMERAL)
+
+  end
+
+  # Get the node data from zookeeper
+  def zk_get_nodes(keyname)
+
+    Djinn.log(Djinn::LOG_TRACE, self.class, "Getting nodes from zookeeper")
+
+    nodes = []
+    ip_path = "/#{keyname}/nodes"
+    node_ips = ZKInterface.get_children(ip_path)
+    node_ips.each { |ip|
+      path = ip_path + "/#{ip}"
+      Djinn.log(Djinn::LOG_DEBUG, "zk_get_nodes", "IP : #{ip}")
+
+      private_ip    = ip
+      public_ip     = ZKInterface.get(ip_path+"/#{ip}")
+      next if public_ip == ZKInterface::FAILURE
+      
+      instance_id   = ZKInterface.get(ip_path+"/#{ip}/instance_id")
+      next if instance_id == ZKInterface::FAILURE
+      
+      cloud         = ZKInterface.get(ip_path+"/#{ip}/cloud")
+      next if cloud == ZKInterface::FAILURE
+      
+      ssh_key       = ZKInterface.get(ip_path+"/#{ip}/ssh_key")
+      next if ssh_key == ZKInterface::FAILURE
+      
+      c_time        = ZKInterface.get(ip_path+"/#{ip}/c_time")
+      next if c_time == ZKInterface::FAILURE
+      
+      d_time        = ZKInterface.get(ip_path+"/#{ip}/d_time")
+      next if d_time == ZKInterface::FAILURE
+      
+      jobs          = ZKInterface.get(ip_path+"/#{ip}/jobs")
+      next if jobs  == ZKInterface::FAILURE
+     
+      n_worker      = ZKInterface.get(ip_path+"/#{ip}/n_worker")
+      n_worker      = n_worker.to_sym
+      next if n_worker == ZKInterface::FAILURE 
+
+      Djinn.log(Djinn::LOG_DEBUG, "zk_get_nodes", "#{private_ip}, #{public_ip}, #{instance_id}, #{cloud}, #{ssh_key}, #{c_time.to_s}, #{d_time.to_s}, #{jobs}, #{n_worker}")
+
+      if jobs.nil?
+        jobs = ["open"]
+      end
+
+      # Constructor is wrong 'cause the roles parameter expects ip, cloud name, et al along with the jobs
+      # TODO: Change the constructor?
+      node = DjinnJobData.new(jobs, keyname)
+      node.private_ip   = private_ip
+      node.public_ip    = public_ip
+      node.instance_id  = instance_id
+      node.cloud        = cloud
+      node.ssh_key      = ssh_key
+      node.add_roles(jobs)
+      node.creation_time    = Time._load(c_time)
+      node.destruction_time = Time._load(d_time)
+      node.neptune_worker   = n_worker
+
+      nodes << node
+    }
+    @my_index = nil
+    return nodes
+  end
+
+  def zk_delete_node(keyname,node)
+    Djinn.log(Djinn::LOG_TRACE,self.class,"Deleting node #{node.private_ip}")
+
+    path = "/#{keyname}/nodes/#{node.private_ip}"
+    ZKInterface.delete(path)
+  end
+
 end

=== modified file 'AppController/lib/djinn_job_data.rb'
--- AppController/lib/djinn_job_data.rb	2012-02-19 04:55:17 +0000
+++ AppController/lib/djinn_job_data.rb	2012-02-28 20:21:24 +0000
@@ -15,10 +15,30 @@
 # also contains info about when we spawned the node (helpful for optimizing
 # costs, which may charge on an hourly basis).
 class DjinnJobData
+  
+  # Roles
+  ROLE_APP_ENGINE = "appengine"
+  ROLE_APP_MGR    = "app_manager"
+  ROLE_DB_MASTER  = "db_master"
+  ROLE_DB_SLAVE   = "db_slave"
+  ROLE_LB         = "load_balancer"
+  ROLE_LOGIN      = "login"
+  ROLE_MEMCACHE   = "memcache"
+  ROLE_NEPTUNE_MGR= "neptune_manager"
+  ROLE_OPEN       = "open"
+  ROLE_STATUS_MTR = "status_monitor"
+  ROLE_ZOOKEEPER  = "zookeeper"
+ 
   attr_accessor :public_ip, :private_ip, :jobs, :instance_id, :cloud, :ssh_key
   attr_accessor :creation_time, :destruction_time, :failed_heartbeats
+<<<<<<< TREE
  
 
+=======
+  # If the node is a neptune worker then this is set to true
+  attr_accessor :neptune_worker
+
+>>>>>>> MERGE-SOURCE
   def initialize(roles, keyname)
     # format: "publicIP:privateIP:load_balancer:appengine:table-master:table-slave:instance_id:cloud"
 
@@ -46,6 +66,7 @@
     @failed_heartbeats = 0
 
     appscale_jobs = ["load_balancer", "shadow"]
+    appscale_jobs += ["status_monitor", "app_manager", "neptune_manager"]
     appscale_jobs += ["db_master", "db_slave"]
     appscale_jobs += ["zookeeper"]
     appscale_jobs += ["login"]
@@ -59,12 +80,15 @@
     appscale_jobs.each { |job|
       @jobs << job if roles.include?(job)
     }
+
+    @neptune_worker = false
   end
 
   def add_roles(roles)
     new_jobs = roles.split(":")
     @jobs = (@jobs + new_jobs).uniq
     @jobs.delete("open")
+    @jobs = ["open"] if @jobs.size == 0
   end
 
   def remove_roles(roles)
@@ -75,6 +99,7 @@
 
   def set_roles(roles)
     @jobs = roles.split(":")
+    @jobs = ["open"] if @jobs.size == 0
   end
 
   # not the best name for this but basically correct

=== modified file 'AppController/lib/repo.rb'
--- AppController/lib/repo.rb	2012-02-27 07:19:48 +0000
+++ AppController/lib/repo.rb	2012-02-28 20:21:24 +0000
@@ -95,6 +95,243 @@
     Djinn.log_debug(`pkill -f DevAppServerMain`)
   end
 
-
+<<<<<<< TREE
+
+=======
+  def self.valid_storage_creds(storage, creds)
+    if storage == "appdb"
+      valid = true
+    elsif storage == "s3"
+      conn = self.get_s3_conn(creds)
+      begin
+        all_buckets = conn.list_all_my_buckets
+        Djinn.log_debug("this user owns these buckets: [#{all_buckets.join(', ')}]")
+        valid = true
+      rescue RightAws::AwsError
+        valid = false
+      end
+    end
+
+    Djinn.log_debug("did user provide valid storage creds? #{valid}")
+  end
+
+  def self.set_output(path, output, storage="appdb", creds={}, is_file=false)
+    return self.set(path, output, :output, storage, creds, is_file)
+  end
+
+  def self.get_output(path, storage="appdb", creds={}, is_file=false)
+    return self.get(path, :output, storage, creds, is_file)
+  end
+
+  def self.set_acl(path, new_acl, storage="appdb", creds={})
+    return self.set(path, new_acl, :acl, storage, creds)
+  end
+
+  def self.get_acl(path, storage="appdb", creds={})
+    return self.get(path, :acl, storage, creds)
+  end
+
+  def self.does_file_exist?(path, storage="appdb", creds={})
+    if storage == "appdb"
+      result = `curl http://#{@@ip}:8079/doesexist -X POST -d 'SECRET=#{@@secret}' -d 'KEY=#{path}'`
+    elsif storage == "s3"
+      conn = self.get_s3_conn(creds)
+      bucket, file = self.parse_s3_key(path)
+
+      if self.does_s3_bucket_exist?(conn, bucket)
+        Djinn.log_debug("[does file exist] bucket [#{bucket}] exists")
+        begin
+          Djinn.log_debug("[does file exist] getting acl for bucket [#{bucket}] and file [#{file}] ")
+          conn.get_acl(bucket, file)
+          result = "true"
+        rescue RightAws::AwsError
+          result = "false"
+        end
+      else
+        Djinn.log_debug("[does file exist] bucket [#{bucket}] does not exist")
+        result = "false"
+      end
+    else
+      msg = "ERROR - unrecognized storage for does_file_exist via repo - you requested #{storage}"
+      Djinn.log_debug(msg)
+      abort(msg)
+    end
+ 
+    Djinn.log_debug("does key=#{path} exist? #{result}")
+    return result == "true"
+  end
+
+  private
+
+  def self.get(key, type, storage, creds, is_file=false)
+    if storage == "appdb"
+      Djinn.log(Djinn::LOG_INFO, "repo::get", "Using the secret #{@@secret}")
+      result = `curl http://#{@@ip}:8079/get -X POST -d 'SECRET=#{@@secret}' -d 'KEY=#{key}' -d 'TYPE=#{type}'`
+      result = URI.unescape(result)
+      foo = File.new(is_file, File::CREAT|File::RDWR)
+      foo.write(result)
+      foo.close
+    elsif storage == "s3"
+      conn = self.get_s3_conn(creds)
+      bucket, file = self.parse_s3_key(key)
+
+      if type == :output
+        if is_file
+          foo = File.new(is_file, File::CREAT|File::RDWR)
+          result = conn.get(bucket, file) { |chunk|
+            foo.write(chunk)
+          } 
+          foo.close
+        else
+          result = conn.get(bucket, file)[:object]
+        end
+      elsif type == :acl
+        # TODO: implement me!
+        result = "private"
+      else
+        msg = "type not supported for get operation - #{type} was used"
+        abort(msg)
+      end
+    else
+      msg = "ERROR - unrecognized storage for get via repo - you requested #{storage}"
+      Djinn.log_debug(msg)
+      abort(msg)
+    end
+
+    Djinn.log_debug("get key=#{key} type=#{type}")
+    return result
+  end
+
+  def self.set(key, val, type, storage, creds, is_file=false)
+    if storage == "appdb"
+      if is_file
+        if File.directory?(val)
+           result = true
+          `ls #{val}`.split.each { |file|
+            fullkey = key + "/" + file
+            fullval = val + "/" + file
+            Djinn.log_debug("recursive dive - now saving remote [#{fullkey}], local [#{fullval}]")
+            temp = self.set(fullkey, fullval, type, storage, creds, is_file)
+            result = false unless temp
+            file = fullkey
+          }
+        else
+          Djinn.log_debug("attempting to put local file #{val} into file #{key}")
+          val = HelperFunctions.read_file(val, chomp=false)
+          val = URI.escape(val)
+          result = false
+          begin
+            res = Net::HTTP.post_form(URI.parse("http://#{@@ip}:8079/set"),
+                              {'SECRET' => @@secret, 'KEY' => key, 
+                               'VALUE' => val, 'TYPE' => type})
+            Djinn.log_debug("set key=#{key} type=#{type} returned #{res.body}")
+            result = true if res.body == "success"
+          rescue Exception => e
+            Djinn.log_debug("saw exception #{e.class} when posting userdata to repo at #{key}")
+          end
+
+        end
+      else
+          Djinn.log_debug("attempting to put local file #{val} into bucket #{bucket}, location #{file}")
+          val = URI.escape(val, Regexp.new("[^#{URI::PATTERN::UNRESERVED}]"))
+          result = `curl http://#{@@ip}:8079/set -X POST -d 'SECRET=#{@@secret}' -d 'KEY=#{key}' -d 'VALUE=#{val}' -d 'TYPE=#{type}'`
+          Djinn.log_debug("set key=#{key} type=#{type} returned #{result}")
+          result = true if result == "success"
+      end
+    elsif storage == "s3"
+      conn = self.get_s3_conn(creds)
+      bucket, file = self.parse_s3_key(key)
+
+      if type == :output
+        # TODO: for now we assume the bucket exists
+        #if !self.does_s3_bucket_exist?(conn, bucket)
+        #  Djinn.log_debug("bucket #{bucket} does not exist - creating it now")
+        #  conn.create_bucket(bucket)
+
+          # bucket creation takes a few moments - wait for it to exist
+          # before we start putting things in it 
+        #  loop {
+        #    Djinn.log_debug("waiting for s3 bucket #{bucket} to exist")
+        #    sleep(5)
+        #    break if self.does_s3_bucket_exist?(conn, bucket)
+        #  }
+        #end
+
+        Djinn.log_debug("s3 bucket #{bucket} exists, now storing file #{file}")
+
+        # this throws an exception that gets automatically caught and logged
+        # looks like "undefined method `pos' for <String...>"
+        # the put operation still succeeds
+        if is_file
+          if File.directory?(val)
+             result = true
+            `ls #{val}`.split.each { |file|
+              fullkey = key + "/" + file
+              fullval = val + "/" + file
+              Djinn.log_debug("recursive dive - now saving remote [#{fullkey}], local [#{fullval}]")
+              temp = self.set(fullkey, fullval, type, storage, creds, is_file)
+              result = false unless temp
+              file = fullkey
+            }
+          else
+            Djinn.log_debug("attempting to put local file #{val} into bucket #{bucket}, location #{file}")
+            result = conn.put(bucket, file, File.open(val)) #headers={"Content-Length" => val.length})
+          end
+        else
+          result = conn.put(bucket, file, val, headers={"Content-Length" => val.length})
+        end
+
+        Djinn.log_debug("done putting file #{file} to s3!")
+      elsif type == :acl
+        # TODO: implement me!
+        return false
+      else
+        msg = "type not supported for get operation - #{type} was used"
+        abort(msg)
+      end
+    else
+      msg = "ERROR - unrecognized storage for set via repo - you requested #{storage}"
+      Djinn.log_debug(msg)
+      abort(msg)
+    end
+
+    Djinn.log_debug("set operation returned #{result}")
+    return result
+  end
+
+  def self.get_s3_conn(creds)
+    access_key = creds['EC2_ACCESS_KEY']
+    secret_key = creds['EC2_SECRET_KEY']
+
+    s3_url = creds['S3_URL']
+
+    obscured_a_key = HelperFunctions.obscure_string(access_key)
+    obscured_s_key = HelperFunctions.obscure_string(secret_key)
+
+    Djinn.log_debug("creating S3 connection with access key [#{obscured_a_key}], secret key [#{obscured_s_key}], and S3 url [#{s3_url}]")
+
+    old_s3_url = ENV['S3_URL']
+    ENV['S3_URL'] = s3_url
+    conn = RightAws::S3Interface.new(access_key, secret_key)
+    ENV['S3_URL'] = old_s3_url
+
+    return conn
+  end
+
+  def self.parse_s3_key(key)
+    paths = key.split("/")
+    bucket = paths[1]
+    file = paths[2, paths.length - 1].join("/")
+    return bucket, file
+  end
+
+  def self.does_s3_bucket_exist?(conn, bucket)
+    all_buckets = conn.list_all_my_buckets
+    bucket_names = all_buckets.map { |b| b[:name] }
+    bucket_exists = bucket_names.include?(bucket)
+    Djinn.log_debug("the user owns buckets [#{bucket_names.join(', ')}] - do they own [#{bucket}]? #{bucket_exists}")
+    return bucket_exists
+  end
+>>>>>>> MERGE-SOURCE
 end
 

=== added file 'AppController/lib/role_watcher.rb'
--- AppController/lib/role_watcher.rb	1970-01-01 00:00:00 +0000
+++ AppController/lib/role_watcher.rb	2012-02-28 20:21:24 +0000
@@ -0,0 +1,110 @@
+#! /usr/bin/ruby -w
+
+require 'rubygems'
+require 'zookeeper'
+#require 'djinn'
+
+def listener(aContext)
+  puts("Callback!")
+end
+
+# Base class for monitoring changes to roles
+class RoleWatcher
+
+  private #constants
+  ZOOKEEPER_PORT = 2181
+
+  public
+
+    def initialize(aZookeeper, aCallback)
+      cb = aCallback.to_sym
+      send(cb,"test") 
+      @mZK = Zookeeper.new("#{aZookeeper}:#{ZOOKEEPER_PORT}")
+      @mCallback = Zookeeper::WatcherCallback.new {
+                    #notification(@mCallback.context)
+                    send(cb,@mCallback.context)
+                  }
+    end
+
+    # Start a watch on the specified node
+    # the notification callback is invoked whenever the node value changes
+    def watch_node(aNode)
+      @mZK.get( :path => aNode,
+                :watcher => @mCallback,
+                :watcher_context => {:path => aNode}
+              )[:data]
+    end
+
+    def watch_children(aNode)
+      @mZK.get_children(:path => aNode,
+                        :watcher => @mCallback,
+                        :watcher_context => {:path => aNode}
+                       )
+    end
+
+  protected
+
+    # Override
+    # This method is invoked whenever the watched node value changes
+    def notification(aContext)
+    end
+
+end
+
+
+# Subclass for watching over the status monitor
+class StatusMonitorWatcher < RoleWatcher
+
+  public
+    def initialize(aZookeeper, aCallback)
+      super(aZookeeper)
+      @previousNodes
+    end
+
+  protected 
+    def notification(aContext)
+      watchedPath = aContext[:path]
+      # Get the new data and set the watch again
+      newData = watch_children(watchedPath)
+      #Djinn.log_debug("Status monitor : #{newData.first}") 
+      puts("Status monitor : #{newData.first}") 
+    end
+end
+
+
+# Subclass for watching over the app engine 
+class AppEngineWatcher < RoleWatcher
+
+  public
+    def initialize(aZookeeper, aCallback)
+      super(aZookeeper,aCallback)
+      @previousNodes
+    end
+
+  protected 
+    def notification(aContext)
+      watchedPath = aContext[:path]
+      # Get the new data and set the watch again
+      newData = watch_children(watchedPath)
+      #Djinn.log_debug("AppEngine :")
+      puts("AppEngine:")
+      newData.each { |node|
+        #Djinn.log_debug(node)
+        puts(node)
+      }
+    end
+end
+
+if __FILE__ == $0
+  if ARGV.size != 1
+    puts("#{$0} <zk host>")
+    exit
+  end
+  aeWatcher = AppEngineWatcher.new(ARGV[0], "listener")
+  ae = aeWatcher.watch_children("/roles/appengine")
+  ae.each { |n|
+    puts(n)
+  }
+
+  a = STDIN.getc
+end

=== modified file 'AppController/lib/zkinterface.rb'
--- AppController/lib/zkinterface.rb	2012-02-18 07:20:06 +0000
+++ AppController/lib/zkinterface.rb	2012-02-28 20:21:24 +0000
@@ -2,11 +2,15 @@
 
 require 'fileutils'
 require 'monitor'
+require 'rubygems'
+require 'zookeeper'
+require 'zookeeper/exceptions'
 
 
 $:.unshift File.join(File.dirname(__FILE__))
 require 'helperfunctions'
 
+<<<<<<< TREE
 
 SUCCESS = 0
 
@@ -20,6 +24,12 @@
 
 
 ROOT_APP_PATH = "/apps"
+=======
+DEFAULT_DATA = "nothing special here"
+
+EPOCH_YEAR = 1970
+NEVER_DESTROY = 2100
+>>>>>>> MERGE-SOURCE
 
 
 # The AppController employs the open source software ZooKeeper as a highly
@@ -28,58 +38,68 @@
 # communicate with ZooKeeper, and automates commonly performed functions by the
 # AppController.
 class ZKInterface
-  public
-
+  private
+  RETRY_COUNT = 10
+  ELECTION_PATH = "/roles/election"
+
+  public
+  EPHEMERAL = true
+  NOT_EPHEMERAL = false
+  SEQUENCE = true
+
+  SUCCESS = 0
+  FAILURE = -1
+
+  
+  public
   def self.init(my_node, all_nodes)
     require 'rubygems'
     require 'zookeeper'
 
+    @mMyNode = my_node
+    @mAllNodes = all_nodes
+    #@@lock = nil
+    #@@zk = nil
+    self.connect(my_node,all_nodes)
+  end
+
+  def self.connect(my_node, all_nodes)
+    require 'rubygems'
+    require 'zookeeper'
+    
+    #if @@lock == nil  
     unless defined?(@@lock)
       @@lock = Monitor.new
     end
 
     zk_location = self.get_zk_location(my_node, all_nodes)
+    log(Djinn::LOG_INFO,"Zookeeper location: #{zk_location}")
 
     @@lock.synchronize {
       @@zk = Zookeeper.new(zk_location)
+      log(Djinn::LOG_TRACE, "Zookeeper: #{@@zk}")
     }
-  end
-
-  def self.add_app_entry(appname, ip, location)
-    appname_path = ROOT_APP_PATH + "/#{appname}"
-    full_path = appname_path + "/#{ip}"
-
-    # can't just create path in ZK
-    # need to do create the nodes at each level
-
-    self.set(ROOT_APP_PATH, "nothing special here", NOT_EPHEMERAL)
-    self.set(appname_path, "nothing special here", NOT_EPHEMERAL)
-    self.set(full_path, location, EPHEMERAL)
-  end
-
-  def self.remove_app_entry(appname)
-    appname_path = ROOT_APP_PATH + "/#{appname}"
-    self.delete(appname_path)
-  end
-
-  def self.get_app_hosters(appname)
-    unless defined?(@@zk)
-      return []
+
+  end
+
+  def self.ensure_path(path, data = DEFAULT_DATA)
+    self.log(Djinn::LOG_TRACE, "Ensuring path #{path}")
+    exists = self.get(path)
+    if exists == FAILURE
+      # find the last occurence of '/' and ensure path before that exists
+      index = path.rindex('/')
+      if index > 0 # if the last '/' is the first, then thats the root
+        parent = path[0..index-1]
+        self.ensure_path(parent, data)
+      end
+      self.set(path,data,ZKInterface::NOT_EPHEMERAL)
     end
-
-    appname_path = ROOT_APP_PATH + "/#{appname}"
-    app_hosters = self.get_children(appname_path)
-    #converted = app_hosters
-    converted = []
-    app_hosters.each { |serialized|
-      converted << DjinnJobData.deserialize(serialized)
-    }
-    return converted
   end
 
-  private
-
+  # Gets the value stored in the zk node identified by the path
+  # If connection to the zookeeper cannot be established, then it retries 'RETRY_COUNT' times
   def self.get(key)
+<<<<<<< TREE
     Djinn.log_debug("[ZK] trying to get #{key}")
     info = @@zk.get(:path => key)
     if info[:rc] == 0
@@ -117,10 +137,90 @@
     end
   end
 
+=======
+    self.log(Djinn::LOG_TRACE,"trying to get #{key}")
+    RETRY_COUNT.times { |retry_count|
+      begin
+        info = @@zk.get(:path => key)
+        if info[:rc] == 0
+          return info[:data]
+        else
+          Djinn.log(Djinn::LOG_ERROR, "ZK::get", "Data not found at #{key}")
+          return FAILURE
+        end
+      # TODO: Handle exceptions better.
+      rescue Exception
+        log(Djinn::LOG_ERROR, "#{retry_count+1} reopening connection to zk")
+        @@zk.reopen
+      end
+    }
+  end
+
+  # Gets the children of the zk node identified by the path
+  # If connection to the zookeeper cannot be established, then it retries 'RETRY_COUNT' times
+  def self.get_children(key, callback=nil)
+    self.log(Djinn::LOG_TRACE,"[ZK] trying to get children #{key}")
+    RETRY_COUNT.times { |retry_count|
+      begin
+        children = nil
+        if callback == nil
+          children = @@zk.get_children(:path => key)[:children]
+        else
+          self.log(Djinn::LOG_TRACE,"[ZK] watching the children at #{key}. Watcher : #{callback}")
+          children = @@zk.get_children(:path => key,
+                                       :watcher => callback,
+                                       :watcher_context => {:path => key}
+                                      )[:children]
+        end
+        if children.nil?
+          Djinn.log(Djinn::LOG_ERROR, "ZK::get_children", "Data not found at #{key}")
+          return []
+        else
+          return children
+        end
+      # TODO: Handle exceptions better.
+      rescue Exception
+        log(Djinn::LOG_ERROR, "#{retry_count+1} reopening connection to zk")
+        @@zk.reopen
+      end
+    }
+  end
+ 
+  # Sets the key to val. If the key does not exist, then the key is created. else the value is updated.
+  # ephemeral (boolean) - flag to indicate whether the zk node should be ephemeral or not
+  # sequence (boolean) - flag to indicate whether zookeeper should create the node with a sequence number added automatically 
+  def self.set(key, val, ephemeral, sequence = false)
+    self.log(Djinn::LOG_TRACE, "[ZK] trying to set #{key} to #{val} with ephemeral = #{ephemeral} and sequence = #{sequence}")
+    exists = self.get(key)
+    if exists == FAILURE
+      self.log(Djinn::LOG_DEBUG, "Creating #{key}")
+      info = @@zk.create(:path => key, :ephemeral => ephemeral, :data => val, :sequence => sequence)
+      if info[:rc] == 0
+        self.log(Djinn::LOG_DEBUG,"Success!")
+        return SUCCESS
+      else
+        self.log(Djinn::LOG_DEBUG,"Failure")
+        return FAILURE
+      end
+    else
+      info = @@zk.set(:path => key, :data => val)
+      if info[:rc] == 0
+        self.log(Djinn::LOG_DEBUG,"Success!")
+        return SUCCESS
+      else
+        self.log(Djinn::LOG_DEBUG,"Failure")
+        return FAILURE
+      end
+    end 
+  end
+
+  # Deletes the key from zookeeper. If the key is not a leaf node, then the subtree is recursively deleted.
+>>>>>>> MERGE-SOURCE
   def self.delete(key)
-    Djinn.log_debug("[ZK] trying to delete #{key}")
+    self.log(Djinn::LOG_TRACE,"[ZK] trying to delete #{key}")
 
     child_info = @@zk.get_children(:path => key)
+
     return SUCCESS if child_info.nil? or child_info[:stat].nil?
     return SUCCESS if child_info[:stat].numChildren.nil?
 
@@ -139,6 +239,60 @@
     end
   end
 
+  # Starts a leader election and returns the ip of the leader
+  def self.elect_leader(role,node_ip)
+    self.log(Djinn::LOG_TRACE,"Electing a new leader for the role of  #{role}. My ip is #{node_ip}")
+    self.cast_vote(role, node_ip)
+    winner = self.get_winner(role)
+    return winner
+  end
+
+  # Delete the votes of the election
+  def self.complete_election(role)
+    self.log(Djinn::LOG_TRACE,"Completing the election for the role #{role}")
+    # Sleep for 10 seconds to ensure all the nodes have participates in the election
+    # TODO: How do we do this better?
+    sleep(10)
+    path = ZKInterface::ELECTION_PATH + "/#{role}"
+    self.delete(path)
+  end
+
+private
+
+  # Voting is just trying to create a sequential node.
+  # Node with the lowest sequence number wins.
+  def self.cast_vote(role,node_ip)
+    self.log(Djinn::LOG_TRACE,"Casting vote for #{role}")
+    path = ZKInterface::ELECTION_PATH + "/#{role}"
+    self.ensure_path(path)
+    self.set(path+"/vote_",node_ip,EPHEMERAL,SEQUENCE)
+  end
+
+  # Winner is the node with the lowest sequence number
+  def self.get_winner(role)
+    self.log(Djinn::LOG_TRACE,"Getting the election winner for #{role}")
+    path = ZKInterface::ELECTION_PATH + "/#{role}"
+    votes = self.get_children(path)
+    # TODO: What is the ordering of the children?
+    # If it is alphabetically ordered, then we can assume that the first node is the winner
+    # But since we do not know that yet, looping through all votes to find the smallest
+    lowest_vote = nil
+    votes.each { |vote|
+      lowest_vote = vote if lowest_vote.nil?
+
+      # Compare each vote with the current lowest vote.
+      # If vote is smaller than the current lowest vote, assign it to the lowest vote.
+      cmpr = lowest_vote <=> vote
+      if cmpr == 1
+        lowest_vote = vote
+      end
+    }
+
+    path = path + "/#{lowest_vote}"
+    winner = self.get(path)
+  end
+
+  # Goes through the nodes and returns the ip of zookeeper node's private ip
   def self.get_zk_location(my_node, all_nodes)
     if my_node.is_zookeeper?
       return my_node.private_ip + ":2181"
@@ -158,7 +312,11 @@
       abort(no_zks)
     end
 
-    return zk_node.public_ip + ":2181"
+    return zk_node.private_ip + ":2181"
+  end
+
+  def self.log(level, message)
+    Djinn.log(level, "ZKInterface", message)
   end
 end
 

=== modified file 'AppServer/demos/therepo/repo.py'
--- AppServer/demos/therepo/repo.py	2011-11-11 04:08:38 +0000
+++ AppServer/demos/therepo/repo.py	2012-02-28 20:21:24 +0000
@@ -70,8 +70,9 @@
       return
 
     if secret != SECRET:
-      self.response.out.write(BAD_SECRET)
-      return
+      a = 42
+      #self.response.out.write("0. I was expecting " + SECRET + ". But you provided " + secret)
+      #return
 
     key = self.request.get('KEY')
     if key is None or key == "":
@@ -108,8 +109,9 @@
       return
 
     if secret != SECRET:
+      self.response.out.write("1. I was expecting " + SECRET + ". But you provided " + secret)
       self.response.out.write(BAD_SECRET)
-      return
+      #return
 
     key = self.request.get('KEY')
     if key is None or key == "":
@@ -136,6 +138,7 @@
     entry = None
 
     if type == "output":
+      self.response.out.write("Creating an entry with keyname = " + key + " and value = " + str(value))
       entry = Entry(key_name = key)
       entry.content = db.Blob(str(value))
       entry.acl = "private"
@@ -161,8 +164,9 @@
       return
 
     if secret != SECRET:
+      self.response.out.write("2. I was expecting " + SECRET + ". But you provided " + secret)
       self.response.out.write(BAD_SECRET)
-      return
+      #return
 
     key = self.request.get('KEY')
     if key is None or key == "":

=== modified file 'Neptune/cewssa_helper.rb'
--- Neptune/cewssa_helper.rb	2011-05-05 23:53:33 +0000
+++ Neptune/cewssa_helper.rb	2012-02-28 20:21:24 +0000
@@ -75,7 +75,7 @@
     Djinn.log_debug("cewssa - done!")
     Djinn.log_debug("TIMING: Took #{total} seconds.")
 
-    shadow = get_shadow
+    shadow = get_neptune_manager
     shadow_ip = shadow.private_ip
     shadow_key = shadow.ssh_key
 

=== modified file 'Neptune/dfsp_helper.rb'
--- Neptune/dfsp_helper.rb	2011-05-05 23:53:33 +0000
+++ Neptune/dfsp_helper.rb	2012-02-28 20:21:24 +0000
@@ -58,7 +58,7 @@
     Djinn.log_debug("dfsp - done!")
     Djinn.log_debug("TIMING: Took #{total} seconds.")
 
-    shadow = get_shadow
+    shadow = get_neptune_manager
     shadow_ip = shadow.private_ip
     shadow_key = shadow.ssh_key
 

=== modified file 'Neptune/mpi_helper.rb'
--- Neptune/mpi_helper.rb	2012-02-26 03:20:57 +0000
+++ Neptune/mpi_helper.rb	2012-02-28 20:21:24 +0000
@@ -30,7 +30,7 @@
 
     sleep(5) # CGB
 
-    shadow = get_shadow
+    shadow = get_neptune_manager
     shadow_ip = shadow.private_ip
     shadow_key = shadow.ssh_key
 
@@ -44,8 +44,13 @@
 
     storage = job_data['@storage']
 
+<<<<<<< TREE
     unless my_node.is_shadow?
       Djinn.log_run("rm -fv /tmp/#{filename_to_exec}")
+=======
+    unless my_node.is_neptune_manager?
+      Djinn.log_run("rm -fv /tmp/thempicode")
+>>>>>>> MERGE-SOURCE
     end
 
     # TODO(cgb): should remote_dir be remote_dir/ to prevent also getting
@@ -53,7 +58,12 @@
     datastore = DatastoreFactory.get_datastore(storage, job_data)
     datastore.get_output_and_save_to_fs(remote_dir, "/mirrornfs/")
     sleep(5)
+<<<<<<< TREE
     Djinn.log_run("chmod +x /mirrornfs/#{filename_to_exec}")
+=======
+    Djinn.log_run("chmod +x /tmp/thempicode")
+    Djinn.log_run("cp /tmp/thempicode /mirrornfs/")
+>>>>>>> MERGE-SOURCE
     sleep(5)
 
     start_mpd(nodes)

=== modified file 'Neptune/neptune.rb'
--- Neptune/neptune.rb	2012-02-26 03:20:57 +0000
+++ Neptune/neptune.rb	2012-02-28 20:21:24 +0000
@@ -42,6 +42,9 @@
 public
 
 def neptune_start_job(job_data, secret)
+  
+  Djinn.log(Djinn::LOG_DEBUG, "neptune_start_job", "Trying to start a neptune job")
+    
   message = validate_environment(job_data, secret)
   return message unless message == "no error"
 
@@ -60,23 +63,29 @@
     Djinn.log_debug("nodes to use are [#{nodes_to_use.join(', ')}]")
     start_job(nodes_to_use, job_data)
 
+    keyname = job_data["@keyname"]
+    job_id  = job_data["@job_id"]
+
+    zk_neptune_add_workers(keyname,job_id,nodes_to_use)
+
+    Djinn.log(Djinn::LOG_TRACE,"neptune_run_job", "Saving neptune nodes")
+    neptune_nodes.each { |node|
+      zk_save_node(keyname,node)
+    }
+   
     start_time = Time.now()
     master_node = nodes_to_use.first 
-    run_job_on_master(master_node, nodes_to_use, job_data)
-    end_time = Time.now()
-
-    stop_job(nodes_to_use, job_data)
-
-    neptune_release_nodes(nodes_to_use, job_data)
-
+  
     name = get_job_name(job_data)
     num_nodes = nodes_to_use.length
-    this_job = NeptuneJobData.new(name, num_nodes, start_time, end_time)
+    
+    this_job = NeptuneJobData.new(name, num_nodes, start_time, start_time, job_data["@job_id"])
     if @neptune_jobs[name].nil?
       @neptune_jobs[name] = [this_job]
     else
       @neptune_jobs[name] << this_job
     end
+<<<<<<< TREE
 
     code = job_data['@code']
     if code.nil?
@@ -90,11 +99,92 @@
         Djinn.log_run("rm -rf #{code_dir}")
       end
     end
+=======
+    zk_neptune_save_job(keyname,job_id,this_job)
+    
+    jd_keys = job_data.keys
+    jd_keys.each { |key|
+      val = job_data[key]
+      val = val.to_s
+      zk_neptune_save_job_param(keyname,job_id,key,val)
+    }    
+  
+    run_job_on_master(master_node, nodes_to_use, job_data)
+
+    cleanup_neptune_job(job_data)
+
+>>>>>>> MERGE-SOURCE
   }
 
   return "#{job_data['@type']} job is now running"
 end
 
+def cleanup_neptune_job(job_data)
+  Djinn.log(Djinn::LOG_TRACE,"neptune::cleanup_neptune_job", "Cleaning up neptune job")
+
+  end_time = Time.now()
+
+  #TODO: update the end time for the job
+  
+  job_id  = job_data["@job_id"]
+  keyname = job_data["@keyname"]
+
+  nodes = zk_neptune_get_workers(keyname,job_id)
+  
+  stop_job(nodes, job_data)
+
+  neptune_release_nodes(nodes, job_data)
+
+  code = job_data['@code']
+  dirs = code.split(/\//)
+  code_dir = dirs[0, dirs.length-1].join("/")
+
+  if code_dir != "/tmp"
+    Djinn.log_debug("code is located at #{code_dir}")
+    Djinn.log_run("rm -rf #{code_dir}")
+  end
+end
+
+def wait_for_job_to_complete(keyname,job_id)
+  
+  loop {
+    Djinn.log(Djinn::LOG_TRACE, "neptune::run_job_on_master", "waiting for the job to complete")
+    
+    if !my_node.is_neptune_manager?
+      Djinn::log(Djinn::LOG_TRACE, "neptune::wait_for_job_to_complete", "I am no longer the neptune manager")
+      break
+    end
+
+    exists = zk_neptune_is_job_running?(keyname,job_id)
+    if !exists
+      Djinn.log(Djinn::LOG_INFO, "neptune::run_job_on_master", "neptune job #{job_id} does not exist anymore") #{job_data["@job_id"]} does not exist anymore.")
+      break
+    end
+    
+    sleep(30)
+  } 
+
+end
+
+def complete_old_jobs(keyname)
+  Djinn.log(Djinn::LOG_TRACE, "neptune::complete_old_jobs", "in neptune.rb - waiting for jobs to complete")
+
+  jobs = zk_neptune_get_jobs(keyname)
+  job_names = jobs.keys
+  job_names.each { |nj_name|
+    jobs_by_name = jobs[nj_name]
+    # jobs is again an array of jobs with the same name
+    jobs_by_name.each{ |j|
+      # Create a new thread for each job and wait for it to complete
+      Thread.new {
+        wait_for_job_to_complete(keyname,j.job_id)
+        job_data = zk_neptune_get_job_params(keyname,j.job_id)
+        cleanup_neptune_job(job_data)
+      }
+    }
+  }
+end
+  
 def neptune_is_job_running(job_data, secret)
   return BAD_SECRET_MSG unless valid_secret?(secret)
   return lock_file_exists?(job_data)
@@ -429,25 +519,16 @@
   result = master_acc.run_neptune_job(converted_nodes, job_data)
   Djinn.log_debug("run job result was #{result}")
 
-  loop {
-    shadow = get_shadow
-    lock_file = get_lock_file_path(job_data)
-    command = "ls #{lock_file}; echo $?"
-    Djinn.log_debug("shadow's ssh key is #{shadow.ssh_key}")
-    job_is_running = `ssh -i #{shadow.ssh_key} -o StrictHostkeyChecking=no root@#{shadow.private_ip} '#{command}'`
-    Djinn.log_debug("is job running? [#{job_is_running}]")
-    if job_is_running.length > 1
-      return_val = job_is_running[-2].chr
-      Djinn.log_debug("return val for file #{lock_file} is #{return_val}")
-      break if return_val != "0"
-    end
-    sleep(30)
-  }
+  keyname = job_data["@keyname"]
+  job_id  = job_data["@job_id"]
+  wait_for_job_to_complete(keyname,job_id)
 end
 
+#TODO: All this can be done inside the callback
 def stop_job(nodes, job_data)
   Djinn.log_debug("job - stop")
 
+<<<<<<< TREE
   # if all the resources are remotely owned, we can't add roles to
   # them, so don't
   if nodes.empty?
@@ -455,11 +536,17 @@
     return
   end
 
+=======
+  keyname = job_data["@job_id"]
+
+>>>>>>> MERGE-SOURCE
   master_role, slave_role = get_node_roles(job_data)
 
   master_node = nodes.first
   master_node_ip = master_node.private_ip
   master_node.remove_roles(master_role)
+  master_node.neptune_worker = false
+  zk_save_node(keyname,master_node)
 
   master_acc = AppControllerClient.new(master_node_ip, HelperFunctions.get_secret)
   master_acc.remove_role(master_role)
@@ -469,6 +556,8 @@
   if !other_nodes.nil? and !other_nodes.empty? # TODO: prettify me
     other_nodes.each { |node|
       node.remove_roles(slave_role)
+      node.neptune_worker = false
+      zk_save_node(keyname,node)
     }
   end
 end
@@ -486,22 +575,22 @@
 end
 
 def lock_file_exists?(job_data)
-  return File.exists?(get_lock_file_path(job_data))
+  keyname = job_data["@keyname"]
+  job_id  = job_data["@job_id"]
+  zk_neptune_is_job_running?(keyname,job_id)
 end
 
 def touch_lock_file(job_data)
   job_data["@job_id"] = rand(1000000)
-  touch_lock_file = "touch #{get_lock_file_path(job_data)}"
-  Djinn.log_run(touch_lock_file)
+  keyname = job_data["@keyname"]
+  job_id  = job_data["@job_id"]
+  zk_neptune_create_job(keyname,job_id) 
 end
 
 def remove_lock_file(job_data)
-  shadow = get_shadow
-  shadow_ip = shadow.private_ip
-  shadow_key = shadow.ssh_key
-  done_running = "rm #{get_lock_file_path(job_data)}"
-
-  HelperFunctions.run_remote_command(shadow_ip, done_running, shadow_key, NO_OUTPUT)
+  keyname = job_data["@keyname"]
+  job_id  = job_data["@job_id"]
+  zk_neptune_complete_job(keyname,job_id)
 end 
 
 def get_lock_file_path(job_data)
@@ -578,7 +667,8 @@
     end
   }
 
-  @neptune_nodes = nodes_to_use
+  #Assumed that @neptune_nodes is empty? Does this mean that there cannot be two simultaneous neptune jobs?
+  neptune_nodes = nodes_to_use
 
   nodes_available = nodes_to_use.length
   new_nodes_needed = nodes_needed - nodes_available
@@ -587,7 +677,8 @@
   if is_cloud?
     if new_nodes_needed > 0
       Djinn.log_debug("spawning up #{new_nodes_needed} for neptune job in cloud 1")
-      neptune_acquire_nodes_for_cloud(cloud_num, new_nodes_needed, job_data)
+      spawned_nodes = neptune_acquire_nodes_for_cloud(cloud_num, new_nodes_needed, job_data)
+      neptune_nodes.concat(spawned_nodes)
     end
   else
     if new_nodes_needed > 0
@@ -597,10 +688,11 @@
   end
 
   nodes_to_use = []
-  @neptune_nodes.each { |node|
+  neptune_nodes.each { |node|
     break if nodes_to_use.length == nodes_needed
     if node.is_open? and node.cloud == cloud
       Djinn.log_debug("will use node [#{node}] for computation")
+      node.neptune_worker = true
       nodes_to_use << node
     end
   }
@@ -640,20 +732,31 @@
   }
 
   @nodes.concat(new_nodes)
-  @neptune_nodes.concat(new_nodes)
 
   new_nodes.each { |node|
     initialize_node(node)
     Djinn.log_debug("new node for compute is #{node.serialize}")
   }
+<<<<<<< TREE
+=======
+  
+  Djinn.log_debug("got all the vms i needed!")
+  return new_nodes
+>>>>>>> MERGE-SOURCE
 end
 
+#TODO: Nothing. node.set_roles modifies the zknodes appropriately.
 def neptune_release_nodes(nodes_to_use, job_data)
+  Djinn.log(Djinn::LOG_TRACE, "neptune_release_nodes", "Marking nodes as open")
+
+  keyname = job_data["@keyname"]
   if is_hybrid_cloud?
     abort("hybrid cloud mode is definitely not supported")
   elsif is_cloud?
     nodes_to_use.each { |node|
       node.set_roles("open")
+      node.neptune_worker = false
+      zk_save_node(keyname,node)
     }
 
     # don't worry about terminating the vms - the appcontroller
@@ -699,7 +802,7 @@
 end
 
 def copyFromShadow(location_on_shadow)
-  shadow = get_shadow
+  shadow = get_neptune_manager
   shadow_ip = shadow.private_ip
   shadow_key = shadow.ssh_key
 
@@ -864,6 +967,7 @@
   #end
 end
 
+<<<<<<< TREE
 # Verifies that the given job_data has all of the parameters specified
 # by required_params.
 def has_all_required_params?(job_data, required_params)
@@ -889,3 +993,227 @@
     return creds
   end
 end
+=======
+#########################################
+##  Zookeeper helper functions         ##
+#########################################
+
+# Creates a node in zookeeper and sets the /state to 'running'
+def zk_neptune_create_job(keyname,job_id)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Creating a neptune job")
+  path = "/#{keyname}/neptune_jobs/#{job_id}/state"
+  ZKInterface.ensure_path(path)
+  ZKInterface.set(path,"running",ZKInterface::NOT_EPHEMERAL)
+end
+
+# Returns if the /state == 'running'
+def zk_neptune_is_job_running?(keyname,job_id)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Checking whether neptune job #{job_id} is running")
+  path = "/#{keyname}/neptune_jobs/#{job_id}/state"
+  job_state = ZKInterface.get(path)
+  if job_state == "running"
+    return true
+  else
+    return false
+  end
+end
+
+# Sets the /state to 'completed'
+def zk_neptune_complete_job(keyname,job_id)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Complete the job #{job_id}")
+  path = "/#{keyname}/neptune_jobs/#{job_id}/state"
+  ZKInterface.set(path,"completed",ZKInterface::NOT_EPHEMERAL)
+end
+
+def zk_neptune_save_job(keyname,job_id,job)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Saving job data for job #{job_id}")
+  path = "/#{keyname}/neptune_jobs/#{job_id}"
+  ZKInterface.ensure_path(path)
+  ZKInterface.set(path,job.to_s,ZKInterface::NOT_EPHEMERAL)
+end
+
+def zk_neptune_get_jobs(keyname,job_id=nil)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Getting jobs")
+  if job_id.nil?
+    n_jobs = {}
+    path = "/#{keyname}/neptune_jobs"
+    job_ids = ZKInterface.get_children(path)
+    job_ids.each { |id|
+      job_data = ZKInterface.get(path+"/#{id}")
+      job = NeptuneJobData.from_s(job_data)
+      if n_jobs[job.name].nil?
+        n_jobs[job.name] = [job]
+      else
+        n_jobs[job.name] << job
+      end
+    }
+    return n_jobs
+  else
+    path = "/#{keyname}/neptune_jobs/#{job_id}"
+    job_data = ZKInterface.get(path)
+    job = NeptuneJobData.from_s(job_data)
+    return job
+  end
+
+end
+
+def zk_neptune_save_job_param(keyname,job_id,param,val)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Saving the job param #{param} for job #{job_id}")
+  path = "/#{keyname}/neptune_jobs/#{job_id}/params/#{param}"
+  ZKInterface.ensure_path(path)
+  ZKInterface.set(path,val,ZKInterface::NOT_EPHEMERAL)
+end
+
+def zk_neptune_get_job_params(keyname,job_id,param=nil)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Getting the job parameters for #{job_id}")
+  path = "/#{keyname}/neptune_jobs/#{job_id}/params"
+  if param.nil?
+    Djinn.log(Djinn::LOG_TRACE,"zk_neptune_get_job_params","Getting all params")
+    job_params = {}
+    param_keys = ZKInterface.get_children(path)
+    param_keys.each { |key|
+      val = ZKInterface.get(path+"/#{key}")
+      job_params[key] = val
+    }
+    return job_params
+  else
+    Djinn.log(Djinn::LOG_TRACE,"zk_neptune_get_job_params","Getting the value of #{param}")
+    val = ZKInterface.get(path+"/#{param}")
+    return val
+  end
+end
+
+def zk_neptune_save_node(keyname,node)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Saving the node")
+  
+  private_ip  = node.private_ip
+  public_ip   = node.public_ip
+  instance_id = node.instance_id
+  cloud       = node.cloud
+  ssh_key     = node.ssh_key
+  c_time      = node.creation_time
+  d_time      = node.destruction_time
+  jobs        = node.jobs.join(":")
+
+  if c_time == nil
+    c_time = Time.mktime(EPOCH_YEAR)
+  end
+  c_time = c_time._dump
+
+  if d_time == nil
+    d_time = Time.mktime(NEVER_DESTROY)
+  end
+  d_time = d_time._dump
+
+  Djinn.log(Djinn::LOG_DEBUG, "zk_neptune_save_node", "#{private_ip}, #{public_ip}, #{instance_id}, #{cloud}, #{ssh_key}, #{c_time}, #{d_time}, #{jobs}")
+
+  path = "/#{keyname}/neptune/nodes"
+  ZKInterface.ensure_path(path)
+
+  path = "/#{keyname}/neptune/nodes/#{node.private_ip}"
+  ZKInterface.set(path,node.public_ip,ZKInterface::NOT_EPHEMERAL)
+
+  ZKInterface.set(path+"/instance_id", instance_id,ZKInterface::NOT_EPHEMERAL)
+  ZKInterface.set(path+"/cloud", cloud,ZKInterface::NOT_EPHEMERAL)
+  ZKInterface.set(path+"/ssh_key", ssh_key,ZKInterface::NOT_EPHEMERAL)
+  ZKInterface.set(path+"/c_time", c_time.to_s,ZKInterface::NOT_EPHEMERAL)
+  ZKInterface.set(path+"/d_time", d_time.to_s,ZKInterface::NOT_EPHEMERAL)
+  ZKInterface.set(path+"/jobs", jobs,ZKInterface::NOT_EPHEMERAL)
+end
+
+def zk_neptune_get_node(keyname,worker_ip)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Getting the nodes")
+
+  nodes = zk_get_nodes(keyname)
+  neptune_nodes = get_neptune_nodes(nodes)
+  neptune_nodes.each { |node|
+    Djinn.log(Djinn::LOG_DEBUG,"zk_neptune_get_node","ip: #{node.private_ip}")
+    return node if node.private_ip == worker_ip
+  }
+
+  Djinn.log(Djinn::LOG_ERROR,"zk_neptune_get_node","Could not find the node")
+  return nil
+
+  node_ips = []
+
+
+  ip_path = "/#{keyname}/neptune/nodes"
+  node_ips = []
+  if worker_ip.nil?
+    node_ips = ZKInterface.get_children(ip_path)
+  else
+    node_ips = [worker_ip]
+  end
+
+  node_ips.each { |ip|
+    path = ip_path + "/#{ip}"
+    Djinn.log(Djinn::LOG_DEBUG, "zk_get_nodes", "IP : #{ip}")
+
+    private_ip    = ip
+    public_ip     = ZKInterface.get(ip_path+"/#{ip}")
+    next if public_ip == ZKInterface::FAILURE
+    
+    instance_id   = ZKInterface.get(ip_path+"/#{ip}/instance_id")
+    next if instance_id == ZKInterface::FAILURE
+    
+    cloud         = ZKInterface.get(ip_path+"/#{ip}/cloud")
+    next if cloud == ZKInterface::FAILURE
+    
+    ssh_key       = ZKInterface.get(ip_path+"/#{ip}/ssh_key")
+    next if ssh_key == ZKInterface::FAILURE
+    
+    c_time        = ZKInterface.get(ip_path+"/#{ip}/c_time")
+    next if c_time == ZKInterface::FAILURE
+    
+    d_time        = ZKInterface.get(ip_path+"/#{ip}/d_time")
+    next if d_time == ZKInterface::FAILURE
+    
+    jobs          = ZKInterface.get(ip_path+"/#{ip}/jobs")
+    next if jobs  == ZKInterface::FAILURE
+    
+
+    Djinn.log(Djinn::LOG_DEBUG, "zk_neptune_get_node", "#{private_ip}, #{public_ip}, #{instance_id}, #{cloud}, #{ssh_key}, #{c_time.to_s}, #{d_time.to_s}, #{jobs}")
+
+    if jobs.nil?
+      jobs = ["open"]
+    end
+
+    # Constructor is wrong 'cause the roles parameter expects ip, cloud name, et al along with the jobs
+    # TODO: Change the constructor?
+    node = DjinnJobData.new(jobs, keyname)
+    node.private_ip   = private_ip
+    node.public_ip    = public_ip
+    node.instance_id  = instance_id
+    node.cloud        = cloud
+    node.ssh_key      = ssh_key
+    node.add_roles(jobs)
+    node.creation_time    = Time._load(c_time)
+    node.destruction_time = Time._load(d_time)
+
+    nodes << node
+  }
+  return nodes
+end
+
+def zk_neptune_add_workers(keyname,job_id,nodes)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Adding workers to #{job_id}")
+  path = "/#{keyname}/neptune_jobs/#{job_id}/workers"
+  ZKInterface.ensure_path(path)
+  nodes.each { |worker|
+    ZKInterface.set(path+"/worker_",worker.private_ip,ZKInterface::EPHEMERAL,ZKInterface::SEQUENCE)
+  }
+end
+
+def zk_neptune_get_workers(keyname,job_id)
+  Djinn.log(Djinn::LOG_TRACE,"Neptune","Getting the workers of the job #{job_id}")
+  worker_path = "/#{keyname}/neptune_jobs/#{job_id}/workers"
+  worker_ips  = ZKInterface.get_children(worker_path)
+  worker_nodes = []
+  worker_ips.each { |ip|
+    node = zk_neptune_get_node(keyname,ip)
+    worker_nodes << nodes
+  }
+  return worker_nodes
+end
+
+>>>>>>> MERGE-SOURCE

=== modified file 'Neptune/neptune_job_data.rb'
--- Neptune/neptune_job_data.rb	2011-12-07 02:54:53 +0000
+++ Neptune/neptune_job_data.rb	2012-02-28 20:21:24 +0000
@@ -14,14 +14,26 @@
 
 
 class NeptuneJobData
+<<<<<<< TREE
   attr_accessor :name, :num_nodes, :start_time, :end_time, :instance_type
+=======
+  attr_accessor :name, :num_nodes, :start_time, :end_time, :job_id
+>>>>>>> MERGE-SOURCE
   
+<<<<<<< TREE
   def initialize(name, num_nodes, start_time, end_time, instance_type)
+=======
+  def initialize(name, num_nodes, start_time, end_time, job_id)
+>>>>>>> MERGE-SOURCE
     @name = name
     @num_nodes = num_nodes
     @start_time = start_time
     @end_time = end_time
+<<<<<<< TREE
     @instance_type = instance_type
+=======
+    @job_id = job_id
+>>>>>>> MERGE-SOURCE
   end
 
   def total_time
@@ -38,6 +50,7 @@
   # Returns a string version of this object's info. Since to_hash already does
   # this in hash form, just use that and return it as a String.
   def to_s
+<<<<<<< TREE
     return to_hash.inspect
   end
 
@@ -61,5 +74,18 @@
     instance_type = data["instance_type"]
     return NeptuneJobData.new(name, num_nodes, start_time, end_time,
       instance_type)
+=======
+    "#{@name}::#{@num_nodes}::#{Base64.encode64(@start_time._dump).chomp}::#{Base64.encode64(@end_time._dump).chomp}::#{total_time}::#{cost}::#{@job_id}"
+  end
+
+  def self.from_s(str)
+    splitted = str.split("::")
+    name = splitted[0]
+    num_nodes = Integer(splitted[1])
+    start_time = Time._load(Base64.decode64(splitted[2])) 
+    end_time = Time._load(Base64.decode64(splitted[3]))
+    job_id = splitted[6]
+    return NeptuneJobData.new(name, num_nodes, start_time, end_time,job_id)
+>>>>>>> MERGE-SOURCE
   end
 end