Merge pull request danielberkompas#27 from infinitered/26-bulk-configs

[danielberkompas#26] Configure bulk settings on indexes

Author: danielberkompas

README.md

@@ -59,15 +59,6 @@ config :my_app, MyApp.ElasticsearchCluster,
   username: "username",
   password: "password",
 
-  # When indexing data using the `mix elasticsearch.build` task,
-  # control the data ingestion rate by raising or lowering the number
-  # of items to send in each bulk request.
-  bulk_page_size: 5000,
-
-  # Likewise, wait a given period between posting pages to give
-  # Elasticsearch time to catch up.
-  bulk_wait_interval: 15_000, # 15 seconds
-
   # If you want to mock the responses of the Elasticsearch JSON API
   # for testing or other purposes, you can inject a different module
   # here. It must implement the Elasticsearch.API behaviour.
@@ -100,7 +91,16 @@ config :my_app, MyApp.ElasticsearchCluster,
       #
       # Each piece of data that is returned by the store must implement the
       # Elasticsearch.Document protocol.
-      sources: [MyApp.Post]
+      sources: [MyApp.Post],
+
+      # When indexing data using the `mix elasticsearch.build` task,
+      # control the data ingestion rate by raising or lowering the number
+      # of items to send in each bulk request.
+      bulk_page_size: 5000,
+
+      # Likewise, wait a given period between posting pages to give
+      # Elasticsearch time to catch up.
+      bulk_wait_interval: 15_000 # 15 seconds
     }
   }
 ```

config/config.exs

@@ -13,4 +13,4 @@ config :elasticsearch, Elasticsearch.Test.Repo,
 
 config :elasticsearch, ecto_repos: [Elasticsearch.Test.Repo]
 
-config :logger, level: :warn
+config :logger, level: :debug

guides/upgrading/0.3.x_to_0.4.x.md

@@ -0,0 +1,55 @@
+# Upgrading from 0.3.x to 0.4.x
+
+Version `0.4.0` moved the `:bulk_page_size` and `:bulk_wait_interval` settings
+into each index's configuration, rather than global on the cluster.
+
+## Rationale
+
+This makes it easier to have different bulk settings for different indexes.
+
+## Changes
+
+**BREAKING**: `:bulk_page_size` and `:bulk_wait_interval` are now configured
+on the index, not the cluster.
+
+**FEATURE**: You can now pass `--bulk-wait-interval` and `--bulk-page-size`
+options to `mix elasticsearch.build`, as requested in [#26](https://github.com/infinitered/elasticsearch-elixir/issues/26).
+
+## How to Update Your App
+
+Move the `:bulk_page_size` and `:bulk_wait_interval` settings to your index
+rather than your entire cluster.
+
+    # BEFORE
+    config :my_app, MyApp.ElasticsearchCluster,
+      api: Elasticsearch.API.HTTP,
+      json_library: Poison,
+      url: "http://localhost:9200",
+      username: "username",
+      password: "password",
+      bulk_page_size: 5000,
+      bulk_wait_interval: 0,
+      indexes: %{
+        posts: %{
+          settings: "test/support/settings/posts.json",
+          store: Elasticsearch.Test.Store,
+          sources: [Post]
+        }
+      }
+
+    # AFTER
+    config :my_app, MyApp.ElasticsearchCluster,
+      api: Elasticsearch.API.HTTP,
+      json_library: Poison,
+      url: "http://localhost:9200",
+      username: "username",
+      password: "password",
+      indexes: %{
+        posts: %{
+          settings: "test/support/settings/posts.json",
+          store: Elasticsearch.Test.Store,
+          sources: [Post],
+          bulk_page_size: 5000,
+          bulk_wait_interval: 0
+        }
+      }

lib/elasticsearch/cluster/config.ex

@@ -34,9 +34,7 @@ defmodule Elasticsearch.Cluster.Config do
              username: [presence: [unless: &(&1[:password] == nil)]],
              password: [presence: [unless: &(&1[:username] == nil)]],
              api: [presence: true, by: &is_module/1],
-             json_library: [by: &(is_nil(&1) || is_module(&1))],
-             bulk_page_size: [presence: true, by: &is_integer/1],
-             bulk_wait_interval: [presence: true, by: &is_integer/1]
+             json_library: [by: &(is_nil(&1) || is_module(&1))]
            ),
          :ok <- validate_indexes(config[:indexes] || %{}) do
       {:ok, config}
@@ -80,7 +78,9 @@ defmodule Elasticsearch.Cluster.Config do
       sources: [
         presence: true,
         by: &(is_list(&1) && Enum.map(&1, fn source -> is_atom(source) end))
-      ]
+      ],
+      bulk_page_size: [presence: true, by: &is_integer/1],
+      bulk_wait_interval: [presence: true, by: &is_integer/1]
     )
   end
 end

lib/elasticsearch/indexing/bulk.ex

@@ -79,28 +79,35 @@ defmodule Elasticsearch.Index.Bulk do
   """
   @spec upload(Cluster.t(), index_name :: String.t(), Elasticsearch.Store.t(), list) ::
           :ok | {:error, [map]}
-  def upload(cluster, index_name, store, sources, errors \\ [])
-  def upload(_cluster, _index_name, _store, [], []), do: :ok
-  def upload(_cluster, _index_name, _store, [], errors), do: {:error, errors}
-
-  def upload(cluster, index_name, store, [source | tail] = _sources, errors)
+  def upload(cluster, index_name, index_config, errors \\ [])
+  def upload(_cluster, _index_name, %{sources: []}, []), do: :ok
+  def upload(_cluster, _index_name, %{sources: []}, errors), do: {:error, errors}
+
+  def upload(
+        cluster,
+        index_name,
+        %{store: store, sources: [source | tail]} = index_config,
+        errors
+      )
       when is_atom(store) do
     config = Cluster.Config.get(cluster)
+    bulk_page_size = index_config[:bulk_page_size] || 5000
+    bulk_wait_interval = index_config[:bulk_wait_interval] || 0
 
     errors =
-      config
-      |> DataStream.stream(source, store)
+      source
+      |> DataStream.stream(store, bulk_page_size)
       |> Stream.map(&encode!(config, &1, index_name))
-      |> Stream.chunk_every(config.bulk_page_size)
-      |> Stream.intersperse(config.bulk_wait_interval)
+      |> Stream.chunk_every(bulk_page_size)
+      |> Stream.intersperse(bulk_wait_interval)
       |> Stream.map(&put_bulk_page(config, index_name, &1))
       |> Enum.reduce(errors, &collect_errors/2)
 
-    upload(cluster, index_name, store, tail, errors)
+    upload(config, index_name, %{index_config | sources: tail}, errors)
   end
 
   defp put_bulk_page(_config, _index_name, wait_interval) when is_integer(wait_interval) do
-    IO.puts("Pausing #{wait_interval}ms between bulk pages")
+    Logger.debug("Pausing #{wait_interval}ms between bulk pages")
     :timer.sleep(wait_interval)
   end
 

lib/elasticsearch/indexing/index.ex

@@ -23,24 +23,20 @@ defmodule Elasticsearch.Index do
 
       iex> file = "test/support/settings/posts.json"
       ...> store = Elasticsearch.Test.Store
-      ...> Index.hot_swap(Cluster, "posts", file, store, [Post])
+      ...> Index.hot_swap(Cluster, "posts", %{settings: file, store: store, sources: [Post]})
       :ok
   """
-  @spec hot_swap(
-          Cluster.t(),
-          alias :: String.t() | atom,
-          settings_path :: String.t(),
-          Elasticsearch.Store.t(),
-          list
-        ) ::
-          :ok
-          | {:error, Elasticsearch.Exception.t()}
-  def hot_swap(cluster, alias, settings_file, store, sources) do
+  @spec hot_swap(Cluster.t(), alias :: String.t() | atom, %{
+          settings: Path.t(),
+          store: module,
+          sources: [any]
+        }) :: :ok | {:error, Elasticsearch.Exception.t()}
+  def hot_swap(cluster, alias, %{settings: settings_file} = index_config) do
     name = build_name(alias)
     config = Config.get(cluster)
 
     with :ok <- create_from_file(config, name, settings_file),
-         :ok <- Bulk.upload(config, name, store, sources),
+         :ok <- Bulk.upload(config, name, index_config),
          :ok <- __MODULE__.alias(config, name, alias),
          :ok <- clean_starting_with(config, alias, 2),
          :ok <- refresh(config, name) do

lib/elasticsearch/storage/data_stream.ex

@@ -13,22 +13,21 @@ defmodule Elasticsearch.DataStream do
 
   ## Example
 
-      iex> stream = DataStream.stream(Cluster, MyApp.Schema, Elasticsearch.Test.Store)
+      iex> stream = DataStream.stream(MyApp.Schema, Elasticsearch.Test.Store, 100)
       ...> is_function(stream)
       true
 
   """
   @spec stream(Cluster.t(), source, Elasticsearch.Store.t()) :: Stream.t()
-  def stream(cluster, source, store) do
-    config = Cluster.Config.get(cluster)
-    Stream.resource(fn -> init(config) end, &next(&1, source, store), &finish/1)
+  def stream(source, store, bulk_page_size) do
+    Stream.resource(fn -> init(bulk_page_size) end, &next(&1, source, store), &finish/1)
   end
 
   # Store state in the following format:
   #
   # {items, offset, limit}
-  defp init(config) do
-    {[], 0, config.bulk_page_size}
+  defp init(bulk_page_size) do
+    {[], 0, bulk_page_size}
   end
 
   # If no items, load another page of items

lib/mix/elasticsearch.build.ex

@@ -11,6 +11,16 @@ defmodule Mix.Tasks.Elasticsearch.Build do
   For a functional version of this approach, see
   `Elasticsearch.Index.hot_swap/4`.
 
+  ## Options
+
+  `--cluster`: The `Elasticsearch.Cluster` to build the indexes to.
+
+  `--bulk-page-size`: (Optional) The number of documents to post to the cluster in each
+  bulk page upload. Default: 5000
+
+  `--bulk-wait-interval`: (Optional) The number of milliseconds to wait between posting
+  each bulk page, to avoid overloading your cluster. Default: 0
+
   ## Example
 
       $ mix elasticsearch.build posts [index2] [index3] --cluster MyApp.Cluster
@@ -19,10 +29,16 @@ defmodule Mix.Tasks.Elasticsearch.Build do
 
       $ mix elasticsearch.build posts --existing --cluster MyApp.Cluster
       Index posts already exists.
+
+  You can also specify `--bulk-page-size` and `--bulk-wait-interval` manually:
+
+      $ mix elasticsearch.build posts --cluster MyApp.Cluster --bulk-page-size 1000 --bulk-wait-interval 500
   """
 
   require Logger
 
+  import Maybe
+
   alias Elasticsearch.{
     Cluster.Config,
     Index
@@ -32,31 +48,33 @@ defmodule Mix.Tasks.Elasticsearch.Build do
   def run(args) do
     Mix.Task.run("app.start", [])
 
-    {cluster, indexes, type} = parse_args!(args)
+    {type, cluster, indexes, settings} = parse_args!(args)
     config = Config.get(cluster)
 
     for alias <- indexes do
-      build(config, alias, type)
+      build(type, config, alias, settings)
     end
   end
 
-  defp build(config, alias, :existing) do
+  defp build(:existing, config, alias, settings) do
     case Index.latest_starting_with(config, alias) do
       {:ok, name} ->
         IO.puts("Index already exists: #{name}")
 
       {:error, :not_found} ->
-        build(config, alias, :rebuild)
+        build(:rebuild, config, alias, settings)
 
       {:error, exception} ->
-        Mix.raise(exception)
+        Mix.raise("""
+        Index could not be built.
+
+            #{inspect(exception)}
+        """)
     end
   end
 
-  defp build(config, alias, :rebuild) do
-    %{settings: settings, store: store, sources: sources} = config.indexes[alias]
-
-    with :ok <- Index.hot_swap(config, alias, settings, store, sources) do
+  defp build(:rebuild, config, alias, settings) do
+    with :ok <- Index.hot_swap(config, alias, Map.merge(config.indexes[alias], settings)) do
       :ok
     else
       {:error, errors} when is_list(errors) ->
@@ -66,12 +84,12 @@ defmodule Mix.Tasks.Elasticsearch.Build do
         Index created, but not aliased: #{alias}
         The following errors occurred:
 
-        #{errors}
+            #{errors}
         """)
 
       {:error, :enoent} ->
         Mix.raise("""
-        Schema file not found at #{settings}.
+        Settings file not found at #{maybe(config, [:indexes, alias, :settings])}.
         """)
 
       {:error, exception} ->
@@ -80,14 +98,20 @@ defmodule Mix.Tasks.Elasticsearch.Build do
 
             #{inspect(exception)}
         """)
-
-      error ->
-        Mix.raise(error)
     end
   end
 
   defp parse_args!(args) do
-    {options, indexes} = OptionParser.parse!(args, strict: [cluster: :string, existing: :boolean])
+    {options, indexes} =
+      OptionParser.parse!(
+        args,
+        strict: [
+          cluster: :string,
+          existing: :boolean,
+          bulk_page_size: :integer,
+          bulk_wait_interval: :integer
+        ]
+      )
 
     cluster =
       if options[:cluster] do
@@ -108,7 +132,12 @@ defmodule Mix.Tasks.Elasticsearch.Build do
 
     type = if options[:existing], do: :existing, else: :rebuild
 
-    {cluster, indexes, type}
+    settings =
+      options
+      |> Keyword.take([:bulk_page_size, :bulk_wait_interval])
+      |> Enum.into(%{})
+
+    {type, cluster, indexes, settings}
   end
 
   defp validate_indexes!(indexes, cluster) do

mix.exs

@@ -6,7 +6,7 @@ defmodule Elasticsearch.Mixfile do
       app: :elasticsearch,
       description: "Elasticsearch without DSLs",
       source_url: "https://github.com/infinitered/elasticsearch-elixir",
-      version: "0.3.1",
+      version: "0.4.0",
       elixir: "~> 1.5",
       start_permanent: Mix.env() == :prod,
       elixirc_paths: elixirc_paths(Mix.env()),
@@ -61,6 +61,7 @@ defmodule Elasticsearch.Mixfile do
       {:poison, ">= 0.0.0", optional: true},
       {:httpoison, ">= 0.0.0"},
       {:vex, "~> 0.6.0"},
+      {:maybe, "~> 1.0.0"},
       {:postgrex, ">= 0.0.0", only: [:dev, :test]},
       {:ex_doc, ">= 0.0.0", only: [:dev, :test]},
       {:ecto, ">= 0.0.0", only: [:dev, :test]},
@@ -73,8 +74,9 @@ defmodule Elasticsearch.Mixfile do
       main: "readme",
       extras: [
         "README.md",
-        "guides/upgrading/0.1.x_to_0.2.x.md": [title: "0.1.x to 0.2.x"],
-        "guides/upgrading/0.2.x_to_0.3.x.md": [title: "0.2.x to 0.3.x"]
+        "guides/upgrading/0.3.x_to_0.4.x.md": [title: "0.3.x to 0.4.x"],
+        "guides/upgrading/0.2.x_to_0.3.x.md": [title: "0.2.x to 0.3.x"],
+        "guides/upgrading/0.1.x_to_0.2.x.md": [title: "0.1.x to 0.2.x"]
       ],
       extra_section: "GUIDES",
       groups_for_extras: [

mix.lock

@@ -14,6 +14,7 @@
   "idna": {:hex, :idna, "5.1.0", "d72b4effeb324ad5da3cab1767cb16b17939004e789d8c0ad5b70f3cea20c89a", [], [{:unicode_util_compat, "0.3.1", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm"},
   "jason": {:hex, :jason, "1.0.0-rc.1", "c8421d4e6e6ef0dd7c2b64ff63589f8561116808fa003dddfd5360cde7bb4625", [], [{:decimal, "~> 1.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm"},
   "jsx": {:hex, :jsx, "2.8.3", "a05252d381885240744d955fbe3cf810504eb2567164824e19303ea59eef62cf", [], [], "hexpm"},
+  "maybe": {:hex, :maybe, "1.0.0", "65311dd7e16659579116666b268d03d7e1d1b3da8776c81a6b199de7177b43d6", [:mix], [], "hexpm"},
   "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [], [], "hexpm"},
   "mimerl": {:hex, :mimerl, "1.0.2", "993f9b0e084083405ed8252b99460c4f0563e41729ab42d9074fd5e52439be88", [], [], "hexpm"},
   "poison": {:hex, :poison, "3.1.0", "d9eb636610e096f86f25d9a46f35a9facac35609a7591b3be3326e99a0484665", [:mix], [], "hexpm"},