Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Amsterdam Elasticsearch meetup: Cross Cluster S...

Amsterdam Elasticsearch meetup: Cross Cluster Search with Elasticsearch

Luca Cavanna

May 11, 2017
Tweet

More Decks by Luca Cavanna

Other Decks in Programming

Transcript

  1. Agenda 2 Why Cross Cluster Search? How does it work?

    Follow-up search improvements 1 2 3 4 How does it compare to Tribe node?
  2. 4

  3. 5

  4. cluster Search api 8 Client node1 logs 2P posts 1P

    node2 logs 3P users 1P node3 logs 1P posts 2P GET /posts/_search posts index: 2 primaries logs index: 3 primaries users index: 1 primary
  5. cluster Search request parsing 9 Client node1 logs 2P posts

    1P node2 logs 3P users 1P node3 logs 1P posts 2P The coordinating node parses the request GET /posts/_search
  6. cluster Query phase 10 Client node1 logs 2P posts 1P

    node2 logs 3P users 1P node3 logs 1P posts 2P GET /posts/_search The query gets executed on the relevant shards The query gets executed on the relevant shards
  7. cluster Reduce phase 11 Client node1 logs 2P posts 1P

    node2 logs 3P users 1P node3 logs 1P posts 2P GET /posts/_search The coordinating node receives "size" hits per shard and performs reduction
  8. cluster Fetch phase 12 Client node1 logs 2P posts 1P

    node2 logs 3P users 1P node3 logs 1P posts 2P GET /posts/_search The coordinating node fetches the top hits from the relevant shards The coordinating node fetches the top hits from the relevant shards
  9. cluster Search response 13 Client node1 logs 2P posts 1P

    node2 logs 3P users 1P node3 logs 1P posts 2P The coordinating node returns the top hits back to the client
  10. 15 europe eu1 posts 1P users 1P eu2 posts 3P

    posts 2P usa us1 posts 2P posts 1P australia au1 posts 1P au2 logs 1P
  11. Register remote clusters PUT /_cluster/settings { "persistent" : { "search.remote"

    : { "australia" : { "seeds": ["host_au1:9300", "host_au2:9300"] }, "usa" : { "seeds": "host_us1:9300" } } } } 16
  12. 17 Client europe eu1 posts 1P users 1P eu2 posts

    3P posts 2P usa us1 posts 2P posts 1P australia au1 posts 1P au2 logs 1P GET /posts,usa:posts,australia:posts/_search The coordinating node parses the request
  13. 18 Client europe eu1 posts 1P users 1P eu2 posts

    3P posts 2P usa us1 posts 2P posts 1P australia The coordinating node fetches info about remote indices and their shards au2 logs 1P au1 posts 1P GET /posts,usa:posts,australia:posts/_search
  14. 19 Client europe eu1 posts 1P users 1P eu2 posts

    3P posts 2P usa us1 posts 2P posts 1P australia posts 1P The query gets executed on the relevant local shards au2 logs 1P au1 posts 1P GET /posts,usa:posts,australia:posts/_search
  15. 20 Client europe eu1 posts 1P users 1P eu2 posts

    3P posts 2P usa us1 posts 2P posts 1P australia The query gets also executed on the relevant remote shards au2 logs 1P au1 posts 1P GET /posts,usa:posts,australia:posts/_search
  16. 21 Client europe eu1 posts 1P users 1P eu2 posts

    3P posts 2P usa us1 posts 2P posts 1P australia au2 logs 1P au1 posts 1P The coordinating node receives "size" hits per shard and performs reduction GET /posts,usa:posts,australia:posts/_search
  17. 22 Client europe eu1 posts 1P users 1P eu2 posts

    3P posts 2P usa us1 posts 2P posts 1P australia The coordinating node fetches the top hits from the relevant shards au2 logs 1P au1 posts 1P GET /posts,usa:posts,australia:posts/_search
  18. 23 Client europe eu1 posts 1P users 1P eu2 posts

    3P posts 2P usa us1 posts 2P posts 1P australia The coordinating node returns the top hits back to the client au2 logs 1P au1 posts 1P GET /posts,usa:posts,australia:posts/_search
  19. Search response "hits" : [ { "_index" : "australia:posts", ...

    }, { "_index" : "posts", ... }, { "_index" : "usa:posts", ... } ] 24
  20. 26 europe eu1 posts 1P PUT /_cluster/settings { "persistent" :

    { "search.remote" : { "australia" : { "seeds": ["host_au1", "host_au3"] } } } } australia au1 posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P eu2 posts 2P
  21. europe 27 eu1 posts 1P australia au1 posts 1P au2

    logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P eu2 posts 2P Fetch the cluster state and select remote nodes PUT /_cluster/settings { "persistent" : { "search.remote" : { "australia" : { "seeds": ["host_au1", "host_au3"] } } } }
  22. Disable cross cluster calls from specific nodes cat config/elasticsearch.yml #

    ---------------- Cluster ---------------- # # Use a descriptive name for your cluster: # cluster.name: europe # # ---------------- Node -------------------- # # Use a descriptive name for the node: # node.name: eu2 # Disable connections to remote clusters from this node search.remote.connect: false 28
  23. europe 29 eu1 posts 1P eu2 posts 2P Fetch the

    cluster state and select remote nodes PUT /_cluster/settings { "persistent" : { "search.remote" : { "australia" : { "seeds": ["host_au1", "host_au3"] } } } } australia au1 posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P
  24. europe 30 eu1 posts 1P eu2 posts 2P Fetch the

    cluster state and select remote nodes PUT /_cluster/settings { "persistent" : { "search.remote" : { "australia" : { "seeds": ["host_au1", "host_au3"] } } } } australia au1 posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P
  25. europe 31 eu1 posts 1P eu2 posts 2P Fetch the

    cluster state and select remote nodes PUT /_cluster/settings { "persistent" : { "search.remote" : { "australia" : { "seeds": ["host_au1", "host_au3"] } } } } australia au1 posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P
  26. Maximum number of remote nodes to connect to cat config/elasticsearch.yml

    # ---------------- Cluster ---------------- # # Use a descriptive name for your cluster: # cluster.name: europe # # ---------------- Node -------------------- # # Use a descriptive name for the node: # node.name: eu1 # Max number of remote nodes to connect to search.remote.connections_per_cluster: 3 32
  27. europe 33 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P Client GET /australia:posts/_search
  28. europe 34 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P Client GET /australia:posts/_search The coordinating node fetches info about remote indices and their shards
  29. europe 35 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P Client GET /australia:posts/_search The query gets executed on the relevant remote shards
  30. europe 36 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P Client GET /australia:posts/_search The coordinating node fetches the top hits from the relevant shards
  31. Pick node attribute to tag remote nodes with cat config/elasticsearch.yml

    # ---------------- Cluster ---------------- # # Use a descriptive name for your cluster: # cluster.name: europe # # ---------------- Node -------------------- # # Use a descriptive name for the node: # node.name: eu1 # Attribute that enables/disables remote connections search.remote.node.attr: gateway 37
  32. cat config/elasticsearch.yml # ---------------- Cluster ---------------- # # Use a

    descriptive name for your cluster: # cluster.name: australia # # ---------------- Node -------------------- # # Use a descriptive name for the node: # node.name: au5 # # Add custom attributes to the node: # node.attr.gateway: true 38 Tag selected nodes as remote gateways (2)
  33. europe 39 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P gateway PUT /_cluster/settings { "persistent" : { "search.remote" : { "australia" : { "seeds": ["host_au1", "host_au3"] } } } } Fetch the cluster state and select remote nodes
  34. europe 40 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P Client GET /australia:posts/_search gateway
  35. europe 41 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P Client GET /australia:posts/_search gateway The coordinating node fetches info about remote indices and their shards
  36. europe 42 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P Client GET /australia:posts/_search The query gets executed on the relevant remote shards gateway
  37. europe 43 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P Client GET /australia:posts/_search gateway The coordinating node fetches the top hits from the relevant shards
  38. europe 45 eu1 posts 1P eu2 posts 2P Remote nodes

    selection: - is node available? - is node gateway? - is version compatible? PUT /_cluster/settings { "persistent" : { "search.remote" : { "australia" : { "seeds": ["host_au1", "host_au3"] } } } } australia au1 posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P
  39. ‹#› Upgrading to 6.0 with Rolling Restarts The hardest part

    about upgrading to a new major version has been the fact that you have to do a full cluster restart to get there. No more! You will be able to upgrade from the latest 5.x version to 6.0 using rolling restarts, without any cluster downtime. https://www.elastic.co/blog/elasticsearch-6-0-0-alpha1-released
  40. europe 47 eu1 posts 1P eu2 posts 2P 6.0 won't

    be able to talk directly to 5.4, only to 5.latest PUT /_cluster/settings { "persistent" : { "search.remote" : { "australia" : { "seeds": ["host_au1", "host_au3"] } } } } australia au1 posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P 5.4 5.4 5.4 5.4 5.4 5.4 6.0 6.0
  41. europe 48 eu1 posts 1P eu2 posts 2P PUT /_cluster/settings

    { "persistent" : { "search.remote" : { "australia" : { "seeds": "host_au7" } } } } australia au1 posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P 5.4 5.4 5.4 5.4 5.4 5.4 6.0 au7 5.latest Fetch the cluster state and select remote nodes 6.0
  42. europe 49 eu1 posts 1P eu2 posts 2P PUT /_cluster/settings

    { "persistent" : { "search.remote" : { "australia" : { "seeds": "host_au7" } } } } australia au1 posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P 5.4 5.4 5.4 5.4 5.4 5.4 au7 5.latest Fetch the cluster state and select remote nodes 6.0 6.0
  43. europe 50 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P 5.4 5.4 5.4 5.4 5.4 5.4 au7 5.latest 6.0 6.0 Client GET /australia:posts/_search
  44. europe 51 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P 5.4 5.4 5.4 5.4 5.4 5.4 au7 5.latest 6.0 6.0 Client GET /australia:posts/_search The coordinating node fetches info about remote indices and their shards
  45. europe 52 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P 5.4 5.4 5.4 5.4 5.4 5.4 au7 5.latest 6.0 6.0 Client GET /australia:posts/_search The query gets executed on the relevant remote shards
  46. europe 53 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 logs 1P au3 posts 2P au4 logs 2P au6 posts 3P au5 logs 3P 5.4 5.4 5.4 5.4 5.4 5.4 au7 5.latest 6.0 6.0 Client GET /australia:posts/_search The coordinating node fetches the top hits from the relevant shards
  47. europe 56 eu1 posts 1P eu2 posts 2P australia au1

    posts 1P au2 posts 4P au3 posts 2P au4 posts 5P au6 posts 6P au5 posts 3P Client GET /australia:posts/_search gateway The coordinating node receives "size" hits per shard and performs reduction
  48. Shard count limit (5.3) PUT /_cluster/settings { "persistent" : {

    "action.search.shard_count.limit" : 1000 } } 57
  49. Multiple reduction phases (5.4) { "took" : 43, "timed_out" :

    false, "num_reduce_phases": 6, "_shards" : { "total" : 3000, "successful" : 3000, "failed" : 0 }, "hits" : { "total" : 3987245, "max_score" : 1.0, "hits" : […] } } 59
  50. Shard count limit (5.4) PUT /_cluster/settings { "persistent" : {

    "action.search.shard_count.limit" : Long.MAX_VALUE } } 60
  51. • Adding remote clusters requires node restart 62 Tribe node

    Cross Cluster Search • Remote clusters can be dynamically registered
  52. • Adding remote clusters requires node restart • Receives all

    cluster state updates from remote clusters 63 Tribe node Cross Cluster Search • Remote clusters can be dynamically registered • Retrieves info on demand from remote clusters
  53. • Adding remote clusters requires node restart • Receives all

    cluster state updates from remote clusters • Doesn't support indices with same names on different clusters 64 Tribe node Cross Cluster Search • Remote clusters can be dynamically registered • Retrieves info on demand from remote clusters • No limitations on indices naming
  54. • Adding remote clusters requires node restart • Receives all

    cluster state updates from remote clusters • Doesn't support indices with same names on different clusters • Requires an additional node (tribe) to join all the remote clusters 65 Tribe node Cross Cluster Search • Remote clusters can be dynamically registered • Retrieves info on demand from remote clusters • No limitations on indices naming • No additional nodes required
  55. • Bi-directional connections to every remote node 66 Tribe node

    Cross Cluster Search • Uni-directional connections to selected gateway nodes
  56. • Bi-directional connections to every remote node • Specific type

    of node with its own handling and edge cases, hard to test and maintain 67 Tribe node Cross Cluster Search • Uni-directional connections to selected gateway nodes • Isolated component integrated into a specific API: easier to test and maintain
  57. • Bi-directional connections to every remote node • Specific type

    of node with its own handling and edge cases, hard to test and maintain • Works with almost every api 68 Tribe node Cross Cluster Search • Uni-directional connections to selected gateway nodes • Isolated component integrated into a specific API: easier to test and maintain • Specific to _search api