Add periodic topology snapshot trigger (#3295)

* Add trigger for periodic topology snapshots (#3188)

Signed-off-by: Julien Tinguely <julien.tinguely@digitalasset.com>

Author: julientinguely-da

apps/app/src/main/resources/application.conf

@@ -1,4 +1,5 @@
 # This is deliberately lower than ConsoleCommandTimeout.defaultRequestTimeout
 pekko.http.server.request-timeout = 38 seconds
 pekko.http.server.parsing.error-handler="org.lfdecentralizedtrust.splice.http.PekkoHttpParsingErrorHandler$"
-pekko.http.server.parsing.ignore-illegal-header-for = ["user-agent"]
+pekko.http.server.parsing.ignore-illegal-header-for = ["user-agent", "etag"]
+pekko.http.client.parsing.ignore-illegal-header-for = ["etag"]

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/PeriodicBackupIntegrationTest.scala

@@ -4,6 +4,7 @@ import org.lfdecentralizedtrust.splice.config.{
   BackupDumpConfig,
   ConfigTransforms,
   GcpBucketConfig,
+  IdentityDumpTest,
   PeriodicBackupDumpConfig,
 }
 import org.lfdecentralizedtrust.splice.identities.NodeIdentitiesDump
@@ -98,7 +99,7 @@ final class DirectoryPeriodicBackupIntegrationTest
 final class GcpBucketPeriodicBackupIntegrationTest
     extends PeriodicBackupIntegrationTestBase[BackupDumpConfig.Gcp] {
   override def backupDumpLocation =
-    BackupDumpConfig.Gcp(GcpBucketConfig.inferForTesting, None)
+    BackupDumpConfig.Gcp(GcpBucketConfig.inferForTesting(IdentityDumpTest), None)
   val bucket = new GcpBucket(backupDumpLocation.bucket, loggerFactory)
   override def readDump(filename: String) = {
     bucket.readStringFromBucket(Paths.get(filename))

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/PeriodicTopologySnapshotIntegrationTest.scala

@@ -0,0 +1,78 @@
+package org.lfdecentralizedtrust.splice.integration.tests
+
+import com.digitalasset.canton.config.NonNegativeFiniteDuration
+import com.digitalasset.canton.logging.SuppressionRule
+import com.google.cloud.storage.Blob
+import org.lfdecentralizedtrust.splice.config.*
+import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition
+import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.IntegrationTest
+import org.lfdecentralizedtrust.splice.util.GcpBucket
+import org.slf4j.event.Level
+
+import java.nio.charset.StandardCharsets
+import java.time.{ZoneOffset, ZonedDateTime}
+import scala.concurrent.duration.DurationInt
+
+class PeriodicTopologySnapshotIntegrationTest[T <: BackupDumpConfig] extends IntegrationTest {
+
+  private val bucket = new GcpBucket(topologySnapshotLocation.bucket, loggerFactory)
+
+  private val topologySnapshotInterval: NonNegativeFiniteDuration =
+    NonNegativeFiniteDuration.ofMinutes(10)
+
+  private def topologySnapshotConfig: PeriodicBackupDumpConfig =
+    PeriodicBackupDumpConfig(topologySnapshotLocation, topologySnapshotInterval)
+
+  private def topologySnapshotLocation: BackupDumpConfig.Gcp =
+    BackupDumpConfig.Gcp(GcpBucketConfig.inferForTesting(TopologySnapshotTest), None)
+
+  private def listDump(utcDate: String): Seq[Blob] =
+    bucket.list(startOffset = s"topology_snapshot_$utcDate", endOffset = "")
+
+  override def environmentDefinition: SpliceEnvironmentDefinition =
+    EnvironmentDefinition
+      .simpleTopology1Sv(this.getClass.getSimpleName)
+      .addConfigTransformsToFront((_, conf) =>
+        ConfigTransforms.updateAllSvAppConfigs((_, c) =>
+          c.copy(topologySnapshotConfig = Some(topologySnapshotConfig))
+        )(conf)
+      )
+      .withManualStart
+
+  "sv1" should {
+    "produces a topology snapshot in the background" in { implicit env =>
+      val utcDate = ZonedDateTime.now(ZoneOffset.UTC).toLocalDate.toString
+      clue("topology snapshot is being produced")(
+        loggerFactory.assertEventuallyLogsSeq(SuppressionRule.Level(Level.INFO))(
+          timeUntilSuccess = 40.seconds,
+          within = {
+            initDsoWithSv1Only()
+          },
+          assertion = logEntries => {
+            forAtLeast(
+              1,
+              logEntries,
+            )(logEntry =>
+              logEntry.message should (include(s"Took a new topology snapshot on $utcDate") or
+                include("Today's topology snapshot already exists."))
+            )
+          },
+        )
+      )
+
+      clue("the 3 files created from the topology snapshot exist in gcp.")({
+        val dumps = listDump(utcDate).filter(_.getSize > 0L)
+        dumps.size shouldBe 3
+        new String(
+          dumps
+            .find(_.getName.endsWith("metadata"))
+            .getOrElse(throw new RuntimeException("Metadata file not found."))
+            .getContent(),
+          StandardCharsets.UTF_8,
+        ) should include("SEQ::sv1") // the sequencerId change through ci runs
+        dumps.exists(_.getName.endsWith("authorized")) shouldBe true
+        dumps.exists(_.getName.endsWith("onboarding-state")) shouldBe true
+      })
+    }
+  }
+}

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/runbook/NonDsoNonDevNetPreflightIntegrationTest.scala

@@ -1,9 +1,10 @@
 package org.lfdecentralizedtrust.splice.integration.tests.runbook
 
+import org.lfdecentralizedtrust.splice.config.IdentityDump
 import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition
 import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.{
-  SpliceTestConsoleEnvironment,
   IntegrationTestWithSharedEnvironment,
+  SpliceTestConsoleEnvironment,
 }
 import org.lfdecentralizedtrust.splice.util.DataExportTestUtil
 
@@ -41,11 +42,11 @@ final class NonDsoNonDevNetPreflightIntegrationTest
   }
 
   "Check that there is a recent participant identities backup on GCP for validator1" in { _ =>
-    testRecentParticipantIdentitiesDump("validator1")
+    testRecentParticipantIdentitiesDump("validator1", IdentityDump)
   }
 
   "Check that there is a recent participant identities backup on GCP for splitwell validator" in {
     _ =>
-      testRecentParticipantIdentitiesDump("splitwell")
+      testRecentParticipantIdentitiesDump("splitwell", IdentityDump)
   }
 }

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/integration/tests/runbook/SvNonDevNetPreflightintegrationTest.scala

@@ -1,11 +1,12 @@
 package org.lfdecentralizedtrust.splice.integration.tests.runbook
 
+import org.lfdecentralizedtrust.splice.config.IdentityDump
 import org.lfdecentralizedtrust.splice.integration.EnvironmentDefinition
 import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.SpliceTestConsoleEnvironment
 import org.lfdecentralizedtrust.splice.integration.tests.FrontendIntegrationTestWithSharedEnvironment
 import org.lfdecentralizedtrust.splice.util.DataExportTestUtil
-
 import org.lfdecentralizedtrust.splice.util.FrontendLoginUtil
+
 import scala.util.Try
 
 abstract class SvNonDevNetPreflightIntegrationTestBase
@@ -113,7 +114,7 @@ abstract class SvNonDevNetPreflightIntegrationTestBase
   }
 
   "Check that there is a recent participant identities backup on GCP" in { _ =>
-    testRecentParticipantIdentitiesDump(svNamespace)
+    testRecentParticipantIdentitiesDump(svNamespace, IdentityDump)
   }
 }
 

apps/app/src/test/scala/org/lfdecentralizedtrust/splice/util/DataExportTestUtil.scala

@@ -1,6 +1,6 @@
 package org.lfdecentralizedtrust.splice.util
 
-import org.lfdecentralizedtrust.splice.config.GcpBucketConfig
+import org.lfdecentralizedtrust.splice.config.{BucketName, GcpBucketConfig}
 import org.lfdecentralizedtrust.splice.identities.{NodeIdentitiesDump, NodeIdentitiesStore}
 import org.lfdecentralizedtrust.splice.integration.tests.SpliceTests.TestCommon
 import com.digitalasset.canton.topology.ParticipantId
@@ -15,8 +15,9 @@ trait DataExportTestUtil extends TestCommon {
       namespace: String,
       getFileName: Instant => Path,
       decode: String => Either[String, A],
+      bucketName: BucketName,
   ) = {
-    val bucket = new GcpBucket(GcpBucketConfig.inferForCluster, loggerFactory)
+    val bucket = new GcpBucket(GcpBucketConfig.inferForCluster(bucketName), loggerFactory)
     import java.time.Instant
     import java.time.temporal.ChronoUnit
     val cluster = sys.env("GCP_CLUSTER_BASENAME")
@@ -33,10 +34,11 @@ trait DataExportTestUtil extends TestCommon {
     }
   }
 
-  def testRecentParticipantIdentitiesDump(namespace: String) =
+  def testRecentParticipantIdentitiesDump(namespace: String, bucketName: BucketName) =
     testRecentDump(
       namespace,
       NodeIdentitiesStore.dumpFilename(_),
       NodeIdentitiesDump.fromJsonString(ParticipantId.tryFromProtoPrimitive, _),
+      bucketName,
     )
 }

apps/common/src/main/scala/org/lfdecentralizedtrust/splice/config/GcpConfig.scala

@@ -48,16 +48,36 @@ final case class GcpBucketConfig(
   def description: String = s"GCP bucket $bucketName in project $projectId"
 }
 
+sealed trait BucketName {
+  def value: String
+}
+
+case object IdentityDumpTest extends BucketName {
+  val value: String = "da-splice-identity-dumps"
+}
+
+case object TopologySnapshotTest extends BucketName {
+  val value: String = "da-splice-topology-snapshots"
+}
+
+case object IdentityDump extends BucketName {
+  val value: String = "da-cn-data-dumps"
+}
+
+case object TopologySnapshot extends BucketName {
+  val value: String = "da-cn-topology-snapshots"
+}
+
 object GcpBucketConfig {
-  def inferForTesting: GcpBucketConfig =
+  def inferForTesting(testBucketName: BucketName): GcpBucketConfig =
     infer(
       "GCP_DATA_EXPORT_INTEGRATION_TEST_SERVICE_ACCOUNT_CREDENTIALS",
       "da-cn-splice",
-      "da-splice-identity-dumps",
+      testBucketName.value,
     )
 
-  def inferForCluster: GcpBucketConfig =
-    infer("GCP_DATA_DUMP_BUCKET_SERVICE_ACCOUNT_CREDENTIALS", "da-cn-devnet", "da-cn-data-dumps")
+  def inferForCluster(bucketName: BucketName): GcpBucketConfig =
+    infer("GCP_DATA_DUMP_BUCKET_SERVICE_ACCOUNT_CREDENTIALS", "da-cn-devnet", bucketName.value)
 
   def inferForBundles: GcpBucketConfig =
     infer(