Deployment manager stats + source field (#3781)

.run/NussknackerApp.run.xml

@@ -15,6 +15,7 @@
       <env name="SCHEMA_REGISTRY_URL" value="http://localhost:3082" />
       <env name="SQL_ENRICHER_URL" value="localhost:5432" />
       <env name="USAGE_REPORTS_FINGERPRINT" value="development" />
+      <env name="USAGE_REPORTS_SOURCE" value="sources" />
     </envs>
     <option name="INCLUDE_PROVIDED_SCOPE" value="true" />
     <option name="MAIN_CLASS_NAME" value="pl.touk.nussknacker.ui.NussknackerApp" />
@@ -31,4 +32,4 @@
       <option name="Make" enabled="true" />
     </method>
   </configuration>
-</component>
+</component>

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

@@ -11,6 +11,8 @@ import sttp.client.{NothingT, SttpBackend}
 
 import scala.concurrent.{ExecutionContext, Future}
 
+// If you are adding a new DeploymentManagerProvider available in the public distribution, please remember
+// to add it's type to UsageStatisticsHtmlSnippet.knownDeploymentManagerTypes
 trait DeploymentManagerProvider extends NamedServiceProvider {
 
   def createDeploymentManager(modelData: BaseModelData, config: Config)
@@ -48,4 +50,4 @@ case class FixedTypeSpecificInitialData(fixedForScenario: ScenarioSpecificData,
 
   override def forFragment(scenarioName: ProcessName, scenarioType: String): FragmentSpecificData = fixedForFragment
 
-}
+}

designer/runServer.sh

@@ -17,6 +17,7 @@ export CONSOLE_THRESHOLD_LEVEL=DEBUG
 export OPENAPI_SERVICE_URL="http://localhost:5000"
 export SQL_ENRICHER_URL="localhost:5432"
 export USAGE_REPORTS_FINGERPRINT="development"
+export USAGE_REPORTS_SOURCE="sources"
 
 USE_DOCKER_ENV=${USE_DOCKER_ENV:-true}
 

designer/server/src/main/resources/defaultDesignerConfig.conf

@@ -201,4 +201,5 @@ notifications {
 usageStatisticsReports {
   enabled: true
   fingerprint: ${?USAGE_REPORTS_FINGERPRINT}
-}
+  source: ${?USAGE_REPORTS_SOURCE}
+}

designer/server/src/main/scala/pl/touk/nussknacker/engine/ProcessingTypeData.scala

@@ -6,6 +6,7 @@ import com.typesafe.config.Config
 import pl.touk.nussknacker.engine.api.component.AdditionalPropertyConfig
 import pl.touk.nussknacker.engine.api.deployment.{DeploymentManager, ProcessingTypeDeploymentService}
 import pl.touk.nussknacker.engine.api.queryablestate.QueryableClient
+import pl.touk.nussknacker.ui.statistics.ProcessingTypeUsageStatistics
 import pl.touk.nussknacker.ui.validation.AdditionalPropertiesValidator
 
 import scala.concurrent.{ExecutionContext, Future}
@@ -16,7 +17,8 @@ case class ProcessingTypeData(deploymentManager: DeploymentManager,
                               additionalPropertiesConfig: Map[String, AdditionalPropertyConfig],
                               additionalValidators: List[CustomProcessValidator],
                               queryableClient: Option[QueryableClient],
-                              supportsSignals: Boolean) extends AutoCloseable {
+                              supportsSignals: Boolean,
+                              usageStatistics: ProcessingTypeUsageStatistics) extends AutoCloseable {
 
   def close(): Unit = {
     modelData.close()
@@ -46,7 +48,8 @@ object ProcessingTypeData {
       additionalProperties,
       deploymentManagerProvider.additionalValidators(managerConfig) ,
       queryableClient,
-      deploymentManagerProvider.supportsSignals)
+      deploymentManagerProvider.supportsSignals,
+      ProcessingTypeUsageStatistics(managerConfig))
   }
 
   def createProcessingTypeData(deploymentManagerProvider: DeploymentManagerProvider, processTypeConfig: ProcessingTypeConfig)

designer/server/src/main/scala/pl/touk/nussknacker/ui/NussknackerApp.scala

@@ -212,7 +212,7 @@ trait NusskanckerDefaultAppRouter extends NusskanckerAppRouter {
     )
 
     val usageStatisticsReportsConfig = config.as[UsageStatisticsReportsConfig]("usageStatisticsReports")
-    val usageStatisticsSnippetOpt = UsageStatisticsHtmlSnippet.prepareWhenEnabledReporting(usageStatisticsReportsConfig)
+    val usageStatisticsSnippetOpt = UsageStatisticsHtmlSnippet.prepareWhenEnabledReporting(usageStatisticsReportsConfig, typeToConfig.mapValues(_.usageStatistics))
 
     //TODO: In the future will be nice to have possibility to pass authenticator.directive to resource and there us it at concrete path resource
     val webResources = new WebResources(config.getString("http.publicPath"), usageStatisticsSnippetOpt)

designer/server/src/main/scala/pl/touk/nussknacker/ui/config/UsageStatisticsReportsConfig.scala

@@ -1,3 +1,7 @@
 package pl.touk.nussknacker.ui.config
 
-case class UsageStatisticsReportsConfig(enabled: Boolean, fingerprint: Option[String])
+case class UsageStatisticsReportsConfig(enabled: Boolean,
+                                        // unique identifier for Designer installation
+                                        fingerprint: Option[String],
+                                        // source from which Nussknacker was downloaded
+                                        source: Option[String])

designer/server/src/main/scala/pl/touk/nussknacker/ui/statistics/ProcessingTypeUsageStatistics.scala

@@ -0,0 +1,14 @@
+package pl.touk.nussknacker.ui.statistics
+
+import com.typesafe.config.Config
+import net.ceedubs.ficus.Ficus._
+
+case class ProcessingTypeUsageStatistics(deploymentManagerType: String, processingMode: Option[String])
+
+object ProcessingTypeUsageStatistics {
+  // TODO: handle only enabled managers by category configuration
+  def apply(managerConfig: Config): ProcessingTypeUsageStatistics =
+    ProcessingTypeUsageStatistics(
+      managerConfig.getString("type"),
+      managerConfig.getAs[String]("mode"))
+}

designer/server/src/main/scala/pl/touk/nussknacker/ui/statistics/UsageStatisticsHtmlSnippet.scala

@@ -1,7 +1,9 @@
 package pl.touk.nussknacker.ui.statistics
 
 import pl.touk.nussknacker.engine.version.BuildInfo
+import pl.touk.nussknacker.restmodel.process.ProcessingType
 import pl.touk.nussknacker.ui.config.UsageStatisticsReportsConfig
+import pl.touk.nussknacker.ui.process.processingtypedata.ProcessingTypeDataProvider
 
 import java.net.URLEncoder
 import java.nio.charset.StandardCharsets
@@ -12,25 +14,68 @@ case class UsageStatisticsHtmlSnippet(value: String)
 
 object UsageStatisticsHtmlSnippet {
 
-  def prepareWhenEnabledReporting(config: UsageStatisticsReportsConfig): Option[UsageStatisticsHtmlSnippet] = {
+  private val knownDeploymentManagerTypes = Set("flinkStreaming", "lite-k8s", "lite-embedded")
+
+  private val streamingProcessingMode = "streaming"
+
+  private val knownProcessingModes = Set(streamingProcessingMode, "request-response")
+
+  // We aggregate custom deployment managers and processing modes as a "custom" to avoid leaking of internal, confidential data
+  private val aggregateForCustomValues = "custom"
+
+  def prepareWhenEnabledReporting(config: UsageStatisticsReportsConfig,
+                                  processingTypeStatistics: ProcessingTypeDataProvider[ProcessingTypeUsageStatistics]): Option[UsageStatisticsHtmlSnippet] = {
     if (config.enabled) {
-      val queryParams = prepareQueryParams(config)
+      val queryParams = prepareQueryParams(config, processingTypeStatistics.all)
       val url = prepareUrl(queryParams)
       Some(UsageStatisticsHtmlSnippet(s"""<img src="$url" alt="anonymous usage reporting" referrerpolicy="origin" hidden />"""))
     } else {
       None
     }
   }
 
-  private def prepareQueryParams(config: UsageStatisticsReportsConfig) = {
+  private[statistics] def prepareQueryParams(config: UsageStatisticsReportsConfig,
+                                             processingTypeStatisticsMap: Map[ProcessingType, ProcessingTypeUsageStatistics]): ListMap[String, String] = {
+    val deploymentManagerTypes = processingTypeStatisticsMap.values.map(_.deploymentManagerType).map {
+      case dm if knownDeploymentManagerTypes.contains(dm) => dm
+      case _ => aggregateForCustomValues
+    }
+    val dmParams = prepareValuesParams(deploymentManagerTypes, "dm")
+
+    val processingModes = processingTypeStatisticsMap.values.map {
+      case ProcessingTypeUsageStatistics(_, Some(mode)) if knownProcessingModes.contains(mode) => mode
+      case ProcessingTypeUsageStatistics(deploymentManagerType, None) if deploymentManagerType.toLowerCase.contains(streamingProcessingMode) => streamingProcessingMode
+      case _ => aggregateForCustomValues
+    }
+    val mParams = prepareValuesParams(processingModes, "m")
+
     ListMap(
-      "fingerprint" -> config.fingerprint.getOrElse(randomFingerprint),
-      "version" -> BuildInfo.version)
+      // We filter out blank fingerprint and source because when smb uses docker-compose, and forwards env variables eg. USAGE_REPORTS_FINGERPRINT
+      // from system and the variable doesn't exist, there is no way to skip variable - it can be only set to empty
+      "fingerprint" -> config.fingerprint.filterNot(_.isBlank).getOrElse(randomFingerprint),
+      // If it is not set, we assume that it is some custom build from source code
+      "source" -> config.source.filterNot(_.isBlank).getOrElse("sources"),
+      "version" -> BuildInfo.version
+    ) ++ dmParams ++ mParams
+  }
+
+  private def prepareValuesParams(values: Iterable[ProcessingType], metricCategoryKeyPart: String) = {
+    val countsParams = values.groupBy(identity).mapValues(_.size).map {
+      case (value, count) =>
+        s"${metricCategoryKeyPart}_$value" -> count.toString
+    }.toList.sortBy(_._1)
+    val singleParamValue = values.toSet.toList match {
+      case Nil => "zero"
+      case single :: Nil => single
+      case _ => "multiple"
+    }
+    ListMap(countsParams: _*) + (s"single_$metricCategoryKeyPart" -> singleParamValue)
   }
 
   private[statistics] def prepareUrl(queryParams: ListMap[String, String]) = {
-    val queryParamsPart = queryParams.toList.map { case (k, v) => s"$k=${URLEncoder.encode(v, StandardCharsets.UTF_8)}" }.mkString("&")
-    s"https://stats.nussknacker.io/?$queryParamsPart"
+    queryParams.toList.map {
+      case (k, v) => s"${URLEncoder.encode(k, StandardCharsets.UTF_8)}=${URLEncoder.encode(v, StandardCharsets.UTF_8)}"
+    }.mkString("https://stats.nussknacker.io/?", "&", "")
   }
 
   private lazy val randomFingerprint = s"gen-${Random.alphanumeric.take(10).mkString}"

designer/server/src/test/resources/designer.conf

@@ -183,4 +183,5 @@ notifications {
 usageStatisticsReports {
   enabled: true
   fingerprint: ${?USAGE_REPORTS_FINGERPRINT}
-}
+  source: ${?USAGE_REPORTS_SOURCE}
+}

designer/server/src/test/scala/pl/touk/nussknacker/ui/component/DefaultComponentServiceSpec.scala

@@ -29,6 +29,7 @@ import pl.touk.nussknacker.ui.process.ProcessCategoryService.Category
 import pl.touk.nussknacker.ui.process.processingtypedata.MapBasedProcessingTypeDataProvider
 import pl.touk.nussknacker.ui.process.{ConfigProcessCategoryService, DBProcessService, ProcessCategoryService}
 import pl.touk.nussknacker.ui.security.api.LoggedUser
+import pl.touk.nussknacker.ui.statistics.ProcessingTypeUsageStatistics
 import sttp.client.{NothingT, SttpBackend}
 
 import java.time.Duration
@@ -369,7 +370,8 @@ class DefaultComponentServiceSpec extends AnyFlatSpec with Matchers with Patient
       Map.empty,
       Nil,
       None,
-      supportsSignals = false)
+      supportsSignals = false,
+      ProcessingTypeUsageStatistics("stubManager", None))
   })
 
   it should "return components for each user" in {
@@ -445,7 +447,8 @@ class DefaultComponentServiceSpec extends AnyFlatSpec with Matchers with Patient
         Map.empty,
         Nil,
         None,
-        supportsSignals = false)
+        supportsSignals = false,
+        ProcessingTypeUsageStatistics("stubManager", None))
     })
 
     val processService = createDbProcessService(categoryService, List(MarketingProcess))

designer/server/src/test/scala/pl/touk/nussknacker/ui/statistics/UsageStatisticsHtmlSnippetTest.scala

@@ -2,12 +2,89 @@ package pl.touk.nussknacker.ui.statistics
 
 import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
+import pl.touk.nussknacker.engine.version.BuildInfo
+import pl.touk.nussknacker.ui.config.UsageStatisticsReportsConfig
 
 import scala.collection.immutable.ListMap
 
-class UsageStatisticsHtmlSnippetTest extends AnyFunSuite with Matchers{
+class UsageStatisticsHtmlSnippetTest extends AnyFunSuite with Matchers {
 
-  test("should generate correct url with encoded params") {
+  val sampleFingerprint = "fooFingerprint"
+
+  test("should generate correct url with encoded paramsForSingleMode") {
     UsageStatisticsHtmlSnippet.prepareUrl(ListMap("f" -> "a b", "v" -> "1.6.5-a&b=c")) shouldBe "https://stats.nussknacker.io/?f=a+b&v=1.6.5-a%26b%3Dc"
   }
+
+  test("should generated statically defined query paramsForSingleMode") {
+    val params = UsageStatisticsHtmlSnippet.prepareQueryParams(
+      UsageStatisticsReportsConfig(enabled = true, Some(sampleFingerprint), None),
+      Map.empty)
+    params should contain ("fingerprint" -> sampleFingerprint)
+    params should contain ("source" -> "sources")
+    params should contain ("version" -> BuildInfo.version)
+  }
+
+  test("should generated random fingerprint if configured is blank") {
+    val params = UsageStatisticsHtmlSnippet.prepareQueryParams(
+      UsageStatisticsReportsConfig(enabled = true, Some(""), None),
+      Map.empty)
+    params("fingerprint") should startWith ("gen-")
+  }
+
+  test("should generated query params for each deployment manager and with single deployment manager field") {
+    val givenDm1 = "flinkStreaming"
+    val paramsForSingleDm = UsageStatisticsHtmlSnippet.prepareQueryParams(
+      UsageStatisticsReportsConfig(enabled = true, Some(sampleFingerprint), None),
+      Map("streaming" -> ProcessingTypeUsageStatistics(givenDm1, None)))
+    paramsForSingleDm should contain ("single_dm" -> givenDm1)
+    paramsForSingleDm should contain ("dm_" + givenDm1 -> "1")
+
+    val givenDm2 = "lite-k8s"
+    val paramsForMultipleDms = UsageStatisticsHtmlSnippet.prepareQueryParams(
+      UsageStatisticsReportsConfig(enabled = true, Some(sampleFingerprint), None),
+      Map(
+        "streaming" -> ProcessingTypeUsageStatistics(givenDm1, None),
+        "streaming2" -> ProcessingTypeUsageStatistics(givenDm2, None),
+        "streaming3" -> ProcessingTypeUsageStatistics(givenDm1, None)))
+    paramsForMultipleDms should contain ("single_dm" -> "multiple")
+    paramsForMultipleDms should contain ("dm_" + givenDm1 -> "2")
+    paramsForMultipleDms should contain ("dm_" + givenDm2 -> "1")
+  }
+
+  test("should generated query params for each processing mode and with single processing mode field") {
+    val streamingMode = "streaming"
+    val paramsForSingleMode = UsageStatisticsHtmlSnippet.prepareQueryParams(
+      UsageStatisticsReportsConfig(enabled = true, Some(sampleFingerprint), None),
+      Map("streaming" -> ProcessingTypeUsageStatistics("fooDm", Some(streamingMode))))
+    paramsForSingleMode should contain ("single_m" -> streamingMode)
+    paramsForSingleMode should contain ("m_" + streamingMode -> "1")
+
+    val requestResponseMode = "request-response"
+    val paramsForMultipleModes = UsageStatisticsHtmlSnippet.prepareQueryParams(
+      UsageStatisticsReportsConfig(enabled = true, Some(sampleFingerprint), None),
+      Map(
+        "streaming" -> ProcessingTypeUsageStatistics("fooDm", Some(streamingMode)),
+        "streaming2" -> ProcessingTypeUsageStatistics("barDm", Some(requestResponseMode)),
+        "streaming3" -> ProcessingTypeUsageStatistics("bazDm", Some(streamingMode))))
+    paramsForMultipleModes should contain ("single_m" -> "multiple")
+    paramsForMultipleModes should contain ("m_" + streamingMode -> "2")
+    paramsForMultipleModes should contain ("m_" + requestResponseMode -> "1")
+  }
+
+  test("should aggregate unknown deployment manager and processing mode as a custom") {
+    val givenCustomDm = "customDm"
+    val paramsForSingleDm = UsageStatisticsHtmlSnippet.prepareQueryParams(
+      UsageStatisticsReportsConfig(enabled = true, Some(sampleFingerprint), None),
+      Map("streaming" -> ProcessingTypeUsageStatistics(givenCustomDm, None)))
+    paramsForSingleDm should contain("single_dm" -> "custom")
+    paramsForSingleDm should contain("dm_custom" -> "1")
+
+    val customMode = "customMode"
+    val paramsForSingleMode = UsageStatisticsHtmlSnippet.prepareQueryParams(
+      UsageStatisticsReportsConfig(enabled = true, Some(sampleFingerprint), None),
+      Map("streaming" -> ProcessingTypeUsageStatistics("fooDm", Some(customMode))))
+    paramsForSingleMode should contain ("single_m" -> "custom")
+    paramsForSingleMode should contain ("m_custom" -> "1")
+  }
+
 }

nussknacker-dist/src/universal/bin/nussknacker-entrypoint.sh

@@ -35,6 +35,10 @@ else
   JAVA_PROMETHEUS_OPTS="-javaagent:$agentPath=$PROMETHEUS_METRICS_PORT:$PROMETHEUS_AGENT_CONFIG_FILE"
 fi
 
+if [ "$USAGE_REPORTS_SOURCE" == "" ]; then
+  export USAGE_REPORTS_SOURCE="docker"
+fi
+
 mkdir -p ${STORAGE_DIR}/db
 
 echo "Starting Nussknacker:"

nussknacker-dist/src/universal/bin/run.sh

@@ -43,6 +43,8 @@ else
   JAVA_PROMETHEUS_OPTS="-javaagent:$agentPath=$PROMETHEUS_METRICS_PORT:$PROMETHEUS_AGENT_CONFIG_FILE"
 fi
 
+export USAGE_REPORTS_SOURCE="binaries"
+
 mkdir -p $LOGS_DIR
 cd $WORKING_DIR