refactor soft deletion of searchindex

Author: jiangphcn

src/main/scala/com/cloudant/clouseau/ClouseauTypeFactory.scala

@@ -34,6 +34,7 @@ case class RenamePathMsg(dbName: String)
 case class CleanupDbMsg(dbName: String, activeSigs: List[String])
 case class DiskSizeMsg(path: String)
 case class CloseLRUByPathMsg(path: String)
+case class SoftDeleteMsg(path: String)
 
 case class Group1Msg(query: String, field: String, refresh: Boolean, groupSort: Any, groupOffset: Int,
                      groupLimit: Int)
@@ -95,6 +96,8 @@ object ClouseauTypeFactory extends TypeFactory {
       Some(DiskSizeMsg(reader.readAs[String]))
     case ('close_lru_by_path, 2) =>
       Some(CloseLRUByPathMsg(reader.readAs[String]))
+    case ('soft_delete, 2) =>
+      Some(SoftDeleteMsg(reader.readAs[String]))
     case ('commit, 2) =>
       Some(CommitMsg(toLong(reader.readTerm)))
     case ('set_update_seq, 2) =>

src/main/scala/com/cloudant/clouseau/IndexCleanupService.scala

@@ -15,9 +15,6 @@ package com.cloudant.clouseau
 import com.yammer.metrics.scala._
 import java.io.File
 import java.util.regex.Pattern
-import java.text.SimpleDateFormat
-import java.util.Calendar
-import java.util.TimeZone
 import org.apache.log4j.Logger
 import scalang._
 
@@ -31,19 +28,6 @@ class IndexCleanupService(ctx: ServiceContext[ConfigurationArgs]) extends Servic
       val dir = new File(rootDir, path)
       logger.info("Removing %s".format(path))
       recursivelyDelete(dir)
-    case RenamePathMsg(dbName: String) =>
-      val srcDir = new File(rootDir, dbName)
-      val sdf = new SimpleDateFormat("yyyyMMdd'.'HHmmss")
-      sdf.setTimeZone(TimeZone.getTimeZone("UTC"))
-      val sdfNow = sdf.format(Calendar.getInstance().getTime())
-      // move timestamp information in dbName to end of destination path
-      // for example, from foo.1234567890 to foo.20170912.092828.deleted.1234567890
-      val destPath = dbName.dropRight(10) + sdfNow + ".deleted." + dbName.takeRight(10)
-      val destDir = new File(rootDir, destPath)
-      logger.info("Renaming '%s' to '%s'".format(
-        srcDir.getAbsolutePath, destDir.getAbsolutePath)
-      )
-      rename(srcDir, destDir)
     case CleanupDbMsg(dbName: String, activeSigs: List[String]) =>
       logger.info("Cleaning up " + dbName)
       val pattern = Pattern.compile("shards/[0-9a-f]+-[0-9a-f]+/" + dbName + "\\.[0-9]+/([0-9a-f]+)$")
@@ -78,14 +62,4 @@ class IndexCleanupService(ctx: ServiceContext[ConfigurationArgs]) extends Servic
       fileOrDir.delete
   }
 
-  private def rename(srcDir: File, destDir: File) {
-    if (!srcDir.isDirectory) {
-      return
-    }
-    if (!srcDir.renameTo(destDir)) {
-      logger.error("Failed to rename directory from '%s' to '%s'".format(
-        srcDir.getAbsolutePath, destDir.getAbsolutePath))
-    }
-  }
-
 }

src/main/scala/com/cloudant/clouseau/IndexManagerService.scala

@@ -136,6 +136,26 @@ class IndexManagerService(ctx: ServiceContext[ConfigurationArgs]) extends Servic
     case CloseLRUByPathMsg(path: String) =>
       lru.closeByPath(path)
       'ok
+    case SoftDeleteMsg(path: String) =>
+      lru.get(path) match {
+        case null =>
+          val manager = self
+          node.spawn((_) => {
+            openTimer.time {
+              IndexService.start(node, ctx.args.config, path, "standard") match {
+                case ('ok, pid: Pid) =>
+                  pid ! 'soft_delete
+                  'ok
+                case error =>
+                  'error
+              }
+            }
+          })
+        case pid =>
+          lru.remove(pid)
+          pid ! 'soft_delete
+          'ok
+      }
     case 'version =>
       ('ok, getClass.getPackage.getImplementationVersion)
   }

src/main/scala/com/cloudant/clouseau/IndexService.scala

@@ -57,6 +57,9 @@ import scalang.Reference
 import com.spatial4j.core.context.SpatialContext
 import com.spatial4j.core.distance.DistanceUtils
 import java.util.HashSet
+import java.util.Calendar
+import java.util.TimeZone
+import java.text.SimpleDateFormat
 
 case class IndexServiceArgs(config: Configuration, name: String, queryParser: QueryParser, writer: IndexWriter)
 case class HighlightParameters(highlighter: Highlighter, highlightFields: List[String], highlightNumber: Int, analyzers: List[Analyzer])
@@ -178,6 +181,11 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
         dir.deleteFile(name)
       }
       exit('deleted)
+    case 'soft_delete =>
+      val dir = ctx.args.writer.getDirectory
+      ctx.args.writer.close()
+      softDelete(ctx.args.name)
+      exit('soft_deleted)
     case 'maybe_commit =>
       commit(pendingSeq, pendingPurgeSeq)
     case ('committed, newUpdateSeq: Long, newPurgeSeq: Long) =>
@@ -220,14 +228,38 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
       ctx.args.writer.rollback()
     } catch {
       case e: AlreadyClosedException => 'ignored
-      case e: IOException =>
-        warn("Error while closing writer", e)
-        ctx.args.writer.close()
+      case e: IOException => warn("Error while closing writer", e)
     } finally {
       super.exit(msg)
     }
   }
 
+  private def softDelete(dbName: String) {
+    val rootDir = new File(ctx.args.config.getString("clouseau.dir", "target/indexes"))
+    val srcDir = new File(rootDir, dbName)
+    val sdf = new SimpleDateFormat("yyyyMMdd'.'HHmmss")
+    sdf.setTimeZone(TimeZone.getTimeZone("UTC"))
+    val sdfNow = sdf.format(Calendar.getInstance().getTime())
+    // move timestamp information in dbName to end of destination path
+    // for example, from foo.1234567890 to foo.20170912.092828.deleted.1234567890
+    val destPath = dbName.dropRight(10) + sdfNow + ".deleted." + dbName.takeRight(10)
+    val destDir = new File(rootDir, destPath)
+    rename(srcDir, destDir)
+  }
+
+  private def rename(srcDir: File, destDir: File) {
+    info("Renaming '%s' to '%s'".format(
+      srcDir.getAbsolutePath, destDir.getAbsolutePath)
+    )
+    if (!srcDir.isDirectory) {
+      return
+    }
+    if (!srcDir.renameTo(destDir)) {
+      warn("Failed to rename directory from '%s' to '%s'".format(
+        srcDir.getAbsolutePath, destDir.getAbsolutePath))
+    }
+  }
+
   private def commit(newUpdateSeq: Long, newPurgeSeq: Long) {
     if (!committing && (newUpdateSeq > updateSeq || newPurgeSeq > purgeSeq)) {
       committing = true

src/test/scala/com/cloudant/clouseau/IndexCleanupServiceSpec.scala

@@ -55,6 +55,22 @@ class IndexManagerServiceSpec extends SpecificationWithJUnit {
 
   }
 
+  "the index soft deletion service" should {
+
+    "softly delete index when database is deleted" in new manager_service {
+      node.call(service, SoftDeleteMsg("foo.5432109876"))
+      Thread.sleep(1000)
+      val indexdir = new File("target", "indexes")
+      var subdirlist = List[String]()
+
+      for (file <- indexdir.listFiles if file.getName contains ".deleted") {
+        subdirlist = file.getName() +: subdirlist
+      }
+      subdirlist.length > 0 must be equalTo true
+    }
+
+  }
+
 }
 
 trait manager_service extends RunningNode {
@@ -90,4 +106,8 @@ trait manager_service extends RunningNode {
     foo2indexdir.mkdirs
   }
 
+  val foo3dir = new File(new File("target", "indexes"), "foo.5432109876")
+  if (!foo3dir.exists) {
+    foo3dir.mkdirs
+  }
 }