rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-28 14:46:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add SASL and TLS config for Kafka client (#4536)

Browse Source 
* feat: add SASL and TLS config

* feat: add SASL/PLAIN and TLS config for Kafka client

* chore: use `ring`

* feat: support SASL SCRAM-SHA-256 and SCRAM-SHA-512

* fix: correct unit test

* test: add integration test

* chore: apply suggestions from CR

* refactor: introduce `KafkaConnectionConfig`

* chore: refine toml examples

* docs: add missing fields

* chore: refine examples

* feat: allow no server ca cert

* chore: refine examples

* chore: fix clippy

* feat: load system ca certs

* chore: fmt toml

* chore: unpin version

* Update src/common/wal/src/error.rs

Co-authored-by: Lei, HUANG <6406592+v0y4g3r@users.noreply.github.com>

---------

Co-authored-by: Lei, HUANG <6406592+v0y4g3r@users.noreply.github.com>
This commit is contained in:
Weny Xu
2024-08-12 12:27:11 +00:00
committed by GitHub
co-authored by Lei, HUANG
parent 9bcaeaaa0e
commit 2e2eacf3b2
24 changed files with 588 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/develop.yml
+1
View File
@@ -719,6 +719,7 @@ jobs:
GT_MINIO_ENDPOINT_URL: http://127.0.0.1:9000
GT_ETCD_ENDPOINTS: http://127.0.0.1:2379
GT_KAFKA_ENDPOINTS: 127.0.0.1:9092
GT_KAFKA_SASL_ENDPOINTS: 127.0.0.1:9093
UNITTEST_LOG_DIR: "__unittest_logs"
- name: Codecov upload
uses: codecov/codecov-action@v4
Cargo.lock
Generated
+46 -14
View File
@@ -2280,6 +2280,9 @@ dependencies = [
"futures-util",
"humantime-serde",
"rskafka",
"rustls 0.23.10",
"rustls-native-certs",
"rustls-pemfile 2.1.2",
"serde",
"serde_json",
"serde_with",
@@ -2445,6 +2448,15 @@ version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06ea2b9bc92be3c2baa9334a323ebca2d6f074ff852cd1d7b11064035cd3868f"
[[package]]
name = "core2"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b49ba7ef1ad6107f8824dbe97de947cbaac53c44e7f9756a1fba0d37c1eec505"
dependencies = [
"memchr",
]
[[package]]
name = "cpp_demangle"
version = "0.4.3"
@@ -8253,7 +8265,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22505a5c94da8e3b7c2996394d1c933236c4d743e81a410bcca4e6989fc066a4"
dependencies = [
"bytes",
"heck 0.4.1",
"heck 0.5.0",
"itertools 0.12.1",
"log",
"multimap",
@@ -8274,7 +8286,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5bb182580f71dd070f88d01ce3de9f4da5021db7115d2e1c3605a754153b77c1"
dependencies = [
"bytes",
"heck 0.4.1",
"heck 0.5.0",
"itertools 0.13.0",
"log",
"multimap",
@@ -9133,11 +9145,27 @@ dependencies = [
"zeroize",
]
[[package]]
name = "rsasl"
version = "2.0.2"
source = "git+https://github.com/wenyxu/rsasl.git?rev=06ebb683d5539c3410de4ce9fa37ff9b97e790a4#06ebb683d5539c3410de4ce9fa37ff9b97e790a4"
dependencies = [
"base64 0.22.1",
"core2",
"digest",
"hmac",
"pbkdf2",
"rand",
"serde_json",
"sha2",
"stringprep",
"thiserror",
]
[[package]]
name = "rskafka"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "132ecfa3cd9c3825208524a80881f115337762904ad3f0174e87975b2d79162c"
source = "git+https://github.com/WenyXu/rskafka.git?rev=940c6030012c5b746fad819fb72e3325b26e39de#940c6030012c5b746fad819fb72e3325b26e39de"
dependencies = [
"async-trait",
"bytes",
@@ -9150,11 +9178,14 @@ dependencies = [
"parking_lot 0.12.3",
"pin-project-lite",
"rand",
"rsasl",
"rustls 0.23.10",
"snap",
"thiserror",
"tokio",
"tokio-rustls 0.26.0",
"tracing",
"zstd 0.12.4",
"zstd 0.13.1",
]
[[package]]
@@ -9423,9 +9454,9 @@ dependencies = [
[[package]]
name = "rustls-native-certs"
version = "0.7.0"
version = "0.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f1fb85efa936c42c6d5fc28d2629bb51e4b2f4b8a5211e297d599cc5a093792"
checksum = "a88d6d420651b496bdd98684116959239430022a115c1240e6c3993be0b15fba"
dependencies = [
"openssl-probe",
"rustls-pemfile 2.1.2",
@@ -10127,12 +10158,13 @@ dependencies = [
[[package]]
name = "serde_json"
version = "1.0.117"
version = "1.0.122"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "455182ea6142b14f93f4bc5320a2b31c1f266b66a4a5c858b013302a5d8cbfc3"
checksum = "784b6203951c57ff748476b126ccb5e8e2959a5c19e5c617ab1956be3dbc68da"
dependencies = [
"indexmap 2.2.6",
"itoa",
"memchr",
"ryu",
"serde",
]
@@ -10565,7 +10597,7 @@ version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "38d1e02fca405f6280643174a50c942219f0bbf4dbf7d480f1dd864d6f211ae5"
dependencies = [
"heck 0.4.1",
"heck 0.5.0",
"proc-macro2",
"quote",
"syn 2.0.66",
@@ -11694,18 +11726,18 @@ checksum = "23d434d3f8967a09480fb04132ebe0a3e088c173e6d0ee7897abbdf4eab0f8b9"
[[package]]
name = "thiserror"
version = "1.0.61"
version = "1.0.63"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c546c80d6be4bc6a00c0f01730c08df82eaa7a7a61f11d656526506112cc1709"
checksum = "c0342370b38b6a11b6cc11d6a805569958d54cfa061a29969c3b5ce2ea405724"
dependencies = [
"thiserror-impl",
]
[[package]]
name = "thiserror-impl"
version = "1.0.61"
version = "1.0.63"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "46c3384250002a6d5af4d114f2845d37b57521033f30d5c3f46c4d70e1197533"
checksum = "a4558b58466b9ad7ca0f102865eccc95938dca1a74a856f2b57b6629050da261"
dependencies = [
"proc-macro2",
"quote",
Cargo.toml
+4 -1
View File
@@ -151,7 +151,10 @@ reqwest = { version = "0.12", default-features = false, features = [
"stream",
"multipart",
] }
rskafka = "0.5"
# SCRAM-SHA-512 requires https://github.com/dequbed/rsasl/pull/48, https://github.com/influxdata/rskafka/pull/247
rskafka = { git = "https://github.com/WenyXu/rskafka.git", rev = "940c6030012c5b746fad819fb72e3325b26e39de", features = [
"transport-tls",
] }
rstest = "0.21"
rstest_reuse = "0.7"
rust_decimal = "1.33"
config/config.md
+5
View File
@@ -67,6 +67,11 @@
| `wal.prefill_log_files` | Bool | `false` | Whether to pre-create log files on start up.<br/>**It's only used when the provider is `raft_engine`**. |
| `wal.sync_period` | String | `10s` | Duration for fsyncing log files.<br/>**It's only used when the provider is `raft_engine`**. |
| `wal.broker_endpoints` | Array | -- | The Kafka broker endpoints.<br/>**It's only used when the provider is `kafka`**. |
| `wal.num_topics` | Integer | `64` | Number of topics to be created upon start.<br/>**It's only used when the provider is `kafka`**. |
| `wal.selector_type` | String | `round_robin` | Topic selector type.<br/>Available selector types:<br/>- `round_robin` (default)<br/>**It's only used when the provider is `kafka`**. |
| `wal.topic_name_prefix` | String | `greptimedb_wal_topic` | A Kafka topic is constructed by concatenating `topic_name_prefix` and `topic_id`.<br/>**It's only used when the provider is `kafka`**. |
| `wal.replication_factor` | Integer | `1` | Expected number of replicas of each partition.<br/>**It's only used when the provider is `kafka`**. |
| `wal.create_topic_timeout` | String | `30s` | Above which a topic creation operation will be cancelled.<br/>**It's only used when the provider is `kafka`**. |
| `wal.max_batch_bytes` | String | `1MB` | The max size of a single producer batch.<br/>Warning: Kafka has a default limit of 1MB per message in a topic.<br/>**It's only used when the provider is `kafka`**. |
| `wal.consumer_wait_timeout` | String | `100ms` | The consumer wait timeout.<br/>**It's only used when the provider is `kafka`**. |
| `wal.backoff_init` | String | `500ms` | The initial backoff delay.<br/>**It's only used when the provider is `kafka`**. |
config/datanode.example.toml
+18
View File
@@ -187,6 +187,24 @@ backoff_base = 2
## **It's only used when the provider is `kafka`**.
backoff_deadline = "5mins"
# The Kafka SASL configuration.
# **It's only used when the provider is `kafka`**.
# Available SASL mechanisms:
# - `PLAIN`
# - `SCRAM-SHA-256`
# - `SCRAM-SHA-512`
# [wal.sasl]
# type = "SCRAM-SHA-512"
# username = "user_kafka"
# password = "secret"
# The Kafka TLS configuration.
# **It's only used when the provider is `kafka`**.
# [wal.tls]
# server_ca_cert_path = "/path/to/server_cert"
# client_cert_path = "/path/to/client_cert"
# client_key_path = "/path/to/key"
# Example of using S3 as the storage.
# [storage]
# type = "S3"
config/metasrv.example.toml
+18
View File
@@ -124,6 +124,24 @@ backoff_base = 2
## Stop reconnecting if the total wait time reaches the deadline. If this config is missing, the reconnecting won't terminate.
backoff_deadline = "5mins"
# The Kafka SASL configuration.
# **It's only used when the provider is `kafka`**.
# Available SASL mechanisms:
# - `PLAIN`
# - `SCRAM-SHA-256`
# - `SCRAM-SHA-512`
# [wal.sasl]
# type = "SCRAM-SHA-512"
# username = "user_kafka"
# password = "secret"
# The Kafka TLS configuration.
# **It's only used when the provider is `kafka`**.
# [wal.tls]
# server_ca_cert_path = "/path/to/server_cert"
# client_cert_path = "/path/to/client_cert"
# client_key_path = "/path/to/key"
## The logging options.
[logging]
## The directory to store the log files.
config/standalone.example.toml
+40
View File
@@ -171,6 +171,28 @@ sync_period = "10s"
## **It's only used when the provider is `kafka`**.
broker_endpoints = ["127.0.0.1:9092"]
## Number of topics to be created upon start.
## **It's only used when the provider is `kafka`**.
num_topics = 64
## Topic selector type.
## Available selector types:
## - `round_robin` (default)
## **It's only used when the provider is `kafka`**.
selector_type = "round_robin"
## A Kafka topic is constructed by concatenating `topic_name_prefix` and `topic_id`.
## **It's only used when the provider is `kafka`**.
topic_name_prefix = "greptimedb_wal_topic"
## Expected number of replicas of each partition.
## **It's only used when the provider is `kafka`**.
replication_factor = 1
## Above which a topic creation operation will be cancelled.
## **It's only used when the provider is `kafka`**.
create_topic_timeout = "30s"
## The max size of a single producer batch.
## Warning: Kafka has a default limit of 1MB per message in a topic.
## **It's only used when the provider is `kafka`**.
@@ -196,6 +218,24 @@ backoff_base = 2
## **It's only used when the provider is `kafka`**.
backoff_deadline = "5mins"
# The Kafka SASL configuration.
# **It's only used when the provider is `kafka`**.
# Available SASL mechanisms:
# - `PLAIN`
# - `SCRAM-SHA-256`
# - `SCRAM-SHA-512`
# [wal.sasl]
# type = "SCRAM-SHA-512"
# username = "user_kafka"
# password = "secret"
# The Kafka TLS configuration.
# **It's only used when the provider is `kafka`**.
# [wal.tls]
# server_ca_cert_path = "/path/to/server_cert"
# client_cert_path = "/path/to/client_cert"
# client_key_path = "/path/to/key"
## Metadata storage options.
[metadata_store]
## Kv file size in bytes.
src/common/meta/src/error.rs
+9 -1
View File
@@ -499,6 +499,13 @@ pub enum Error {
error: rskafka::client::error::Error,
},
#[snafu(display("Failed to create TLS Config"))]
TlsConfig {
#[snafu(implicit)]
location: Location,
source: common_wal::error::Error,
},
#[snafu(display("Failed to resolve Kafka broker endpoint."))]
ResolveKafkaEndpoint { source: common_wal::error::Error },
@@ -714,7 +721,8 @@ impl ErrorExt for Error {
| AlterLogicalTablesInvalidArguments { .. }
| CreateLogicalTablesInvalidArguments { .. }
| MismatchPrefix { .. }
| DelimiterNotFound { .. } => StatusCode::InvalidArguments,
| DelimiterNotFound { .. }
| TlsConfig { .. } => StatusCode::InvalidArguments,
FlowNotFound { .. } => StatusCode::FlowNotFound,
FlowRouteNotFound { .. } => StatusCode::Unexpected,
src/common/meta/src/wal_options_allocator.rs
+5 -2
View File
@@ -123,7 +123,7 @@ pub fn prepare_wal_options(
#[cfg(test)]
mod tests {
use common_wal::config::kafka::common::KafkaTopicConfig;
use common_wal::config::kafka::common::{KafkaConnectionConfig, KafkaTopicConfig};
use common_wal::config::kafka::MetasrvKafkaConfig;
use common_wal::test_util::run_test_with_kafka_wal;
@@ -166,7 +166,10 @@ mod tests {
..Default::default()
};
let config = MetasrvKafkaConfig {
broker_endpoints,
connection: KafkaConnectionConfig {
broker_endpoints,
..Default::default()
},
kafka_topic,
..Default::default()
};
src/common/meta/src/wal_options_allocator/kafka/topic_manager.rs
+19 -9
View File
@@ -30,7 +30,7 @@ use snafu::{ensure, ResultExt};
use crate::error::{
BuildKafkaClientSnafu, BuildKafkaCtrlClientSnafu, BuildKafkaPartitionClientSnafu,
CreateKafkaWalTopicSnafu, DecodeJsonSnafu, EncodeJsonSnafu, InvalidNumTopicsSnafu,
ProduceRecordSnafu, ResolveKafkaEndpointSnafu, Result,
ProduceRecordSnafu, ResolveKafkaEndpointSnafu, Result, TlsConfigSnafu,
};
use crate::kv_backend::KvBackendRef;
use crate::rpc::store::PutRequest;
@@ -117,15 +117,22 @@ impl TopicManager {
base: self.config.backoff.base as f64,
deadline: self.config.backoff.deadline,
};
let broker_endpoints = common_wal::resolve_to_ipv4(&self.config.broker_endpoints)
.await
.context(ResolveKafkaEndpointSnafu)?;
let client = ClientBuilder::new(broker_endpoints)
.backoff_config(backoff_config)
let broker_endpoints =
common_wal::resolve_to_ipv4(&self.config.connection.broker_endpoints)
.await
.context(ResolveKafkaEndpointSnafu)?;
let mut builder = ClientBuilder::new(broker_endpoints).backoff_config(backoff_config);
if let Some(sasl) = &self.config.connection.sasl {
builder = builder.sasl_config(sasl.config.clone().into_sasl_config());
};
if let Some(tls) = &self.config.connection.tls {
builder = builder.tls_config(tls.to_tls_config().await.context(TlsConfigSnafu)?)
};
let client = builder
.build()
.await
.with_context(|_| BuildKafkaClientSnafu {
broker_endpoints: self.config.broker_endpoints.clone(),
broker_endpoints: self.config.connection.broker_endpoints.clone(),
})?;
let control_client = client
@@ -242,7 +249,7 @@ impl TopicManager {
#[cfg(test)]
mod tests {
use common_wal::config::kafka::common::KafkaTopicConfig;
use common_wal::config::kafka::common::{KafkaConnectionConfig, KafkaTopicConfig};
use common_wal::test_util::run_test_with_kafka_wal;
use super::*;
@@ -289,7 +296,10 @@ mod tests {
..Default::default()
};
let config = MetasrvKafkaConfig {
broker_endpoints,
connection: KafkaConnectionConfig {
broker_endpoints,
..Default::default()
},
kafka_topic,
..Default::default()
};
src/common/wal/Cargo.toml
+3
View File
@@ -18,6 +18,9 @@ common-telemetry.workspace = true
futures-util.workspace = true
humantime-serde.workspace = true
rskafka.workspace = true
rustls = { version = "0.23", default-features = false, features = ["ring", "logging", "std", "tls12"] }
rustls-native-certs = "0.7"
rustls-pemfile = "2.1"
serde.workspace = true
serde_with.workspace = true
snafu.workspace = true
src/common/wal/src/config.rs
+40 -4
View File
@@ -23,6 +23,7 @@ use crate::config::raft_engine::RaftEngineConfig;
/// Wal configurations for metasrv.
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Default)]
#[serde(tag = "provider", rename_all = "snake_case")]
#[allow(clippy::large_enum_variant)]
pub enum MetasrvWalConfig {
#[default]
RaftEngine,
@@ -48,7 +49,7 @@ impl From<DatanodeWalConfig> for MetasrvWalConfig {
match config {
DatanodeWalConfig::RaftEngine(_) => Self::RaftEngine,
DatanodeWalConfig::Kafka(config) => Self::Kafka(MetasrvKafkaConfig {
broker_endpoints: config.broker_endpoints,
connection: config.connection,
backoff: config.backoff,
kafka_topic: config.kafka_topic,
}),
@@ -61,7 +62,7 @@ impl From<MetasrvWalConfig> for DatanodeWalConfig {
match config {
MetasrvWalConfig::RaftEngine => Self::RaftEngine(RaftEngineConfig::default()),
MetasrvWalConfig::Kafka(config) => Self::Kafka(DatanodeKafkaConfig {
broker_endpoints: config.broker_endpoints,
connection: config.connection,
backoff: config.backoff,
kafka_topic: config.kafka_topic,
..Default::default()
@@ -75,6 +76,9 @@ mod tests {
use std::time::Duration;
use common_base::readable_size::ReadableSize;
use kafka::common::{
KafkaClientSasl, KafkaClientSaslConfig, KafkaClientTls, KafkaConnectionConfig,
};
use tests::kafka::common::KafkaTopicConfig;
use super::*;
@@ -144,12 +148,31 @@ mod tests {
replication_factor = 1
create_topic_timeout = "30s"
topic_name_prefix = "greptimedb_wal_topic"
[tls]
server_ca_cert_path = "/path/to/server.pem"
[sasl]
type = "SCRAM-SHA-512"
username = "hi"
password = "test"
"#;
// Deserialized to MetasrvWalConfig.
let metasrv_wal_config: MetasrvWalConfig = toml::from_str(toml_str).unwrap();
let expected = MetasrvKafkaConfig {
broker_endpoints: vec!["127.0.0.1:9092".to_string()],
connection: KafkaConnectionConfig {
broker_endpoints: vec!["127.0.0.1:9092".to_string()],
sasl: Some(KafkaClientSasl {
config: KafkaClientSaslConfig::ScramSha512 {
username: "hi".to_string(),
password: "test".to_string(),
},
}),
tls: Some(KafkaClientTls {
server_ca_cert_path: Some("/path/to/server.pem".to_string()),
client_cert_path: None,
client_key_path: None,
}),
},
backoff: BackoffConfig {
init: Duration::from_millis(500),
max: Duration::from_secs(10),
@@ -170,7 +193,20 @@ mod tests {
// Deserialized to DatanodeWalConfig.
let datanode_wal_config: DatanodeWalConfig = toml::from_str(toml_str).unwrap();
let expected = DatanodeKafkaConfig {
broker_endpoints: vec!["127.0.0.1:9092".to_string()],
connection: KafkaConnectionConfig {
broker_endpoints: vec!["127.0.0.1:9092".to_string()],
sasl: Some(KafkaClientSasl {
config: KafkaClientSaslConfig::ScramSha512 {
username: "hi".to_string(),
password: "test".to_string(),
},
}),
tls: Some(KafkaClientTls {
server_ca_cert_path: Some("/path/to/server.pem".to_string()),
client_cert_path: None,
client_key_path: None,
}),
},
max_batch_bytes: ReadableSize::mb(1),
consumer_wait_timeout: Duration::from_millis(100),
backoff: BackoffConfig {
src/common/wal/src/config/kafka/common.rs
+136 -2
View File
@@ -12,16 +12,22 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::io::Cursor;
use std::sync::Arc;
use std::time::Duration;
use rskafka::client::{Credentials, SaslConfig};
use rustls::{ClientConfig, RootCertStore};
use serde::{Deserialize, Serialize};
use serde_with::with_prefix;
use snafu::{OptionExt, ResultExt};
use crate::{TopicSelectorType, TOPIC_NAME_PREFIX};
use crate::error::{self, Result};
use crate::{TopicSelectorType, BROKER_ENDPOINT, TOPIC_NAME_PREFIX};
with_prefix!(pub backoff_prefix "backoff_");
/// Backoff configurations for kafka clients.
/// Backoff configurations for kafka client.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct BackoffConfig {
@@ -49,6 +55,134 @@ impl Default for BackoffConfig {
}
}
/// The SASL configurations for kafka client.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct KafkaClientSasl {
#[serde(flatten)]
pub config: KafkaClientSaslConfig,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "type", rename_all = "SCREAMING-KEBAB-CASE")]
pub enum KafkaClientSaslConfig {
Plain {
username: String,
password: String,
},
#[serde(rename = "SCRAM-SHA-256")]
ScramSha256 {
username: String,
password: String,
},
#[serde(rename = "SCRAM-SHA-512")]
ScramSha512 {
username: String,
password: String,
},
}
impl KafkaClientSaslConfig {
/// Converts to [`SaslConfig`].
pub fn into_sasl_config(self) -> SaslConfig {
match self {
KafkaClientSaslConfig::Plain { username, password } => {
SaslConfig::Plain(Credentials::new(username, password))
}
KafkaClientSaslConfig::ScramSha256 { username, password } => {
SaslConfig::ScramSha256(Credentials::new(username, password))
}
KafkaClientSaslConfig::ScramSha512 { username, password } => {
SaslConfig::ScramSha512(Credentials::new(username, password))
}
}
}
}
/// The TLS configurations for kafka client.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct KafkaClientTls {
pub server_ca_cert_path: Option<String>,
pub client_cert_path: Option<String>,
pub client_key_path: Option<String>,
}
impl KafkaClientTls {
/// Builds the [`ClientConfig`].
pub async fn to_tls_config(&self) -> Result<Arc<ClientConfig>> {
let builder = ClientConfig::builder();
let mut roots = RootCertStore::empty();
if let Some(server_ca_cert_path) = &self.server_ca_cert_path {
let root_cert_bytes =
tokio::fs::read(&server_ca_cert_path)
.await
.context(error::ReadFileSnafu {
path: server_ca_cert_path,
})?;
let mut cursor = Cursor::new(root_cert_bytes);
for cert in rustls_pemfile::certs(&mut cursor)
.collect::<std::result::Result<Vec<_>, _>>()
.context(error::ReadCertsSnafu {
path: server_ca_cert_path,
})?
{
roots.add(cert).context(error::AddCertSnafu)?;
}
};
roots.add_parsable_certificates(
rustls_native_certs::load_native_certs().context(error::LoadSystemCertsSnafu)?,
);
let builder = builder.with_root_certificates(roots);
let config = if let (Some(cert_path), Some(key_path)) =
(&self.client_cert_path, &self.client_key_path)
{
let cert_bytes = tokio::fs::read(cert_path)
.await
.context(error::ReadFileSnafu { path: cert_path })?;
let client_certs = rustls_pemfile::certs(&mut Cursor::new(cert_bytes))
.collect::<std::result::Result<Vec<_>, _>>()
.context(error::ReadCertsSnafu { path: cert_path })?;
let key_bytes = tokio::fs::read(key_path)
.await
.context(error::ReadFileSnafu { path: key_path })?;
let client_key = rustls_pemfile::private_key(&mut Cursor::new(key_bytes))
.context(error::ReadKeySnafu { path: key_path })?
.context(error::KeyNotFoundSnafu { path: key_path })?;
builder
.with_client_auth_cert(client_certs, client_key)
.context(error::SetClientAuthCertSnafu)?
} else {
builder.with_no_client_auth()
};
Ok(Arc::new(config))
}
}
/// The connection configurations for kafka clients.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct KafkaConnectionConfig {
/// The broker endpoints of the Kafka cluster.
pub broker_endpoints: Vec<String>,
/// Client SASL.
pub sasl: Option<KafkaClientSasl>,
/// Client TLS config
pub tls: Option<KafkaClientTls>,
}
impl Default for KafkaConnectionConfig {
fn default() -> Self {
Self {
broker_endpoints: vec![BROKER_ENDPOINT.to_string()],
sasl: None,
tls: None,
}
}
}
/// Topic configurations for kafka clients.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
src/common/wal/src/config/kafka/datanode.rs
+5 -4
View File
@@ -17,15 +17,16 @@ use std::time::Duration;
use common_base::readable_size::ReadableSize;
use serde::{Deserialize, Serialize};
use super::common::KafkaConnectionConfig;
use crate::config::kafka::common::{backoff_prefix, BackoffConfig, KafkaTopicConfig};
use crate::BROKER_ENDPOINT;
/// Kafka wal configurations for datanode.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(default)]
pub struct DatanodeKafkaConfig {
/// The broker endpoints of the Kafka cluster.
pub broker_endpoints: Vec<String>,
/// The kafka connection config.
#[serde(flatten)]
pub connection: KafkaConnectionConfig,
/// TODO(weny): Remove the alias once we release v0.9.
/// The max size of a single producer batch.
#[serde(alias = "max_batch_size")]
@@ -44,7 +45,7 @@ pub struct DatanodeKafkaConfig {
impl Default for DatanodeKafkaConfig {
fn default() -> Self {
Self {
broker_endpoints: vec![BROKER_ENDPOINT.to_string()],
connection: KafkaConnectionConfig::default(),
// Warning: Kafka has a default limit of 1MB per message in a topic.
max_batch_bytes: ReadableSize::mb(1),
consumer_wait_timeout: Duration::from_millis(100),
src/common/wal/src/config/kafka/metasrv.rs
+5 -15
View File
@@ -14,15 +14,16 @@
use serde::{Deserialize, Serialize};
use super::common::KafkaConnectionConfig;
use crate::config::kafka::common::{backoff_prefix, BackoffConfig, KafkaTopicConfig};
use crate::BROKER_ENDPOINT;
/// Kafka wal configurations for metasrv.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
#[serde(default)]
pub struct MetasrvKafkaConfig {
/// The broker endpoints of the Kafka cluster.
pub broker_endpoints: Vec<String>,
/// The kafka connection config.
#[serde(flatten)]
pub connection: KafkaConnectionConfig,
/// The backoff config.
#[serde(flatten, with = "backoff_prefix")]
pub backoff: BackoffConfig,
@@ -30,14 +31,3 @@ pub struct MetasrvKafkaConfig {
#[serde(flatten)]
pub kafka_topic: KafkaTopicConfig,
}
impl Default for MetasrvKafkaConfig {
fn default() -> Self {
let broker_endpoints = vec![BROKER_ENDPOINT.to_string()];
Self {
broker_endpoints,
backoff: BackoffConfig::default(),
kafka_topic: KafkaTopicConfig::default(),
}
}
}
src/common/wal/src/error.rs
+66 -2
View File
@@ -13,7 +13,7 @@
// limitations under the License.
use common_macro::stack_trace_debug;
use snafu::Snafu;
use snafu::{Location, Snafu};
#[derive(Snafu)]
#[snafu(visibility(pub))]
@@ -24,10 +24,74 @@ pub enum Error {
broker_endpoint: String,
#[snafu(source)]
error: std::io::Error,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to find ipv4 endpoint: {:?}", broker_endpoint))]
EndpointIPV4NotFound { broker_endpoint: String },
EndpointIPV4NotFound {
broker_endpoint: String,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to read file, path: {}", path))]
ReadFile {
path: String,
#[snafu(source)]
error: std::io::Error,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to add root cert"))]
AddCert {
#[snafu(source)]
error: rustls::Error,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to read cert, path: {}", path))]
ReadCerts {
path: String,
#[snafu(source)]
error: std::io::Error,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to read key, path: {}", path))]
ReadKey {
path: String,
#[snafu(source)]
error: std::io::Error,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to parse key, path: {}", path))]
KeyNotFound {
path: String,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to set client auth cert"))]
SetClientAuthCert {
#[snafu(source)]
error: rustls::Error,
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Failed to load ca certs from system"))]
LoadSystemCerts {
#[snafu(source)]
error: std::io::Error,
#[snafu(implicit)]
location: Location,
},
}
pub type Result<T> = std::result::Result<T, Error>;
src/common/wal/src/lib.rs
+43
View File
@@ -61,6 +61,9 @@ async fn resolve_to_ipv4_one<T: AsRef<str>>(endpoint: T) -> Result<String> {
mod tests {
use std::assert_matches::assert_matches;
use common_telemetry::warn;
use rskafka::client::{Credentials, SaslConfig};
use super::*;
use crate::error::Error;
@@ -86,4 +89,44 @@ mod tests {
let got = resolve_to_ipv4_one(host).await;
assert_matches!(got.unwrap_err(), Error::ResolveEndpoint { .. });
}
#[tokio::test]
async fn test_sasl() {
common_telemetry::init_default_ut_logging();
let Ok(broker_endpoints) = std::env::var("GT_KAFKA_SASL_ENDPOINTS") else {
warn!("The endpoints is empty, skipping the test 'test_sasl'");
return;
};
let broker_endpoints = broker_endpoints
.split(',')
.map(|s| s.trim().to_string())
.collect::<Vec<_>>();
let username = "user_kafka";
let password = "secret";
let _ = rskafka::client::ClientBuilder::new(broker_endpoints.clone())
.sasl_config(SaslConfig::Plain(Credentials::new(
username.to_string(),
password.to_string(),
)))
.build()
.await
.unwrap();
let _ = rskafka::client::ClientBuilder::new(broker_endpoints.clone())
.sasl_config(SaslConfig::ScramSha256(Credentials::new(
username.to_string(),
password.to_string(),
)))
.build()
.await
.unwrap();
let _ = rskafka::client::ClientBuilder::new(broker_endpoints)
.sasl_config(SaslConfig::ScramSha512(Credentials::new(
username.to_string(),
password.to_string(),
)))
.build()
.await
.unwrap();
}
}
src/log-store/src/error.rs
+7
View File
@@ -27,6 +27,13 @@ use crate::kafka::producer::ProduceRequest;
#[snafu(visibility(pub))]
#[stack_trace_debug]
pub enum Error {
#[snafu(display("Failed to create TLS Config"))]
TlsConfig {
#[snafu(implicit)]
location: Location,
source: common_wal::error::Error,
},
#[snafu(display("Invalid provider type, expected: {}, actual: {}", expected, actual))]
InvalidProvider {
#[snafu(implicit)]
src/log-store/src/kafka/client_manager.rs
+18 -10
View File
@@ -25,7 +25,7 @@ use tokio::sync::{Mutex, RwLock};
use super::producer::OrderedBatchProducer;
use crate::error::{
BuildClientSnafu, BuildPartitionClientSnafu, ResolveKafkaEndpointSnafu, Result,
BuildClientSnafu, BuildPartitionClientSnafu, ResolveKafkaEndpointSnafu, Result, TlsConfigSnafu,
};
use crate::kafka::producer::OrderedBatchProducerRef;
@@ -80,16 +80,20 @@ impl ClientManager {
base: config.backoff.base as f64,
deadline: config.backoff.deadline,
};
let broker_endpoints = common_wal::resolve_to_ipv4(&config.broker_endpoints)
let broker_endpoints = common_wal::resolve_to_ipv4(&config.connection.broker_endpoints)
.await
.context(ResolveKafkaEndpointSnafu)?;
let client = ClientBuilder::new(broker_endpoints)
.backoff_config(backoff_config)
.build()
.await
.with_context(|_| BuildClientSnafu {
broker_endpoints: config.broker_endpoints.clone(),
})?;
let mut builder = ClientBuilder::new(broker_endpoints).backoff_config(backoff_config);
if let Some(sasl) = &config.connection.sasl {
builder = builder.sasl_config(sasl.config.clone().into_sasl_config());
};
if let Some(tls) = &config.connection.tls {
builder = builder.tls_config(tls.to_tls_config().await.context(TlsConfigSnafu)?)
};
let client = builder.build().await.with_context(|_| BuildClientSnafu {
broker_endpoints: config.connection.broker_endpoints.clone(),
})?;
Ok(Self {
client,
@@ -161,6 +165,7 @@ impl ClientManager {
#[cfg(test)]
mod tests {
use common_wal::config::kafka::common::KafkaConnectionConfig;
use common_wal::test_util::run_test_with_kafka_wal;
use tokio::sync::Barrier;
@@ -206,7 +211,10 @@ mod tests {
.await;
let config = DatanodeKafkaConfig {
broker_endpoints,
connection: KafkaConnectionConfig {
broker_endpoints,
..Default::default()
},
..Default::default()
};
let manager = ClientManager::try_new(&config).await.unwrap();
src/log-store/src/kafka/log_store.rs
+9 -2
View File
@@ -360,6 +360,7 @@ mod tests {
use common_base::readable_size::ReadableSize;
use common_telemetry::info;
use common_telemetry::tracing::warn;
use common_wal::config::kafka::common::KafkaConnectionConfig;
use common_wal::config::kafka::DatanodeKafkaConfig;
use futures::TryStreamExt;
use rand::prelude::SliceRandom;
@@ -461,7 +462,10 @@ mod tests {
.map(|s| s.trim().to_string())
.collect::<Vec<_>>();
let config = DatanodeKafkaConfig {
broker_endpoints,
connection: KafkaConnectionConfig {
broker_endpoints,
..Default::default()
},
max_batch_bytes: ReadableSize::kb(32),
..Default::default()
};
@@ -530,7 +534,10 @@ mod tests {
.map(|s| s.trim().to_string())
.collect::<Vec<_>>();
let config = DatanodeKafkaConfig {
broker_endpoints,
connection: KafkaConnectionConfig {
broker_endpoints,
..Default::default()
},
max_batch_bytes: ReadableSize::kb(8),
..Default::default()
};
src/log-store/src/test_util/log_store_util.rs
+5 -1
View File
@@ -15,6 +15,7 @@
use std::path::Path;
use common_base::readable_size::ReadableSize;
use common_wal::config::kafka::common::KafkaConnectionConfig;
use common_wal::config::kafka::DatanodeKafkaConfig;
use common_wal::config::raft_engine::RaftEngineConfig;
@@ -34,7 +35,10 @@ pub async fn create_tmp_local_file_log_store<P: AsRef<Path>>(path: P) -> RaftEng
/// Create a [KafkaLogStore].
pub async fn create_kafka_log_store(broker_endpoints: Vec<String>) -> KafkaLogStore {
KafkaLogStore::try_new(&DatanodeKafkaConfig {
broker_endpoints,
connection: KafkaConnectionConfig {
broker_endpoints,
..Default::default()
},
..Default::default()
})
.await
tests-integration/fixtures/kafka/docker-compose-standalone.yml
+12 -5
View File
@@ -1,21 +1,28 @@
version: '3.8'
services:
zookeeper:
image: bitnami/zookeeper:3.7
ports:
- '2181:2181'
environment:
- ALLOW_ANONYMOUS_LOGIN=yes
kafka:
image: bitnami/kafka:3.6.0
container_name: kafka
ports:
- 9092:9092
- 9093:9093
environment:
# KRaft settings
KAFKA_KRAFT_CLUSTER_ID: Kmp-xkTnSf-WWXhWmiorDg
KAFKA_ENABLE_KRAFT: "yes"
KAFKA_CFG_NODE_ID: "1"
KAFKA_CFG_PROCESS_ROLES: broker,controller
KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@127.0.0.1:2181
# Listeners
KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://127.0.0.1:9092
KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://127.0.0.1:9092,SECURE://localhost:9093
KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:2181
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SECURE:SASL_PLAINTEXT
KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:2181,SECURE://:9093
ALLOW_PLAINTEXT_LISTENER: "yes"
KAFKA_BROKER_ID: "1"
KAFKA_CLIENT_USERS: "user_kafka"
KAFKA_CLIENT_PASSWORDS: "secret"
tests-integration/src/tests/test_util.rs
+17 -5
View File
@@ -21,7 +21,7 @@ use common_query::Output;
use common_recordbatch::util;
use common_telemetry::warn;
use common_test_util::find_workspace_path;
use common_wal::config::kafka::common::KafkaTopicConfig;
use common_wal::config::kafka::common::{KafkaConnectionConfig, KafkaTopicConfig};
use common_wal::config::kafka::{DatanodeKafkaConfig, MetasrvKafkaConfig};
use common_wal::config::{DatanodeWalConfig, MetasrvWalConfig};
use frontend::instance::Instance;
@@ -227,11 +227,17 @@ pub(crate) async fn standalone_with_kafka_wal() -> Option<Box<dyn RebuildableMoc
let test_name = uuid::Uuid::new_v4().to_string();
let builder = GreptimeDbStandaloneBuilder::new(&test_name)
.with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig {
broker_endpoints: endpoints.clone(),
connection: KafkaConnectionConfig {
broker_endpoints: endpoints.clone(),
..Default::default()
},
..Default::default()
}))
.with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
broker_endpoints: endpoints,
connection: KafkaConnectionConfig {
broker_endpoints: endpoints,
..Default::default()
},
kafka_topic: KafkaTopicConfig {
topic_name_prefix: test_name.to_string(),
num_topics: 3,
@@ -260,11 +266,17 @@ pub(crate) async fn distributed_with_kafka_wal() -> Option<Box<dyn RebuildableMo
let builder = GreptimeDbClusterBuilder::new(&test_name)
.await
.with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig {
broker_endpoints: endpoints.clone(),
connection: KafkaConnectionConfig {
broker_endpoints: endpoints.clone(),
..Default::default()
},
..Default::default()
}))
.with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
broker_endpoints: endpoints,
connection: KafkaConnectionConfig {
broker_endpoints: endpoints,
..Default::default()
},
kafka_topic: KafkaTopicConfig {
topic_name_prefix: test_name.to_string(),
num_topics: 3,
tests-integration/tests/region_migration.rs
+57 -15
View File
@@ -23,7 +23,7 @@ use common_recordbatch::RecordBatches;
use common_telemetry::info;
use common_test_util::recordbatch::check_output_stream;
use common_test_util::temp_dir::create_temp_dir;
use common_wal::config::kafka::common::KafkaTopicConfig;
use common_wal::config::kafka::common::{KafkaConnectionConfig, KafkaTopicConfig};
use common_wal::config::kafka::{DatanodeKafkaConfig, MetasrvKafkaConfig};
use common_wal::config::{DatanodeWalConfig, MetasrvWalConfig};
use datatypes::prelude::ScalarVector;
@@ -114,11 +114,17 @@ pub async fn test_region_migration(store_type: StorageType, endpoints: Vec<Strin
.with_datanodes(datanodes as u32)
.with_store_config(store_config)
.with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig {
broker_endpoints: endpoints.clone(),
connection: KafkaConnectionConfig {
broker_endpoints: endpoints.clone(),
..Default::default()
},
..Default::default()
}))
.with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
broker_endpoints: endpoints,
connection: KafkaConnectionConfig {
broker_endpoints: endpoints,
..Default::default()
},
kafka_topic: KafkaTopicConfig {
num_topics: 3,
topic_name_prefix: Uuid::new_v4().to_string(),
@@ -246,11 +252,17 @@ pub async fn test_metric_table_region_migration_by_sql(
.with_datanodes(datanodes as u32)
.with_store_config(store_config)
.with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig {
broker_endpoints: endpoints.clone(),
connection: KafkaConnectionConfig {
broker_endpoints: endpoints.clone(),
..Default::default()
},
..Default::default()
}))
.with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
broker_endpoints: endpoints,
connection: KafkaConnectionConfig {
broker_endpoints: endpoints,
..Default::default()
},
kafka_topic: KafkaTopicConfig {
num_topics: 3,
topic_name_prefix: Uuid::new_v4().to_string(),
@@ -371,11 +383,17 @@ pub async fn test_region_migration_by_sql(store_type: StorageType, endpoints: Ve
.with_datanodes(datanodes as u32)
.with_store_config(store_config)
.with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig {
broker_endpoints: endpoints.clone(),
connection: KafkaConnectionConfig {
broker_endpoints: endpoints.clone(),
..Default::default()
},
..Default::default()
}))
.with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
broker_endpoints: endpoints,
connection: KafkaConnectionConfig {
broker_endpoints: endpoints,
..Default::default()
},
kafka_topic: KafkaTopicConfig {
num_topics: 3,
topic_name_prefix: Uuid::new_v4().to_string(),
@@ -495,11 +513,17 @@ pub async fn test_region_migration_multiple_regions(
.with_datanodes(datanodes as u32)
.with_store_config(store_config)
.with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig {
broker_endpoints: endpoints.clone(),
connection: KafkaConnectionConfig {
broker_endpoints: endpoints.clone(),
..Default::default()
},
..Default::default()
}))
.with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
broker_endpoints: endpoints,
connection: KafkaConnectionConfig {
broker_endpoints: endpoints,
..Default::default()
},
kafka_topic: KafkaTopicConfig {
num_topics: 3,
topic_name_prefix: Uuid::new_v4().to_string(),
@@ -634,11 +658,17 @@ pub async fn test_region_migration_all_regions(store_type: StorageType, endpoint
.with_datanodes(datanodes as u32)
.with_store_config(store_config)
.with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig {
broker_endpoints: endpoints.clone(),
connection: KafkaConnectionConfig {
broker_endpoints: endpoints.clone(),
..Default::default()
},
..Default::default()
}))
.with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
broker_endpoints: endpoints,
connection: KafkaConnectionConfig {
broker_endpoints: endpoints,
..Default::default()
},
kafka_topic: KafkaTopicConfig {
num_topics: 3,
topic_name_prefix: Uuid::new_v4().to_string(),
@@ -768,11 +798,17 @@ pub async fn test_region_migration_incorrect_from_peer(
.with_datanodes(datanodes as u32)
.with_store_config(store_config)
.with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig {
broker_endpoints: endpoints.clone(),
connection: KafkaConnectionConfig {
broker_endpoints: endpoints.clone(),
..Default::default()
},
..Default::default()
}))
.with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
broker_endpoints: endpoints,
connection: KafkaConnectionConfig {
broker_endpoints: endpoints,
..Default::default()
},
kafka_topic: KafkaTopicConfig {
num_topics: 3,
topic_name_prefix: Uuid::new_v4().to_string(),
@@ -845,11 +881,17 @@ pub async fn test_region_migration_incorrect_region_id(
.with_datanodes(datanodes as u32)
.with_store_config(store_config)
.with_datanode_wal_config(DatanodeWalConfig::Kafka(DatanodeKafkaConfig {
broker_endpoints: endpoints.clone(),
connection: KafkaConnectionConfig {
broker_endpoints: endpoints.clone(),
..Default::default()
},
..Default::default()
}))
.with_metasrv_wal_config(MetasrvWalConfig::Kafka(MetasrvKafkaConfig {
broker_endpoints: endpoints,
connection: KafkaConnectionConfig {
broker_endpoints: endpoints,
..Default::default()
},
kafka_topic: KafkaTopicConfig {
num_topics: 3,
topic_name_prefix: Uuid::new_v4().to_string(),