[NU-2009] feature: active scenarios limit (global and per processing type limits) (#7970)

Author: mk-software-pl

.run/RunEnvForLocalDesigner.run.xml

@@ -12,4 +12,4 @@
       <option name="Make" enabled="true" />
     </method>
   </configuration>
-</component>
+</component>

designer/client/src/components/graph/node-modal/fragment-input-definition/settings/variants/fields/UserDefinedListInput.tsx

@@ -1,22 +1,21 @@
-import { useSelector } from "react-redux";
-import { getProcessingType } from "../../../../../../../reducers/selectors/graph";
-import { GenericValidationRequest } from "../../../../../../../actions/nk/adhocTesting";
-import { debounce } from "lodash";
-import { useSettings } from "../../SettingsProvider";
 import { Box, Button, CircularProgress, FormControl, Stack } from "@mui/material";
+import { debounce } from "lodash";
 import React, { useMemo, useRef, useState } from "react";
 import type AceEditor from "react-ace";
-
-
 import type { IAceEditor } from "react-ace/lib/types";
 import { useTranslation } from "react-i18next";
+import { useSelector } from "react-redux";
+
+import type { GenericValidationRequest } from "../../../../../../../actions/nk/adhocTesting";
 import HttpService from "../../../../../../../http/HttpService";
+import { getProcessingType } from "../../../../../../../reducers/selectors/graph";
 import type { NodeValidationError, ReturnedType, VariableTypes } from "../../../../../../../types";
 import { useDelayedEnterAction } from "../../../../../../toolbars/scenarioDetails/useDelayedEnterAction";
 import { SpelEditor } from "../../../../editors/expression/SpelEditor";
 import { ExpressionLang } from "../../../../editors/expression/types";
 import { getValidationErrorsForField, mandatoryValueValidator, uniqueValueValidator } from "../../../../editors/Validators";
 import type { FieldName, FixedValuesOption, onChangeType } from "../../../item";
+import { useSettings } from "../../SettingsProvider";
 import { ListItems } from "./ListItems";
 import { SettingLabelStyled } from "./StyledSettingsComponnets";
 

designer/client/src/components/graph/node-modal/parametersListAdvanced.tsx

@@ -1,9 +1,9 @@
 import { partition } from "lodash";
-import type { PropsWithChildren} from "react";
+import type { PropsWithChildren } from "react";
 import React, { useCallback, useMemo } from "react";
 import { useTranslation } from "react-i18next";
 
-import type { Parameter} from "../../../types";
+import type { Parameter } from "../../../types";
 import { ParameterCategory } from "../../../types";
 import { Expandable } from "../../common/Expandable";
 import type { ParameterExpressionFieldProps } from "./ParameterExpressionField";

designer/deployment-manager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/DeploymentManager.scala

@@ -1,6 +1,6 @@
 package pl.touk.nussknacker.engine.api.deployment
 
-import cats.effect.{IO, Resource, SyncIO}
+import cats.effect.{Resource, SyncIO}
 import com.typesafe.config.Config
 import pl.touk.nussknacker.engine.api.definition.EngineScenarioCompilationDependencies
 import pl.touk.nussknacker.engine.api.deployment.scheduler.services._
@@ -35,7 +35,6 @@ trait DeploymentManager extends AutoCloseable {
 
   protected final def notImplemented: Future[Nothing] =
     Future.failed(new NotImplementedError())
-
 }
 
 trait ManagerSpecificScenarioActivitiesStoredByManager { self: DeploymentManager =>

designer/deployment-manager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/DeploymentStatus.scala

@@ -8,15 +8,9 @@ import io.circe.generic.extras.semiauto.deriveUnwrappedCodec
 // Currently DeploymentStatus are limited set of allowed statuses. Only ProblemDeploymentStatus can have different
 // descriptions depending on DM implementation. It makes implementation of logic based on statuses easier. In case
 // if we have requirement to make it more flexible, we can relax this restriction.
-sealed trait DeploymentStatus extends EnumEntry with UpperSnakecase {
-  def name: DeploymentStatusName = DeploymentStatusName(entryName)
-}
-
-sealed abstract class NoAttributesDeploymentStatus extends DeploymentStatus
-
-final case class ProblemDeploymentStatus(description: String) extends DeploymentStatus {
-  override def name: DeploymentStatusName = ProblemDeploymentStatus.name
-}
+sealed trait DeploymentStatus                                        extends EnumEntry with UpperSnakecase
+sealed abstract class NoAttributesDeploymentStatus                   extends DeploymentStatus
+final case class ProblemDeploymentStatus(problemDescription: String) extends DeploymentStatus
 
 object DeploymentStatus extends Enum[DeploymentStatus] {
 
@@ -41,19 +35,45 @@ object DeploymentStatus extends Enum[DeploymentStatus] {
   case object DuringCancel extends NoAttributesDeploymentStatus
   case object Canceled     extends NoAttributesDeploymentStatus
 
-}
+  implicit class IsActive(val status: DeploymentStatus) extends AnyVal {
 
-object ProblemDeploymentStatus {
-  def name: DeploymentStatusName = DeploymentStatusName("PROBLEM")
+    def isActive: Boolean = {
+      status match {
+        case DuringDeploy | Running | Restarting                             => true
+        case Finished | DuringCancel | Canceled | ProblemDeploymentStatus(_) => false
+      }
+    }
+
+  }
 
-  def extractDescription(status: DeploymentStatus): Option[String] =
-    status match {
-      case problem: ProblemDeploymentStatus =>
-        Some(problem.description)
-      case _: NoAttributesDeploymentStatus =>
-        None
+  implicit class ProblemDescription(val status: DeploymentStatus) extends AnyVal {
+
+    def problemDescription: Option[String] = status match {
+      case _: NoAttributesDeploymentStatus             => None
+      case ProblemDeploymentStatus(problemDescription) => Some(problemDescription)
     }
 
+  }
+
+  implicit class ToDeploymentStatusName(val status: DeploymentStatus) extends AnyVal {
+
+    def name: DeploymentStatusName = status match {
+      case status: NoAttributesDeploymentStatus => DeploymentStatusName(status.entryName)
+      case ProblemDeploymentStatus(_)           => DeploymentStatusName.problemStatusName
+    }
+
+  }
+
+  def from(name: DeploymentStatusName, description: Option[String]): DeploymentStatus = {
+    name match {
+      case DeploymentStatusName.problemStatusName =>
+        val desc = description.getOrElse(throw new IllegalStateException("No description for ProblemDeploymentStatus"))
+        ProblemDeploymentStatus(desc)
+      case other =>
+        DeploymentStatus.withName(other.value)
+    }
+  }
+
 }
 
 final case class DeploymentStatusName(value: String) {
@@ -64,4 +84,5 @@ object DeploymentStatusName {
 
   implicit val codec: Codec[DeploymentStatusName] = deriveUnwrappedCodec[DeploymentStatusName]
 
+  val problemStatusName: DeploymentStatusName = DeploymentStatusName("PROBLEM")
 }

designer/deployment-manager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/simple/SimpleProcessStateDefinitionManager.scala

@@ -12,10 +12,11 @@ import pl.touk.nussknacker.engine.api.deployment.StateStatus.StatusName
 import pl.touk.nussknacker.engine.api.deployment.simple.SimpleStateStatus.{statusActionsPF, ProblemStateStatus}
 
 /**
-  * Base [[ProcessStateDefinitionManager]] with basic state definitions and state transitions.
-  * Provides methods to handle erroneous edge cases.
-  * @see [[SimpleStateStatus]]
-  */
+ * Base [[ProcessStateDefinitionManager]] with basic state definitions and state transitions.
+ * Provides methods to handle erroneous edge cases.
+ *
+ * @see [[SimpleStateStatus]]
+ */
 object SimpleProcessStateDefinitionManager extends ProcessStateDefinitionManager {
 
   override def statusActions(input: ScenarioStatusWithScenarioContext): Set[ScenarioActionName] =
@@ -26,16 +27,15 @@ object SimpleProcessStateDefinitionManager extends ProcessStateDefinitionManager
   )
 
   private[nussknacker] def statusDescription(status: StateStatus): String = status match {
-    case _ @ProblemStateStatus(message, _, _) => message
-    case _                                    => SimpleStateStatus.definitions(status.name).description
+    case s: ProblemStateStatus => s.description
+    case _                     => SimpleStateStatus.definitions(status.name).description
   }
 
   override def statusTooltip(input: ScenarioStatusWithScenarioContext): String = statusTooltip(input.scenarioStatus)
 
   private[nussknacker] def statusTooltip(status: StateStatus): String = status match {
-    case _ @ProblemStateStatus(message, _, Some(tooltip)) => tooltip
-    case _ @ProblemStateStatus(message, _, _)             => message
-    case _                                                => SimpleStateStatus.definitions(status.name).tooltip
+    case s: ProblemStateStatus => s.tooltip.getOrElse(s.description)
+    case _                     => SimpleStateStatus.definitions(status.name).tooltip
   }
 
   override def stateDefinitions: Map[StatusName, StateDefinitionDetails] =

designer/deployment-manager-api/src/main/scala/pl/touk/nussknacker/engine/api/deployment/simple/SimpleStateStatus.scala

@@ -2,7 +2,11 @@ package pl.touk.nussknacker.engine.api.deployment.simple
 
 import pl.touk.nussknacker.engine.api.deployment._
 import pl.touk.nussknacker.engine.api.deployment.StateStatus.StatusName
-import pl.touk.nussknacker.engine.api.deployment.simple.SimpleStateStatus.ProblemStateStatus.defaultActions
+import pl.touk.nussknacker.engine.api.deployment.simple.SimpleStateStatus.ProblemStateStatus.{
+  GeneralProblemStateStatus,
+  MultipleJobsRunning,
+  ShouldNotBeRunning
+}
 import pl.touk.nussknacker.engine.api.process.VersionId
 import pl.touk.nussknacker.engine.deployment.DeploymentId
 
@@ -12,23 +16,56 @@ object SimpleStateStatus {
 
   def fromDeploymentStatus(deploymentStatus: DeploymentStatus): StateStatus = {
     deploymentStatus match {
-      case noAttributes: NoAttributesDeploymentStatus => NoAttributesStateStatus(noAttributes.name.value)
+      case status: NoAttributesDeploymentStatus => NoAttributesStateStatus(status.name.value)
       // We assume that all deployment status have default allowedActions. Non-default allowedActions have only
       // statuses that are not deployment statuses but scenario statuses.
-      case problem: ProblemDeploymentStatus => ProblemStateStatus(problem.description)
+      case status: ProblemDeploymentStatus => GeneralProblemStateStatus(status.problemDescription)
     }
   }
 
-  // Represents general problem.
-  final case class ProblemStateStatus(
-      description: String,
-      allowedActions: Set[ScenarioActionName] = defaultActions,
-      tooltip: Option[String] = None
-  ) extends StateStatus {
+  sealed trait ProblemStateStatus extends StateStatus {
     override def name: StatusName = ProblemStateStatus.name
+
+    def description: String
+    def allowedActions: Set[ScenarioActionName]
+    def tooltip: Option[String]
   }
 
   object ProblemStateStatus {
+
+    final case class ShouldNotBeRunning(deployed: Boolean) extends ProblemStateStatus {
+
+      override val description: String =
+        if (deployed) "Scenario has been canceled but still is running."
+        else "Scenario has been never deployed but now is running."
+
+      override val allowedActions: Set[ScenarioActionName] = defaultActions
+      override val tooltip: Option[String]                 = None
+    }
+
+    final case class MultipleJobsRunning(
+        firstNonFinalDeployment: (DeploymentId, StateStatus),
+        secondNonFinalDeployment: (DeploymentId, StateStatus),
+        otherNonFinalDeployments: (DeploymentId, StateStatus)*
+    ) extends ProblemStateStatus {
+
+      override val allowedActions: Set[ScenarioActionName] = Set(ScenarioActionName.Cancel)
+      override val description: String                     = "More than one deployment is running."
+
+      override val tooltip: Option[StatusName] = Some {
+        (firstNonFinalDeployment :: secondNonFinalDeployment :: otherNonFinalDeployments.toList)
+          .map { case (deploymentId, deploymentStatus) => s"$deploymentId - $deploymentStatus" }
+          .mkString("Expected one job, instead: ", ", ", "")
+      }
+
+    }
+
+    final case class GeneralProblemStateStatus(
+        override val description: String,
+        override val allowedActions: Set[ScenarioActionName] = defaultActions,
+        override val tooltip: Option[String] = None
+    ) extends ProblemStateStatus
+
     val name: String = "PROBLEM"
 
     def isProblemStatus(status: StateStatus): Boolean = status.name == name
@@ -40,48 +77,54 @@ object SimpleStateStatus {
 
     // Problem factory methods
 
-    val Failed: ProblemStateStatus = ProblemStateStatus(defaultDescription)
+    val Failed: ProblemStateStatus = GeneralProblemStateStatus(defaultDescription)
 
     val ArchivedShouldBeCanceled: ProblemStateStatus =
-      ProblemStateStatus("Archived scenario should be canceled.", Set(ScenarioActionName.Cancel))
+      GeneralProblemStateStatus("Archived scenario should be canceled.", Set(ScenarioActionName.Cancel))
 
     val FailedToGet: ProblemStateStatus =
-      ProblemStateStatus(s"Failed to get a state of the scenario.")
+      GeneralProblemStateStatus(s"Failed to get a state of the scenario.")
 
     def shouldBeRunning(deployedVersionId: VersionId, user: String): ProblemStateStatus =
-      ProblemStateStatus(s"Scenario deployed in version $deployedVersionId by $user is not running.")
+      GeneralProblemStateStatus(s"Scenario deployed in version $deployedVersionId by $user is not running.")
 
     def mismatchDeployedVersion(
         deployedVersionId: VersionId,
         exceptedVersionId: VersionId,
         user: String
     ): ProblemStateStatus =
-      ProblemStateStatus(
+      GeneralProblemStateStatus(
         s"Scenario deployed in version $deployedVersionId by $user, expected version $exceptedVersionId."
       )
 
-    def shouldNotBeRunning(deployed: Boolean): ProblemStateStatus = {
-      val shouldNotBeRunningMessage =
-        if (deployed) "Scenario has been canceled but still is running."
-        else "Scenario has been never deployed but now is running."
-      ProblemStateStatus(shouldNotBeRunningMessage)
-    }
+    def shouldNotBeRunning(deployed: Boolean): ProblemStateStatus =
+      ShouldNotBeRunning(deployed)
 
     def missingDeployedVersion(exceptedVersionId: VersionId, user: String): ProblemStateStatus =
-      ProblemStateStatus(s"Scenario deployed without version by $user, expected version $exceptedVersionId.")
-
-    def multipleJobsRunning(nonFinalDeploymentIds: List[(DeploymentId, StateStatus)]): ProblemStateStatus =
-      ProblemStateStatus(
-        description = "More than one deployment is running.",
-        allowedActions = Set(ScenarioActionName.Cancel),
-        tooltip = Some(
-          nonFinalDeploymentIds
-            .map { case (deploymentId, deploymentStatus) =>
-              s"$deploymentId - $deploymentStatus"
-            }
-            .mkString("Expected one job, instead: ", ", ", "")
-        )
-      )
+      GeneralProblemStateStatus(s"Scenario deployed without version by $user, expected version $exceptedVersionId.")
+
+    def multipleJobsRunning(
+        first: (DeploymentId, StateStatus),
+        second: (DeploymentId, StateStatus),
+        others: (DeploymentId, StateStatus)*
+    ): ProblemStateStatus =
+      MultipleJobsRunning(first, second, others: _*)
+
+  }
+
+  implicit class CanBeConsideredAsActiveStatus(val status: StateStatus) extends AnyVal {
+
+    def isActive: Boolean = {
+      status match {
+        case problemStatus: ProblemStateStatus =>
+          problemStatus match {
+            case _: ShouldNotBeRunning | _: MultipleJobsRunning => true
+            case _: GeneralProblemStateStatus                   => false
+          }
+        case `Restarting` => true
+        case status       => DefaultFollowingDeployStatuses.contains(status)
+      }
+    }
 
   }
 
@@ -119,7 +162,8 @@ object SimpleStateStatus {
       Set(ScenarioActionName.Deploy, ScenarioActionName.Cancel)
     // When Failed - process is in terminal state in Flink and it doesn't require any cleanup in Flink, but in NK it does
     // - that's why Cancel action is available
-    case SimpleStateStatus.ProblemStateStatus(_, allowedActions, _) => allowedActions
+    case s: ShouldNotBeRunning                           => s.allowedActions
+    case GeneralProblemStateStatus(_, allowedActions, _) => allowedActions
   }
 
   val definitions: Map[StatusName, StateDefinitionDetails] = Map(

designer/server/src/main/scala/pl/touk/nussknacker/ui/api/DeploymentApiHttpService.scala

@@ -57,6 +57,10 @@ class DeploymentApiHttpService(
                     ScenarioGraphValidationError(errors)
                   case DeploymentService.DeployValidationError(message) =>
                     DeployValidationError(message)
+                  case DeploymentService.MaxActiveScenariosCountExceededError(maxCount) =>
+                    DeployValidationError(
+                      s"The limit of active scenarios has been reached. You can have a maximum of $maxCount active scenarios."
+                    )
                 }
               case ActivityService.CommentValidationError(message) => CommentValidationError(message)
             })
@@ -75,7 +79,7 @@ class DeploymentApiHttpService(
               _.map { statusWithModifiedAt =>
                 GetDeploymentStatusResponse(
                   statusWithModifiedAt.value.name,
-                  ProblemDeploymentStatus.extractDescription(statusWithModifiedAt.value),
+                  statusWithModifiedAt.value.problemDescription,
                   statusWithModifiedAt.modifiedAt.toInstant
                 )
               }.left.map {

designer/server/src/main/scala/pl/touk/nussknacker/ui/api/description/DeploymentApiEndpoints.scala

@@ -138,7 +138,7 @@ class DeploymentApiEndpoints(auth: EndpointInput[AuthCredentials]) extends BaseE
               Example.of(
                 GetDeploymentStatusResponse(
                   DeploymentStatus.Problem.Failed.name,
-                  Some(DeploymentStatus.Problem.Failed.description),
+                  Some(DeploymentStatus.Problem.Failed.problemDescription),
                   exampleInstant
                 ),
                 Some("PROBLEM status")