Make the opengeometadata indexer more robust

- Gracefully handle indexing failures
- Use Blacklight's solr connection if available

Closes #168
Closes #166

Author: thatbudakguy

geo_combine.gemspec

@@ -26,6 +26,7 @@ Gem::Specification.new do |spec|
   spec.add_dependency 'thor'
   spec.add_dependency 'faraday-net_http_persistent', '~> 2.0'
   spec.add_dependency 'git'
+  spec.add_dependency 'faraday-retry', '~> 2.2'
 
   spec.add_development_dependency 'bundler'
   spec.add_development_dependency 'rake'

lib/geo_combine/indexer.rb

@@ -1,47 +1,126 @@
 # frozen_string_literal: true
 
 require 'rsolr'
+require 'faraday/retry'
 require 'faraday/net_http_persistent'
+require 'geo_combine/logger'
 
 module GeoCombine
   # Indexes Geoblacklight documents into Solr
   class Indexer
     attr_reader :solr
 
-    def self.solr(url: ENV.fetch('SOLR_URL', 'http://127.0.0.1:8983/solr/blacklight-core'))
-      RSolr.connect url: url, adapter: :net_http_persistent
+    def initialize(solr: nil, logger: GeoCombine::Logger.logger)
+      @logger = logger
+      @batch_size = ENV.fetch('SOLR_BATCH_SIZE', 100).to_i
+
+      # If SOLR_URL is set, use it; if in a Geoblacklight app, use its solr core
+      solr_url = ENV.fetch('SOLR_URL', nil)
+      solr_url ||= Blacklight.default_index.connection.base_uri.to_s if defined? Blacklight
+
+      # If neither, warn and try to use local Blacklight default solr core
+      if solr_url.nil?
+        @logger.warn 'SOLR_URL not set; using Blacklight default'
+        solr_url = 'http://localhost:8983/solr/blacklight-core'
+      end
+
+      @solr = solr || RSolr.connect(client, url: solr_url)
     end
 
-    def initialize(solr: GeoCombine::Indexer.solr)
-      @solr = solr
+    # Index everything and return the number of docs successfully indexed
+    def index(docs)
+      # Track total indexed and time spent
+      @logger.info "indexing into #{solr_url}"
+      total_indexed = 0
+      start_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+      # Index in batches; set batch size via BATCH_SIZE
+      batch = []
+      docs.each do |doc, path|
+        if batch.size < @batch_size
+          batch << [doc, path]
+        else
+          total_indexed += index_batch(batch)
+          batch = []
+        end
+      end
+      total_indexed += index_batch(batch) unless batch.empty?
+
+      # Issue a commit to make sure all documents are indexed
+      @solr.commit
+      end_time = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      sec = end_time - start_time
+      @logger.info format('indexed %<total_indexed>d documents in %<sec>.2f seconds', total_indexed:, sec:)
+      total_indexed
     end
 
+    # URL to the solr instance being used
     def solr_url
       @solr.options[:url]
     end
 
-    # Index everything and return the number of docs successfully indexed
-    def index(docs, commit_within: ENV.fetch('SOLR_COMMIT_WITHIN', 5000).to_i)
-      indexed_count = 0
-
-      docs.each do |record, path|
-        # log the unique identifier for the record for debugging
-        id = record['id'] || record['dc_identifier_s']
-        puts "Indexing #{id}: #{path}" if $DEBUG
-
-        # index the record into solr
-        @solr.update params: { commitWithin: commit_within, overwrite: true },
-                     data: [record].to_json,
-                     headers: { 'Content-Type' => 'application/json' }
-
-        # count the number of records successfully indexed
-        indexed_count += 1
-      rescue RSolr::Error::Http => e
-        puts e
-      end
+    private
 
-      @solr.commit
-      indexed_count
+    # Index a batch of documents; if we fail, index them all individually
+    def index_batch(batch)
+      docs = batch.map(&:first)
+      @solr.update(data: batch_json(docs), params:, headers:)
+      @logger.debug "indexed batch (#{batch.size} docs)"
+      batch.size
+    rescue RSolr::Error::Http => e
+      @logger.error "error indexing batch (#{batch.size} docs): #{format_error(e)}"
+      @logger.warn 'retrying documents individually'
+      batch.map { |doc, path| index_single(doc, path) }.compact.size
+    end
+
+    # Index a single document; if it fails, log the error and continue
+    def index_single(doc, path)
+      @solr.add(doc, params:, headers:)
+      @logger.debug "indexed #{path}"
+      doc
+    rescue RSolr::Error::Http => e
+      @logger.error "error indexing #{path}: #{format_error(e)}"
+      nil
+    end
+
+    # Generate a JSON string to send to solr update API for a batch of documents
+    def batch_json(batch)
+      batch.map { |doc| "add: { doc: #{doc.to_json} }" }.join(",\n").prepend('{ ').concat(' }')
+    end
+
+    # Generate a friendly error message for logging including status code and message
+    def format_error(error)
+      code = error.response[:status]
+      status_info = "#{code} #{RSolr::Error::Http::STATUS_CODES[code.to_i]}"
+      error_info = parse_solr_error(error)
+      [status_info, error_info].compact.join(' - ')
+    end
+
+    # Extract the specific error message from a solr JSON error response, if any
+    def parse_solr_error(error)
+      JSON.parse(error.response[:body]).dig('error', 'msg')
+    rescue StandardError
+      nil
+    end
+
+    def headers
+      { 'Content-Type' => 'application/json' }
+    end
+
+    def params
+      { overwrite: true }
+    end
+
+    def client
+      @client ||= Faraday.new do |conn|
+        conn.request :retry, max: 3, interval: 1, backoff_factor: 2, exceptions: [
+          Faraday::TimeoutError,
+          Faraday::ConnectionFailed,
+          Faraday::TooManyRequestsError
+        ]
+        conn.response :raise_error
+        conn.adapter :net_http_persistent
+      end
     end
   end
 end

spec/lib/geo_combine/indexer_spec.rb

@@ -3,10 +3,22 @@
 require 'geo_combine/indexer'
 require 'spec_helper'
 
+# Mock an available Blacklight installation
+class FakeBlacklight
+  def self.default_index
+    Repository
+  end
+
+  class Repository
+    def self.connection; end
+  end
+end
+
 RSpec.describe GeoCombine::Indexer do
-  subject(:indexer) { described_class.new(solr: solr) }
+  subject(:indexer) { described_class.new(solr:, logger:) }
 
-  let(:solr) { instance_double(RSolr::Client) }
+  let(:logger) { instance_double(Logger, warn: nil, info: nil, error: nil, debug: nil) }
+  let(:solr) { instance_double(RSolr::Client, options: { url: 'TEST' }) }
   let(:docs) do
     [
       [{ 'id' => '1' }, 'path/to/record1.json'],              # v1.0 schema
@@ -21,36 +33,69 @@
 
   describe '#initialize' do
     before do
-      stub_const('ENV', 'SOLR_URL' => 'http://localhost:8983/solr/geoblacklight')
       allow(RSolr).to receive(:connect).and_return(solr)
     end
 
-    it 'connects to a solr instance if set in the environment' do
-      described_class.new
-      expect(RSolr).to have_received(:connect).with(
-        url: 'http://localhost:8983/solr/geoblacklight',
-        adapter: :net_http_persistent
-      )
+    context 'when solr url is set in the environment' do
+      before do
+        stub_const('ENV', 'SOLR_URL' => 'http://localhost:8983/solr/geoblacklight')
+      end
+
+      it 'connects to the solr instance' do
+        described_class.new(logger:)
+        expect(RSolr).to have_received(:connect).with(
+          be_a(Faraday::Connection),
+          url: 'http://localhost:8983/solr/geoblacklight'
+        )
+      end
+    end
+
+    context 'when there is a configured Blacklight connection' do
+      before do
+        stub_const('Blacklight', FakeBlacklight)
+        allow(FakeBlacklight::Repository).to receive(:connection).and_return(
+          instance_double(RSolr::Client, base_uri: URI('http://localhost:8983/solr/geoblacklight'))
+        )
+      end
+
+      it 'connects to the solr instance' do
+        described_class.new(logger:)
+        expect(RSolr).to have_received(:connect).with(
+          be_a(Faraday::Connection),
+          url: 'http://localhost:8983/solr/geoblacklight'
+        )
+      end
+    end
+
+    context 'when solr url is not set' do
+      before do
+        stub_const('ENV', {})
+      end
+
+      it 'falls back to the Blacklight default' do
+        described_class.new(logger:)
+        expect(RSolr).to have_received(:connect).with(
+          be_a(Faraday::Connection),
+          url: 'http://localhost:8983/solr/blacklight-core'
+        )
+      end
     end
   end
 
   describe '#index' do
-    it 'posts each record to solr as JSON' do
-      indexer.index([docs[0]], commit_within: 1)
+    let(:solr_error_msg) {  { error: { msg: 'error message' } }.to_json }
+    let(:solr_response) { { status: '400', body: solr_error_msg } }
+    let(:error) { RSolr::Error::Http.new({ uri: URI('') }, solr_response) }
+
+    it 'sends records in batches to solr' do
+      indexer.index(docs)
       expect(solr).to have_received(:update).with(
-        params: { commitWithin: 1, overwrite: true },
-        data: [docs[0][0]].to_json,
-        headers: { 'Content-Type' => 'application/json' }
+        data: "{ add: { doc: {\"id\":\"1\"} },\nadd: { doc: {\"dc_identifier_s\":\"2\"} } }",
+        headers: { 'Content-Type' => 'application/json' },
+        params: { overwrite: true }
       )
     end
 
-    it 'prints the id and path of each record in debug mode' do
-      $DEBUG = true
-      expect { indexer.index([docs[0]]) }.to output("Indexing 1: path/to/record1.json\n").to_stdout
-      expect { indexer.index([docs[1]]) }.to output("Indexing 2: path/to/record2.json\n").to_stdout
-      $DEBUG = false
-    end
-
     it 'commits changes to solr after indexing' do
       indexer.index(docs)
       expect(solr).to have_received(:commit).once
@@ -59,5 +104,31 @@
     it 'returns the count of records successfully indexed' do
       expect(indexer.index(docs)).to eq 2
     end
+
+    context 'when an error occurs during batch indexing' do
+      before do
+        allow(solr).to receive(:update).and_raise(error)
+        allow(solr).to receive(:add)
+      end
+
+      it 'attempts to index records individually' do
+        total = indexer.index(docs)
+        expect(solr).to have_received(:add).twice
+        expect(total).to eq 2
+      end
+    end
+
+    context 'when an error occurs during individual indexing' do
+      before do
+        allow(solr).to receive(:update).and_raise(error)
+        allow(solr).to receive(:add).with(docs[0][0], anything).and_raise(error)
+        allow(solr).to receive(:add).with(docs[1][0], anything)
+      end
+
+      it 'continues indexing' do
+        total = indexer.index(docs)
+        expect(total).to eq 1
+      end
+    end
   end
 end