Live data preview and node transition throughput on Flink minicluster

build.sbt

@@ -647,6 +647,7 @@ lazy val flinkDeploymentManager = (project in flink("management"))
   )
   .dependsOn(
     deploymentManagerApi % Provided,
+    liveDataCollector    % Provided,
     scenarioCompilerFlinkDeps,
     flinkMiniCluster,
     commonUtils          % Provided,
@@ -2023,6 +2024,20 @@ lazy val deploymentManagerApi = (project in file("designer/deployment-manager-ap
   )
   .dependsOn(extensionsApi, testUtils % Test)
 
+lazy val liveDataCollector = (project in file("designer/live-data-collector"))
+  .settings(commonSettings)
+  .settings(
+    name := "nussknacker-live-data-collector",
+    libraryDependencies ++= Seq(
+      "org.scalatest" %% "scalatest" % scalaTestV % Test,
+    ),
+  )
+  .dependsOn(
+    deploymentManagerApi % Provided,
+    // For testResultsVariableEncoder purpose
+    scenarioCompiler     % Provided,
+  )
+
 lazy val prepareDesignerTests     = taskKey[Unit]("Prepare all necessary artifacts before running designer module tests")
 lazy val prepareDesignerSlowTests =
   taskKey[Unit]("Prepare all necessary artifacts before running designer module slow tests")
@@ -2166,6 +2181,7 @@ lazy val designer = (project in file("designer/server"))
     processReports,
     security,
     deploymentManagerApi,
+    liveDataCollector,
     componentsApi,
     restmodel,
     listenerApi,
@@ -2317,6 +2333,7 @@ lazy val modules = List[ProjectReference](
   customHttpServiceApi,
   configLoaderApi,
   deploymentManagerApi,
+  liveDataCollector,
   designer,
   sqlComponents,
   schemedKafkaComponentsUtils,

components-api/src/main/scala/pl/touk/nussknacker/engine/ModelConfig.scala

@@ -4,11 +4,13 @@ import com.typesafe.config.Config
 import net.ceedubs.ficus.Ficus.toFicusConfig
 import net.ceedubs.ficus.readers.AnyValReaders._
 import net.ceedubs.ficus.readers.OptionReader._
+import pl.touk.nussknacker.engine.ModelConfig.LiveDataPreviewMode
 import pl.touk.nussknacker.engine.api.namespaces.NamingStrategy
 
 final case class ModelConfig(
     allowEndingScenarioWithoutSink: Boolean,
     namingStrategy: NamingStrategy,
+    liveDataPreviewMode: LiveDataPreviewMode,
     // TODO: we should parse this underlying config as ModelConfig class fields instead of passing raw config
     underlyingConfig: Config,
 ) {
@@ -23,8 +25,33 @@ object ModelConfig {
     ModelConfig(
       allowEndingScenarioWithoutSink = rawModelConfig.getOrElse[Boolean]("allowEndingScenarioWithoutSink", false),
       namingStrategy = NamingStrategy.fromConfig(rawModelConfig),
+      liveDataPreviewMode = parseLiveDataPreviewMode(rawModelConfig),
       underlyingConfig = rawModelConfig,
     )
   }
 
+  sealed trait LiveDataPreviewMode
+
+  object LiveDataPreviewMode {
+
+    case object Disabled extends LiveDataPreviewMode
+
+    final case class Enabled(
+        maxNumberOfSamples: Int,
+        throughputTimeWindowInSeconds: Int,
+    ) extends LiveDataPreviewMode
+
+  }
+
+  private def parseLiveDataPreviewMode(config: Config): LiveDataPreviewMode = {
+    if (config.getOrElse("liveDataPreview.enabled", false)) {
+      LiveDataPreviewMode.Enabled(
+        maxNumberOfSamples = config.getOrElse("liveDataPreview.maxNumberOfSamples", 10),
+        throughputTimeWindowInSeconds = config.getOrElse("liveDataPreview.throughputTimeWindowInSeconds", 60),
+      )
+    } else {
+      LiveDataPreviewMode.Disabled
+    }
+  }
+
 }

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

@@ -2,10 +2,13 @@ package pl.touk.nussknacker.engine.api.deployment
 
 import cats.effect.{Resource, SyncIO}
 import com.typesafe.config.Config
+import io.circe.Json
 import pl.touk.nussknacker.engine.api.definition.EngineScenarioCompilationDependencies
+import pl.touk.nussknacker.engine.api.deployment.LiveDataPreviewSupported.{LiveData, LiveDataError}
 import pl.touk.nussknacker.engine.api.deployment.scheduler.services._
 import pl.touk.nussknacker.engine.api.process.{ProcessIdWithName, ProcessName}
 import pl.touk.nussknacker.engine.newdeployment
+import pl.touk.nussknacker.engine.testmode.TestProcess.{NodeTransition, TestResults}
 
 import java.time.Instant
 import scala.concurrent.Future
@@ -33,6 +36,8 @@ trait DeploymentManager extends AutoCloseable {
 
   def scenarioCompilationDependenciesResource: Resource[SyncIO, EngineScenarioCompilationDependencies]
 
+  def liveDataPreviewSupport: LiveDataPreviewSupport
+
   protected final def notImplemented: Future[Nothing] =
     Future.failed(new NotImplementedError())
 }
@@ -89,3 +94,30 @@ trait SchedulingSupported extends SchedulingSupport {
 }
 
 case object NoSchedulingSupport extends SchedulingSupport
+
+sealed trait LiveDataPreviewSupport
+
+trait LiveDataPreviewSupported extends LiveDataPreviewSupport {
+
+  def getLiveData(
+      processIdWithName: ProcessIdWithName,
+  ): Future[Either[LiveDataError, LiveData]]
+
+}
+
+object LiveDataPreviewSupported {
+
+  final case class LiveData(
+      liveDataSamples: TestResults[Json],
+      nodeTransitionThroughput: Map[NodeTransition, BigDecimal],
+  )
+
+  sealed trait LiveDataError
+
+  object LiveDataError {
+    case object NoLiveDataAvailableForScenario extends LiveDataError
+  }
+
+}
+
+case object NoLiveDataPreviewSupport extends LiveDataPreviewSupport

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

@@ -19,6 +19,7 @@ class CachingProcessStateDeploymentManager(
     override val deploymentSynchronisationSupport: DeploymentSynchronisationSupport,
     override val deploymentsStatusesQueryForAllScenariosSupport: DeploymentsStatusesQueryForAllScenariosSupport,
     override val schedulingSupport: SchedulingSupport,
+    override val liveDataPreviewSupport: LiveDataPreviewSupport,
 ) extends DeploymentManager {
 
   private val cache: AsyncCache[ProcessName, List[DeploymentStatusDetails]] = Caffeine
@@ -74,6 +75,7 @@ object CachingProcessStateDeploymentManager extends LazyLogging {
           delegate.deploymentSynchronisationSupport,
           delegate.deploymentsStatusesQueryForAllScenariosSupport,
           delegate.schedulingSupport,
+          delegate.liveDataPreviewSupport,
         )
       }
       .getOrElse {

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

@@ -60,6 +60,8 @@ class DeploymentManagerStub(implicit ec: ExecutionContext) extends BaseDeploymen
 
   override def schedulingSupport: SchedulingSupport = NoSchedulingSupport
 
+  override def liveDataPreviewSupport: LiveDataPreviewSupport = NoLiveDataPreviewSupport
+
   override def close(): Unit = {}
 
   override def scenarioCompilationDependenciesResource: Resource[SyncIO, EngineScenarioCompilationDependencies] =

designer/deployment-manager-api/src/test/scala/pl/touk/nussknacker/engine/api/deployment/cache/CachingProcessStateDeploymentManagerSpec.scala

@@ -32,6 +32,7 @@ class CachingProcessStateDeploymentManagerSpec
       NoDeploymentSynchronisationSupport,
       NoDeploymentsStatusesQueryForAllScenariosSupport,
       NoSchedulingSupport,
+      NoLiveDataPreviewSupport,
     )
 
     val results = List(
@@ -52,6 +53,7 @@ class CachingProcessStateDeploymentManagerSpec
       NoDeploymentSynchronisationSupport,
       NoDeploymentsStatusesQueryForAllScenariosSupport,
       NoSchedulingSupport,
+      NoLiveDataPreviewSupport,
     )
 
     val firstInvocation = cachingManager.getProcessStatesDeploymentIdNow(DataFreshnessPolicy.CanBeCached)
@@ -71,6 +73,7 @@ class CachingProcessStateDeploymentManagerSpec
       NoDeploymentSynchronisationSupport,
       NoDeploymentsStatusesQueryForAllScenariosSupport,
       NoSchedulingSupport,
+      NoLiveDataPreviewSupport,
     )
 
     val resultForFresh = cachingManager.getProcessStatesDeploymentIdNow(DataFreshnessPolicy.Fresh)

designer/live-data-collector/src/main/scala/pl/touk/nussknacker/engine/livedata/LiveDataCollectingListener.scala

@@ -0,0 +1,104 @@
+package pl.touk.nussknacker.engine.livedata
+
+import io.circe.Json
+import pl.touk.nussknacker.engine.api._
+import pl.touk.nussknacker.engine.api.exception.NuExceptionInfo
+import pl.touk.nussknacker.engine.api.process.ProcessName
+import pl.touk.nussknacker.engine.testmode.TestInterpreterRunner
+import pl.touk.nussknacker.engine.testmode.TestProcess.TestResults
+
+import scala.util.Try
+
+// This class must be serializable. It means, that when deserializing, we lose the reference to it.
+// The actual data is stored in the LiveDataCollectingListenerHolder, and all instances of LiveDataCollectingListener can access the data.
+class LiveDataCollectingListener private[livedata] (
+    processName: ProcessName,
+    maxNumberOfSamples: Int,
+    throughputTimeWindowInSeconds: Int,
+) extends ProcessListener
+    with Serializable {
+
+  private val variableEncoder: Any => io.circe.Json = TestInterpreterRunner.testResultsVariableEncoder
+
+  private def storage = LiveDataCollectingListenerHolder.storage(
+    processName = processName,
+    maxNumberOfSamples = maxNumberOfSamples,
+    throughputTimeWindowInSeconds = throughputTimeWindowInSeconds,
+  )
+
+  override def nodeEntered(nodeId: String, context: Context, processMetaData: MetaData): Unit = {
+    updateResults(context, _.updateNodeResult(nodeId, context, variableEncoder))
+  }
+
+  override def transitionToNextNode(
+      nodeId: String,
+      nextNodeId: String,
+      context: Context,
+      processMetaData: MetaData,
+  ): Unit = {
+    storage.registerTransitionBetweenNodes(nodeId, Some(nextNodeId))
+    updateResults(context, _.updateNodeOutputResult(nodeId, Some(nextNodeId), context, variableEncoder))
+  }
+
+  override def processingFinishedInNode(
+      nodeId: String,
+      context: Context,
+      processMetaData: MetaData,
+  ): Unit = {
+    storage.registerTransitionBetweenNodes(nodeId, None)
+    updateResults(context, _.updateNodeOutputResult(nodeId, None, context, variableEncoder))
+  }
+
+  override def endEncountered(
+      nodeId: String,
+      ref: String,
+      context: Context,
+      processMetaData: MetaData
+  ): Unit = ()
+
+  override def deadEndEncountered(
+      lastNodeId: String,
+      context: Context,
+      processMetaData: MetaData
+  ): Unit = {}
+
+  override def expressionEvaluated(
+      nodeId: String,
+      expressionId: String,
+      expression: String,
+      context: Context,
+      processMetaData: MetaData,
+      result: Any
+  ): Unit = {
+    updateResults(context, _.updateExpressionResult(nodeId, context, expressionId, result, variableEncoder))
+  }
+
+  override def serviceInvoked(
+      nodeId: String,
+      id: String,
+      context: Context,
+      processMetaData: MetaData,
+      result: Try[Any]
+  ): Unit = {}
+
+  override def exceptionThrown(exceptionInfo: NuExceptionInfo): Unit = {
+    updateResults(exceptionInfo.context, _.updateExceptionResult(exceptionInfo, variableEncoder))
+  }
+
+  override final def close(): Unit = ()
+
+  private def updateResults(context: Context, action: TestResults[Json] => TestResults[Json]): Unit = {
+    // todo: The ContextId will be refactored to be represented by case class with fields,
+    //       but for now we have to use regex in order to parse the initial context id as generated by the source.
+    val pattern = raw"^(.+?-.+?-\d+-\d+)(-.+)?".r
+    val initialContextIdOpt = context.id match {
+      case pattern(initialContextId, _*) => Some(initialContextId)
+      case _                             => None
+    }
+    initialContextIdOpt match {
+      case Some(initialContextId) => storage.updateResults(initialContextId, action)
+      case None                   => ()
+    }
+  }
+
+}

designer/live-data-collector/src/main/scala/pl/touk/nussknacker/engine/livedata/LiveDataCollectingListenerHolder.scala

@@ -0,0 +1,56 @@
+package pl.touk.nussknacker.engine.livedata
+
+import com.github.benmanes.caffeine.cache.Caffeine
+import pl.touk.nussknacker.engine.api.deployment.LiveDataPreviewSupported.{LiveData, LiveDataError}
+import pl.touk.nussknacker.engine.api.process.ProcessName
+
+import java.time.Clock
+import scala.compat.java8.FunctionConverters._
+
+object LiveDataCollectingListenerHolder {
+
+  private implicit val clock: Clock = Clock.systemUTC()
+
+  private val listenerStorages =
+    Caffeine
+      .newBuilder()
+      .expireAfterAccess(java.time.Duration.ofHours(1))
+      .build[String, LiveDataCollectingListenerStorage]()
+
+  def createListenerFor(
+      processName: ProcessName,
+      maxNumberOfSamples: Int,
+      throughputTimeWindowInSeconds: Int,
+  ): LiveDataCollectingListener = {
+    cleanResults(processName)
+    new LiveDataCollectingListener(processName, maxNumberOfSamples, throughputTimeWindowInSeconds)
+  }
+
+  def getLiveDataPreview(processName: ProcessName): Either[LiveDataError, LiveData] = {
+    Option(listenerStorages.getIfPresent(processName.value)).map(_.getLiveData) match {
+      case Some(liveData) => Right(liveData)
+      case None           => Left(LiveDataError.NoLiveDataAvailableForScenario)
+    }
+  }
+
+  private[livedata] def storage(
+      processName: ProcessName,
+      maxNumberOfSamples: Int,
+      throughputTimeWindowInSeconds: Int,
+  ): LiveDataCollectingListenerStorage = {
+    listenerStorages.get(
+      processName.value,
+      asJavaFunction((_: String) =>
+        new LiveDataCollectingListenerStorage(maxNumberOfSamples, throughputTimeWindowInSeconds)
+      )
+    )
+  }
+
+  // We want to store and present the live data from the most recent deployment:
+  //  - the data from the old run is stored until the listener storage cache expires (currently hardcoded 1 hour)
+  //  - or new deployment is done - then we discard all old data
+  private def cleanResults(processName: ProcessName): Unit = {
+    listenerStorages.invalidate(processName.value)
+  }
+
+}

designer/live-data-collector/src/main/scala/pl/touk/nussknacker/engine/livedata/LiveDataCollectingListenerStorage.scala

@@ -0,0 +1,38 @@
+package pl.touk.nussknacker.engine.livedata
+
+import io.circe.Json
+import pl.touk.nussknacker.engine.api.deployment.LiveDataPreviewSupported.LiveData
+import pl.touk.nussknacker.engine.testmode.TestProcess._
+
+import java.time.{Clock, Instant}
+
+private[livedata] class LiveDataCollectingListenerStorage(
+    maxNumberOfSamples: Int,
+    throughputTimeWindowInSeconds: Int,
+)(implicit clock: Clock) {
+
+  private val results            = new RingBuffer[String, TestResults[Json]](maxNumberOfSamples)
+  private val transitionsCounter = new SlidingWindowCounter[NodeTransition](Instant.now, throughputTimeWindowInSeconds)
+
+  def getLiveData: LiveData = {
+    LiveData(
+      liveDataSamples = TestResults.aggregate(results.values),
+      nodeTransitionThroughput = transitionsCounter.getThroughput,
+    )
+  }
+
+  def updateResults(contextId: String, action: TestResults[Json] => TestResults[Json]): Unit = {
+    results.update(
+      contextId,
+      {
+        case Some(resultsSoFar) => action(resultsSoFar)
+        case None               => action(TestResults.empty[Json])
+      }
+    )
+  }
+
+  def registerTransitionBetweenNodes(from: String, to: Option[String]): Unit = {
+    transitionsCounter.add(NodeTransition(from, to))
+  }
+
+}