diff --git a/src/main/kotlin/io/cloudshiftdev/mavensync/MavenHttpClient.kt b/src/main/kotlin/io/cloudshiftdev/mavensync/MavenHttpClient.kt index 45945a4..df6bd5b 100644 --- a/src/main/kotlin/io/cloudshiftdev/mavensync/MavenHttpClient.kt +++ b/src/main/kotlin/io/cloudshiftdev/mavensync/MavenHttpClient.kt @@ -30,13 +30,15 @@ internal class MavenHttpClient( httpClient.close() } - internal suspend fun upload(url: Url, file: File) { + internal suspend fun upload(url: Url, file: File): Long { val resp = httpClient.put(url) { contentType(ContentType.Application.OctetStream) setBody(LocalFileContent(file)) } - logger.info { "Uploaded $url: status=${resp.status} size=${file.length()}" } + val size = file.length() + logger.info { "Uploaded $url: status=${resp.status} size=$size" } + return size } internal suspend fun upload(url: Url, content: String) { diff --git a/src/main/kotlin/io/cloudshiftdev/mavensync/MavenHttpRepository.kt b/src/main/kotlin/io/cloudshiftdev/mavensync/MavenHttpRepository.kt index 12999cd..733e2c1 100644 --- a/src/main/kotlin/io/cloudshiftdev/mavensync/MavenHttpRepository.kt +++ b/src/main/kotlin/io/cloudshiftdev/mavensync/MavenHttpRepository.kt @@ -22,9 +22,9 @@ internal interface MavenHttpRepository : AutoCloseable { includeSignatures: Boolean, ): List - suspend fun copyAsset(asset: ArtifactVersionAsset, targetRepository: MavenHttpRepository) + suspend fun copyAsset(asset: ArtifactVersionAsset, targetRepository: MavenHttpRepository): Long - suspend fun uploadAsset(asset: ArtifactVersionAsset, file: Path) + suspend fun uploadAsset(asset: ArtifactVersionAsset, file: Path): Long suspend fun releaseVersion(coordinates: Coordinates) @@ -100,14 +100,14 @@ internal class DefaultMavenHttpRepository( override suspend fun copyAsset( asset: ArtifactVersionAsset, targetRepository: MavenHttpRepository, - ) { - mavenHttpClient.download(url(asset.coordinates, asset.name)) { _, file -> + ): Long { + return mavenHttpClient.download(url(asset.coordinates, asset.name)) { _, file -> targetRepository.uploadAsset(asset, file.toPath()) } } - override suspend fun uploadAsset(asset: ArtifactVersionAsset, file: Path) { - mavenHttpClient.upload(url(asset.coordinates, asset.name), file.toFile()) + override suspend fun uploadAsset(asset: ArtifactVersionAsset, file: Path): Long { + return mavenHttpClient.upload(url(asset.coordinates, asset.name), file.toFile()) } override suspend fun releaseVersion(coordinates: Coordinates) { diff --git a/src/main/kotlin/io/cloudshiftdev/mavensync/MavenSyncEngine.kt b/src/main/kotlin/io/cloudshiftdev/mavensync/MavenSyncEngine.kt index 6aac29b..3bc262e 100644 --- a/src/main/kotlin/io/cloudshiftdev/mavensync/MavenSyncEngine.kt +++ b/src/main/kotlin/io/cloudshiftdev/mavensync/MavenSyncEngine.kt @@ -1,6 +1,8 @@ package io.cloudshiftdev.mavensync import io.github.oshai.kotlinlogging.KotlinLogging +import kotlin.coroutines.cancellation.CancellationException +import kotlin.time.TimeSource import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.channels.consumeEach import kotlinx.coroutines.coroutineScope @@ -15,6 +17,7 @@ internal class MavenSyncEngine( private val source: MavenHttpRepository, private val target: MavenHttpRepository, private val options: SyncOptions, + private val metrics: SyncMetrics, ) { @OptIn(ExperimentalCoroutinesApi::class) suspend fun sync() = coroutineScope { @@ -32,6 +35,9 @@ internal class MavenSyncEngine( val targetMetadata = target.queryArtifactMetadata(metadata.group, metadata.artifact) val sourceVersions = metadata.artifactVersions.toSet() val targetVersions = targetMetadata.artifactVersions.toSet() + val inSyncVersions = sourceVersions intersect targetVersions + metrics.recordInSync(metadata.group, metadata.artifact, inSyncVersions) + val missingVersions = sourceVersions - targetVersions if (missingVersions.isEmpty()) { logger.debug { "No missing versions for ${metadata.group}:${metadata.artifact}" } @@ -42,21 +48,38 @@ internal class MavenSyncEngine( } missingVersions .map { Coordinates(metadata.group, metadata.artifact, it) } - .forEach { coordinates -> - val assets = - source.listArtifactVersionAssets( - coordinates, - options.transferChecksums, - options.transferSignatures, - ) - - if (assets.isNotEmpty()) { - assets.forEach { asset -> source.copyAsset(asset, target) } - - target.releaseVersion(coordinates) - - delay(options.downloadDelay) - } - } + .forEach { coordinates -> syncVersion(coordinates) } + } + + private suspend fun syncVersion(coordinates: Coordinates) { + val mark = TimeSource.Monotonic.markNow() + try { + val assets = + source.listArtifactVersionAssets( + coordinates, + options.transferChecksums, + options.transferSignatures, + ) + + if (assets.isEmpty()) return + + var bytes = 0L + assets.forEach { asset -> bytes += source.copyAsset(asset, target) } + target.releaseVersion(coordinates) + metrics.recordSynced( + coordinates = coordinates, + assetCount = assets.size, + bytes = bytes, + duration = mark.elapsedNow(), + ) + + delay(options.downloadDelay) + } catch (e: CancellationException) { + throw e + } catch (e: Exception) { + val msg = e.message ?: e.toString() + logger.error(e) { "Failed to sync $coordinates: $msg" } + metrics.recordFailure(coordinates, msg, mark.elapsedNow()) + } } } diff --git a/src/main/kotlin/io/cloudshiftdev/mavensync/MavenSyncMain.kt b/src/main/kotlin/io/cloudshiftdev/mavensync/MavenSyncMain.kt index d44b406..4fd569b 100644 --- a/src/main/kotlin/io/cloudshiftdev/mavensync/MavenSyncMain.kt +++ b/src/main/kotlin/io/cloudshiftdev/mavensync/MavenSyncMain.kt @@ -27,10 +27,17 @@ public suspend fun main(args: Array) { logger.info { "Effective configuration: $config" } - config.source.toMavenHttpRepository().use { source -> - config.target.toMavenHttpRepository().use { target -> - MavenSyncEngine(source, target, config.toSyncOptions()).sync() + val metrics = SyncMetrics() + try { + config.source.toMavenHttpRepository().use { source -> + config.target.toMavenHttpRepository().use { target -> + MavenSyncEngine(source, target, config.toSyncOptions(), metrics).sync() + } } + } finally { + val report = metrics.snapshot() + logger.info { "\n" + report.renderDetailed() } + logger.info { "\n" + report.renderSummary() } } } diff --git a/src/main/kotlin/io/cloudshiftdev/mavensync/SyncMetrics.kt b/src/main/kotlin/io/cloudshiftdev/mavensync/SyncMetrics.kt new file mode 100644 index 0000000..b3c110b --- /dev/null +++ b/src/main/kotlin/io/cloudshiftdev/mavensync/SyncMetrics.kt @@ -0,0 +1,154 @@ +package io.cloudshiftdev.mavensync + +import java.util.concurrent.ConcurrentHashMap +import kotlin.time.Duration +import kotlin.time.TimeSource + +internal class SyncMetrics { + private val mark = TimeSource.Monotonic.markNow() + private val artifacts = ConcurrentHashMap, Entry>() + + fun recordInSync(group: Group, artifact: Artifact, versions: Collection) { + if (versions.isEmpty()) return + entry(group, artifact).inSync.addAll(versions) + } + + fun recordSynced(coordinates: Coordinates, assetCount: Int, bytes: Long, duration: Duration) { + entry(coordinates.group, coordinates.artifact) + .synced + .add( + VersionResult.Success( + version = coordinates.artifactVersion, + assetCount = assetCount, + bytes = bytes, + duration = duration, + ) + ) + } + + fun recordFailure(coordinates: Coordinates, error: String, duration: Duration) { + entry(coordinates.group, coordinates.artifact) + .failed + .add( + VersionResult.Failure( + version = coordinates.artifactVersion, + error = error, + duration = duration, + ) + ) + } + + fun snapshot(): SyncReport { + val artifactMetrics = + artifacts.values + .map { e -> + ArtifactMetrics( + group = e.group, + artifact = e.artifact, + inSync = e.inSync.toList(), + synced = e.synced.toList(), + failed = e.failed.toList(), + ) + } + .sortedWith(compareBy({ it.group.value }, { it.artifact.value })) + return SyncReport(totalDuration = mark.elapsedNow(), artifacts = artifactMetrics) + } + + private fun entry(group: Group, artifact: Artifact): Entry = + artifacts.computeIfAbsent(group to artifact) { Entry(group, artifact) } + + private class Entry(val group: Group, val artifact: Artifact) { + val inSync: MutableList = mutableListOf() + val synced: MutableList = mutableListOf() + val failed: MutableList = mutableListOf() + } +} + +internal data class ArtifactMetrics( + val group: Group, + val artifact: Artifact, + val inSync: List, + val synced: List, + val failed: List, +) + +internal sealed interface VersionResult { + val version: ArtifactVersion + val duration: Duration + + data class Success( + override val version: ArtifactVersion, + val assetCount: Int, + val bytes: Long, + override val duration: Duration, + ) : VersionResult + + data class Failure( + override val version: ArtifactVersion, + val error: String, + override val duration: Duration, + ) : VersionResult +} + +internal data class SyncReport(val totalDuration: Duration, val artifacts: List) { + val inSyncTotal: Int = artifacts.sumOf { it.inSync.size } + val syncedTotal: Int = artifacts.sumOf { it.synced.size } + val failedTotal: Int = artifacts.sumOf { it.failed.size } + val assetsTotal: Int = artifacts.sumOf { a -> a.synced.sumOf { it.assetCount } } + val bytesTotal: Long = artifacts.sumOf { a -> a.synced.sumOf { it.bytes } } + + fun renderDetailed(): String = + buildString { + appendLine("========== SYNC METRICS (detailed) ==========") + if (artifacts.isEmpty()) { + appendLine("(no artifacts processed)") + return@buildString + } + artifacts.forEach { a -> + appendLine("${a.group.value}:${a.artifact.value}") + if (a.inSync.isNotEmpty()) { + appendLine( + " in sync (${a.inSync.size}): ${a.inSync.joinToString(", ") { it.value }}" + ) + } + if (a.synced.isNotEmpty()) { + appendLine(" synced (${a.synced.size}):") + a.synced.forEach { s -> + appendLine( + " ${s.version.value} — ${s.assetCount} assets, ${formatBytes(s.bytes)}, ${s.duration}" + ) + } + } + if (a.failed.isNotEmpty()) { + appendLine(" failed (${a.failed.size}):") + a.failed.forEach { f -> + appendLine(" ${f.version.value} — ${f.error} (${f.duration})") + } + } + } + } + .trimEnd() + + fun renderSummary(): String = buildString { + appendLine("========== SYNC METRICS (summary) ==========") + appendLine("artifacts: ${artifacts.size}") + appendLine("versions in sync: $inSyncTotal") + appendLine("versions synced: $syncedTotal") + appendLine("versions failed: $failedTotal") + appendLine("assets copied: $assetsTotal") + appendLine("bytes transferred: ${formatBytes(bytesTotal)}") + append("duration: $totalDuration") + } +} + +private fun formatBytes(bytes: Long): String { + if (bytes < 1024) return "$bytes B" + val units = listOf("KB", "MB", "GB", "TB") + var value = bytes.toDouble() / 1024.0 + var idx = 0 + while (value >= 1024.0 && idx < units.lastIndex) { + value /= 1024.0 + idx++ + } + return "%.1f %s".format(value, units[idx]) +} diff --git a/src/test/kotlin/io/cloudshiftdev/mavensync/FakeMavenHttpRepository.kt b/src/test/kotlin/io/cloudshiftdev/mavensync/FakeMavenHttpRepository.kt index 78c5429..7346bb6 100644 --- a/src/test/kotlin/io/cloudshiftdev/mavensync/FakeMavenHttpRepository.kt +++ b/src/test/kotlin/io/cloudshiftdev/mavensync/FakeMavenHttpRepository.kt @@ -35,15 +35,19 @@ internal class FakeMavenHttpRepository(private val label: String) : MavenHttpRep return assets[coordinates].orEmpty() } + var copyAssetBehavior: suspend (ArtifactVersionAsset) -> Long = { 0L } + override suspend fun copyAsset( asset: ArtifactVersionAsset, targetRepository: MavenHttpRepository, - ) { + ): Long { copyCalls += asset to targetRepository + return copyAssetBehavior(asset) } - override suspend fun uploadAsset(asset: ArtifactVersionAsset, file: Path) { + override suspend fun uploadAsset(asset: ArtifactVersionAsset, file: Path): Long { uploadCalls += asset to file + return 0L } override suspend fun releaseVersion(coordinates: Coordinates) { diff --git a/src/test/kotlin/io/cloudshiftdev/mavensync/MavenSyncEngineTest.kt b/src/test/kotlin/io/cloudshiftdev/mavensync/MavenSyncEngineTest.kt index 5f28399..bd42b79 100644 --- a/src/test/kotlin/io/cloudshiftdev/mavensync/MavenSyncEngineTest.kt +++ b/src/test/kotlin/io/cloudshiftdev/mavensync/MavenSyncEngineTest.kt @@ -4,6 +4,7 @@ import io.kotest.core.spec.style.FunSpec import io.kotest.matchers.collections.shouldBeEmpty import io.kotest.matchers.collections.shouldContainExactly import io.kotest.matchers.collections.shouldContainExactlyInAnyOrder +import io.kotest.matchers.shouldBe import kotlin.time.Duration private fun defaultOptions(transferChecksums: Boolean = false, transferSignatures: Boolean = true) = @@ -33,7 +34,7 @@ class MavenSyncEngineTest : listOf(ArtifactVersion("1.0"), ArtifactVersion("1.1")), ) ) - val engine = MavenSyncEngine(source, target, defaultOptions()) + val engine = MavenSyncEngine(source, target, defaultOptions(), SyncMetrics()) engine.handleArtifact( ArtifactMetadata( @@ -58,7 +59,7 @@ class MavenSyncEngineTest : source.assets[v11] = listOf(ArtifactVersionAsset(v11, Filename("foo-1.1.jar"))) source.assets[v20] = listOf(ArtifactVersionAsset(v20, Filename("foo-2.0.jar"))) - val engine = MavenSyncEngine(source, target, defaultOptions()) + val engine = MavenSyncEngine(source, target, defaultOptions(), SyncMetrics()) engine.handleArtifact( ArtifactMetadata( @@ -81,7 +82,7 @@ class MavenSyncEngineTest : val v10 = coords("1.0") source.assets[v10] = emptyList() - val engine = MavenSyncEngine(source, target, defaultOptions()) + val engine = MavenSyncEngine(source, target, defaultOptions(), SyncMetrics()) engine.handleArtifact(ArtifactMetadata(group, artifact, listOf(ArtifactVersion("1.0")))) @@ -101,10 +102,86 @@ class MavenSyncEngineTest : source, target, defaultOptions(transferChecksums = true, transferSignatures = false), + SyncMetrics(), ) engine.handleArtifact(ArtifactMetadata(group, artifact, listOf(ArtifactVersion("1.0")))) source.listAssetCalls shouldContainExactly listOf(Triple(v, true, false)) } + + test("records in-sync versions in metrics when target already has them") { + val source = FakeMavenHttpRepository("source") + val target = FakeMavenHttpRepository("target") + target.seedMetadata( + ArtifactMetadata( + group, + artifact, + listOf(ArtifactVersion("1.0"), ArtifactVersion("1.1")), + ) + ) + val metrics = SyncMetrics() + val engine = MavenSyncEngine(source, target, defaultOptions(), metrics) + + engine.handleArtifact( + ArtifactMetadata( + group, + artifact, + listOf(ArtifactVersion("1.0"), ArtifactVersion("1.1")), + ) + ) + + val report = metrics.snapshot() + report.inSyncTotal shouldBe 2 + report.syncedTotal shouldBe 0 + report.failedTotal shouldBe 0 + } + + test("records synced versions with asset count and bytes") { + val source = FakeMavenHttpRepository("source") + val target = FakeMavenHttpRepository("target") + val v11 = coords("1.1") + source.assets[v11] = + listOf( + ArtifactVersionAsset(v11, Filename("foo-1.1.jar")), + ArtifactVersionAsset(v11, Filename("foo-1.1.pom")), + ) + source.copyAssetBehavior = { 100L } + val metrics = SyncMetrics() + val engine = MavenSyncEngine(source, target, defaultOptions(), metrics) + + engine.handleArtifact(ArtifactMetadata(group, artifact, listOf(ArtifactVersion("1.1")))) + + val report = metrics.snapshot() + report.syncedTotal shouldBe 1 + report.assetsTotal shouldBe 2 + report.bytesTotal shouldBe 200L + } + + test("records failure and continues to the next version when copyAsset throws") { + val source = FakeMavenHttpRepository("source") + val target = FakeMavenHttpRepository("target") + val v11 = coords("1.1") + val v20 = coords("2.0") + source.assets[v11] = listOf(ArtifactVersionAsset(v11, Filename("foo-1.1.jar"))) + source.assets[v20] = listOf(ArtifactVersionAsset(v20, Filename("foo-2.0.jar"))) + source.copyAssetBehavior = { asset -> + if (asset.coordinates == v11) error("boom") else 50L + } + val metrics = SyncMetrics() + val engine = MavenSyncEngine(source, target, defaultOptions(), metrics) + + engine.handleArtifact( + ArtifactMetadata( + group, + artifact, + listOf(ArtifactVersion("1.1"), ArtifactVersion("2.0")), + ) + ) + + val report = metrics.snapshot() + report.syncedTotal shouldBe 1 + report.failedTotal shouldBe 1 + target.releaseCalls shouldContainExactly listOf(v20) + } }) diff --git a/src/test/kotlin/io/cloudshiftdev/mavensync/SyncMetricsTest.kt b/src/test/kotlin/io/cloudshiftdev/mavensync/SyncMetricsTest.kt new file mode 100644 index 0000000..f47bb13 --- /dev/null +++ b/src/test/kotlin/io/cloudshiftdev/mavensync/SyncMetricsTest.kt @@ -0,0 +1,90 @@ +package io.cloudshiftdev.mavensync + +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.collections.shouldContainExactlyInAnyOrder +import io.kotest.matchers.shouldBe +import io.kotest.matchers.string.shouldContain +import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.Duration.Companion.seconds + +private val g = Group("com.example") +private val a = Artifact("foo") + +private fun v(s: String) = ArtifactVersion(s) + +private fun c(s: String) = Coordinates(g, a, v(s)) + +class SyncMetricsTest : + FunSpec({ + test("empty snapshot has zero totals") { + val report = SyncMetrics().snapshot() + report.inSyncTotal shouldBe 0 + report.syncedTotal shouldBe 0 + report.failedTotal shouldBe 0 + report.assetsTotal shouldBe 0 + report.bytesTotal shouldBe 0L + } + + test("aggregates in-sync, synced and failed totals across artifacts") { + val metrics = SyncMetrics() + metrics.recordInSync(g, a, listOf(v("1.0"), v("1.1"))) + metrics.recordSynced( + c("1.2"), + assetCount = 3, + bytes = 1024L, + duration = 500.milliseconds, + ) + metrics.recordSynced(c("1.3"), assetCount = 2, bytes = 2048L, duration = 1.seconds) + metrics.recordFailure(c("1.4"), error = "HTTP 500", duration = 200.milliseconds) + + val report = metrics.snapshot() + report.inSyncTotal shouldBe 2 + report.syncedTotal shouldBe 2 + report.failedTotal shouldBe 1 + report.assetsTotal shouldBe 5 + report.bytesTotal shouldBe 3072L + } + + test("groups results by artifact and preserves order of records within an artifact") { + val metrics = SyncMetrics() + metrics.recordInSync(g, a, listOf(v("1.0"))) + metrics.recordSynced(c("1.1"), 1, 10L, 1.seconds) + metrics.recordSynced(c("1.2"), 1, 20L, 1.seconds) + metrics.recordFailure(c("1.3"), "bad", 1.seconds) + + val report = metrics.snapshot() + report.artifacts.size shouldBe 1 + val am = report.artifacts.single() + am.inSync.map { it.value } shouldContainExactlyInAnyOrder listOf("1.0") + am.synced.map { it.version.value } shouldBe listOf("1.1", "1.2") + am.failed.map { it.version.value } shouldBe listOf("1.3") + } + + test("renderDetailed and renderSummary include the expected fields") { + val metrics = SyncMetrics() + metrics.recordInSync(g, a, listOf(v("1.0"))) + metrics.recordSynced(c("1.1"), 2, 1024L, 1500.milliseconds) + metrics.recordFailure(c("1.2"), "HTTP 500", 200.milliseconds) + + val report = metrics.snapshot() + val detailed = report.renderDetailed() + detailed shouldContain "com.example:foo" + detailed shouldContain "in sync (1)" + detailed shouldContain "1.0" + detailed shouldContain "synced (1)" + detailed shouldContain "1.1 — 2 assets" + detailed shouldContain "failed (1)" + detailed shouldContain "HTTP 500" + + val summary = report.renderSummary() + summary shouldContain "artifacts: 1" + summary shouldContain "versions in sync: 1" + summary shouldContain "versions synced: 1" + summary shouldContain "versions failed: 1" + summary shouldContain "assets copied: 2" + } + + test("renderDetailed handles an empty report") { + SyncMetrics().snapshot().renderDetailed() shouldContain "(no artifacts processed)" + } + })