close writer before soft-deletione

Author: jiangphcn

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

@@ -30,9 +30,9 @@ case class SearchRequest(options: Map[Symbol, Any])
 
 case class OpenIndexMsg(peer: Pid, path: String, options: Any)
 case class CleanupPathMsg(path: String)
-case class RenamePathMsg(dbName: String)
 case class CleanupDbMsg(dbName: String, activeSigs: List[String])
 case class DiskSizeMsg(path: String)
+case class SoftDeletePathMsg(path: String)
 
 case class Group1Msg(query: String, field: String, refresh: Boolean, groupSort: Any, groupOffset: Int,
                      groupLimit: Int)
@@ -54,8 +54,6 @@ object ClouseauTypeFactory extends TypeFactory {
       Some(OpenIndexMsg(reader.readAs[Pid], reader.readAs[String], reader.readTerm))
     case ('cleanup, 2) =>
       Some(CleanupPathMsg(reader.readAs[String]))
-    case ('rename, 2) =>
-      Some(RenamePathMsg(reader.readAs[String]))
     case ('cleanup, 3) =>
       Some(CleanupDbMsg(reader.readAs[String], reader.readAs[List[String]]))
     case ('search, 2) =>
@@ -92,6 +90,8 @@ object ClouseauTypeFactory extends TypeFactory {
       Some(DeleteDocMsg(reader.readAs[String]))
     case ('disk_size, 2) =>
       Some(DiskSizeMsg(reader.readAs[String]))
+    case ('soft_delete, 2) =>
+      Some(SoftDeletePathMsg(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,10 @@ 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 SoftDeletePathMsg(path: String) =>
+      logger.info("Soft-deleting " + path)
+      val pattern = Pattern.compile(path + "/([0-9a-f]+)$")
+      softDelete(rootDir, path, pattern)
     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,13 +66,24 @@ class IndexCleanupService(ctx: ServiceContext[ConfigurationArgs]) extends Servic
       fileOrDir.delete
   }
 
-  private def rename(srcDir: File, destDir: File) {
-    if (!srcDir.isDirectory) {
+  private def softDelete(fileOrDir: File, path: String, includePattern: Pattern) {
+    if (!fileOrDir.isDirectory) {
       return
     }
-    if (!srcDir.renameTo(destDir)) {
-      logger.error("Failed to rename directory from '%s' to '%s'".format(
-        srcDir.getAbsolutePath, destDir.getAbsolutePath))
+    for (file <- fileOrDir.listFiles) {
+      softDelete(file, path, includePattern)
+    }
+
+    logger.info(fileOrDir.getAbsolutePath)
+    val m = includePattern.matcher(fileOrDir.getAbsolutePath)
+    if (m.find) {
+      logger.info("Soft-deleting index " + m.group)
+      call('main, ('soft_delete, m.group)) match {
+        case 'ok =>
+          'ok
+        case ('error, 'not_found) =>
+          Utils.rename(rootDir, path, fileOrDir.getName)
+      }
     }
   }
 

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

@@ -119,6 +119,14 @@ class IndexManagerService(ctx: ServiceContext[ConfigurationArgs]) extends Servic
           pid ! 'delete
           'ok
       }
+    case ('soft_delete, path: String) =>
+      lru.get(path) match {
+        case null =>
+          ('error, 'not_found)
+        case pid: Pid =>
+          pid ! 'soft_delete
+          'ok
+      }
     case DiskSizeMsg(path: String) =>
       getDiskSize(path)
     case 'close_lru =>

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

@@ -178,6 +178,13 @@ class IndexService(ctx: ServiceContext[IndexServiceArgs]) extends Service(ctx) w
         dir.deleteFile(name)
       }
       exit('deleted)
+    case 'soft_delete =>
+      ctx.args.writer.close()
+      val rootDir = new File(ctx.args.config.getString("clouseau.dir", "target/indexes"))
+      val path = ctx.args.name.substring(0, ctx.args.name.lastIndexOf('/'))
+      val sig = ctx.args.name.substring(ctx.args.name.lastIndexOf('/') + 1)
+      Utils.rename(rootDir, path, sig)
+      exit('deleted)
     case 'maybe_commit =>
       commit(pendingSeq, pendingPurgeSeq)
     case ('committed, newUpdateSeq: Long, newPurgeSeq: Long) =>
@@ -220,7 +227,16 @@ 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)
+      case e: IOException =>
+        try {
+          ctx.args.writer.close();
+        } finally {
+          val dir = ctx.args.writer.getDirectory
+          if (IndexWriter.isLocked(dir)) {
+            IndexWriter.unlock(dir);
+          }
+        }
+        warn("Error while closing writer", e)
     } finally {
       super.exit(msg)
     }

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

@@ -15,6 +15,12 @@ package com.cloudant.clouseau
 import org.apache.lucene.index.Term
 import org.apache.lucene.util.BytesRef
 import org.apache.lucene.util.NumericUtils
+import org.apache.log4j.Logger
+import java.io.File
+import java.io.IOException
+import java.util.Calendar
+import java.util.TimeZone
+import java.text.SimpleDateFormat
 
 object Utils {
 
@@ -29,4 +35,33 @@ object Utils {
     new BytesRef(string)
   }
 
+  def rename(rootDir: File, dbName: String, sig: String) {
+    val logger = Logger.getLogger("clouseau.utils")
+    val srcParentDir = 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 destParentPath = dbName.dropRight(10) + sdfNow + ".deleted." + dbName.takeRight(10)
+    val destParentDir = new File(rootDir, destParentPath)
+    logger.info("Renaming '%s' to '%s'".format(
+      srcParentDir.getAbsolutePath, destParentDir.getAbsolutePath)
+    )
+    if (!srcParentDir.isDirectory) {
+      return
+    }
+    if (!destParentDir.exists) {
+      destParentDir.mkdirs
+    }
+
+    val srcDir = new File(srcParentDir, sig)
+    val destDir = new File(destParentDir, sig)
+
+    if (!srcDir.renameTo(destDir)) {
+      logger.error("Failed to rename directory from '%s' to '%s'".format(
+        srcDir.getAbsolutePath, destDir.getAbsolutePath))
+    }
+  }
+
 }

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

@@ -22,8 +22,8 @@ class IndexCleanupServiceSpec extends SpecificationWithJUnit {
 
   "the index clean-up service" should {
 
-    "rename index when database is deleted" in new cleanup_service {
-      node.cast(service, RenamePathMsg("foo.1234567890")) must be equalTo 'ok
+    "soft-delete index when database is deleted" in new cleanup_service {
+      node.cast(service, SoftDeletePathMsg("foo.1234567890")) must be equalTo 'ok
       Thread.sleep(1000)
       val indexdir = new File("target", "indexes")
       var subdirlist = List[String]()