data-courier 0.1.0-beta.4

Async Rust framework for composable data pipelines
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

use std::fmt;

use serde_json::Value;

use crate::pipeline::ErrorPolicy;
use crate::retry::RetryPolicy;

use super::redact::{RedactedJsonValue, RedactedOptionRetryPolicy, RedactedStr};

#[derive(Clone, Default, PartialEq)]
pub struct Config {
    pub pipelines: Vec<PipelineSpec>,
    pub observability: Option<super::observability::ObservabilityConfig>,
}

impl fmt::Debug for Config {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Config")
            .field("pipelines", &self.pipelines)
            .field("observability", &self.observability)
            .finish()
    }
}

#[derive(Clone, PartialEq)]
pub struct PipelineSpec {
    pub name: String,
    pub source: SourceSpec,
    pub transforms: Vec<TransformSpec>,
    pub sinks: Vec<SinkSpec>,
    pub channel_capacity: Option<usize>,
}

#[derive(Clone, PartialEq)]
pub struct SourceSpec {
    pub kind: String,
    pub config: Value,
    pub retry: Option<RetryPolicy>,
}

#[derive(Clone, PartialEq)]
pub struct TransformSpec {
    pub kind: String,
    pub config: Value,
    pub on_error: Option<ErrorPolicyConfig>,
}

#[derive(Clone, PartialEq)]
pub struct SinkSpec {
    pub kind: String,
    pub config: Value,
    pub on_error: Option<ErrorPolicyConfig>,
    pub retry: Option<RetryPolicy>,
}

impl fmt::Debug for PipelineSpec {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("PipelineSpec")
            .field("name", &RedactedStr(&self.name))
            .field("source", &self.source)
            .field("transforms", &self.transforms)
            .field("sinks", &self.sinks)
            .field("channel_capacity", &self.channel_capacity)
            .finish()
    }
}

impl fmt::Debug for SourceSpec {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SourceSpec")
            .field("kind", &RedactedStr(&self.kind))
            .field("config", &RedactedJsonValue(&self.config))
            .field("retry", &RedactedOptionRetryPolicy(self.retry.as_ref()))
            .finish()
    }
}

impl fmt::Debug for TransformSpec {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("TransformSpec")
            .field("kind", &RedactedStr(&self.kind))
            .field("config", &RedactedJsonValue(&self.config))
            .field("on_error", &self.on_error)
            .finish()
    }
}

impl fmt::Debug for SinkSpec {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SinkSpec")
            .field("kind", &RedactedStr(&self.kind))
            .field("config", &RedactedJsonValue(&self.config))
            .field("on_error", &self.on_error)
            .field("retry", &RedactedOptionRetryPolicy(self.retry.as_ref()))
            .finish()
    }
}

#[derive(Debug, Clone, Copy, Default, Eq, PartialEq)]
pub enum ErrorPolicyConfig {
    #[default]
    Drop,
    FailPipeline,
}

impl From<ErrorPolicyConfig> for ErrorPolicy {
    fn from(value: ErrorPolicyConfig) -> Self {
        match value {
            ErrorPolicyConfig::Drop => ErrorPolicy::Drop,
            ErrorPolicyConfig::FailPipeline => ErrorPolicy::FailPipeline,
        }
    }
}