feat: operators can prefix elasticsearch indexes for multi-tenancy

api/search.rb

@@ -47,7 +47,7 @@ def get_thread_ids(context, group_ids, local_params, search_text)
     }
   }
 
-  response = Elasticsearch::Model.client.search(index: TaskHelpers::ElasticsearchHelper::INDEX_NAMES, body: body)
+  response = Elasticsearch::Model.client.search(index: TaskHelpers::ElasticsearchHelper::prefixed_index_names, body: body)
 
   thread_ids = Set.new
   response['hits']['hits'].each do |hit|

app.rb

@@ -67,7 +67,7 @@ def get_logger(progname, threshold=nil)
 # NOTE: You can also add a logger, but it will log some FATAL warning during index creation.
 # Example: Elasticsearch::Client.new(logger: get_logger('elasticsearch', Logger::WARN))
 Elasticsearch::Model.client = Elasticsearch::Client.new(
-    host: CommentService.config[:elasticsearch_server],
+    url: CommentService.config[:elasticsearch_server],
     log: false
 )
 

config/application.yml

@@ -7,3 +7,4 @@ default_locale: <%= ENV['SERVICE_LANGUAGE'] || 'en-US' %>
 manual_pagination_batch_size: <%= ENV['MANUAL_PAGINATION_BATCH_SIZE'] || 500 %>
 thread_response_default_size: <%= ENV['THREAD_RESPONSE_DEFAULT_SIZE'] || 100 %>
 thread_response_size_limit: <%= ENV['THREAD_RESPONSE_SIZE_LIMIT'] || 200 %>
+elasticsearch_index_prefix: <%= ENV['ELASTICSEARCH_INDEX_PREFIX'] || "" %>

lib/task_helpers.rb

@@ -10,6 +10,15 @@ module ElasticsearchHelper
     # local variable which store actual indices for future deletion
     @@temporary_index_names = []
 
+    def self.prefix_index(index_name)
+      prefix = CommentService.config[:elasticsearch_index_prefix]
+      "#{prefix}#{index_name}"
+    end
+
+    def self.prefixed_index_names
+      INDEX_NAMES.map{|n| self.prefix_index(n) }
+    end
+
     def self.temporary_index_names
       @@temporary_index_names
     end
@@ -47,7 +56,7 @@ def self.rebuild_indices(batch_size=500, extra_catchup_minutes=5)
       index_names.each do |index_name|
         current_batch = 1
         model = get_index_model_rel(index_name)
-        model_index_name = model.index_name
+        model_index_name = self.prefix_index(model.index_name)
         alias_names.push(model_index_name)
         move_alias(model_index_name, index_name, force_delete: true)
       end
@@ -87,7 +96,7 @@ def self.create_indices
       time_now = Time.now.strftime('%Y%m%d%H%M%S%L')
 
       INDEX_MODELS.each do |model|
-        index_name = "#{model.index_name}_#{time_now}"
+        index_name = "#{self.prefix_index(model.index_name)}_#{time_now}"
         index_names.push(index_name)
         Elasticsearch::Model.client.indices.create(
           index: index_name,
@@ -186,19 +195,19 @@ def self.move_alias(alias_name, index_name, force_delete=false)
 
     def self.refresh_indices
       if temporary_index_names.length > 0
-        Elasticsearch::Model.client.indices.refresh(index: INDEX_NAMES)
+        Elasticsearch::Model.client.indices.refresh(index: self.prefixed_index_names)
       else
         fail "No indices to refresh"
       end
     end
 
     def self.initialize_indices(force_new_index = false)
       # When force_new_index is true, fresh indices will be created even if it already exists.
-      if force_new_index or not exists_aliases(INDEX_NAMES)
+      if force_new_index or not exists_aliases(self.prefixed_index_names)
         index_names = create_indices
         index_names.each do |index_name|
           model = get_index_model_rel(index_name)
-          move_alias(model.index_name, index_name, force_delete: true)
+          move_alias(self.prefix_index(model.index_name), index_name, force_delete: true)
         end
         add_temporary_index_names(index_names)
       else
@@ -211,7 +220,7 @@ def self.initialize_indices(force_new_index = false)
     # Validates that each index includes the proper mappings.
     # There is no return value, but an exception is raised if the index is invalid.
     def self.validate_indices
-      actual_mappings = Elasticsearch::Model.client.indices.get_mapping(index: INDEX_NAMES)
+      actual_mappings = Elasticsearch::Model.client.indices.get_mapping(index: self.prefixed_index_names)
 
       if actual_mappings.length == 0
         fail "Indices are not exist!"
@@ -243,11 +252,11 @@ def self.validate_indices
           end
         end
         if missing_fields.any? or invalid_field_types.any?
-          fail "Index '#{model.index_name}' has missing or invalid field mappings.  Missing fields: #{missing_fields}. Invalid types: #{invalid_field_types}."
+          fail "Index '#{self.prefix_index(model.index_name)}' has missing or invalid field mappings.  Missing fields: #{missing_fields}. Invalid types: #{invalid_field_types}."
         end
 
         # Check that expected field mappings of the correct type exist
-        LOG.info "Passed: Index '#{model.index_name}' exists with up-to-date mappings."
+        LOG.info "Passed: Index '#{self.prefix_index(model.index_name)}' exists with up-to-date mappings."
       end
 
     end

spec/lib/task_helpers_spec.rb

@@ -47,7 +47,7 @@ def assert_alias_points_to_index(alias_name, index_name)
         create(:comment_thread, body: 'the best test body', course_id: 'test_course_id')
         TaskHelpers::ElasticsearchHelper.refresh_indices
         expect(Elasticsearch::Model.client.search(
-            index: TaskHelpers::ElasticsearchHelper::INDEX_NAMES
+            index: TaskHelpers::ElasticsearchHelper::prefixed_index_names
         )['hits']['total']['value']).to be > 0
       end
 

spec/lib/tasks/search_rake_spec.rb

@@ -30,7 +30,7 @@
 
 describe "search:catchup" do
   include_context "rake"
-  let(:indices) { TaskHelpers::ElasticsearchHelper::INDEX_NAMES }
+  let(:indices) { TaskHelpers::ElasticsearchHelper::prefixed_index_names }
   let(:comments_index_name) { Comment.index_name }
   let(:comment_threads_index_name) { CommentThread.index_name }