Live data preview and node transition frequency on Flink minicluster #8047

mgoworko · 2025-05-07T16:52:25Z

Describe your changes

added listener, that collects live data processed by the scenario
the data is stored in a ConcurrentHashMap-based ring buffer, samples corresponding to the last N contextIds are stored
also added node transition frequency calculation - node transition counts are stored in buckets for each epoch second and calculated in time windows

The data can be fetched using /api/scenarioTesting/{scenario_name}/liveData

At the moment only for Flink minicluster, because it is run a part of the JVM and we can pass listener to the job.

Checklist before merge

Related issue ID is placed at the beginning of PR title in [brackets] (can be GH issue or Nu Jira issue)
Code is cleaned from temporary changes and commented out lines
Parts of the code that are not easy to understand are documented in the code
Changes are covered by automated tests
Showcase in dev-application.conf added to demonstrate the feature
Documentation added or updated
Added entry in Changelog.md describing the change from the perspective of a public distribution user
Added MigrationGuide.md entry in the appropriate subcategory if introducing a breaking change
Verify that PR will be squashed during merge

# Conflicts: # engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/StubbedFlinkProcessCompilerDataFactory.scala

# Conflicts: # components-api/src/main/scala/pl/touk/nussknacker/engine/ModelConfig.scala # engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/FlinkProcessCompilerDataFactory.scala

# Conflicts: # docs/Changelog.md

mgoworko · 2025-05-20T12:57:14Z

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

 * @param id            correlation id/trace id used for tracing (logs, error presentation) and for tests mechanism, it should be always defined
 * @param variables     variables available in evaluation
 * @param parentContext context used for scopes handling, mainly for fragment invocation purpose
 */
 case class Context(
+    initialId: String,


This change will no longer be needed after the ContextId refactor. The initialId is basically ContextId without prefix.

So let's add a TODO/FIXME

I removed the initialId field. Instead of it, the initial context id is now recognized by regex. There is a todo added about it. After the ContextId refactor the regex won't be necessary, because we will be able co compare parts of the ContextId case class instead.

arkadius · 2025-05-20T10:43:24Z

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

+
+  final case class LiveDataPreview(
+      liveDataSamples: TestResults[Json],
+      nodeTransitionFrequency: Map[NodeTransition, BigDecimal],


I think that using "throughput" word is more natural in this context. The "frequency" is used when we talk about some cyclic mechanism to show how often it is triggered.

arkadius · 2025-05-20T10:46:13Z

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

+
+  def getLiveData(
+      processIdWithName: ProcessIdWithName,
+  ): Future[Option[LiveDataPreview]]


This Option is mysterious. It could be interpreted in many ways. Let's use Either with explicit error instead. For deployed scenarios, we should always return defined, empty LiveDataPreview. The error should be returned when someone ask for scenario that is not deployed or is finished/stopped.

Done, either with error in Left. The error only says that the information is not available for the scenario. In that place we do not have detailed information as to why.

arkadius · 2025-05-20T13:07:23Z

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

 }

 case class ContextId(value: String)

 /**
 * Context is container for variables used in expression evaluation
 *
+ * @param initialId     the initial id of the context (id may change during the processing, but the initial id is preserved in this field


Can you add one more sentence in the comment, why do we need this field with preserved id? Also, maybe we could replace case class with normal class to make changes of initialId statically impossible?

I removed this field, details in the other comment.

arkadius · 2025-05-20T13:14:53Z

...ala/pl/touk/nussknacker/ui/api/description/scenarioTesting/ScenarioTestingApiEndpoints.scala

+    Any
+  ] =
+    baseNuApiEndpoint
+      .summary("Perform test")


What test this endpoint perform? :P

arkadius · 2025-05-20T13:15:47Z

...ala/pl/touk/nussknacker/ui/api/description/scenarioTesting/ScenarioTestingApiEndpoints.scala

+      .summary("Perform test")
+      .tag("Testing")
+      .get
+      .in("scenarioTesting" / path[ProcessName]("scenarioName") / "liveData")


Why endpoint is nested in scenarioTesing resource? Let's create a new liveData endpoint. I know that under the hood we reuse some features from the scenario testing mechanism, but we don't need to expose it in the interface.

arkadius · 2025-05-20T13:36:58Z

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

 * @param id            correlation id/trace id used for tracing (logs, error presentation) and for tests mechanism, it should be always defined
 * @param variables     variables available in evaluation
 * @param parentContext context used for scopes handling, mainly for fragment invocation purpose
 */
 case class Context(
+    initialId: String,


So let's add a TODO/FIXME

arkadius · 2025-05-20T13:41:33Z

...touk/nussknacker/engine/management/jobrunner/livedata/LiveDataCollectingListenerHolder.scala

+    Option(listenerStorages.getIfPresent(processName.value)).map(_.getLiveDataPreview)
+  }
+
+  private[livedata] def cleanResults(processName: ProcessName): Unit = {


I assume that you don't clean results after the job finish, because you want to make it still available for the user for some period of time? Maybe let's write a comment about that because from the first glance it looks like some memory leak.

done, comment is now above this method and describes when we want to clean the data

arkadius · 2025-05-20T13:51:52Z

...ouk/nussknacker/engine/management/jobrunner/livedata/LiveDataCollectingListenerStorage.scala

+    val isNewKey     = !results.containsKey(key)
+    if (isNewKey) orderOfResults.add(key)
+    results.put(key, newValue)
+    while (orderOfResults.size() > maxNumberOfSamples) {


size has O(n) complexity. Have you considered using the ready solution such as EvictingQueue from guave? we do synchronisation, so I can't see too much added value from using thread-safe structures. Keeping two information in two data structures: results and orderOfResults is hard to analyze in code, e.g. as a reviewer I can't guratanee that it is correctly cleaned.

Unfortunately guava EvictingQueue won't work here. It does not support modifying elements by key. But my solution was not good either.

I changed the solution. I created RingBuffer util, with unit tests. It is based on the LinkedHashMap, with added synchronization.

arkadius · 2025-05-20T14:17:48Z

...ouk/nussknacker/engine/management/jobrunner/livedata/LiveDataCollectingListenerStorage.scala

+
+  private val results                             = new ConcurrentHashMap[K, V]()
+  private val orderOfResults                      = new ConcurrentLinkedQueue[K]()
+  private val transitionsByEpochSecond            = new ConcurrentHashMap[Long, ConcurrentLinkedQueue[NodeTransition]]()


Maybe we could just use a MultiMap to make it simpler? Or a Map NodeTransition ->InstantRateMeter?

Unfortunately there is no ready-to-use util to do exactly what is needed here. I asked LLM's about it too :) . (The main difficulty is that we have to drop the transitions, that are older than the treshold)

I created an util class SlidingWindowCounter. It wraps the logic around it and I think it is quite simple and easy to understand. I also added unit tests of that util.

# Conflicts: # docs/Changelog.md

github-actions · 2025-05-21T13:19:51Z

created: #8133
⚠️ Be careful! Snapshot changes are not necessarily the cause of the error. Check the logs.

# Conflicts: # e2e-tests/src/test/scala/pl/touk/nussknacker/BatchDataGenerationSpec.scala

# Conflicts: # designer/server/src/main/scala/pl/touk/nussknacker/ui/api/ScenarioTestingApiHttpService.scala # designer/server/src/main/scala/pl/touk/nussknacker/ui/api/description/scenarioTesting/Dtos.scala # designer/server/src/main/scala/pl/touk/nussknacker/ui/api/description/scenarioTesting/ScenarioTestingApiEndpoints.scala # designer/server/src/test/scala/pl/touk/nussknacker/ui/api/testing/ScenarioTestingApiHttpServiceSpec.scala # designer/server/src/test/scala/pl/touk/nussknacker/ui/api/testing/SchemalessKafkaJsonTypeTests.scala # docs-internal/api/nu-designer-openapi.yaml # docs/Changelog.md

# Conflicts: # components/openapi/src/main/scala/pl/touk/nussknacker/openapi/enrichers/OpenAPIEnricherFactory.scala

mgoworko · 2025-05-23T18:10:45Z

build.sbt

@@ -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"))


I created a new module, reusable by other DMs, instead of adding it in the Flink engine module.

live data draft

b7f6fe9

github-actions bot mentioned this pull request May 7, 2025

Update Cypress snapshots in live-data #8048

Closed

mgoworko added 2 commits May 12, 2025 15:48

Merge remote-tracking branch 'origin/staging' into live-data

549018e

# Conflicts: # engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/StubbedFlinkProcessCompilerDataFactory.scala

config and API refactor

caf5ad6

mgoworko marked this pull request as ready for review May 13, 2025 09:33

mgoworko added 3 commits May 13, 2025 11:34

Merge remote-tracking branch 'origin/staging' into live-data

75b40ae

# Conflicts: # components-api/src/main/scala/pl/touk/nussknacker/engine/ModelConfig.scala # engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/compiler/FlinkProcessCompilerDataFactory.scala

qs

a8a7772

fix scala 2.12

f0f3713

mgoworko changed the title ~~Draft, do not review - Flink minicluster live data preview~~ In progress - Flink minicluster live data preview May 13, 2025

mgoworko added 11 commits May 13, 2025 12:52

qs

5fab9d1

qs

29eb20a

qs

922bc1b

Merge remote-tracking branch 'origin/staging' into live-data

875ed75

tests

b18a021

tests and frequency

a353c19

tests and frequency

c63a84f

qs

8acef21

qs

c5a5483

qs

94c6242

qs

8fe11c8

github-actions bot added the docs label May 18, 2025

mgoworko changed the title ~~In progress - Flink minicluster live data preview~~ Live data preview and node transition frequency on Flink minicluster May 18, 2025

mgoworko added 3 commits May 18, 2025 19:58

qs

66f3af5

qs

07ffb12

Merge remote-tracking branch 'origin/staging' into live-data

e2e24fd

# Conflicts: # docs/Changelog.md

github-actions bot added client client main fe ui labels May 18, 2025

mgoworko added 2 commits May 19, 2025 00:05

qs

83cc4fe

qs

140a62b

TouK deleted a comment from github-actions bot May 19, 2025

mgoworko requested a review from arkadius May 19, 2025 07:58

mgoworko commented May 20, 2025

View reviewed changes

arkadius reviewed May 20, 2025

View reviewed changes

Merge remote-tracking branch 'origin/staging' into live-data

8ed4c8b

# Conflicts: # docs/Changelog.md

github-actions bot removed ui client client main fe labels May 21, 2025

qs

d1d7004

github-actions bot mentioned this pull request May 21, 2025

Update Cypress snapshots in live-data #8133

Closed

mgoworko added 10 commits May 22, 2025 15:33

Merge remote-tracking branch 'origin/staging' into live-data

fd4ce84

# Conflicts: # e2e-tests/src/test/scala/pl/touk/nussknacker/BatchDataGenerationSpec.scala

qs

8063a7f

qs

3a8111e

qs

f3bf584

review changes

35e7599

review changes

9bde15c

Merge remote-tracking branch 'origin/staging' into live-data

38bbde3

# Conflicts: # components/openapi/src/main/scala/pl/touk/nussknacker/openapi/enrichers/OpenAPIEnricherFactory.scala

review changes

68dfadf

review changes

e0ccee6

mgoworko commented May 23, 2025

View reviewed changes

mgoworko added 5 commits May 23, 2025 21:13

qs

f1e1b92

qs

47be346

qs

936f68f

qs

cbf2ef9

qs

38f530c

Live data preview and node transition frequency on Flink minicluster #8047

Are you sure you want to change the base?

Live data preview and node transition frequency on Flink minicluster #8047

Conversation

mgoworko commented May 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Describe your changes

Checklist before merge

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

arkadius May 20, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented May 21, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mgoworko commented May 7, 2025 •

edited

Loading

arkadius May 20, 2025 •

edited

Loading