Skip to content

Commit 029816d

Browse files
maltesanderclaudeadwk67
authored
refactor: extract dereference/validate pipeline from reconcile (#687)
* refactor(app): scaffold dereference and validate submodules Empty modules with snafu Error stubs. Wires up `mod` declarations so later commits can fill them in without further plumbing changes. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(app): add dereference module with template/S3/logdir fetch Defines DereferencedSparkApplication and an async dereference() function that consolidates template merging, S3 connection resolution, the early S3 TLS sanity check, and log directory resolution. Not yet wired into reconcile(). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(app): add validate module with image + product-config validation Defines ValidatedSparkApplication and a sync validate() function that resolves the product image (moved from reconcile()) and runs validated_role_config. Not yet wired into reconcile(). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(app): wire reconcile through dereference + validate reconcile() now delegates template merging, S3 resolution, TLS sanity checking, log directory resolution, image resolution, and product-config validation to the new submodules. The function body shrinks by ~70 lines and the controller's Error enum loses six variants now living in the submodules. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * chore(history): rename history_controller.rs to controller.rs Aligns history with connect (which already uses controller.rs) ahead of splitting it into a controller/ submodule folder. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(history): scaffold dereference and validate submodules Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(history): add dereference module with log directory fetch Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(history): add validate module with image + product-config validation Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(history): wire reconcile through dereference + validate Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(connect): scaffold dereference and validate submodules Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(connect): add dereference module with S3 connection fetch Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(connect): add validate module with image + role config resolution Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * refactor(connect): wire reconcile through dereference + validate Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * style: apply cargo fmt to wired controllers Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * fix: correct error message * fix: remove obsolete errors * fix: regenerate crate hashes * test(spark-connect): trim 10-assert to bare CR Available wait Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * test(smoke): add SparkHistoryServer resource snapshot assert Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * test(smoke): add SparkHistoryServer ConfigMap data snapshot Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * test(smoke): add SparkApplication resource snapshot assert Drift-detect the 3 ConfigMaps, ServiceAccount, and RoleBinding that persist after SparkApplication/spark-pi-s3-1 reaches phase: Succeeded. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * test(smoke): add SparkApplication ConfigMap data snapshot Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * test(spark-connect): add SparkConnectServer resource snapshot assert Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * test(spark-connect): add SparkConnectServer ConfigMap data snapshot Snapshot the full `.data` of spark-connect-server and spark-connect-executor ConfigMaps. Uses `.yaml.j2` because spark-defaults.conf and template.yaml differ between s3-use-tls dimensions (https vs http, extraJavaOptions, initContainers, minio-tls-ca volume). The spark.kubernetes.namespace line is substituted at runtime via sed/__NAMESPACE__ to survive randomised kuttl namespaces. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * docs: adapt changelog * fix(smoke): improve diff output * fix(tests): fix test asserts for vector aggregator * fix(connect): add vector aggregator fixes * move TLS check into the validate step * thread dereferenced fields through the validated structs * fix: correct error handling for no-tls s3 bucket * fix: remove obsolete tls validation check --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com> Co-authored-by: Andrew Kenworthy <andrew.kenworthy@stackable.tech>
1 parent 9f44126 commit 029816d

22 files changed

Lines changed: 4752 additions & 282 deletions

File tree

CHANGELOG.md

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,15 +19,16 @@ All notable changes to this project will be documented in this file.
1919
- `SparkApplication`: `spark-env.sh` and `security.properties`
2020
- `SparkHistoryServer`: `spark-defaults.conf`, `spark-env.sh` and `security.properties`
2121
- `SparkConnectServer`: `spark-defaults.conf`, `metrics.properties` and `security.properties`
22-
2322
Previously, arbitrary file names were silently accepted and ignored ([#679]).
2423
- Bump `stackable-operator` to 0.110.1 ([#679]).
24+
- Internal operator refactoring: introduce dereference() and validate() steps in the reconciler for spark application, spark connect and spark history server([#687]).
2525
- test: Bump vector-aggregator to 0.55.0, replace /graphql call with gRPC call ([#689]).
2626

2727
[#674]: https://github.com/stackabletech/spark-k8s-operator/pull/674
2828
[#679]: https://github.com/stackabletech/spark-k8s-operator/pull/679
2929
[#680]: https://github.com/stackabletech/spark-k8s-operator/pull/680
3030
[#684]: https://github.com/stackabletech/spark-k8s-operator/pull/684
31+
[#687]: https://github.com/stackabletech/spark-k8s-operator/pull/687
3132
[#689]: https://github.com/stackabletech/spark-k8s-operator/pull/689
3233

3334
## [26.3.0] - 2026-03-16

Cargo.nix

Lines changed: 9 additions & 9 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

crate-hashes.json

Lines changed: 9 additions & 9 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

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

Lines changed: 39 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ use std::sync::Arc;
33
use snafu::{ResultExt, Snafu};
44
use stackable_operator::{
55
cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
6-
commons::{product_image_selection, rbac::build_rbac_resources},
6+
commons::rbac::build_rbac_resources,
77
kube::{
88
Resource, ResourceExt,
99
core::{DeserializeGuard, error_boundary},
@@ -21,10 +21,13 @@ use strum::{EnumDiscriminants, IntoStaticStr};
2121
use super::crd::{CONNECT_APP_NAME, CONNECT_CONTROLLER_NAME, v1alpha1};
2222
use crate::{
2323
Ctx,
24-
connect::{common, crd::SparkConnectServerStatus, executor, s3, server, service},
25-
crd::constants::{CONTAINER_IMAGE_BASE_NAME, OPERATOR_NAME},
24+
connect::{common, crd::SparkConnectServerStatus, executor, server, service},
25+
crd::constants::OPERATOR_NAME,
2626
};
2727

28+
pub mod dereference;
29+
pub mod validate;
30+
2831
#[derive(Snafu, Debug, EnumDiscriminants)]
2932
#[strum_discriminants(derive(IntoStaticStr))]
3033
#[allow(clippy::enum_variant_names)]
@@ -124,11 +127,6 @@ pub enum Error {
124127
BuildRbacResources {
125128
source: stackable_operator::commons::rbac::Error,
126129
},
127-
#[snafu(display("failed to build connect server configuration"))]
128-
ServerConfig { source: crate::connect::crd::Error },
129-
130-
#[snafu(display("failed to build connect executor configuration"))]
131-
ExecutorConfig { source: crate::connect::crd::Error },
132130

133131
#[snafu(display("failed to build connect executor pod template"))]
134132
ExecutorPodTemplate {
@@ -138,16 +136,14 @@ pub enum Error {
138136
#[snafu(display("failed to serialize executor pod template"))]
139137
ExecutorPodTemplateSerde { source: serde_yaml::Error },
140138

141-
#[snafu(display("failed to resolve product image"))]
142-
ResolveProductImage {
143-
source: product_image_selection::Error,
144-
},
145-
146-
#[snafu(display("failed to resolve S3 connections for SparkConnectServer {name:?}"))]
147-
ResolveS3Connections { source: s3::Error, name: String },
148-
149139
#[snafu(display("failed to build connect server S3 properties"))]
150140
S3SparkProperties { source: crate::connect::s3::Error },
141+
142+
#[snafu(display("failed to dereference SparkConnectServer"))]
143+
DereferenceSparkConnectServer { source: dereference::Error },
144+
145+
#[snafu(display("failed to validate SparkConnectServer"))]
146+
ValidateSparkConnectServer { source: validate::Error },
151147
}
152148

153149
type Result<T, E = Error> = std::result::Result<T, E>;
@@ -170,12 +166,22 @@ pub async fn reconcile(
170166
.map_err(error_boundary::InvalidObject::clone)
171167
.context(InvalidSparkConnectServerSnafu)?;
172168

173-
let server_config = scs.server_config().context(ServerConfigSnafu)?;
174-
let server_role_config = &scs.spec.server.role_config;
175-
let executor_config = scs.executor_config().context(ExecutorConfigSnafu)?;
176-
177169
let client = &ctx.client;
178170

171+
let dereferenced = dereference::dereference(client, scs)
172+
.await
173+
.context(DereferenceSparkConnectServerSnafu)?;
174+
175+
let validated = validate::validate(scs, dereferenced, &ctx.operator_environment)
176+
.context(ValidateSparkConnectServerSnafu)?;
177+
178+
let server_config = &validated.server_config;
179+
let executor_config = &validated.executor_config;
180+
let resolved_product_image = &validated.resolved_product_image;
181+
let resolved_s3 = &validated.resolved_s3;
182+
183+
let server_role_config = &scs.spec.server.role_config;
184+
179185
let mut cluster_resources = ClusterResources::new(
180186
CONNECT_APP_NAME,
181187
OPERATOR_NAME,
@@ -186,23 +192,6 @@ pub async fn reconcile(
186192
)
187193
.context(CreateClusterResourcesSnafu)?;
188194

189-
let resolved_product_image = scs
190-
.spec
191-
.image
192-
.resolve(
193-
CONTAINER_IMAGE_BASE_NAME,
194-
&ctx.operator_environment.image_repository,
195-
crate::built_info::PKG_VERSION,
196-
)
197-
.context(ResolveProductImageSnafu)?;
198-
199-
// Resolve any S3 connections early to fail fast if there are issues.
200-
let resolved_s3 = s3::ResolvedS3::resolve(client, scs)
201-
.await
202-
.with_context(|_| ResolveS3ConnectionsSnafu {
203-
name: scs.name_unchecked(),
204-
})?;
205-
206195
// Use a dedicated service account for connect server pods.
207196
let (service_account, role_binding) = build_rbac_resources(
208197
scs,
@@ -251,21 +240,21 @@ pub async fn reconcile(
251240
.context(S3SparkPropertiesSnafu)?,
252241
server::server_properties(
253242
scs,
254-
&server_config,
243+
server_config,
255244
&applied_headless_service,
256245
&service_account,
257-
&resolved_product_image,
246+
resolved_product_image,
258247
)
259248
.context(ServerPropertiesSnafu)?,
260-
executor::executor_properties(scs, &executor_config, &resolved_product_image)
249+
executor::executor_properties(scs, executor_config, resolved_product_image)
261250
.context(ExecutorPropertiesSnafu)?,
262251
])
263252
.context(SerializePropertiesSnafu)?;
264253

265254
// ========================================
266255
// Executor config map and pod template
267256
let executor_config_map =
268-
executor::executor_config_map(scs, &executor_config, &resolved_product_image).context(
257+
executor::executor_config_map(scs, executor_config, resolved_product_image).context(
269258
BuildExecutorConfigMapSnafu {
270259
name: scs.name_unchecked(),
271260
},
@@ -280,10 +269,10 @@ pub async fn reconcile(
280269
let executor_pod_template = serde_yaml::to_string(
281270
&executor::executor_pod_template(
282271
scs,
283-
&executor_config,
284-
&resolved_product_image,
272+
executor_config,
273+
resolved_product_image,
285274
&executor_config_map,
286-
&resolved_s3,
275+
resolved_s3,
287276
)
288277
.context(ExecutorPodTemplateSnafu)?,
289278
)
@@ -293,8 +282,8 @@ pub async fn reconcile(
293282
// Server config map
294283
let server_config_map = server::server_config_map(
295284
scs,
296-
&server_config,
297-
&resolved_product_image,
285+
server_config,
286+
resolved_product_image,
298287
&spark_props,
299288
&executor_pod_template,
300289
)
@@ -310,7 +299,7 @@ pub async fn reconcile(
310299

311300
// ========================================
312301
// Server listener
313-
let listener = server::build_listener(scs, server_role_config, &resolved_product_image)
302+
let listener = server::build_listener(scs, server_role_config, resolved_product_image)
314303
.context(BuildListenerSnafu)?;
315304

316305
let applied_listener = cluster_resources
@@ -323,13 +312,13 @@ pub async fn reconcile(
323312
let args = server::command_args(&scs.spec.args);
324313
let stateful_set = server::build_stateful_set(
325314
scs,
326-
&server_config,
327-
&resolved_product_image,
315+
server_config,
316+
resolved_product_image,
328317
&service_account,
329318
&server_config_map,
330319
&applied_listener.name_any(),
331320
args,
332-
&resolved_s3,
321+
resolved_s3,
333322
)
334323
.context(BuildServerStatefulSetSnafu)?;
335324

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
//! The dereference step in the SparkConnectServer controller.
2+
//!
3+
//! Fetches the resolved S3 configuration referenced by the SparkConnectServer spec.
4+
5+
use snafu::{ResultExt, Snafu};
6+
use stackable_operator::{client::Client, kube::ResourceExt};
7+
8+
use crate::connect::{crd::v1alpha1, s3};
9+
10+
#[derive(Snafu, Debug)]
11+
pub enum Error {
12+
#[snafu(display("failed to resolve S3 connection for {name}"))]
13+
ResolveS3Connections { source: s3::Error, name: String },
14+
}
15+
16+
type Result<T, E = Error> = std::result::Result<T, E>;
17+
18+
pub struct DereferencedSparkConnectServer {
19+
pub resolved_s3: s3::ResolvedS3,
20+
}
21+
22+
pub async fn dereference(
23+
client: &Client,
24+
scs: &v1alpha1::SparkConnectServer,
25+
) -> Result<DereferencedSparkConnectServer> {
26+
let resolved_s3 = s3::ResolvedS3::resolve(client, scs)
27+
.await
28+
.with_context(|_| ResolveS3ConnectionsSnafu {
29+
name: scs.name_unchecked(),
30+
})?;
31+
32+
Ok(DereferencedSparkConnectServer { resolved_s3 })
33+
}

0 commit comments

Comments
 (0)