indexer.rb

require 'manticore'

require 'json'

HOST = "http://localhost:9200"
JSON_HEADERS = {"Content-Type" => "application/json"} 

#client = Manticore::Client.new(:auth => {:user => 'elastic', :password => 'changeme' }, :ssl => { :verify => :none })
client = Manticore::Client.new

puts client.get(HOST).body

pipeline_body = %q[
{
  "processors": [
          {
        "script": {
          "lang": "painless",
          "source": "ctx.putIfAbsent(\"event\", [:]); ctx.event.ingested = metadata().now.format(DateTimeFormatter.ISO_INSTANT);"
        }
      }
  ]
}
]

puts JSON.parse(client.put("#{HOST}/_ingest/pipeline/my-pipeline", :body => pipeline_body, :headers => JSON_HEADERS).body)

template_body = %q[
{
  "index_patterns": ["test-*"],
  "settings": {
    "index.lifecycle.name": "my_policy",
    "index.lifecycle.rollover_alias": "test-alias",
    "index.default_pipeline": "my-pipeline",
    "index.number_of_shards": 5,
    "index.number_of_replicas": 0
  },
  "mappings": {
    "properties": {
      "event": {
        "properties": {
          "ingested": {
            "type": "date_nanos",
            "format": "strict_date_optional_time_nanos"
          }
        }
      }
    }
  }
}
]

puts JSON.parse(client.put("#{HOST}/_template/my_template", :body => template_body, :headers => JSON_HEADERS).body)

puts JSON.parse(client.delete("#{HOST}/test-000001").body)

create_index = %q[
{
  "aliases": {
    "test-alias":{
      "is_write_index": true
    }
  }
}
]
puts JSON.parse(client.put("#{HOST}/test-000001", :body => create_index, :headers => JSON_HEADERS).body)

total = 0

bulk_action = %q[{ "create" : { "_index" : "test-alias" } }
{ "sequence" : :sequence }
]

loop do
  bulk_body = 2000.times.map {|i| bulk_action.sub(":sequence", (i+total).to_s) }.join

  start = Time.now
  begin
  response_code = client.post("#{HOST}/_bulk", :body => bulk_body, :headers => JSON_HEADERS).code
  rescue => e
    puts e.inspect
    sleep 1
    retry
  end
  elapsed = Time.now - start
  puts "Bulk took too long: #{elapsed} seconds" if elapsed > 1

  total += 2000; puts "Written: #{total} documents" if total % 100_000 == 0
  break if total == 10_000_000
  sleep 0.01 # don't go too fast
end

client.post("#{HOST}/test-alias/_refresh").body
puts JSON.parse(client.get("#{HOST}/test-alias/_count").body)["count"]