diff --git a/build.sbt b/build.sbt index ee602a9..333dbfb 100644 --- a/build.sbt +++ b/build.sbt @@ -25,15 +25,15 @@ lazy val versions = new { // pekko val pekko = "1.6.0" val pekkoHttp = "1.3.0" - val pekkoJson = "3.9.1" + val pekkoJson = "3.9.2" // persistence val slick = "3.6.1" val h2 = "2.4.240" // telemetry - val openTelemetry = "1.62.0" - val openTelemetryPrometheus = "1.62.0-alpha" + val openTelemetry = "1.64.0" + val openTelemetryPrometheus = "1.64.0-alpha" val prometheus = "0.16.0" // testing @@ -45,7 +45,7 @@ lazy val versions = new { // misc val playJson = "2.10.8" val jose4j = "0.9.6" - val logback = "1.5.32" + val logback = "1.5.38" } lazy val root = project diff --git a/lib/src/main/scala/io/github/sndnv/layers/telemetry/analytics/DefaultAnalyticsCollector.scala b/lib/src/main/scala/io/github/sndnv/layers/telemetry/analytics/DefaultAnalyticsCollector.scala index 9378f9f..8bc6636 100644 --- a/lib/src/main/scala/io/github/sndnv/layers/telemetry/analytics/DefaultAnalyticsCollector.scala +++ b/lib/src/main/scala/io/github/sndnv/layers/telemetry/analytics/DefaultAnalyticsCollector.scala @@ -2,6 +2,7 @@ package io.github.sndnv.layers.telemetry.analytics import java.time.Instant +import scala.concurrent.ExecutionContext import scala.concurrent.Future import scala.concurrent.duration.* import scala.util.Failure @@ -66,17 +67,31 @@ object DefaultAnalyticsCollector { Behaviors.withTimers { implicit scheduler => Behaviors.receive { case (ctx, LoadState) => - ctx.pipeToSelf(persistence.restore()) { + implicit val ec: ExecutionContext = ctx.executionContext + ctx.pipeToSelf( + persistence.restore().flatMap { + case Some(entry) if entry.runtime.app != app.asString() => + val fresh = AnalyticsEntry.collected(app) + persistence + .transmit(entry) + .map { _ => persistence.cache(entry = fresh); fresh } + .recover { case _ => entry.asCollected() } + + case Some(entry) => + Future.successful(entry.asCollected()) + + case None => + Future.successful(AnalyticsEntry.collected(app)) + } + ) { case Success(entry) => - val actual = entry.map(_.asCollected()).getOrElse(AnalyticsEntry.collected(app)) - ctx.log.debug( "Analytics state successfully loaded with [events={},failures={}]", - actual.events.length, - actual.failures.length + entry.events.length, + entry.failures.length ) - StateLoaded(entry = actual) + StateLoaded(entry = entry) case Failure(e) => ctx.log.error( diff --git a/lib/src/test/scala/io/github/sndnv/layers/telemetry/analytics/DefaultAnalyticsCollectorSpec.scala b/lib/src/test/scala/io/github/sndnv/layers/telemetry/analytics/DefaultAnalyticsCollectorSpec.scala index 42f8e2f..f40605d 100644 --- a/lib/src/test/scala/io/github/sndnv/layers/telemetry/analytics/DefaultAnalyticsCollectorSpec.scala +++ b/lib/src/test/scala/io/github/sndnv/layers/telemetry/analytics/DefaultAnalyticsCollectorSpec.scala @@ -127,6 +127,78 @@ class DefaultAnalyticsCollectorSpec extends UnitSpec with Eventually { } } + it should "roll over cached state when the app version changes" in withRetry { + val persistence = MockAnalyticsPersistence( + existing = AnalyticsEntry + .collected(app = previousApp) + .withEvent(name = "existing_event", attributes = Map.empty) + .withFailure(message = "Existing failure") + ) + + val collector = DefaultAnalyticsCollector( + name = "test-analytics-collector", + config = config, + persistence = persistence, + app = currentApp + ) + + collector.recordEvent("test_event") + + collector.state.map { state => + state.runtime.app should be(currentApp.asString()) + state.events.map(_.event) should be(Seq("test_event")) + state.failures should be(empty) + + persistence.transmitted.toList match { + case transmitted :: Nil => + transmitted.runtime.app should be(previousApp.asString()) + transmitted.events.map(_.event) should be(Seq("existing_event")) + transmitted.failures.map(_.message) should be(Seq("Existing failure")) + + case other => + fail(s"Unexpected result received: [$other]") + } + + persistence.cached.toList match { + case cached :: Nil => + cached.runtime.app should be(currentApp.asString()) + cached.events should be(empty) + cached.failures should be(empty) + + case other => + fail(s"Unexpected result received: [$other]") + } + } + } + + it should "retain cached state when roll-over transmission fails" in withRetry { + val existing = AnalyticsEntry + .collected(app = previousApp) + .withEvent(name = "existing_event", attributes = Map.empty) + + val persistence = new MockAnalyticsPersistence(existing = Success(Some(existing))) { + override def transmit(entry: AnalyticsEntry): Future[Done] = + Future.failed(new RuntimeException("Test failure")) + } + + val collector = DefaultAnalyticsCollector( + name = "test-analytics-collector", + config = config, + persistence = persistence, + app = currentApp + ) + + collector.recordEvent("test_event") + + collector.state.map { state => + state.runtime.app should be(previousApp.asString()) + state.events.map(_.event) should be(Seq("existing_event", "test_event")) + + persistence.cached should be(empty) + persistence.transmitted should be(empty) + } + } + it should "handle failures when loading cached state" in withRetry { val persistence = MockAnalyticsPersistence(existing = new RuntimeException("Test failure")) @@ -439,4 +511,16 @@ class DefaultAnalyticsCollectorSpec extends UnitSpec with Eventually { persistenceInterval = 3.seconds, transmissionInterval = 10.minutes ) + + private val previousApp: ApplicationInformation = new ApplicationInformation { + override val name: String = "test-app" + override val version: String = "previous" + override val buildTime: Long = 0L + } + + private val currentApp: ApplicationInformation = new ApplicationInformation { + override val name: String = "test-app" + override val version: String = "current" + override val buildTime: Long = 0L + } } diff --git a/project/build.properties b/project/build.properties index dabdb15..7c95fc1 100644 --- a/project/build.properties +++ b/project/build.properties @@ -1 +1 @@ -sbt.version=1.12.11 +sbt.version=1.12.13 diff --git a/project/plugins.sbt b/project/plugins.sbt index ea87b63..85cdd3b 100644 --- a/project/plugins.sbt +++ b/project/plugins.sbt @@ -1,7 +1,8 @@ addSbtPlugin("org.scoverage" % "sbt-scoverage" % "2.4.4") -addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.6.0") -addSbtPlugin("org.wartremover" % "sbt-wartremover" % "3.5.6") -addSbtPlugin("com.timushev.sbt" % "sbt-updates" % "0.6.3") -addSbtPlugin("com.github.sbt" % "sbt-ci-release" % "1.11.2") -addSbtPlugin("com.typesafe" % "sbt-mima-plugin" % "1.1.5") +addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.6.1") +addSbtPlugin("org.wartremover" % "sbt-wartremover" % "3.6.1") +addSbtPlugin("com.timushev.sbt" % "sbt-updates" % "0.7.0") +addSbtPlugin("com.github.sbt" % "sbt-ci-release" % "1.12.0") +addSbtPlugin("com.typesafe" % "sbt-mima-plugin" % "1.1.6") addSbtPlugin("com.github.sbt" %% "sbt-sbom" % "0.5.0") +addSbtPlugin("com.github.sbt" % "sbt-git" % "2.1.0")