Skip to content

refactor: extract dereference/validate pipeline from reconcile_hive #707

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
adwk67 wants to merge 11 commits into
base: main
Choose a base branch
from
Open

refactor: extract dereference/validate pipeline from reconcile_hive #707

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
30acbeb
refactor: extract dereference/validate pipeline from reconcile_hive
adwk67 May 12, 2026
6da580c
refactor: extract validate_cluster into controller::validate module
adwk67 May 12, 2026
63744a2
refactor: align error variant naming with airflow and hbase operators
adwk67 May 13, 2026
1b61308
refactor: move product image resolution from dereference to validate
adwk67 May 13, 2026
213bac3
changelog
adwk67 May 13, 2026
49e4a40
Merge branch 'main' into feature/validated-config-types-2
adwk67 May 13, 2026
55ed469
Update rust/operator-binary/src/controller.rs
adwk67 May 15, 2026
6db46c1
assume single role in ValidatedHiveCluster
adwk67 May 15, 2026
bd9b277
move db connection details from dereferenced to the controller and re…
adwk67 May 15, 2026
1758748
move validated cluster to the controller which will be the central de…
adwk67 May 15, 2026
9cdd1e1
replace parameter with constant
adwk67 May 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,11 +17,13 @@ All notable changes to this project will be documented in this file.
This means you need to replace your simple database connection string with a typed struct.
This struct is consistent between different CRDs, so that you can easily copy/paste it between stacklets.
Read on the [Hive database documentation](https://docs.stackable.tech/home/nightly/hive/usage-guide/database-driver) for details ([#674]).
- Internal operator refactoring: introduce dereference() and validate() steps in the reconciler ([#707]).

[#674]: https://github.com/stackabletech/hive-operator/pull/674
[#693]: https://github.com/stackabletech/hive-operator/pull/693
[#695]: https://github.com/stackabletech/hive-operator/pull/695
[#702]: https://github.com/stackabletech/hive-operator/pull/702
[#707]: https://github.com/stackabletech/hive-operator/pull/707

## [26.3.0] - 2026-03-16

Expand Down
215 changes: 70 additions & 145 deletions rust/operator-binary/src/controller.rs
View file
Open in desktop
Copy link
Copy Markdown
Member

@siegfriedweber siegfriedweber May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The smoke test of the opensearch-operator has an assertion that covers all generated resources (https://github.com/stackabletech/opensearch-operator/blob/26.3.0/tests/templates/kuttl/smoke/10-assert.yaml.j2). The other integration tests only add assertions for the resources specific to the test case (https://github.com/stackabletech/opensearch-operator/blob/26.3.0/tests/templates/kuttl/security-config/11-assert.yaml#L29-L41). That leads to relatively high test coverage.

The pull requests for stackabletech/issues#850 should only perform internal changes. If we also add an assertion to the smoke test with the resources produced before this change, then we could be more confident that no unwanted changes are introduced.

Copy link
Copy Markdown
Member Author

@adwk67 adwk67 May 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice idea!

Original file line number Diff line number Diff line change
@@ -1,7 +1,9 @@
//! Ensures that `Pod`s are configured and running for each [`v1alpha1::HiveCluster`]

mod dereference;
mod validate;

use std::{
borrow::Cow,
collections::{BTreeMap, HashMap},
hash::Hasher,
sync::Arc,
Expand Down Expand Up @@ -35,8 +37,7 @@ use stackable_operator::{
cli::OperatorEnvironmentOptions,
cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
commons::{
product_image_selection::{self, ResolvedProductImage},
rbac::build_rbac_resources,
product_image_selection::ResolvedProductImage, rbac::build_rbac_resources,
secret_class::SecretClassVolumeProvisionParts,
},
constants::RESTART_CONTROLLER_ENABLED_LABEL,
Expand All @@ -63,7 +64,6 @@ use stackable_operator::{
kvp::{Labels, ObjectLabels},
logging::controller::ReconcilerError,
memory::{BinaryMultiple, MemoryQuantity},
product_config_utils::{transform_all_roles_to_config, validate_all_roles_and_groups_config},
product_logging::{
self,
framework::{
Expand All @@ -74,7 +74,7 @@ use stackable_operator::{
CustomContainerLogConfig,
},
},
role_utils::{GenericRoleConfig, RoleGroupRef},
role_utils::RoleGroupRef,
shared::time::Duration,
status::condition::{
compute_conditions, operations::ClusterOperationsConditionBuilder,
Expand All @@ -92,6 +92,7 @@ use crate::{
jvm::{construct_hadoop_heapsize_env, construct_non_heap_jvm_args},
opa::{HiveOpaConfig, OPA_TLS_VOLUME_NAME},
},
controller::validate::{ValidatedRoleConfig, ValidatedRoleGroupConfig},
crd::{
APP_NAME, CORE_SITE_XML, Container, HIVE_PORT, HIVE_PORT_NAME, HIVE_SITE_XML,
HiveClusterStatus, HiveRole, JVM_SECURITY_PROPERTIES_FILE, METRICS_PORT, METRICS_PORT_NAME,
Expand All @@ -100,7 +101,7 @@ use crate::{
STACKABLE_LOG_CONFIG_MOUNT_DIR, STACKABLE_LOG_CONFIG_MOUNT_DIR_NAME, STACKABLE_LOG_DIR,
STACKABLE_LOG_DIR_NAME,
databases::{MetadataDatabaseConnection, derby_driver_class},
v1alpha1::{self, HiveMetastoreRoleConfig},
v1alpha1,
},
discovery::{self},
kerberos::{
Expand All @@ -116,7 +117,7 @@ use crate::{
pub const HIVE_CONTROLLER_NAME: &str = "hivecluster";
pub const HIVE_FULL_CONTROLLER_NAME: &str = concatcp!(HIVE_CONTROLLER_NAME, '.', OPERATOR_NAME);

const CONTAINER_IMAGE_BASE_NAME: &str = "hive";
pub const CONTAINER_IMAGE_BASE_NAME: &str = "hive";

pub const MAX_HIVE_LOG_FILES_SIZE: MemoryQuantity = MemoryQuantity {
value: 10.0,
Expand All @@ -136,9 +137,6 @@ pub enum Error {
#[snafu(display("object defines no namespace"))]
ObjectHasNoNamespace,

#[snafu(display("object defines no metastore role"))]
NoMetaStoreRole,

#[snafu(display("failed to apply Service for {rolegroup}"))]
ApplyRoleGroupService {
source: stackable_operator::cluster_resources::Error,
Expand All @@ -163,16 +161,6 @@ pub enum Error {
rolegroup: RoleGroupRef<v1alpha1::HiveCluster>,
},

#[snafu(display("failed to generate product config"))]
GenerateProductConfig {
source: stackable_operator::product_config_utils::Error,
},

#[snafu(display("invalid product config"))]
InvalidProductConfig {
source: stackable_operator::product_config_utils::Error,
},

#[snafu(display("object is missing metadata to build owner reference"))]
ObjectMissingMetadataForOwnerRef {
source: stackable_operator::builder::meta::Error,
Expand Down Expand Up @@ -201,9 +189,6 @@ pub enum Error {
))]
S3TlsNoVerificationNotSupported,

#[snafu(display("failed to resolve and merge resource config for role and role group"))]
FailedToResolveResourceConfig { source: crate::crd::Error },

#[snafu(display("failed to create hive container [{name}]"))]
FailedToCreateHiveContainer {
source: stackable_operator::builder::pod::container::Error,
Expand Down Expand Up @@ -320,25 +305,18 @@ pub enum Error {
#[snafu(display("failed to configure service"))]
ServiceConfiguration { source: crate::service::Error },

#[snafu(display("failed to resolve product image"))]
ResolveProductImage {
source: product_image_selection::Error,
#[snafu(display("failed to dereference cluster resources"))]
Dereference {
source: crate::controller::dereference::Error,
},

#[snafu(display("invalid OpaConfig"))]
InvalidOpaConfig {
source: stackable_operator::commons::opa::Error,
},
#[snafu(display("failed to validate cluster configuration"))]
Validate { source: validate::Error },

#[snafu(display("failed to build TLS certificate SecretClass Volume"))]
TlsCertSecretClassVolumeBuild {
source: stackable_operator::builder::pod::volume::SecretOperatorVolumeSourceBuilderError,
},

#[snafu(display("invalid metadata database connection"))]
InvalidMetadataDatabaseConnection {
source: stackable_operator::database_connections::Error,
},
}
type Result<T, E = Error> = std::result::Result<T, E>;

Expand All @@ -348,6 +326,17 @@ impl ReconcilerError for Error {
}
}

/// The validated cluster: proves that product-config validation and config merging
/// succeeded for every role and role group before any resources are created.
/// Placed in the controller so that subsequent steps that reference this struct
/// only depend on the controller.
pub struct ValidatedHiveCluster {
pub image: ResolvedProductImage,
pub role_groups: BTreeMap<String, ValidatedRoleGroupConfig>,
pub role_config: Option<ValidatedRoleConfig>,
pub metadata_database_connection_details: JdbcDatabaseConnectionDetails,
}

pub async fn reconcile_hive(
hive: Arc<DeserializeGuard<v1alpha1::HiveCluster>>,
ctx: Arc<Ctx>,
Expand All @@ -361,69 +350,17 @@ pub async fn reconcile_hive(
let client = &ctx.client;
let hive_namespace = hive.namespace().context(ObjectHasNoNamespaceSnafu)?;

let resolved_product_image = hive
.spec
.image
.resolve(
CONTAINER_IMAGE_BASE_NAME,
&ctx.operator_environment.image_repository,
crate::built_info::PKG_VERSION,
)
.context(ResolveProductImageSnafu)?;
let role = hive.spec.metastore.as_ref().context(NoMetaStoreRoleSnafu)?;
let hive_role = HiveRole::MetaStore;

let s3_connection_spec: Option<s3::v1alpha1::ConnectionSpec> =
if let Some(s3) = &hive.spec.cluster_config.s3 {
Some(
s3.clone()
.resolve(
client,
&hive.namespace().ok_or(Error::ObjectHasNoNamespace)?,
)
.await
.context(ConfigureS3ConnectionSnafu)?,
)
} else {
None
};

let metadata_database_connection_details = hive
.spec
.cluster_config
.metadata_database
.jdbc_connection_details("METADATA")
.context(InvalidMetadataDatabaseConnectionSnafu)?;

let validated_config = validate_all_roles_and_groups_config(
&resolved_product_image.product_version,
&transform_all_roles_to_config(
hive,
&[(
HiveRole::MetaStore.to_string(),
(
vec![
PropertyNameKind::Env,
PropertyNameKind::Cli,
PropertyNameKind::File(HIVE_SITE_XML.to_string()),
PropertyNameKind::File(JVM_SECURITY_PROPERTIES_FILE.to_string()),
],
role.clone(),
),
)]
.into(),
)
.context(GenerateProductConfigSnafu)?,
let dereferenced = crate::controller::dereference::dereference(client, hive)
.await
.context(DereferenceSnafu)?;

let validated = validate::validate_cluster(
hive,
&ctx.operator_environment.image_repository,
crate::built_info::PKG_VERSION,
&ctx.product_config,
false,
false,
)
.context(InvalidProductConfigSnafu)?;

let metastore_config = validated_config
.get(&HiveRole::MetaStore.to_string())
.map(Cow::Borrowed)
.unwrap_or_default();
.context(ValidateSnafu)?;

let mut cluster_resources = ClusterResources::new(
APP_NAME,
Expand Down Expand Up @@ -454,55 +391,42 @@ pub async fn reconcile_hive(
.await
.context(ApplyRoleBindingSnafu)?;

let hive_opa_config = match hive.get_opa_config() {
Some(opa_config) => Some(
HiveOpaConfig::from_opa_config(client, hive, opa_config)
.await
.context(InvalidOpaConfigSnafu)?,
),
None => None,
};

let mut ss_cond_builder = StatefulSetConditionBuilder::default();

for (rolegroup_name, rolegroup_config) in metastore_config.iter() {
for (rolegroup_name, validated_rg_config) in &validated.role_groups {
let rolegroup = hive.metastore_rolegroup_ref(rolegroup_name);

let config = hive
.merged_config(&HiveRole::MetaStore, &rolegroup)
.context(FailedToResolveResourceConfigSnafu)?;

let rg_metrics_service =
build_rolegroup_metrics_service(hive, &resolved_product_image, &rolegroup)
build_rolegroup_metrics_service(hive, &validated.image, &rolegroup)
.context(ServiceConfigurationSnafu)?;

let rg_headless_service =
build_rolegroup_headless_service(hive, &resolved_product_image, &rolegroup)
build_rolegroup_headless_service(hive, &validated.image, &rolegroup)
.context(ServiceConfigurationSnafu)?;

let rg_configmap = build_metastore_rolegroup_config_map(
hive,
&hive_namespace,
&resolved_product_image,
&validated.image,
&rolegroup,
rolegroup_config,
&metadata_database_connection_details,
s3_connection_spec.as_ref(),
&config,
&validated_rg_config.product_config_properties,
&validated.metadata_database_connection_details,
dereferenced.s3_connection_spec.as_ref(),
&validated_rg_config.merged_config,
&client.kubernetes_cluster_info,
hive_opa_config.as_ref(),
dereferenced.hive_opa_config.as_ref(),
)?;
let rg_statefulset = build_metastore_rolegroup_statefulset(
hive,
&hive_role,
&resolved_product_image,
&HiveRole::MetaStore,
&validated.image,
&rolegroup,
rolegroup_config,
&metadata_database_connection_details,
s3_connection_spec.as_ref(),
&config,
&validated_rg_config.product_config_properties,
&validated.metadata_database_connection_details,
dereferenced.s3_connection_spec.as_ref(),
&validated_rg_config.merged_config,
&rbac_sa.name_any(),
hive_opa_config.as_ref(),
dereferenced.hive_opa_config.as_ref(),
)?;

cluster_resources
Expand Down Expand Up @@ -539,38 +463,39 @@ pub async fn reconcile_hive(
);
}

let role_config = hive.role_config(&hive_role);
if let Some(HiveMetastoreRoleConfig {
common: GenericRoleConfig {
pod_disruption_budget: pdb,
},
..
}) = role_config
{
add_pdbs(pdb, hive, &hive_role, client, &mut cluster_resources)
.await
.context(FailedToCreatePdbSnafu)?;
}

// std's SipHasher is deprecated, and DefaultHasher is unstable across Rust releases.
// We don't /need/ stability, but it's still nice to avoid spurious changes where possible.
let mut discovery_hash = FnvHasher::with_key(0);

if let Some(HiveMetastoreRoleConfig { listener_class, .. }) = role_config {
let role_listener: Listener =
build_role_listener(hive, &resolved_product_image, &hive_role, listener_class)
.context(ListenerConfigurationSnafu)?;
if let Some(role_config) = validated.role_config {
add_pdbs(
&role_config.pdb,
hive,
&HiveRole::MetaStore,
client,
&mut cluster_resources,
)
.await
.context(FailedToCreatePdbSnafu)?;

let role_listener: Listener = build_role_listener(
hive,
&validated.image,
&HiveRole::MetaStore,
&role_config.listener_class,
)
.context(ListenerConfigurationSnafu)?;
let listener = cluster_resources.add(client, role_listener).await.context(
ApplyGroupListenerSnafu {
role: hive_role.to_string(),
role: HiveRole::MetaStore.to_string(),
},
)?;

for discovery_cm in discovery::build_discovery_configmaps(
hive,
hive,
hive_role,
&resolved_product_image,
HiveRole::MetaStore,
&validated.image,
None,
listener,
)
Expand Down
Loading
Loading