Fix storage issues - the init container is still needed (#1036)

* Fix storage issues - the init container is stil needed

* Use correct user ID

Author: scholzj

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

@@ -10,6 +10,7 @@
 import io.fabric8.kubernetes.api.model.ConfigMapVolumeSource;
 import io.fabric8.kubernetes.api.model.ConfigMapVolumeSourceBuilder;
 import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
 import io.fabric8.kubernetes.api.model.ContainerPort;
 import io.fabric8.kubernetes.api.model.ContainerPortBuilder;
 import io.fabric8.kubernetes.api.model.EnvVar;
@@ -23,6 +24,8 @@
 import io.fabric8.kubernetes.api.model.OwnerReferenceBuilder;
 import io.fabric8.kubernetes.api.model.PersistentVolumeClaim;
 import io.fabric8.kubernetes.api.model.PersistentVolumeClaimBuilder;
+import io.fabric8.kubernetes.api.model.PodSecurityContext;
+import io.fabric8.kubernetes.api.model.PodSecurityContextBuilder;
 import io.fabric8.kubernetes.api.model.Probe;
 import io.fabric8.kubernetes.api.model.ProbeBuilder;
 import io.fabric8.kubernetes.api.model.Quantity;
@@ -86,6 +89,11 @@ public abstract class AbstractModel {
     protected static final int CERTS_EXPIRATION_DAYS = 365;
     protected static final String DEFAULT_JVM_XMS = "128M";
 
+    private static final String VOLUME_MOUNT_HACK_IMAGE = "busybox";
+    protected static final String VOLUME_MOUNT_HACK_NAME = "volume-mount-hack";
+    private static final Long VOLUME_MOUNT_HACK_USERID = 1001L;
+    private static final Long VOLUME_MOUNT_HACK_GROUPID = 0L;
+
     public static final String ANCILLARY_CM_KEY_METRICS = "metrics-config.yml";
     public static final String ANCILLARY_CM_KEY_LOG_CONFIG = "log4j.properties";
     public static final String ENV_VAR_DYNAMIC_HEAP_FRACTION = "DYNAMIC_HEAP_FRACTION";
@@ -744,9 +752,11 @@ protected Service createHeadlessService(List<ServicePort> ports, Map<String, Str
     protected StatefulSet createStatefulSet(
             List<Volume> volumes,
             List<PersistentVolumeClaim> volumeClaims,
+            List<VolumeMount> volumeMounts,
             Affinity affinity,
             List<Container> initContainers,
-            List<Container> containers) {
+            List<Container> containers,
+            boolean isOpenShift) {
 
         Map<String, String> annotations = new HashMap<>();
 
@@ -755,6 +765,25 @@ protected StatefulSet createStatefulSet(
                         && ((PersistentClaimStorage) storage).isDeleteClaim()));
 
         List<Container> initContainersInternal = new ArrayList<>();
+        PodSecurityContext securityContext = null;
+        // if a persistent volume claim is requested and the running cluster is a Kubernetes one
+        // there is an hack on volume mounting which needs an "init-container"
+        if (this.storage instanceof PersistentClaimStorage && !isOpenShift) {
+            String chown = String.format("chown -R %d:%d %s",
+                    AbstractModel.VOLUME_MOUNT_HACK_USERID,
+                    AbstractModel.VOLUME_MOUNT_HACK_GROUPID,
+                    volumeMounts.get(0).getMountPath());
+            Container initContainer = new ContainerBuilder()
+                    .withName(AbstractModel.VOLUME_MOUNT_HACK_NAME)
+                    .withImage(AbstractModel.VOLUME_MOUNT_HACK_IMAGE)
+                    .withVolumeMounts(volumeMounts.get(0))
+                    .withCommand("sh", "-c", chown)
+                    .build();
+            initContainersInternal.add(initContainer);
+            securityContext = new PodSecurityContextBuilder()
+                    .withFsGroup(AbstractModel.VOLUME_MOUNT_HACK_GROUPID)
+                    .build();
+        }
         // add all the other init containers provided by the specific model implementation
         if (initContainers != null) {
             initContainersInternal.addAll(initContainers);
@@ -782,6 +811,7 @@ protected StatefulSet createStatefulSet(
                         .withNewSpec()
                             .withServiceAccountName(getServiceAccountName())
                             .withAffinity(affinity)
+                            .withSecurityContext(securityContext)
                             .withInitContainers(initContainersInternal)
                             .withContainers(containers)
                             .withVolumes(volumes)

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

@@ -516,9 +516,11 @@ public StatefulSet generateStatefulSet(boolean isOpenShift) {
         return createStatefulSet(
                 getVolumes(isOpenShift),
                 getVolumeClaims(),
+                getVolumeMounts(),
                 getMergedAffinity(),
                 getInitContainers(),
-                getContainers());
+                getContainers(),
+                isOpenShift);
     }
 
     /**

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

@@ -277,9 +277,11 @@ public StatefulSet generateStatefulSet(boolean isOpenShift) {
         return createStatefulSet(
                 getVolumes(isOpenShift),
                 getVolumeClaims(),
+                getVolumeMounts(),
                 getMergedAffinity(),
                 getInitContainers(),
-                getContainers());
+                getContainers(),
+                isOpenShift);
     }
 
     /**

cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaClusterTest.java

@@ -23,6 +23,7 @@
 import io.strimzi.api.kafka.model.KafkaBuilder;
 import io.strimzi.api.kafka.model.KafkaClusterSpec;
 import io.strimzi.api.kafka.model.KafkaListenerAuthenticationTls;
+import io.strimzi.api.kafka.model.PersistentClaimStorage;
 import io.strimzi.api.kafka.model.PersistentClaimStorageBuilder;
 import io.strimzi.api.kafka.model.Rack;
 import io.strimzi.api.kafka.model.TlsSidecarLogLevel;
@@ -250,6 +251,20 @@ private void checkStatefulSet(StatefulSet ss, Kafka cm, boolean isOpenShift) {
         assertEquals(KafkaCluster.TLS_SIDECAR_KAFKA_CERTS_VOLUME_MOUNT, containers.get(1).getVolumeMounts().get(0).getMountPath());
         assertEquals(KafkaCluster.TLS_SIDECAR_CLUSTER_CA_CERTS_VOLUME_MOUNT, containers.get(1).getVolumeMounts().get(1).getMountPath());
 
+        if (cm.getSpec().getKafka().getStorage() != null) {
+            io.strimzi.api.kafka.model.Storage storage = cm.getSpec().getKafka().getStorage();
+            if (storage instanceof PersistentClaimStorage && !isOpenShift) {
+                PodSpec podSpec = ss.getSpec().getTemplate().getSpec();
+                // check that pod spec contains the volume hack container for Kubernetes
+                List<Container> initContainers = podSpec.getInitContainers();
+                assertNotNull(initContainers);
+                assertTrue(initContainers.size() > 0);
+                boolean isVolumeHack =
+                        initContainers.stream().anyMatch(container -> container.getName().equals(AbstractModel.VOLUME_MOUNT_HACK_NAME));
+                assertTrue(isVolumeHack);
+            }
+        }
+
         if (cm.getSpec().getKafka().getRack() != null) {
 
             Rack rack = cm.getSpec().getKafka().getRack();