Refactor out the common methods used for reconciling the internal secrets into ReconcilerUtils

Signed-off-by: Gantigmaa Selenge <[email protected]>

Author: tinaselenge

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

@@ -1681,6 +1681,7 @@ public ClusterRoleBinding generateClusterRoleBinding(String assemblyNamespace) {
     public Role generateRole() {
         List<String> certSecretNames = new ArrayList<>();
         certSecretNames.add(KafkaResources.clusterCaCertificateSecretName(cluster));
+        certSecretNames.add(KafkaResources.clientsCaCertificateSecretName(cluster));
         certSecretNames.add(KafkaResources.internalAuthzTrustedCertsSecretName(cluster));
         certSecretNames.add(KafkaResources.internalOauthTrustedCertsSecretName(cluster));
         certSecretNames.addAll(nodes().stream().map(NodeRef::podName).toList());

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

@@ -940,16 +940,15 @@ public RoleBinding generateRoleBindingForRole() {
     }
 
     /**
-     * Creates a secret that contains the TLS certificates from one or more secrets
+     * Creates a secret with the given name and data
      * in the same namespace as the resource.
-     * This is used for loading truststore certificates from the secret directly.
-     **
+     *
      * @param secretData secret data
      * @param secretName secret name
      *
-     * @return secret for tls certificates
+     * @return secret
      */
-    public Secret generateTlsTrustedCertsSecret(Map<String, String> secretData, String secretName) {
+    public Secret generateSecret(Map<String, String> secretData, String secretName) {
         return ModelUtils.createSecret(secretName, namespace, labels, ownerReference, secretData, Map.of(), Map.of());
     }
 

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

@@ -29,6 +29,7 @@
 import io.strimzi.api.kafka.model.connector.KafkaConnector;
 import io.strimzi.api.kafka.model.connector.KafkaConnectorSpec;
 import io.strimzi.api.kafka.model.connector.ListOffsets;
+import io.strimzi.api.kafka.model.kafka.KafkaResources;
 import io.strimzi.api.kafka.model.kafka.Status;
 import io.strimzi.api.kafka.model.mirrormaker2.KafkaMirrorMaker2;
 import io.strimzi.operator.cluster.ClusterOperatorConfig;
@@ -62,7 +63,6 @@
 import io.strimzi.operator.common.BackOff;
 import io.strimzi.operator.common.Reconciliation;
 import io.strimzi.operator.common.ReconciliationLogger;
-import io.strimzi.operator.common.Util;
 import io.strimzi.operator.common.model.InvalidResourceException;
 import io.strimzi.operator.common.model.Labels;
 import io.strimzi.operator.common.model.OrderedProperties;
@@ -84,7 +84,6 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.TreeMap;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -274,7 +273,7 @@ protected Future<ReconcileResult<NetworkPolicy>> connectNetworkPolicy(Reconcilia
      *
      * @return  Future which completes when the reconciliation is done
      */
-    protected Future<Void> tlsTrustedCertsSecret(Reconciliation reconciliation, String namespace, KafkaConnectCluster connect) {
+    protected Future<Void> tlsTrustedCertsSecret(Reconciliation reconciliation, KafkaConnectCluster connect) {
         ClientTls tls = connect.getTls();
         Set<String> secretsToCopy = new HashSet<>();
 
@@ -286,27 +285,7 @@ protected Future<Void> tlsTrustedCertsSecret(Reconciliation reconciliation, Stri
             return Future.succeededFuture();
         }
 
-        ConcurrentHashMap<String, String> secretData = new ConcurrentHashMap<>();
-        return Future.join(secretsToCopy.stream()
-                        .map(secretName -> secretOperations.getAsync(namespace, secretName)
-                                .compose(secret -> {
-                                    if (secret == null) {
-                                        return Future.failedFuture("Secret " + secretName + " not found");
-                                    } else {
-                                        secret.getData().entrySet().stream()
-                                                .filter(e -> e.getKey().contains(".crt"))
-                                                // In case secrets contain the same key, append the secret name into the key
-                                                .forEach(e -> secretData.put(secretName + "-" + e.getKey(), e.getValue()));
-                                    }
-                                    return Future.succeededFuture();
-                                }))
-                        .collect(Collectors.toList()))
-                .compose(ignore -> secretOperations.reconcile(
-                                reconciliation,
-                                namespace,
-                                KafkaConnectResources.internalTlsTrustedCertsSecretName(connect.getCluster()),
-                                connect.generateTlsTrustedCertsSecret(secretData, KafkaConnectResources.internalTlsTrustedCertsSecretName(connect.getCluster())))
-                        .mapEmpty());
+        return ReconcilerUtils.generateTlsTrustedCertsSecret(reconciliation, secretsToCopy, KafkaConnectResources.internalTlsTrustedCertsSecretName(connect.getCluster()), secretOperations, connect::generateSecret).mapEmpty();
     }
 
     /**
@@ -316,7 +295,7 @@ protected Future<Void> tlsTrustedCertsSecret(Reconciliation reconciliation, Stri
      *
      * @return  Future which completes when the reconciliation is done
      */
-    protected Future<Void> oauthTrustedCertsSecret(Reconciliation reconciliation, String namespace, KafkaConnectCluster connect) {
+    protected Future<Void> oauthTrustedCertsSecret(Reconciliation reconciliation, KafkaConnectCluster connect) {
         KafkaClientAuthentication authentication = connect.getAuthentication();
         Set<String> secretsToCopy = new HashSet<>();
 
@@ -328,42 +307,7 @@ protected Future<Void> oauthTrustedCertsSecret(Reconciliation reconciliation, St
             return Future.succeededFuture();
         }
 
-        List<String> certs = new ArrayList<>();
-        String oauthSecret = KafkaConnectResources.internalOauthTrustedCertsSecretName(connect.getCluster());
-        return Future.join(secretsToCopy.stream()
-                        .map(secretName -> secretOperations.getAsync(namespace, secretName)
-                                .compose(secret -> {
-                                    if (secret == null) {
-                                        return Future.failedFuture("Secret " + secretName + " not found");
-                                    } else {
-                                        secret.getData().entrySet().stream()
-                                                .filter(e -> e.getKey().contains(".crt"))
-                                                // In case secrets contain the same key, append the secret name into the key
-                                                .forEach(e -> certs.add(e.getValue()));
-                                    }
-                                    return Future.succeededFuture();
-                                }))
-                        .collect(Collectors.toList()))
-                .compose(ignore -> secretOperations.reconcile(
-                                reconciliation,
-                                namespace,
-                                oauthSecret,
-                                connect.generateTlsTrustedCertsSecret(Map.of(oauthSecret + ".crt", mergeAndEncodeCerts(certs)), oauthSecret))
-                        .mapEmpty());
-    }
-
-    private String mergeAndEncodeCerts(List<String> certs) {
-        if (certs.size() > 1) {
-            String decodedAndMergedCerts = certs.stream()
-                    .map(Util::decodeFromBase64)
-                    .collect(Collectors.joining("\n"));
-
-            return Util.encodeToBase64(decodedAndMergedCerts);
-        } else if (certs.size() < 1) {
-            return "";
-        } else {
-            return certs.get(0);
-        }
+        return ReconcilerUtils.generateOauthTrustedCertsSecret(reconciliation, secretsToCopy, KafkaResources.internalOauthTrustedCertsSecretName(connect.getCluster()), secretOperations, connect::generateSecret);
     }
 
     /**

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

@@ -180,8 +180,8 @@ protected Future<KafkaConnectStatus> createOrUpdate(Reconciliation reconciliatio
                 })
                 .compose(i -> serviceOperations.reconcile(reconciliation, namespace, connect.getServiceName(), connect.generateService()))
                 .compose(i -> serviceOperations.reconcile(reconciliation, namespace, connect.getComponentName(), connect.generateHeadlessService()))
-                .compose(i -> tlsTrustedCertsSecret(reconciliation, namespace, connect))
-                .compose(i -> oauthTrustedCertsSecret(reconciliation, namespace, connect))
+                .compose(i -> tlsTrustedCertsSecret(reconciliation, connect))
+                .compose(i -> oauthTrustedCertsSecret(reconciliation, connect))
                 .compose(i -> generateMetricsAndLoggingConfigMap(reconciliation, connect))
                 .compose(logAndMetricsConfigMap -> {
                     String logging = logAndMetricsConfigMap.getData().get(connect.logging().configMapKey());

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

@@ -126,8 +126,8 @@ protected Future<KafkaMirrorMaker2Status> createOrUpdate(Reconciliation reconcil
                 .compose(i -> manualRollingUpdate(reconciliation, mirrorMaker2Cluster))
                 .compose(i -> serviceOperations.reconcile(reconciliation, namespace, mirrorMaker2Cluster.getServiceName(), mirrorMaker2Cluster.generateService()))
                 .compose(i -> serviceOperations.reconcile(reconciliation, namespace, mirrorMaker2Cluster.getComponentName(), mirrorMaker2Cluster.generateHeadlessService()))
-                .compose(i -> tlsTrustedCertsSecret(reconciliation, namespace, mirrorMaker2Cluster))
-                .compose(i -> oauthTrustedCertsSecret(reconciliation, namespace, mirrorMaker2Cluster))
+                .compose(i -> tlsTrustedCertsSecret(reconciliation, mirrorMaker2Cluster))
+                .compose(i -> oauthTrustedCertsSecret(reconciliation, mirrorMaker2Cluster))
                 .compose(i -> generateMetricsAndLoggingConfigMap(reconciliation, mirrorMaker2Cluster))
                 .compose(logAndMetricsConfigMap -> {
                     String logging = logAndMetricsConfigMap.getData().get(mirrorMaker2Cluster.logging().configMapKey());

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

@@ -872,7 +872,6 @@ protected Future<Void> updateCertificateSecrets(List<Secret> secrets) {
     @Deprecated
     protected Future<Void> authzTrustedCertsSecret() {
         Set<String> secretsToCopy = new HashSet<>();
-        List<Integer> certHashes = new ArrayList<>();
 
         if (kafka.getAuthorization() instanceof KafkaAuthorizationOpa opaAuthz && opaAuthz.getTlsTrustedCertificates() != null) {
             secretsToCopy.addAll(opaAuthz.getTlsTrustedCertificates().stream().map(CertSecretSource::getSecretName).toList());
@@ -886,31 +885,9 @@ protected Future<Void> authzTrustedCertsSecret() {
             return Future.succeededFuture();
         }
 
-        ConcurrentHashMap<String, String> secretData = new ConcurrentHashMap<>();
-        return Future.join(secretsToCopy.stream()
-                        .map(secretName -> secretOperator.getAsync(reconciliation.namespace(), secretName)
-                                .compose(secret -> {
-                                    if (secret == null) {
-                                        return Future.failedFuture("Secret " + secretName + " not found");
-                                    } else {
-                                        secret.getData().entrySet().stream()
-                                                .filter(e -> e.getKey().contains(".crt"))
-                                                // In case secrets contain the same key, append the secret name into the key
-                                                .forEach(e -> {
-                                                    secretData.put(secretName + "-" + e.getKey(), e.getValue());
-                                                    certHashes.add(e.getValue().hashCode());
-                                                });
-                                    }
-                                    return Future.succeededFuture();
-                                }))
-                        .collect(Collectors.toList()))
-                .compose(ignore -> secretOperator.reconcile(
-                                reconciliation,
-                                reconciliation.namespace(),
-                                KafkaResources.internalAuthzTrustedCertsSecretName(kafka.getCluster()),
-                                kafka.generateSecret(secretData, KafkaResources.internalAuthzTrustedCertsSecretName(kafka.getCluster()))))
-                .compose(ignore -> {
-                    authorizerServerCertificateHash = certHashes.stream().mapToInt(e -> e).sum();
+        return ReconcilerUtils.generateTlsTrustedCertsSecret(reconciliation, secretsToCopy, KafkaResources.internalAuthzTrustedCertsSecretName(kafka.getCluster()), secretOperator, kafka::generateSecret)
+                .compose(certHashes -> {
+                    authorizerServerCertificateHash = certHashes;
                     return Future.succeededFuture();
                 });
     }
@@ -936,43 +913,9 @@ protected Future<Void> oauthTrustedCertsSecret() {
             return Future.succeededFuture();
         }
 
-        List<String> certs = new ArrayList<>();
-        String oauthSecret = KafkaResources.internalOauthTrustedCertsSecretName(kafka.getCluster());
-        return Future.join(secretsToCopy.stream()
-                        .map(secretName -> secretOperator.getAsync(reconciliation.namespace(), secretName)
-                                .compose(secret -> {
-                                    if (secret == null) {
-                                        return Future.failedFuture("Secret " + secretName + " not found");
-                                    } else {
-                                        secret.getData().entrySet().stream()
-                                                .filter(e -> e.getKey().contains(".crt"))
-                                                // certificates appended under a same key so that we create a single volume mounted file
-                                                .forEach(e -> certs.add(e.getValue()));
-                                    }
-                                    return Future.succeededFuture();
-                                }))
-                        .collect(Collectors.toList()))
-                .compose(ignore -> secretOperator.reconcile(
-                                reconciliation,
-                                reconciliation.namespace(),
-                                oauthSecret,
-                                kafka.generateSecret(Map.of(oauthSecret + ".crt", mergeAndEncodeCerts(certs)), oauthSecret))
-                        .mapEmpty());
+        return ReconcilerUtils.generateOauthTrustedCertsSecret(reconciliation, secretsToCopy, KafkaResources.internalOauthTrustedCertsSecretName(kafka.getCluster()), secretOperator, kafka::generateSecret);
     }
 
-    private String mergeAndEncodeCerts(List<String> certs) {
-        if (certs.size() > 1) {
-            String decodedAndMergedCerts = certs.stream()
-                    .map(Util::decodeFromBase64)
-                    .collect(Collectors.joining("\n"));
-
-            return Util.encodeToBase64(decodedAndMergedCerts);
-        } else if (certs.size() < 1) {
-            return "";
-        } else {
-            return certs.get(0);
-        }
-    }
     /**
      * Manages the secret with JMX credentials when JMX is enabled
      *