extend the EntityOperator, CruiseControl and KafkaExporter to support PDB

CHANGELOG.md

@@ -10,6 +10,7 @@
 * Added support for Strimzi Metrics Reporter to Kafka Connect, Mirror Maker 2 and Kafka Bridge.
 * Add new feature gate `ServerSideApplyPhase1` (disabled by default) that adds support for Server Side Apply for `ConfigMap`, `Ingress`, `PVC`, `Service`, and `ServiceAccount` according to [Strimzi Proposal #105](https://github.com/strimzi/proposals/blob/main/105-server-side-apply-implementation-fg-timelines.md).
 * Added distinction between changes of "cluster-wide" broker properties applied dynamically at cluster level, and "per-broker" broker properties applied dynamically at broker level.
+* Extend the EntityOperator, Cruise Control and KafkaExporter deployment to support PDB via the template section in the CR spec.
 
 ### Major changes, deprecations and removals
 

api/src/main/java/io/strimzi/api/kafka/model/kafka/entityoperator/EntityOperatorTemplate.java

@@ -11,6 +11,7 @@
 import io.strimzi.api.kafka.model.common.UnknownPropertyPreserving;
 import io.strimzi.api.kafka.model.common.template.ContainerTemplate;
 import io.strimzi.api.kafka.model.common.template.DeploymentTemplate;
+import io.strimzi.api.kafka.model.common.template.PodDisruptionBudgetTemplate;
 import io.strimzi.api.kafka.model.common.template.PodTemplate;
 import io.strimzi.api.kafka.model.common.template.ResourceTemplate;
 import io.strimzi.crdgenerator.annotations.Description;
@@ -29,7 +30,7 @@
         builderPackage = Constants.FABRIC8_KUBERNETES_API
 )
 @JsonInclude(JsonInclude.Include.NON_NULL)
-@JsonPropertyOrder({"deployment", "pod", "topicOperatorContainer", "userOperatorContainer", "tlsSidecarContainer", "serviceAccount", "entityOperatorRole", "topicOperatorRoleBinding", "userOperatorRoleBinding"})
+@JsonPropertyOrder({"deployment", "pod", "topicOperatorContainer", "userOperatorContainer", "tlsSidecarContainer", "serviceAccount", "podDisruptionBudget", "entityOperatorRole", "topicOperatorRoleBinding", "userOperatorRoleBinding"})
 @EqualsAndHashCode
 @ToString
 public class EntityOperatorTemplate implements UnknownPropertyPreserving {
@@ -42,6 +43,7 @@ public class EntityOperatorTemplate implements UnknownPropertyPreserving {
     private ContainerTemplate userOperatorContainer;
     private ContainerTemplate tlsSidecarContainer;
     private ResourceTemplate serviceAccount;
+    private PodDisruptionBudgetTemplate podDisruptionBudget;
     private Map<String, Object> additionalProperties;
 
     @Description("Template for Entity Operator `Deployment`.")
@@ -136,6 +138,15 @@ public void setServiceAccount(ResourceTemplate serviceAccount) {
         this.serviceAccount = serviceAccount;
     }
 
+    @Description("Template for the Entity Operator Pod Disruption Budget.")
+    @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    public PodDisruptionBudgetTemplate getPodDisruptionBudget() {
+        return podDisruptionBudget;
+    }
+    public void setPodDisruptionBudget(PodDisruptionBudgetTemplate podDisruptionBudget) {
+        this.podDisruptionBudget = podDisruptionBudget;
+    }
+
     @Override
     public Map<String, Object> getAdditionalProperties() {
         return this.additionalProperties != null ? this.additionalProperties : Map.of();

api/src/main/java/io/strimzi/api/kafka/model/kafka/exporter/KafkaExporterTemplate.java

@@ -11,6 +11,7 @@
 import io.strimzi.api.kafka.model.common.UnknownPropertyPreserving;
 import io.strimzi.api.kafka.model.common.template.ContainerTemplate;
 import io.strimzi.api.kafka.model.common.template.DeploymentTemplate;
+import io.strimzi.api.kafka.model.common.template.PodDisruptionBudgetTemplate;
 import io.strimzi.api.kafka.model.common.template.PodTemplate;
 import io.strimzi.api.kafka.model.common.template.ResourceTemplate;
 import io.strimzi.crdgenerator.annotations.Description;
@@ -30,7 +31,7 @@
         builderPackage = Constants.FABRIC8_KUBERNETES_API
 )
 @JsonInclude(JsonInclude.Include.NON_NULL)
-@JsonPropertyOrder({"deployment", "pod", "service", "container", "serviceAccount"})
+@JsonPropertyOrder({"deployment", "pod", "service", "container", "serviceAccount", "podDisruptionBudget"})
 @EqualsAndHashCode
 @ToString
 public class KafkaExporterTemplate implements UnknownPropertyPreserving {
@@ -39,6 +40,7 @@ public class KafkaExporterTemplate implements UnknownPropertyPreserving {
     private ResourceTemplate service;
     private ContainerTemplate container;
     private ResourceTemplate serviceAccount;
+    private PodDisruptionBudgetTemplate podDisruptionBudget;
     private Map<String, Object> additionalProperties;
 
     @Description("Template for Kafka Exporter `Deployment`.")
@@ -94,6 +96,15 @@ public void setServiceAccount(ResourceTemplate serviceAccount) {
         this.serviceAccount = serviceAccount;
     }
 
+    @Description("Template for the Pod Disruption Budget for Kafka Exporter pods.")
+    @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    public PodDisruptionBudgetTemplate getPodDisruptionBudget() {
+        return podDisruptionBudget;
+    }
+    public void setPodDisruptionBudget(PodDisruptionBudgetTemplate podDisruptionBudget) {
+        this.podDisruptionBudget = podDisruptionBudget;
+    }
+
     @Override
     public Map<String, Object> getAdditionalProperties() {
         return this.additionalProperties != null ? this.additionalProperties : Map.of();
@@ -106,4 +117,5 @@ public void setAdditionalProperty(String name, Object value) {
         }
         this.additionalProperties.put(name, value);
     }
+
 }

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/CruiseControl.java

@@ -20,11 +20,13 @@
 import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicy;
 import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyIngressRule;
 import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyPeer;
+import io.fabric8.kubernetes.api.model.policy.v1.PodDisruptionBudget;
 import io.strimzi.api.kafka.model.common.JvmOptions;
 import io.strimzi.api.kafka.model.common.metrics.JmxPrometheusExporterMetrics;
 import io.strimzi.api.kafka.model.common.metrics.StrimziMetricsReporter;
 import io.strimzi.api.kafka.model.common.template.DeploymentTemplate;
 import io.strimzi.api.kafka.model.common.template.InternalServiceTemplate;
+import io.strimzi.api.kafka.model.common.template.PodDisruptionBudgetTemplate;
 import io.strimzi.api.kafka.model.common.template.PodTemplate;
 import io.strimzi.api.kafka.model.kafka.Kafka;
 import io.strimzi.api.kafka.model.kafka.KafkaClusterSpec;
@@ -140,6 +142,7 @@ public class CruiseControl extends AbstractModel implements SupportsMetrics, Sup
     private DeploymentTemplate templateDeployment;
     private PodTemplate templatePod;
     private InternalServiceTemplate templateService;
+    private PodDisruptionBudgetTemplate templatePodDisruptionBudget;
 
     private static final Map<String, String> DEFAULT_POD_LABELS = new HashMap<>();
     static {
@@ -235,6 +238,7 @@ public static CruiseControl fromCrd(
                 result.templateService = template.getApiService();
                 result.templateServiceAccount = template.getServiceAccount();
                 result.templateContainer = template.getCruiseControlContainer();
+                result.templatePodDisruptionBudget = template.getPodDisruptionBudget();
             }
 
             return result;
@@ -542,4 +546,19 @@ public ConfigMap generateConfigMap(MetricsAndLogging metricsAndLogging) {
                         configMapData
                 );
     }
+    /**
+     * Generates the PodDisruptionBudget for Cruise Control
+     *
+     * @return The PodDisruptionBudget for Cruise Control
+     */
+    public PodDisruptionBudget generatePodDisruptionBudget() {
+        return PodDisruptionBudgetUtils.createCustomControllerPodDisruptionBudget(
+                componentName,
+                namespace,
+                labels,
+                ownerReference,
+                templatePodDisruptionBudget,
+                1
+        );
+    }
 }

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/EntityOperator.java

@@ -13,11 +13,13 @@
 import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicy;
 import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyIngressRule;
+import io.fabric8.kubernetes.api.model.policy.v1.PodDisruptionBudget;
 import io.fabric8.kubernetes.api.model.rbac.ClusterRole;
 import io.fabric8.kubernetes.api.model.rbac.PolicyRule;
 import io.fabric8.kubernetes.api.model.rbac.Role;
 import io.strimzi.api.kafka.model.common.Probe;
 import io.strimzi.api.kafka.model.common.template.DeploymentTemplate;
+import io.strimzi.api.kafka.model.common.template.PodDisruptionBudgetTemplate;
 import io.strimzi.api.kafka.model.common.template.PodTemplate;
 import io.strimzi.api.kafka.model.common.template.ResourceTemplate;
 import io.strimzi.api.kafka.model.kafka.Kafka;
@@ -81,6 +83,7 @@ public class EntityOperator extends AbstractModel {
     private ResourceTemplate templateRole;
     private DeploymentTemplate templateDeployment;
     private PodTemplate templatePod;
+    private PodDisruptionBudgetTemplate templatePodDisruptionBudget;
 
     private static final Map<String, String> DEFAULT_POD_LABELS = new HashMap<>();
     static {
@@ -135,6 +138,7 @@ public static EntityOperator fromCrd(Reconciliation reconciliation,
                 result.templateDeployment = template.getDeployment();
                 result.templatePod = template.getPod();
                 result.templateServiceAccount = template.getServiceAccount();
+                result.templatePodDisruptionBudget = template.getPodDisruptionBudget();
 
                 if (topicOperator != null) {
                     topicOperator.templateContainer = template.getTopicOperatorContainer();
@@ -305,4 +309,19 @@ public NetworkPolicy generateNetworkPolicy() {
                 rules
         );
     }
+    /**
+     * Generates the PodDisruptionBudget for the Entity Operator
+     *
+     * @return The PodDisruptionBudget for the Entity Operator
+     */
+    public PodDisruptionBudget generatePodDisruptionBudget() {
+        return PodDisruptionBudgetUtils.createCustomControllerPodDisruptionBudget(
+                componentName,
+                namespace,
+                labels,
+                ownerReference,
+                templatePodDisruptionBudget,
+                1
+        );
+    }
 }

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaExporter.java

@@ -15,9 +15,11 @@
 import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicy;
 import io.fabric8.kubernetes.api.model.networking.v1.NetworkPolicyIngressRule;
+import io.fabric8.kubernetes.api.model.policy.v1.PodDisruptionBudget;
 import io.strimzi.api.kafka.model.common.Probe;
 import io.strimzi.api.kafka.model.common.ProbeBuilder;
 import io.strimzi.api.kafka.model.common.template.DeploymentTemplate;
+import io.strimzi.api.kafka.model.common.template.PodDisruptionBudgetTemplate;
 import io.strimzi.api.kafka.model.common.template.PodTemplate;
 import io.strimzi.api.kafka.model.kafka.Kafka;
 import io.strimzi.api.kafka.model.kafka.KafkaClusterSpec;
@@ -83,6 +85,7 @@ public class KafkaExporter extends AbstractModel {
 
     private DeploymentTemplate templateDeployment;
     private PodTemplate templatePod;
+    private PodDisruptionBudgetTemplate templatePodDisruptionBudget;
     private static final Map<String, String> DEFAULT_POD_LABELS = new HashMap<>();
     static {
         String value = System.getenv(CO_ENV_VAR_CUSTOM_KAFKA_EXPORTER_POD_LABELS);
@@ -148,6 +151,7 @@ public static KafkaExporter fromCrd(Reconciliation reconciliation, Kafka kafkaAs
                 result.templatePod = template.getPod();
                 result.templateServiceAccount = template.getServiceAccount();
                 result.templateContainer = template.getContainer();
+                result.templatePodDisruptionBudget = template.getPodDisruptionBudget();
             }
 
             result.version = versions.supportedVersion(kafkaAssembly.getSpec().getKafka().getVersion()).version();
@@ -313,4 +317,19 @@ public NetworkPolicy generateNetworkPolicy() {
                 rules
         );
     }
+    /**
+     * Generates the PodDisruptionBudget for Kafka Exporter
+     *
+     * @return The PodDisruptionBudget.
+     */
+    public PodDisruptionBudget generatePodDisruptionBudget() {
+        return PodDisruptionBudgetUtils.createCustomControllerPodDisruptionBudget(
+                componentName,
+                namespace,
+                labels,
+                ownerReference,
+                templatePodDisruptionBudget,
+                1
+        );
+    }
 }

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/CruiseControlReconciler.java

@@ -24,6 +24,7 @@
 import io.strimzi.operator.cluster.operator.resource.kubernetes.ConfigMapOperator;
 import io.strimzi.operator.cluster.operator.resource.kubernetes.DeploymentOperator;
 import io.strimzi.operator.cluster.operator.resource.kubernetes.NetworkPolicyOperator;
+import io.strimzi.operator.cluster.operator.resource.kubernetes.PodDisruptionBudgetOperator;
 import io.strimzi.operator.cluster.operator.resource.kubernetes.SecretOperator;
 import io.strimzi.operator.cluster.operator.resource.kubernetes.ServiceAccountOperator;
 import io.strimzi.operator.cluster.operator.resource.kubernetes.ServiceOperator;
@@ -63,6 +64,9 @@ public class CruiseControlReconciler {
     private final NetworkPolicyOperator networkPolicyOperator;
     private final ConfigMapOperator configMapOperator;
     private final PasswordGenerator passwordGenerator;
+    private final boolean isPodDisruptionBudgetGeneration;
+    private final PodDisruptionBudgetOperator podDisruptionBudgetOperator;
+
 
     private String certificateHash = "";
     private String serverConfigurationHash = "";
@@ -114,6 +118,8 @@ public CruiseControlReconciler(
         this.serviceOperator = supplier.serviceOperations;
         this.networkPolicyOperator = supplier.networkPolicyOperator;
         this.configMapOperator = supplier.configMapOperations;
+        this.isPodDisruptionBudgetGeneration = config.isPodDisruptionBudgetGeneration();
+        this.podDisruptionBudgetOperator = supplier.podDisruptionBudgetOperator;
     }
 
     /**
@@ -135,6 +141,7 @@ public Future<Void> reconcile(boolean isOpenShift, ImagePullPolicy imagePullPoli
                 .compose(i -> certificatesSecret(clock))
                 .compose(i -> apiSecret())
                 .compose(i -> service())
+                .compose(i -> podDisruptionBudget())
                 .compose(i -> deployment(isOpenShift, imagePullPolicy, imagePullSecrets))
                 .compose(i -> waitForDeploymentReadiness());
     }
@@ -173,7 +180,24 @@ protected Future<Void> serviceAccount() {
                         cruiseControl != null ? cruiseControl.generateServiceAccount() : null
                 ).mapEmpty();
     }
-
+    /**
+     * Manages the Cruise Control Pod Disruption Budget
+     *
+     * @return  Future which completes when the reconciliation is done
+     */
+    protected Future<Void> podDisruptionBudget() {
+        if (isPodDisruptionBudgetGeneration) {
+            return podDisruptionBudgetOperator
+                    .reconcile(
+                            reconciliation,
+                            reconciliation.namespace(),
+                            CruiseControlResources.componentName(reconciliation.name()),
+                            cruiseControl != null ? cruiseControl.generatePodDisruptionBudget() : null
+                    ).mapEmpty();
+        } else {
+            return Future.succeededFuture();
+        }
+    }
     /**
      * Manages the Cruise Control ConfigMap which contains the following:
      * (1) Cruise Control server configuration

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/EntityOperatorReconciler.java

@@ -19,6 +19,7 @@
 import io.strimzi.operator.cluster.operator.resource.kubernetes.ConfigMapOperator;
 import io.strimzi.operator.cluster.operator.resource.kubernetes.DeploymentOperator;
 import io.strimzi.operator.cluster.operator.resource.kubernetes.NetworkPolicyOperator;
+import io.strimzi.operator.cluster.operator.resource.kubernetes.PodDisruptionBudgetOperator;
 import io.strimzi.operator.cluster.operator.resource.kubernetes.RoleBindingOperator;
 import io.strimzi.operator.cluster.operator.resource.kubernetes.RoleOperator;
 import io.strimzi.operator.cluster.operator.resource.kubernetes.SecretOperator;
@@ -50,11 +51,13 @@ public class EntityOperatorReconciler {
     private final SecretOperator secretOperator;
     private final ServiceAccountOperator serviceAccountOperator;
     private final boolean isNetworkPolicyGeneration;
+    private final boolean isPodDisruptionBudgetGeneration;
     private final RoleOperator roleOperator;
     private final RoleBindingOperator roleBindingOperator;
     private final ConfigMapOperator configMapOperator;
     private final NetworkPolicyOperator networkPolicyOperator;
     private final boolean isCruiseControlEnabled;
+    private final PodDisruptionBudgetOperator podDistruptionBudgetOperator;
 
     private String toCertificateHash = "";
     private String uoCertificateHash = "";
@@ -83,6 +86,7 @@ public EntityOperatorReconciler(
         this.maintenanceWindows = kafkaAssembly.getSpec().getMaintenanceTimeWindows();
         this.isNetworkPolicyGeneration = config.isNetworkPolicyGeneration();
         this.isCruiseControlEnabled = kafkaAssembly.getSpec().getCruiseControl() != null;
+        this.isPodDisruptionBudgetGeneration = config.isPodDisruptionBudgetGeneration();
 
         this.deploymentOperator = supplier.deploymentOperations;
         this.secretOperator = supplier.secretOperations;
@@ -91,6 +95,7 @@ public EntityOperatorReconciler(
         this.roleBindingOperator = supplier.roleBindingOperations;
         this.configMapOperator = supplier.configMapOperations;
         this.networkPolicyOperator = supplier.networkPolicyOperator;
+        this.podDistruptionBudgetOperator = supplier.podDisruptionBudgetOperator;
     }
 
     /**
@@ -111,6 +116,7 @@ public Future<Void> reconcile(boolean isOpenShift, ImagePullPolicy imagePullPoli
                 .compose(i -> topicOperatorRole())
                 .compose(i -> userOperatorRole())
                 .compose(i -> networkPolicy())
+                .compose(i -> podDistruptionBudget())
                 .compose(i -> topicOperatorRoleBindings())
                 .compose(i -> userOperatorRoleBindings())
                 .compose(i -> topicOperatorConfigMap())
@@ -416,7 +422,24 @@ protected Future<Void> networkPolicy() {
             return Future.succeededFuture();
         }
     }
-
+    /**
+     * Manages the Entity Operator Pod Disruption Budget
+     *
+     * @return  Future which completes when the reconciliation is done
+     */
+    protected Future<Void> podDistruptionBudget() {
+        if (isPodDisruptionBudgetGeneration) {
+            return podDistruptionBudgetOperator
+                    .reconcile(
+                            reconciliation,
+                            reconciliation.namespace(),
+                            KafkaResources.entityOperatorDeploymentName(reconciliation.name()),
+                            entityOperator != null ? entityOperator.generatePodDisruptionBudget() : null
+                    ).mapEmpty();
+        } else {
+            return Future.succeededFuture();
+        }
+    }
     /**
      * Manages the Entity Operator Deployment.
      *