Skip to content

Commit b5cb26c

Browse files
maltesandermaltesander
committed
refactor: move raw cluster methods to ValidatedCluster
1 parent 910ba94 commit b5cb26c

7 files changed

Lines changed: 197 additions & 182 deletions

File tree

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

Lines changed: 123 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ use stackable_operator::{
1212

1313
use crate::{
1414
controller::{ValidatedCluster, ValidatedTrinoConfig},
15-
crd::{DEFAULT_WORKER_GRACEFUL_SHUTDOWN_TIMEOUT, TrinoRole, v1alpha1},
15+
crd::{DEFAULT_WORKER_GRACEFUL_SHUTDOWN_TIMEOUT, TrinoRole},
1616
};
1717

1818
/// Corresponds to "shutdown.grace-period", which defaults to 2 min.
@@ -62,10 +62,8 @@ pub fn graceful_shutdown_config_properties(
6262
}
6363
}
6464

65-
/// Returns the minimal `gracefulShutdownTimeout` across all worker role-groups.
66-
///
67-
/// Mirrors [`v1alpha1::TrinoCluster::min_worker_graceful_shutdown_timeout`] but
68-
/// reads from [`ValidatedCluster::role_group_configs`].
65+
/// Returns the minimal `gracefulShutdownTimeout` across all worker role-groups, read from the
66+
/// validated [`ValidatedCluster::role_group_configs`].
6967
fn min_worker_graceful_shutdown_timeout(
7068
cluster: &ValidatedCluster,
7169
) -> stackable_operator::shared::time::Duration {
@@ -80,7 +78,7 @@ fn min_worker_graceful_shutdown_timeout(
8078
}
8179

8280
pub fn add_graceful_shutdown_config(
83-
trino: &v1alpha1::TrinoCluster,
81+
cluster: &ValidatedCluster,
8482
role: &TrinoRole,
8583
merged_config: &ValidatedTrinoConfig,
8684
pod_builder: &mut PodBuilder,
@@ -127,9 +125,9 @@ pub fn add_graceful_shutdown_config(
127125
echo 'Successfully sent graceful shutdown command' >> /proc/1/fd/1 2>&1
128126
echo 'Sleeping {termination_grace_period_seconds} seconds' >> /proc/1/fd/1 2>&1
129127
sleep {termination_grace_period_seconds}",
130-
protocol = trino.exposed_protocol(),
128+
protocol = cluster.exposed_protocol(),
131129
host = "127.0.0.1",
132-
port = trino.exposed_port(),
130+
port = cluster.exposed_port(),
133131
),
134132
]),
135133
}),
@@ -141,3 +139,120 @@ pub fn add_graceful_shutdown_config(
141139

142140
Ok(())
143141
}
142+
143+
#[cfg(test)]
144+
mod tests {
145+
use stackable_operator::shared::time::Duration;
146+
147+
use super::*;
148+
use crate::controller::build::properties::test_support::validated_cluster_from_yaml;
149+
150+
/// A worker role group without an explicit `gracefulShutdownTimeout` falls back to the
151+
/// product default.
152+
#[test]
153+
fn min_worker_timeout_defaults() {
154+
let cluster = validated_cluster_from_yaml(
155+
r#"
156+
apiVersion: trino.stackable.tech/v1alpha1
157+
kind: TrinoCluster
158+
metadata:
159+
name: simple-trino
160+
namespace: default
161+
uid: "e6ac237d-a6d4-43a1-8135-f36506110912"
162+
spec:
163+
image:
164+
productVersion: "479"
165+
clusterConfig:
166+
catalogLabelSelector: {}
167+
coordinators:
168+
roleGroups:
169+
default:
170+
replicas: 1
171+
workers:
172+
roleGroups:
173+
default:
174+
replicas: 1
175+
"#,
176+
);
177+
assert_eq!(
178+
min_worker_graceful_shutdown_timeout(&cluster),
179+
DEFAULT_WORKER_GRACEFUL_SHUTDOWN_TIMEOUT
180+
);
181+
}
182+
183+
/// A role-level `gracefulShutdownTimeout` is merged into every worker role group.
184+
#[test]
185+
fn min_worker_timeout_from_role() {
186+
let cluster = validated_cluster_from_yaml(
187+
r#"
188+
apiVersion: trino.stackable.tech/v1alpha1
189+
kind: TrinoCluster
190+
metadata:
191+
name: simple-trino
192+
namespace: default
193+
uid: "e6ac237d-a6d4-43a1-8135-f36506110912"
194+
spec:
195+
image:
196+
productVersion: "479"
197+
clusterConfig:
198+
catalogLabelSelector: {}
199+
coordinators:
200+
roleGroups:
201+
default:
202+
replicas: 1
203+
workers:
204+
config:
205+
gracefulShutdownTimeout: 42h
206+
roleGroups:
207+
default:
208+
replicas: 1
209+
"#,
210+
);
211+
assert_eq!(
212+
min_worker_graceful_shutdown_timeout(&cluster),
213+
Duration::from_hours_unchecked(42)
214+
);
215+
}
216+
217+
/// The minimum is taken across all worker role groups (role <- role-group merge applied).
218+
#[test]
219+
fn min_worker_timeout_across_role_groups() {
220+
let cluster = validated_cluster_from_yaml(
221+
r#"
222+
apiVersion: trino.stackable.tech/v1alpha1
223+
kind: TrinoCluster
224+
metadata:
225+
name: simple-trino
226+
namespace: default
227+
uid: "e6ac237d-a6d4-43a1-8135-f36506110912"
228+
spec:
229+
image:
230+
productVersion: "479"
231+
clusterConfig:
232+
catalogLabelSelector: {}
233+
coordinators:
234+
roleGroups:
235+
default:
236+
replicas: 1
237+
workers:
238+
config:
239+
gracefulShutdownTimeout: 42h
240+
roleGroups:
241+
normal:
242+
replicas: 1
243+
short:
244+
replicas: 1
245+
config:
246+
gracefulShutdownTimeout: 5m
247+
long:
248+
replicas: 1
249+
config:
250+
gracefulShutdownTimeout: 7d
251+
"#,
252+
);
253+
assert_eq!(
254+
min_worker_graceful_shutdown_timeout(&cluster),
255+
Duration::from_minutes_unchecked(5)
256+
);
257+
}
258+
}

‎rust/operator-binary/src/controller/build/resource/listener.rs‎

Lines changed: 12 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,10 @@ use stackable_operator::{
1010
kvp::{Labels, ObjectLabels},
1111
};
1212

13-
use crate::crd::{TrinoRole, v1alpha1};
13+
use crate::{
14+
controller::ValidatedCluster,
15+
crd::{TrinoRole, v1alpha1},
16+
};
1417

1518
pub const LISTENER_VOLUME_NAME: &str = "listener";
1619
pub const LISTENER_VOLUME_DIR: &str = "/stackable/listener";
@@ -34,23 +37,23 @@ pub enum Error {
3437
}
3538

3639
pub fn build_group_listener(
37-
trino: &v1alpha1::TrinoCluster,
38-
object_labels: ObjectLabels<v1alpha1::TrinoCluster>,
40+
cluster: &ValidatedCluster,
41+
object_labels: ObjectLabels<ValidatedCluster>,
3942
listener_class: String,
4043
listener_group_name: String,
4144
) -> Result<Listener, Error> {
4245
Ok(Listener {
4346
metadata: ObjectMetaBuilder::new()
44-
.name_and_namespace(trino)
47+
.name_and_namespace(cluster)
4548
.name(listener_group_name)
46-
.ownerreference_from_resource(trino, None, Some(true))
49+
.ownerreference_from_resource(cluster, None, Some(true))
4750
.context(ObjectMissingMetadataForOwnerRefSnafu)?
4851
.with_recommended_labels(&object_labels)
4952
.context(BuildObjectMetaSnafu)?
5053
.build(),
5154
spec: ListenerSpec {
5255
class_name: Some(listener_class),
53-
ports: Some(listener_ports(trino)),
56+
ports: Some(listener_ports(cluster)),
5457
..ListenerSpec::default()
5558
},
5659
status: None,
@@ -91,9 +94,9 @@ pub fn secret_volume_listener_scope(role: &TrinoRole) -> Option<String> {
9194
}
9295

9396
/// We only use the http/https port here and intentionally omit the metrics one.
94-
fn listener_ports(trino: &v1alpha1::TrinoCluster) -> Vec<ListenerPort> {
95-
let name = trino.exposed_protocol().to_string();
96-
let port = trino.exposed_port().into();
97+
fn listener_ports(cluster: &ValidatedCluster) -> Vec<ListenerPort> {
98+
let name = cluster.exposed_protocol().to_string();
99+
let port = cluster.exposed_port().into();
97100

98101
vec![ListenerPort {
99102
name,

‎rust/operator-binary/src/controller/build/resource/service.rs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ use stackable_operator::{
99

1010
use crate::{
1111
controller::ValidatedCluster,
12-
crd::{METRICS_PORT, METRICS_PORT_NAME, TrinoRole, v1alpha1},
12+
crd::{METRICS_PORT, METRICS_PORT_NAME, TrinoRole},
1313
};
1414

1515
#[derive(Snafu, Debug)]
@@ -99,9 +99,9 @@ pub fn build_rolegroup_metrics_service(
9999
})
100100
}
101101

102-
pub(crate) fn headless_service_ports(trino: &v1alpha1::TrinoCluster) -> Vec<ServicePort> {
103-
let name = trino.exposed_protocol().to_string();
104-
let port = trino.exposed_port().into();
102+
pub(crate) fn headless_service_ports(cluster: &ValidatedCluster) -> Vec<ServicePort> {
103+
let name = cluster.exposed_protocol().to_string();
104+
let port = cluster.exposed_port().into();
105105

106106
vec![ServicePort {
107107
name: Some(name),

‎rust/operator-binary/src/controller/build/resource/statefulset.rs‎

Lines changed: 25 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -214,7 +214,7 @@ pub fn build_rolegroup_statefulset(
214214
)
215215
.context(InvalidAuthenticationConfigSnafu)?;
216216
build::graceful_shutdown::add_graceful_shutdown_config(
217-
trino,
217+
cluster,
218218
trino_role,
219219
merged_config,
220220
&mut pod_builder,
@@ -363,12 +363,12 @@ pub fn build_rolegroup_statefulset(
363363
.context(AddVolumeMountSnafu)?
364364
.add_volume_mount("log", STACKABLE_LOG_DIR)
365365
.context(AddVolumeMountSnafu)?
366-
.add_container_ports(container_ports(trino))
366+
.add_container_ports(container_ports(cluster))
367367
.resources(merged_config.resources.clone().into())
368368
// The probes are set on coordinators and workers
369-
.startup_probe(startup_probe(trino))
370-
.readiness_probe(readiness_probe(trino))
371-
.liveness_probe(liveness_probe(trino))
369+
.startup_probe(startup_probe(cluster))
370+
.readiness_probe(readiness_probe(cluster))
371+
.liveness_probe(liveness_probe(cluster))
372372
.build();
373373

374374
// add trino container first to better default into that container (e.g. instead of vector)
@@ -529,38 +529,36 @@ fn env_var_from_secret(secret_name: &str, secret_key: Option<&str>, env_var: &st
529529
}
530530
}
531531

532-
fn container_ports(trino: &v1alpha1::TrinoCluster) -> Vec<ContainerPort> {
532+
fn container_ports(cluster: &ValidatedCluster) -> Vec<ContainerPort> {
533533
let mut ports = vec![ContainerPort {
534534
name: Some(METRICS_PORT_NAME.to_string()),
535535
container_port: METRICS_PORT.into(),
536536
protocol: Some("TCP".to_string()),
537537
..ContainerPort::default()
538538
}];
539539

540-
if trino.expose_http_port() {
541-
ports.push(ContainerPort {
542-
name: Some(HTTP_PORT_NAME.to_string()),
543-
container_port: HTTP_PORT.into(),
544-
protocol: Some("TCP".to_string()),
545-
..ContainerPort::default()
546-
})
547-
}
548-
549-
if trino.expose_https_port() {
540+
if cluster.server_tls_enabled() {
550541
ports.push(ContainerPort {
551542
name: Some(HTTPS_PORT_NAME.to_string()),
552543
container_port: HTTPS_PORT.into(),
553544
protocol: Some("TCP".to_string()),
554545
..ContainerPort::default()
555546
});
547+
} else {
548+
ports.push(ContainerPort {
549+
name: Some(HTTP_PORT_NAME.to_string()),
550+
container_port: HTTP_PORT.into(),
551+
protocol: Some("TCP".to_string()),
552+
..ContainerPort::default()
553+
})
556554
}
557555

558556
ports
559557
}
560558

561-
fn startup_probe(trino: &v1alpha1::TrinoCluster) -> Probe {
559+
fn startup_probe(cluster: &ValidatedCluster) -> Probe {
562560
Probe {
563-
exec: Some(finished_starting_probe(trino)),
561+
exec: Some(finished_starting_probe(cluster)),
564562
period_seconds: Some(5),
565563
// Give the coordinator or worker 10 minutes to start up
566564
failure_threshold: Some(120),
@@ -569,19 +567,19 @@ fn startup_probe(trino: &v1alpha1::TrinoCluster) -> Probe {
569567
}
570568
}
571569

572-
fn readiness_probe(trino: &v1alpha1::TrinoCluster) -> Probe {
570+
fn readiness_probe(cluster: &ValidatedCluster) -> Probe {
573571
Probe {
574-
http_get: Some(http_get_probe(trino)),
572+
http_get: Some(http_get_probe(cluster)),
575573
period_seconds: Some(5),
576574
failure_threshold: Some(1),
577575
timeout_seconds: Some(3),
578576
..Probe::default()
579577
}
580578
}
581579

582-
fn liveness_probe(trino: &v1alpha1::TrinoCluster) -> Probe {
580+
fn liveness_probe(cluster: &ValidatedCluster) -> Probe {
583581
Probe {
584-
http_get: Some(http_get_probe(trino)),
582+
http_get: Some(http_get_probe(cluster)),
585583
period_seconds: Some(5),
586584
// Coordinators are currently not highly available, so you always have a singe instance.
587585
// Restarting it causes all queries to fail, so let's not restart it directly after the first
@@ -597,8 +595,8 @@ fn liveness_probe(trino: &v1alpha1::TrinoCluster) -> Probe {
597595
///
598596
/// This is the same probe as the [upstream helm-chart](https://github.com/trinodb/charts/blob/7cd0a7bff6c52e0ee6ca6d5394cd72c150ad4379/charts/trino/templates/deployment-coordinator.yaml#L214)
599597
/// is using.
600-
fn http_get_probe(trino: &v1alpha1::TrinoCluster) -> HTTPGetAction {
601-
let (schema, port_name) = if trino.expose_https_port() {
598+
fn http_get_probe(cluster: &ValidatedCluster) -> HTTPGetAction {
599+
let (schema, port_name) = if cluster.server_tls_enabled() {
602600
("HTTPS", HTTPS_PORT_NAME)
603601
} else {
604602
("HTTP", HTTP_PORT_NAME)
@@ -615,9 +613,9 @@ fn http_get_probe(trino: &v1alpha1::TrinoCluster) -> HTTPGetAction {
615613
/// Wait until `/v1/info` returns `"starting":false`.
616614
///
617615
/// This probe works on coordinators and workers.
618-
fn finished_starting_probe(trino: &v1alpha1::TrinoCluster) -> ExecAction {
619-
let port = trino.exposed_port();
620-
let schema = if trino.expose_https_port() {
616+
fn finished_starting_probe(cluster: &ValidatedCluster) -> ExecAction {
617+
let port = cluster.exposed_port();
618+
let schema = if cluster.server_tls_enabled() {
621619
"https"
622620
} else {
623621
"http"

0 commit comments

Comments
 (0)