From b10668a4548c2b1ffd5f6e29d4a8f72ad025db09 Mon Sep 17 00:00:00 2001 From: Molly Struve Date: Wed, 15 Apr 2020 09:08:59 -0500 Subject: [PATCH] [deploy] Create a Bulk Index Elasticsearch Worker (#7265) --- app/models/article.rb | 2 + app/models/chat_channel_membership.rb | 2 + app/models/comment.rb | 2 + app/models/podcast.rb | 1 + app/models/reaction.rb | 1 + app/models/tag.rb | 2 + app/models/user.rb | 1 + app/services/search/base.rb | 40 +++++++++++++++++++ .../bulk_index_to_elasticsearch_worker.rb | 16 ++++++++ spec/services/search/base_spec.rb | 24 +++++++++++ ...bulk_index_to_elasticsearch_worker_spec.rb | 22 ++++++++++ 11 files changed, 113 insertions(+) create mode 100644 app/workers/search/bulk_index_to_elasticsearch_worker.rb create mode 100644 spec/workers/search/bulk_index_to_elasticsearch_worker_spec.rb diff --git a/app/models/article.rb b/app/models/article.rb index c85884115..44dbd9ab9 100644 --- a/app/models/article.rb +++ b/app/models/article.rb @@ -172,6 +172,8 @@ class Article < ApplicationRecord scope :with_video, -> { published.where.not(video: [nil, ""], video_thumbnail_url: [nil, ""]).where("score > ?", -4) } + scope :eager_load_serialized_data, -> { includes(:user, :organization, :tags) } + algoliasearch per_environment: true, auto_remove: false, enqueue: :trigger_index do attribute :title add_index "searchables", id: :index_id, per_environment: true, enqueue: :trigger_index do diff --git a/app/models/chat_channel_membership.rb b/app/models/chat_channel_membership.rb index 56e4c4cd7..6cd73b08d 100644 --- a/app/models/chat_channel_membership.rb +++ b/app/models/chat_channel_membership.rb @@ -19,6 +19,8 @@ class ChatChannelMembership < ApplicationRecord delegate :channel_type, to: :chat_channel + scope :eager_load_serialized_data, -> { includes(:user, :channel) } + def channel_last_message_at chat_channel.last_message_at end diff --git a/app/models/comment.rb b/app/models/comment.rb index eaf5e1e39..f7a807463 100644 --- a/app/models/comment.rb +++ b/app/models/comment.rb @@ -48,6 +48,8 @@ class Comment < ApplicationRecord before_validation :evaluate_markdown, if: -> { body_markdown } validate :permissions, if: :commentable + scope :eager_load_serialized_data, -> { includes(:user, :commentable) } + alias touch_by_reaction save def self.tree_for(commentable, limit = 0) diff --git a/app/models/podcast.rb b/app/models/podcast.rb index 3d52f9fa2..9062e6b6e 100644 --- a/app/models/podcast.rb +++ b/app/models/podcast.rb @@ -22,6 +22,7 @@ class Podcast < ApplicationRecord scope :reachable, -> { where(id: PodcastEpisode.reachable.select(:podcast_id)) } scope :published, -> { where(published: true) } scope :available, -> { reachable.published } + scope :eager_load_serialized_data, -> { includes(:user, :podcast, :tags) } alias_attribute :path, :slug alias_attribute :profile_image_url, :image_url diff --git a/app/models/reaction.rb b/app/models/reaction.rb index 513a29e08..ff1d0daa1 100644 --- a/app/models/reaction.rb +++ b/app/models/reaction.rb @@ -20,6 +20,7 @@ class Reaction < ApplicationRecord scope :positive, -> { where("points > ?", 0) } scope :readinglist, -> { where(category: "readinglist") } + scope :eager_load_serialized_data, -> { includes(:reactable, :user) } validates :category, inclusion: { in: CATEGORIES } validates :reactable_type, inclusion: { in: REACTABLE_TYPES } diff --git a/app/models/tag.rb b/app/models/tag.rb index 83c10a67e..3980a37ad 100644 --- a/app/models/tag.rb +++ b/app/models/tag.rb @@ -36,6 +36,8 @@ class Tag < ActsAsTaggableOn::Tag after_commit :sync_related_elasticsearch_docs, on: [:update] after_commit :remove_from_elasticsearch, on: [:destroy] + scope :eager_load_serialized_data, -> {} + include Searchable SEARCH_SERIALIZER = Search::TagSerializer SEARCH_CLASS = Search::Tag diff --git a/app/models/user.rb b/app/models/user.rb index 57afaf9c8..6cb49ed2f 100644 --- a/app/models/user.rb +++ b/app/models/user.rb @@ -159,6 +159,7 @@ class User < ApplicationRecord scope :top_commenters, lambda { |number = 10| includes(:counters).order(Arel.sql("user_counters.data -> 'comments_these_7_days' DESC")).limit(number) } + scope :eager_load_serialized_data, -> { includes(:roles) } after_save :bust_cache after_save :subscribe_to_mailchimp_newsletter diff --git a/app/services/search/base.rb b/app/services/search/base.rb index 114f23a0a..6729cdd23 100644 --- a/app/services/search/base.rb +++ b/app/services/search/base.rb @@ -9,6 +9,22 @@ module Search ) end + def bulk_index(data_hashes) + indexing_hashes = data_hashes.map do |data_hash| + model_hash = data_hash.with_indifferent_access + model_hash[:last_indexed_at] = Time.current + { + index: { + _index: self::INDEX_ALIAS, + _id: model_hash[:id], + data: model_hash + } + } + end + + process_hashes(indexing_hashes) + end + def find_document(doc_id) Search::Client.get(id: doc_id, index: self::INDEX_ALIAS) end @@ -80,6 +96,30 @@ module Search def index_settings raise "Search classes must implement their own index settings" end + + def process_hashes(indexing_hashes) + indexing_hashes.in_groups_of(1000, false).flat_map do |hashes| + indexing_chunks(hashes).select(&:any?).map { |chunk| Search::Client.bulk(body: chunk) } + end + end + + def indexing_chunks(hashes) + return [] unless hashes.any? + return to_enum(__method__, hashes) unless block_given? + + size = 0 + chunk = [] + hashes.each do |hash| + chunk << hash + size += hash.to_s.bytesize + next unless size > 5.megabytes + + yield chunk + size = 0 + chunk = [] + end + yield chunk + end end end end diff --git a/app/workers/search/bulk_index_to_elasticsearch_worker.rb b/app/workers/search/bulk_index_to_elasticsearch_worker.rb new file mode 100644 index 000000000..c7f21d4af --- /dev/null +++ b/app/workers/search/bulk_index_to_elasticsearch_worker.rb @@ -0,0 +1,16 @@ +module Search + class BulkIndexToElasticsearchWorker + include Sidekiq::Worker + + sidekiq_options queue: :high_priority, lock: :until_executing + + def perform(object_class, ids) + data_hashes = object_class.constantize. + eager_load_serialized_data. + where(id: ids). + find_each.map(&:serialized_search_hash) + + "Search::#{object_class}".constantize.bulk_index(data_hashes) + end + end +end diff --git a/spec/services/search/base_spec.rb b/spec/services/search/base_spec.rb index 88e432021..c0d8531c6 100644 --- a/spec/services/search/base_spec.rb +++ b/spec/services/search/base_spec.rb @@ -27,6 +27,30 @@ RSpec.describe Search::Base, type: :service, elasticsearch: true do end end + describe "::bulk_index" do + it "indexes a set of data hashes to Elasticsearch" do + id_list = [123, 456, 789] + id_list.each do |document_id| + expect { described_class.find_document(document_id) }.to raise_error(Search::Errors::Transport::NotFound) + end + data_hashes = id_list.map { |id| { id: id, name: "i_am_a_tag" } } + described_class.bulk_index(data_hashes) + + id_list.each do |document_id| + doc = described_class.find_document(document_id) + expect(doc.dig("_source", "id")).to eql(document_id) + end + end + + it "sets last_indexed_at field" do + Timecop.freeze(Time.current) do + described_class.bulk_index([{ id: document_id, name: "i_am_a_tag" }]) + last_indexed_at = described_class.find_document(document_id).dig("_source", "last_indexed_at") + expect(Time.zone.parse(last_indexed_at).to_i).to eq(Time.current.to_i) + end + end + end + describe "::find_document" do it "fetches a document for a given ID from elasticsearch" do described_class.index(document_id, id: document_id) diff --git a/spec/workers/search/bulk_index_to_elasticsearch_worker_spec.rb b/spec/workers/search/bulk_index_to_elasticsearch_worker_spec.rb new file mode 100644 index 000000000..44fddd7e3 --- /dev/null +++ b/spec/workers/search/bulk_index_to_elasticsearch_worker_spec.rb @@ -0,0 +1,22 @@ +require "rails_helper" + +RSpec.describe Search::BulkIndexToElasticsearchWorker, type: :worker, elasticsearch: true do + let(:worker) { subject } + let(:article) { create(:article) } + + include_examples "#enqueues_on_correct_queue", "high_priority", ["Reaction", 1] + + it "indexes documents for a set of given ids and object class" do + reactions = [create(:reaction, reactable: article), create(:reaction), create(:reaction)] + Sidekiq::Worker.clear_all + + reactions.each do |reaction| + expect { reaction.elasticsearch_doc }.to raise_error(Search::Errors::Transport::NotFound) + end + worker.perform("Reaction", reactions.map(&:id)) + + reactions.each do |reaction| + expect(reaction.elasticsearch_doc.dig("_source", "id")).to eql(reaction.id) + end + end +end