Skip to content

Commit 1e7e99c

Browse files
chore(redis sink): Refactor to use StreamSink (vectordotdev#18220)
* Update Redis sink to stream sink Signed-off-by: Stephen Wakely <[email protected]> * Update Redis sink to stream sink Signed-off-by: Stephen Wakely <[email protected]> * Clippy Signed-off-by: Stephen Wakely <[email protected]> * Removed unused encoder file Signed-off-by: Stephen Wakely <[email protected]> * Small bit of module documentation Signed-off-by: Stephen Wakely <[email protected]> * Tidies up comments and added a datavolume integration test Signed-off-by: Stephen Wakely <[email protected]> * Delete unused redis.rs Signed-off-by: Stephen Wakely <[email protected]> * Remove spurious comment Signed-off-by: Stephen Wakely <[email protected]> * Update request builder doc comments Signed-off-by: Stephen Wakely <[email protected]> * Removed Response impl Signed-off-by: Stephen Wakely <[email protected]> --------- Signed-off-by: Stephen Wakely <[email protected]>
1 parent d09b2af commit 1e7e99c

File tree

8 files changed

+943
-680
lines changed

8 files changed

+943
-680
lines changed

src/sinks/redis.rs

-680
This file was deleted.

src/sinks/redis/config.rs

+162
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,162 @@
1+
use redis::{aio::ConnectionManager, RedisResult};
2+
use snafu::prelude::*;
3+
4+
use crate::sinks::prelude::*;
5+
6+
use super::{sink::RedisSink, RedisCreateFailedSnafu};
7+
8+
/// Redis data type to store messages in.
9+
#[configurable_component]
10+
#[derive(Clone, Copy, Debug, Derivative)]
11+
#[derivative(Default)]
12+
#[serde(rename_all = "lowercase")]
13+
pub enum DataTypeConfig {
14+
/// The Redis `list` type.
15+
///
16+
/// This resembles a deque, where messages can be popped and pushed from either end.
17+
///
18+
/// This is the default.
19+
#[derivative(Default)]
20+
List,
21+
22+
/// The Redis `channel` type.
23+
///
24+
/// Redis channels function in a pub/sub fashion, allowing many-to-many broadcasting and receiving.
25+
Channel,
26+
}
27+
28+
/// List-specific options.
29+
#[configurable_component]
30+
#[derive(Clone, Copy, Debug, Derivative, Eq, PartialEq)]
31+
#[serde(rename_all = "lowercase")]
32+
pub struct ListOption {
33+
/// The method to use for pushing messages into a `list`.
34+
pub(super) method: Method,
35+
}
36+
37+
/// Method for pushing messages into a `list`.
38+
#[configurable_component]
39+
#[derive(Clone, Copy, Debug, Derivative, Eq, PartialEq)]
40+
#[derivative(Default)]
41+
#[serde(rename_all = "lowercase")]
42+
pub enum Method {
43+
/// Use the `rpush` method.
44+
///
45+
/// This pushes messages onto the tail of the list.
46+
///
47+
/// This is the default.
48+
#[derivative(Default)]
49+
RPush,
50+
51+
/// Use the `lpush` method.
52+
///
53+
/// This pushes messages onto the head of the list.
54+
LPush,
55+
}
56+
57+
#[derive(Clone, Copy, Debug, Default)]
58+
pub struct RedisDefaultBatchSettings;
59+
60+
impl SinkBatchSettings for RedisDefaultBatchSettings {
61+
const MAX_EVENTS: Option<usize> = Some(1);
62+
const MAX_BYTES: Option<usize> = None;
63+
const TIMEOUT_SECS: f64 = 1.0;
64+
}
65+
66+
/// Configuration for the `redis` sink.
67+
#[configurable_component(sink("redis", "Publish observability data to Redis."))]
68+
#[derive(Clone, Debug)]
69+
#[serde(deny_unknown_fields)]
70+
pub struct RedisSinkConfig {
71+
#[configurable(derived)]
72+
pub(super) encoding: EncodingConfig,
73+
74+
#[configurable(derived)]
75+
#[serde(default)]
76+
pub(super) data_type: DataTypeConfig,
77+
78+
#[configurable(derived)]
79+
#[serde(alias = "list")]
80+
pub(super) list_option: Option<ListOption>,
81+
82+
/// The URL of the Redis endpoint to connect to.
83+
///
84+
/// The URL _must_ take the form of `protocol://server:port/db` where the protocol can either be
85+
/// `redis` or `rediss` for connections secured via TLS.
86+
#[configurable(metadata(docs::examples = "redis://127.0.0.1:6379/0"))]
87+
#[serde(alias = "url")]
88+
pub(super) endpoint: String,
89+
90+
/// The Redis key to publish messages to.
91+
#[configurable(validation(length(min = 1)))]
92+
#[configurable(metadata(docs::examples = "syslog:{{ app }}", docs::examples = "vector"))]
93+
pub(super) key: Template,
94+
95+
#[configurable(derived)]
96+
#[serde(default)]
97+
pub(super) batch: BatchConfig<RedisDefaultBatchSettings>,
98+
99+
#[configurable(derived)]
100+
#[serde(default)]
101+
pub(super) request: TowerRequestConfig,
102+
103+
#[configurable(derived)]
104+
#[serde(
105+
default,
106+
deserialize_with = "crate::serde::bool_or_struct",
107+
skip_serializing_if = "crate::serde::skip_serializing_if_default"
108+
)]
109+
pub(super) acknowledgements: AcknowledgementsConfig,
110+
}
111+
112+
impl GenerateConfig for RedisSinkConfig {
113+
fn generate_config() -> toml::Value {
114+
toml::from_str(
115+
r#"
116+
url = "redis://127.0.0.1:6379/0"
117+
key = "vector"
118+
data_type = "list"
119+
list.method = "lpush"
120+
encoding.codec = "json"
121+
batch.max_events = 1
122+
"#,
123+
)
124+
.unwrap()
125+
}
126+
}
127+
128+
#[async_trait::async_trait]
129+
#[typetag::serde(name = "redis")]
130+
impl SinkConfig for RedisSinkConfig {
131+
async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
132+
if self.key.is_empty() {
133+
return Err("`key` cannot be empty.".into());
134+
}
135+
let conn = self.build_client().await.context(RedisCreateFailedSnafu)?;
136+
let healthcheck = RedisSinkConfig::healthcheck(conn.clone()).boxed();
137+
let sink = RedisSink::new(self, conn)?;
138+
Ok((super::VectorSink::from_event_streamsink(sink), healthcheck))
139+
}
140+
141+
fn input(&self) -> Input {
142+
Input::new(self.encoding.config().input_type() & DataType::Log)
143+
}
144+
145+
fn acknowledgements(&self) -> &AcknowledgementsConfig {
146+
&self.acknowledgements
147+
}
148+
}
149+
150+
impl RedisSinkConfig {
151+
pub(super) async fn build_client(&self) -> RedisResult<ConnectionManager> {
152+
let client = redis::Client::open(self.endpoint.as_str())?;
153+
client.get_tokio_connection_manager().await
154+
}
155+
156+
async fn healthcheck(mut conn: ConnectionManager) -> crate::Result<()> {
157+
redis::cmd("PING")
158+
.query_async(&mut conn)
159+
.await
160+
.map_err(Into::into)
161+
}
162+
}

0 commit comments

Comments
 (0)