final touches to k8s deployment

Author: Damion Werner

malstrom-core/src/coordinator/coordinator.rs

@@ -245,7 +245,7 @@ async fn coordinator_loop<C: Send + CoordinatorWorkerComm, P: Send + Persistence
     state: SerializableCoordinatorState,
     requests: flume::Receiver<CoordinatorRequest>,
     communication_backend: C,
-    persistence_backend: P,
+    mut persistence_backend: P,
 ) -> Result<(), CoordinatorLoopError> {
     let mut state = CoordinatorState::from_serialized(state, &communication_backend).await?;
     // start job on all workers
@@ -290,10 +290,13 @@ async fn coordinator_loop<C: Send + CoordinatorWorkerComm, P: Send + Persistence
                     perform_snapshot_all(&state, next_version).await;
 
                     let serialized_state = serialize_state(&state.get_serializable().await);
-                    persistence_backend
-                        .for_version(COORDINATOR_ID, &next_version)
-                        .persist(&serialized_state, &0);
-                    persistence_backend.commit_version(&next_version);
+                    persistence_backend = tokio::runtime::Handle::current().spawn_blocking(move || {
+                        persistence_backend
+                            .for_version(COORDINATOR_ID, &next_version)
+                            .persist(&serialized_state, &0);
+                        persistence_backend.commit_version(&next_version);
+                        persistence_backend
+                    }).await?;
                     snapshot_version = Some(next_version);
                 }
                 RequestOperation::Scale(desired) => {

malstrom-k8s/artifact-manager/Rocket.toml

@@ -2,5 +2,5 @@
 address = "0.0.0.0"
 
 [default.limits]
-file = "100MiB"
-"data-form" = "100MiB"
+file = "1024MiB"
+"data-form" = "1024MiB"

malstrom-k8s/operator/Cargo.toml

@@ -11,8 +11,6 @@ readme = "Readme.md"
 authors = [
     "nico@acupof.xyz"
 ]
-[lints]
-workspace = true
 
 [[bin]]
 name = "k8s-operator"

malstrom-k8s/operator/crds/src/lib.rs

@@ -133,7 +133,7 @@ impl MalstromJob {
             spec: Some(ServiceSpec {
                 cluster_ip: None,
                 selector: Some(labels),
-                publish_not_ready_addresses: Some(false),
+                publish_not_ready_addresses: Some(true),
                 ports: Some(Vec::from_iter([ServicePort {
                     port: 29091,
                     target_port: Some(IntOrString::Int(29091)),

malstrom-kafka/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 [dependencies]
 bon = { version = "3.5.1", features = ["experimental-overwritable"] }
 malstrom = { path = "../malstrom-core" }
-rdkafka = { version = "0.36.2"}
+rdkafka = { version = "0.37.0"}
 thiserror = "2.0.12"
 tracing = "0.1.41"
 

website/guide/Kubernetes.md

@@ -101,6 +101,10 @@ number greater than 0. The operator will then automatically take care of rescali
 
 ## Operator values
 
+> [!WARNING]
+> Currently the operator only supports modifying the job replica count. Any other modification
+> requires manually deleting and re-creating the job. See [issue 25](https://github.com/MalstromDevelopers/malstrom/issues/25)
+
 The operator helm chart supports these configuration values:
 
 <<< @../../malstrom-k8s/operator/helm/malstrom-operator/values.yaml