vector/sinks/greptimedb/logs/
sink.rs

1use crate::sinks::{
2    greptimedb::logs::http_request_builder::{
3        GreptimeDBLogsHttpRequestBuilder, KeyPartitioner, PartitionKey,
4    },
5    prelude::*,
6    util::http::HttpRequest,
7};
8
9pub struct LogsSinkSetting {
10    pub dbname: ConfinedTemplate,
11    pub table: ConfinedTemplate,
12    pub pipeline_name: ConfinedTemplate,
13    pub pipeline_version: Option<ConfinedTemplate>,
14}
15
16/// A sink that ingests logs into GreptimeDB.
17pub struct GreptimeDBLogsHttpSink<S> {
18    batcher_settings: BatcherSettings,
19    service: S,
20    request_builder: GreptimeDBLogsHttpRequestBuilder,
21    logs_sink_setting: LogsSinkSetting,
22}
23
24impl<S> GreptimeDBLogsHttpSink<S>
25where
26    S: Service<HttpRequest<PartitionKey>> + Send + 'static,
27    S::Future: Send + 'static,
28    S::Response: DriverResponse + Send + 'static,
29    S::Error: std::fmt::Debug + Into<crate::Error> + Send,
30{
31    pub const fn new(
32        batcher_settings: BatcherSettings,
33        service: S,
34        request_builder: GreptimeDBLogsHttpRequestBuilder,
35        logs_sink_setting: LogsSinkSetting,
36    ) -> Self {
37        Self {
38            batcher_settings,
39            service,
40            request_builder,
41            logs_sink_setting,
42        }
43    }
44
45    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
46        let batcher_settings = self.batcher_settings;
47        input
48            .batched_partitioned(
49                KeyPartitioner::new(
50                    self.logs_sink_setting.dbname,
51                    self.logs_sink_setting.table,
52                    self.logs_sink_setting.pipeline_name,
53                    self.logs_sink_setting.pipeline_version,
54                ),
55                batcher_settings.timeout,
56                |_| batcher_settings.as_byte_size_config(),
57            )
58            .filter_map(|(key, batch)| async move { key.map(move |k| (k, batch)) })
59            .request_builder(
60                default_request_builder_concurrency_limit(),
61                self.request_builder,
62            )
63            .filter_map(|request| async {
64                match request {
65                    Err(error) => {
66                        emit!(SinkRequestBuildError { error });
67                        None
68                    }
69                    Ok(req) => Some(req),
70                }
71            })
72            .into_driver(self.service)
73            .run()
74            .await
75    }
76}
77
78#[async_trait::async_trait]
79impl<S> StreamSink<Event> for GreptimeDBLogsHttpSink<S>
80where
81    S: Service<HttpRequest<PartitionKey>> + Send + 'static,
82    S::Future: Send + 'static,
83    S::Response: DriverResponse + Send + 'static,
84    S::Error: std::fmt::Debug + Into<crate::Error> + Send,
85{
86    async fn run(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
87        self.run_inner(input).await
88    }
89}
vector/sinks/greptimedb/logs/sink.rs

vector/sinks/greptimedb/logs/
sink.rs
