diff --git a/CHANGELOG.md b/CHANGELOG.md index 56cd009a9..251b59ee7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file. ## [Unreleased] +### Added + +- Coordinators now have a default affinity to the OPA Pods when OPA authorization is configured ([#923]). + ### Changed - Internal operator refactoring: introduce a build() step in the reconciler that @@ -17,6 +21,7 @@ All notable changes to this project will be documented in this file. [#909]: https://github.com/stackabletech/trino-operator/pull/909 [#913]: https://github.com/stackabletech/trino-operator/pull/913 [#918]: https://github.com/stackabletech/trino-operator/pull/918 +[#923]: https://github.com/stackabletech/trino-operator/pull/923 ## [26.7.0] - 2026-07-21 diff --git a/docs/modules/trino/pages/usage-guide/operations/pod-placement.adoc b/docs/modules/trino/pages/usage-guide/operations/pod-placement.adoc index bb6366cbd..61d326574 100644 --- a/docs/modules/trino/pages/usage-guide/operations/pod-placement.adoc +++ b/docs/modules/trino/pages/usage-guide/operations/pod-placement.adoc @@ -7,7 +7,12 @@ The default affinities created by the operator are: 1. Co-locate all the Trino Pods (weight 20) 2. Distribute all Pods within the same role (coordinators, workers) (weight 70) +In case OPA authorization is configured the following affinity is added + +1. Co-locate the coordinator with the OPA Pods (weight 50). + This basically has no effect in case the OpaCluster is deployed as a DaemonSet, but makes sense for a Deployment, as Trino does many calls to OPA. + Additionally the operator looks through every `TrinoCatalog` you configure and sets up the following affinities: -1. Hive + Iceberg connector: Co-locate the coordinators with the hive metastores (weight 50) +1. Hive + Iceberg connector: Co-locate the coordinator with the hive metastores (weight 50) 2. Hive + Iceberg connector: Co-locate the workers with the hdfs datanodes (if hdfs is used) (weight 50) diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index ee84bdcea..5b2ee4423 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -237,6 +237,7 @@ pub fn validate( &trino.name_any(), &trino_role, &dereferenced_objects.catalog_definitions, + trino.get_opa_config(), ); let mut groups = BTreeMap::new(); for (rg_name, rg) in &role.role_groups { @@ -349,8 +350,12 @@ pub(crate) fn merged_role_group_config( trino_catalogs: &[crate::crd::catalog::v1alpha1::TrinoCatalog], ) -> TrinoRoleGroupConfig { let role = trino.role(trino_role); - let default_config = - v1alpha1::TrinoConfig::default_config(&trino.name_any(), trino_role, trino_catalogs); + let default_config = v1alpha1::TrinoConfig::default_config( + &trino.name_any(), + trino_role, + trino_catalogs, + trino.get_opa_config(), + ); let rg = role .role_groups .get(role_group) diff --git a/rust/operator-binary/src/crd/affinity.rs b/rust/operator-binary/src/crd/affinity.rs index b993b3610..fdcb9f00f 100644 --- a/rust/operator-binary/src/crd/affinity.rs +++ b/rust/operator-binary/src/crd/affinity.rs @@ -8,12 +8,14 @@ use stackable_operator::{ use crate::crd::{ APP_NAME, TrinoRole, catalog::{self, TrinoCatalogConnector}, + v1alpha1, }; pub fn get_affinity( cluster_name: &str, role: &TrinoRole, trino_catalogs: &[catalog::v1alpha1::TrinoCatalog], + opa_config: Option<&v1alpha1::TrinoAuthorizationOpaConfig>, ) -> StackableAffinityFragment { let affinity_between_cluster_pods = affinity_between_cluster_pods(APP_NAME, cluster_name, 20); let mut affinities = vec![affinity_between_cluster_pods]; @@ -75,6 +77,20 @@ pub fn get_affinity( .collect(), }; affinities.extend(additional_affinities); + + // Only the coordinator talks to OPA (it does the authorization checks for the whole cluster), + // so we only co-locate the coordinator with the OPA Pods. + if let Some(opa_config) = opa_config + && role == &TrinoRole::Coordinator + { + affinities.push(affinity_between_role_pods( + "opa", + &opa_config.opa.config_map_name, // The discovery cm has the same name as the OpaCluster itself + "server", + 50, + )); + } + StackableAffinityFragment { pod_affinity: Some(PodAffinity { preferred_during_scheduling_ignored_during_execution: Some(affinities), @@ -104,6 +120,7 @@ mod tests { }, apimachinery::pkg::apis::meta::v1::LabelSelector, }, + utils::yaml_from_str_singleton_map, }; use super::*; @@ -200,7 +217,7 @@ mod tests { #[rstest] #[case(TrinoRole::Coordinator)] #[case(TrinoRole::Worker)] - fn test_hms_and_hdfs_affinity(#[case] role: TrinoRole) { + fn test_hms_hdfs_and_opa_affinity(#[case] role: TrinoRole) { let input = r#" apiVersion: trino.stackable.tech/v1alpha1 kind: TrinoCluster @@ -213,6 +230,10 @@ mod tests { catalogLabelSelector: matchLabels: trino: simple-trino + authorization: + opa: + configMapName: simple-opa + package: trino coordinators: roleGroups: default: @@ -223,7 +244,7 @@ mod tests { replicas: 1 "#; let trino: v1alpha1::TrinoCluster = - serde_yaml::from_str(input).expect("illegal test input"); + yaml_from_str_singleton_map(input).expect("illegal test input"); let input = r#" apiVersion: trino.stackable.tech/v1alpha1 @@ -240,9 +261,8 @@ mod tests { hdfs: configMap: simple-hdfs "#; - let deserializer = serde_yaml::Deserializer::from_str(input); let hive_catalog_1: catalog::v1alpha1::TrinoCatalog = - serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap(); + yaml_from_str_singleton_map(input).expect("illegal test input"); let input = r#" apiVersion: trino.stackable.tech/v1alpha1 @@ -255,9 +275,8 @@ mod tests { connector: tpch: {} "#; - let deserializer = serde_yaml::Deserializer::from_str(input); let tpch_catalog: catalog::v1alpha1::TrinoCatalog = - serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap(); + yaml_from_str_singleton_map(input).expect("illegal test input"); let input = r#" apiVersion: trino.stackable.tech/v1alpha1 @@ -274,9 +293,8 @@ mod tests { s3: reference: minio "#; - let deserializer = serde_yaml::Deserializer::from_str(input); let hive_catalog_2: catalog::v1alpha1::TrinoCatalog = - serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap(); + yaml_from_str_singleton_map(input).expect("illegal test input"); let merged_config = crate::controller::validate::merged_role_group_config( &trino, @@ -348,6 +366,27 @@ mod tests { }, weight: 50, }); + expected_affinities.push(WeightedPodAffinityTerm { + pod_affinity_term: PodAffinityTerm { + label_selector: Some(LabelSelector { + match_labels: Some(BTreeMap::from([ + ("app.kubernetes.io/name".to_string(), "opa".to_string()), + ( + "app.kubernetes.io/instance".to_string(), + "simple-opa".to_string(), + ), + ( + "app.kubernetes.io/component".to_string(), + "server".to_string(), + ), + ])), + ..LabelSelector::default() + }), + topology_key: "kubernetes.io/hostname".to_string(), + ..PodAffinityTerm::default() + }, + weight: 50, + }); } TrinoRole::Worker => { expected_affinities.push(WeightedPodAffinityTerm { diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 0a47bd41c..a6a4253ce 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -464,6 +464,7 @@ impl v1alpha1::TrinoConfig { cluster_name: &str, role: &TrinoRole, trino_catalogs: &[catalog::v1alpha1::TrinoCatalog], + opa_config: Option<&v1alpha1::TrinoAuthorizationOpaConfig>, ) -> v1alpha1::TrinoConfigFragment { let (cpu_min, cpu_max, memory) = match role { TrinoRole::Coordinator => ("500m", "2", "4Gi"), @@ -483,7 +484,7 @@ impl v1alpha1::TrinoConfig { v1alpha1::TrinoConfigFragment { logging: product_logging::spec::default_logging(), - affinity: get_affinity(cluster_name, role, trino_catalogs), + affinity: get_affinity(cluster_name, role, trino_catalogs, opa_config), resources: ResourcesFragment { cpu: CpuLimitsFragment { min: Some(Quantity(cpu_min.to_string())),