on local runs, only pull the image if it isn't present locally alread…

…y. add the ability to specify a different docker image in task config.

build.gradle.kts

@@ -8,7 +8,7 @@ plugins {
 }
 
 group = "io.krews"
-version = "0.10.12"
+version = "0.11.0"
 
 repositories {
     maven { setUrl("http://dl.bintray.com/kotlin/kotlin-eap") }

scripts/test.sh

@@ -4,5 +4,6 @@ set -e
 
 # cd to project root directory
 cd "$(dirname "$(dirname "$0")")"
+cd src/test/kotlin/krews/util/test-image && docker build -t krewstest:latest . && cd ../../../../../..
 
-./gradlew test -Dkotlintest.tags.exclude=E2E
+./gradlew test -Dkotlintest.tags.exclude=E2E

src/main/kotlin/krews/config/BaseConfig.kt

@@ -10,7 +10,9 @@ data class TaskConfig (
     // The maximum allowed parallelism for this task
     val parallelism: Parallelism = UnlimitedParallelism,
     // The number of tasks "executions" that will be run with the same job / vm.
-    val grouping: Int = 1
+    val grouping: Int = 1,
+    // Docker image; defaults to one provided by task
+    val dockerImage: String? = null
 )
 
 data class WorkflowConfig (

src/main/kotlin/krews/core/TaskRunContext.kt

@@ -60,7 +60,7 @@ class TaskRunContextBuilder<I : Any, O : Any> internal constructor(
 
         return TaskRunContext(
             taskName = taskName,
-            dockerImage = checkNotNull(dockerImage),
+            dockerImage = checkNotNull(rawTaskParams.get("dockerImage") as String? ?: dockerImage),
             inputsDir = inputsDir,
             outputsDir = outputsDir,
             input = input,

src/main/kotlin/krews/executor/local/LocalExecutor.kt

@@ -84,8 +84,13 @@ class LocalExecutor(workflowConfig: WorkflowConfig) : LocallyDirectedExecutor {
             val runBasePath = workflowBasePath.resolve(workflowRunDir)
 
             // Pull image from remote
-            log.info { "Pulling image \"${taskRunContext.dockerImage}\" from remote..." }
-            dockerClient.pullImageCmd(taskRunContext.dockerImage).exec(PullImageResultCallback()).awaitCompletion()
+            try {
+                dockerClient.inspectImageCmd(taskRunContext.dockerImage).exec()
+                log.info { "Found ${taskRunContext.dockerImage} locally; skipping pull." }
+            } catch (e: Throwable) {
+                log.info { "Pulling image \"${taskRunContext.dockerImage}\" from remote..." }
+                dockerClient.pullImageCmd(taskRunContext.dockerImage).exec(PullImageResultCallback()).awaitCompletion()
+            }
 
             // Download InputFiles from remote sources
             for (downloadInputFile in taskRunContext.inputFiles) {

src/test/kotlin/krews/AppTests.kt

@@ -24,6 +24,13 @@ class AppTests {
         task.base64.params = {
             some-val = test
         }
+        
+        task.echo {
+            params {
+                value = "test"
+            }
+            docker-image = "krewstest:latest"
+        }
         """.trimIndent()
 
     @BeforeAll

src/test/kotlin/krews/LocalExecutorTests.kt

@@ -11,7 +11,6 @@ import org.junit.jupiter.api.*
 import java.nio.file.*
 import kotlin.streams.toList
 
-@Disabled
 @TestMethodOrder(MethodOrderer.OrderAnnotation::class)
 class LocalExecutorTests {
     private val testDir = Paths.get("local-workflow-test")!!
@@ -20,6 +19,7 @@ class LocalExecutorTests {
     private val unusedFilesDir = testDir.resolve("unused-files")
     private val base64Dir = outputsDir.resolve("base64")
     private val gzipDir = outputsDir.resolve("gzip")
+    private val echoDir = outputsDir.resolve("echo")
 
     private fun config(taskParam: String) =
         """
@@ -42,6 +42,12 @@ class LocalExecutorTests {
                 }
             ]
         }
+        task.echo {
+            params {
+                value = "test"
+            }
+            docker-image = "krewstest:latest"
+        }
         task.default {
             grouping = 2
         }
@@ -75,10 +81,11 @@ class LocalExecutorTests {
             assertThat(gzipDir.resolve("test-$i.b64.gz")).exists()
             assertThat(gzipDir.resolve("test-$i.b64.fake.none")).exists()
         }
+        assertThat(echoDir.resolve("output.txt")).exists()
 
         // Confirm that logs and an html report were generated
         val runPath = testDir.resolve("run/1/")
-        assertThat(Files.list(runPath.resolve(LOGS_DIR)).toList().size).isEqualTo(6)
+        assertThat(runPath.resolve(LOGS_DIR)).exists()
         assertThat(runPath.resolve(REPORT_FILENAME)).exists()
         assertThat(runPath.resolve(STATUS_JSON_FILENAME)).exists()
     }
@@ -159,4 +166,4 @@ class LocalExecutorTests {
         runner.run()
         return executor
     }
-}
+}

src/test/kotlin/krews/SlurmTest.kt

@@ -55,6 +55,12 @@ class SlurmExecutorTests {
                 comment = "\"This is just a test.\""
             }
         }
+        task.echo {
+            params {
+                value = "test"
+            }
+            docker-image = "krewstest:latest"
+        }
 
         """.trimIndent()
 

src/test/kotlin/krews/util/Workflows.kt

@@ -14,12 +14,16 @@ data class TestComplexInputType (
     override val file: File
 ) : TestBaseInputType
 
-data class Bast64TaskParams(val someVal: String, val someFiles: List<File>?)
+data class Base64TaskParams(val someVal: String, val someFiles: List<File>?)
+
+data class EchoTaskParams(val value: String)
 
 data class LocalGzipOutput(val gzip: File, val missingOptional: File)
 
 data class Base64TaskOutput(val base64: File, val extra: OutputDirectory)
 
+data class EchoTaskOutput(val output: File)
+
 fun localFilesWorkflow() = workflow("local-files-workflow") {
     val params = params<LocalWorkflowParams>()
     val sampleFiles = Files.newDirectoryStream(Paths.get(params.sampleFilesDir)).sortedBy { f -> f.fileName }
@@ -29,7 +33,7 @@ fun localFilesWorkflow() = workflow("local-files-workflow") {
         .toFlux()
 
     val base64 = task<TestBaseInputType, Base64TaskOutput>("base64", sampleFiles) {
-        val taskParams = taskParams<Bast64TaskParams>()
+        val taskParams = taskParams<Base64TaskParams>()
         val file = input.file
         dockerImage = "alpine:3.8"
         output = Base64TaskOutput(OutputFile("base64/${file.filenameNoExt()}.b64"), OutputDirectory("extra"))
@@ -44,6 +48,17 @@ fun localFilesWorkflow() = workflow("local-files-workflow") {
             """
     }
 
+    task<Any, EchoTaskOutput>("echo", listOf(0).toFlux()) {
+        val taskParams = taskParams<EchoTaskParams>()
+        dockerImage = "krewstest"
+        output = EchoTaskOutput(OutputFile("echo/output.txt"))
+        command =
+            """
+            mkdir -p $(dirname ${output!!.output.dockerPath})
+            /test.sh ${taskParams.value} ${output!!.output.dockerPath}
+            """
+    }
+
     task<File, LocalGzipOutput>("gzip", base64.map { it.base64 }) {
         dockerImage = "alpine:3.8"
         val outGz = OutputFile("gzip/${input.filename()}.gz")
@@ -66,6 +81,7 @@ fun localFilesWorkflow() = workflow("local-files-workflow") {
             gzip -c ${input.dockerPath} > ${output!!.dockerPath}
             """
     }
+
 }
 
 private data class GSWorkflowParams(

src/test/kotlin/krews/util/test-image/Dockerfile

@@ -0,0 +1,2 @@
+FROM alpine:3.8
+COPY test.sh /

src/test/kotlin/krews/util/test-image/test.sh

@@ -0,0 +1 @@
+echo $1 > $2