job_flow.rb in elasticity-4.0

- old
+ new

@@ -2,38 +2,38 @@
 
   class JobFlowRunningError < StandardError; end
   class JobFlowNotStartedError < StandardError; end
   class JobFlowMissingStepsError < StandardError; end
   class LogUriMissingError < StandardError; end
+  class UnknownPlacementError < StandardError; end
 
   class JobFlow
 
     attr_accessor :action_on_failure
     attr_accessor :ec2_key_name
     attr_accessor :name
-    attr_accessor :hadoop_version
     attr_accessor :instance_count
     attr_accessor :log_uri
     attr_accessor :master_instance_type
     attr_accessor :slave_instance_type
     attr_accessor :ami_version
     attr_accessor :keep_job_flow_alive_when_no_steps
     attr_accessor :ec2_subnet_id
     attr_accessor :placement
+    attr_accessor :region
     attr_accessor :visible_to_all_users
     attr_accessor :enable_debugging
 
     attr_reader :access_key
     attr_reader :secret_key
 
     def initialize(access=nil, secret=nil)
       @action_on_failure = 'TERMINATE_JOB_FLOW'
-      @hadoop_version = '1.0.3'
       @name = 'Elasticity Job Flow'
       @ami_version = 'latest'
       @keep_job_flow_alive_when_no_steps = false
-      @placement = 'us-east-1a'
+      self.placement = 'us-east-1a'
       @enable_debugging = false
 
       @access_key = access
       @secret_key = secret
       @visible_to_all_users = false
@@ -59,10 +59,24 @@
         j.instance_variable_set(:@jobflow_id, jobflow_id)
         j.instance_variable_set(:@installed_steps, j.status.installed_steps)
       end
     end
 
+    def placement=(new_placement)
+      @placement = new_placement
+      return unless @placement
+
+      # The region has to be set so we know where to launch the job flow, and it has
+      # to be in sync with the placement.  Not setting this will lead to strange, difficult to
+      # track errors, for example the job flow won't exist where you think it should, etc.
+      if @placement =~ /\w+-\w+-\d+/
+        @region = @placement.match(/(\w+-\w+-\d+)/)[0]
+      else
+        raise UnknownPlacementError, "'#{@placement}' is not a valid EMR placement"
+      end
+    end
+
     def enable_debugging=(enabled)
       if enabled
         raise LogUriMissingError, 'To enable debugging, please set a #log_uri' unless @log_uri
       end
       @enable_debugging = enabled
@@ -156,11 +170,10 @@
       jf_status = status
       return status.state == 'RUNNING' || status.state == 'STARTING', jf_status
     end
 
     def emr
-      @region ||= @placement.match(/(\w+-\w+-\d+)/)[0]
       @emr ||= Elasticity::EMR.new(@access_key, @secret_key, :region => @region)
     end
 
     def is_jobflow_running?
       !@jobflow_id.nil?
@@ -181,10 +194,9 @@
         :name => @name,
         :ami_version => @ami_version,
         :visible_to_all_users => @visible_to_all_users,
         :instances => {
           :keep_job_flow_alive_when_no_steps => @keep_job_flow_alive_when_no_steps,
-          :hadoop_version => @hadoop_version,
           :instance_groups => jobflow_instance_groups,
           :placement => {
             :availability_zone => @placement
           }
         }
\ No newline at end of file