mattp5657 commented on code in PR #3873: URL: https://github.com/apache/iggy/pull/3873#discussion_r3929962464
########## core/connectors/sinks/opensearch_sink/src/lib.rs: ########## @@ -0,0 +1,3085 @@ +// 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::{BufMut, Bytes, BytesMut}; +use iggy_common::{HeaderKey, HeaderValue, IggyTimestamp, calculate_256}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, + convert::owned_value_to_serde_json, + retry::{exponential_backoff, is_transient_status, jitter, parse_duration}, + sink_connector, +}; +use opensearch::{ + BulkParts, OpenSearch, + auth::Credentials, + cluster::ClusterHealthParts, + http::{ + StatusCode, + transport::{SingleNodeConnectionPool, TransportBuilder}, + }, + indices::{IndicesCreateParts, IndicesExistsParts}, + params::Refresh, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use serde_json::{Map, Value, json}; +use std::{ + collections::BTreeMap, + future::Future, + net::IpAddr, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; +use tokio::time::sleep; +use tracing::{debug, error, info, warn}; +use url::Url; + +sink_connector!(OpenSearchSink); + +const DEFAULT_CREATE_INDEX_IF_NOT_EXISTS: bool = true; +const DEFAULT_INCLUDE_METADATA: bool = true; +const DEFAULT_BATCH_SIZE: usize = 1000; +const DEFAULT_TIMEOUT: &str = "30s"; +const DEFAULT_RETRY_DELAY: &str = "500ms"; +const DEFAULT_MAX_RETRY_DELAY: &str = "5s"; +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_MAX_OPEN_RETRIES: u32 = 5; +const ENCODING_BASE64: &str = "base64"; +const ENCODING_UTF8: &str = "utf8"; +const GENERATED_ID_PREFIX: &str = "iggy_"; +const INDEX_ALREADY_EXISTS_ERROR: &str = "resource_already_exists_exception"; + +/// OpenSearch rejects `_id` values longer than 512 bytes. Payload-supplied IDs +/// are checked before the batch is built because that rejection fails the whole +/// `_bulk` call with an `action_request_validation_exception` rather than the +/// one item: a single oversized ID would cost every document in its chunk. +const MAX_DOCUMENT_ID_BYTES: usize = 512; + +// No `Serialize`: nothing serializes this type, and the only in-tree helper for +// a `SecretString` field writes the credential in plaintext. +#[derive(Debug, Default, Deserialize)] +pub struct OpenSearchSinkConfig { + pub url: String, + pub index: String, + pub username: Option<String>, + pub password: Option<SecretString>, + pub document_id_field: Option<String>, + pub create_index_if_not_exists: Option<bool>, + pub index_mapping: Option<Value>, + pub include_metadata: Option<bool>, + pub batch_size: Option<usize>, + pub timeout: Option<String>, + pub refresh: Option<Refresh>, + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub max_retry_delay: Option<String>, + pub max_open_retries: Option<u32>, + pub verbose_logging: Option<bool>, +} + +pub struct OpenSearchSink { + id: u32, + config: ResolvedOpenSearchSinkConfig, + client: Option<OpenSearch>, + invocations_count: AtomicU64, + documents_indexed: AtomicU64, + errors_count: AtomicU64, +} + +// `OpenSearch` derives `Debug` down through its `Transport`, and +// `opensearch::auth::Credentials::Basic` derives `Debug` on its raw +// `(String, String)` without redaction, so a derived `Debug` on this struct +// would print the Basic-auth password in plaintext once `client` is set. +impl std::fmt::Debug for OpenSearchSink { Review Comment: > This hides the authenticated client and relies on SecretString for the dedicated password field, but it still prints self.config.url verbatim through self.config. Since embedded URL credentials are rejected only later in open(), formatting a newly constructed sink from https://admin:hunter2@host exposes hunter2. Let's redact or sanitize the URL in the resolved config's Debug output. -- 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]
