Merge branch 'release/1.8.3' into show-my-search-results

# Conflicts:
#	CHANGELOG.md

Author: lmarini

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## 1.8.3 - 2020-03-12
 
+### Changed
+- Elasticsearch indexer will new add new metadata fields as strings to avoid unexpected behavior on date fields.
+
 ### Fixed
 - Ability to delete tags from sections on file [CATS-1042](https://opensource.ncsa.illinois.edu/jira/browse/CATS-1046)
 - Ability to delete tags on file page.

app/api/Admin.scala

@@ -34,7 +34,7 @@ class Admin @Inject() (userService: UserService,
    */
   def deleteAllData(resetAll: Boolean) = ServerAdminAction { implicit request =>
     current.plugin[MongoSalatPlugin].map(_.dropAllData(resetAll))
-    current.plugin[ElasticsearchPlugin].map(_.deleteAll)
+    current.plugin[ElasticsearchPlugin].map(_.deleteAll())
 
     Ok(toJson("done"))
   }

app/api/Comments.scala

@@ -38,7 +38,7 @@ class Comments @Inject()(datasets: DatasetService, comments: CommentService, eve
                     datasets.get(parent.dataset_id.get) match {
                       case Some(dataset) => {
                         current.plugin[ElasticsearchPlugin].foreach {
-                          _.index(dataset, false)
+                          _.index(dataset, false, None)
                         }
                       }
                       case None => Logger.error("Dataset not found: " + id)

app/models/QueuedAction.scala

@@ -12,13 +12,15 @@ import play.api.libs.json._
  */
 // ElasticsearchQueue service action parameters
 case class ElasticsearchParameters(
-                                    recursive: Boolean = false
+                                    recursive: Boolean = false,
+                                    index: Option[String] = None
                                   ) {}
 
 object ElasticsearchParameters {
   implicit val elasticParamsWrites = new Writes[ElasticsearchParameters] {
     def writes(params: ElasticsearchParameters): JsValue = Json.obj(
-      "recursive" -> params.recursive.toString
+      "recursive" -> params.recursive.toString,
+      "index" -> params.index.getOrElse("").toString
     )
   }
 }

app/services/CollectionService.scala

@@ -243,10 +243,10 @@ trait CollectionService {
   def syncUpRootSpaces(collectionId: UUID, initialParents: List[UUID])
 
   /** Queue all collections to be indexed in Elasticsearch. */
-  def indexAll()
+  def indexAll(idx: Option[String] = None)
 
   /** Queue a collection to be indexed in Elasticsearch. */
-  def index(id: UUID)
+  def index(id: UUID, idx: Option[String] = None)
 
   def incrementViews(id: UUID, user: Option[User]): (Int, Date)
 

app/services/DatasetService.scala

@@ -250,10 +250,10 @@ trait DatasetService {
   def selectNewThumbnailFromFiles(datasetId: UUID)
 
   /** Queue all datasets to be indexed in Elasticsearch. */
-  def indexAll()
+  def indexAll(idx: Option[String] = None)
 
   /** Queue a dataset to be indexed in Elasticsearch. */
-  def index(id: UUID)
+  def index(id: UUID, idx: Option[String] = None)
 
   def removeTags(id: UUID, userIdStr: Option[String], eid: Option[String], tags: List[String])
 

app/services/ElasticsearchPlugin.scala

@@ -6,26 +6,28 @@ import org.elasticsearch.common.xcontent.XContentBuilder
 import org.elasticsearch.search.aggregations.AggregationBuilders
 import org.elasticsearch.search.aggregations.bucket.terms.StringTerms.Bucket
 import play.api.libs.json.Json._
+
 import scala.util.Try
-import scala.collection.mutable.{MutableList, ListBuffer}
+import scala.collection.mutable.{ListBuffer, MutableList}
 import scala.collection.immutable.List
-import play.api.{Plugin, Logger, Application}
+import play.api.{Application, Logger, Plugin}
 import org.elasticsearch.common.settings.Settings
 import org.elasticsearch.client.transport.TransportClient
 import org.elasticsearch.common.transport.InetSocketTransportAddress
 import java.net.InetAddress
 import java.util.regex.Pattern
 
 import org.elasticsearch.common.xcontent.XContentFactory._
-import org.elasticsearch.action.search.{SearchPhaseExecutionException, SearchType, SearchResponse}
+import org.elasticsearch.action.search.{SearchPhaseExecutionException, SearchResponse, SearchType}
 import org.elasticsearch.client.transport.NoNodeAvailableException
 import org.elasticsearch.ElasticsearchException
 import org.elasticsearch.indices.IndexAlreadyExistsException
-
-import models.{Collection, Dataset, File, Folder, UUID, ResourceRef, Section, ElasticsearchResult, User}
+import org.elasticsearch.index.reindex.{ReindexAction, ReindexPlugin, ReindexRequestBuilder}
+import models.{Collection, Dataset, ElasticsearchResult, File, Folder, ResourceRef, Section, UUID, User}
 import play.api.Play.current
 import play.api.libs.json._
 import _root_.util.SearchUtils
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest
 
 
 /**
@@ -67,7 +69,7 @@ class ElasticsearchPlugin(application: Application) extends Plugin {
       } else {
         Settings.settingsBuilder().build()
       }
-      client = Some(TransportClient.builder().settings(settings).build()
+      client = Some(TransportClient.builder().settings(settings).addPlugin(classOf[ReindexPlugin]).build()
         .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(serverAddress), serverPort)))
       Logger.debug("--- Elasticsearch Client is being created----")
       client match {
@@ -324,16 +326,33 @@ class ElasticsearchPlugin(application: Application) extends Plugin {
       }
       case None =>
     }
+  }
 
+  def swapIndex(idx: String): Unit = {
+    client match {
+      case Some(x) => {
+        // Check if swap index exists before swapping
+        if (x.admin.indices.exists(new IndicesExistsRequest(idx)).get().isExists()) {
+          Logger.debug("Deleting "+nameOfIndex+" index...")
+          deleteAll(nameOfIndex)
+          createIndex(nameOfIndex)
+          Logger.debug("Replacing with "+idx+"...")
+          ReindexAction.INSTANCE.newRequestBuilder(x).source(idx).destination(nameOfIndex).get()
+          Logger.debug("Deleting "+idx+" index...")
+          deleteAll(idx)
+        }
+      }
+      case None =>
+    }
   }
 
-  /** Delete all indices */
-  def deleteAll {
+  /** Delete all documents in default index */
+  def deleteAll(idx: String = nameOfIndex) {
     connect()
     client match {
       case Some(x) => {
         try {
-          val response = x.admin().indices().prepareDelete(nameOfIndex).get()
+          val response = x.admin().indices().prepareDelete(idx).get()
           if (!response.isAcknowledged())
             Logger.error("Did not delete all data from elasticsearch.")
         } catch {
@@ -398,51 +417,51 @@ class ElasticsearchPlugin(application: Application) extends Plugin {
    * Reindex the given collection, if recursive is set to true it will
    * also reindex all datasets and files.
    */
-  def index(collection: Collection, recursive: Boolean) {
+  def index(collection: Collection, recursive: Boolean, idx: Option[String]) {
     connect()
     // Perform recursion first if necessary
     if (recursive) {
       for (dataset <- datasets.listCollection(collection.id.toString)) {
-        index(dataset, recursive)
+        index(dataset, recursive, idx)
       }
     }
-    index(SearchUtils.getElasticsearchObject(collection))
+    index(SearchUtils.getElasticsearchObject(collection), idx.getOrElse(nameOfIndex))
   }
 
   /**
    * Reindex the given dataset, if recursive is set to true it will
    * also reindex all files.
    */
-  def index(dataset: Dataset, recursive: Boolean) {
+  def index(dataset: Dataset, recursive: Boolean, idx: Option[String]) {
     connect()
     // Perform recursion first if necessary
     if (recursive) {
-      files.get(dataset.files).found.foreach(f => index(f))
+      files.get(dataset.files).found.foreach(f => index(f, idx))
       for (folderid <- dataset.folders) {
         folders.get(folderid) match {
           case Some(f) => {
-            files.get(f.files).found.foreach(fi => index(fi))
+            files.get(f.files).found.foreach(fi => index(fi, idx))
           }
           case None => Logger.error(s"Error getting file $folderid for recursive indexing")
         }
       }
     }
-    index(SearchUtils.getElasticsearchObject(dataset))
+    index(SearchUtils.getElasticsearchObject(dataset), idx.getOrElse(nameOfIndex))
   }
 
   /** Reindex the given file. */
-  def index(file: File) {
+  def index(file: File, idx: Option[String]) {
     connect()
     // Index sections first so they register for tag counts
     for (section <- file.sections) {
-      index(section)
+      index(section, idx)
     }
-    index(SearchUtils.getElasticsearchObject(file))
+    index(SearchUtils.getElasticsearchObject(file), idx.getOrElse(nameOfIndex))
   }
 
-  def index(section: Section) {
+  def index(section: Section, idx: Option[String]) {
     connect()
-    index(SearchUtils.getElasticsearchObject(section))
+    index(SearchUtils.getElasticsearchObject(section), idx.getOrElse(nameOfIndex))
   }
 
   /** Index document using an arbitrary map of fields. */
@@ -687,19 +706,12 @@ class ElasticsearchPlugin(application: Application) extends Plugin {
 
   /** Return string-encoded JSON object describing field types */
   def getElasticsearchObjectMappings(): String = {
-    """dynamic_templates": [{
-       "nonindexer": {
-          "match": "*",
-          "match_mapping_type":"string",
-          "mapping": {
-            "type": "string",
-            "index": "not_analyzed"
-          }
-        }
-      }
-    ],"""
-
+    /** The dynamic template will restrict all dynamic metadata fields to be indexed
+     * as strings for datatypes besides Objects. In the future, this could
+     * be removed, but only once the Search API better supports those data types (e.g. Date).
+     */
     """{"clowder_object": {
+          |"date_detection": false,
           |"properties": {
             |"name": {"type": "string"},
             |"description": {"type": "string"},

app/services/FileService.scala

@@ -123,9 +123,9 @@ trait FileService {
    */
   def first(): Option[File]
 
-  def indexAll()
+  def indexAll(idx: Option[String] = None)
 
-  def index(id: UUID)
+  def index(id: UUID, idx: Option[String] = None)
 
   /**
    * Directly insert file into database, for example if the file path is local.

app/services/mongodb/ElasticsearchQueue.scala

@@ -29,29 +29,31 @@ class ElasticsearchQueue @Inject() (
   // process the next entry in the queue
   def handler(action: QueuedAction) = {
     val recursive = action.elastic_parameters.fold(false)(_.recursive)
+    val idx: Option[String] = action.elastic_parameters.fold[Option[String]](None)(_.index)
 
     action.target match {
       case Some(targ) => {
         action.action match {
           case "index_file" => {
             val target = files.get(targ.id) match {
-              case Some(f) => current.plugin[ElasticsearchPlugin].foreach(p => p.index(f))
+              case Some(f) => current.plugin[ElasticsearchPlugin].foreach(p => p.index(f, idx))
               case None => throw new NullPointerException(s"File ${targ.id.stringify} no longer found for indexing")
             }
           }
           case "index_dataset" => {
             val target = datasets.get(targ.id) match {
-              case Some(ds) => current.plugin[ElasticsearchPlugin].foreach(p => p.index(ds, recursive))
+              case Some(ds) => current.plugin[ElasticsearchPlugin].foreach(p => p.index(ds, recursive, idx))
               case None => throw new NullPointerException(s"Dataset ${targ.id.stringify} no longer found for indexing")
             }
           }
           case "index_collection" => {
             val target = collections.get(targ.id) match {
-              case Some(c) => current.plugin[ElasticsearchPlugin].foreach(p => p.index(c, recursive))
+              case Some(c) => current.plugin[ElasticsearchPlugin].foreach(p => p.index(c, recursive, idx))
               case None => throw new NullPointerException(s"Collection ${targ.id.stringify} no longer found for indexing")
             }
           }
           case "index_all" => _indexAll()
+          case "index_swap" => _swapIndex()
           case _ => throw new IllegalArgumentException(s"Unrecognized action: ${action.action}")
         }
       }
@@ -61,6 +63,7 @@ class ElasticsearchQueue @Inject() (
           case "index_dataset" => throw new IllegalArgumentException(s"No target specified for action ${action.action}")
           case "index_collection" => throw new IllegalArgumentException(s"No target specified for action ${action.action}")
           case "index_all" => _indexAll()
+          case "index_swap" => _swapIndex()
           case _ => throw new IllegalArgumentException(s"Unrecognized action: ${action.action}")
         }
       }
@@ -70,13 +73,25 @@ class ElasticsearchQueue @Inject() (
   def _indexAll() = {
     // Add all individual entries to the queue and delete this action
     current.plugin[ElasticsearchPlugin].foreach(p => {
-      // delete & recreate index
-      p.deleteAll
-      p.createIndex()
+      val idx = p.nameOfIndex + "_reindex_temp_swap"
+      Logger.debug("Reindexing database into temporary reindex file: "+idx)
+      p.createIndex(idx)
+
       // queue everything for each resource type
-      collections.indexAll()
-      datasets.indexAll()
-      files.indexAll()
+      collections.indexAll(Some(idx))
+      datasets.indexAll(Some(idx))
+      files.indexAll(Some(idx))
+
+      // queue action to swap index once we're done reindexing
+      p.queue.queue("index_swap")
+    })
+  }
+
+  // Replace the main index with the newly reindexed temp file
+  def _swapIndex() = {
+    Logger.debug("Swapping temporary reindex for main index")
+    current.plugin[ElasticsearchPlugin].foreach(p => {
+      p.swapIndex(p.nameOfIndex + "_reindex_temp_swap")
     })
   }
 }

app/services/mongodb/MongoDBCollectionService.scala

@@ -892,16 +892,16 @@ class MongoDBCollectionService @Inject() (
     }
   }
 
-  def indexAll() = {
+  def indexAll(idx: Option[String] = None) = {
     // Bypass Salat in case any of the file records are malformed to continue past them
     Collection.dao.collection.find(MongoDBObject(), MongoDBObject("_id" -> 1)).foreach(c => {
-      index(new UUID(c.get("_id").toString))
+      index(new UUID(c.get("_id").toString), idx)
     })
   }
 
-  def index(id: UUID) {
+  def index(id: UUID, idx: Option[String] = None) {
     try
-      esqueue.queue("index_collection", new ResourceRef('collection, id))
+      esqueue.queue("index_collection", new ResourceRef('collection, id), new ElasticsearchParameters(index=idx))
     catch {
       case except: Throwable => Logger.error(s"Error queuing collection ${id.stringify}: ${except}")
       case _ => Logger.error(s"Error queuing collection ${id.stringify}")