feat(catalog): add support for redacting sensitive connector properties

Cargo.toml

@@ -51,6 +51,8 @@ members = [
   "src/tests/state_cleaning_test",
   "src/utils/local_stats_alloc",
   "src/utils/pgwire",
+  "src/utils/redaction/redaction",
+  "src/utils/redaction/redaction_derive",
   "src/utils/runtime",
   "src/utils/sync-point",
   "src/utils/variables",

src/connector/Cargo.toml

@@ -70,6 +70,8 @@ itertools = "0.12"
 jni = { version = "0.21.1", features = ["invocation"] }
 jst = { package = 'jsonschema-transpiler', git = "https://github.com/mozilla/jsonschema-transpiler", rev = "c1a89d720d118843d8bcca51084deb0ed223e4b4" }
 maplit = "1.0.2"
+mark_redaction = { path = "../utils/redaction/redaction" }
+mark_redaction_derive = { path = "../utils/redaction/redaction_derive" }
 moka = { version = "0.12", features = ["future"] }
 mysql_async = { version = "0.33", default-features = false, features = [
     "default",

src/connector/src/source/cdc/mod.rs

@@ -20,6 +20,7 @@ use std::marker::PhantomData;
 
 pub use enumerator::*;
 use itertools::Itertools;
+use mark_redaction_derive::MarkRedaction;
 use risingwave_common::catalog::{ColumnDesc, Field, Schema};
 use risingwave_pb::catalog::PbSource;
 use risingwave_pb::connector_service::{PbSourceType, PbTableSchema, SourceType, TableSchema};
@@ -68,7 +69,7 @@ impl CdcSourceType {
     }
 }
 
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug, Default, MarkRedaction)]
 pub struct CdcProperties<T: CdcSourceTypeTrait> {
     /// Properties specified in the WITH clause by user
     pub properties: HashMap<String, String>,

src/connector/src/source/datagen/mod.rs

@@ -19,6 +19,7 @@ pub mod split;
 use std::collections::HashMap;
 
 pub use enumerator::*;
+use mark_redaction_derive::MarkRedaction;
 use serde::Deserialize;
 use serde_with::{serde_as, DisplayFromStr};
 pub use source::*;
@@ -29,7 +30,7 @@ use crate::source::SourceProperties;
 pub const DATAGEN_CONNECTOR: &str = "datagen";
 
 #[serde_as]
-#[derive(Clone, Debug, Deserialize)]
+#[derive(Clone, Debug, Deserialize, MarkRedaction)]
 pub struct DatagenProperties {
     /// split_num means data source partition
     #[serde(rename = "datagen.split.num")]

src/connector/src/source/filesystem/opendal_source/mod.rs

@@ -21,6 +21,8 @@ use serde::Deserialize;
 pub mod opendal_enumerator;
 pub mod opendal_reader;
 
+use mark_redaction_derive::MarkRedaction;
+
 use self::opendal_enumerator::OpendalEnumerator;
 use self::opendal_reader::OpendalReader;
 use super::{OpendalFsSplit, S3Properties};
@@ -30,7 +32,7 @@ pub const GCS_CONNECTOR: &str = "gcs";
 // The new s3_v2 will use opendal.
 pub const OPENDAL_S3_CONNECTOR: &str = "s3_v2";
 
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, MarkRedaction)]
 pub struct GcsProperties {
     #[serde(rename = "gcs.bucket_name")]
     pub bucket_name: String,
@@ -78,7 +80,7 @@ impl OpendalSource for OpendalGcs {
     }
 }
 
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, MarkRedaction)]
 pub struct OpendalS3Properties {
     pub s3_properties: S3Properties,
     #[serde(rename = "s3.assume_role", default)]

src/connector/src/source/filesystem/s3/mod.rs

@@ -15,6 +15,7 @@ pub mod enumerator;
 
 pub use enumerator::S3SplitEnumerator;
 mod source;
+use mark_redaction_derive::MarkRedaction;
 use serde::Deserialize;
 pub use source::S3FileReader;
 
@@ -24,7 +25,7 @@ use crate::source::SourceProperties;
 
 pub const S3_CONNECTOR: &str = "s3";
 
-#[derive(Clone, Debug, Deserialize, PartialEq)]
+#[derive(Clone, Debug, Deserialize, PartialEq, MarkRedaction)]
 pub struct S3Properties {
     #[serde(rename = "s3.region_name")]
     pub region_name: String,
@@ -33,8 +34,10 @@ pub struct S3Properties {
     #[serde(rename = "match_pattern", default)]
     pub match_pattern: Option<String>,
     #[serde(rename = "s3.credentials.access", default)]
+    #[mark_redaction(rename = "s3.credentials.access")]
     pub access: Option<String>,
     #[serde(rename = "s3.credentials.secret", default)]
+    #[mark_redaction(rename = "s3.credentials.secret")]
     pub secret: Option<String>,
     #[serde(rename = "s3.endpoint_url")]
     pub endpoint_url: Option<String>,

src/connector/src/source/google_pubsub/mod.rs

@@ -19,6 +19,7 @@ pub mod source;
 pub mod split;
 
 pub use enumerator::*;
+use mark_redaction_derive::MarkRedaction;
 use serde_with::{serde_as, DisplayFromStr};
 pub use source::*;
 pub use split::*;
@@ -29,7 +30,7 @@ use crate::source::SourceProperties;
 pub const GOOGLE_PUBSUB_CONNECTOR: &str = "google_pubsub";
 
 #[serde_as]
-#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Hash, WithOptions)]
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Hash, WithOptions, MarkRedaction)]
 pub struct PubsubProperties {
     #[serde_as(as = "DisplayFromStr")]
     #[serde(rename = "pubsub.split_count")]

src/connector/src/source/kafka/mod.rs

@@ -22,6 +22,7 @@ pub mod split;
 pub mod stats;
 
 pub use enumerator::*;
+use mark_redaction_derive::MarkRedaction;
 pub use private_link::*;
 pub use source::*;
 pub use split::*;
@@ -87,7 +88,7 @@ pub struct RdKafkaPropertiesConsumer {
     pub enable_auto_commit: Option<bool>,
 }
 
-#[derive(Clone, Debug, Deserialize, WithOptions)]
+#[derive(Clone, Debug, Deserialize, WithOptions, MarkRedaction)]
 pub struct KafkaProperties {
     /// This parameter is not intended to be exposed to users.
     /// This parameter specifies only for one parallelism. The parallelism of kafka source

src/connector/src/source/kinesis/mod.rs

@@ -16,6 +16,7 @@ pub mod enumerator;
 pub mod source;
 pub mod split;
 
+use mark_redaction_derive::MarkRedaction;
 use serde::Deserialize;
 use serde_with::{serde_as, DisplayFromStr};
 use with_options::WithOptions;
@@ -29,7 +30,7 @@ use crate::source::SourceProperties;
 pub const KINESIS_CONNECTOR: &str = "kinesis";
 
 #[serde_as]
-#[derive(Clone, Debug, Deserialize, WithOptions)]
+#[derive(Clone, Debug, Deserialize, WithOptions, MarkRedaction)]
 pub struct KinesisProperties {
     #[serde(rename = "scan.startup.mode", alias = "kinesis.scan.startup.mode")]
     // accepted values: "latest", "earliest", "timestamp"

src/connector/src/source/nats/mod.rs

@@ -16,6 +16,7 @@ pub mod enumerator;
 pub mod source;
 pub mod split;
 
+use mark_redaction_derive::MarkRedaction;
 use serde::Deserialize;
 use with_options::WithOptions;
 
@@ -26,7 +27,7 @@ use crate::source::SourceProperties;
 
 pub const NATS_CONNECTOR: &str = "nats";
 
-#[derive(Clone, Debug, Deserialize, WithOptions)]
+#[derive(Clone, Debug, Deserialize, WithOptions, MarkRedaction)]
 pub struct NatsProperties {
     #[serde(flatten)]
     pub common: NatsCommon,

src/connector/src/source/nexmark/mod.rs

@@ -16,9 +16,11 @@ pub mod enumerator;
 pub mod source;
 pub mod split;
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 pub use enumerator::*;
+use mark_redaction::MarkRedaction;
+use mark_redaction_derive::MarkRedaction;
 use nexmark::config::{NexmarkConfig, RateShape};
 use nexmark::event::EventType;
 use serde::Deserialize;
@@ -45,8 +47,14 @@ const fn none<T>() -> Option<T> {
 
 pub type NexmarkProperties = Box<NexmarkPropertiesInner>;
 
+impl MarkRedaction for NexmarkProperties {
+    fn marks() -> HashSet<String> {
+        NexmarkPropertiesInner::marks()
+    }
+}
+
 #[serde_as]
-#[derive(Clone, Debug, Deserialize, WithOptions)]
+#[derive(Clone, Debug, Deserialize, WithOptions, MarkRedaction)]
 pub struct NexmarkPropertiesInner {
     #[serde_as(as = "DisplayFromStr")]
     #[serde(rename = "nexmark.split.num", default = "identity_i32::<1>")]

src/connector/src/source/pulsar/mod.rs

@@ -18,6 +18,7 @@ pub mod split;
 pub mod topic;
 
 pub use enumerator::*;
+use mark_redaction_derive::MarkRedaction;
 use serde::Deserialize;
 use serde_with::serde_as;
 pub use split::*;
@@ -37,7 +38,7 @@ impl SourceProperties for PulsarProperties {
     const SOURCE_NAME: &'static str = PULSAR_CONNECTOR;
 }
 
-#[derive(Clone, Debug, Deserialize, WithOptions)]
+#[derive(Clone, Debug, Deserialize, WithOptions, MarkRedaction)]
 #[serde_as]
 pub struct PulsarProperties {
     #[serde(rename = "scan.startup.mode", alias = "pulsar.scan.startup.mode")]

src/connector/src/source/test_source.rs

@@ -17,6 +17,7 @@ use std::sync::{Arc, OnceLock};
 
 use anyhow::anyhow;
 use async_trait::async_trait;
+use mark_redaction_derive::MarkRedaction;
 use parking_lot::Mutex;
 use risingwave_common::types::JsonbVal;
 use serde_derive::{Deserialize, Serialize};
@@ -114,7 +115,7 @@ pub fn registry_test_source(box_source: BoxSource) -> TestSourceRegistryGuard {
 
 pub const TEST_CONNECTOR: &str = "test";
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, MarkRedaction)]
 pub struct TestSourceProperties {
     properties: HashMap<String, String>,
 }

src/frontend/Cargo.toml

@@ -38,6 +38,8 @@ futures-async-stream = { workspace = true }
 iana-time-zone = "0.1"
 itertools = "0.12"
 maplit = "1"
+mark_redaction = { path = "../utils/redaction/redaction" }
+mark_redaction_derive = { path = "../utils/redaction/redaction_derive" }
 md5 = "0.7.0"
 num-integer = "0.1"
 parking_lot = "0.12"

src/frontend/src/utils/mod.rs

@@ -32,6 +32,8 @@ pub use index_set::*;
 pub(crate) mod group_by;
 pub mod infer_stmt_row_desc;
 pub mod overwrite_options;
+pub mod redaction;
+
 pub use group_by::*;
 pub use overwrite_options::*;