ryerraguntla commented on code in PR #2933: URL: https://github.com/apache/iggy/pull/2933#discussion_r2988393354
########## core/connectors/sinks/influxdb_sink/src/lib.rs: ########## @@ -0,0 +1,1067 @@ +/* Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +use async_trait::async_trait; +use base64::{Engine as _, engine::general_purpose}; +use bytes::Bytes; +use iggy_connector_sdk::retry::{ + CircuitBreaker, build_retry_client, exponential_backoff, jitter, parse_duration, +}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Sink, TopicMetadata, sink_connector, +}; +use reqwest::Url; +use reqwest_middleware::ClientWithMiddleware; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; +use std::time::SystemTime; +use std::time::UNIX_EPOCH; +use tracing::{debug, error, info, warn}; +sink_connector!(InfluxDbSink); + +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_RETRY_DELAY: &str = "1s"; +const DEFAULT_TIMEOUT: &str = "30s"; +const DEFAULT_PRECISION: &str = "us"; +// Maximum attempts for open() connectivity retries +const DEFAULT_MAX_OPEN_RETRIES: u32 = 10; +// Cap for exponential backoff in open() — never wait longer than this +const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "60s"; +// Cap for exponential backoff on per-write retries — kept short so a +// transient InfluxDB blip does not stall message delivery for too long +const DEFAULT_RETRY_MAX_DELAY: &str = "5s"; +// How many consecutive batch failures open the circuit breaker +const DEFAULT_CIRCUIT_BREAKER_THRESHOLD: u32 = 5; +// How long the circuit stays open before allowing a probe attempt +const DEFAULT_CIRCUIT_COOL_DOWN: &str = "30s"; + +// --------------------------------------------------------------------------- +// Main connector structs +// --------------------------------------------------------------------------- + +#[derive(Debug)] +pub struct InfluxDbSink { + pub id: u32, + config: InfluxDbSinkConfig, + /// `None` until `open()` is called. Wraps `reqwest::Client` with + /// [`InfluxDbRetryMiddleware`] so retry/back-off/jitter is handled + /// transparently by the middleware stack instead of a hand-rolled loop. + client: Option<ClientWithMiddleware>, + /// Cached once in `open()` — config fields never change at runtime. + write_url: Option<Url>, + messages_attempted: AtomicU64, + write_success: AtomicU64, + write_errors: AtomicU64, + verbose: bool, + retry_delay: Duration, + circuit_breaker: Arc<CircuitBreaker>, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct InfluxDbSinkConfig { + pub url: String, + pub org: String, + pub bucket: String, + pub token: String, + pub measurement: Option<String>, + pub precision: Option<String>, + pub batch_size: Option<u32>, + pub include_metadata: Option<bool>, + pub include_checksum: Option<bool>, + pub include_origin_timestamp: Option<bool>, + pub include_stream_tag: Option<bool>, + pub include_topic_tag: Option<bool>, + pub include_partition_tag: Option<bool>, + pub payload_format: Option<String>, + pub verbose_logging: Option<bool>, + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub timeout: Option<String>, + // How many times open() will retry before giving up + pub max_open_retries: Option<u32>, + // Upper cap on open() backoff delay — can be set high (e.g. "60s") for + // patient startup without affecting per-write retry behaviour + pub open_retry_max_delay: Option<String>, + // Upper cap on per-write retry backoff — kept short so a transient blip + // does not stall message delivery; independent of open_retry_max_delay + pub retry_max_delay: Option<String>, + // Circuit breaker configuration + pub circuit_breaker_threshold: Option<u32>, + pub circuit_breaker_cool_down: Option<String>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +enum PayloadFormat { + #[default] + Json, + Text, + Base64, +} + +impl PayloadFormat { + fn from_config(value: Option<&str>) -> Self { + match value.map(|v| v.to_ascii_lowercase()).as_deref() { + Some("text") | Some("utf8") => PayloadFormat::Text, + Some("base64") | Some("raw") => PayloadFormat::Base64, + Some("json") => PayloadFormat::Json, + other => { + warn!( + "Unrecognized payload_format value {:?}, falling back to JSON. \ + Valid values are: \"json\", \"text\", \"utf8\", \"base64\", \"raw\".", + other + ); + PayloadFormat::Json + } + } + } +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Write an escaped measurement name into `buf`. +/// Escapes: `\` → `\\`, `,` → `\,`, ` ` → `\ ` +fn write_measurement(buf: &mut String, value: &str) { + for ch in value.chars() { + match ch { + '\\' => buf.push_str("\\\\"), + ',' => buf.push_str("\\,"), + ' ' => buf.push_str("\\ "), + _ => buf.push(ch), + } + } +} + +/// Write an escaped tag key/value into `buf`. +/// Escapes: `\` → `\\`, `,` → `\,`, `=` → `\=`, ` ` → `\ ` +fn write_tag_value(buf: &mut String, value: &str) { + for ch in value.chars() { + match ch { + '\\' => buf.push_str("\\\\"), + ',' => buf.push_str("\\,"), + '=' => buf.push_str("\\="), + ' ' => buf.push_str("\\ "), + _ => buf.push(ch), + } + } +} + +/// Write an escaped string field value (without surrounding quotes) into `buf`. +/// Escapes: `\` → `\\`, `"` → `\"` +fn write_field_string(buf: &mut String, value: &str) { Review Comment: Done. Also let me check for any other characters. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
