Make ScyllaDB replication factor configurable

ndr-ds · ndr-ds · commit 6fd307a685d1 · 2025-04-16T19:21:46.000-03:00
diff --git a/docker/Dockerfile b/docker/Dockerfile
@@ -115,6 +115,7 @@ COPY --from=binaries \
 COPY --chmod=755 \
     docker/server-entrypoint.sh \
     docker/server-init.sh \
+    docker/proxy-entrypoint.sh \
     docker/proxy-init.sh \
     docker/compose-server-entrypoint.sh \
     docker/compose-proxy-entrypoint.sh \
diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml
@@ -42,7 +42,7 @@ services:
   shard-init:
     image: "${LINERA_IMAGE:-linera}"
     container_name: shard-init
-    command: [ "./compose-server-init.sh" ]
+    command: [ "./compose-server-init.sh", "1"]
     volumes:
       - .:/config
     depends_on:
diff --git a/docker/proxy-entrypoint.sh b/docker/proxy-entrypoint.sh
@@ -0,0 +1,9 @@
+#!/bin/sh
+
+storage_replication_factor=$1
+
+exec ./linera-proxy \
+    --storage scylladb:tcp:scylla-client.scylla.svc.cluster.local:9042 \
+    --genesis /config/genesis.json \
+    --storage-replication-factor $storage_replication_factor \
+    /config/server.json
diff --git a/docker/server-entrypoint.sh b/docker/server-entrypoint.sh
@@ -1,10 +1,13 @@
 #!/bin/sh
 
+storage_replication_factor=$1
+
 # Extract the ordinal number from the pod hostname
 ORDINAL="${HOSTNAME##*-}"
 
 exec ./linera-server run \
   --storage scylladb:tcp:scylla-client.scylla.svc.cluster.local:9042 \
   --server /config/server.json \
   --shard $ORDINAL \
-  --genesis /config/genesis.json
+  --genesis /config/genesis.json \
+  --storage-replication-factor $storage_replication_factor
diff --git a/docker/server-init.sh b/docker/server-init.sh
@@ -1,5 +1,6 @@
 #!/bin/sh
 
+storage_replication_factor=$1
 while true; do
   ./linera storage check-existence --storage "scylladb:tcp:scylla-client.scylla.svc.cluster.local:9042"
   status=$?
@@ -11,7 +12,8 @@ while true; do
     echo "Database does not exist, attempting to initialize..."
     if ./linera-server initialize \
       --storage scylladb:tcp:scylla-client.scylla.svc.cluster.local:9042 \
-      --genesis /config/genesis.json; then
+      --genesis /config/genesis.json \
+      --storage-replication-factor $storage_replication_factor; then
       echo "Initialization successful."
       exit 0
     else
diff --git a/kubernetes/linera-validator/helmfile.yaml b/kubernetes/linera-validator/helmfile.yaml
@@ -42,7 +42,7 @@ releases:
       - scylla-manager/scylla-manager
       - scylla-operator/scylla-operator
     values:
-      - {{ env "LINERA_HELMFILE_VALUES_SCYLLA" | default "scylla.values.yaml" }}
+      - {{ env "LINERA_HELMFILE_VALUES_SCYLLA" | default "scylla.values.yaml.gotmpl" }}
   - name: scylla-manager
     version: v1.13.0
     namespace: scylla-manager
diff --git a/kubernetes/linera-validator/scylla.values.yaml.gotmpl b/kubernetes/linera-validator/scylla.values.yaml.gotmpl
@@ -4,7 +4,7 @@ sysctls:
 datacenter: validator
 racks:
   - name: rack-1
-    members: 1
+    members: {{ env "LINERA_HELMFILE_SET_SCYLLA_REPLICATION_FACTOR" | default 1 }}
     scyllaConfig: "scylla-config"
     storage:
       capacity: 2Gi
diff --git a/kubernetes/linera-validator/templates/proxy.yaml b/kubernetes/linera-validator/templates/proxy.yaml
@@ -69,15 +69,7 @@ spec:
               name: linera-port
             - containerPort: 20100
               name: linera-port-int
-          command:
-            [
-              "./linera-proxy",
-              "--storage",
-              "scylladb:tcp:scylla-client.scylla.svc.cluster.local:9042",
-              "--genesis",
-              "/config/genesis.json",
-            ]
-          args: ["/config/server.json"]
+          command: ["./proxy-entrypoint.sh", {{ .Values.storageReplicationFactor | quote }}]
           env:
             - name: RUST_LOG
               value: {{ .Values.logLevel }}
diff --git a/kubernetes/linera-validator/templates/shards.yaml b/kubernetes/linera-validator/templates/shards.yaml
@@ -35,7 +35,7 @@ spec:
         - name: linera-server-initializer
           image: {{ .Values.lineraImage }}
           imagePullPolicy: {{ .Values.lineraImagePullPolicy }}
-          command: ["./server-init.sh"]
+          command: ["./server-init.sh", {{ .Values.storageReplicationFactor | quote }}]
           env:
             - name: RUST_LOG
               value: {{ .Values.logLevel }}
@@ -49,7 +49,7 @@ spec:
         - name: linera-server
           image: {{ .Values.lineraImage }}
           imagePullPolicy: {{ .Values.lineraImagePullPolicy }}
-          command: ["./server-entrypoint.sh"]
+          command: ["./server-entrypoint.sh", {{ .Values.storageReplicationFactor | quote }}]
           env:
             - name: RUST_LOG
               value: {{ .Values.logLevel }}
diff --git a/kubernetes/linera-validator/values-local.yaml.gotmpl b/kubernetes/linera-validator/values-local.yaml.gotmpl
@@ -7,6 +7,7 @@ logLevel: "debug"
 proxyPort: 19100
 metricsPort: 21100
 numShards: {{ env "LINERA_HELMFILE_SET_NUM_SHARDS" | default 10 }}
+storageReplicationFactor: {{ env "LINERA_HELMFILE_SET_STORAGE_REPLICATION_FACTOR" | default 1 }}
 
 # Loki
 loki-stack:
diff --git a/linera-indexer/lib/src/scylla_db.rs b/linera-indexer/lib/src/scylla_db.rs
@@ -41,6 +41,10 @@ pub struct ScyllaDbConfig {
     /// The maximal number of entries in the storage cache.
     #[arg(long, default_value = "1000")]
     pub max_cache_entries: usize,
+
+    /// The replication factor for the keyspace
+    #[arg(long, default_value = "1")]
+    pub replication_factor: u32,
 }
 
 pub type ScyllaDbRunner = Runner<ScyllaDbStore, ScyllaDbConfig>;
@@ -59,7 +63,11 @@ impl ScyllaDbRunner {
             storage_cache_config,
         };
         let namespace = config.client.table.clone();
-        let store_config = ScyllaDbStoreConfig::new(config.client.uri.clone(), common_config);
+        let store_config = ScyllaDbStoreConfig::new(
+            config.client.uri.clone(),
+            common_config,
+            config.client.replication_factor,
+        );
         let store = ScyllaDbStore::connect(&store_config, &namespace).await?;
         Self::new(config, store).await
     }
diff --git a/linera-service/src/linera/main.rs b/linera-service/src/linera/main.rs
@@ -1315,6 +1315,10 @@ struct ClientOptions {
     /// Subcommand.
     #[command(subcommand)]
     command: ClientCommand,
+
+    /// The replication factor for the keyspace
+    #[arg(long, default_value = "1")]
+    storage_replication_factor: u32,
 }
 
 impl ClientOptions {
@@ -1339,7 +1343,7 @@ impl ClientOptions {
         let storage_config = self.storage_config()?;
         debug!("Running command using storage configuration: {storage_config}");
         let store_config = storage_config
-            .add_common_config(self.common_config())
+            .add_common_config(self.common_config(), self.storage_replication_factor)
             .await?;
         let genesis_config = self.wallet().await?.genesis_config().clone();
         let output = Box::pin(store_config.run_with_storage(
@@ -1355,7 +1359,7 @@ impl ClientOptions {
         let storage_config = self.storage_config()?;
         debug!("Running command using storage configuration: {storage_config}");
         let store_config = storage_config
-            .add_common_config(self.common_config())
+            .add_common_config(self.common_config(), self.storage_replication_factor)
             .await?;
         let output = Box::pin(store_config.run_with_store(job)).await?;
         Ok(output)
@@ -1365,7 +1369,7 @@ impl ClientOptions {
         let storage_config = self.storage_config()?;
         debug!("Initializing storage using configuration: {storage_config}");
         let store_config = storage_config
-            .add_common_config(self.common_config())
+            .add_common_config(self.common_config(), self.storage_replication_factor)
             .await?;
         let wallet = self.wallet().await?;
         store_config.initialize(wallet.genesis_config()).await?;
diff --git a/linera-service/src/proxy/main.rs b/linera-service/src/proxy/main.rs
@@ -95,6 +95,10 @@ pub struct ProxyOptions {
     /// Path to the file describing the initial user chains (aka genesis state)
     #[arg(long = "genesis")]
     genesis_config_path: PathBuf,
+
+    /// The replication factor for the keyspace
+    #[arg(long, default_value = "1")]
+    storage_replication_factor: u32,
 }
 
 /// A Linera Proxy, either gRPC or over 'Simple Transport', meaning TCP or UDP.
@@ -410,7 +414,10 @@ impl ProxyOptions {
             storage_cache_config,
         };
         let genesis_config: GenesisConfig = util::read_json(&self.genesis_config_path)?;
-        let store_config = self.storage_config.add_common_config(common_config).await?;
+        let store_config = self
+            .storage_config
+            .add_common_config(common_config, self.storage_replication_factor)
+            .await?;
         store_config
             .run_with_storage(&genesis_config, None, ProxyContext::from_options(self)?)
             .boxed()
diff --git a/linera-service/src/server.rs b/linera-service/src/server.rs
@@ -382,6 +382,10 @@ enum ServerCommand {
         /// The maximal number of entries in the storage cache.
         #[arg(long, default_value = "1000")]
         max_cache_entries: usize,
+
+        /// The replication factor for the storage.
+        #[arg(long, default_value = "1")]
+        storage_replication_factor: u32,
     },
 
     /// Act as a trusted third-party and generate all server configurations
@@ -431,6 +435,10 @@ enum ServerCommand {
         /// The maximal number of entries in the storage cache.
         #[arg(long, default_value = "1000")]
         max_cache_entries: usize,
+
+        /// The replication factor for the storage.
+        #[arg(long, default_value = "1")]
+        storage_replication_factor: u32,
     },
 
     /// Replaces the configurations of the shards by following the given template.
@@ -533,6 +541,7 @@ async fn run(options: ServerOptions) {
             max_cache_size,
             max_entry_size,
             max_cache_entries,
+            storage_replication_factor,
         } => {
             linera_version::VERSION_INFO.log();
 
@@ -561,7 +570,7 @@ async fn run(options: ServerOptions) {
                 storage_cache_config,
             };
             let store_config = storage_config
-                .add_common_config(common_config)
+                .add_common_config(common_config, storage_replication_factor)
                 .await
                 .unwrap();
             store_config
@@ -621,6 +630,7 @@ async fn run(options: ServerOptions) {
             max_cache_size,
             max_entry_size,
             max_cache_entries,
+            storage_replication_factor,
         } => {
             let genesis_config: GenesisConfig =
                 util::read_json(&genesis_config_path).expect("Failed to read initial chain config");
@@ -635,7 +645,7 @@ async fn run(options: ServerOptions) {
                 storage_cache_config,
             };
             let store_config = storage_config
-                .add_common_config(common_config)
+                .add_common_config(common_config, storage_replication_factor)
                 .await
                 .unwrap();
             tracing::info!(
diff --git a/linera-service/src/storage.rs b/linera-service/src/storage.rs
@@ -388,6 +388,7 @@ impl StorageConfigNamespace {
     pub async fn add_common_config(
         &self,
         common_config: CommonStoreConfig,
+        _replication_factor: u32,
     ) -> Result<StoreConfig, anyhow::Error> {
         let namespace = self.namespace.clone();
         match &self.storage_config {
@@ -424,7 +425,8 @@ impl StorageConfigNamespace {
             }
             #[cfg(feature = "scylladb")]
             StorageConfig::ScyllaDb { uri } => {
-                let config = ScyllaDbStoreConfig::new(uri.to_string(), common_config);
+                let config =
+                    ScyllaDbStoreConfig::new(uri.to_string(), common_config, _replication_factor);
                 Ok(StoreConfig::ScyllaDb(config, namespace))
             }
             #[cfg(all(feature = "rocksdb", feature = "scylladb"))]
@@ -438,7 +440,8 @@ impl StorageConfigNamespace {
                     path_with_guard.clone(),
                     common_config.clone(),
                 );
-                let second_config = ScyllaDbStoreConfig::new(uri.to_string(), common_config);
+                let second_config =
+                    ScyllaDbStoreConfig::new(uri.to_string(), common_config, _replication_factor);
                 let config = DualStoreConfig {
                     first_config,
                     second_config,
diff --git a/linera-views/src/backends/scylla_db.rs b/linera-views/src/backends/scylla_db.rs
@@ -632,6 +632,8 @@ pub struct ScyllaDbStoreInternalConfig {
     pub uri: String,
     /// The common configuration of the key value store
     common_config: CommonStoreInternalConfig,
+    /// The replication factor for the keyspace
+    pub replication_factor: u32,
 }
 
 impl AdminKeyValueStore for ScyllaDbStoreInternal {
@@ -811,10 +813,13 @@ impl AdminKeyValueStore for ScyllaDbStoreInternal {
             .boxed()
             .await?;
         // Create a keyspace if it doesn't exist
-        let query = "CREATE KEYSPACE IF NOT EXISTS kv WITH REPLICATION = { \
-            'class' : 'SimpleStrategy', \
-            'replication_factor' : 1 \
-        }";
+        let query = format!(
+            "CREATE KEYSPACE IF NOT EXISTS kv WITH REPLICATION = {{ \
+                'class' : 'SimpleStrategy', \
+                'replication_factor' : {} \
+            }}",
+            config.replication_factor
+        );
 
         // Execute the query
         let prepared = session.prepare(query).await?;
@@ -890,7 +895,11 @@ impl TestKeyValueStore for JournalingKeyValueStore<ScyllaDbStoreInternal> {
             max_concurrent_queries: Some(TEST_SCYLLA_DB_MAX_CONCURRENT_QUERIES),
             max_stream_queries: TEST_SCYLLA_DB_MAX_STREAM_QUERIES,
         };
-        Ok(ScyllaDbStoreInternalConfig { uri, common_config })
+        Ok(ScyllaDbStoreInternalConfig {
+            uri,
+            common_config,
+            replication_factor: 1,
+        })
     }
 }
 
@@ -914,10 +923,15 @@ pub type ScyllaDbStoreConfig = LruCachingConfig<ScyllaDbStoreInternalConfig>;
 
 impl ScyllaDbStoreConfig {
     /// Creates a `ScyllaDbStoreConfig` from the inputs.
-    pub fn new(uri: String, common_config: crate::store::CommonStoreConfig) -> ScyllaDbStoreConfig {
+    pub fn new(
+        uri: String,
+        common_config: crate::store::CommonStoreConfig,
+        replication_factor: u32,
+    ) -> ScyllaDbStoreConfig {
         let inner_config = ScyllaDbStoreInternalConfig {
             uri,
             common_config: common_config.reduced(),
+            replication_factor,
         };
         ScyllaDbStoreConfig {
             inner_config,
