Skip to content

Commit 4d32b6f

Browse files
razvanrazvan
committed
client_protocol: refactor s3 config into crd::s3 and config::s3 to make it reusable
1 parent 0ba2fed commit 4d32b6f

8 files changed

Lines changed: 335 additions & 256 deletions

File tree

‎rust/operator-binary/src/command.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,14 @@ use stackable_operator::{
99
use crate::{
1010
authentication::TrinoAuthenticationConfig,
1111
catalog::config::CatalogConfig,
12+
config::client_protocol,
1213
controller::{STACKABLE_LOG_CONFIG_DIR, STACKABLE_LOG_DIR},
1314
crd::{
1415
CONFIG_DIR_NAME, Container, EXCHANGE_MANAGER_PROPERTIES, LOG_PROPERTIES,
1516
RW_CONFIG_DIR_NAME, SPOOLING_MANAGER_PROPERTIES, STACKABLE_CLIENT_TLS_DIR,
1617
STACKABLE_INTERNAL_TLS_DIR, STACKABLE_MOUNT_INTERNAL_TLS_DIR,
1718
STACKABLE_MOUNT_SERVER_TLS_DIR, STACKABLE_SERVER_TLS_DIR, STACKABLE_TLS_STORE_PASSWORD,
18-
SYSTEM_TRUST_STORE, SYSTEM_TRUST_STORE_PASSWORD, TrinoRole, client_protocol,
19+
SYSTEM_TRUST_STORE, SYSTEM_TRUST_STORE_PASSWORD, TrinoRole,
1920
fault_tolerant_execution::ResolvedFaultTolerantExecutionConfig, v1alpha1,
2021
},
2122
};

‎rust/operator-binary/src/config/client_protocol.rs‎

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
// Consolidate Trino S3 properties in a single reusable struct.
2+
3+
use std::collections::BTreeMap;
4+
5+
use snafu::{self, ResultExt, Snafu};
6+
use stackable_operator::{
7+
client::Client,
8+
k8s_openapi::api::core::v1::{Volume, VolumeMount},
9+
};
10+
11+
use crate::{
12+
config,
13+
crd::{
14+
ENV_SPOOLING_SECRET,
15+
client_protocol::{ClientProtocolConfig, SpoolingFileSystemConfig},
16+
},
17+
};
18+
19+
#[derive(Snafu, Debug)]
20+
pub enum Error {
21+
#[snafu(display("Failed to resolve S3 connection"))]
22+
ResolveS3Connection { source: config::s3::Error },
23+
24+
#[snafu(display("trino does not support disabling the TLS verification of S3 servers"))]
25+
S3TlsNoVerificationNotSupported,
26+
27+
#[snafu(display("failed to convert data size for [{field}] to bytes"))]
28+
QuantityConversion {
29+
source: stackable_operator::memory::Error,
30+
field: &'static str,
31+
},
32+
}
33+
34+
pub struct ResolvedClientProtocolConfig {
35+
/// Properties to add to config.properties
36+
pub config_properties: BTreeMap<String, String>,
37+
38+
// Properties for spooling-manager.properties
39+
pub spooling_manager_properties: BTreeMap<String, String>,
40+
41+
/// Volumes required for the configuration (e.g., for S3 credentials)
42+
pub volumes: Vec<Volume>,
43+
44+
/// Volume mounts required for the configuration
45+
pub volume_mounts: Vec<VolumeMount>,
46+
47+
/// Additional commands that need to be executed before starting Trino
48+
/// Used to add TLS certificates to the client's trust store.
49+
pub init_container_extra_start_commands: Vec<String>,
50+
}
51+
52+
impl ResolvedClientProtocolConfig {
53+
/// Resolve S3 connection properties from Kubernetes resources
54+
/// and prepare spooling filesystem configuration.
55+
pub async fn from_config(
56+
config: &ClientProtocolConfig,
57+
client: Option<&Client>,
58+
namespace: &str,
59+
) -> Result<Self, Error> {
60+
let mut resolved_config = Self {
61+
config_properties: BTreeMap::new(),
62+
spooling_manager_properties: BTreeMap::new(),
63+
volumes: Vec::new(),
64+
volume_mounts: Vec::new(),
65+
init_container_extra_start_commands: Vec::new(),
66+
};
67+
68+
match config {
69+
ClientProtocolConfig::Spooling(spooling_config) => {
70+
// Resolve external resources if Kubernetes client is available
71+
// This should always be the case, except for when this function is called during unit tests
72+
if let Some(client) = client {
73+
match &spooling_config.filesystem {
74+
SpoolingFileSystemConfig::S3(s3_config) => {
75+
let resolved_s3_config = config::s3::ResolvedS3Config::from_config(
76+
s3_config, client, namespace,
77+
)
78+
.await
79+
.context(ResolveS3ConnectionSnafu)?;
80+
81+
// Enable S3 filesystem after successful resolution
82+
resolved_config
83+
.spooling_manager_properties
84+
.insert("fs.s3.enabled".to_string(), "true".to_string());
85+
86+
// Copy the S3 configuration over
87+
resolved_config
88+
.spooling_manager_properties
89+
.extend(resolved_s3_config.properties);
90+
resolved_config.volumes.extend(resolved_s3_config.volumes);
91+
resolved_config
92+
.volume_mounts
93+
.extend(resolved_s3_config.volume_mounts);
94+
resolved_config
95+
.init_container_extra_start_commands
96+
.extend(resolved_s3_config.init_container_extra_start_commands);
97+
}
98+
}
99+
}
100+
101+
resolved_config.spooling_manager_properties.extend([
102+
("fs.location".to_string(), spooling_config.location.clone()),
103+
(
104+
"spooling-manager.name".to_string(),
105+
"filesystem".to_string(),
106+
),
107+
]);
108+
109+
// Enable spooling protocol
110+
resolved_config.config_properties.extend([
111+
("protocol.spooling.enabled".to_string(), "true".to_string()),
112+
(
113+
"protocol.spooling.shared-secret-key".to_string(),
114+
format!("${{ENV:{secret}}}", secret = ENV_SPOOLING_SECRET),
115+
),
116+
]);
117+
}
118+
}
119+
120+
Ok(resolved_config)
121+
}
122+
}
123+
124+
#[cfg(test)]
125+
mod tests {
126+
use super::*;
127+
use crate::crd::{client_protocol::ClientSpoolingProtocolConfig, s3::S3Config};
128+
129+
#[tokio::test]
130+
async fn test_spooling_config() {
131+
let config = ClientProtocolConfig::Spooling(ClientSpoolingProtocolConfig {
132+
location: "s3://my-bucket/spooling".to_string(),
133+
filesystem: SpoolingFileSystemConfig::S3(S3Config {
134+
connection:
135+
stackable_operator::crd::s3::v1alpha1::InlineConnectionOrReference::Reference(
136+
"test-s3-connection".to_string(),
137+
),
138+
iam_role: None,
139+
external_id: None,
140+
max_error_retries: None,
141+
upload_part_size: None,
142+
}),
143+
});
144+
145+
let resolved_spooling_config = ResolvedClientProtocolConfig::from_config(
146+
&config, None, // No client, so no external resolution
147+
"default",
148+
)
149+
.await
150+
.unwrap();
151+
152+
let expected_props = BTreeMap::from([
153+
(
154+
"fs.location".to_string(),
155+
"s3://my-bucket/spooling".to_string(),
156+
),
157+
(
158+
"spooling-manager.name".to_string(),
159+
"filesystem".to_string(),
160+
),
161+
]);
162+
assert_eq!(
163+
expected_props,
164+
resolved_spooling_config.spooling_manager_properties
165+
);
166+
}
167+
}

‎rust/operator-binary/src/config/mod.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1 +1,3 @@
1+
pub mod client_protocol;
12
pub mod jvm;
3+
pub mod s3;

‎rust/operator-binary/src/config/s3.rs‎

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
use std::collections::BTreeMap;
2+
3+
use snafu::{ResultExt, Snafu};
4+
use stackable_operator::{
5+
client::Client,
6+
commons::tls_verification::{CaCert, TlsServerVerification, TlsVerification},
7+
crd::s3,
8+
k8s_openapi::api::core::v1::{Volume, VolumeMount},
9+
};
10+
11+
use crate::{
12+
command,
13+
crd::{STACKABLE_CLIENT_TLS_DIR, s3 as trino_s3},
14+
};
15+
16+
#[derive(Snafu, Debug)]
17+
pub enum Error {
18+
#[snafu(display("Failed to resolve S3 connection"))]
19+
S3Connection {
20+
source: s3::v1alpha1::ConnectionError,
21+
},
22+
23+
#[snafu(display("trino does not support disabling the TLS verification of S3 servers"))]
24+
S3TlsNoVerificationNotSupported,
25+
26+
#[snafu(display("failed to convert data size for [{field}] to bytes"))]
27+
QuantityConversion {
28+
source: stackable_operator::memory::Error,
29+
field: &'static str,
30+
},
31+
}
32+
33+
pub struct ResolvedS3Config {
34+
/// Properties to add to config.properties
35+
pub properties: BTreeMap<String, String>,
36+
37+
/// Volumes required for the configuration (e.g., for S3 credentials)
38+
pub volumes: Vec<Volume>,
39+
40+
/// Volume mounts required for the configuration
41+
pub volume_mounts: Vec<VolumeMount>,
42+
43+
/// Additional commands that need to be executed before starting Trino
44+
/// Used to add TLS certificates to the client's trust store.
45+
pub init_container_extra_start_commands: Vec<String>,
46+
}
47+
48+
impl ResolvedS3Config {
49+
/// Resolve S3 connection properties from Kubernetes resources
50+
/// and prepare spooling filesystem configuration.
51+
pub async fn from_config(
52+
config: &trino_s3::S3Config,
53+
client: &Client,
54+
namespace: &str,
55+
) -> Result<Self, Error> {
56+
let mut resolved_config = Self {
57+
properties: BTreeMap::new(),
58+
volumes: Vec::new(),
59+
volume_mounts: Vec::new(),
60+
init_container_extra_start_commands: Vec::new(),
61+
};
62+
63+
let s3_connection = config
64+
.connection
65+
.clone()
66+
.resolve(client, namespace)
67+
.await
68+
.context(S3ConnectionSnafu)?;
69+
70+
let (volumes, mounts) = s3_connection
71+
.volumes_and_mounts()
72+
.context(S3ConnectionSnafu)?;
73+
resolved_config.volumes.extend(volumes);
74+
resolved_config.volume_mounts.extend(mounts);
75+
76+
resolved_config
77+
.properties
78+
.insert("s3.region".to_string(), s3_connection.region.name.clone());
79+
resolved_config.properties.insert(
80+
"s3.endpoint".to_string(),
81+
s3_connection
82+
.endpoint()
83+
.context(S3ConnectionSnafu)?
84+
.to_string(),
85+
);
86+
resolved_config.properties.insert(
87+
"s3.path-style-access".to_string(),
88+
(s3_connection.access_style == s3::v1alpha1::S3AccessStyle::Path).to_string(),
89+
);
90+
91+
if let Some((access_key_path, secret_key_path)) = s3_connection.credentials_mount_paths() {
92+
resolved_config.properties.extend([
93+
(
94+
"s3.aws-access-key".to_string(),
95+
format!("${{file:UTF-8:{access_key_path}}}"),
96+
),
97+
(
98+
"s3.aws-secret-key".to_string(),
99+
format!("${{file:UTF-8:{secret_key_path}}}"),
100+
),
101+
]);
102+
}
103+
104+
if let Some(tls) = s3_connection.tls.tls.as_ref() {
105+
match &tls.verification {
106+
TlsVerification::None {} => return S3TlsNoVerificationNotSupportedSnafu.fail(),
107+
TlsVerification::Server(TlsServerVerification {
108+
ca_cert: CaCert::WebPki {},
109+
}) => {}
110+
TlsVerification::Server(TlsServerVerification {
111+
ca_cert: CaCert::SecretClass(_),
112+
}) => {
113+
if let Some(ca_cert) = s3_connection.tls.tls_ca_cert_mount_path() {
114+
resolved_config.init_container_extra_start_commands.extend(
115+
command::add_cert_to_truststore(
116+
&ca_cert,
117+
STACKABLE_CLIENT_TLS_DIR,
118+
"resolved-s3-ca-cert",
119+
),
120+
);
121+
}
122+
}
123+
}
124+
}
125+
126+
Ok(resolved_config)
127+
}
128+
}

‎rust/operator-binary/src/controller.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ use crate::{
7979
authorization::opa::TrinoOpaConfig,
8080
catalog::{FromTrinoCatalogError, config::CatalogConfig},
8181
command, config,
82+
config::client_protocol,
8283
crd::{
8384
ACCESS_CONTROL_PROPERTIES, APP_NAME, CONFIG_DIR_NAME, CONFIG_PROPERTIES, Container,
8485
DISCOVERY_URI, ENV_INTERNAL_SECRET, ENV_SPOOLING_SECRET, EXCHANGE_MANAGER_PROPERTIES,
@@ -88,7 +89,7 @@ use crate::{
8889
STACKABLE_CLIENT_TLS_DIR, STACKABLE_INTERNAL_TLS_DIR, STACKABLE_MOUNT_INTERNAL_TLS_DIR,
8990
STACKABLE_MOUNT_SERVER_TLS_DIR, STACKABLE_SERVER_TLS_DIR, TrinoRole,
9091
authentication::resolve_authentication_classes,
91-
catalog, client_protocol,
92+
catalog,
9293
discovery::{TrinoDiscovery, TrinoDiscoveryProtocol, TrinoPodRef},
9394
fault_tolerant_execution::ResolvedFaultTolerantExecutionConfig,
9495
rolegroup_headless_service_name, v1alpha1,
@@ -1810,7 +1811,7 @@ mod tests {
18101811
use stackable_operator::commons::networking::DomainName;
18111812

18121813
use super::*;
1813-
use crate::crd::client_protocol::ResolvedClientProtocolConfig;
1814+
use crate::config::client_protocol::ResolvedClientProtocolConfig;
18141815

18151816
#[test]
18161817
fn test_config_overrides() {

0 commit comments

Comments
 (0)