Preseed stats on boot with measurements from database

Author: Michel Zimmer

Dockerfile

@@ -12,10 +12,10 @@ LABEL org.opencontainers.image.vendor="neuland – Büro für Informatik GmbH"
 LABEL org.opencontainers.image.licenses="Apache-2.0"
 LABEL org.opencontainers.image.title="bandwhichd-server"
 LABEL org.opencontainers.image.description="bandwhichd server collecting measurements and calculating statistics"
-LABEL org.opencontainers.image.version="0.6.0-rc4"
+LABEL org.opencontainers.image.version="0.6.0-rc5"
 USER guest
 ENTRYPOINT ["/opt/java/openjdk/bin/java"]
 CMD ["-jar", "/opt/bandwhichd-server.jar"]
 EXPOSE 8080
 STOPSIGNAL SIGTERM
-COPY --from=build --chown=root:root /tmp/bandwhichd-server/target/scala-3.1.3/bandwhichd-server-assembly-0.6.0-rc4.jar /opt/bandwhichd-server.jar
+COPY --from=build --chown=root:root /tmp/bandwhichd-server/target/scala-3.1.3/bandwhichd-server-assembly-0.6.0-rc5.jar /opt/bandwhichd-server.jar

build.sbt

@@ -2,7 +2,7 @@ lazy val root = (project in file("."))
   .settings(
     organization := "de.neuland-bfi",
     name := "bandwhichd-server",
-    version := "0.6.0-rc4",
+    version := "0.6.0-rc5",
     scalaVersion := "3.1.3",
     Compile / scalaSource := baseDirectory.value / "src" / "main" / "scala",
     Test / scalaSource := baseDirectory.value / "src" / "test" / "scala",

src/main/scala/de/neuland/bandwhichd/server/boot/App.scala

@@ -13,9 +13,13 @@ import de.neuland.bandwhichd.server.application.{
   MeasurementApplicationService,
   StatsApplicationService
 }
-import de.neuland.bandwhichd.server.domain.measurement.MeasurementsRepository
-import de.neuland.bandwhichd.server.domain.stats.StatsRepository
+import de.neuland.bandwhichd.server.domain.measurement.{
+  MeasurementsRepository,
+  Timing
+}
+import de.neuland.bandwhichd.server.domain.stats.{Stats, StatsRepository}
 import de.neuland.bandwhichd.server.lib.cassandra.CassandraContext
+import de.neuland.bandwhichd.server.lib.time.Interval
 import de.neuland.bandwhichd.server.lib.time.cats.TimeContext
 import org.http4s.dsl.io.*
 import org.http4s.ember.server.EmberServerBuilder
@@ -25,6 +29,7 @@ import org.typelevel.log4cats.slf4j.Slf4jLogger
 import org.typelevel.log4cats.{Logger, SelfAwareStructuredLogger}
 
 import java.io.{BufferedReader, InputStreamReader}
+import java.time.Duration
 import java.util.Scanner
 import scala.io.StdIn
 
@@ -91,11 +96,32 @@ object App extends IOApp.Simple {
       _ <- Resource.eval(
         CassandraMigration(cassandraContext).migrate(configuration)
       )
-      main = App[IO](
-        TimeContext.systemTimeContext,
+      timeContext = TimeContext.systemTimeContext[IO]
+      main: App[IO] = App[IO](
+        timeContext,
         cassandraContext,
         configuration
       )
+      _ <- Resource.eval {
+        for {
+          now <- timeContext.now
+          preseedDuration = Stats.defaultTimeframeDuration.plus(
+            Duration.ofMinutes(15)
+          )
+          initialStats <- main.statsRepository.get
+          preseededStats <- main.measurementsRepository
+            .get(
+              Timing.Timeframe(
+                Interval(now.minus(preseedDuration), preseedDuration)
+              )
+            )
+            .compile
+            .fold(initialStats) { case (stats, measurement) =>
+              stats.append(measurement)
+            }
+          _ <- main.statsRepository.safe(preseededStats)
+        } yield ()
+      }
       server <- EmberServerBuilder
         .default[IO]
         .withHostOption(None)

src/main/scala/de/neuland/bandwhichd/server/boot/Configuration.scala

@@ -19,8 +19,7 @@ case class Configuration(
     measurementNetworkConfigurationTTL: Duration,
     measurementNetworkUtilizationTTL: Duration,
     recordMeasurementQueryTimeout: Duration,
-    getAllMeasurementsQueryTimeout: Duration,
-    aggregationSchedulerInterval: Duration
+    getAllMeasurementsQueryTimeout: Duration
 )
 
 object Configuration {
@@ -33,8 +32,7 @@ object Configuration {
       measurementNetworkConfigurationTTL: String,
       measurementNetworkUtilizationTTL: String,
       recordMeasurementQueryTimeout: String,
-      getAllMeasurementsQueryTimeout: String,
-      aggregationSchedulerInterval: String
+      getAllMeasurementsQueryTimeout: String
   ): F[Configuration] = {
 
     val maybeHostnameContactPoints = contactPoints
@@ -74,9 +72,7 @@ object Configuration {
       recordMeasurementQueryTimeout =
         Duration.parse(recordMeasurementQueryTimeout),
       getAllMeasurementsQueryTimeout =
-        Duration.parse(getAllMeasurementsQueryTimeout),
-      aggregationSchedulerInterval =
-        Duration.parse(aggregationSchedulerInterval)
+        Duration.parse(getAllMeasurementsQueryTimeout)
     )
   }
 
@@ -94,8 +90,7 @@ object Configuration {
       scala.util.Properties
         .envOrElse("RECORD_MEASUREMENT_QUERY_TIMEOUT", "PT2S"),
       scala.util.Properties
-        .envOrElse("GET_ALL_MEASUREMENTS_QUERY_TIMEOUT", "PT8S"),
-      scala.util.Properties.envOrElse("AGGREGATION_SCHEDULER_INTERVAL", "PT10S")
+        .envOrElse("GET_ALL_MEASUREMENTS_QUERY_TIMEOUT", "PT8S")
     )
 
   def resource[F[_]: Sync]: Resource[F, Configuration] =

src/main/scala/de/neuland/bandwhichd/server/domain/stats/Stats.scala

@@ -56,13 +56,6 @@ object Stats {
 
   val empty: MonitoredStats = new Stats(Map.empty)
 
-  def apply[F[_]: Concurrent](
-      measurements: Stream[F, Measurement[Timing]]
-  ): F[MonitoredStats] =
-    measurements.compile.fold(Stats.empty) { case (stats, measurement) =>
-      stats.append(measurement)
-    }
-
   extension (stats: MonitoredStats) {
     def append(
         measurement: Measurement[Timing]

src/test/scala/de/neuland/bandwhichd/server/boot/ConfigurationFixtures.scala

@@ -16,8 +16,7 @@ object ConfigurationFixtures {
       measurementNetworkConfigurationTTL = Duration.ofHours(2),
       measurementNetworkUtilizationTTL = Duration.ofHours(2),
       recordMeasurementQueryTimeout = Duration.ofSeconds(4),
-      getAllMeasurementsQueryTimeout = Duration.ofSeconds(8),
-      aggregationSchedulerInterval = Duration.ofSeconds(10)
+      getAllMeasurementsQueryTimeout = Duration.ofSeconds(8)
     )
 
   def testDefaults(container: CassandraContainer): Configuration =