hubcio commented on code in PR #3523:
URL: https://github.com/apache/iggy/pull/3523#discussion_r3853116029
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
Review Comment:
this branch never fires against quickwit. v0.8.2 (the version pinned in the
integration fixture) and main both answer a duplicate `POST /api/v1/indexes`
with 400, never 409, and the body says `already exist(s)` - so
`contains("already exists")` is false and the race loser still dies with
`InitError`. the create also runs through the retry middleware now, so a create
that committed but lost its response gets re-posted and lands here too.
simplest robust fix: on any non-2xx, re-probe `has_index()` and treat `true`
as success; drop the 409 arm and the substring match. worth a small mock-http
test (surrealdb_sink has a `TcpListener` pattern for this) - it would have
caught this.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
Review Comment:
`new()` runs inside `iggy_sink_open`, an `extern "C"` fn, so this `expect`
panicking aborts the whole connectors process, not just this plugin. the
crate's own `config.toml` ships `index = ""`, which trips it (yaml eof error).
since `new()` can't fail, move the `index_id` extraction into `open()` and
return `Error::InvalidConfigValue`. also reject an empty id there - it turns
into `/api/v1//ingest`.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
+ {
+ info!(
+ "Quickwit index already exists ({status}): {} for
connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else {
+ error!(
+ "Failed creating Quickwit index: {} for connector ID: {}.
status: {status}, reason: {reason}",
+ self.index_id, self.id
+ );
+ Err(Error::InitError(format!(
+ "Failed to create index '{0}': {status} {reason}",
+ self.index_id
+ )))
+ }
}
-
- info!("Created index: {}", self.index_id);
- Ok(())
}
pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) ->
Result<(), Error> {
+ let client = self.client()?;
+ // At-least-once during transient retries, but at-most-once on final
failure:
+ // Quickwit ingest carries no dedup key, so a retry after a transient
5xx/timeout
+ // that actually committed double-writes those rows. Conversely, if a
batch permanently
+ // fails (e.g. 4xx client error or retries exhausted), the offset was
already committed
+ // at poll, so the batch is silently dropped and never redelivered.
let url = format!(
"{}/api/v1/{}/ingest?commit=auto",
- self.config.url, self.index_id
+ self.base_url, self.index_id
);
- info!("Ingesting messages for index: {}...", self.index_id);
- let messages_count = messages.len();
- let messages = messages
- .into_iter()
- .filter_map(|record| simd_json::to_string(&record).ok())
- .collect::<Vec<_>>()
- .join("\n");
-
- let response = self
- .client
+ let mut ndjson = String::with_capacity(messages.len() * 512);
+ let mut messages_count = 0;
+ for record in messages {
+ if let Ok(json_str) = simd_json::to_string(&record) {
+ if !ndjson.is_empty() {
+ ndjson.push('\n');
+ }
+ ndjson.push_str(&json_str);
+ messages_count += 1;
+ }
+ }
+
+ if messages_count == 0 {
+ return Ok(());
+ }
+
+ let response = client
.post(&url)
- .body(messages)
+ .header("Content-Type", "application/x-ndjson")
+ .body(ndjson)
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to ingest messages for index:
{}. {error}",
- self.index_id
+ "Failed to ingest {messages_count} messages into Quickwit
index: {} for connector ID: {}. {e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
+ let status = response.status();
+ if status.is_success() {
+ debug!(
+ "Ingested {messages_count} messages into Quickwit index: {}
for connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else if status.is_client_error() && status !=
StatusCode::TOO_MANY_REQUESTS {
let text = response.text().await.unwrap_or_default();
error!(
- "Received an invalid HTTP response when ingesting messages for
index: {}. Status code: {status}, reason: {text}",
- self.index_id
+ "Permanent error ingesting into Quickwit index: {} for
connector ID: {}. status: {status}, reason: {text}",
+ self.index_id, self.id
);
- return Err(Error::HttpRequestFailed(format!(
- "Status code: {status}, reason: {text}"
- )));
+ Err(Error::PermanentHttpError(format!(
+ "status: {status}, reason: {text}"
+ )))
+ } else {
+ let text = response.text().await.unwrap_or_default();
+ error!(
+ "Transient error ingesting into Quickwit index: {} for
connector ID: {}. status: {status}, reason: {text}",
+ self.index_id, self.id
+ );
+ Err(Error::HttpRequestFailed(format!(
+ "status: {status}, reason: {text}"
+ )))
}
+ }
- info!(
- "Ingested {messages_count} messages for index: {}",
- self.index_id
- );
- Ok(())
+ fn extract_json_payloads(
+ &self,
+ messages: Vec<ConsumedMessage>,
+ schema: Schema,
+ ) -> Vec<OwnedValue> {
+ let mut json_payloads = Vec::with_capacity(messages.len());
+ for message in messages {
+ let val = match message.payload {
+ Payload::Json(value) => value,
+ Payload::Raw(bytes) => {
+ let mut bytes_copy = bytes.clone();
+ match simd_json::from_slice::<OwnedValue>(&mut bytes_copy)
{
+ Ok(value) => value,
+ Err(_) => match String::from_utf8(bytes) {
+ Ok(text) => simd_json::json!({
+ "data": text,
+ "data_type": "raw"
+ }),
+ Err(err) => simd_json::json!({
+ "data":
general_purpose::STANDARD.encode(err.into_bytes()),
+ "data_type": "raw"
+ }),
+ },
+ }
+ }
+ Payload::Text(text) => simd_json::json!({
+ "text": text,
+ "data_type": "text"
+ }),
+ _ => {
+ warn!(
+ "Quickwit sink connector ID: {} unsupported payload
schema: {}",
+ self.id, schema
+ );
+ continue;
+ }
+ };
+ json_payloads.push(val);
+ }
+ json_payloads
}
}
#[async_trait]
impl Sink for QuickwitSink {
async fn open(&mut self) -> Result<(), Error> {
- info!(
- "Opened Quickwit sink connector with ID: {} for URL: {}",
- self.id, self.config.url
+ let retry_delay = parse_duration(self.config.retry_delay.as_deref(),
DEFAULT_RETRY_DELAY);
+ let retry_max_delay = parse_duration(
+ self.config.retry_max_delay.as_deref(),
+ DEFAULT_RETRY_MAX_DELAY,
);
+ let max_open_retries = self
+ .config
+ .max_open_retries
+ .unwrap_or(DEFAULT_MAX_OPEN_RETRIES);
+ let open_retry_max_delay = parse_duration(
+ self.config.open_retry_max_delay.as_deref(),
+ DEFAULT_OPEN_RETRY_MAX_DELAY,
+ );
+
+ let timeout = parse_duration(self.config.timeout.as_deref(),
DEFAULT_TIMEOUT);
Review Comment:
`parse_duration` in the sdk falls back to a hardcoded 1s on a parse error,
not to the default passed in. so `timeout = "30"` (unit missing) silently
becomes a 1s http timeout with one warning at open, and every large batch then
times out and gets dropped. either validate the four durations here and fail
`open()` with `InvalidConfigValue`, or fix the sdk helper to fall back to the
parsed default (http_sink's local helper already does that).
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
Review Comment:
a non-404 status here (401/403 behind a proxy, say) returns `Err` with no
log, and the sdk container discards the error value, so the runtime only prints
"plugin initialization failed". same for the `Url::parse` error at line 311.
log an `error!` in these arms like `create_index` does, or once at the tail of
`open()`. the old code did log the send error at least.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
+ {
+ info!(
+ "Quickwit index already exists ({status}): {} for
connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else {
+ error!(
+ "Failed creating Quickwit index: {} for connector ID: {}.
status: {status}, reason: {reason}",
+ self.index_id, self.id
+ );
+ Err(Error::InitError(format!(
+ "Failed to create index '{0}': {status} {reason}",
+ self.index_id
+ )))
+ }
}
-
- info!("Created index: {}", self.index_id);
- Ok(())
}
pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) ->
Result<(), Error> {
+ let client = self.client()?;
+ // At-least-once during transient retries, but at-most-once on final
failure:
+ // Quickwit ingest carries no dedup key, so a retry after a transient
5xx/timeout
+ // that actually committed double-writes those rows. Conversely, if a
batch permanently
+ // fails (e.g. 4xx client error or retries exhausted), the offset was
already committed
+ // at poll, so the batch is silently dropped and never redelivered.
let url = format!(
"{}/api/v1/{}/ingest?commit=auto",
- self.config.url, self.index_id
+ self.base_url, self.index_id
);
- info!("Ingesting messages for index: {}...", self.index_id);
- let messages_count = messages.len();
- let messages = messages
- .into_iter()
- .filter_map(|record| simd_json::to_string(&record).ok())
- .collect::<Vec<_>>()
- .join("\n");
-
- let response = self
- .client
+ let mut ndjson = String::with_capacity(messages.len() * 512);
+ let mut messages_count = 0;
+ for record in messages {
+ if let Ok(json_str) = simd_json::to_string(&record) {
+ if !ndjson.is_empty() {
+ ndjson.push('\n');
+ }
+ ndjson.push_str(&json_str);
+ messages_count += 1;
+ }
+ }
+
+ if messages_count == 0 {
+ return Ok(());
+ }
+
+ let response = client
.post(&url)
- .body(messages)
+ .header("Content-Type", "application/x-ndjson")
+ .body(ndjson)
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to ingest messages for index:
{}. {error}",
- self.index_id
+ "Failed to ingest {messages_count} messages into Quickwit
index: {} for connector ID: {}. {e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
+ let status = response.status();
+ if status.is_success() {
+ debug!(
+ "Ingested {messages_count} messages into Quickwit index: {}
for connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else if status.is_client_error() && status !=
StatusCode::TOO_MANY_REQUESTS {
let text = response.text().await.unwrap_or_default();
error!(
- "Received an invalid HTTP response when ingesting messages for
index: {}. Status code: {status}, reason: {text}",
- self.index_id
+ "Permanent error ingesting into Quickwit index: {} for
connector ID: {}. status: {status}, reason: {text}",
+ self.index_id, self.id
);
- return Err(Error::HttpRequestFailed(format!(
- "Status code: {status}, reason: {text}"
- )));
+ Err(Error::PermanentHttpError(format!(
+ "status: {status}, reason: {text}"
+ )))
+ } else {
+ let text = response.text().await.unwrap_or_default();
+ error!(
+ "Transient error ingesting into Quickwit index: {} for
connector ID: {}. status: {status}, reason: {text}",
+ self.index_id, self.id
+ );
+ Err(Error::HttpRequestFailed(format!(
+ "status: {status}, reason: {text}"
+ )))
}
+ }
- info!(
- "Ingested {messages_count} messages for index: {}",
- self.index_id
- );
- Ok(())
+ fn extract_json_payloads(
+ &self,
+ messages: Vec<ConsumedMessage>,
+ schema: Schema,
+ ) -> Vec<OwnedValue> {
+ let mut json_payloads = Vec::with_capacity(messages.len());
+ for message in messages {
+ let val = match message.payload {
+ Payload::Json(value) => value,
+ Payload::Raw(bytes) => {
+ let mut bytes_copy = bytes.clone();
+ match simd_json::from_slice::<OwnedValue>(&mut bytes_copy)
{
Review Comment:
`from_slice` happily parses `42`, `"text"`, `[1,2]` or `null`, and those go
out as bare ndjson lines. quickwit accepts the request with 200 and then drops
every non-object line in the indexer, so the sink logs "ingested n" while the
data is gone - and a plain-text payload would have been wrapped and kept. gate
the parse on the first non-whitespace byte being `{` (that also skips the clone
+ parse for text payloads). the `Payload::Json` arm has the same hole for
array/scalar bodies.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
+ {
+ info!(
+ "Quickwit index already exists ({status}): {} for
connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else {
+ error!(
+ "Failed creating Quickwit index: {} for connector ID: {}.
status: {status}, reason: {reason}",
+ self.index_id, self.id
+ );
+ Err(Error::InitError(format!(
+ "Failed to create index '{0}': {status} {reason}",
+ self.index_id
+ )))
+ }
}
-
- info!("Created index: {}", self.index_id);
- Ok(())
}
pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) ->
Result<(), Error> {
+ let client = self.client()?;
+ // At-least-once during transient retries, but at-most-once on final
failure:
+ // Quickwit ingest carries no dedup key, so a retry after a transient
5xx/timeout
+ // that actually committed double-writes those rows. Conversely, if a
batch permanently
+ // fails (e.g. 4xx client error or retries exhausted), the offset was
already committed
+ // at poll, so the batch is silently dropped and never redelivered.
let url = format!(
"{}/api/v1/{}/ingest?commit=auto",
- self.config.url, self.index_id
+ self.base_url, self.index_id
);
- info!("Ingesting messages for index: {}...", self.index_id);
- let messages_count = messages.len();
- let messages = messages
- .into_iter()
- .filter_map(|record| simd_json::to_string(&record).ok())
- .collect::<Vec<_>>()
- .join("\n");
-
- let response = self
- .client
+ let mut ndjson = String::with_capacity(messages.len() * 512);
+ let mut messages_count = 0;
+ for record in messages {
+ if let Ok(json_str) = simd_json::to_string(&record) {
+ if !ndjson.is_empty() {
+ ndjson.push('\n');
+ }
+ ndjson.push_str(&json_str);
+ messages_count += 1;
+ }
+ }
+
+ if messages_count == 0 {
+ return Ok(());
+ }
+
+ let response = client
.post(&url)
- .body(messages)
+ .header("Content-Type", "application/x-ndjson")
+ .body(ndjson)
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to ingest messages for index:
{}. {error}",
- self.index_id
+ "Failed to ingest {messages_count} messages into Quickwit
index: {} for connector ID: {}. {e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
+ let status = response.status();
+ if status.is_success() {
+ debug!(
+ "Ingested {messages_count} messages into Quickwit index: {}
for connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else if status.is_client_error() && status !=
StatusCode::TOO_MANY_REQUESTS {
let text = response.text().await.unwrap_or_default();
error!(
- "Received an invalid HTTP response when ingesting messages for
index: {}. Status code: {status}, reason: {text}",
- self.index_id
+ "Permanent error ingesting into Quickwit index: {} for
connector ID: {}. status: {status}, reason: {text}",
+ self.index_id, self.id
);
- return Err(Error::HttpRequestFailed(format!(
- "Status code: {status}, reason: {text}"
- )));
+ Err(Error::PermanentHttpError(format!(
+ "status: {status}, reason: {text}"
+ )))
+ } else {
+ let text = response.text().await.unwrap_or_default();
+ error!(
+ "Transient error ingesting into Quickwit index: {} for
connector ID: {}. status: {status}, reason: {text}",
+ self.index_id, self.id
+ );
+ Err(Error::HttpRequestFailed(format!(
+ "status: {status}, reason: {text}"
+ )))
}
+ }
- info!(
- "Ingested {messages_count} messages for index: {}",
- self.index_id
- );
- Ok(())
+ fn extract_json_payloads(
+ &self,
+ messages: Vec<ConsumedMessage>,
+ schema: Schema,
+ ) -> Vec<OwnedValue> {
+ let mut json_payloads = Vec::with_capacity(messages.len());
+ for message in messages {
+ let val = match message.payload {
+ Payload::Json(value) => value,
+ Payload::Raw(bytes) => {
+ let mut bytes_copy = bytes.clone();
+ match simd_json::from_slice::<OwnedValue>(&mut bytes_copy)
{
+ Ok(value) => value,
+ Err(_) => match String::from_utf8(bytes) {
+ Ok(text) => simd_json::json!({
+ "data": text,
+ "data_type": "raw"
+ }),
+ Err(err) => simd_json::json!({
+ "data":
general_purpose::STANDARD.encode(err.into_bytes()),
Review Comment:
utf-8 text and base64 both land under `data_type: "raw"`, so a reader can't
tell which one it got (short text is itself valid base64). add a
`data_encoding` field - `utf8` / `base64` - like meilisearch_sink does.
flattening the two arms into `let (data, encoding) = match
String::from_utf8(bytes) { ... }` and one `json!` also removes the duplicated
literal.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -16,29 +16,65 @@
// under the License.
use async_trait::async_trait;
+use base64::{Engine as _, engine::general_purpose};
+use iggy_connector_sdk::retry::{
+ ConnectivityConfig, build_retry_client, check_connectivity_with_retry,
parse_duration,
+};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
sink_connector,
+ ConsumedMessage, Error, MessagesMetadata, Payload, Schema, Sink,
TopicMetadata, sink_connector,
};
-use serde::{Deserialize, Serialize};
-use tracing::{error, info, warn};
+use reqwest::StatusCode;
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use serde::Deserialize;
+use simd_json::OwnedValue;
+use tracing::{debug, error, info, warn};
sink_connector!(QuickwitSink);
+const DEFAULT_MAX_RETRIES: u32 = 3;
Review Comment:
with these defaults the readiness probe sleeps 0.4 + 0.8 + 1.6 + 3.2s and
gives up after about 6s, and the 30s cap never kicks in (it needs 8 attempts),
so `open_retry_max_delay` is effectively inert and the doc on line 71
overstates it. a quickwit that takes longer to come up leaves the sink dead
until someone restarts it.
same base drives the ingest retries: 3 total attempts give 0.4 + 0.8s before
the batch is gone, and quickwit answers 429 without `Retry-After` when its
ingest queue hits the disk/memory limit, which lasts until the indexer drains.
influxdb uses 10 attempts with a 1s base - `DEFAULT_RETRY_DELAY = "1s"` and
`DEFAULT_MAX_OPEN_RETRIES = 10` fixes both windows.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -16,29 +16,65 @@
// under the License.
use async_trait::async_trait;
+use base64::{Engine as _, engine::general_purpose};
+use iggy_connector_sdk::retry::{
+ ConnectivityConfig, build_retry_client, check_connectivity_with_retry,
parse_duration,
+};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
sink_connector,
+ ConsumedMessage, Error, MessagesMetadata, Payload, Schema, Sink,
TopicMetadata, sink_connector,
};
-use serde::{Deserialize, Serialize};
-use tracing::{error, info, warn};
+use reqwest::StatusCode;
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use serde::Deserialize;
+use simd_json::OwnedValue;
+use tracing::{debug, error, info, warn};
sink_connector!(QuickwitSink);
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "200ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 5;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "30s";
+const DEFAULT_TIMEOUT: &str = "30s";
+
#[derive(Debug)]
pub struct QuickwitSink {
id: u32,
config: QuickwitSinkConfig,
- client: reqwest::Client,
+ client: Option<ClientWithMiddleware>,
+ verbose: bool,
index_id: String,
+ base_url: String,
}
-#[derive(Debug, Serialize, Deserialize)]
+/// Configuration for the Quickwit sink connector, deserialized from
[plugin_config] in config.toml.
Review Comment:
`[plugin_config]` without backticks is a rustdoc shortcut link, so this
fails `cargo doc` under `-D warnings` (CI runs that now, so this goes red on
rebase). backticks fix it, see `http_sink` for the same line.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
+ {
+ info!(
+ "Quickwit index already exists ({status}): {} for
connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else {
+ error!(
+ "Failed creating Quickwit index: {} for connector ID: {}.
status: {status}, reason: {reason}",
+ self.index_id, self.id
+ );
+ Err(Error::InitError(format!(
+ "Failed to create index '{0}': {status} {reason}",
+ self.index_id
+ )))
+ }
}
-
- info!("Created index: {}", self.index_id);
- Ok(())
}
pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) ->
Result<(), Error> {
+ let client = self.client()?;
+ // At-least-once during transient retries, but at-most-once on final
failure:
+ // Quickwit ingest carries no dedup key, so a retry after a transient
5xx/timeout
+ // that actually committed double-writes those rows. Conversely, if a
batch permanently
Review Comment:
"silently dropped" isn't quite right - the batch is logged at `error!` here
and again by the sdk container, it's just never redelivered. also the readme
still lists only `url` and `index`; the 7 new keys and the
at-least-once-on-retry semantics (with `max_retries = 1` as the at-most-once
knob) need documenting there.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -16,29 +16,65 @@
// under the License.
use async_trait::async_trait;
+use base64::{Engine as _, engine::general_purpose};
+use iggy_connector_sdk::retry::{
+ ConnectivityConfig, build_retry_client, check_connectivity_with_retry,
parse_duration,
+};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
sink_connector,
+ ConsumedMessage, Error, MessagesMetadata, Payload, Schema, Sink,
TopicMetadata, sink_connector,
};
-use serde::{Deserialize, Serialize};
-use tracing::{error, info, warn};
+use reqwest::StatusCode;
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use serde::Deserialize;
+use simd_json::OwnedValue;
+use tracing::{debug, error, info, warn};
sink_connector!(QuickwitSink);
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "200ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 5;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "30s";
+const DEFAULT_TIMEOUT: &str = "30s";
+
#[derive(Debug)]
pub struct QuickwitSink {
id: u32,
config: QuickwitSinkConfig,
- client: reqwest::Client,
+ client: Option<ClientWithMiddleware>,
+ verbose: bool,
index_id: String,
+ base_url: String,
}
-#[derive(Debug, Serialize, Deserialize)]
+/// Configuration for the Quickwit sink connector, deserialized from
[plugin_config] in config.toml.
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
pub struct QuickwitSinkConfig {
- url: String,
- index: String,
+ /// Target URL for the Quickwit service.
+ pub url: String,
+ /// Full Quickwit index config YAML, passed to `POST /api/v1/indexes` on
first open.
+ /// `index_id` is extracted from this YAML to build ingest URLs.
+ pub index: String,
+ /// Enable verbose logging for ingested messages (default: false).
+ pub verbose_logging: Option<bool>,
+ /// Maximum number of retries for transient ingest errors (default: 3).
Review Comment:
`max_retries` and `max_open_retries` are total attempt counts in the sdk
(`attempts < max_retries` after incrementing), so `max_retries = 1` means no
retry and 5 probes = 4 retries. say "total attempts including the first; 1
disables retries" here and on line 69 (doris_sink.toml has the wording).
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -16,29 +16,65 @@
// under the License.
use async_trait::async_trait;
+use base64::{Engine as _, engine::general_purpose};
+use iggy_connector_sdk::retry::{
+ ConnectivityConfig, build_retry_client, check_connectivity_with_retry,
parse_duration,
+};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
sink_connector,
+ ConsumedMessage, Error, MessagesMetadata, Payload, Schema, Sink,
TopicMetadata, sink_connector,
};
-use serde::{Deserialize, Serialize};
-use tracing::{error, info, warn};
+use reqwest::StatusCode;
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use serde::Deserialize;
+use simd_json::OwnedValue;
+use tracing::{debug, error, info, warn};
sink_connector!(QuickwitSink);
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "200ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 5;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "30s";
+const DEFAULT_TIMEOUT: &str = "30s";
+
#[derive(Debug)]
pub struct QuickwitSink {
id: u32,
config: QuickwitSinkConfig,
- client: reqwest::Client,
+ client: Option<ClientWithMiddleware>,
+ verbose: bool,
index_id: String,
+ base_url: String,
}
-#[derive(Debug, Serialize, Deserialize)]
+/// Configuration for the Quickwit sink connector, deserialized from
[plugin_config] in config.toml.
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
pub struct QuickwitSinkConfig {
- url: String,
- index: String,
+ /// Target URL for the Quickwit service.
+ pub url: String,
+ /// Full Quickwit index config YAML, passed to `POST /api/v1/indexes` on
first open.
+ /// `index_id` is extracted from this YAML to build ingest URLs.
+ pub index: String,
+ /// Enable verbose logging for ingested messages (default: false).
Review Comment:
the flag only gates the "received n messages" line; "ingested n" at line 219
stays `debug!` even when verbose. the old code logged at info per batch, now
there's nothing at info by default. gate both or fix the doc.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
+ {
+ info!(
+ "Quickwit index already exists ({status}): {} for
connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else {
+ error!(
+ "Failed creating Quickwit index: {} for connector ID: {}.
status: {status}, reason: {reason}",
+ self.index_id, self.id
+ );
+ Err(Error::InitError(format!(
+ "Failed to create index '{0}': {status} {reason}",
+ self.index_id
+ )))
+ }
}
-
- info!("Created index: {}", self.index_id);
- Ok(())
}
pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) ->
Result<(), Error> {
Review Comment:
`ingest` has no callers outside this file - can be private.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
+ {
+ info!(
+ "Quickwit index already exists ({status}): {} for
connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else {
+ error!(
+ "Failed creating Quickwit index: {} for connector ID: {}.
status: {status}, reason: {reason}",
+ self.index_id, self.id
+ );
+ Err(Error::InitError(format!(
+ "Failed to create index '{0}': {status} {reason}",
+ self.index_id
+ )))
+ }
}
-
- info!("Created index: {}", self.index_id);
- Ok(())
}
pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) ->
Result<(), Error> {
+ let client = self.client()?;
+ // At-least-once during transient retries, but at-most-once on final
failure:
+ // Quickwit ingest carries no dedup key, so a retry after a transient
5xx/timeout
+ // that actually committed double-writes those rows. Conversely, if a
batch permanently
+ // fails (e.g. 4xx client error or retries exhausted), the offset was
already committed
+ // at poll, so the batch is silently dropped and never redelivered.
let url = format!(
"{}/api/v1/{}/ingest?commit=auto",
- self.config.url, self.index_id
+ self.base_url, self.index_id
);
- info!("Ingesting messages for index: {}...", self.index_id);
- let messages_count = messages.len();
- let messages = messages
- .into_iter()
- .filter_map(|record| simd_json::to_string(&record).ok())
- .collect::<Vec<_>>()
- .join("\n");
-
- let response = self
- .client
+ let mut ndjson = String::with_capacity(messages.len() * 512);
+ let mut messages_count = 0;
+ for record in messages {
+ if let Ok(json_str) = simd_json::to_string(&record) {
+ if !ndjson.is_empty() {
+ ndjson.push('\n');
+ }
+ ndjson.push_str(&json_str);
+ messages_count += 1;
+ }
+ }
+
+ if messages_count == 0 {
+ return Ok(());
+ }
+
+ let response = client
.post(&url)
- .body(messages)
+ .header("Content-Type", "application/x-ndjson")
+ .body(ndjson)
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to ingest messages for index:
{}. {error}",
- self.index_id
+ "Failed to ingest {messages_count} messages into Quickwit
index: {} for connector ID: {}. {e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
+ let status = response.status();
+ if status.is_success() {
+ debug!(
+ "Ingested {messages_count} messages into Quickwit index: {}
for connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else if status.is_client_error() && status !=
StatusCode::TOO_MANY_REQUESTS {
let text = response.text().await.unwrap_or_default();
error!(
- "Received an invalid HTTP response when ingesting messages for
index: {}. Status code: {status}, reason: {text}",
- self.index_id
+ "Permanent error ingesting into Quickwit index: {} for
connector ID: {}. status: {status}, reason: {text}",
+ self.index_id, self.id
);
- return Err(Error::HttpRequestFailed(format!(
- "Status code: {status}, reason: {text}"
- )));
+ Err(Error::PermanentHttpError(format!(
+ "status: {status}, reason: {text}"
+ )))
+ } else {
+ let text = response.text().await.unwrap_or_default();
+ error!(
+ "Transient error ingesting into Quickwit index: {} for
connector ID: {}. status: {status}, reason: {text}",
+ self.index_id, self.id
+ );
+ Err(Error::HttpRequestFailed(format!(
+ "status: {status}, reason: {text}"
+ )))
}
+ }
- info!(
- "Ingested {messages_count} messages for index: {}",
- self.index_id
- );
- Ok(())
+ fn extract_json_payloads(
+ &self,
+ messages: Vec<ConsumedMessage>,
+ schema: Schema,
+ ) -> Vec<OwnedValue> {
+ let mut json_payloads = Vec::with_capacity(messages.len());
+ for message in messages {
+ let val = match message.payload {
+ Payload::Json(value) => value,
+ Payload::Raw(bytes) => {
+ let mut bytes_copy = bytes.clone();
+ match simd_json::from_slice::<OwnedValue>(&mut bytes_copy)
{
+ Ok(value) => value,
+ Err(_) => match String::from_utf8(bytes) {
+ Ok(text) => simd_json::json!({
+ "data": text,
+ "data_type": "raw"
+ }),
+ Err(err) => simd_json::json!({
+ "data":
general_purpose::STANDARD.encode(err.into_bytes()),
+ "data_type": "raw"
+ }),
+ },
+ }
+ }
+ Payload::Text(text) => simd_json::json!({
+ "text": text,
+ "data_type": "text"
+ }),
+ _ => {
+ warn!(
+ "Quickwit sink connector ID: {} unsupported payload
schema: {}",
+ self.id, schema
+ );
+ continue;
+ }
+ };
+ json_payloads.push(val);
+ }
+ json_payloads
}
}
#[async_trait]
impl Sink for QuickwitSink {
async fn open(&mut self) -> Result<(), Error> {
- info!(
- "Opened Quickwit sink connector with ID: {} for URL: {}",
- self.id, self.config.url
+ let retry_delay = parse_duration(self.config.retry_delay.as_deref(),
DEFAULT_RETRY_DELAY);
+ let retry_max_delay = parse_duration(
+ self.config.retry_max_delay.as_deref(),
+ DEFAULT_RETRY_MAX_DELAY,
);
+ let max_open_retries = self
+ .config
+ .max_open_retries
+ .unwrap_or(DEFAULT_MAX_OPEN_RETRIES);
+ let open_retry_max_delay = parse_duration(
+ self.config.open_retry_max_delay.as_deref(),
+ DEFAULT_OPEN_RETRY_MAX_DELAY,
+ );
+
+ let timeout = parse_duration(self.config.timeout.as_deref(),
DEFAULT_TIMEOUT);
+ let raw_client = reqwest::Client::builder()
+ .timeout(timeout)
+ .build()
+ .map_err(|e| Error::InitError(format!("reqwest client: {e}")))?;
+ let health_url = Url::parse(&format!("{}/health/readyz",
self.base_url))
+ .map_err(|e| Error::InvalidConfigValue(format!("url: {e}")))?;
+
+ check_connectivity_with_retry(
+ &raw_client,
+ health_url,
+ "Quickwit sink connector",
Review Comment:
the sdk template already appends "connector ... for connector ID", so this
reads "quickwit sink connector health check failed ... for connector ID: 1".
pass `"Quickwit sink"` like influxdb does.
##########
core/connectors/runtime/example_config/connectors/quickwit_sink.toml:
##########
@@ -53,6 +53,13 @@ fields = ["email", "created_at"]
[plugin_config]
url = "http://localhost:7280";
+verbose_logging = false
+# max_retries = 3
Review Comment:
`max_retries` is a total attempt count (1 = no retry) - worth the same
one-line comment doris_sink.toml has, here and on `max_open_retries`.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -16,29 +16,65 @@
// under the License.
use async_trait::async_trait;
+use base64::{Engine as _, engine::general_purpose};
+use iggy_connector_sdk::retry::{
+ ConnectivityConfig, build_retry_client, check_connectivity_with_retry,
parse_duration,
+};
use iggy_connector_sdk::{
- ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
sink_connector,
+ ConsumedMessage, Error, MessagesMetadata, Payload, Schema, Sink,
TopicMetadata, sink_connector,
};
-use serde::{Deserialize, Serialize};
-use tracing::{error, info, warn};
+use reqwest::StatusCode;
+use reqwest::Url;
+use reqwest_middleware::ClientWithMiddleware;
+use serde::Deserialize;
+use simd_json::OwnedValue;
+use tracing::{debug, error, info, warn};
sink_connector!(QuickwitSink);
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_RETRY_DELAY: &str = "200ms";
+const DEFAULT_RETRY_MAX_DELAY: &str = "5s";
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 5;
+const DEFAULT_OPEN_RETRY_MAX_DELAY: &str = "30s";
+const DEFAULT_TIMEOUT: &str = "30s";
+
#[derive(Debug)]
pub struct QuickwitSink {
id: u32,
config: QuickwitSinkConfig,
- client: reqwest::Client,
+ client: Option<ClientWithMiddleware>,
+ verbose: bool,
index_id: String,
+ base_url: String,
}
-#[derive(Debug, Serialize, Deserialize)]
+/// Configuration for the Quickwit sink connector, deserialized from
[plugin_config] in config.toml.
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
Review Comment:
heads-up on `deny_unknown_fields`: the runtime's env sweep turns
`IGGY_CONNECTORS_SINK_QUICKWIT_PLUGIN_CONFIG_FORMAT` (the typed override for
`plugin_config_format`) into a `format` key inside `plugin_config`, which now
fails deserialization and kills init - it was silently ignored before.
doris_sink renamed its field to dodge exactly this. keep the attribute, but
call out the behavior change in the PR description; the real fix is the runtime
skipping `FORMAT` in the sweep.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
+ {
+ info!(
+ "Quickwit index already exists ({status}): {} for
connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else {
+ error!(
+ "Failed creating Quickwit index: {} for connector ID: {}.
status: {status}, reason: {reason}",
+ self.index_id, self.id
+ );
+ Err(Error::InitError(format!(
+ "Failed to create index '{0}': {status} {reason}",
+ self.index_id
+ )))
+ }
}
-
- info!("Created index: {}", self.index_id);
- Ok(())
}
pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) ->
Result<(), Error> {
+ let client = self.client()?;
+ // At-least-once during transient retries, but at-most-once on final
failure:
+ // Quickwit ingest carries no dedup key, so a retry after a transient
5xx/timeout
+ // that actually committed double-writes those rows. Conversely, if a
batch permanently
+ // fails (e.g. 4xx client error or retries exhausted), the offset was
already committed
+ // at poll, so the batch is silently dropped and never redelivered.
let url = format!(
"{}/api/v1/{}/ingest?commit=auto",
- self.config.url, self.index_id
+ self.base_url, self.index_id
);
- info!("Ingesting messages for index: {}...", self.index_id);
- let messages_count = messages.len();
- let messages = messages
- .into_iter()
- .filter_map(|record| simd_json::to_string(&record).ok())
- .collect::<Vec<_>>()
- .join("\n");
-
- let response = self
- .client
+ let mut ndjson = String::with_capacity(messages.len() * 512);
+ let mut messages_count = 0;
+ for record in messages {
+ if let Ok(json_str) = simd_json::to_string(&record) {
+ if !ndjson.is_empty() {
+ ndjson.push('\n');
+ }
+ ndjson.push_str(&json_str);
+ messages_count += 1;
+ }
+ }
+
+ if messages_count == 0 {
+ return Ok(());
+ }
+
+ let response = client
.post(&url)
- .body(messages)
+ .header("Content-Type", "application/x-ndjson")
+ .body(ndjson)
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to ingest messages for index:
{}. {error}",
- self.index_id
+ "Failed to ingest {messages_count} messages into Quickwit
index: {} for connector ID: {}. {e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
+ let status = response.status();
+ if status.is_success() {
+ debug!(
+ "Ingested {messages_count} messages into Quickwit index: {}
for connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else if status.is_client_error() && status !=
StatusCode::TOO_MANY_REQUESTS {
Review Comment:
the two non-2xx arms differ only in the log label and the error variant;
read the body once and pick both from `is_transient_status(status)` (public in
the sdk), like influxdb_sink does. also worth saying in the comment above: the
permanent/transient split is log-only here - both variants become return code 1
at the ffi boundary, the runtime doesn't read it, and there's no circuit
breaker in this sink.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
+ {
+ info!(
+ "Quickwit index already exists ({status}): {} for
connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else {
+ error!(
+ "Failed creating Quickwit index: {} for connector ID: {}.
status: {status}, reason: {reason}",
+ self.index_id, self.id
+ );
+ Err(Error::InitError(format!(
+ "Failed to create index '{0}': {status} {reason}",
+ self.index_id
+ )))
+ }
}
-
- info!("Created index: {}", self.index_id);
- Ok(())
}
pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) ->
Result<(), Error> {
+ let client = self.client()?;
+ // At-least-once during transient retries, but at-most-once on final
failure:
+ // Quickwit ingest carries no dedup key, so a retry after a transient
5xx/timeout
+ // that actually committed double-writes those rows. Conversely, if a
batch permanently
+ // fails (e.g. 4xx client error or retries exhausted), the offset was
already committed
+ // at poll, so the batch is silently dropped and never redelivered.
let url = format!(
"{}/api/v1/{}/ingest?commit=auto",
- self.config.url, self.index_id
+ self.base_url, self.index_id
);
- info!("Ingesting messages for index: {}...", self.index_id);
- let messages_count = messages.len();
- let messages = messages
- .into_iter()
- .filter_map(|record| simd_json::to_string(&record).ok())
- .collect::<Vec<_>>()
- .join("\n");
-
- let response = self
- .client
+ let mut ndjson = String::with_capacity(messages.len() * 512);
+ let mut messages_count = 0;
+ for record in messages {
+ if let Ok(json_str) = simd_json::to_string(&record) {
+ if !ndjson.is_empty() {
+ ndjson.push('\n');
+ }
+ ndjson.push_str(&json_str);
+ messages_count += 1;
+ }
+ }
+
+ if messages_count == 0 {
+ return Ok(());
+ }
+
+ let response = client
.post(&url)
- .body(messages)
+ .header("Content-Type", "application/x-ndjson")
+ .body(ndjson)
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to ingest messages for index:
{}. {error}",
- self.index_id
+ "Failed to ingest {messages_count} messages into Quickwit
index: {} for connector ID: {}. {e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
+ let status = response.status();
+ if status.is_success() {
+ debug!(
+ "Ingested {messages_count} messages into Quickwit index: {}
for connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else if status.is_client_error() && status !=
StatusCode::TOO_MANY_REQUESTS {
let text = response.text().await.unwrap_or_default();
error!(
- "Received an invalid HTTP response when ingesting messages for
index: {}. Status code: {status}, reason: {text}",
- self.index_id
+ "Permanent error ingesting into Quickwit index: {} for
connector ID: {}. status: {status}, reason: {text}",
+ self.index_id, self.id
);
- return Err(Error::HttpRequestFailed(format!(
- "Status code: {status}, reason: {text}"
- )));
+ Err(Error::PermanentHttpError(format!(
+ "status: {status}, reason: {text}"
+ )))
+ } else {
+ let text = response.text().await.unwrap_or_default();
+ error!(
+ "Transient error ingesting into Quickwit index: {} for
connector ID: {}. status: {status}, reason: {text}",
+ self.index_id, self.id
+ );
+ Err(Error::HttpRequestFailed(format!(
+ "status: {status}, reason: {text}"
+ )))
}
+ }
- info!(
- "Ingested {messages_count} messages for index: {}",
- self.index_id
- );
- Ok(())
+ fn extract_json_payloads(
+ &self,
+ messages: Vec<ConsumedMessage>,
+ schema: Schema,
+ ) -> Vec<OwnedValue> {
+ let mut json_payloads = Vec::with_capacity(messages.len());
+ for message in messages {
+ let val = match message.payload {
+ Payload::Json(value) => value,
+ Payload::Raw(bytes) => {
+ let mut bytes_copy = bytes.clone();
+ match simd_json::from_slice::<OwnedValue>(&mut bytes_copy)
{
+ Ok(value) => value,
+ Err(_) => match String::from_utf8(bytes) {
+ Ok(text) => simd_json::json!({
+ "data": text,
+ "data_type": "raw"
+ }),
+ Err(err) => simd_json::json!({
+ "data":
general_purpose::STANDARD.encode(err.into_bytes()),
+ "data_type": "raw"
+ }),
+ },
+ }
+ }
+ Payload::Text(text) => simd_json::json!({
+ "text": text,
+ "data_type": "text"
+ }),
+ _ => {
+ warn!(
Review Comment:
two things with this arm. it fires one `warn!` per message, each through the
ffi log callback, so an avro or flatbuffer stream at the default 1000-message
batches floods the log. and it drops data that's actually json: with the
default `extract_as_json`, avro payloads arrive here as json bytes labelled
`Payload::Avro`, so a default avro stream is dropped entirely with the offset
already committed. `Payload::Proto` never reaches a sink at all (the schema
decodes to json or raw).
handle `Raw | Avro | FlatBuffer` through the raw path and `Text | Proto`
through the text wrapper like http_sink does - that removes the `_` arm, the
warn and the `schema` param.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
+ {
+ info!(
+ "Quickwit index already exists ({status}): {} for
connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else {
+ error!(
+ "Failed creating Quickwit index: {} for connector ID: {}.
status: {status}, reason: {reason}",
+ self.index_id, self.id
+ );
+ Err(Error::InitError(format!(
+ "Failed to create index '{0}': {status} {reason}",
+ self.index_id
+ )))
+ }
}
-
- info!("Created index: {}", self.index_id);
- Ok(())
}
pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) ->
Result<(), Error> {
+ let client = self.client()?;
+ // At-least-once during transient retries, but at-most-once on final
failure:
+ // Quickwit ingest carries no dedup key, so a retry after a transient
5xx/timeout
+ // that actually committed double-writes those rows. Conversely, if a
batch permanently
+ // fails (e.g. 4xx client error or retries exhausted), the offset was
already committed
+ // at poll, so the batch is silently dropped and never redelivered.
let url = format!(
"{}/api/v1/{}/ingest?commit=auto",
- self.config.url, self.index_id
+ self.base_url, self.index_id
);
- info!("Ingesting messages for index: {}...", self.index_id);
- let messages_count = messages.len();
- let messages = messages
- .into_iter()
- .filter_map(|record| simd_json::to_string(&record).ok())
- .collect::<Vec<_>>()
- .join("\n");
-
- let response = self
- .client
+ let mut ndjson = String::with_capacity(messages.len() * 512);
+ let mut messages_count = 0;
+ for record in messages {
+ if let Ok(json_str) = simd_json::to_string(&record) {
Review Comment:
the loop allocates a `String` per record and copies it in, and the `if let
Ok` silently skips a serialization that can't really fail.
`simd_json::to_writer(&mut body, record)?` into one `Vec<u8>` with
`body.push(b'\n')` after each record does the same with no per-record copy (a
trailing newline is fine, quickwit skips blank lines). then `messages.len()` is
the count and the `== 0` guard here is redundant with the one in `consume`.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -47,127 +83,265 @@ impl QuickwitSink {
pub fn new(id: u32, config: QuickwitSinkConfig) -> Self {
let index_config =
serde_yaml_ng::from_str::<IndexConfig>(&config.index).expect("Invalid index
config.");
- QuickwitSink {
+ let verbose = config.verbose_logging.unwrap_or(false);
+ let base_url = config.url.trim_end_matches('/').to_owned();
+ Self {
id,
config,
+ client: None,
+ verbose,
index_id: index_config.index_id,
- client: reqwest::Client::new(),
+ base_url,
}
}
+ fn client(&self) -> Result<&ClientWithMiddleware, Error> {
+ self.client
+ .as_ref()
+ .ok_or_else(|| Error::InitError("Quickwit sink client not
initialized".into()))
+ }
+
async fn has_index(&self) -> Result<bool, Error> {
- let url = format!("{}/api/v1/indexes/{}", self.config.url,
self.index_id);
- let response = self.client.get(&url).send().await.map_err(|error| {
- error!(
- "Failed to send HTTP request to check if index with ID: {}
exists. {error}",
- self.index_id
- );
- Error::HttpRequestFailed(error.to_string())
- })?;
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes/{}", self.base_url,
self.index_id);
+ let response = client
+ .get(&url)
+ .send()
+ .await
+ .map_err(|e| Error::HttpRequestFailed(e.to_string()))?;
let status = response.status();
if status.is_success() {
Ok(true)
- } else if status == reqwest::StatusCode::NOT_FOUND {
+ } else if status == StatusCode::NOT_FOUND {
Ok(false)
} else {
Err(Error::HttpRequestFailed(format!(
- "Unexpected status code: {status}",
+ "Unexpected status checking Quickwit index: {status}"
)))
}
}
async fn create_index(&self) -> Result<(), Error> {
- info!("Creating index: {}", self.index_id);
- let url = format!("{}/api/v1/indexes", self.config.url);
- let response = self
- .client
+ info!(
+ "Creating Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
+ );
+ let client = self.client()?;
+ let url = format!("{}/api/v1/indexes", self.base_url);
+ let response = client
.post(&url)
- .header("content-type", "application/yaml")
- .body(self.config.index.to_owned())
+ .header("Content-Type", "application/yaml")
+ .body(self.config.index.clone())
.send()
.await
- .map_err(|error| {
+ .map_err(|e| {
error!(
- "Failed to send HTTP request to create index: {}. {error}",
- self.index_id
+ "Failed to create Quickwit index: {} for connector ID: {}.
{e}",
+ self.index_id, self.id
);
- Error::HttpRequestFailed(error.to_string())
+ Error::HttpRequestFailed(e.to_string())
})?;
- if !response.status().is_success() {
- let status = response.status();
- let reason = response.text().await.unwrap_or_default();
- error!(
- "Received an invalid HTTP response when creating index: {}.
Status code: {status}, reason: {reason}",
- self.index_id
+ let status = response.status();
+ if status.is_success() {
+ info!(
+ "Created Quickwit index: {} for connector ID: {}",
+ self.index_id, self.id
);
- return Err(Error::InitError(format!(
- "Failed to create index: {}. {reason}",
- self.index_id
- )));
+ Ok(())
+ } else {
+ let reason = response.text().await.unwrap_or_default();
+ if status == StatusCode::CONFLICT
+ || (status == StatusCode::BAD_REQUEST
+ && reason.to_lowercase().contains("already exists"))
+ {
+ info!(
+ "Quickwit index already exists ({status}): {} for
connector ID: {}",
+ self.index_id, self.id
+ );
+ Ok(())
+ } else {
+ error!(
+ "Failed creating Quickwit index: {} for connector ID: {}.
status: {status}, reason: {reason}",
+ self.index_id, self.id
+ );
+ Err(Error::InitError(format!(
+ "Failed to create index '{0}': {status} {reason}",
+ self.index_id
+ )))
+ }
}
-
- info!("Created index: {}", self.index_id);
- Ok(())
}
pub async fn ingest(&self, messages: Vec<simd_json::OwnedValue>) ->
Result<(), Error> {
+ let client = self.client()?;
+ // At-least-once during transient retries, but at-most-once on final
failure:
+ // Quickwit ingest carries no dedup key, so a retry after a transient
5xx/timeout
+ // that actually committed double-writes those rows. Conversely, if a
batch permanently
+ // fails (e.g. 4xx client error or retries exhausted), the offset was
already committed
+ // at poll, so the batch is silently dropped and never redelivered.
let url = format!(
Review Comment:
this sends the whole batch as one post. quickwit's default
`content_length_limit` is 10 mib, so 1000 messages of ~10 kib each get a 413
and the batch is dropped. chunk the body at a fixed cap (8 mib leaves headroom)
and post per chunk.
##########
core/connectors/sinks/quickwit_sink/src/lib.rs:
##########
@@ -177,33 +351,172 @@ impl Sink for QuickwitSink {
messages_metadata: MessagesMetadata,
messages: Vec<ConsumedMessage>,
) -> Result<(), Error> {
- info!(
- "Quickwit sink with ID: {} received: {} messages, format: {}",
- self.id,
- messages.len(),
- messages_metadata.schema
- );
-
- let mut json_payloads = Vec::with_capacity(messages.len());
- for message in messages {
- match message.payload {
- Payload::Json(value) => json_payloads.push(value),
- _ => {
- warn!("Unsupported payload format: {}",
messages_metadata.schema);
- }
- }
+ let total = messages.len();
+ if self.verbose {
+ info!(
+ "Quickwit sink connector ID: {} received {total} messages,
schema: {}",
+ self.id, messages_metadata.schema
+ );
+ } else {
+ debug!(
+ "Quickwit sink connector ID: {} received {total} messages,
schema: {}",
+ self.id, messages_metadata.schema
+ );
}
+ let json_payloads = self.extract_json_payloads(messages,
messages_metadata.schema);
if json_payloads.is_empty() {
return Ok(());
}
- self.ingest(json_payloads).await?;
- Ok(())
+ self.ingest(json_payloads).await
}
async fn close(&mut self) -> Result<(), Error> {
- info!("Quickwit sink connector with ID: {} is closed.", self.id);
+ let _ = self.client.take();
+ info!("Closed Quickwit sink connector ID: {}", self.id);
Ok(())
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn test_config() -> QuickwitSinkConfig {
+ QuickwitSinkConfig {
+ url: "http://localhost:7280".to_string(),
+ index: "index_id: test\nversion: 0.8\n".to_string(),
+ verbose_logging: None,
+ max_retries: None,
+ retry_delay: None,
+ retry_max_delay: None,
+ max_open_retries: None,
+ open_retry_max_delay: None,
+ timeout: None,
+ }
+ }
+
+ #[test]
+ fn given_default_config_verbose_should_be_false() {
+ let sink = QuickwitSink::new(1, test_config());
+ assert!(!sink.verbose);
+ }
+
+ #[test]
+ fn given_verbose_logging_enabled_should_set_verbose_flag() {
+ let mut config = test_config();
+ config.verbose_logging = Some(true);
+ let sink = QuickwitSink::new(1, config);
+ assert!(sink.verbose);
+ }
+
+ #[test]
Review Comment:
these two test the constructor literally (`unwrap_or(false)` with
`Some(false)`, and `client: None`). the missing test is the interesting one:
`create_index` getting a 400 "already exist(s)" and recovering - a small mock
http server (surrealdb_sink has the pattern) would have caught the dead branch
above. the 400 vs 503 classification in `ingest` is untested too.
--
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]
