build(dataflow): Add ktlint plugin in Gradle, run formatter, and fix lint issues

scheduler/data-flow/.gitignore

@@ -341,4 +341,7 @@ $RECYCLE.BIN/
 # Windows shortcuts
 *.lnk
 
+# VSCode
+.vscode/
+
 # End of https://www.gitignore.io/api/java,linux,macos,windows,android,intellij,androidstudio

scheduler/data-flow/Makefile

@@ -7,3 +7,15 @@ licenses:
 	chmod +x ./scripts/generate_license.sh
 	./scripts/generate_license.sh licenses/dependency-license.json licenses/dependency-license.txt
 	cp ../../LICENSE licenses/license.txt
+
+.PHONY: lint
+lint:
+	./gradlew ktlintCheck --no-daemon --no-build-cache --continue
+
+.PHONY: format
+format:
+	./gradlew ktlintFormat --no-daemon --no-build-cache
+
+.PHONY: build
+build:
+	./gradlew build --no-daemon --no-build-cache

scheduler/data-flow/README.md

@@ -54,6 +54,14 @@ $ ./gradlew build
 BUILD SUCCESSFUL in 536ms
 6 actionable tasks: 6 up-to-date
 ```
+You can also run lint check and formatting using `ktlint`, which are also available as make targets i.e. `make lint` and `make format`.
+
+```bash
+$ ./gradlew ktlintCheck 
+$ ./gradlew ktlintFormat
+```
+
+After a successful run, they generate reports in `build/reports/ktlint`.
 
 <details>
 <summary>Unsupported class file major version</summary>

scheduler/data-flow/build.gradle.kts

@@ -1,12 +1,12 @@
-import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
 import com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar
+import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
 
 
 plugins {
     id("com.github.hierynomus.license-report") version "0.16.1"
     id("com.github.johnrengelman.shadow") version "8.1.1"
-    kotlin("jvm") version "1.8.20"  // the kotlin version
-
+    kotlin("jvm") version "1.8.20" // the kotlin version
+    id("org.jlleitschuh.gradle.ktlint") version "12.1.0"
     java
     application
 }
@@ -98,3 +98,12 @@ downloadLicenses {
     includeProjectDependencies = true
     dependencyConfiguration = "compileClasspath"
 }
+
+ktlint {
+    verbose = true
+    debug = true
+    // Ignore generated code from proto
+    filter {
+        exclude { element -> element.file.path.contains("apis/mlops") }
+    }
+}

scheduler/data-flow/settings.gradle.kts

@@ -1,2 +1 @@
 rootProject.name = "dataflow"
-

scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/Cli.kt

@@ -9,14 +9,25 @@ the Change License after the Change Date as each is defined in accordance with t
 
 package io.seldon.dataflow
 
-import com.natpryce.konfig.*
+import com.natpryce.konfig.CommandLineOption
+import com.natpryce.konfig.Configuration
+import com.natpryce.konfig.ConfigurationProperties
+import com.natpryce.konfig.EnvironmentVariables
+import com.natpryce.konfig.Key
+import com.natpryce.konfig.booleanType
+import com.natpryce.konfig.enumType
+import com.natpryce.konfig.intType
+import com.natpryce.konfig.longType
+import com.natpryce.konfig.overriding
+import com.natpryce.konfig.parseArgs
+import com.natpryce.konfig.stringType
 import io.klogging.Level
 import io.klogging.noCoLogger
 import io.seldon.dataflow.kafka.security.KafkaSaslMechanisms
 import io.seldon.dataflow.kafka.security.KafkaSecurityProtocols
 
 object Cli {
-    private const val envVarPrefix = "SELDON_"
+    private const val ENV_VAR_PREFIX = "SELDON_"
     private val logger = noCoLogger(Cli::class)
 
     // General setup
@@ -94,18 +105,19 @@ object Cli {
 
     fun configWith(rawArgs: Array<String>): Configuration {
         val fromProperties = ConfigurationProperties.fromResource("local.properties")
-        val fromEnv = EnvironmentVariables(prefix = envVarPrefix)
+        val fromEnv = EnvironmentVariables(prefix = ENV_VAR_PREFIX)
         val fromArgs = parseArguments(rawArgs)
 
         return fromArgs overriding fromEnv overriding fromProperties
     }
 
     private fun parseArguments(rawArgs: Array<String>): Configuration {
-        val (config, unparsedArgs) = parseArgs(
-            rawArgs,
-            *this.args().map { CommandLineOption(it) }.toTypedArray(),
-            programName = "seldon-dataflow-engine",
-        )
+        val (config, unparsedArgs) =
+            parseArgs(
+                rawArgs,
+                *this.args().map { CommandLineOption(it) }.toTypedArray(),
+                programName = "seldon-dataflow-engine",
+            )
         if (unparsedArgs.isNotEmpty()) {
             logUnknownArguments(unparsedArgs)
         }

scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/DataflowStatus.kt

@@ -13,10 +13,10 @@ import kotlinx.coroutines.runBlocking
  * return values to indicate errors/status updates that require special handling in the code.
  */
 interface DataflowStatus {
-    var exception : Exception?
-    var message : String?
+    var exception: Exception?
+    var message: String?
 
-    fun getDescription() : String? {
+    fun getDescription(): String? {
         val exceptionMsg = this.exception?.message
         return if (exceptionMsg != null) {
             "${this.message} Exception: $exceptionMsg"
@@ -26,7 +26,10 @@ interface DataflowStatus {
     }
 
     // log status when logger is in a coroutine
-    fun log(logger: Klogger, levelIfNoException: Level) {
+    fun log(
+        logger: Klogger,
+        levelIfNoException: Level,
+    ) {
         val exceptionMsg = this.exception?.message
         val exceptionCause = this.exception?.cause ?: Exception("")
         val statusMsg = this.message
@@ -42,7 +45,10 @@ interface DataflowStatus {
     }
 
     // log status when logger is outside coroutines
-    fun log(logger: NoCoLogger, levelIfNoException: Level) {
+    fun log(
+        logger: NoCoLogger,
+        levelIfNoException: Level,
+    ) {
         val exceptionMsg = this.exception?.message
         val exceptionCause = this.exception?.cause ?: Exception("")
         if (exceptionMsg != null) {
@@ -53,13 +59,12 @@ interface DataflowStatus {
     }
 }
 
-fun <T: DataflowStatus> T.withException(e: Exception) : T {
+fun <T : DataflowStatus> T.withException(e: Exception): T {
     this.exception = e
     return this
 }
 
-fun <T: DataflowStatus> T.withMessage(msg: String): T {
+fun <T : DataflowStatus> T.withMessage(msg: String): T {
     this.message = msg
     return this
 }
-

scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/GrpcServiceConfigProvider.kt

@@ -15,27 +15,32 @@ object GrpcServiceConfigProvider {
     // Details: https://github.com/grpc/proposal/blob/master/A6-client-retries.md#validation-of-retrypolicy
     // Example: https://github.com/grpc/grpc-java/blob/v1.35.0/examples/src/main/resources/io/grpc/examples/retrying/retrying_service_config.json
     // However does not work: https://github.com/grpc/grpc-kotlin/issues/277
-    val config = mapOf<String, Any>(
-        "methodConfig" to listOf(
-            mapOf(
-                "name" to listOf(
+    val config =
+        mapOf<String, Any>(
+            "methodConfig" to
+                listOf(
                     mapOf(
-                        "service" to "io.seldon.mlops.chainer.Chainer",
-                        "method" to "SubscribePipelineUpdates",
+                        "name" to
+                            listOf(
+                                mapOf(
+                                    "service" to "io.seldon.mlops.chainer.Chainer",
+                                    "method" to "SubscribePipelineUpdates",
+                                ),
+                            ),
+                        "retryPolicy" to
+                            mapOf(
+                                "maxAttempts" to "100",
+                                "initialBackoff" to "1s",
+                                "maxBackoff" to "30s",
+                                "backoffMultiplier" to 1.5,
+                                "retryableStatusCodes" to
+                                    listOf(
+                                        Status.UNAVAILABLE.code.toString(),
+                                        Status.CANCELLED.code.toString(),
+                                        Status.FAILED_PRECONDITION.code.toString(),
+                                    ),
+                            ),
                     ),
                 ),
-                "retryPolicy" to mapOf(
-                    "maxAttempts" to "100",
-                    "initialBackoff" to "1s",
-                    "maxBackoff" to "30s",
-                    "backoffMultiplier" to 1.5,
-                    "retryableStatusCodes" to listOf(
-                        Status.UNAVAILABLE.code.toString(),
-                        Status.CANCELLED.code.toString(),
-                        Status.FAILED_PRECONDITION.code.toString(),
-                    )
-                )
-            ),
-        ),
-    )
-}
+        )
+}

scheduler/data-flow/src/main/kotlin/io/seldon/dataflow/Logging.kt

@@ -15,21 +15,23 @@ import io.klogging.rendering.RENDER_ISO8601
 import io.klogging.sending.STDOUT
 
 object Logging {
-    private const val stdoutSink = "stdout"
+    private const val STDOUT_SINK = "stdout"
 
-    fun configure(appLevel: Level = Level.INFO, kafkaLevel: Level = Level.WARN) =
-        loggingConfiguration {
-            kloggingMinLogLevel(appLevel)
-            sink(stdoutSink, RENDER_ISO8601, STDOUT)
-            logging {
-                fromLoggerBase("io.seldon")
-                toSink(stdoutSink)
-            }
-            logging {
-                fromMinLevel(kafkaLevel) {
-                    fromLoggerBase("org.apache")
-                    toSink(stdoutSink)
-                }
+    fun configure(
+        appLevel: Level = Level.INFO,
+        kafkaLevel: Level = Level.WARN,
+    ) = loggingConfiguration {
+        kloggingMinLogLevel(appLevel)
+        sink(STDOUT_SINK, RENDER_ISO8601, STDOUT)
+        logging {
+            fromLoggerBase("io.seldon")
+            toSink(STDOUT_SINK)
+        }
+        logging {
+            fromMinLevel(kafkaLevel) {
+                fromLoggerBase("org.apache")
+                toSink(STDOUT_SINK)
             }
         }
-}
+    }
+}