forked from vectordotdev/vector
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmod.rs
More file actions
121 lines (112 loc) · 3.4 KB
/
Copy pathmod.rs
File metadata and controls
121 lines (112 loc) · 3.4 KB
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
121
use crate::Event;
use futures::{future::BoxFuture, Sink, Stream, StreamExt};
use snafu::Snafu;
use std::fmt;
pub mod util;
#[cfg(feature = "sinks-aws_cloudwatch_logs")]
pub mod aws_cloudwatch_logs;
#[cfg(feature = "sinks-aws_cloudwatch_metrics")]
pub mod aws_cloudwatch_metrics;
#[cfg(feature = "sinks-aws_kinesis_firehose")]
pub mod aws_kinesis_firehose;
#[cfg(feature = "sinks-aws_kinesis_streams")]
pub mod aws_kinesis_streams;
#[cfg(feature = "sinks-aws_s3")]
pub mod aws_s3;
#[cfg(feature = "sinks-aws_sqs")]
pub mod aws_sqs;
#[cfg(feature = "sinks-azure_monitor_logs")]
pub mod azure_monitor_logs;
#[cfg(feature = "sinks-blackhole")]
pub mod blackhole;
#[cfg(feature = "sinks-clickhouse")]
pub mod clickhouse;
#[cfg(feature = "sinks-console")]
pub mod console;
#[cfg(feature = "sinks-datadog")]
pub mod datadog;
#[cfg(feature = "sinks-elasticsearch")]
pub mod elasticsearch;
#[cfg(feature = "sinks-file")]
pub mod file;
#[cfg(feature = "sinks-gcp")]
pub mod gcp;
#[cfg(feature = "sinks-honeycomb")]
pub mod honeycomb;
#[cfg(feature = "sinks-http")]
pub mod http;
#[cfg(feature = "sinks-humio")]
pub mod humio;
#[cfg(any(feature = "sinks-influxdb", feature = "prometheus-integration-tests"))]
pub mod influxdb;
#[cfg(all(feature = "sinks-kafka", feature = "rdkafka"))]
pub mod kafka;
#[cfg(feature = "sinks-logdna")]
pub mod logdna;
#[cfg(feature = "sinks-loki")]
pub mod loki;
#[cfg(feature = "sinks-nats")]
pub mod nats;
#[cfg(feature = "sinks-new_relic_logs")]
pub mod new_relic_logs;
#[cfg(feature = "sinks-papertrail")]
pub mod papertrail;
#[cfg(feature = "sinks-prometheus")]
pub mod prometheus;
#[cfg(feature = "sinks-pulsar")]
pub mod pulsar;
#[cfg(feature = "sinks-sematext")]
pub mod sematext;
#[cfg(feature = "sinks-socket")]
pub mod socket;
#[cfg(feature = "sinks-splunk_hec")]
pub mod splunk_hec;
#[cfg(feature = "sinks-statsd")]
pub mod statsd;
#[cfg(feature = "sinks-vector")]
pub mod vector;
pub enum VectorSink {
Sink(Box<dyn Sink<Event, Error = ()> + Send + Unpin>),
Stream(Box<dyn util::StreamSink + Send>),
}
pub type Healthcheck = BoxFuture<'static, crate::Result<()>>;
/// Common build errors
#[derive(Debug, Snafu)]
pub enum BuildError {
#[snafu(display("Unable to resolve DNS for {:?}", address))]
DNSFailure { address: String },
#[snafu(display("DNS errored {}", source))]
DNSError { source: crate::dns::DnsError },
#[snafu(display("Socket address problem: {}", source))]
SocketAddressError { source: std::io::Error },
#[snafu(display("URI parse error: {}", source))]
UriParseError { source: ::http::uri::InvalidUri },
}
/// Common healthcheck errors
#[derive(Debug, Snafu)]
pub enum HealthcheckError {
#[snafu(display("Unexpected status: {}", status))]
UnexpectedStatus { status: ::http::StatusCode },
}
impl VectorSink {
pub async fn run<S>(mut self, input: S) -> Result<(), ()>
where
S: Stream<Item = Event> + Send,
{
match self {
Self::Sink(sink) => input.map(Ok).forward(sink).await,
Self::Stream(ref mut s) => s.run(Box::pin(input)).await,
}
}
pub fn into_sink(self) -> Box<dyn Sink<Event, Error = ()> + Send + Unpin> {
match self {
Self::Sink(sink) => sink,
_ => panic!("Failed type coercion, {:?} is not a Sink", self),
}
}
}
impl fmt::Debug for VectorSink {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("VectorSink").finish()
}
}