qs

Author: mgoworko

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

@@ -38,7 +38,7 @@ object ModelConfig {
 
     final case class Enabled(
         maxNumberOfSamples: Int,
-        frequencyWindowInSeconds: Int,
+        throughputTimeWindowInSeconds: Int,
     ) extends LiveDataPreviewMode
 
   }
@@ -47,7 +47,7 @@ object ModelConfig {
     if (config.getOrElse("liveDataPreview.enabled", false)) {
       LiveDataPreviewMode.Enabled(
         maxNumberOfSamples = config.getOrElse("liveDataPreview.maxNumberOfSamples", 10),
-        frequencyWindowInSeconds = config.getOrElse("liveDataPreview.frequencyWindowInSeconds", 60),
+        throughputTimeWindowInSeconds = config.getOrElse("liveDataPreview.throughputTimeWindowInSeconds", 60),
       )
     } else {
       LiveDataPreviewMode.Disabled

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

@@ -109,7 +109,7 @@ object LiveDataPreviewSupported {
 
   final case class LiveDataPreview(
       liveDataSamples: TestResults[Json],
-      nodeTransitionFrequency: Map[NodeTransition, BigDecimal],
+      nodeTransitionThroughput: Map[NodeTransition, BigDecimal],
   )
 
 }

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

@@ -217,7 +217,7 @@ class ManagementResources(
                                 mapResultsToHttpResponse(
                                   ResultsWithCountsDto.from(
                                     resultsWithCounts = value,
-                                    nodeTransitionFrequency = None,
+                                    nodeTransitionThroughput = None,
                                     skipResultsPerNode = SkipResultsPerNode(skipResultsPerNode),
                                     skipResultsPerTransition = SkipResultsPerTransition(skipResultsPerTransition)
                                   )

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

@@ -240,7 +240,7 @@ class ScenarioTestingApiHttpService(
             )
           } yield ResultsWithCountsDto.from(
             resultsWithCounts,
-            Some(liveDataPreview.nodeTransitionFrequency),
+            Some(liveDataPreview.nodeTransitionThroughput),
             skipResultsPerNode.getOrElse(SkipResultsPerNode(false)),
             skipResultsPerTransition.getOrElse(SkipResultsPerTransition(false))
           )

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

@@ -201,21 +201,21 @@ object Dtos {
   final case class ResultsWithCountsDto(
       results: TestResultsDto,
       counts: Map[String, NodeCount],
-      nodeTransitionFrequency: Option[List[NodeTransitionFrequencyDto]],
+      nodeTransitionThroughput: Option[List[NodeTransitionThroughputDto]],
   )
 
   object ResultsWithCountsDto {
 
     def from(
         resultsWithCounts: ResultsWithCounts,
-        nodeTransitionFrequency: Option[Map[NodeTransition, BigDecimal]],
+        nodeTransitionThroughput: Option[Map[NodeTransition, BigDecimal]],
         skipResultsPerNode: SkipResultsPerNode,
         skipResultsPerTransition: SkipResultsPerTransition
     ): ResultsWithCountsDto = {
       ResultsWithCountsDto(
         results = TestResultsDto.from(resultsWithCounts.results, skipResultsPerNode, skipResultsPerTransition),
         counts = resultsWithCounts.counts,
-        nodeTransitionFrequency = nodeTransitionFrequency.map(NodeTransitionFrequency.from),
+        nodeTransitionThroughput = nodeTransitionThroughput.map(NodeTransitionThroughput.from),
       )
     }
 
@@ -260,17 +260,17 @@ object Dtos {
       results: List[ResultContext[Json]]
   )
 
-  final case class NodeTransitionFrequencyDto(
+  final case class NodeTransitionThroughputDto(
       sourceNodeId: String,
       destinationNodeId: Option[String],
-      frequency: BigDecimal,
+      throughput: BigDecimal,
   )
 
-  object NodeTransitionFrequency {
+  object NodeTransitionThroughput {
 
-    def from(nodeTransitionFrequency: Map[NodeTransition, BigDecimal]): List[NodeTransitionFrequencyDto] = {
-      nodeTransitionFrequency.map { case (k, v) =>
-        NodeTransitionFrequencyDto(k.sourceNodeId, k.destinationNodeId, v)
+    def from(nodeTransitionThroughput: Map[NodeTransition, BigDecimal]): List[NodeTransitionThroughputDto] = {
+      nodeTransitionThroughput.map { case (k, v) =>
+        NodeTransitionThroughputDto(k.sourceNodeId, k.destinationNodeId, v)
       }.toList
     }
 
@@ -284,7 +284,7 @@ object Dtos {
   implicit def nodeTransitionResultSchema: Schema[NodeTransitionResult]                   = Schema.derived
   implicit def testResultsSchema: Schema[TestResultsDto]                                  = Schema.derived
   implicit def nodeCountSchema: Schema[NodeCount]                                         = Schema.anyObject
-  implicit def nodeTransitionFrequencyDtoSchema: Schema[NodeTransitionFrequencyDto]       = Schema.derived
+  implicit def nodeTransitionThroughputDtoSchema: Schema[NodeTransitionThroughputDto]     = Schema.derived
   implicit def resultsWithCountsSchema: Schema[ResultsWithCountsDto]                      = Schema.derived
   implicit def typingResultDecoder: Decoder[TypingResult] = Decoder.decodeJson.map(_ => typing.Unknown)
   implicit def scenarioGraphSchema: Schema[ScenarioGraph] = Schema.anyObject

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

@@ -195,10 +195,10 @@ class ScenarioTestingApiEndpoints(auth: EndpointInput[AuthCredentials]) extends
     Any
   ] =
     baseNuApiEndpoint
-      .summary("Perform test")
-      .tag("Testing")
+      .summary("Preview of the data samples currently processed by the scenario")
+      .tag("Live data")
       .get
-      .in("scenarioTesting" / path[ProcessName]("scenarioName") / "liveData")
+      .in("liveData" / path[ProcessName]("scenarioName"))
       .in(skipResultsPerNodeQueryParam)
       .in(skipResultsPerTransitionQueryParam)
       .out(statusCode(Ok).and(jsonBody[ResultsWithCountsDto]))

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

@@ -9,8 +9,8 @@ import pl.touk.nussknacker.engine.testmode.TestProcess.{
   ResultContext
 }
 import pl.touk.nussknacker.ui.api.description.scenarioTesting.Dtos.{
-  NodeTransitionFrequencyDto,
   NodeTransitionResult,
+  NodeTransitionThroughputDto,
   ResultsWithCountsDto,
   TestResultsDto
 }
@@ -60,7 +60,7 @@ object TestResultsCodecs {
 
   }
 
-  implicit val nodeTransitionFrequencyDto: Encoder[NodeTransitionFrequencyDto] = deriveConfiguredEncoder
+  implicit val nodeTransitionThroughputDto: Encoder[NodeTransitionThroughputDto] = deriveConfiguredEncoder
 
   implicit val testResultsDecoder: Decoder[TestResultsDto] =
     Decoder.failed(DecodingFailure("Not implemented", List.empty))

designer/server/src/test/scala/pl/touk/nussknacker/ui/api/testing/ScenarioTestingApiHttpServiceSpec.scala

@@ -449,7 +449,7 @@ trait ScenarioTestingApiHttpServiceSpec
         }
         .when()
         .basicAuthAllPermUser()
-        .get(s"$nuDesignerHttpAddress/api/scenarioTesting/${exampleScenario.name}/liveData")
+        .get(s"$nuDesignerHttpAddress/api/liveData/${exampleScenario.name}")
         .Then()
         .statusCode(StatusCodes.OK.intValue)
         .body(

docs-internal/api/nu-designer-openapi.yaml

@@ -4574,12 +4574,12 @@ paths:
       security:
       - {}
       - httpAuth: []
-  /api/scenarioTesting/{scenarioName}/liveData:
+  /api/liveData/{scenarioName}:
     get:
       tags:
-      - Testing
-      summary: Perform test
-      operationId: getApiScenariotestingScenarionameLivedata
+      - Live data
+      summary: Preview of the data samples currently processed by the scenario
+      operationId: getApiLivedataScenarioname
       parameters:
       - name: Nu-Impersonate-User-Identity
         in: header
@@ -6597,37 +6597,37 @@ components:
             type: string
           type:
             $ref: '#/components/schemas/NodeTypes12'
-    NodeTransitionFrequencyDto:
-      title: NodeTransitionFrequencyDto
+    NodeTransitionResult:
+      title: NodeTransitionResult
       type: object
       required:
       - sourceNodeId
-      - frequency
       properties:
         sourceNodeId:
           type: string
         destinationNodeId:
           type:
           - string
           - 'null'
-        frequency:
-          type: number
-    NodeTransitionResult:
-      title: NodeTransitionResult
+        results:
+          type: array
+          items:
+            $ref: '#/components/schemas/ResultContext_Json'
+    NodeTransitionThroughputDto:
+      title: NodeTransitionThroughputDto
       type: object
       required:
       - sourceNodeId
+      - throughput
       properties:
         sourceNodeId:
           type: string
         destinationNodeId:
           type:
           - string
           - 'null'
-        results:
-          type: array
-          items:
-            $ref: '#/components/schemas/ResultContext_Json'
+        throughput:
+          type: number
     NodeTypes:
       title: NodeTypes
       type: string
@@ -7104,12 +7104,12 @@ components:
           $ref: '#/components/schemas/TestResultsDto'
         counts:
           $ref: '#/components/schemas/Map_NodeCount'
-        nodeTransitionFrequency:
+        nodeTransitionThroughput:
           type:
           - array
           - 'null'
           items:
-            $ref: '#/components/schemas/NodeTransitionFrequencyDto'
+            $ref: '#/components/schemas/NodeTransitionThroughputDto'
     RunDeploymentRequest:
       title: RunDeploymentRequest
       type: object

docs/Changelog.md

@@ -177,13 +177,13 @@ description: Stay informed with detailed changelogs covering new features, impro
 * [#8042](https://github.com/TouK/nussknacker/pull/8042) Merge OpenAPI components into one with multiple services.
 * [#8047](https://github.com/TouK/nussknacker/pull/8047) Added functionality of collecting live data samples and node transition frequencies
     * live data preview is optional and available for now only for Flink minicluster 
-    * there is a new endpoint `/scenarioTesting/{scenarioName}/liveData`, which returns live data samples and frequencies
+    * there is a new endpoint `/liveData/{scenarioName}`, which returns live data samples and throughput information
     * the functionality can be configured by setting in the 'modelConfig' section of the scenario type:
       ```hocon
       liveDataPreview {              // optional config section, functionality disabled by default
         enabled: true                // disabled by default
         maxNumberOfSamples: 20       // max number of latest live data samples that will be returned
-        frequencyWindowInSeconds: 60 // the time windows, for which the node transition frequency will be calculated
+        throughputTimeWindowInSeconds: 60 // the time windows, for which the node transition throughput will be calculated
       }
       ```
 

engine/flink/management/src/it/resources/application.conf

@@ -10,6 +10,6 @@ modelConfig {
   liveDataPreview {
     enabled: true
     maxNumberOfSamples: 20
-    frequencyWindowInSeconds: 60
+    throughputTimeWindowInSeconds: 60
   }
 }

engine/flink/management/src/it/scala/pl/touk/nussknacker/engine/management/streaming/BaseFlinkDeploymentManagerSpec.scala

@@ -81,7 +81,7 @@ trait BaseFlinkDeploymentManagerSpec extends AnyFunSuiteLike with Matchers with
     LiveDataCollectingListenerHolder.createListenerFor(
       processName = processName,
       maxNumberOfSamples = 20,
-      frequencyWindowInSeconds = 60
+      throughputTimeWindowInSeconds = 60
     )
     val externalDeploymentIdOpt = deployProcessAndWaitIfRunning(
       process = process,

engine/flink/management/src/main/scala/pl/touk/nussknacker/engine/management/jobrunner/FlinkMiniClusterScenarioJobRunner.scala

@@ -52,12 +52,12 @@ class FlinkMiniClusterScenarioJobRunner(
           modelDataProvider.getCurrentModelData().modelConfig.liveDataPreviewMode match {
             case LiveDataPreviewMode.Disabled =>
               None
-            case LiveDataPreviewMode.Enabled(maxNumberOfSamples, frequencyWindowInSeconds) =>
+            case LiveDataPreviewMode.Enabled(maxNumberOfSamples, throughputTimeWindowInSeconds) =>
               Some(
                 LiveDataCollectingListenerHolder.createListenerFor(
                   command.processVersion.processName,
                   maxNumberOfSamples,
-                  frequencyWindowInSeconds
+                  throughputTimeWindowInSeconds
                 )
               )
           }

engine/flink/management/src/main/scala/pl/touk/nussknacker/engine/management/jobrunner/livedata/LiveDataCollectingListener.scala

@@ -12,7 +12,7 @@ import scala.util.Try
 class LiveDataCollectingListener private[livedata] (
     processName: ProcessName,
     maxNumberOfSamples: Int,
-    frequencyWindowInSeconds: Int,
+    throughputTimeWindowInSeconds: Int,
 ) extends ProcessListener
     with Serializable {
 
@@ -21,7 +21,7 @@ class LiveDataCollectingListener private[livedata] (
   private def storage = LiveDataCollectingListenerHolder.storage(
     processName = processName,
     maxNumberOfSamples = maxNumberOfSamples,
-    frequencyWindowInSeconds = frequencyWindowInSeconds,
+    throughputTimeWindowInSeconds = throughputTimeWindowInSeconds,
   )
 
   override def nodeEntered(nodeId: String, context: Context, processMetaData: MetaData): Unit = {

engine/flink/management/src/main/scala/pl/touk/nussknacker/engine/management/jobrunner/livedata/LiveDataCollectingListenerHolder.scala

@@ -17,12 +17,12 @@ object LiveDataCollectingListenerHolder {
   def createListenerFor(
       processName: ProcessName,
       maxNumberOfSamples: Int,
-      frequencyWindowInSeconds: Int,
+      throughputTimeWindowInSeconds: Int,
   ): LiveDataCollectingListener = {
     // We want to store and present only the live data from the current deployment,
     // so when we start a new job, we discard all old data
     cleanResults(processName)
-    new LiveDataCollectingListener(processName, maxNumberOfSamples, frequencyWindowInSeconds)
+    new LiveDataCollectingListener(processName, maxNumberOfSamples, throughputTimeWindowInSeconds)
   }
 
   def getLiveDataPreview(processName: ProcessName): Option[LiveDataPreview] = {
@@ -36,11 +36,13 @@ object LiveDataCollectingListenerHolder {
   private[livedata] def storage(
       processName: ProcessName,
       maxNumberOfSamples: Int,
-      frequencyWindowInSeconds: Int,
+      throughputTimeWindowInSeconds: Int,
   ): LiveDataCollectingListenerStorage = {
     listenerStorages.get(
       processName.value,
-      asJavaFunction((_: String) => new LiveDataCollectingListenerStorage(maxNumberOfSamples, frequencyWindowInSeconds))
+      asJavaFunction((_: String) =>
+        new LiveDataCollectingListenerStorage(maxNumberOfSamples, throughputTimeWindowInSeconds)
+      )
     )
   }
 

engine/flink/management/src/main/scala/pl/touk/nussknacker/engine/management/jobrunner/livedata/LiveDataCollectingListenerStorage.scala

@@ -12,7 +12,7 @@ import scala.jdk.CollectionConverters._
 
 private[livedata] class LiveDataCollectingListenerStorage(
     maxNumberOfSamples: Int,
-    frequencyWindowInSeconds: Int,
+    throughputTimeWindowInSeconds: Int,
 ) {
 
   private type K = String
@@ -65,7 +65,7 @@ private[livedata] class LiveDataCollectingListenerStorage(
     TestResults.aggregate(orderOfResults.asScala.toList.flatMap(key => Option(results.get(key))))
 
   private def getTransitionFrequencies: Map[NodeTransition, BigDecimal] = {
-    val cutoff                  = Instant.now().getEpochSecond - frequencyWindowInSeconds
+    val cutoff                  = Instant.now().getEpochSecond - throughputTimeWindowInSeconds
     val oldestSampleEpochSecond = transitionsByEpochSecond.keys().asScala.min
     val newestSampleEpochSecond = transitionsByEpochSecond.keys().asScala.max
     val samplingInterval        = newestSampleEpochSecond - oldestSampleEpochSecond + 1
@@ -83,7 +83,7 @@ private[livedata] class LiveDataCollectingListenerStorage(
   }
 
   private def cleanOldTransitions(currentEpochSecond: Long): Unit = {
-    val cutoff = currentEpochSecond - frequencyWindowInSeconds
+    val cutoff = currentEpochSecond - throughputTimeWindowInSeconds
     transitionsByEpochSecond
       .keySet()
       .asScala

nussknacker-dist/src/universal/conf/dev-application.conf

@@ -134,7 +134,7 @@ scenarioTypes {
       liveDataPreview {
         enabled: true
         maxNumberOfSamples: 20
-        frequencyWindowInSeconds: 60
+        throughputTimeWindowInSeconds: 60
       }
     }