Cache all uploads and downloads to GCS (#639)

* Cache all uploads and downloads to GCS

Currently, when using local apk artifacts, Flank always uploads every
app/test apk pair even if the app apk has already been uploaded. This is
also true for pulling the old JUnitReport from GCS for smart flank where
every test apk re-downloads the results. To fix this we simply cache
uploads and downloads in a ConcurrentHashMap where the key is the GCS
path for downloads and the filename for uploads.

This change also fixes a bug where local test apks would be uploaded and
then re-downloaded parse their tests. The sharding loop now loops over
the local AppTestPair where it uploads them if necessary and scans for
tests on the local test apk.

All in all this shaves almost 4 minutes off of shard generation for a
project with 20+ modules. There is also likely more work that can be
done to parallelize matrix creation (uploading apks, parsing tests,
creating the shard), but it is out of scope of this change.

* Ensure resources used to download from GCS are closed

Make sure that when an exception is ignored, we still close our file
system resources (ReadChannel/FileOutputStream). Also rename some
variables and methods to better describe what they are and how they are
used.

README.md

@@ -316,11 +316,12 @@ flank:
   ## Default: false  
   # keep-file-path: false
 
-  ## Include additional app/test apk pairs in the run. If app is omitted, then the top level app is used for that pair.
+  ## Include additional app/test apk pairs in the run. Apks are unique by just filename and not by path!
+  ## If app is omitted, then the top level app is used for that pair.
   # additional-app-test-apks:
   #  - app: ../test_app/apks/app-debug.apk
-  #    test: ../test_app/apks/app-debug-androidTest.apk
-  #  - test: ../test_app/apks/app-debug-androidTest.apk
+  #    test: ../test_app/apks/app1-debug-androidTest.apk
+  #  - test: ../test_app/apks/app2-debug-androidTest.apk
 ```
 
 ### Android code coverage

test_runner/src/main/kotlin/ftl/args/yml/AndroidFlankYml.kt

@@ -4,7 +4,7 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties
 import com.fasterxml.jackson.annotation.JsonProperty
 
 data class AppTestPair(
-    val app: String?,
+    val app: String,
     val test: String
 )
 

test_runner/src/main/kotlin/ftl/gc/GcStorage.kt

@@ -20,9 +20,13 @@ import java.io.FileOutputStream
 import java.net.URI
 import java.nio.file.Files
 import java.nio.file.Paths
+import java.util.concurrent.ConcurrentHashMap
 
 object GcStorage {
 
+    private val uploadCache: ConcurrentHashMap<String, String> = ConcurrentHashMap()
+    private val downloadCache: ConcurrentHashMap<String, String> = ConcurrentHashMap()
+
     val storageOptions: StorageOptions by lazy {
         val builder = StorageOptions.newBuilder()
         if (FtlConstants.useMock) builder.setHost(FtlConstants.localhost)
@@ -99,43 +103,45 @@ object GcStorage {
 
     private fun upload(file: String, fileBytes: ByteArray, rootGcsBucket: String, runGcsPath: String): String {
         val fileName = Paths.get(file).fileName.toString()
-        val gcsFilePath = GCS_PREFIX + join(rootGcsBucket, runGcsPath, fileName)
-
-        // 404 Not Found error when rootGcsBucket does not exist
-        val fileBlob = BlobInfo.newBuilder(rootGcsBucket, join(runGcsPath, fileName)).build()
-
-        val progress = ProgressBar()
-        try {
-            progress.start("Uploading $fileName")
-            storage.create(fileBlob, fileBytes)
-        } catch (e: Exception) {
-            fatalError(e)
-        } finally {
-            progress.stop()
+        return uploadCache[fileName] ?: uploadCache.computeIfAbsent(fileName) {
+            val gcsFilePath = GCS_PREFIX + join(rootGcsBucket, runGcsPath, fileName)
+
+            // 404 Not Found error when rootGcsBucket does not exist
+            val fileBlob = BlobInfo.newBuilder(rootGcsBucket, join(runGcsPath, fileName)).build()
+
+            val progress = ProgressBar()
+            try {
+                progress.start("Uploading $fileName")
+                storage.create(fileBlob, fileBytes)
+            } catch (e: Exception) {
+                fatalError(e)
+            } finally {
+                progress.stop()
+            }
+            gcsFilePath
         }
-
-        return gcsFilePath
     }
 
     fun download(gcsUriString: String, ignoreError: Boolean = false): String {
         val gcsURI = URI.create(gcsUriString)
         val bucket = gcsURI.authority
         val path = gcsURI.path.drop(1) // Drop leading slash
-
-        val outputFile = File.createTempFile("tmp", null)
-        outputFile.deleteOnExit()
-
-        try {
-            val blob = storage.get(bucket, path)
-            val readChannel = blob.reader()
-            val output = FileOutputStream(outputFile)
-            output.channel.transferFrom(readChannel, 0, Long.MAX_VALUE)
-            output.close()
-        } catch (e: Exception) {
-            if (ignoreError) return ""
-            fatalError(e)
+        return downloadCache[path] ?: downloadCache.computeIfAbsent(path) {
+            val outputFile = File.createTempFile("tmp", null)
+            outputFile.deleteOnExit()
+
+            try {
+                val blob = storage.get(bucket, path)
+                blob.reader().use { readChannel ->
+                    FileOutputStream(outputFile).use {
+                        it.channel.transferFrom(readChannel, 0, Long.MAX_VALUE)
+                    }
+                }
+            } catch (e: Exception) {
+                if (ignoreError) return@computeIfAbsent ""
+                fatalError(e)
+            }
+            outputFile.path
         }
-
-        return outputFile.path
     }
 }

test_runner/src/main/kotlin/ftl/run/AndroidTestRunner.kt

@@ -19,71 +19,67 @@ import kotlinx.coroutines.coroutineScope
 
 object AndroidTestRunner {
 
-    suspend fun runTests(androidArgs: AndroidArgs): Pair<MatrixMap, List<List<String>>> = coroutineScope {
-        val (stopwatch, runGcsPath) = GenericTestRunner.beforeRunTests(androidArgs)
+    suspend fun runTests(args: AndroidArgs): Pair<MatrixMap, List<List<String>>> = coroutineScope {
+        val (stopwatch, runGcsPath) = GenericTestRunner.beforeRunTests(args)
 
         // GcAndroidMatrix => GcAndroidTestMatrix
         // GcAndroidTestMatrix.execute() 3x retry => matrix id (string)
-        val androidDeviceList = GcAndroidDevice.build(androidArgs.devices)
+        val androidDeviceList = GcAndroidDevice.build(args.devices)
 
         val jobs = arrayListOf<Deferred<TestMatrix>>()
-        val runCount = androidArgs.repeatTests
+        val runCount = args.repeatTests
         val shardCounter = ShardCounter()
-        val history = GcToolResults.createToolResultsHistory(androidArgs)
-        val apks = resolveApks(androidArgs, runGcsPath)
-        val allTestShardChunks: MutableList<List<String>> = mutableListOf()
-
-        apks.forEach { apk ->
-            // ensure we only shard tests that are part of the test apk
-            val testShardChunks = AndroidTestShard.getTestShardChunks(androidArgs, apk.test)
-            allTestShardChunks += testShardChunks
+        val history = GcToolResults.createToolResultsHistory(args)
+        val apkPairsInArgs = listOf(AppTestPair(app = args.appApk, test = args.testApk)) + args.additionalAppTestApks
+        val allTestShardChunks: List<List<String>> = apkPairsInArgs.map { unresolvedApkPair ->
+            val resolvedApkPair = resolveApkPair(unresolvedApkPair, args, runGcsPath)
+            // Ensure we only shard tests that are part of the test apk. Use the unresolved test apk path to make sure
+            // we don't re-download an apk it is on the local file system.
+            val testShards = AndroidTestShard.getTestShardChunks(args, unresolvedApkPair.test)
             repeat(runCount) {
-                testShardChunks.forEach { testTargets ->
+                testShards.forEach { testTargets ->
                     // specify dispatcher to avoid inheriting main runBlocking context that runs in the main thread
                     // https://kotlinlang.org/docs/reference/coroutines/coroutine-context-and-dispatchers.html
                     jobs += async(Dispatchers.IO) {
                         GcAndroidTestMatrix.build(
-                            appApkGcsPath = apk.app ?: androidArgs.appApk,
-                            testApkGcsPath = apk.test,
+                            appApkGcsPath = resolvedApkPair.app,
+                            testApkGcsPath = resolvedApkPair.test,
                             runGcsPath = runGcsPath,
                             androidDeviceList = androidDeviceList,
                             testTargets = testTargets,
-                            args = androidArgs,
+                            args = args,
                             shardCounter = shardCounter,
                             toolResultsHistory = history
                         ).executeWithRetry()
                     }
                 }
             }
-        }
+            testShards
+        }.flatten()
 
-        println(GenericTestRunner.beforeRunMessage(androidArgs, allTestShardChunks))
-        val matrixMap = GenericTestRunner.afterRunTests(jobs.awaitAll(), runGcsPath, stopwatch, androidArgs)
+        println(GenericTestRunner.beforeRunMessage(args, allTestShardChunks))
+        val matrixMap = GenericTestRunner.afterRunTests(jobs.awaitAll(), runGcsPath, stopwatch, args)
         matrixMap to allTestShardChunks
     }
 
     /**
-     * Upload APKs if the path given is local
+     * Upload an APK pair if the path given is local
      *
-     * @return Pair(gcs uri for app apk, gcs uri for test apk)
+     * @return AppTestPair with their GCS paths
      */
-    private suspend fun resolveApks(args: AndroidArgs, runGcsPath: String): List<AppTestPair> = coroutineScope {
+    private suspend fun resolveApkPair(
+        apk: AppTestPair,
+        args: AndroidArgs,
+        runGcsPath: String
+    ): AppTestPair = coroutineScope {
         val gcsBucket = args.resultsBucket
-        val appTestApks = listOf(AppTestPair(app = args.appApk, test = args.testApk)) + args.additionalAppTestApks
-        val result = mutableListOf<AppTestPair>()
-
-        appTestApks.forEach { apks ->
-            val appApkGcsPath = async(Dispatchers.IO) { GcStorage.upload(apks.app ?: args.appApk, gcsBucket, runGcsPath) }
-            val testApkGcsPath = async(Dispatchers.IO) { GcStorage.upload(apks.test, gcsBucket, runGcsPath) }
 
-            result.add(
-                AppTestPair(
-                    app = appApkGcsPath.await(),
-                    test = testApkGcsPath.await()
-                )
-            )
-        }
+        val appApkGcsPath = async(Dispatchers.IO) { GcStorage.upload(apk.app, gcsBucket, runGcsPath) }
+        val testApkGcsPath = async(Dispatchers.IO) { GcStorage.upload(apk.test, gcsBucket, runGcsPath) }
 
-        result
+        AppTestPair(
+            app = appApkGcsPath.await(),
+            test = testApkGcsPath.await()
+        )
     }
 }