Add Kubernetes pod affinity support for activation job pods

src/aap_eda/api/serializers/activation.py

@@ -417,6 +417,7 @@ class Meta:
             "k8s_pod_annotations",
             "k8s_pod_node_selector",
             "k8s_pod_tolerations",
+            "k8s_pod_affinity",
         ]
         read_only_fields = [
             "id",
@@ -498,6 +499,7 @@ class Meta:
             "k8s_pod_annotations",
             "k8s_pod_node_selector",
             "k8s_pod_tolerations",
+            "k8s_pod_affinity",
         ]
         read_only_fields = [
             "id",
@@ -570,6 +572,7 @@ def to_representation(self, activation):
             "rule_engine_credential_id": activation.rule_engine_credential_id,
             **_activation_k8s_pod_metadata_payload(activation),
             "k8s_pod_tolerations": activation.k8s_pod_tolerations,
+            "k8s_pod_affinity": activation.k8s_pod_affinity,
         }
 
 
@@ -606,6 +609,7 @@ class Meta:
             "k8s_pod_annotations",
             "k8s_pod_node_selector",
             "k8s_pod_tolerations",
+            "k8s_pod_affinity",
         ]
 
     rulebook_id = serializers.IntegerField(
@@ -659,6 +663,11 @@ class Meta:
         default=list,
         validators=[validators.validate_k8s_pod_tolerations],
     )
+    k8s_pod_affinity = serializers.JSONField(
+        required=False,
+        default=dict,
+        validators=[validators.check_if_k8s_pod_affinity_valid],
+    )
 
     def validate(self, data):
         _validate_credentials_and_token_and_rulebook(data=data, creating=True)
@@ -736,6 +745,7 @@ def copy(self) -> dict:
         validators.validate_k8s_pod_tolerations(
             activation.k8s_pod_tolerations or []
         )
+        validators.check_if_k8s_pod_affinity_valid(activation.k8s_pod_affinity)
         validators.check_if_rulebook_exists(activation.rulebook_id)
 
         copied_data = {
@@ -765,6 +775,7 @@ def copy(self) -> dict:
             "rule_engine_credential_id": activation.rule_engine_credential_id,
             **pod_metadata,
             "k8s_pod_tolerations": activation.k8s_pod_tolerations,
+            "k8s_pod_affinity": activation.k8s_pod_affinity,
         }
         if activation.eda_system_vault_credential:
             inputs = yaml.safe_load(
@@ -814,6 +825,7 @@ class Meta:
             "k8s_pod_annotations",
             "k8s_pod_node_selector",
             "k8s_pod_tolerations",
+            "k8s_pod_affinity",
         ]
 
     rulebook_id = serializers.IntegerField(
@@ -859,28 +871,30 @@ class Meta:
         default=list,
         validators=[validators.validate_k8s_pod_tolerations],
     )
+    k8s_pod_affinity = serializers.JSONField(
+        required=False,
+        default=dict,
+        validators=[validators.check_if_k8s_pod_affinity_valid],
+    )
 
     def refill_needed_data(
         self, data: dict, activation: models.Activation
     ) -> None:
-        if "name" not in data:
-            data["name"] = activation.name
-        if "k8s_service_name" not in data:
-            data["k8s_service_name"] = activation.k8s_service_name
-        if "k8s_pod_service_account_name" not in data:
-            data[
-                "k8s_pod_service_account_name"
-            ] = activation.k8s_pod_service_account_name
-        if "k8s_pod_labels" not in data:
-            data["k8s_pod_labels"] = activation.k8s_pod_labels or {}
-        if "k8s_pod_annotations" not in data:
-            data["k8s_pod_annotations"] = activation.k8s_pod_annotations or {}
-        if "k8s_pod_node_selector" not in data:
-            data["k8s_pod_node_selector"] = (
-                activation.k8s_pod_node_selector or {}
-            )
-        if "k8s_pod_tolerations" not in data:
-            data["k8s_pod_tolerations"] = activation.k8s_pod_tolerations or []
+        default_field_values = {
+            "name": activation.name,
+            "k8s_service_name": activation.k8s_service_name,
+            "k8s_pod_service_account_name": (
+                activation.k8s_pod_service_account_name
+            ),
+            "k8s_pod_labels": activation.k8s_pod_labels or {},
+            "k8s_pod_annotations": activation.k8s_pod_annotations or {},
+            "k8s_pod_node_selector": activation.k8s_pod_node_selector or {},
+            "k8s_pod_tolerations": activation.k8s_pod_tolerations or [],
+            "k8s_pod_affinity": activation.k8s_pod_affinity or {},
+        }
+        for field_name, default_value in default_field_values.items():
+            if field_name not in data:
+                data[field_name] = default_value
         if "extra_var" not in data:
             data["extra_var"] = activation.extra_var
         data["extra_var"] = _get_user_extra_vars(activation, data["extra_var"])
@@ -1017,6 +1031,7 @@ def to_representation(self, activation):
             "enable_persistence": activation.enable_persistence,
             "rule_engine_credential_id": activation.rule_engine_credential_id,
             "k8s_pod_tolerations": activation.k8s_pod_tolerations,
+            "k8s_pod_affinity": activation.k8s_pod_affinity,
         }
 
 
@@ -1170,6 +1185,7 @@ class Meta:
             "rule_engine_credential_id",
             "rule_engine_credential",
             "k8s_pod_tolerations",
+            "k8s_pod_affinity",
         ]
         read_only_fields = [
             "id",
@@ -1314,6 +1330,7 @@ def to_representation(self, activation):
             "rule_engine_credential_id": activation.rule_engine_credential_id,
             "rule_engine_credential": rule_engine_credential,
             "k8s_pod_tolerations": activation.k8s_pod_tolerations,
+            "k8s_pod_affinity": activation.k8s_pod_affinity,
         }
 
 
@@ -1362,6 +1379,11 @@ class PostActivationSerializer(
         default=list,
         validators=[validators.validate_k8s_pod_tolerations],
     )
+    k8s_pod_affinity = serializers.JSONField(
+        required=False,
+        default=dict,
+        validators=[validators.check_if_k8s_pod_affinity_valid],
+    )
 
     def validate(self, data):
         _validate_credentials_and_token_and_rulebook(data=data, creating=False)
@@ -1392,6 +1414,7 @@ class Meta:
             "k8s_pod_annotations",
             "k8s_pod_node_selector",
             "k8s_pod_tolerations",
+            "k8s_pod_affinity",
             "source_mappings",
             "skip_audit_events",
             "enable_persistence",
@@ -1428,6 +1451,7 @@ def is_activation_valid(activation: models.Activation) -> tuple[bool, str]:
     data["rule_engine_credential_id"] = activation.rule_engine_credential_id
     data.update(_activation_k8s_pod_metadata_payload(activation))
     data["k8s_pod_tolerations"] = activation.k8s_pod_tolerations or []
+    data["k8s_pod_affinity"] = activation.k8s_pod_affinity or {}
     serializer = PostActivationSerializer(data=data)
 
     valid = serializer.is_valid()

src/aap_eda/core/migrations/0074_activation_k8s_pod_affinity.py

@@ -0,0 +1,23 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0073_activation_k8s_pod_tolerations"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="activation",
+            name="k8s_pod_affinity",
+            field=models.JSONField(
+                blank=True,
+                default=dict,
+                help_text=(
+                    "Kubernetes affinity rules (nodeAffinity, podAffinity, "
+                    "podAntiAffinity) applied to activation job pods for "
+                    "scheduling constraints."
+                ),
+            ),
+        ),
+    ]

src/aap_eda/core/models/activation.py

@@ -197,6 +197,16 @@ class Meta:
             "so they can be scheduled onto tainted nodes."
         ),
     )
+    k8s_pod_affinity = models.JSONField(
+        default=dict,
+        blank=True,
+        help_text=(
+            "Kubernetes affinity rules (nodeAffinity, podAffinity, "
+            "podAntiAffinity) applied to activation job pods for "
+            "scheduling constraints."
+        ),
+    )
+
     event_streams = models.ManyToManyField(
         EventStream, related_name="activations", default=None
     )

src/aap_eda/core/validators.py

@@ -411,6 +411,35 @@ def check_if_k8s_pod_node_selector_valid(value) -> None:
         _validate_label_value(k, v)
 
 
+_K8S_AFFINITY_TOP_LEVEL_KEYS = frozenset(
+    {"nodeAffinity", "podAffinity", "podAntiAffinity"}
+)
+
+
+def check_if_k8s_pod_affinity_valid(value) -> None:
+    """Validate affinity dict at a structural level only."""
+    if value in (None, {}):
+        return
+    if settings.DEPLOYMENT_TYPE != "k8s":
+        return
+    if not isinstance(value, dict):
+        raise serializers.ValidationError(
+            "k8s_pod_affinity must be a JSON object"
+        )
+    unknown = set(value.keys()) - _K8S_AFFINITY_TOP_LEVEL_KEYS
+    if unknown:
+        raise serializers.ValidationError(
+            f"k8s_pod_affinity has unknown top-level keys: "
+            f"{sorted(unknown)}. Allowed: "
+            f"{sorted(_K8S_AFFINITY_TOP_LEVEL_KEYS)}"
+        )
+    for key, sub_value in value.items():
+        if not isinstance(sub_value, dict):
+            raise serializers.ValidationError(
+                f"k8s_pod_affinity.{key} must be a JSON object"
+            )
+
+
 def check_credential_types(
     eda_credential_id: int,
     types: list[enums.DefaultCredentialType],

src/aap_eda/services/activation/engine/common.py

@@ -131,6 +131,7 @@ class ContainerRequest(BaseModel):
     k8s_pod_annotations: tp.Optional[dict] = None
     k8s_pod_node_selector: tp.Optional[dict] = None
     k8s_pod_tolerations: tp.Optional[list[dict]] = None
+    k8s_pod_affinity: tp.Optional[dict] = None
     k8s_mem_limit: tp.Optional[str] = None
     k8s_cpu_limit: tp.Optional[str] = None
     log_tracking_id: tp.Optional[str] = None
@@ -184,6 +185,7 @@ def get_container_request(self) -> ContainerRequest:
             k8s_pod_annotations=self.k8s_pod_annotations or {},
             k8s_pod_node_selector=self.k8s_pod_node_selector or {},
             k8s_pod_tolerations=self.k8s_pod_tolerations or [],
+            k8s_pod_affinity=self.k8s_pod_affinity or {},
             k8s_mem_limit=settings.K8S_MEM_LIMIT,
             k8s_cpu_limit=settings.K8S_CPU_LIMIT,
             log_tracking_id=self.log_tracking_id,

src/aap_eda/services/activation/engine/kubernetes.py

@@ -419,8 +419,19 @@ def _create_pod_template_spec(
                 for t in tolerations
             ]
 
-        spec = k8sclient.V1PodSpec(**spec_kwargs)
+        affinity = request.k8s_pod_affinity or {}
+
+        if affinity:
+            # affinity is passed through as the raw
+            # dict the user supplied (validated for shape only in
+            # core/validators.py). V1Affinity's nested structure
+            # (nodeAffinity/podAffinity/podAntiAffinity, each several
+            # levels deep) is accepted by the client's serializer as a
+            # plain dict without needing manual construction of the
+            # typed sub-objects confirmed against kubernetes==26.1.0
+            spec_kwargs["affinity"] = affinity
 
+        spec = k8sclient.V1PodSpec(**spec_kwargs)
         pod_template = k8sclient.V1PodTemplateSpec(
             spec=spec,
             metadata=k8sclient.V1ObjectMeta(**pod_meta),

tests/integration/api/test_activation.py

@@ -971,6 +971,7 @@ def assert_activation_base_data(
     assert data["created_at"] == activation.created_at
     assert data["modified_at"] <= activation.modified_at
     assert data["status_message"]
+    assert data["k8s_pod_affinity"] == (activation.k8s_pod_affinity or {})
 
 
 def assert_activation_related_object_fks(
@@ -1038,6 +1039,33 @@ def test_is_activation_valid(
     assert error == "{}"  # noqa P103
 
 
+@pytest.mark.django_db
+def test_is_activation_valid_with_k8s_pod_affinity(
+    default_activation: models.Activation, preseed_credential_types
+):
+    """Test that is_activation_valid succeeds when k8s_pod_affinity is set."""
+    default_activation.k8s_pod_affinity = {
+        "nodeAffinity": {
+            "requiredDuringSchedulingIgnoredDuringExecution": {
+                "nodeSelectorTerms": [
+                    {
+                        "matchExpressions": [
+                            {
+                                "key": "eda-lab/zone",
+                                "operator": "In",
+                                "values": ["a"],
+                            }
+                        ]
+                    }
+                ]
+            }
+        }
+    }
+    valid, error = is_activation_valid(default_activation)
+    assert valid is True
+    assert error == "{}"  # noqa P103
+
+
 @pytest.mark.django_db
 @patch(
     "aap_eda.api.views.activation.check_dispatcherd_workers_health",
@@ -1274,6 +1302,52 @@ def test_update_activation(
     assert activation.status == enums.ActivationStatus.PENDING
 
 
+@pytest.mark.django_db
+@patch("aap_eda.api.serializers.activation.settings.DEPLOYMENT_TYPE", "k8s")
+@patch(
+    "aap_eda.api.views.activation.check_dispatcherd_workers_health",
+    return_value=True,
+)
+def test_update_activation_k8s_pod_affinity(
+    mock_health_check,
+    activation_payload: Dict[str, Any],
+    default_rulebook: models.Rulebook,
+    admin_client: APIClient,
+):
+    """Test that k8s_pod_affinity can be set and read back via update."""
+    activation_payload["is_enabled"] = False
+    response = admin_client.post(
+        f"{api_url_v1}/activations/", data=activation_payload
+    )
+    assert response.status_code == status.HTTP_201_CREATED
+    id = response.data["id"]
+    affinity = {
+        "nodeAffinity": {
+            "requiredDuringSchedulingIgnoredDuringExecution": {
+                "nodeSelectorTerms": [
+                    {
+                        "matchExpressions": [
+                            {
+                                "key": "eda-lab/zone",
+                                "operator": "In",
+                                "values": ["a"],
+                            }
+                        ]
+                    }
+                ]
+            }
+        }
+    }
+    response = admin_client.patch(
+        f"{api_url_v1}/activations/{id}/",
+        data={"k8s_pod_affinity": affinity},
+    )
+    assert response.status_code == status.HTTP_200_OK
+    assert response.data["k8s_pod_affinity"] == affinity
+    activation = models.Activation.objects.get(id=id)
+    assert activation.k8s_pod_affinity == affinity
+
+
 @pytest.mark.django_db
 def test_update_activation_invalid_body(
     activation_payload: Dict[str, Any],