feat: Add affinity to OPA Pods

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
 

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)

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)

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 {

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())),