allow deactivating recipes (#1848)

* adds deactivate button to the dashboard to improve the performance of the initial load

* adds allActive method to RecipeManager so that only active recipes can be fetched from the possible storage backend

Author: bekiroguz

core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/AkkaBaker.scala

@@ -85,8 +85,7 @@ class AkkaBaker private[runtime](config: AkkaBakerConfig) extends scaladsl.Baker
 
   config.bakerActorProvider.initialize(system)
 
-  private val recipeManager: RecipeManager =
-    config.recipeManager
+  private val recipeManager: RecipeManager = config.recipeManager
 
   private val processIndexActor: ActorRef =
     config.bakerActorProvider.createProcessIndexActor(config.interactions, recipeManager, system.settings.config)
@@ -106,7 +105,7 @@ class AkkaBaker private[runtime](config: AkkaBakerConfig) extends scaladsl.Baker
       val updated = recipeRecord.updated
       if (!recipeRecord.validate || config.bakerValidationSettings.allowAddingRecipeWithoutRequiringInstances) {
         logger.debug(s"Recipe implementation errors are ignored for ${recipe.name}:${recipe.recipeId}")
-        addToManager(recipe, updated)
+        recipeManager.put(recipeRecord.copy(updated = updated))
       } else {
         logger.debug(s"Recipe ${recipe.name}:${recipe.recipeId} is validated for compatibility with interactions")
         getImplementationErrors(recipe).flatMap { implementationErrors =>
@@ -115,16 +114,13 @@ class AkkaBaker private[runtime](config: AkkaBakerConfig) extends scaladsl.Baker
           } else if (recipe.validationErrors.nonEmpty) {
             Future.failed(RecipeValidationException(s"Recipe ${recipe.name}:${recipe.recipeId} has validation errors: ${recipe.validationErrors.mkString(", ")}"))
           } else {
-            addToManager(recipe, updated)
+            recipeManager.put(recipeRecord.copy(updated = updated))
           }
         }
       }
     }
   } yield result
 
-  private def addToManager(compiledRecipe: CompiledRecipe, timeCreated: Long): Future[String] =
-    recipeManager.put(RecipeRecord.of(compiledRecipe, updated = timeCreated))
-
   private def getImplementationErrors(compiledRecipe: CompiledRecipe): Future[Set[String]] = {
     compiledRecipe.interactionTransitions.toList
       .traverse(x => config.interactions.incompatibilities(x).map((_, x.originalInteractionName)))
@@ -141,7 +137,7 @@ class AkkaBaker private[runtime](config: AkkaBakerConfig) extends scaladsl.Baker
     * @return
     */
   override def getRecipe(recipeId: String): Future[RecipeInformation] = {
-    // here we ask the RecipeManager actor to return us the recipe for the given id
+    // here we ask the RecipeManager to return us the recipe for the given id
     recipeManager.get(recipeId).flatMap {
       case Some(r: RecipeRecord) =>
         getImplementationErrors(r.recipe).map(errors => RecipeInformation(r.recipe, r.updated, errors, r.validate, r.recipe.sensoryEvents ))
@@ -155,12 +151,12 @@ class AkkaBaker private[runtime](config: AkkaBakerConfig) extends scaladsl.Baker
     getRecipe(recipeId).map(r => RecipeVisualizer.visualizeRecipe(r.compiledRecipe, style))
 
   /**
-    * Returns all recipes added to this baker instance.
+    * Returns all 'active' recipes added to this baker instance.
     *
     * @return All recipes in the form of map of recipeId -> CompiledRecipe
     */
   override def getAllRecipes: Future[Map[String, RecipeInformation]] = {
-    recipeManager.all
+    recipeManager.allActive
       .flatMap(
         _.toList
           .traverse(ri => getImplementationErrors(ri.recipe)

core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_index/ProcessIndex.scala

@@ -147,7 +147,6 @@ class ProcessIndex(recipeInstanceIdleTimeout: Option[FiniteDuration],
   private val restartRandomFactor: Double =  config.getDouble("baker.process-instance.restart-randomFactor")
 
   private val index: mutable.Map[String, ActorMetadata] = mutable.Map[String, ActorMetadata]()
-  private val recipeCache: mutable.Map[String, (CompiledRecipe, Long)] = mutable.Map[String, (CompiledRecipe, Long)]()
 
   //TODO chose if to use the CassandraBakerCleanup or the ActorBasedBakerCleanup
   private val cleanup: BakerCleanup = {
@@ -168,41 +167,37 @@ class ProcessIndex(recipeInstanceIdleTimeout: Option[FiniteDuration],
   }
 
   // TODO this is a synchronous ask on an actor which createProcessActor is considered bad practice, alternative?
-  private def getRecipeIdFromManager(recipeId: String): Option[(CompiledRecipe, Long)] = {
-    log.debug("Adding recipe to recipe cache: {}", recipeId)
-    val futureResult: Future[Option[RecipeRecord]] = recipeManager.get(recipeId)
-    val result = Await.result(futureResult, updateCacheTimeout)
-    recipeCache ++= result.map(r => r.recipeId -> (r.recipe, r.updated))
-    log.debug("Added recipe to recipe cache: {}, current size: {}", recipeId, recipeCache.size)
-    result.map(r => r.recipe -> r.updated)
+  private def getRecipeRecord(recipeId: String, reactivate: Boolean): Option[RecipeRecord] = {
+    val eventualRecord = recipeManager.get(recipeId).flatMap {
+      case Some(recipeRecord) if recipeRecord.isActive => Future.successful(Some(recipeRecord))
+      case Some(recipeRecord) if !reactivate => Future.successful(Some(recipeRecord)) // inactive recipe, but reactivation is not needed
+      case Some(recipeRecord) if reactivate => // inactive recipe, reactivate it
+        log.info(s"Inactive recipe $recipeId being reactivated.")
+        val activeRecipe = recipeRecord.copy(isActive = true)
+        recipeManager.put(activeRecipe)
+        Future.successful(Some(activeRecipe))
+      case None => Future.successful(None)
+    }
+    Await.result(eventualRecord, updateCacheTimeout)
   }
 
   def getRecipeIdFromActor(actorRef: ActorRef) : String = actorRef.path.name
 
-  def getRecipeWithTimeStamp(recipeId: String): Option[(CompiledRecipe, Long)] =
-    recipeCache.get(recipeId) match {
-      case None =>
-        getRecipeIdFromManager(recipeId)
-      case other => other
-    }
-
-  def getCompiledRecipe(recipeId: String): Option[CompiledRecipe] =
-    getRecipeWithTimeStamp(recipeId).fold[Option[CompiledRecipe]] {
+  private def getCompiledRecipe(recipeId: String, reactivate: Boolean = true): Option[CompiledRecipe] = {
+    getRecipeRecord(recipeId, reactivate).fold[Option[CompiledRecipe]] {
       log.warning(s"No recipe found for $recipeId")
       None
     } {
-      case (recipe, _) => Some(recipe)
+      case r:RecipeRecord => Some(r.recipe)
       case _ => None
     }
-
-  def getOrCreateProcessActor(recipeInstanceId: String): Option[ActorRef] =
-    context.child(recipeInstanceId).orElse(createProcessActor(recipeInstanceId))
+  }
 
   def getProcessActor(recipeInstanceId: String): Option[ActorRef] =
     context.child(recipeInstanceId)
 
-  def createProcessActor(recipeInstanceId: String): Option[ActorRef] =
-    getCompiledRecipe(index(recipeInstanceId).recipeId).map(createProcessActor(recipeInstanceId, _))
+  private def createProcessActor(recipeInstanceId: String, reactivateRecipe: Boolean = true): Option[ActorRef] =
+    getCompiledRecipe(index(recipeInstanceId).recipeId, reactivateRecipe).map(createProcessActor(recipeInstanceId, _))
 
   // creates a ProcessInstanceActor, does not do any validation
   def createProcessActor(recipeInstanceId: String, compiledRecipe: CompiledRecipe): ActorRef = {
@@ -238,7 +233,7 @@ class ProcessIndex(recipeInstanceIdleTimeout: Option[FiniteDuration],
 
   def shouldDelete(meta: ActorMetadata): Boolean = {
     if(meta.processStatus != Deleted)
-      getCompiledRecipe(meta.recipeId) match {
+      getCompiledRecipe(meta.recipeId, reactivate = false) match {
         case Some(recipe) =>
           recipe.retentionPeriod.exists { p => meta.createdDateTime + p.toMillis < System.currentTimeMillis() }
         case None =>
@@ -257,7 +252,7 @@ class ProcessIndex(recipeInstanceIdleTimeout: Option[FiniteDuration],
           actorRef ! Stop(delete = true)
         case None =>
           log.debug(s"Deleting ${meta.recipeInstanceId} via cleanup tool")
-          getCompiledRecipe(meta.recipeId) match {
+          getCompiledRecipe(meta.recipeId, reactivate = false) match {
             case Some(compiledRecipe) =>
               val persistenceId = ProcessInstance.recipeInstanceId2PersistenceId(compiledRecipe.name, meta.recipeInstanceId)
               log.debug(s"Deleting with persistenceId: ${persistenceId}")
@@ -284,6 +279,10 @@ class ProcessIndex(recipeInstanceIdleTimeout: Option[FiniteDuration],
     }
   }
 
+  // This util function is used only for delete process functionality, therefore passing reactivateRecipe=false to avoid reactivating the recipe
+  private def getOrCreateProcessActor(recipeInstanceId: String): Option[ActorRef] =
+    context.child(recipeInstanceId).orElse(createProcessActor(recipeInstanceId, reactivateRecipe = false))
+
   private def forgetProcesses(): Unit = {
     rememberProcessDuration.map {
       duration: Duration =>
@@ -520,8 +519,8 @@ class ProcessIndex(recipeInstanceIdleTimeout: Option[FiniteDuration],
       index.get(recipeInstanceId) match {
         case Some(processMetadata) if processMetadata.isDeleted => sender() ! ProcessDeleted(recipeInstanceId)
         case Some(processMetadata) =>
-          getRecipeWithTimeStamp(processMetadata.recipeId) match {
-            case Some((compiledRecipe, timestamp)) => sender() ! RecipeFound(compiledRecipe, timestamp)
+          getRecipeRecord(processMetadata.recipeId, reactivate = true) match {
+            case Some(RecipeRecord(_, _, updated, recipe, _, _)) => sender() ! RecipeFound(recipe, updated)
             case None => sender() ! NoSuchProcess(recipeInstanceId)
           }
         case None => sender() ! NoSuchProcess(recipeInstanceId)

core/akka-runtime/src/main/scala/com/ing/baker/runtime/recipe_manager/ActorBasedRecipeManager.scala

@@ -13,27 +13,52 @@ import com.ing.baker.runtime.common.RecipeRecord
 import com.typesafe.config.Config
 import net.ceedubs.ficus.Ficus._
 
+import scala.collection.concurrent.TrieMap
 import scala.concurrent.{ExecutionContext, Future}
 
 private class ActorBasedRecipeManager(actor: ActorRef, timeouts: Timeouts)(implicit val ex: ExecutionContext) extends RecipeManager {
 
+  /**
+   * A cache to store the recipes that have been retrieved from the actor system.
+   * Needed to avoid querying the actor too often.
+   */
+  private val cache: TrieMap[String, RecipeRecord] = TrieMap.empty
+
   override def put(recipeRecord: RecipeRecord): Future[String] = {
     implicit val timeout: Timeout = timeouts.defaultAddRecipeTimeout
-    (actor ? AddRecipe(recipeRecord.recipe)).mapTo[AddRecipeResponse].map(_.recipeId)
+    (actor ? AddRecipe(recipeRecord.recipe)).mapTo[AddRecipeResponse].map { response =>
+      cache += ((response.recipeId, recipeRecord))
+      response.recipeId
+    }
   }
 
   override def get(recipeId: String): Future[Option[RecipeRecord]] = {
-    implicit val timeout: Timeout = timeouts.defaultInquireTimeout
-    (actor ? GetRecipe(recipeId)).map {
-      case RecipeFound(compiledRecipe, timestamp) => Some(RecipeRecord.of(compiledRecipe, updated = timestamp))
-      case NoRecipeFound(_) => None
+    cache.get(recipeId) match {
+      case Some(recipeRecord) => Future.successful(Some(recipeRecord))
+      case None =>
+        implicit val timeout: Timeout = timeouts.defaultInquireTimeout
+        (actor ? GetRecipe(recipeId)).map {
+          case RecipeFound(compiledRecipe, timestamp) =>
+            val recipeRecord = RecipeRecord.of(compiledRecipe, updated = timestamp)
+            cache += ((recipeId, recipeRecord))
+            Some(recipeRecord)
+          case NoRecipeFound(_) => None
+        }
     }
   }
 
   override def all: Future[Seq[RecipeRecord]] = {
-        implicit val timeout: Timeout = timeouts.defaultInquireTimeout
-        (actor ? GetAllRecipes).mapTo[AllRecipes].map(_.recipes.map { r => RecipeRecord.of(r.compiledRecipe, updated = r.timestamp) })
+    implicit val timeout: Timeout = timeouts.defaultInquireTimeout
+    (actor ? GetAllRecipes).mapTo[AllRecipes].map { response =>
+      val recipes = response.recipes.map { r => RecipeRecord.of(r.compiledRecipe, updated = r.timestamp) }
+      recipes.foreach(recipe => cache += ((recipe.recipeId, recipe)))
+      recipes
+    }
   }
+
+  // active/inactive recipes are not supported yet in the actor based recipe manager,
+  // therefore we are returning all recipes here
+  override def allActive: Future[Seq[RecipeRecord]] = all
 }
 
 object ActorBasedRecipeManager {

core/akka-runtime/src/main/scala/com/ing/baker/runtime/recipe_manager/DefaultRecipeManager.scala

@@ -9,9 +9,8 @@ class DefaultRecipeManager(implicit val ex: ExecutionContext) extends RecipeMana
 
   val state:TrieMap[String, RecipeRecord] = TrieMap.empty
 
-
   override def put(recipeRecord: RecipeRecord): Future[String] = {
-    state. += ((recipeRecord.recipeId,recipeRecord))
+    state += ((recipeRecord.recipeId,recipeRecord))
     Future.successful(recipeRecord.recipeId)
   }
 
@@ -22,6 +21,10 @@ class DefaultRecipeManager(implicit val ex: ExecutionContext) extends RecipeMana
   override def all: Future[Seq[RecipeRecord]] = {
     Future.successful(state.values.toSeq)
   }
+
+  override def allActive: Future[Seq[RecipeRecord]] = {
+    Future.successful(state.values.filter(_.isActive).toSeq)
+  }
 }
 
 object DefaultRecipeManager {

core/akka-runtime/src/main/scala/com/ing/baker/runtime/recipe_manager/PollingAware.scala

@@ -5,6 +5,9 @@ import com.typesafe.scalalogging.LazyLogging
 
 import scala.concurrent.{ExecutionContext, Future}
 
+/**
+ * This trait is used to avoid replacing recipes with elder or same creation time.
+ */
 trait PollingAware extends RecipeManager with LazyLogging {
 
   implicit def ex: ExecutionContext

core/akka-runtime/src/main/scala/com/ing/baker/runtime/recipe_manager/RecipeManager.scala

@@ -10,4 +10,6 @@ trait RecipeManager {
   def get(recipeId: String): Future[Option[RecipeRecord]]
 
   def all: Future[Seq[RecipeRecord]]
+
+  def allActive: Future[Seq[RecipeRecord]]
 }

core/akka-runtime/src/test/java/com/ing/baker/BakerTest.java

@@ -61,7 +61,7 @@ public void shouldSetupJBakerWithDefaultActorFramework() throws BakerException,
 
         String recipeInstanceId = UUID.randomUUID().toString();
         Baker jBaker = AkkaBaker.java(config, actorSystem, implementationsList);
-        java.util.Map<String, Value> ingredients = jBaker.addRecipe(RecipeRecord.of(compiledRecipe, System.currentTimeMillis(), false))
+        java.util.Map<String, Value> ingredients = jBaker.addRecipe(RecipeRecord.of(compiledRecipe, System.currentTimeMillis(), false, true))
                 .thenCompose(recipeId -> {
                     assertEquals(compiledRecipe.getValidationErrors().size(), 0);
                     return jBaker.bake(recipeId, recipeInstanceId);
@@ -84,7 +84,7 @@ public void shouldSetupJBakerWithGivenActorFramework() throws BakerException, Ex
         assertEquals(compiledRecipe.getValidationErrors().size(), 0);
 
         Baker jBaker = AkkaBaker.java(config, actorSystem, implementationsList);
-        String recipeId = jBaker.addRecipe(RecipeRecord.of(compiledRecipe, System.currentTimeMillis(), false)).get();
+        String recipeId = jBaker.addRecipe(RecipeRecord.of(compiledRecipe, System.currentTimeMillis(), false, true)).get();
 
         String requestId = UUID.randomUUID().toString();
         jBaker.bake(recipeId, requestId).get();
@@ -104,7 +104,7 @@ public void shouldFailWhenMissingImplementations() throws BakerException, Execut
         CompiledRecipe compiledRecipe = RecipeCompiler.compileRecipe(JavaCompiledRecipeTest.setupComplexRecipe());
         Baker jBaker = AkkaBaker.java(config, actorSystem);
 
-        jBaker.addRecipe(RecipeRecord.of(compiledRecipe, System.currentTimeMillis(), true)).get();
+        jBaker.addRecipe(RecipeRecord.of(compiledRecipe, System.currentTimeMillis(), true, true)).get();
     }
 
     @Test
@@ -119,7 +119,7 @@ public void shouldExecuteCompleteFlow() throws BakerException, ExecutionExceptio
 
         // Setup recipe
         CompiledRecipe compiledRecipe = RecipeCompiler.compileRecipe(JavaCompiledRecipeTest.setupComplexRecipe());
-        String recipeId = jBaker.addRecipe(RecipeRecord.of(compiledRecipe, System.currentTimeMillis(), false)).get();
+        String recipeId = jBaker.addRecipe(RecipeRecord.of(compiledRecipe, System.currentTimeMillis(), false, true)).get();
         EventInstance eventOne = EventInstance.from(new JavaCompiledRecipeTest.EventOne());
         assertEquals(eventOne.getName(), "EventOne");
         assertTrue(eventOne.getProvidedIngredients().isEmpty());

core/akka-runtime/src/test/java/com/ing/baker/Webshop.java

@@ -155,7 +155,7 @@ public void testWebshop() throws ExecutionException, InterruptedException {
         ActorSystem system = ActorSystem.create("webshop");
         Baker baker = AkkaBaker.java(config, system, ImmutableList.of(shipGoodsMock, sendInvoiceMock, manufactureGoodsMock, validateOrderMock));
 
-        String recipeId = baker.addRecipe(RecipeRecord.of(recipe, System.currentTimeMillis(), false)).get();
+        String recipeId = baker.addRecipe(RecipeRecord.of(recipe, System.currentTimeMillis(), false, true)).get();
 
         String recipeInstanceId = "56a70f82-a24d-497f-b3ac-57366adbb39c"; //UUID.randomUUID().toString();