cluster.rb in elasticsearch-extensions-0.0.23

- old
+ new

@@ -159,21 +159,24 @@
                 -E script.stored=true \
                 -E node.attr.testattr=test \
                 -E path.repo=/tmp \
                 -E repositories.url.allowed_urls=http://snapshot.test* \
                 -E discovery.zen.minimum_master_nodes=#{arguments[:number_of_nodes]-1} \
+                -E node.max_local_storage_nodes=#{arguments[:number_of_nodes]} \
                 -E logger.level=DEBUG \
                 #{arguments[:es_params]} \
                 > /dev/null
               COMMAND
             }
           }
+          COMMANDS['6.0'] = COMMANDS['5.0']
+          COMMANDS.freeze
 
           # Create a new instance of the Cluster class
           #
           # @option arguments [String]  :cluster_name Cluster name (default: `elasticsearch_test`)
-          # @option arguments [Integer] :nodes        Number of desired nodes (default: 2)
+          # @option arguments [Integer] :number_of_nodes Number of desired nodes (default: 2)
           # @option arguments [String]  :command      Elasticsearch command (default: `elasticsearch`)
           # @option arguments [String]  :port         Starting port number; will be auto-incremented (default: 9250)
           # @option arguments [String]  :node_name    The node name (will be appended with a number)
           # @option arguments [String]  :path_data    Path to the directory to store data in
           # @option arguments [String]  :path_work    Path to the directory with auxiliary files
@@ -186,11 +189,11 @@
           # You can also use environment variables to set the constructor options (see source).
           #
           # @see Cluster#start
           #
           def initialize(arguments={})
-            @arguments = arguments
+            @arguments = arguments.dup
 
             @arguments[:command]           ||= ENV.fetch('TEST_CLUSTER_COMMAND',   'elasticsearch')
             @arguments[:port]              ||= ENV.fetch('TEST_CLUSTER_PORT',      9250).to_i
             @arguments[:cluster_name]      ||= ENV.fetch('TEST_CLUSTER_NAME',      __default_cluster_name).chomp
             @arguments[:node_name]         ||= ENV.fetch('TEST_CLUSTER_NODE_NAME', 'node')
@@ -219,11 +222,11 @@
           #      Elasticsearch::Extensions::Test::Cluster::Cluster.new.start
           #
           # @example Start a cluster with a custom configuration
           #      Elasticsearch::Extensions::Test::Cluster::Cluster.new(
           #        cluster_name: 'my-cluster',
-          #        nodes: 3,
+          #        number_of_nodes: 3,
           #        node_name: 'my-node',
           #        port: 9350
           #      ).start
           #
           # @example Start a cluster with a different Elasticsearch version
@@ -238,11 +241,11 @@
             if self.running?
               STDOUT.print "[!] Elasticsearch cluster already running".ansi(:red)
               return false
             end
 
-            __remove_cluster_data
+            __remove_cluster_data if @clear_cluster
 
             STDOUT.print "Starting ".ansi(:faint) + arguments[:number_of_nodes].to_s.ansi(:bold, :faint) +
                          " Elasticsearch nodes..".ansi(:faint)
             pids = []
 
@@ -254,10 +257,11 @@
               STDERR.puts command.gsub(/ {1,}/, ' ') if ENV['DEBUG']
 
               pid = Process.spawn(command)
               Process.detach pid
               pids << pid
+              sleep 1
             end
 
             __check_for_running_processes(pids)
             wait_for_green
             __print_cluster_info
@@ -363,13 +367,13 @@
           def __default_network_host
             case version
               when /^0|^1/
                 '0.0.0.0'
               when /^2/
-                '0.0.0.0'
-              when /^5/
                 '_local_'
+              when /^5|^6/
+                '_local_'
               else
                 raise RuntimeError, "Cannot determine default network host from version [#{version}]"
             end
           end
 
@@ -416,22 +420,38 @@
                 m[1]
               else
                 raise RuntimeError, "Cannot determine Elasticsearch version from jar [#{jar}]"
               end
             else
-              STDERR.puts "[!] Cannot find Elasticsearch .jar from path to command [#{arguments[:command]}], using `elasticsearch --version`" if ENV['DEBUG']
+              STDERR.puts "[!] Cannot find Elasticsearch .jar from path to command [#{arguments[:command]}], using `#{arguments[:command]} --version`" if ENV['DEBUG']
 
+              unless File.exist? arguments[:command]
+                raise Errno::ENOENT, "File [#{arguments[:command]}] does not exist -- did you pass a correct path to the Elasticsearch launch script"
+              end
+
               output = ''
 
               begin
                 # First, try the new `--version` syntax...
                 STDERR.puts "Running [#{arguments[:command]} --version] to determine version" if ENV['DEBUG']
-                Timeout::timeout(10) { output = `#{arguments[:command]} --version` }
+                rout, wout = IO.pipe
+                pid = Process.spawn("#{arguments[:command]} --version", out: wout)
+
+                Timeout::timeout(10) do
+                  Process.wait(pid)
+                  wout.close unless wout.closed?
+                  output = rout.read unless rout.closed?
+                  rout.close unless rout.closed?
+                end
               rescue Timeout::Error
-                # ...else, the new `-v` syntax
+                # ...else, the old `-v` syntax
                 STDERR.puts "Running [#{arguments[:command]} -v] to determine version" if ENV['DEBUG']
                 output = `#{arguments[:command]} -v`
+              ensure
+                Process.kill('INT', pid) if pid
+                wout.close unless wout.closed?
+                rout.close unless rout.closed?
               end
 
               STDERR.puts "> #{output}" if ENV['DEBUG']
 
               if output.empty?
@@ -452,10 +472,12 @@
                 '1.0'
               when /^2\..*/
                 '2.0'
               when /^5\..*/
                 '5.0'
+              when /^6\..*/
+                '6.0'
               else
                 raise RuntimeError, "Cannot determine major version from [#{version}]"
             end
           end
 
@@ -483,21 +505,29 @@
           # @api private
           #
           # @return Boolean
           #
           def __wait_for_status(status='green', timeout=30)
-            Timeout::timeout(timeout) do
-              loop do
-                response = __get_cluster_health(status)
+            begin
+              Timeout::timeout(timeout) do
+                loop do
+                  response = __get_cluster_health(status)
+                  STDERR.puts response if ENV['DEBUG']
 
-                if response && response['status'] == status && ( arguments[:number_of_nodes].nil? || arguments[:number_of_nodes].to_i == response['number_of_nodes'].to_i  )
-                  break
-                end
+                  if response && response['status'] == status && ( arguments[:number_of_nodes].nil? || arguments[:number_of_nodes].to_i == response['number_of_nodes'].to_i  )
+                    break
+                  end
 
-                STDOUT.print '.'.ansi(:faint)
-                sleep 1
+                  STDOUT.print '.'.ansi(:faint)
+                  sleep 1
+                end
               end
+            rescue Timeout::Error => e
+              message = "\nTimeout while waiting for cluster status [#{status}]"
+              message += " and [#{arguments[:number_of_nodes]}] nodes" if arguments[:number_of_nodes]
+              STDOUT.puts message.ansi(:red, :bold)
+              raise e
             end
 
             return true
           end
 
@@ -551,16 +581,15 @@
             end
 
             JSON.parse(response)
           end
 
-          # Remove the data directory (unless it has been disabled by arguments)
+          # Remove the data directory
           #
           # @api private
           #
           def __remove_cluster_data
-            # Wipe out data on disk for this cluster name by default
-            FileUtils.rm_rf "#{arguments[:path_data]}/#{arguments[:cluster_name]}" if @clear_cluster
+            FileUtils.rm_rf arguments[:path_data]
           end
 
 
           # Check whether process for PIDs are running
           #