WIP: Add support for configuring Iceberg REST catalog

Cargo.toml

@@ -19,7 +19,7 @@ clap = "4.6"
 const_format = "0.2"
 futures = { version = "0.3", features = ["compat"] }
 indoc = "2.0"
-rstest = "0.26"
+rstest = "0.27"
 semver = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
@@ -28,7 +28,8 @@ snafu = "0.9"
 strum = { version = "0.28", features = ["derive"] }
 tokio = { version = "1.53", features = ["full"] }
 tracing = "0.1"
+url = "2.5"
 
 [patch."https://github.com/stackabletech/operator-rs.git"]
-# stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "main"}
+stackable-operator = { git = "https://github.com/stackabletech//operator-rs.git", branch = "feat/from_docs" }
 # stackable-operator = { path = "../operator-rs/crates/stackable-operator" }

rust/operator-binary/Cargo.toml

@@ -32,6 +32,7 @@ snafu.workspace = true
 strum.workspace = true
 tokio.workspace = true
 tracing.workspace = true
+url.workspace = true
 
 [dev-dependencies]
 rstest.workspace = true

rust/operator-binary/src/catalog/commons.rs

@@ -23,13 +23,13 @@ use crate::{
         CONFIG_DIR_NAME,
         catalog::{
             TrinoCatalogName,
-            commons::{HdfsConnection, MetastoreConnection},
+            commons::{HdfsConnection, HiveMetastoreConnection},
         },
     },
 };
 
 #[async_trait]
-impl ExtendCatalogConfig for MetastoreConnection {
+impl ExtendCatalogConfig for HiveMetastoreConnection {
     async fn extend_catalog_config(
         &self,
         catalog_config: &mut CatalogConfig,

rust/operator-binary/src/catalog/config.rs

@@ -10,7 +10,10 @@ use stackable_operator::{
 
 use crate::{
     catalog::{FromTrinoCatalogError, ToCatalogConfig},
-    crd::catalog::{TrinoCatalogConnector, TrinoCatalogName, v1alpha1},
+    crd::catalog::{
+        TrinoCatalogName,
+        v1alpha2::{self, TrinoCatalogConnector},
+    },
 };
 
 #[derive(Clone, Debug)]
@@ -107,7 +110,7 @@ impl CatalogConfig {
 
     pub async fn from_catalog(
         catalog_name: &TrinoCatalogName,
-        catalog: &v1alpha1::TrinoCatalog,
+        catalog: &v1alpha2::TrinoCatalog,
         client: &Client,
         catalog_namespace: &NamespaceName,
     ) -> Result<CatalogConfig, FromTrinoCatalogError> {
@@ -144,7 +147,7 @@ impl CatalogConfig {
     }
 }
 
-fn calculate_env_name(catalog_name: &TrinoCatalogName, property: impl Into<String>) -> String {
+pub fn calculate_env_name(catalog_name: &TrinoCatalogName, property: impl Into<String>) -> String {
     let catalog = catalog_name.to_string().replace(['.', '-'], "_");
     let property = property.into().replace(['.', '-'], "_");
     format!("CATALOG_{catalog}_{property}").to_uppercase()

rust/operator-binary/src/catalog/iceberg.rs

@@ -1,9 +1,23 @@
 use async_trait::async_trait;
-use stackable_operator::{client::Client, v2::types::kubernetes::NamespaceName};
+use stackable_operator::{
+    client::Client,
+    k8s_openapi::api::core::v1::{EnvVar, EnvVarSource, SecretKeySelector},
+    v2::types::kubernetes::NamespaceName,
+};
 
 use crate::{
-    catalog::{ExtendCatalogConfig, FromTrinoCatalogError, ToCatalogConfig, config::CatalogConfig},
-    crd::catalog::{TrinoCatalogName, iceberg::IcebergConnector},
+    catalog::{
+        ExtendCatalogConfig, FromTrinoCatalogError, ToCatalogConfig,
+        config::{CatalogConfig, calculate_env_name},
+    },
+    crd::catalog::{
+        TrinoCatalogName,
+        iceberg::{
+            IcebergCatalogConnection, IcebergRestCatalogConnection,
+            IcebergRestCatalogOAuthCredential, IcebergRestCatalogSecurity,
+        },
+        v1alpha2::IcebergConnector,
+    },
 };
 
 pub const CONNECTOR_NAME: &str = "iceberg";
@@ -25,10 +39,20 @@ impl ToCatalogConfig for IcebergConnector {
         // See https://trino.io/docs/current/connector/iceberg.html
         config.add_property("iceberg.security", "allow-all");
 
-        if let Some(metastore) = &self.metastore {
-            metastore
-                .extend_catalog_config(&mut config, catalog_name, catalog_namespace, client)
-                .await?;
+        match &self.catalog {
+            IcebergCatalogConnection::Rest(iceberg_rest_catalog_connection) => {
+                iceberg_rest_catalog_connection
+                    .extend_catalog_config(&mut config, catalog_name, catalog_namespace, client)
+                    .await?;
+            }
+            IcebergCatalogConnection::HiveMetastore(hive_metastore_connection) => {
+                hive_metastore_connection
+                    .extend_catalog_config(&mut config, catalog_name, catalog_namespace, client)
+                    .await?;
+            }
+            IcebergCatalogConnection::UserProvided {} => {
+                // Nothing to do, the user will set it up
+            }
         }
 
         if let Some(ref s3) = self.s3 {
@@ -44,3 +68,86 @@ impl ToCatalogConfig for IcebergConnector {
         Ok(config)
     }
 }
+
+#[async_trait]
+impl ExtendCatalogConfig for IcebergRestCatalogConnection {
+    async fn extend_catalog_config(
+        &self,
+        catalog_config: &mut CatalogConfig,
+        catalog_name: &TrinoCatalogName,
+        _catalog_namespace: &NamespaceName,
+        _client: &Client,
+    ) -> Result<(), FromTrinoCatalogError> {
+        catalog_config.add_property("iceberg.catalog.type", "rest");
+        catalog_config.add_property("iceberg.rest-catalog.uri", self.uri.as_str());
+
+        // We explicitly use a match here to catch further additions
+        match &self.security {
+            IcebergRestCatalogSecurity::None {} => {
+                catalog_config.add_property("iceberg.rest-catalog.security", "NONE");
+            }
+            IcebergRestCatalogSecurity::OAuth2 {
+                server_uri,
+                credential,
+            } => {
+                catalog_config.add_property("iceberg.rest-catalog.security", "OAUTH2");
+                catalog_config.add_property(
+                    "iceberg.rest-catalog.oauth2.server-uri",
+                    server_uri.as_str(),
+                );
+                match credential {
+                    IcebergRestCatalogOAuthCredential::TokenSecretName(secret_name) => {
+                        // We can't use `add_env_property_from_secret`, as we need to concatenate
+                        // user and password in the Trino configuration. So instead we come up with
+                        // our own envs and bind them.
+                        let property = "iceberg.rest-catalog.oauth2.credential";
+                        let base_env_name = calculate_env_name(catalog_name, property);
+                        let username_env_name = format!("{base_env_name}_USERNAME");
+                        let password_env_name = format!("{base_env_name}_PASSWORD");
+                        catalog_config.add_property(
+                            property,
+                            format!("${{ENV:{username_env_name}}}:${{ENV:{password_env_name}}}"),
+                        );
+
+                        catalog_config.env_bindings.push(EnvVar {
+                            name: username_env_name,
+                            value_from: Some(EnvVarSource {
+                                secret_key_ref: Some(SecretKeySelector {
+                                    name: secret_name.to_owned(),
+                                    key: "username".to_owned(),
+                                    ..Default::default()
+                                }),
+                                ..Default::default()
+                            }),
+                            ..Default::default()
+                        });
+                        catalog_config.env_bindings.push(EnvVar {
+                            name: password_env_name,
+                            value_from: Some(EnvVarSource {
+                                secret_key_ref: Some(SecretKeySelector {
+                                    name: secret_name.to_owned(),
+                                    key: "password".to_owned(),
+                                    ..Default::default()
+                                }),
+                                ..Default::default()
+                            }),
+                            ..Default::default()
+                        });
+                    }
+                    IcebergRestCatalogOAuthCredential::CredentialSecretName(secret_name) => {
+                        catalog_config.add_env_property_from_secret(
+                            "iceberg.rest-catalog.oauth2.token",
+                            SecretKeySelector {
+                                name: secret_name.to_owned(),
+                                key: "token".to_owned(),
+                                ..Default::default()
+                            },
+                        )
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+}

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

@@ -49,7 +49,7 @@ pub enum Error {
     #[snafu(display("failed to parse {catalog}"))]
     ParseCatalog {
         source: FromTrinoCatalogError,
-        catalog: ObjectRef<catalog::v1alpha1::TrinoCatalog>,
+        catalog: ObjectRef<catalog::v1alpha2::TrinoCatalog>,
     },
 
     #[snafu(display("failed to configure fault tolerant execution"))]
@@ -77,7 +77,7 @@ type Result<T, E = Error> = std::result::Result<T, E>;
 /// Kubernetes objects referenced from the TrinoCluster spec, fetched from the cluster.
 pub struct DereferencedObjects {
     pub resolved_authentication_classes: Vec<ResolvedAuthenticationClassRef>,
-    pub catalog_definitions: Vec<catalog::v1alpha1::TrinoCatalog>,
+    pub catalog_definitions: Vec<catalog::v1alpha2::TrinoCatalog>,
     pub catalogs: Vec<CatalogConfig>,
     pub trino_opa_config: Option<TrinoOpaConfig>,
     pub resolved_fte_config: Option<ResolvedFaultTolerantExecutionConfig>,
@@ -97,7 +97,7 @@ pub async fn dereference(
             .context(AuthenticationClassRetrievalSnafu)?;
 
     let catalog_definitions = client
-        .list_with_label_selector::<catalog::v1alpha1::TrinoCatalog>(
+        .list_with_label_selector::<catalog::v1alpha2::TrinoCatalog>(
             namespace.as_ref(),
             &trino.spec.cluster_config.catalog_label_selector,
         )
@@ -168,12 +168,12 @@ pub async fn dereference(
 /// Determines the Trino catalog name based on user settings and the Kubernetes TrinoCatalog object
 /// name.
 fn determine_catalog_name(
-    catalog_name_spec: &catalog::v1alpha1::TrinoCatalogNameSpec,
+    catalog_name_spec: &catalog::v1alpha2::TrinoCatalogNameSpec,
     catalog_object_name: &str,
 ) -> Result<TrinoCatalogName> {
     // A `match` is used because we might support other ways of naming (e.g. custom) later.
     match catalog_name_spec {
-        catalog::v1alpha1::TrinoCatalogNameSpec::Inferred {
+        catalog::v1alpha2::TrinoCatalogNameSpec::Inferred {
             replace_hyphens_with_underscores,
         } => {
             let mut catalog_name = catalog_object_name.to_owned();
@@ -194,7 +194,7 @@ mod tests {
     fn determine_catalog_name_replaces_hyphens() {
         let inferred = |replace_hyphens_with_underscores| {
             determine_catalog_name(
-                &catalog::v1alpha1::TrinoCatalogNameSpec::Inferred {
+                &catalog::v1alpha2::TrinoCatalogNameSpec::Inferred {
                     replace_hyphens_with_underscores,
                 },
                 "my-postgres",

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

@@ -375,7 +375,7 @@ pub(crate) fn merged_role_group_config(
     trino: &v1alpha1::TrinoCluster,
     trino_role: &TrinoRole,
     role_group: &str,
-    trino_catalogs: &[crate::crd::catalog::v1alpha1::TrinoCatalog],
+    trino_catalogs: &[crate::crd::catalog::v1alpha2::TrinoCatalog],
 ) -> TrinoRoleGroupConfig {
     // The shared test clusters do not enable the Vector agent, so no aggregator ConfigMap name is
     // required here.