Skip to content

Commit 5907824

Browse files
abhizerabhizer
committed
s3_sync: start from any checkpoint in object store
* `sync.start_from_checkpoint` takes either a UUID to a checkpoint or `latest` which just pulls the latest checkpoint * checkpoints now include state.json inside the checkpoint directory as well * syncs individual pipelines to object store, as now they are self contained Signed-off-by: Abhinav Gyawali <22275402+abhizer@users.noreply.github.com>
1 parent 060ea9e commit 5907824

9 files changed

Lines changed: 115 additions & 51 deletions

File tree

‎crates/adapters/src/controller/mod.rs‎

Lines changed: 19 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -238,7 +238,7 @@ enum Command {
238238
GraphProfile(GraphProfileCallbackFn),
239239
Checkpoint(CheckpointCallbackFn),
240240
Suspend(SuspendCallbackFn),
241-
SyncCheckpoint(SyncCheckpointCallbackFn),
241+
SyncCheckpoint((uuid::Uuid, SyncCheckpointCallbackFn)),
242242
}
243243

244244
impl Command {
@@ -249,7 +249,7 @@ impl Command {
249249
callback(Err(Arc::new(ControllerError::ControllerExit)))
250250
}
251251
Command::Suspend(callback) => callback(Err(Arc::new(ControllerError::ControllerExit))),
252-
Command::SyncCheckpoint(callback) => {
252+
Command::SyncCheckpoint((_, callback)) => {
253253
callback(Err(Arc::new(ControllerError::ControllerExit)))
254254
}
255255
}
@@ -589,8 +589,9 @@ impl Controller {
589589
///
590590
/// The callback-based nature of this function makes it useful in
591591
/// asynchronous contexts.
592-
pub fn start_sync_checkpoint(&self, cb: SyncCheckpointCallbackFn) {
593-
self.inner.send_command(Command::SyncCheckpoint(cb));
592+
pub fn start_sync_checkpoint(&self, checkpoint: uuid::Uuid, cb: SyncCheckpointCallbackFn) {
593+
self.inner
594+
.send_command(Command::SyncCheckpoint((checkpoint, cb)));
594595
}
595596

596597
/// Checkpoints the pipeline.
@@ -711,7 +712,7 @@ struct CircuitThread {
711712
checkpoint_requests: Vec<CheckpointRequest>,
712713

713714
/// Currently only allows one request at a time.
714-
sync_checkpoint_request: Option<SyncCheckpointCallbackFn>,
715+
sync_checkpoint_request: Option<(uuid::Uuid, SyncCheckpointCallbackFn)>,
715716

716717
/// Storage backend for writing checkpoints.
717718
storage: Option<Arc<dyn StorageBackend>>,
@@ -1053,6 +1054,7 @@ impl CircuitThread {
10531054
.commit()
10541055
.map_err(|e| Arc::new(ControllerError::from(e)))
10551056
.and_then(|circuit| {
1057+
let uuid = circuit.uuid.to_string();
10561058
let checkpoint = Checkpoint {
10571059
circuit: Some(circuit),
10581060
step: this.step,
@@ -1069,6 +1071,12 @@ impl CircuitThread {
10691071
&**this.storage.as_ref().unwrap(),
10701072
&StoragePath::from(STATE_FILE),
10711073
)
1074+
.map_err(Arc::new)?;
1075+
checkpoint
1076+
.write(
1077+
&**this.storage.as_ref().unwrap(),
1078+
&StoragePath::from(uuid).child(STATE_FILE),
1079+
)
10721080
.map(|()| checkpoint)
10731081
.map_err(Arc::new)
10741082
})?;
@@ -1161,8 +1169,8 @@ impl CircuitThread {
11611169
self.checkpoint_requests
11621170
.push(CheckpointRequest::SuspendCommand(reply_callback));
11631171
}
1164-
Command::SyncCheckpoint(reply_callback) => {
1165-
self.sync_checkpoint_request = Some(reply_callback);
1172+
Command::SyncCheckpoint((uuid, reply_callback)) => {
1173+
self.sync_checkpoint_request = Some((uuid, reply_callback));
11661174
}
11671175
}
11681176
}
@@ -1455,18 +1463,10 @@ impl CircuitThread {
14551463
}
14561464

14571465
fn sync_checkpoint(&mut self) {
1458-
let Some(cb) = self.sync_checkpoint_request.take() else {
1466+
let Some((uuid, cb)) = self.sync_checkpoint_request.take() else {
14591467
return;
14601468
};
14611469

1462-
let cpms = match self.list_checkpoints() {
1463-
Ok(x) => x,
1464-
Err(e) => {
1465-
cb(Err(e));
1466-
return;
1467-
}
1468-
};
1469-
14701470
let Some((_, options)) = self.controller.status.pipeline_config.storage() else {
14711471
cb(Err(Arc::new(ControllerError::storage_error(
14721472
"cannot sync checkpoints when storage is disabled".to_owned(),
@@ -1511,16 +1511,9 @@ impl CircuitThread {
15111511
let config = sync.to_owned();
15121512

15131513
std::thread::spawn(move || {
1514-
let mut errs = vec![];
1515-
for cpm in cpms {
1516-
if let Err(err) = synchronizer.push(&cpm, storage.clone(), config.clone()) {
1517-
errs.push(err.to_string());
1518-
}
1519-
}
1520-
1521-
if !errs.is_empty() {
1514+
if let Err(err) = synchronizer.push(uuid, storage.clone(), config.clone()) {
15221515
cb(Err(Arc::new(ControllerError::checkpoint_push_error(
1523-
errs.join("\n"),
1516+
err.to_string(),
15241517
))));
15251518
} else {
15261519
cb(Ok(()))
@@ -2026,7 +2019,7 @@ impl ControllerInit {
20262019
..
20272020
}) = storage.options.backend
20282021
{
2029-
if sync.start_from_checkpoint {
2022+
if sync.start_from_checkpoint.is_some() {
20302023
let Some(synchronizer) = inventory::iter::<&dyn CheckpointSynchronizer>
20312024
.into_iter()
20322025
.next()

‎crates/adapters/src/server/mod.rs‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -931,9 +931,12 @@ async fn checkpoint_sync(state: WebData<ServerState>) -> Result<impl Responder,
931931
};
932932

933933
let state = state.sync_checkpoint_state.clone();
934-
controller.start_sync_checkpoint(Box::new(move |result| {
935-
state.lock().unwrap().completed(last_checkpoint, result);
936-
}));
934+
controller.start_sync_checkpoint(
935+
last_checkpoint,
936+
Box::new(move |result| {
937+
state.lock().unwrap().completed(last_checkpoint, result);
938+
}),
939+
);
937940
last_checkpoint
938941
}
939942
};

‎crates/feldera-types/src/config.rs‎

Lines changed: 56 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -257,6 +257,56 @@ pub enum StorageCompression {
257257
Snappy,
258258
}
259259

260+
#[derive(Debug, Clone, Eq, PartialEq, ToSchema)]
261+
pub enum StartFromCheckpoint {
262+
Latest,
263+
Uuid(uuid::Uuid),
264+
}
265+
266+
impl<'de> Deserialize<'de> for StartFromCheckpoint {
267+
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
268+
where
269+
D: Deserializer<'de>,
270+
{
271+
struct StartFromCheckpointVisitor;
272+
273+
impl<'de> Visitor<'de> for StartFromCheckpointVisitor {
274+
type Value = StartFromCheckpoint;
275+
276+
fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
277+
formatter.write_str("a UUID string or the string \"latest\"")
278+
}
279+
280+
fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
281+
where
282+
E: de::Error,
283+
{
284+
if value == "latest" {
285+
Ok(StartFromCheckpoint::Latest)
286+
} else {
287+
uuid::Uuid::parse_str(value)
288+
.map(StartFromCheckpoint::Uuid)
289+
.map_err(|_| E::invalid_value(serde::de::Unexpected::Str(value), &self))
290+
}
291+
}
292+
}
293+
294+
deserializer.deserialize_str(StartFromCheckpointVisitor)
295+
}
296+
}
297+
298+
impl Serialize for StartFromCheckpoint {
299+
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
300+
where
301+
S: serde::Serializer,
302+
{
303+
match self {
304+
StartFromCheckpoint::Latest => serializer.serialize_str("latest"),
305+
StartFromCheckpoint::Uuid(uuid) => serializer.serialize_str(&uuid.to_string()),
306+
}
307+
}
308+
}
309+
260310
#[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize, ToSchema)]
261311
pub struct SyncConfig {
262312
/// The endpoint URL for the storage service.
@@ -299,9 +349,12 @@ pub struct SyncConfig {
299349
/// this can be left empty to allow automatic authentication via the pod's service account.
300350
pub secret_key: Option<String>,
301351

302-
/// If `true`, will try to pull the latest checkpoint from the configured
303-
/// object store and resume from that point.
304-
pub start_from_checkpoint: bool,
352+
/// When set, the pipeline will fetch either the specified checkpoint
353+
/// from the object store.
354+
///
355+
/// If the checkpoint doesn't exist in object store, the pipeline will
356+
/// fail to initialize.
357+
pub start_from_checkpoint: Option<StartFromCheckpoint>,
305358
}
306359

307360
#[derive(Debug, Clone, Default, Eq, PartialEq, Serialize, Deserialize, ToSchema)]

‎crates/pipeline-manager/src/api/main.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,7 @@ only the program-related core fields, and is used by the compiler to discern whe
166166
feldera_types::config::StorageOptions,
167167
feldera_types::config::StorageBackendConfig,
168168
feldera_types::config::SyncConfig,
169+
feldera_types::config::StartFromCheckpoint,
169170
feldera_types::config::FileBackendConfig,
170171
feldera_types::config::StorageCompression,
171172
feldera_types::config::RuntimeConfig,

‎crates/storage/src/checkpoint_synchronizer.rs‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,15 @@
11
use std::sync::Arc;
22
use std::{error::Error, fmt::Display};
33

4-
use feldera_types::{checkpoint::CheckpointMetadata, config::SyncConfig};
4+
use feldera_types::config::SyncConfig;
55

66
use crate::error::StorageError;
77
use crate::StorageBackend;
88

99
pub trait CheckpointSynchronizer: Sync {
1010
fn push(
1111
&self,
12-
checkpoint: &CheckpointMetadata,
12+
checkpoint: uuid::Uuid,
1313
storage: Arc<dyn StorageBackend>,
1414
remote_config: SyncConfig,
1515
) -> Result<(), SyncError>;
@@ -35,10 +35,7 @@ impl Display for SyncError {
3535
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
3636
match self {
3737
SyncError::Io(error) => write!(f, "synchronizer IO err: {error}",),
38-
SyncError::Serde(error) => write!(
39-
f,
40-
"synchronizer: error parsing checkpoint metadata: {error}",
41-
),
38+
SyncError::Serde(error) => write!(f, "synchronizer: serde error: {error}",),
4239
SyncError::RcloneExitCode(error) => write!(f, "synchronizer: rclone: '{error}'"),
4340
SyncError::Storage(error) => write!(f, "synchronizer: storage error: '{error}'"),
4441
}

‎crates/storage/src/lib.rs‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -185,14 +185,14 @@ impl dyn StorageBackend {
185185
}
186186
}
187187

188-
pub fn gather_batches_for_checkpoint(
188+
pub fn gather_batches_for_checkpoint_uuid(
189189
&self,
190-
cpm: &CheckpointMetadata,
190+
cpm: uuid::Uuid,
191191
) -> Result<HashSet<StoragePath>, StorageError> {
192-
assert!(!cpm.uuid.is_nil());
192+
assert!(!cpm.is_nil());
193193

194194
let mut spines = Vec::new();
195-
self.list(&cpm.uuid.to_string().into(), &mut |path, _file_type| {
195+
self.list(&cpm.to_string().into(), &mut |path, _file_type| {
196196
if path
197197
.filename()
198198
.is_some_and(|filename| filename.starts_with("pspine-batches"))
@@ -212,6 +212,13 @@ impl dyn StorageBackend {
212212
Ok(batch_files_in_commit)
213213
}
214214

215+
pub fn gather_batches_for_checkpoint(
216+
&self,
217+
cpm: &CheckpointMetadata,
218+
) -> Result<HashSet<StoragePath>, StorageError> {
219+
self.gather_batches_for_checkpoint_uuid(cpm.uuid)
220+
}
221+
215222
/// Writes `content` to `name` as JSON, automatically creating any parent
216223
/// directories within `name` that don't already exist.
217224
pub fn write_json<V>(&self, name: &StoragePath, value: &V) -> Result<(), StorageError>

‎docs.feldera.com/docs/pipelines/checkpoint-sync.md‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
# Synchronizing checkpoints to object store
22

3+
:::caution Experimental feature
4+
Synchronizing checkpoints to object store is a highly experimental feature.
5+
:::
6+
37
Feldera allows synchronizing pipeline checkpoints to an object store and
48
restoring them at startup.
59

@@ -21,7 +25,7 @@ Here is a sample configuration:
2125
"provider": "AWS",
2226
"access_key": "ACCESS_KEY",
2327
"secret_key": "SECRET_KEY",
24-
"start_from_checkpoint": true
28+
"start_from_checkpoint": false
2529
}
2630
}
2731
}
@@ -32,12 +36,12 @@ Here is a sample configuration:
3236

3337
| Field | Type | Description |
3438
| ------------------------- | --------- | ---------------------------------------------------------------------------------------------------------------------------------- |
35-
| `endpoint` | `string` | The S3-compatible object store endpoint (e.g., for MinIO, AWS). |
36-
| `bucket`\* | `string` | The bucket name and optional prefix to store checkpoints (e.g., `mybucket/checkpoints`). |
37-
| `provider`\* | `string` | The S3 provider identifier. Must match [rclone's list](https://rclone.org/s3/#providers). Case-sensitive. Use `"Other"` if unsure. |
38-
| `access_key` | `string` | Your S3 access key. Not required if using environment-based authentication (e.g., IRSA). |
39-
| `secret_key` | `string` | Your S3 secret key. Same rules as `access_key`. |
40-
| `start_from_checkpoint`\* | `boolean` | If `true`, Feldera will restore the latest checkpoint from the object store on startup. |
39+
| `endpoint` | `string` | The S3-compatible object store endpoint (e.g., for MinIO, AWS). |
40+
| `bucket`\* | `string` | The bucket name and optional prefix to store checkpoints (e.g., `mybucket/checkpoints`). |
41+
| `provider`\* | `string` | The S3 provider identifier. Must match [rclone's list](https://rclone.org/s3/#providers). Case-sensitive. Use `"Other"` if unsure. |
42+
| `access_key` | `string` | Your S3 access key. Not required if using environment-based authentication (e.g., IRSA). |
43+
| `secret_key` | `string` | Your S3 secret key. Same rules as `access_key`. |
44+
| `start_from_checkpoint` | `string` | Provide a checkpoint UUID to resume from it, or use `latest` to restore from the latest one. The provided UUID must exist in object store. |
4145

4246
## S3 permissions
4347

‎python/feldera/pipeline.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
from datetime import datetime
44

55
import pandas
6+
from uuid import UUID
67

78
from typing import List, Dict, Callable, Optional, Generator, Mapping, Any
89
from collections import deque
@@ -636,6 +637,8 @@ def sync_checkpoint_status(self, uuid: str) -> CheckpointStatus:
636637
resp = self.client.sync_checkpoint_status(self.name)
637638
success = resp.get("success")
638639

640+
fail = resp.get("failure") or {}
641+
639642
if uuid == success:
640643
return CheckpointStatus.Success
641644

@@ -645,6 +648,9 @@ def sync_checkpoint_status(self, uuid: str) -> CheckpointStatus:
645648
failure.error = fail.get("error", "")
646649
return failure
647650

651+
if (success is None) or UUID(uuid) > UUID(success):
652+
return CheckpointStatus.InProgress
653+
648654
return CheckpointStatus.Unknown
649655

650656
def query(self, query: str) -> Generator[Mapping[str, Any], None, None]:

‎python/tests/test_shared_pipeline1.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212

1313
def storage_cfg(
1414
endpoint: Optional[str] = None,
15-
start_from_checkpoint: bool = False,
15+
start_from_checkpoint: Optional[str] = None,
1616
auth_err: bool = False,
1717
) -> dict:
1818
return {
@@ -56,7 +56,7 @@ def test_checkpoint_sync(self, auth_err: bool = False):
5656
self.pipeline.clear_storage()
5757

5858
# Restart pipeline from checkpoint
59-
storage_config = storage_cfg(start_from_checkpoint=True, auth_err=auth_err)
59+
storage_config = storage_cfg(start_from_checkpoint="latest", auth_err=auth_err)
6060
self.set_runtime_config(RuntimeConfig(storage=Storage(config=storage_config)))
6161
self.pipeline.start()
6262
got_after = list(self.pipeline.query("SELECT * FROM v0"))

0 commit comments

Comments
 (0)