Use aliases to enable zero down time when re-indexing (#433)

* improve error messages on log

* different datafile payloads of update and create

* Use aliases to enable zero down time when re-indexing

Refactoring

Namespace legacy syncing services

* Split Reindex Service into separate concerns. Add tests

Update logging

Add test for IndexDeleter

* Fix code climate issues and disable a couple of checks:
- Missing top level class documentation comment
- Missing frozen string literal comment

* Improve error handling and readability for AliasUpdater and IndexDeleter

* Use 'drop' method instead of 'slice'

Author: TobyRet

Diff for: .codeclimate.yml

@@ -1,6 +1,11 @@
 engines:
   rubocop:
     enabled: true
+    checks:
+      Rubocop/Style/Documentation:
+        enabled: false
+      Rubocop/Style/FrozenStringLiteralComment:
+        enabled: false
   eslint:
     enabled: true
   csslint:

Diff for: app/models/dataset.rb

@@ -158,7 +158,7 @@ def timeseries?
   private
 
   def sync_with_legacy
-    LegacySyncService.new(self).sync
+    Legacy::LegacyToBetaSyncService.new(self).sync
   end
 
   def send_to_search_index

Diff for: app/models/legacy/datafile.rb

@@ -1,37 +1,40 @@
 class Legacy::Datafile < SimpleDelegator
 
   def update_payload
-    { "id" => uuid,
-      "description" => name,
-      "format" => format,
-      "date" => build_date,
-      "resource_type" => build_datafile_type,
-      "url" => url,
-      "created" => created_at
+    {
+      'id' => uuid,
+      'description' => name,
+      'format' => format,
+      'date' => build_date,
+      'resource_type' => build_datafile_type,
+      'url' => url,
+      'created' => created_at
     }.compact.to_json
   end
 
   def create_payload
-    { "package_id" => dataset.ckan_uuid,
-      "url" => url,
-      "description" => name,
-      "format" => format,
-      "name" => name,
-      "resource_type" => build_datafile_type,
-      "size" => size,
-      "created" => created_at
+    {
+      'package_id' => dataset.ckan_uuid,
+      'url' => url,
+      'description' => name,
+      'format' => format,
+      'name' => name,
+      'resource_type' => build_datafile_type,
+      'size' => size,
+      'created' => created_at
+
     }.compact.to_json
   end
 
   private
 
   def build_date
-    return "" unless dataset.timeseries?
-    end_date.presence.strftime("%d/%m/%Y")
+    return '' unless dataset.timeseries?
+    end_date.presence.strftime('%d/%m/%Y')
   end
 
   def build_datafile_type
-    return "" if type.blank?
-    type == "Doc" ? "documentation" : "file"
+    return '' if type.blank?
+    type == 'Doc' ? 'documentation' : 'file'
   end
 end

Diff for: app/services/alias_updater_service.rb

@@ -0,0 +1,52 @@
+class AliasUpdaterService
+  def initialize(args)
+    @new_index_name = args[:new_index_name]
+    @index_alias = args[:index_alias]
+    @client = args[:client]
+    @logger = args[:logger]
+  end
+
+  def run
+    remove_alias_from_old_index
+    assign_alias_to_new_index
+    logger.info "Alias '#{index_alias}' now pointing to '#{new_index_name}'"
+  end
+
+  private
+
+  attr_reader :logger, :client, :index_alias, :new_index_name
+
+  def remove_alias_from_old_index
+    client.indices.update_aliases body: {
+      actions: [
+        {
+          remove: {
+            index: Dataset.index_name,
+            alias: index_alias
+          }
+        }
+      ]
+    }
+  rescue => e
+    msg = "Could not remove alias.\n #{e.message}"
+    logger.error msg
+    Raven.capture_error msg
+  end
+
+  def assign_alias_to_new_index
+    client.indices.update_aliases body: {
+      actions: [
+        {
+          add: {
+            index: new_index_name,
+            alias: index_alias
+          }
+        }
+      ]
+    }
+  rescue => e
+    msg = "Could not update alias.\n #{e.message}"
+    logger.error msg
+    Raven.capture_error msg
+  end
+end

Diff for: app/services/datasets_indexer_service.rb

@@ -0,0 +1,92 @@
+class DatasetsIndexerService
+  def initialize(args)
+    @batch_size = args[:batch_size]
+    @date = args[:date]
+    @new_index_name = args[:new_index_name]
+    @client = args[:client]
+    @logger = args[:logger]
+  end
+
+  def run
+    number_datasets_processed = 0
+
+    create_new_index
+
+    Dataset.published.find_in_batches(batch_size: batch_size) do |datasets|
+      logger.info "Batching #{datasets.length} datasets"
+      bulk_index(datasets)
+      number_datasets_processed += batch_size
+    end
+
+    logger.info "Datasets indexed to #{new_index_name}"
+  end
+
+  private
+
+  attr_reader :date, :new_index_name, :batch_size, :client, :logger
+
+  def create_new_index
+    client.indices.create(
+      index: new_index_name,
+      body: { mappings: index_mapping }
+    )
+  end
+
+  def bulk_index(datasets)
+    client.bulk(
+      index: new_index_name,
+      type: ::Dataset.__elasticsearch__.document_type,
+      body: prepare_records(datasets)
+    )
+  rescue => e
+    msg = "There was an error indexing datasets:\n#{e.message}"
+    logger.error msg
+    Raven.capture msg
+  end
+
+  def index_mapping
+    {
+      dataset: {
+        properties: {
+          name: {
+            type: 'string',
+            index: 'not_analyzed'
+          },
+          uuid: {
+            type: 'string',
+            index: 'not_analyzed'
+          },
+          location1: {
+            type: 'string',
+            fields: {
+              raw: {
+                type: 'string',
+                index: 'not_analyzed'
+              }
+            }
+          },
+          organisation: {
+            type: 'nested',
+            properties: {
+              title: {
+                type: 'string',
+                fields: {
+                  raw: {
+                    type: 'string',
+                    index: 'not_analyzed'
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+  end
+
+  def prepare_records(datasets)
+    datasets.map do |dataset|
+      { index: { _id: dataset.id, data: dataset.as_indexed_json } }
+    end
+  end
+end

Diff for: app/services/index_deletion_service.rb

@@ -0,0 +1,36 @@
+class IndexDeletionService
+  def initialize(args)
+    @index_alias = args[:index_alias]
+    @client = args[:client]
+    @logger = args[:logger]
+  end
+
+  def run
+    indexes = client.indices.get_aliases.keys
+    indexes_to_be_deleted = select_indexes_for_deletion(indexes)
+    delete(indexes_to_be_deleted)
+  rescue => e
+    msg = "Failed to delete old indexes.\n#{e.message}"
+    logger.error msg
+    Raven.capture_error msg
+  end
+
+  private
+
+  attr_reader :client, :index_alias, :logger
+
+  def select_indexes_for_deletion(indexes)
+    # Ensure that the three most recent indexes are not deleted
+    indexes
+      .select { |index_name| index_name.include? index_alias }
+      .sort_by { |index_name| Time.parse(index_name.gsub(/"#{index_alias}_"/, '')) }
+      .drop(3)
+  end
+
+  def delete(indexes)
+    indexes.each do |index|
+      client.indices.delete index: index
+      logger.info "Deleted #{index}"
+    end
+  end
+end

Diff for: app/services/beta_sync_service.rb renamed to app/services/legacy/beta_to_legacy_sync_service.rb

@@ -1,6 +1,6 @@
 require 'util/metadata_tools'
 
-class BetaSyncService
+class Legacy::BetaToLegacySyncService
   ENDPOINTS = {
     modified_datasets: 'api/3/action/package_search?q=metadata_modified:[NOW-1DAY%20TO%20NOW]&rows=5000'.freeze,
     new_datasets: 'api/3/action/package_search?q=metadata_created:[NOW-1DAY%20TO%20NOW]&rows=5000'.freeze
@@ -36,10 +36,10 @@ def new_datasets
 
   def import(dataset)
     begin
-      @logger.info "Attempting to save legacy dataset to postgres and elasticsearch - legacy_id: #{dataset["id"]}"
+      @logger.info "Attempting to save legacy dataset to postgres and elasticsearch - legacy_id: #{dataset['id']}"
       MetadataTools.persist(dataset, @orgs_cache, @theme_cache)
       MetadataTools.index(dataset)
-      @logger.info "Legacy dataset saved - legacy_id: #{dataset["id"]}"
+      @logger.info "Legacy dataset saved - legacy_id: #{dataset['id']}"
     rescue => e
       Raven.capture_exception e.message
     end

Diff for: app/services/legacy_sync_service.rb renamed to app/services/legacy/legacy_to_beta_sync_service.rb

@@ -1,4 +1,4 @@
-class LegacySyncService
+class Legacy::LegacyToBetaSyncService
   attr_reader :dataset
 
   def initialize(dataset)

Diff for: app/services/reindex_service.rb

@@ -0,0 +1,24 @@
+class ReindexService
+  attr_reader :logger, :index_alias, :indexer, :alias_updater, :index_deleter
+
+  def initialize(args)
+    @indexer = args[:indexer]
+    @alias_updater = args[:alias_updater]
+    @index_deleter = args[:index_deleter]
+    @logger = args[:logger]
+  end
+
+  def run
+    logger.info "Indexing #{published_datasets_count} datasets"
+    indexer.run
+    alias_updater.run
+    index_deleter.run
+    logger.info 'Reindexing complete!'
+  end
+
+  private
+
+  def published_datasets_count
+    Dataset.published.count
+  end
+end