Latest changes from Toradex branch

.gitlab-ci.yml

@@ -29,7 +29,7 @@ default:
         - --character-set-server=utf8
         - --collation-server=utf8_unicode_ci
         - --max_connections=1000
-    - name: docker:dind
+    - name: docker:28.5.0-dind
 
 include:
   - project: toradex/torizon-cloud/ci-container-build

build.sbt

@@ -15,32 +15,32 @@ scalacOptions := Seq(
   "-Wconf:cat=other-match-analysis:error"
 )
 
-resolvers += "sonatype-snapshots".at("https://s01.oss.sonatype.org/content/repositories/snapshots")
-resolvers += "sonatype-releases".at("https://s01.oss.sonatype.org/content/repositories/releases")
+resolvers += Resolver.mavenCentral
+resolvers += "maven-snapshots"at "https://central.sonatype.com/repository/maven-snapshots"
 
 Global / bloopAggregateSourceDependencies := true
 
 libraryDependencies ++= {
-  val akkaV = "2.8.5"
-  val akkaHttpV = "10.5.2"
-  val tufV = "3.2.11"
+  val pekkoV = "1.1.5"
+  val pekkoHttpV = "1.2.0"
+  val tufV = "5.0.0"
   val scalaTestV = "3.2.19"
-  val bouncyCastleV = "1.83"
-  val libatsV = "2.6.6"
+  val bouncyCastleV = "1.80"
+  val libatsV = "5.0.0"
 
   Seq(
-    "com.typesafe.akka" %% "akka-actor" % akkaV,
-    "com.typesafe.akka" %% "akka-stream" % akkaV,
-    "com.typesafe.akka" %% "akka-http" % akkaHttpV,
-    "com.typesafe.akka" %% "akka-http-testkit" % akkaHttpV,
-    "com.typesafe.akka" %% "akka-stream-testkit" % akkaV,
-    "com.typesafe.akka" %% "akka-slf4j" % akkaV,
+    "org.apache.pekko" %% "pekko-actor" % pekkoV,
+    "org.apache.pekko" %% "pekko-stream" % pekkoV,
+    "org.apache.pekko" %% "pekko-http" % pekkoHttpV,
+    "org.apache.pekko" %% "pekko-http-testkit" % pekkoHttpV,
+    "org.apache.pekko" %% "pekko-stream-testkit" % pekkoV,
+    "org.apache.pekko" %% "pekko-slf4j" % pekkoV,
     "org.scalatest" %% "scalatest" % scalaTestV % Test,
     "org.scalacheck" %% "scalacheck" % "1.19.0" % Test,
     "io.github.uptane" %% "libats" % libatsV,
     "io.github.uptane" %% "libats-messaging" % libatsV,
     "io.github.uptane" %% "libats-messaging-datatype" % libatsV,
-    "io.github.uptane" %% "libats-metrics-akka" % libatsV,
+    "io.github.uptane" %% "libats-metrics-pekko" % libatsV,
     "io.github.uptane" %% "libats-metrics-prometheus" % libatsV,
     "io.github.uptane" %% "libats-http-tracing" % libatsV,
     "io.github.uptane" %% "libats-slick" % libatsV,
@@ -56,8 +56,8 @@ libraryDependencies ++= {
     "com.beachape" %% "enumeratum-circe" % "1.9.0",
 
     // Device registry specific dependencies
-    "com.lightbend.akka" %% "akka-stream-alpakka-csv" % "2.0.0",
-    "io.circe" %% "circe-testing" % "0.14.15",
+    "org.apache.pekko" %% "pekko-connectors-csv" % "1.0.0",
+    "io.circe" %% "circe-testing" % "0.14.13",
     "tech.sparse" %% "toml-scala" % "0.2.2",
     "org.tpolecat" %% "atto-core" % "0.9.5",
     "org.scalatestplus" %% "scalacheck-1-16" % "3.2.14.0" % Test

src/main/resources/application.conf

@@ -1,16 +1,16 @@
-akka {
-  loggers = ["akka.event.slf4j.Slf4jLogger"]
+pekko {
+  loggers = ["org.apache.pekko.event.slf4j.Slf4jLogger"]
   loglevel = "DEBUG"
-  loglevel = ${?AKKA_LOGLEVEL}
+  loglevel = ${?PEKKO_LOGLEVEL}
   log-config-on-start = off
-  log-config-on-start = ${?AKKA_LOG_CONFIG_ON_START}
+  log-config-on-start = ${?PEKKO_LOG_CONFIG_ON_START}
 
   http {
     server {
       max-connections = 2048
-      max-connections = ${?AKKA_HTTP_MAX_CONNECTIONS}
+      max-connections = ${?PEKKO_HTTP_MAX_CONNECTIONS}
       backlog = 10000
-      backlog = ${?AKKA_HTTP_BACKLOG}
+      backlog = ${?PEKKO_HTTP_BACKLOG}
     }
 
     # older versions of RAC send an "invalid" header
@@ -20,15 +20,15 @@ akka {
       # The maximum number of parallel connections that a connection pool to a
       # single host endpoint is allowed to establish. Must be greater than zero.
       max-connections = 2048
-      max-connections = ${?AKKA_HTTP_CLIENT_MAX_CONNECTIONS}
+      max-connections = ${?PEKKO_HTTP_CLIENT_MAX_CONNECTIONS}
       # The maximum number of open requests accepted into the pool across all
       # materializations of any of its client flows.
       # Protects against (accidentally) overloading a single pool with too many client flow materializations.
       # Note that with N concurrent materializations the max number of open request in the pool
       # will never exceed N * max-connections * pipelining-limit.
       # Must be a power of 2 and > 0!
       max-open-requests = 4096
-      max-open-requests = ${?AKKA_HTTP_CLIENT_MAX_OPEN_REQUESTS}
+      max-open-requests = ${?PEKKO_HTTP_CLIENT_MAX_OPEN_REQUESTS}
     }
   }
 }

src/main/resources/db/migration/director/V16__create_updates_table.sql

@@ -0,0 +1,66 @@
+CREATE TABLE `updates` (
+  `namespace` varchar(255) NOT NULL,
+  `id` char(36) NOT NULL,
+  `device_id` char(36) NOT NULL,
+  `hardware_update_id` char(36) NOT NULL,
+  `correlation_id` varchar(255) NOT NULL,
+  `scheduled_for` datetime(3) NULL,
+  `status` Enum('Assigned', 'Scheduled', 'Completed', 'PartiallyCompleted', 'Cancelled') NOT NULL,
+  `status_info` json DEFAULT NULL,
+  `completed_at` datetime(3) DEFAULT NULL,
+  `created_at` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  `updated_at` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`, `device_id`),
+
+  CONSTRAINT fk_updates_hardware_update
+  FOREIGN KEY (hardware_update_id) REFERENCES hardware_updates (id),
+  FOREIGN KEY (device_id) REFERENCES provisioned_devices(id),
+
+  KEY `updates_status_scheduled_for` (`status`,`scheduled_for`),
+  KEY `idx_updates_device_id` (`device_id`),
+  KEY `idx_updates_namespace` (`namespace`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_unicode_ci;
+
+CREATE INDEX `idx_assignments_namespace_correlation_id` ON `assignments` (`namespace`, `correlation_id`);
+
+INSERT INTO `updates` (
+  `namespace`, 
+  `id`, 
+  `device_id`, 
+  `hardware_update_id`, 
+  `correlation_id`, 
+  `scheduled_for`,
+  `status`, 
+  `status_info`, 
+  `completed_at`, 
+  `created_at`, 
+  `updated_at`
+)
+SELECT 
+  `namespace`, 
+  `id`, 
+  `device_id`, 
+  `hardware_update_id`, 
+  CONCAT('urn:here-ota:mtu:', `hardware_update_id`), 
+  `scheduled_at`, 
+  `status`, 
+  `status_info`, 
+  CASE 
+    WHEN `status` IN ('Completed', 'Cancelled') THEN `updated_at`
+    ELSE NULL
+  END as `completed_at`, 
+  `created_at`, 
+  `updated_at`
+FROM `scheduled_updates`
+ON DUPLICATE KEY UPDATE
+  `status` = VALUES(`status`),
+  `status_info` = VALUES(`status_info`),
+  `completed_at` = CASE 
+    WHEN VALUES(`status`) IN ('Completed', 'Cancelled') THEN VALUES(`updated_at`)
+    ELSE NULL
+  END,
+  `updated_at` = VALUES(`updated_at`)
+  ;
+
+alter table EcuInstallationResult ADD `description` TEXT NULL
+;

src/main/resources/db/migration/director/V17__device_manifests_checksum.sql

@@ -0,0 +1,6 @@
+
+ALTER TABLE device_manifests CHANGE COLUMN sha256 checksum char(8)
+;
+
+CREATE INDEX device_manifests_received_at ON device_manifests(device_id, received_at)
+;

src/main/resources/db/migration/director/V18__device_manifests_index.sql

@@ -0,0 +1,3 @@
+ALTER TABLE device_manifests DROP INDEX IF EXISTS device_manifests_device_id_received_at_idx;
+
+ALTER TABLE device_manifests ADD INDEX device_manifests_device_id_received_at_idx (device_id, received_at);

src/main/resources/db/migration/director/V19__add_update_seen_col.sql

@@ -0,0 +1,3 @@
+ALTER TABLE `updates` MODIFY COLUMN
+  `status` Enum('Scheduled', 'Assigned', 'Seen', 'Completed', 'PartiallyCompleted', 'Cancelled') NOT NULL
+;

src/main/scala/com/advancedtelematic/director/Boot.scala

@@ -1,11 +1,11 @@
 package com.advancedtelematic.director
 
 import com.advancedtelematic.director.http.deviceregistry.TomlSupport.`application/toml`
-import akka.actor.ActorSystem
-import akka.http.scaladsl.Http
-import akka.http.scaladsl.Http.ServerBinding
-import akka.http.scaladsl.server.{Directives, Route}
-import akka.http.scaladsl.settings.{ParserSettings, ServerSettings}
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.http.scaladsl.Http
+import org.apache.pekko.http.scaladsl.Http.ServerBinding
+import org.apache.pekko.http.scaladsl.server.{Directives, Route}
+import org.apache.pekko.http.scaladsl.settings.{ParserSettings, ServerSettings}
 import com.advancedtelematic.director.db.deviceregistry.DeviceRepository
 import com.advancedtelematic.director.deviceregistry.AllowUUIDPath
 import com.advancedtelematic.director.http.DirectorRoutes
@@ -17,23 +17,14 @@ import com.advancedtelematic.libats.http.VersionDirectives.versionHeaders
 import com.advancedtelematic.libats.http.monitoring.ServiceHealthCheck
 import com.advancedtelematic.libats.http.tracing.Tracing
 import com.advancedtelematic.libats.http.tracing.Tracing.ServerRequestTracing
-import com.advancedtelematic.libats.http.{
-  BootApp,
-  BootAppDatabaseConfig,
-  BootAppDefaultConfig,
-  NamespaceDirectives
-}
-import com.advancedtelematic.libats.messaging.MessageBus
+import com.advancedtelematic.libats.http.{BootApp, BootAppDatabaseConfig, BootAppDefaultConfig, NamespaceDirectives}
+import com.advancedtelematic.libats.messaging.{MessageBus, MessageBusPublisher}
 import com.advancedtelematic.libats.messaging_datatype.DataType.DeviceId
 import com.advancedtelematic.libats.slick.db.{CheckMigrations, DatabaseSupport}
 import com.advancedtelematic.libats.slick.monitoring.{DatabaseMetrics, DbHealthResource}
 import com.advancedtelematic.libtuf_server.keyserver.KeyserverHttpClient
 import com.advancedtelematic.metrics.prometheus.PrometheusMetricsSupport
-import com.advancedtelematic.metrics.{
-  AkkaHttpConnectionMetrics,
-  AkkaHttpRequestMetrics,
-  MetricsSupport
-}
+import com.advancedtelematic.metrics.{PekkoHttpConnectionMetrics, PekkoHttpRequestMetrics, MetricsSupport}
 import com.codahale.metrics.MetricRegistry
 import com.typesafe.config.Config
 import org.bouncycastle.jce.provider.BouncyCastleProvider
@@ -53,8 +44,8 @@ class DirectorBoot(override val globalConfig: Config,
     with DatabaseSupport
     with MetricsSupport
     with DatabaseMetrics
-    with AkkaHttpRequestMetrics
-    with AkkaHttpConnectionMetrics
+    with PekkoHttpRequestMetrics
+    with PekkoHttpConnectionMetrics
     with PrometheusMetricsSupport
     with CheckMigrations {
 
@@ -69,7 +60,7 @@ class DirectorBoot(override val globalConfig: Config,
 
   private def keyserverClient(implicit tracing: ServerRequestTracing) = KeyserverHttpClient(tufUri)
 
-  private implicit val msgPublisher: com.advancedtelematic.libats.messaging.MessageBusPublisher =
+  private implicit val msgPublisher: MessageBusPublisher =
     MessageBus.publisher(system, globalConfig)
 
   private lazy val authNamespace = NamespaceDirectives.fromConfig()

src/main/scala/com/advancedtelematic/director/Settings.scala

@@ -1,7 +1,7 @@
 package com.advancedtelematic.director
 
-import akka.event.Logging
-import akka.http.scaladsl.model.Uri
+import org.apache.pekko.event.Logging
+import org.apache.pekko.http.scaladsl.model.Uri
 import com.typesafe.config.ConfigFactory
 
 trait Settings {

src/main/scala/com/advancedtelematic/director/daemon/DaemonBoot.scala

@@ -1,9 +1,9 @@
 package com.advancedtelematic.director.daemon
 
-import akka.actor.ActorSystem
-import akka.http.scaladsl.Http
-import akka.http.scaladsl.Http.ServerBinding
-import akka.http.scaladsl.server.Directives
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.http.scaladsl.Http
+import org.apache.pekko.http.scaladsl.Http.ServerBinding
+import org.apache.pekko.http.scaladsl.server.Directives
 import com.advancedtelematic.director.{Settings, VersionInfo}
 import com.advancedtelematic.libats.http.{BootApp, BootAppDatabaseConfig, BootAppDefaultConfig}
 import com.advancedtelematic.libats.messaging.{
@@ -68,6 +68,8 @@ class DirectorDaemonBoot(override val globalConfig: Config,
     startMonitoredListener[DeviceUpdateEvent](new DeviceUpdateEventListener(messageBus))
     startMonitoredListener[DeviceMqttLifecycle](new MqttLifecycleListener)
 
+    new DeviceManifestReportedListener(globalConfig).start()
+
     val routes = versionHeaders(version) {
       prometheusMetricsRoutes ~
         DbHealthResource(