Add fault tolerance for builds which hang upon retrieval (#52)

Also:
- Adds info-level logging of what builds are in flight to help identity
  hanging scans
- Simplifies the `DevelocityService.build` function to eliminate the
  vestigial transform argument

Author: mcumings

README.md

@@ -64,6 +64,11 @@ extensions.configure<MetricsForDevelocityExtension> {
         // when querying build data.  Defaults to 24.
         develocityMaxConcurrency.set(10)
 
+        // Optional: Set the maximum number of seconds to wait for an individual build scan to be
+        // processed before skipping it and proceeding with data retrieval.  This provides some
+        // mitigation if Develocity is slow to return a scan.  Defaults to 120 seconds.
+        buildScanRetrievalTimeout.set(60)
+        
         // Optional: Add custom summarizers to the plugin.  These summarizers will be used to
         // build 
         summarizers.add(MyCustomSummarizer())

src/main/kotlin/com/ebay/plugins/metrics/develocity/GatherHourlyTask.kt

@@ -10,11 +10,11 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ReceiveChannel
-import kotlinx.coroutines.channels.consumeEach
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeoutOrNull
 import org.gradle.api.DefaultTask
 import org.gradle.api.file.DirectoryProperty
 import org.gradle.api.model.ObjectFactory
@@ -31,6 +31,7 @@ import java.time.OffsetDateTime
 import java.time.ZoneId
 import java.time.ZoneOffset
 import java.time.format.DateTimeFormatter
+import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.atomic.AtomicInteger
 import javax.inject.Inject
 import kotlin.math.roundToInt
@@ -85,11 +86,15 @@ open class GatherHourlyTask @Inject constructor(
     @get:Internal
     override val summarizersProperty: ListProperty<MetricSummarizer<*>> = objectFactory.listProperty(MetricSummarizer::class.java)
 
+    @get:Internal
+    val buildScanRetrievalTimeoutProperty: Property<Int> = objectFactory.property(Int::class.java)
+
     @get:OutputDirectory
     override val outputDirectoryProperty: DirectoryProperty = objectFactory.directoryProperty()
 
     private val refCount = AtomicInteger()
     private val processedCount = AtomicInteger()
+    private val processingBuilds = CopyOnWriteArrayList<String>()
 
     @TaskAction
     fun gather() {
@@ -179,13 +184,21 @@ open class GatherHourlyTask @Inject constructor(
             val detailsQuery = BuildQuery(
                 models = modelsNeeded
             )
-            channel.consumeEach { buildRef ->
-                develocityServiceProperty.get().build(buildRef.id, detailsQuery) { build ->
+            val timeout = buildScanRetrievalTimeoutProperty.get().seconds
+            for (buildRef in channel) {
+                processingBuilds.add(buildRef.id)
+                val build = withTimeoutOrNull(timeout) {
+                    develocityServiceProperty.get().build(buildRef.id, detailsQuery)
+                }
+                if (build == null) {
+                    logger.warn("Unable to process build ${buildRef.id}!  Skipping!")
+                } else {
                     summarizerStates.forEach { state ->
                         state.ingestBuild(build)
                     }
-                    processedCount.incrementAndGet()
                 }
+                processingBuilds.remove(buildRef.id)
+                processedCount.incrementAndGet()
             }
         }
     }
@@ -213,6 +226,18 @@ open class GatherHourlyTask @Inject constructor(
         } else {
             "%.2f / second".format(processedSoFar / elapsedSeconds)
         }
-        logger.info("Processed $processedSoFar build(s) in ${elapsedSeconds.roundToInt()} second(s): $rate")
+        val processingBuilds = processingBuilds.toArray().let { processing ->
+            if (processing.isEmpty()) {
+                ""
+            } else {
+                processing.joinToString(
+                    prefix = " (Processing ${processing.size} build(s): ",
+                    separator = ", ",
+                    postfix = ")"
+                )
+            }
+
+        }
+        logger.info("Processed $processedSoFar build(s) in ${elapsedSeconds.roundToInt()} second(s): $rate $processingBuilds")
     }
 }

src/main/kotlin/com/ebay/plugins/metrics/develocity/MetricsForDevelocityExtension.kt

@@ -42,6 +42,14 @@ abstract class MetricsForDevelocityExtension : ExtensionAware {
      */
     abstract val develocityMaxConcurrency: Property<Int>
 
+    /**
+     * The maximum amount of time, in seconds, to wait for an individual build scan to be processed.
+     * If a build scan takes longer than this amount of time, it will be discarded / skipped,
+     * mainly to prevent hanging the entire data collection process.  By default, the timeout
+     * is set to 2 minutes.
+     */
+    abstract val buildScanRetrievalTimeout: Property<Int>
+
     /**
      * Custom build data summarizers to apply to the build data.  These capture the details
      * of the build that are important to report upon and summarize them in a way that can

src/main/kotlin/com/ebay/plugins/metrics/develocity/MetricsForDevelocityInternalExtension.kt

@@ -60,6 +60,8 @@ internal class MetricsForDevelocityProjectPlugin @Inject constructor(
                 develocityQueryFilter.convention(
                     providerFactory.gradleProperty(QUERY_FILTER_PROPERTY)
                         .orElse("project:${project.name}"))
+
+                buildScanRetrievalTimeout.convention(120)
             }
 
         // Register the build service used to query Develocity for build data
@@ -278,6 +280,7 @@ internal class MetricsForDevelocityProjectPlugin @Inject constructor(
                         queryProperty.set(ext.develocityQueryFilter)
                         summarizersProperty.set(ext.summarizers)
                         develocityServiceProperty.set(buildServiceProvider)
+                        buildScanRetrievalTimeoutProperty.set(ext.buildScanRetrievalTimeout)
                         outputDirectoryProperty.set(PathUtil.hourlyOutputDir(project.layout, timeSpec))
                         maxConcurrencyProperty.set(ext.develocityMaxConcurrency)
                         usesService(buildServiceProvider)

src/main/kotlin/com/ebay/plugins/metrics/develocity/MetricsForDevelocityProjectPlugin.kt

@@ -80,16 +80,15 @@ abstract class DevelocityBuildService @Inject constructor(
         } ?: flow
     }
 
-    override suspend fun <T> build(
+    override suspend fun build(
         buildId: String,
         params: BuildQuery,
-        transform: (Build) -> T,
-    ): T? = withContext(dispatcher) {
-        transform.invoke(api.buildsApi.getBuild(
+    ): Build? = withContext(dispatcher) {
+        api.buildsApi.getBuild(
             id = buildId,
             models = params.models,
             allModels = params.allModels,
-        ))
+        )
     }
 
     private fun resolveServer(): String {

src/main/kotlin/com/ebay/plugins/metrics/develocity/service/DevelocityBuildService.kt

@@ -19,10 +19,8 @@ interface DevelocityService {
     /**
      * Get build details for the build with the given ID.
      */
-    suspend fun <T> build(
+    suspend fun build(
         buildId: String,
         params: BuildQuery,
-        transform: (Build) -> T,
-    ): T?
-
-}
+    ): Build?
+}