Skip to content

Commit d981c55

Browse files
agavraagavra
authored
add README.md to ingest crate (#362)
1 parent dc54dc7 commit d981c55

1 file changed

Lines changed: 142 additions & 0 deletions

File tree

‎ingest/README.md‎

Lines changed: 142 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,142 @@
1+
# opendata-ingest
2+
3+
A shared, stateless ingestion library for [OpenData](https://github.com/opendata-oss/opendata) databases.
4+
5+
Provides write-path infrastructure that all OpenData databases (Timeseries, Log, Vector) can reuse. Ingestors accept opaque byte entries, buffer them in memory, and periodically flush batched data files to object storage. A manifest-backed queue coordinates producers (ingestors) and consumers (collectors) in a stateless, crash-safe way.
6+
7+
## Why stateless ingest?
8+
9+
- **Fault tolerance** — ingestors are stateless. If one fails, any other running ingestor can take over without a rebalancing protocol.
10+
- **Decoupled from writes** — if the downstream database is slow or unavailable, ingested data is safely persisted in object storage rather than dropped or back-pressured.
11+
- **Cost savings** — data flows through object storage rather than across availability zones, avoiding cross-zonal transfer fees.
12+
13+
## Architecture
14+
15+
```text
16+
╔═Ingestors══════════════════════════════════════════════════╗
17+
║ ║░
18+
║ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ║░
19+
║ │ Ingestor 1 │ │ Ingestor 2 │ │ Ingestor N │ ║░
20+
║ │ q-producer │ │ q-producer │ │ q-producer │ ║░
21+
║ └───────┬──────┘ └──────────────┘ └───────┬──────┘ ║░
22+
║ │ │ ║░
23+
╚══════════╪═══════════════════════════════════════╪═════════╝░
24+
░░░░░░░░░░│░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░░│░░░░░░░░░░░
25+
└flush enqueue
26+
│ │
27+
╔═Object Store Bucket═════════════════════════╪══════════════╗
28+
║ │ │ ║░
29+
║ │ │ ║░
30+
║ ┏━data━━━━━━━▼━━━━━━━━━━━┓ ┏━queue━━━━━▼━━━━━━━━━━┓ ║░
31+
║ ┃ ┃ ┃ ┃ ║░
32+
║ ┃ ┃ ┃ ╔════════════════╗ ┃ ║░
33+
║ ┃ 01J5T4R3.batch ┃ ┃ ║ ║ ┃ ║░
34+
║ ┃ 01J5T4R7.batch ┃ ┃ ║ q-manifest ║ ┃ ║░
35+
║ ┃ 01J5T4RB.batch ┃ ┃ ║ ║ ┃ ║░
36+
║ ┃ ┃ ┃ ╚════════════════╝ ┃ ║░
37+
║ ┃ ┃ ┃ ┃ ║░
38+
║ ┃ ┃ ┃ ┃ ║░
39+
║ ┗━━━━━━━━━━━━┬━━━━━━━━━━━┛ ┗━━━━━━━━━━━▲━━━━━━━━━━┛ ║░
40+
║ │ │ ║░
41+
║ │ │ ║░
42+
╚═══════════════╪═════════════════════════════╪══════════════╝░
43+
░░░░░░░░░░░░░░░│░░░░░░░░░░░░░░░░░░░░░░░░░░░░░│░░░░░░░░░░░░░░░░
44+
│ ┌───────poll─────────┘
45+
read │
46+
│ │
47+
╔═Writer═╪════════╪═════════╗
48+
║ │ │ ║░
49+
║ │┌───────┴──────┐ ║░ ╔════════════════╗
50+
║ ││ Collector │ ║░ ║ ║
51+
║ └▶ q-consumer ├──write──▶ Database ║
52+
║ └──────────────┘ ║░ ║ ║
53+
║ ║░ ╚════════════════╝
54+
╚═══════════════════════════╝░
55+
░░░░░░░░░░░░░░░░░░░░░░░░░░░░░
56+
```
57+
58+
## Usage
59+
60+
### Ingestor
61+
62+
The ingestor buffers entries and flushes them as compressed batches to object storage, appending their locations to the queue manifest.
63+
64+
```rust
65+
use ingest::{Ingestor, IngestorConfig};
66+
use bytes::Bytes;
67+
use std::sync::Arc;
68+
69+
let ingestor = Ingestor::new(IngestorConfig::default(), clock)?;
70+
71+
// Ingest entries with metadata — returns a handle to await durability
72+
let handle = ingestor.ingest(
73+
vec![Bytes::from("entry-1"), Bytes::from("entry-2")],
74+
Bytes::from("my-metadata"),
75+
).await?;
76+
77+
// Block until the batch is flushed to object storage and enqueued
78+
handle.watcher.await_durable().await?;
79+
80+
// Flush remaining entries and shut down
81+
ingestor.close().await?;
82+
```
83+
84+
#### Configuration
85+
86+
| Field | Default | Description |
87+
|-------|---------|-------------|
88+
| `flush_interval` | 100 ms | Time interval that triggers a flush |
89+
| `flush_size_bytes` | 64 MiB | Batch size threshold that triggers a flush |
90+
| `max_buffered_inputs` | 1000 | Max buffered `ingest()` calls before backpressure |
91+
| `batch_compression` | `None` | Compression algorithm (`None` or `Zstd`) |
92+
| `data_path_prefix` | `"ingest"` | Object storage prefix for data batches |
93+
| `manifest_path` | `"ingest/manifest"` | Path to the queue manifest |
94+
95+
### Collector
96+
97+
The collector reads batches from the queue in ingestion order and makes them available to a database writer.
98+
99+
```rust
100+
use ingest::{Collector, CollectorConfig};
101+
102+
let collector = Collector::new(CollectorConfig::default(), clock)?;
103+
104+
// Initialize the consumer — fences any previous collector via epoch bump
105+
collector.initialize(None).await?;
106+
107+
// Read batches in order
108+
while let Some(batch) = collector.next_batch().await? {
109+
// batch.entries: Vec<Bytes>
110+
// batch.sequence: u64
111+
// batch.metadata: Vec<Metadata>
112+
process(&batch);
113+
collector.ack(batch.sequence).await?;
114+
}
115+
116+
// Force-flush acked entries from the manifest
117+
collector.flush().await?;
118+
```
119+
120+
## Delivery guarantees
121+
122+
Exactly-once delivery is achievable when the caller atomically writes both the batch and its sequence number to the downstream database. After a failure, the collector resumes from the last committed sequence — no data is processed twice.
123+
124+
On the ingestor side, callers that track progress and re-ingest unacknowledged entries achieve at-least-once delivery.
125+
126+
## Data batch format
127+
128+
Each batch is a compact binary file with an optionally compressed block of length-prefixed records followed by a fixed-size footer:
129+
130+
```text
131+
+------------------------------------------+
132+
| compressed record block (variable): |
133+
| record 0: [len: u32 LE][data] |
134+
| record 1: [len: u32 LE][data] |
135+
| ... |
136+
+------------------------------------------+
137+
| footer (7 bytes): |
138+
| compression_type : u8 |
139+
| record_count : u32 LE |
140+
| version : u16 LE (= 1) |
141+
+------------------------------------------+
142+
```

0 commit comments

Comments
 (0)