refactor: split opa/s3 into fetch & validate

Author: maltesander

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

@@ -1,11 +1,4 @@
 //! The dereference step in the DruidCluster controller
-//!
-//! Fetches all Kubernetes objects referenced by the DruidCluster spec and returns them in
-//! [`DereferencedObjects`]. AuthenticationClasses are fetched raw here
-//! ([`fetch_authentication_classes`]) and validated later in the validate step. The remaining
-//! helpers (`DruidCluster::get_s3_connection`, `S3Bucket::resolve`,
-//! `OpaConfig::full_document_url_from_config_map`) still mix fetching and validation; their
-//! outputs are treated as "dereferenced" for now. Splitting those is a follow-up.
 
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
@@ -50,11 +43,13 @@ pub enum Error {
         cm_name: String,
     },
 
-    #[snafu(display("failed to get valid S3 connection"))]
-    GetS3Connection { source: crate::crd::Error },
+    #[snafu(display("failed to resolve the ingestion S3 connection"))]
+    ResolveS3Connection {
+        source: stackable_operator::crd::s3::v1alpha1::ConnectionError,
+    },
 
-    #[snafu(display("failed to get deep storage bucket"))]
-    GetDeepStorageBucket {
+    #[snafu(display("failed to resolve the deep storage S3 bucket"))]
+    ResolveS3Bucket {
         source: stackable_operator::crd::s3::v1alpha1::BucketError,
     },
 
@@ -66,13 +61,19 @@ pub enum Error {
 
 type Result<T, E = Error> = std::result::Result<T, E>;
 
-/// Kubernetes objects referenced from the DruidCluster spec, already fetched (and, for now,
-/// partly validated by the existing helper functions).
+/// Kubernetes objects referenced from the DruidCluster spec, already fetched. Validation and
+/// derivation of final values happens in the validate step.
 pub struct DereferencedObjects {
     pub zookeeper_connection_string: String,
-    pub opa_connection_string: Option<String>,
-    pub s3_connection: Option<s3::v1alpha1::ConnectionSpec>,
-    pub deep_storage_bucket_name: Option<String>,
+    /// The rule-agnostic OPA document URL (package level, no rule). The validate step appends the
+    /// concrete authorization rule to produce the connection string.
+    pub opa_base_document_url: Option<String>,
+    /// The resolved ingestion S3 connection (if any). The validate step checks it against the deep
+    /// storage connection.
+    pub s3_ingestion_connection: Option<s3::v1alpha1::ConnectionSpec>,
+    /// The resolved deep storage S3 bucket (if deep storage uses S3). Carries both the bucket name
+    /// and its connection.
+    pub s3_deep_storage_bucket: Option<s3::v1alpha1::ResolvedBucket>,
     /// The raw, fetched `AuthenticationClass` objects (in spec order). Validation of these happens
     /// in the validate step via [`AuthenticationClassesResolved::from_fetched`].
     pub authentication_classes: Vec<core::v1alpha1::AuthenticationClass>,
@@ -106,12 +107,14 @@ pub async fn dereference(
             cm_name: zk_confmap.clone(),
         })?;
 
-    let opa_connection_string = if let Some(DruidAuthorization { opa: opa_config }) =
+    // Fetch the rule-agnostic OPA document URL (package level, no rule). The validate step appends
+    // the concrete authorization rule.
+    let opa_base_document_url = if let Some(DruidAuthorization { opa: opa_config }) =
         &druid.spec.cluster_config.authorization
     {
         Some(
             opa_config
-                .full_document_url_from_config_map(client, druid, Some("allow"), &OpaApiVersion::V1)
+                .full_document_url_from_config_map(client, druid, None, &OpaApiVersion::V1)
                 .await
                 .context(GetOpaConnStringSnafu {
                     cm_name: opa_config.config_map_name.clone(),
@@ -121,20 +124,34 @@ pub async fn dereference(
         None
     };
 
-    let s3_connection = druid
-        .get_s3_connection(client)
-        .await
-        .context(GetS3ConnectionSnafu)?;
+    // Resolve the ingestion and deep storage S3 references separately. Checking that they are
+    // compatible (and extracting the bucket name) is done in the validate step.
+    let s3_ingestion_connection = if let Some(ingestion_connection) = druid
+        .spec
+        .cluster_config
+        .ingestion
+        .as_ref()
+        .and_then(|ingestion| ingestion.s3connection.as_ref())
+    {
+        Some(
+            ingestion_connection
+                .clone()
+                .resolve(client, namespace)
+                .await
+                .context(ResolveS3ConnectionSnafu)?,
+        )
+    } else {
+        None
+    };
 
-    let deep_storage_bucket_name = match &druid.spec.cluster_config.deep_storage {
+    let s3_deep_storage_bucket = match &druid.spec.cluster_config.deep_storage {
         DeepStorageSpec::S3(s3_spec) => Some(
             s3_spec
                 .bucket
                 .clone()
                 .resolve(client, namespace)
                 .await
-                .context(GetDeepStorageBucketSnafu)?
-                .bucket_name,
+                .context(ResolveS3BucketSnafu)?,
         ),
         _ => None,
     };
@@ -145,9 +162,9 @@ pub async fn dereference(
 
     Ok(DereferencedObjects {
         zookeeper_connection_string,
-        opa_connection_string,
-        s3_connection,
-        deep_storage_bucket_name,
+        opa_base_document_url,
+        s3_ingestion_connection,
+        s3_deep_storage_bucket,
         authentication_classes,
     })
 }

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

@@ -9,7 +9,7 @@ use std::{
     str::FromStr,
 };
 
-use snafu::{ResultExt, Snafu};
+use snafu::{ResultExt, Snafu, ensure};
 use stackable_operator::{
     builder::meta::ObjectMetaBuilder,
     cli::OperatorEnvironmentOptions,
@@ -70,6 +70,11 @@ pub enum Error {
     ResolveAuthenticationClasses {
         source: crate::crd::authentication::Error,
     },
+
+    #[snafu(display(
+        "two differing S3 connections were given (ingestion and deep storage), this is unsupported by Druid"
+    ))]
+    IncompatibleS3Connections,
 }
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -373,16 +378,46 @@ pub fn validate(
         .jdbc_connection_details("metadata")
         .context(InvalidMetadataDatabaseConnectionSnafu)?;
 
+    // The deep storage bucket carries its own S3 connection; if ingestion also configures one, the
+    // two must be identical, as Druid only supports a single S3 connection.
+    let s3_deep_storage_connection = dereferenced_objects
+        .s3_deep_storage_bucket
+        .as_ref()
+        .map(|bucket| &bucket.connection);
+    let s3_connection = match (
+        dereferenced_objects.s3_ingestion_connection.as_ref(),
+        s3_deep_storage_connection,
+    ) {
+        (Some(ingestion), Some(deep_storage)) => {
+            ensure!(ingestion == deep_storage, IncompatibleS3ConnectionsSnafu);
+            Some(ingestion.clone())
+        }
+        (Some(connection), None) | (None, Some(connection)) => Some(connection.clone()),
+        (None, None) => None,
+    };
+    let deep_storage_bucket_name = dereferenced_objects
+        .s3_deep_storage_bucket
+        .as_ref()
+        .map(|bucket| bucket.bucket_name.clone());
+
+    // The dereference step fetched the rule-agnostic OPA document URL; append the authorization
+    // rule Druid queries.
+    // TODO(@maltesander): This would rather be a preprocess step?
+    let opa_connection_string = dereferenced_objects
+        .opa_base_document_url
+        .as_ref()
+        .map(|base_url| format!("{base_url}/allow"));
+
     Ok(ValidatedCluster::new(
         name,
         namespace,
         uid,
         image,
         ValidatedClusterConfig {
             zookeeper_connection_string: dereferenced_objects.zookeeper_connection_string.clone(),
-            opa_connection_string: dereferenced_objects.opa_connection_string.clone(),
-            s3_connection: dereferenced_objects.s3_connection.clone(),
-            deep_storage_bucket_name: dereferenced_objects.deep_storage_bucket_name.clone(),
+            opa_connection_string,
+            s3_connection,
+            deep_storage_bucket_name,
             druid_tls_security,
             druid_auth_config,
             metadata_database: druid.spec.cluster_config.metadata_database.clone(),
@@ -553,9 +588,9 @@ mod tests {
     fn dereferenced_objects() -> DereferencedObjects {
         DereferencedObjects {
             zookeeper_connection_string: "zookeeper-connection-string".to_string(),
-            opa_connection_string: None,
-            s3_connection: None,
-            deep_storage_bucket_name: None,
+            opa_base_document_url: None,
+            s3_ingestion_connection: None,
+            s3_deep_storage_bucket: None,
             authentication_classes: Vec::new(),
         }
     }

rust/operator-binary/src/crd/mod.rs

@@ -8,7 +8,6 @@ use security::add_cert_to_jvm_trust_store_cmd;
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
-    client::Client,
     commons::{
         affinity::StackableAffinity,
         cluster_operation::ClusterOperation,
@@ -152,25 +151,9 @@ pub enum Error {
     #[snafu(display("missing secret lifetime"))]
     MissingSecretLifetime,
 
-    #[snafu(display("failed to resolve S3 connection"))]
-    ResolveS3Connection {
-        source: stackable_operator::crd::s3::v1alpha1::ConnectionError,
-    },
-
-    #[snafu(display("failed to resolve S3 bucket"))]
-    ResolveS3Bucket {
-        source: stackable_operator::crd::s3::v1alpha1::BucketError,
-    },
-
-    #[snafu(display("2 differing s3 connections were given, this is unsupported by Druid"))]
-    IncompatibleS3Connections,
-
     #[snafu(display("the role group {rolegroup_name} is not defined"))]
     CannotRetrieveRoleGroup { rolegroup_name: String },
 
-    #[snafu(display("missing namespace for resource {name}"))]
-    MissingNamespace { name: String },
-
     #[snafu(display("fragment validation failure"))]
     FragmentValidationFailure { source: ValidationError },
 
@@ -363,76 +346,6 @@ impl HasStatusCondition for v1alpha1::DruidCluster {
 }
 
 impl v1alpha1::DruidCluster {
-    /// If an s3 connection for ingestion is given, as well as an s3 connection for deep storage, they need to be the same.
-    /// This function returns the resolved connection, or raises an Error if the connections are not identical.
-    pub async fn get_s3_connection(
-        &self,
-        client: &Client,
-    ) -> Result<Option<s3::v1alpha1::ConnectionSpec>, Error> {
-        // retrieve connection for ingestion (can be None)
-        let ingestion_conn = if let Some(ic) = self
-            .spec
-            .cluster_config
-            .ingestion
-            .as_ref()
-            .and_then(|is| is.s3connection.as_ref())
-        {
-            Some(
-                ic.clone()
-                    .resolve(
-                        client,
-                        self.namespace()
-                            .context(MissingNamespaceSnafu {
-                                name: &self.name_unchecked(),
-                            })?
-                            .as_ref(),
-                    )
-                    .await
-                    .context(ResolveS3ConnectionSnafu)?,
-            )
-        } else {
-            None
-        };
-
-        // retrieve connection for deep storage (can be None)
-        let storage_conn = match &self.spec.cluster_config.deep_storage {
-            DeepStorageSpec::S3(s3_spec) => {
-                let inlined_bucket = s3_spec
-                    .bucket
-                    .clone()
-                    .resolve(
-                        client,
-                        self.namespace()
-                            .context(MissingNamespaceSnafu {
-                                name: &self.name_unchecked(),
-                            })?
-                            .as_ref(),
-                    )
-                    .await
-                    .context(ResolveS3BucketSnafu)?;
-                Some(inlined_bucket.connection)
-            }
-            _ => None,
-        };
-
-        // if both connections are specified and are identical, return it
-        // if they differ, raise an error
-        // if only one connection is specified, return it
-        if ingestion_conn.is_some() && storage_conn.is_some() {
-            if ingestion_conn == storage_conn {
-                Ok(ingestion_conn)
-            } else {
-                Err(Error::IncompatibleS3Connections)
-            }
-        } else if ingestion_conn.is_some() {
-            Ok(ingestion_conn)
-        } else if storage_conn.is_some() {
-            Ok(storage_conn)
-        } else {
-            Ok(None)
-        }
-    }
-
     /// Merges and validates all role groups of the given role. Invoked from the validate step
     /// (`controller::validate`).
     ///