fix: add OpenSearch/Elasticsearch 7+ compatibility

README.md

@@ -1,3 +1,5 @@
+> This repository is a fork of https://github.com/omc/searchyll used for indexing the [developers.italia.it](https://github.com/italia/developers.italia.it) website. All the changes were contributed upstream, so when all our [pull requests](https://github.com/omc/searchyll/pulls?utf8=✓&q=is%3Apr+author%3Aalranel+) are merged we can get rid of this fork.
+
 # Searchyll
 
 Search for Jekyll apps. A plugin for indexing your pages into a search engine.
@@ -33,6 +35,20 @@ elasticsearch:
   default_type: "post"              # Optional. Default type is "post".
   custom_settings: _es_settings.yml # Optional. No default. Relative to your src folder
   custom_mappings: _es_mappings.yml # Optional. No default. Relative to your src folder
+  ignore:                           # Optional. No default.
+    - /news/*
+```
+
+### ES Auth
+Elasticsearch authentication can be provided both by passing user and password as URI parameter:
+```yaml
+url: https://user:[email protected]
+```
+Or by easily setting those ENV variables:
+```sh
+ELASTICSEARCH_URL=http://localhost:9200/
+ELASTICSEARCH_USER=elastic
+ELASTICSEARCH_PASS=changeme
 ```
 
 ### Custom Settings File Example

lib/searchyll.rb

@@ -37,11 +37,11 @@
     # puts %(        indexing page #{page.url})
 
     if (indexer = indexers[page.site])
-      indexer << page.data.merge({
-        id:     page.name,
-        url:    page.url,
-        text:   nokogiri_doc.xpath("//article//text()").to_s.gsub(/\s+/, " ")
-      })
+      indexer << ({
+        "id"   => page.url,
+        "url"  => page.url,
+        "text" => nokogiri_doc.xpath("//article//text()").to_s.gsub(/\s+/, " ")
+      }).merge(page.data)
     end
   end
 
@@ -53,11 +53,11 @@
     # puts %(        indexing document #{document.url})
 
     if (indexer = indexers[document.site])
-      indexer << document.data.merge({
-        id:     document.id,
-        url:    document.url,
-        text:   nokogiri_doc.xpath("//article//text()").to_s.gsub(/\s+/, " ")
-      })
+      indexer << ({
+        "id"   =>  document.id,
+        "url"  =>  document.url,
+        "text" =>  nokogiri_doc.xpath("//article//text()").to_s.gsub(/\s+/, " ")
+      }).merge(document.data)
     end
   end
 

lib/searchyll/configuration.rb

@@ -11,6 +11,12 @@ def elasticsearch_url
       ENV['BONSAI_URL'] || ENV['ELASTICSEARCH_URL'] ||
         ((site.config||{})['elasticsearch']||{})['url']
     end
+    def elasticsearch_user
+      ENV['ELASTICSEARCH_USER']
+    end
+    def elasticsearch_pass
+      ENV['ELASTICSEARCH_PASS']
+    end
 
     def valid?
       elasticsearch_url && !elasticsearch_url.empty? && elasticsearch_url.start_with?('http')
@@ -69,6 +75,11 @@ def elasticsearch_default_type
       site.config['elasticsearch']['default_type'] || 'post'
     end
 
+    # Getter for the ignore regex
+    def elasticsearch_ignore
+      site.config['elasticsearch']['ignore'] || []
+    end
+
     # Getter for es mapping
     def elasticsearch_mapping_path
       site.config['elasticsearch']['custom_mappings']
@@ -84,18 +95,15 @@ def elasticsearch_mapping
     end
 
     def elasticsearch_settings
-      shards = site.config['elasticsearch']['number_of_shards'] || 1
-      replicas = site.config['elasticsearch']['number_of_replicas'] || 1
-      read_yaml(elasticsearch_settings_path, {
-        'index' => {
-          'number_of_shards'   => shards,
-          'number_of_replicas' => replicas,
-          'refresh_interval'   => -1
-        }
-      })
-    end
-
-    def read_yaml(path, default)
+      settings = read_yaml(elasticsearch_settings_path)
+      settings['index'] ||= {}
+      settings['index']['number_of_shards']   ||= site.config['elasticsearch']['number_of_shards'] || 1
+      settings['index']['number_of_replicas'] ||= site.config['elasticsearch']['number_of_replicas'] || 1
+      settings['index']['refresh_interval']   ||= -1
+      settings
+    end
+
+    def read_yaml(path, default = {})
       if path
         joined_path = File.join(@site.source, path)
         expanded_path = File.expand_path(joined_path)

lib/searchyll/indexer.rb

@@ -19,22 +19,35 @@ class Indexer
     attr_accessor :queue
     attr_accessor :timestamp
     attr_accessor :uri
+    attr_accessor :es_user
+    attr_accessor :es_pass
     attr_accessor :working
+    attr_accessor :ignore_regex
 
     # Initialize a basic indexer, with a Jekyll site configuration, waiting
     # to be supplied with documents for indexing.
     def initialize(configuration)
       self.configuration = configuration
       self.uri           = URI(configuration.elasticsearch_url)
+      self.es_user       = configuration.elasticsearch_user
+      self.es_pass       = configuration.elasticsearch_pass
       self.queue         = Queue.new
       self.working       = true
       self.timestamp     = Time.now
       self.batch_size    = BATCH_SIZE
+
+      # Compute a regex for detecting paths to ignore
+      escaped = (configuration.elasticsearch_ignore.map {|i| Regexp.escape(i).gsub('\*','.+?')}).join('|')
+      self.ignore_regex = Regexp.new "^(#{escaped})$", Regexp::IGNORECASE
     end
 
     # Public: Add new documents for batch indexing.
     def <<(doc)
-      queue << doc
+      if doc['url'] =~ self.ignore_regex
+        #puts %(        ...ignoring)
+      else
+        queue << doc
+      end
     end
 
     # Public: start the indexer and wait for documents to index.
@@ -97,7 +110,7 @@ def working?
     # A versioned index name, based on the time of the indexing run.
     # Will be later added to an alias for hot reindexing.
     def elasticsearch_index_name
-      "#{configuration.elasticsearch_index_base_name}-#{timestamp.strftime('%Y%m%d%H%M%S')}"
+      @elasticsearch_index_name ||= "#{configuration.elasticsearch_index_base_name}-#{timestamp.strftime('%Y%m%d%H%M%S')}"
     end
 
     # Prepare an HTTP connection
@@ -117,8 +130,8 @@ def prepare_index
       }
 
       if configuration.elasticsearch_mapping
-        payload['mappings'] = {}
-        payload['mappings'].store(configuration.elasticsearch_default_type, configuration.elasticsearch_mapping)
+        # OpenSearch/ES 7+ doesn't support type in mappings - use flat structure
+        payload['mappings'] = configuration.elasticsearch_mapping
       end
 
       json_payload = payload.to_json
@@ -150,23 +163,30 @@ def http_request(klass, path)
       req = klass.new(path)
       req.content_type = 'application/json'
       req['Accept']    = 'application/json'
-      # Append auth credentials if the exist
-      req.basic_auth(uri.user, uri.password) if uri.user && uri.password
+      # Append auth credentials if they exist
+      # it trying to get them from env and then from
+      # elasticsearch uri itself
+      user = es_user || uri.user
+      pass = es_pass || uri.password
+      req.basic_auth(user, pass) if user && pass
       req
     end
 
     # Given a batch (array) of documents, index them into Elasticsearch
     # using its Bulk Update API.
     # https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html
     def es_bulk_insert!(http, batch)
-      bulk_insert = http_post("/#{elasticsearch_index_name}/#{configuration.elasticsearch_default_type}/_bulk")
+      return if batch.empty?
+      # OpenSearch/ES 7+ doesn't support type in bulk endpoint
+      bulk_insert = http_post("/#{elasticsearch_index_name}/_bulk")
+      bulk_insert.content_type = 'application/x-ndjson'
       bulk_insert.body = batch.map do |doc|
         [{ index: {} }.to_json, doc.to_json].join("\n")
-      end.join("\n") + "\n"
+      end.join("\n").force_encoding('ascii-8bit') + "\n"
       res = http.request(bulk_insert)
       if !res.kind_of?(Net::HTTPSuccess)
         $stderr.puts "Elasticsearch returned an error when performing bulk insert: " + res.message + " " + res.body
-        exit
+        exit(1)
       end
     end
 
@@ -184,12 +204,12 @@ def current_batch
 
     # List the indices currently in the cluster, caching the call in an ivar
     def old_indices
-      # return if defined?(@old_indices)
+      return @old_indices if defined?(@old_indices)
       resp = http_start { |h| h.request(http_get('/_cat/indices?h=index')) }
       indices = JSON.parse(resp.body).map { |i| i['index'] }
       indices = indices.select { |i| i =~ /\A#{configuration.elasticsearch_index_base_name}/ }
       indices -= [elasticsearch_index_name]
-      # @old_indices = indices
+      @old_indices = indices
       indices
     end
 
@@ -225,27 +245,48 @@ def finalize_replication(http)
     # hot swap the index into the canonical alias
     def finalize_aliases(http)
       update_aliases = http_post('/_aliases')
+
+      # perform removal and addition in two different calls so that
+      # the second one is performed even if the first one fails
+      if !old_indices.empty?
+        update_aliases.body = {
+          actions: [
+            { remove: {
+              index: old_indices.join(','),
+              alias: configuration.elasticsearch_index_base_name
+            } }
+          ]
+        }.to_json
+        res = http.request(update_aliases)
+        if !res.kind_of?(Net::HTTPSuccess)
+          $stderr.puts "Elasticsearch returned an error when removing old aliases: " + res.message + " " + res.body
+        end
+      end
+
       update_aliases.body = {
         actions: [
-          { remove: {
-            index: old_indices.join(','),
-            alias: configuration.elasticsearch_index_base_name
-          } },
           { add: {
             index: elasticsearch_index_name,
             alias: configuration.elasticsearch_index_base_name
           } }
         ]
       }.to_json
-      http.request(update_aliases)
+      res = http.request(update_aliases)
+      if !res.kind_of?(Net::HTTPSuccess)
+        $stderr.puts "Elasticsearch returned an error when assigning the new alias: " + res.message + " " + res.body
+        exit(1)
+      end
     end
 
     # delete old indices after a successful reindexing run
     def finalize_cleanup(http)
       return if old_indices.nil? || old_indices.empty?
-      cleanup_indices = http_delete("/#{old_indices.join(',')}")
       puts %(       Old indices: #{old_indices.join(', ')})
-      http.request(cleanup_indices)
+      cleanup_indices = http_delete("/#{old_indices.join(',')}")
+      res = http.request(cleanup_indices)
+      if !res.kind_of?(Net::HTTPSuccess)
+        $stderr.puts "Elasticsearch returned an error when deleting old indices: " + res.message + " " + res.body
+      end
     end
   end
 end