Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(new sink): Adding greptimedb metrics sink (#17198)
This patch adds [greptimedb](https://github.com/greptimeteam/greptimedb) sink. For now, we use greptimedb's git repository for adding its client. We will eventually split it from repo and publish it to crates.io, I hope this is not a road blocker at the moment. TODOs: - [x] website docs - [x] spellcheck cleanup - [x] integration test verification --------- Co-authored-by: Doug Smith <dsmith3197@users.noreply.github.com>
- Loading branch information
1 parent
5b1219f
commit da36a5e
Showing
22 changed files
with
1,465 additions
and
1 deletion.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -296,6 +296,9 @@ gpg | |
| gql | ||
| grafana | ||
| graphiql | ||
| greptime | ||
| greptimecloud | ||
| greptimedb | ||
| gvisor | ||
| gws | ||
| hadoop | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,8 @@ | ||
| version: '3' | ||
|
|
||
| services: | ||
| greptimedb: | ||
| image: docker.io/greptime/greptimedb:${CONFIG_VERSION} | ||
| command: "standalone start --http-addr=0.0.0.0:4000 --rpc-addr=0.0.0.0:4001" | ||
| healthcheck: | ||
| test: "curl -f localhost:4000/health || exit 1" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| features: | ||
| - greptimedb-integration-tests | ||
|
|
||
| test_filter: '::greptimedb::' | ||
|
|
||
| runner: | ||
| env: | ||
| GREPTIMEDB_ENDPOINT: greptimedb:4001 | ||
| GREPTIMEDB_HTTP: http://greptimedb:4000 | ||
|
|
||
| matrix: | ||
| version: ['latest'] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| use vector_core::{ | ||
| event::{Metric, MetricValue}, | ||
| stream::batcher::limiter::ItemBatchSize, | ||
| }; | ||
|
|
||
| use super::request_builder::{ | ||
| DISTRIBUTION_QUANTILES, DISTRIBUTION_STAT_FIELD_COUNT, SUMMARY_STAT_FIELD_COUNT, | ||
| }; | ||
|
|
||
| const F64_BYTE_SIZE: usize = 8; | ||
| const I64_BYTE_SIZE: usize = 8; | ||
|
|
||
| #[derive(Default)] | ||
| pub(super) struct GreptimeDBBatchSizer; | ||
|
|
||
| impl GreptimeDBBatchSizer { | ||
| pub(super) fn estimated_size_of(&self, item: &Metric) -> usize { | ||
| // Metric name. | ||
| item.series().name().name().len() | ||
| // Metric namespace, with an additional 1 to account for the namespace separator. | ||
| + item.series().name().namespace().map(|s| s.len() + 1).unwrap_or(0) | ||
| // Metric tags, with an additional 1 per tag to account for the tag key/value separator. | ||
| + item.series().tags().map(|t| { | ||
| t.iter_all().map(|(k, v)| { | ||
| k.len() + 1 + v.map(|v| v.len()).unwrap_or(0) | ||
| }) | ||
| .sum() | ||
| }) | ||
| .unwrap_or(0) | ||
| // timestamp | ||
| + I64_BYTE_SIZE | ||
| + | ||
| // value size | ||
| match item.value() { | ||
| MetricValue::Counter { .. } | MetricValue::Gauge { .. } | MetricValue::Set { ..} => F64_BYTE_SIZE, | ||
| MetricValue::Distribution { .. } => F64_BYTE_SIZE * (DISTRIBUTION_QUANTILES.len() + DISTRIBUTION_STAT_FIELD_COUNT), | ||
| MetricValue::AggregatedHistogram { buckets, .. } => F64_BYTE_SIZE * (buckets.len() + SUMMARY_STAT_FIELD_COUNT), | ||
| MetricValue::AggregatedSummary { quantiles, .. } => F64_BYTE_SIZE * (quantiles.len() + SUMMARY_STAT_FIELD_COUNT), | ||
| MetricValue::Sketch { .. } => F64_BYTE_SIZE * (DISTRIBUTION_QUANTILES.len() + DISTRIBUTION_STAT_FIELD_COUNT), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| impl ItemBatchSize<Metric> for GreptimeDBBatchSizer { | ||
| fn size(&self, item: &Metric) -> usize { | ||
| self.estimated_size_of(item) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| use chrono::{DateTime, Duration, Utc}; | ||
| use futures::stream; | ||
| use vector_core::event::{Event, Metric, MetricKind, MetricValue}; | ||
| use vector_core::metric_tags; | ||
|
|
||
| use crate::sinks::util::test::load_sink; | ||
| use crate::{ | ||
| config::{SinkConfig, SinkContext}, | ||
| test_util::{ | ||
| components::{run_and_assert_sink_compliance, SINK_TAGS}, | ||
| trace_init, | ||
| }, | ||
| }; | ||
|
|
||
| use super::GreptimeDBConfig; | ||
|
|
||
| #[tokio::test] | ||
| async fn test_greptimedb_sink() { | ||
| trace_init(); | ||
| let cfg = format!( | ||
| r#"endpoint= "{}" | ||
| "#, | ||
| std::env::var("GREPTIMEDB_ENDPOINT").unwrap_or_else(|_| "localhost:4001".to_owned()) | ||
| ); | ||
|
|
||
| let (config, _) = load_sink::<GreptimeDBConfig>(&cfg).unwrap(); | ||
| let (sink, _hc) = config.build(SinkContext::default()).await.unwrap(); | ||
|
|
||
| let query_client = query_client(); | ||
|
|
||
| // Drop the table and data inside | ||
| let _ = query_client | ||
| .get(&format!( | ||
| "{}/v1/sql", | ||
| std::env::var("GREPTIMEDB_HTTP").unwrap_or_else(|_| "http://localhost:4000".to_owned()) | ||
| )) | ||
| .query(&[("sql", "DROP TABLE ns_my_counter")]) | ||
| .send() | ||
| .await | ||
| .unwrap(); | ||
|
|
||
| let base_time = Utc::now(); | ||
| let events: Vec<_> = (0..10).map(|idx| create_event(idx, base_time)).collect(); | ||
| run_and_assert_sink_compliance(sink, stream::iter(events), &SINK_TAGS).await; | ||
|
|
||
| let query_response = query_client | ||
| .get(&format!( | ||
| "{}/v1/sql", | ||
| std::env::var("GREPTIMEDB_HTTP").unwrap_or_else(|_| "http://localhost:4000".to_owned()) | ||
| )) | ||
| .query(&[("sql", "SELECT region, val FROM ns_my_counter")]) | ||
| .send() | ||
| .await | ||
| .unwrap() | ||
| .text() | ||
| .await | ||
| .expect("Fetch json from greptimedb failed"); | ||
| let result: serde_json::Value = | ||
| serde_json::from_str(&query_response).expect("Invalid json returned from greptimedb query"); | ||
| assert_eq!( | ||
| result | ||
| .pointer("/output/0/records/rows") | ||
| .and_then(|v| v.as_array()) | ||
| .expect("Error getting greptimedb response array") | ||
| .len(), | ||
| 10 | ||
| ) | ||
| } | ||
|
|
||
| fn query_client() -> reqwest::Client { | ||
| reqwest::Client::builder().build().unwrap() | ||
| } | ||
|
|
||
| fn create_event(i: i32, base_time: DateTime<Utc>) -> Event { | ||
| Event::Metric( | ||
| Metric::new( | ||
| "my_counter".to_owned(), | ||
| MetricKind::Incremental, | ||
| MetricValue::Counter { value: i as f64 }, | ||
| ) | ||
| .with_namespace(Some("ns")) | ||
| .with_tags(Some(metric_tags!( | ||
| "region" => "us-west-1", | ||
| "production" => "true", | ||
| ))) | ||
| .with_timestamp(Some(base_time + Duration::seconds(i as i64))), | ||
| ) | ||
| } |
Oops, something went wrong.