opendata-buffer 0.2.0

Stateless object storage buffer library for OpenData systems
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

use std::time::Duration;

use common::ObjectStoreConfig;
use serde::{Deserialize, Serialize};
use serde_with::{DurationMilliSeconds, serde_as};

use crate::model::CompressionType;

/// Configuration for a [`Producer`](crate::Producer).
///
/// Controls where data batches and the queue manifest are stored, how often
/// batches are flushed, and when backpressure is applied.
#[serde_as]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProducerConfig {
    /// Determines where and how ingest data is persisted. See [`ObjectStoreConfig`].
    pub object_store: ObjectStoreConfig,

    /// Path prefix for data batch objects in object storage.
    ///
    /// Defaults to `"ingest"`.
    #[serde(default = "default_data_path_prefix")]
    pub data_path_prefix: String,

    /// Path to the queue manifest in object storage.
    ///
    /// Defaults to `"ingest/manifest"`.
    #[serde(default = "default_manifest_path")]
    pub manifest_path: String,

    /// Time interval that triggers the flush of the current batch to object storage when elapsed.
    ///
    /// Defaults to 100 ms.
    #[serde_as(as = "DurationMilliSeconds<u64>")]
    #[serde(default = "default_flush_interval")]
    pub flush_interval: Duration,

    /// Batch size in bytes (entries and metadata) that triggers a flush when exceeded.
    ///
    /// Defaults to 64 MiB.
    #[serde(default = "default_flush_size_bytes")]
    pub flush_size_bytes: usize,

    /// Maximum number of input entries vectors that can be buffered for the background
    /// batch writer before backpressure is applied.
    ///
    /// Defaults to 1000.
    #[serde(default = "default_max_buffered_inputs")]
    pub max_buffered_inputs: usize,

    /// Compression algorithm applied to the record block in data batches.
    ///
    /// Defaults to `None` (uncompressed).
    #[serde(default)]
    pub batch_compression: CompressionType,
}

/// Configuration for a [`Consumer`](crate::Consumer).
///
/// Controls where the queue manifest and data batches are read from.
#[serde_as]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConsumerConfig {
    /// Determines where and how ingest data is read. See [`ObjectStoreConfig`].
    pub object_store: ObjectStoreConfig,

    /// Path to the queue manifest in object storage.
    ///
    /// Defaults to `"ingest/manifest"`.
    #[serde(default = "default_manifest_path")]
    pub manifest_path: String,

    /// Path prefix for data batch objects in object storage.
    ///
    /// Must match the producer's `data_path_prefix`. Defaults to `"ingest"`.
    #[serde(default = "default_data_path_prefix")]
    pub data_path_prefix: String,

    /// How often garbage collection runs.
    ///
    /// Defaults to 5 minutes.
    #[serde_as(as = "DurationMilliSeconds<u64>")]
    #[serde(default = "default_gc_interval")]
    pub gc_interval: Duration,

    /// Minimum age of an unreferenced batch file before it is eligible for
    /// deletion by the garbage collector.
    ///
    /// Defaults to 10 minutes.
    #[serde_as(as = "DurationMilliSeconds<u64>")]
    #[serde(default = "default_gc_grace_period")]
    pub gc_grace_period: Duration,
}

fn default_data_path_prefix() -> String {
    "ingest".to_string()
}

fn default_manifest_path() -> String {
    "ingest/manifest".to_string()
}

fn default_flush_interval() -> Duration {
    Duration::from_millis(100)
}

fn default_flush_size_bytes() -> usize {
    64 * 1024 * 1024
}

fn default_max_buffered_inputs() -> usize {
    1000
}
fn default_gc_interval() -> Duration {
    Duration::from_mins(5)
}

fn default_gc_grace_period() -> Duration {
    Duration::from_mins(10)
}