WIP: Add cert-manager support to Kafka Exporter

katheris · katheris · commit 32b9a7ccbf87 · 2026-01-28T10:15:37.000Z
Signed-off-by: Kate Stanley &lt;11195226+katheris@users.noreply.github.com&gt;
diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/CertManagerUtils.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/CertManagerUtils.java
@@ -104,6 +104,26 @@ public static Certificate buildCertManagerCertificate(String namespace,
      * Used when cert-manager has issued the CA Secret.
      *
      * @param clusterCa         The Cluster CA
+     * @param certManagerSecret cert-manager Secret containing the Cluster CA
+     * @param namespace         Namespace for the Secret
+     * @param secretName        Name of the Secret
+     * @param keyCertName       Key under which the certificate will be stored in the new Secret
+     * @param labels            Labels
+     * @param ownerReference    Owner reference
+     * @return Newly built Secret
+     */
+    public static Secret buildTrustedCertificateSecretFromCertManager(ClusterCa clusterCa, Secret certManagerSecret, String namespace,
+                                                                      String secretName, String keyCertName, Labels labels,
+                                                                      OwnerReference ownerReference) {
+        // TO, UO, KE etc do not track the clientsCa generation annotation on their Secret, so pass null
+        return buildTrustedCertificateSecretFromCertManager(clusterCa, null, certManagerSecret, namespace, secretName, keyCertName, labels, ownerReference);
+    }
+
+    /**
+     * Builds a certificate Secret based on the cert-manager provided Secret.
+     * Used when cert-manager has issued the CA Secret.
+     *
+     * @param clusterCa         The Cluster CA
      * @param clientsCa         The Clients CA. If this is not null the Clients CA generation is added
      * @param certManagerSecret cert-manager Secret containing the Cluster CA
      * @param namespace         Namespace for the Secret
diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaExporter.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaExporter.java
@@ -4,6 +4,7 @@
  */
 package io.strimzi.operator.cluster.model;
 
+import io.fabric8.certmanager.api.model.v1.Certificate;
 import io.fabric8.kubernetes.api.model.Container;
 import io.fabric8.kubernetes.api.model.ContainerPort;
 import io.fabric8.kubernetes.api.model.EnvVar;
@@ -33,13 +34,18 @@
 import io.strimzi.operator.cluster.model.securityprofiles.PodSecurityProviderContextImpl;
 import io.strimzi.operator.common.Reconciliation;
 import io.strimzi.operator.common.Util;
+import io.strimzi.operator.common.auth.PemTrustSet;
+import io.strimzi.operator.common.model.Ca;
 
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
 import static io.strimzi.api.kafka.model.common.template.DeploymentStrategy.ROLLING_UPDATE;
+import static io.strimzi.operator.common.model.Ca.x509Certificate;
 
 /**
  * Kafka Exporter model
@@ -280,9 +286,25 @@ private List<VolumeMount> getVolumeMounts() {
         return volumeList;
     }
 
+    /**
+     * Creates the Certificate resource for the Kafka Exporter used when cert-manager is issuing certificates
+     *
+     * @param clusterCa                 The CA for cluster certificates
+     *
+     * @return List of Certificate resources
+     */
+    public Certificate generateCertificateResource(ClusterCa clusterCa) {
+        return CertManagerUtils.buildCertManagerCertificate(namespace,
+                KafkaExporterResources.secretName(cluster),
+                clusterCa.getCertManagerCert(componentName, Ca.IO_STRIMZI),
+                labels,
+                ownerReference);
+    }
+
     /**
      * Generate the Secret containing the Kafka Exporter certificate signed by the cluster CA certificate used for TLS based
      * internal communication with Kafka. It also contains the related Kafka Exporter private key.
+     * Used when Strimzi is issuing certificates.
      *
      * @param clusterCa                             The cluster CA.
      * @param existingSecret                        The existing secret with Kafka certificates
@@ -291,11 +313,60 @@ private List<VolumeMount> getVolumeMounts() {
      *
      * @return The generated Secret.
      */
-    public Secret generateCertificatesSecret(ClusterCa clusterCa, Secret existingSecret, boolean isMaintenanceTimeWindowsSatisfied) {
+    public Secret generateCertificatesSecretForStrimziCa(ClusterCa clusterCa, Secret existingSecret, boolean isMaintenanceTimeWindowsSatisfied) {
         return CertUtils.buildTrustedCertificateSecret(reconciliation, clusterCa, existingSecret, namespace, KafkaExporterResources.secretName(cluster), componentName,
                 COMPONENT_TYPE, labels, ownerReference, isMaintenanceTimeWindowsSatisfied);
     }
 
+    /**
+     * Generate the Secret containing the Kafka Exporter certificate signed by the cluster CA certificate used for TLS based
+     * internal communication with Kafka. It also contains the related Kafka Exporter private key.
+     * Used when cert-manager is issuing certificates.
+     *
+     * @param clusterCa                             The cluster CA.
+     * @param existingSecret                        Existing Secret.
+     * @param certManagerSecret                     Secret managed by cert-manager, may be null.
+     * @param pemTrustSet                           Trust set to use to determine if new certificates are trusted
+     *
+     * @return The generated Secret.
+     */
+    public Secret generateCertificatesSecretForCertManagerCA(ClusterCa clusterCa, Secret existingSecret, Secret certManagerSecret, PemTrustSet pemTrustSet) {
+        Secret newSecret = CertManagerUtils.buildTrustedCertificateSecretFromCertManager(clusterCa, certManagerSecret, namespace, KafkaExporterResources.secretName(cluster),
+                COMPONENT_TYPE, labels, ownerReference);
+        if (existingSecret == null) {
+            return newSecret;
+        } else if (CertManagerUtils.certManagerCertUpdated(existingSecret, newSecret)) {
+            if (certManagerSecretNotTrusted(pemTrustSet, existingSecret)) {
+                LOGGER.infoCr(reconciliation, "New certificate for Kafka Exporter, but not trusted yet so keeping existing certificate Secret.");
+                return existingSecret;
+            } else {
+                LOGGER.infoCr(reconciliation, "New certificate for Kafka Exporter, updating Secret {}/{}", namespace, existingSecret.getMetadata().getName());
+                return newSecret;
+            }
+        } else {
+            // Certificate has no changed
+            return existingSecret;
+        }
+    }
+
+    /**
+     * Checks if the cert-manager Secret is trusted by the current CA cert
+     *
+     * @param certManagerSecret Secret containing cert-manager provided cert
+     * @return Whether the cert is trusted
+     */
+    private boolean certManagerSecretNotTrusted(PemTrustSet pemTrustSet, Secret certManagerSecret) {
+        X509Certificate x509CaCert;
+        X509Certificate certManagerCert;
+        try {
+            x509CaCert = x509Certificate(pemTrustSet.trustedCertificatesPemBytes());
+            certManagerCert = x509Certificate(Util.decodeBytesFromBase64(certManagerSecret.getData().get("tls.crt")));
+        } catch (CertificateException e) {
+            throw new RuntimeException(e);
+        }
+        return !CertUtils.certIsTrusted(reconciliation, certManagerCert, x509CaCert);
+    }
+
     /**
      * Generates the NetworkPolicies relevant for Kafka Exporter
      *
diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/CaReconciler.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/CaReconciler.java
@@ -430,7 +430,6 @@ Future<Void> reconcileClusterOperatorSecret(Clock clock, Secret certManagerSecre
                     if (clusterCaCertManagerType == CertificateManagerType.CERT_MANAGER_IO) {
                         Secret newCoSecret = CertManagerUtils.buildTrustedCertificateSecretFromCertManager(
                                 clusterCa,
-                                null, //we don't need the Clients CA generation present on this Secret
                                 certManagerSecret,
                                 reconciliation.namespace(),
                                 KafkaResources.clusterOperatorCertsSecretName(reconciliation.name()),
diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaExporterReconciler.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaExporterReconciler.java
@@ -7,15 +7,18 @@
 import io.fabric8.kubernetes.api.model.LocalObjectReference;
 import io.fabric8.kubernetes.api.model.Secret;
 import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.strimzi.api.kafka.model.common.CertificateManagerType;
 import io.strimzi.api.kafka.model.kafka.Kafka;
 import io.strimzi.api.kafka.model.kafka.exporter.KafkaExporterResources;
 import io.strimzi.operator.cluster.ClusterOperatorConfig;
+import io.strimzi.operator.cluster.model.CertManagerUtils;
 import io.strimzi.operator.cluster.model.CertUtils;
 import io.strimzi.operator.cluster.model.ClusterCa;
 import io.strimzi.operator.cluster.model.ImagePullPolicy;
 import io.strimzi.operator.cluster.model.KafkaExporter;
 import io.strimzi.operator.cluster.model.KafkaVersion;
 import io.strimzi.operator.cluster.operator.resource.ResourceOperatorSupplier;
+import io.strimzi.operator.cluster.operator.resource.kubernetes.CertManagerCertificateOperator;
 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;
@@ -49,6 +52,7 @@ public class KafkaExporterReconciler {
     private final ServiceAccountOperator serviceAccountOperator;
     private final NetworkPolicyOperator networkPolicyOperator;
     private final PodDisruptionBudgetOperator podDisruptionBudgetOperator;
+    private final CertManagerCertificateOperator certManagerCertificateOperator;
 
     private String certificateHash = "";
 
@@ -83,6 +87,7 @@ public KafkaExporterReconciler(
         this.serviceAccountOperator = supplier.serviceAccountOperations;
         this.networkPolicyOperator = supplier.networkPolicyOperator;
         this.podDisruptionBudgetOperator = supplier.podDisruptionBudgetOperator;
+        this.certManagerCertificateOperator = supplier.certManagerCertificateOperator;
     }
 
     /**
@@ -99,7 +104,8 @@ public KafkaExporterReconciler(
      */
     public Future<Void> reconcile(boolean isOpenShift, ImagePullPolicy imagePullPolicy, List<LocalObjectReference> imagePullSecrets, Clock clock)    {
         return serviceAccount()
-                .compose(i -> certificatesSecret(clock))
+                .compose(i -> maybeReconcileCertManagerCertificates())
+                .compose(secret -> certificatesSecret(clock, secret))
                 .compose(i -> networkPolicy())
                 .compose(i -> podDisruptionBudget())
                 .compose(i -> deployment(isOpenShift, imagePullPolicy, imagePullSecrets))
@@ -121,27 +127,51 @@ private Future<Void> serviceAccount() {
                 ).mapEmpty();
     }
 
+    /**
+     * Manages the Certificate object that is used when cert-manager is the Certificate issuer
+     *
+     * @return Completes when the Certificate object was successfully created, deleted or updated and returns the related Secret
+     */
+    protected Future<Secret> maybeReconcileCertManagerCertificates() {
+        //TODO handle empty reconciles when kafka exporter not enabled
+        if (CertificateManagerType.CERT_MANAGER_IO.equals(clusterCa.getType())) {
+            return certManagerCertificateOperator.reconcile(reconciliation, reconciliation.namespace(), KafkaExporterResources.secretName(reconciliation.name()), kafkaExporter.generateCertificateResource(clusterCa))
+                    .compose(v -> certManagerCertificateOperator.waitForReady(reconciliation, reconciliation.namespace(), KafkaExporterResources.secretName(reconciliation.name())))
+                    .compose(v -> secretOperator.getAsync(reconciliation.namespace(), CertManagerUtils.certManagerSecretName(KafkaExporterResources.secretName(reconciliation.name()))));
+        } else {
+            return Future.succeededFuture();
+        }
+    }
+
     /**
      * Manages the Kafka Exporter Secret with certificates.
      *
-     * @param clock The clock for supplying the reconciler with the time instant of each reconciliation cycle.
-     *              That time is used for checking maintenance windows
+     * @param clock             The clock for supplying the reconciler with the time instant of each reconciliation cycle.
+     *                          That time is used for checking maintenance windows
+     * @param certManagerSecret Secret managed by cert-manager containing the Kafka Exporter certificate, may be null if Strimzi is issuing certificates.
      *
-     * @return      Future which completes when the reconciliation is done
+     * @return Future which completes when the reconciliation is done
      */
-    private Future<Void> certificatesSecret(Clock clock) {
+    private Future<Void> certificatesSecret(Clock clock, Secret certManagerSecret) {
         if (kafkaExporter != null) {
             return secretOperator.getAsync(reconciliation.namespace(), KafkaExporterResources.secretName(reconciliation.name()))
                     .compose(oldSecret -> {
-                        Secret newSecret = kafkaExporter.generateCertificatesSecret(clusterCa, oldSecret, Util.isMaintenanceTimeWindowsSatisfied(reconciliation, maintenanceWindows, clock.instant()));
-
-                        return secretOperator
-                                .reconcile(reconciliation, reconciliation.namespace(), KafkaExporterResources.secretName(reconciliation.name()), newSecret)
-                                .compose(i -> {
-                                    certificateHash = CertUtils.getCertificateShortThumbprint(newSecret, Ca.SecretEntry.CRT.asKey(KafkaExporter.COMPONENT_TYPE));
+                        Future<Secret> secretFuture;
+                        if (CertificateManagerType.CERT_MANAGER_IO.equals(clusterCa.getType())) {
+                            secretFuture = ReconcilerUtils.clusterCaPemTrustSet(reconciliation, secretOperator)
+                                    .map(pemTrustSet -> kafkaExporter.generateCertificatesSecretForCertManagerCA(clusterCa, oldSecret, certManagerSecret, pemTrustSet));
+                        } else {
+                            secretFuture = Future.succeededFuture(kafkaExporter.generateCertificatesSecretForStrimziCa(clusterCa, oldSecret, Util.isMaintenanceTimeWindowsSatisfied(reconciliation, maintenanceWindows, clock.instant())));
+                        }
 
+                        return secretFuture.compose(secret -> secretOperator.reconcile(reconciliation,
+                                        reconciliation.namespace(),
+                                        KafkaExporterResources.secretName(reconciliation.name()),
+                                        secret)
+                                .compose(result -> {
+                                    certificateHash = CertUtils.getCertificateShortThumbprint(secret, Ca.SecretEntry.CRT.asKey(KafkaExporter.COMPONENT_TYPE));
                                     return Future.succeededFuture();
-                                });
+                                }));
                     });
         } else {
             return secretOperator
diff --git a/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/ReconcilerUtils.java b/cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/ReconcilerUtils.java
@@ -135,7 +135,7 @@ public static Future<TlsPemIdentity> coTlsPemIdentity(Reconciliation reconciliat
      *
      * @return  Future containing the trust set to use for client authentication.
      */
-    private static Future<PemTrustSet> clusterCaPemTrustSet(Reconciliation reconciliation, SecretOperator secretOperator) {
+    public static Future<PemTrustSet> clusterCaPemTrustSet(Reconciliation reconciliation, SecretOperator secretOperator) {
         return getSecret(secretOperator, reconciliation.namespace(), KafkaResources.clusterCaCertificateSecretName(reconciliation.name()))
                 .map(PemTrustSet::new);
     }
diff --git a/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaExporterReconcilerCertManagerTest.java b/cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/assembly/KafkaExporterReconcilerCertManagerTest.java

Original file line number	Diff line number	Diff line change
`@@ -135,7 +135,7 @@ public static Future<TlsPemIdentity> coTlsPemIdentity(Reconciliation reconciliat`
`135`	`135`	`*`
`136`	`136`	`* @return Future containing the trust set to use for client authentication.`
`137`	`137`	`*/`
`138`		`- private static Future<PemTrustSet> clusterCaPemTrustSet(Reconciliation reconciliation, SecretOperator secretOperator) {`
	`138`	`+ public static Future<PemTrustSet> clusterCaPemTrustSet(Reconciliation reconciliation, SecretOperator secretOperator) {`
`139`	`139`	`return getSecret(secretOperator, reconciliation.namespace(), KafkaResources.clusterCaCertificateSecretName(reconciliation.name()))`
`140`	`140`	`.map(PemTrustSet::new);`
`141`	`141`	`}`