/**
 * Creates a secret volume
 *
 * @param name        Name of the Volume
 * @param secretName  Name of the Secret
 * @param isOpenshift true if underlying cluster OpenShift
 * @return The Volume created
 */
public static Volume createSecretVolume(String name, String secretName, boolean isOpenshift) {
    String validName = getValidVolumeName(name);

    int mode = 0444;
    if (isOpenshift) {
        mode = 0440;
    }

    SecretVolumeSource secretVolumeSource = new SecretVolumeSourceBuilder()
            .withDefaultMode(mode)
            .withSecretName(secretName)
            .build();

    Volume volume = new VolumeBuilder()
            .withName(validName)
            .withSecret(secretVolumeSource)
            .build();
    log.trace("Created secret Volume named '{}' with source secret '{}'", validName, secretName);
    return volume;
}
 

/**
 * Creates a secret volume with given items
 *
 * @param name        Name of the Volume
 * @param secretName  Name of the Secret
 * @param items       contents of the Secret
 * @param isOpenshift true if underlying cluster OpenShift
 * @return The Volume created
 */
public static Volume createSecretVolume(String name, String secretName, Map<String, String> items, boolean isOpenshift) {
    String validName = getValidVolumeName(name);

    int mode = 0444;
    if (isOpenshift) {
        mode = 0440;
    }

    List<KeyToPath> keysPaths = new ArrayList<>();

    for (Map.Entry<String, String> item : items.entrySet()) {
        KeyToPath keyPath = new KeyToPathBuilder()
                .withNewKey(item.getKey())
                .withNewPath(item.getValue())
                .build();

        keysPaths.add(keyPath);
    }

    SecretVolumeSource secretVolumeSource = new SecretVolumeSourceBuilder()
            .withDefaultMode(mode)
            .withSecretName(secretName)
            .withItems(keysPaths)
            .build();

    Volume volume = new VolumeBuilder()
            .withName(validName)
            .withSecret(secretVolumeSource)
            .build();
    log.trace("Created secret Volume named '{}' with source secret '{}'", validName, secretName);
    return volume;
}
 

@Test
public void testExternalConfigurationSecretVolumes() {
    ExternalConfigurationVolumeSource volume = new ExternalConfigurationVolumeSourceBuilder()
            .withName("my-volume")
            .withSecret(new SecretVolumeSourceBuilder().withSecretName("my-secret").build())
            .build();

    KafkaConnect resource = new KafkaConnectBuilder(this.resource)
            .editSpec()
                .withNewExternalConfiguration()
                    .withVolumes(volume)
                .endExternalConfiguration()
            .endSpec()
            .build();
    KafkaConnectCluster kc = KafkaConnectCluster.fromCrd(resource, VERSIONS);

    // Check Deployment
    Deployment dep = kc.generateDeployment(emptyMap(), true, null, null);
    List<Volume> volumes = dep.getSpec().getTemplate().getSpec().getVolumes();
    List<Volume> selected = volumes.stream().filter(vol -> vol.getName().equals(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(1));
    assertThat(selected.get(0).getName(), is(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume"));
    assertThat(selected.get(0).getSecret(), is(volume.getSecret()));

    List<VolumeMount> volumeMounths = dep.getSpec().getTemplate().getSpec().getContainers().get(0).getVolumeMounts();
    List<VolumeMount> selectedVolumeMounths = volumeMounths.stream().filter(vol -> vol.getName().equals(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(1));
    assertThat(selectedVolumeMounths.get(0).getName(), is(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume"));
    assertThat(selectedVolumeMounths.get(0).getMountPath(), is(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_MOUNT_BASE_PATH + "my-volume"));
}
 

@Test
public void testExternalConfigurationInvalidVolumes() {
    ExternalConfigurationVolumeSource volume = new ExternalConfigurationVolumeSourceBuilder()
            .withName("my-volume")
            .withConfigMap(new ConfigMapVolumeSourceBuilder().withName("my-map").build())
            .withSecret(new SecretVolumeSourceBuilder().withSecretName("my-secret").build())
            .build();

    KafkaConnect resource = new KafkaConnectBuilder(this.resource)
            .editSpec()
                .withNewExternalConfiguration()
                    .withVolumes(volume)
                .endExternalConfiguration()
            .endSpec()
            .build();
    KafkaConnectCluster kc = KafkaConnectCluster.fromCrd(resource, VERSIONS);

    // Check Deployment
    Deployment dep = kc.generateDeployment(emptyMap(), true, null, null);
    List<Volume> volumes = dep.getSpec().getTemplate().getSpec().getVolumes();
    List<Volume> selected = volumes.stream().filter(vol -> vol.getName().equals(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(0));

    List<VolumeMount> volumeMounths = dep.getSpec().getTemplate().getSpec().getContainers().get(0).getVolumeMounts();
    List<VolumeMount> selectedVolumeMounths = volumeMounths.stream().filter(vol -> vol.getName().equals(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(0));
}
 

@Test
public void testExternalConfigurationSecretVolumes() {
    ExternalConfigurationVolumeSource volume = new ExternalConfigurationVolumeSourceBuilder()
            .withName("my-volume")
            .withSecret(new SecretVolumeSourceBuilder().withSecretName("my-secret").build())
            .build();

    KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource)
            .editSpec()
                .withNewExternalConfiguration()
                    .withVolumes(volume)
                .endExternalConfiguration()
            .endSpec()
            .build();
    KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS);

    // Check Deployment
    Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null);
    List<Volume> volumes = dep.getSpec().getTemplate().getSpec().getVolumes();
    List<Volume> selected = volumes.stream().filter(vol -> vol.getName().equals(KafkaMirrorMaker2Cluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(1));
    assertThat(selected.get(0).getName(), is(KafkaMirrorMaker2Cluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume"));
    assertThat(selected.get(0).getSecret(), is(volume.getSecret()));

    List<VolumeMount> volumeMounths = getContainer(dep).getVolumeMounts();
    List<VolumeMount> selectedVolumeMounths = volumeMounths.stream().filter(vol -> vol.getName().equals(KafkaMirrorMaker2Cluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(1));
    assertThat(selectedVolumeMounths.get(0).getName(), is(KafkaMirrorMaker2Cluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume"));
    assertThat(selectedVolumeMounths.get(0).getMountPath(), is(KafkaMirrorMaker2Cluster.EXTERNAL_CONFIGURATION_VOLUME_MOUNT_BASE_PATH + "my-volume"));
}
 

@Test
public void testExternalConfigurationInvalidVolumes() {
    ExternalConfigurationVolumeSource volume = new ExternalConfigurationVolumeSourceBuilder()
            .withName("my-volume")
            .withConfigMap(new ConfigMapVolumeSourceBuilder().withName("my-map").build())
            .withSecret(new SecretVolumeSourceBuilder().withSecretName("my-secret").build())
            .build();

    KafkaMirrorMaker2 resource = new KafkaMirrorMaker2Builder(this.resource)
            .editSpec()
                .withNewExternalConfiguration()
                    .withVolumes(volume)
                .endExternalConfiguration()
            .endSpec()
            .build();
    KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(resource, VERSIONS);

    // Check Deployment
    Deployment dep = kmm2.generateDeployment(emptyMap(), true, null, null);
    List<Volume> volumes = dep.getSpec().getTemplate().getSpec().getVolumes();
    List<Volume> selected = volumes.stream().filter(vol -> vol.getName().equals(KafkaMirrorMaker2Cluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(0));

    List<VolumeMount> volumeMounths = getContainer(dep).getVolumeMounts();
    List<VolumeMount> selectedVolumeMounths = volumeMounths.stream().filter(vol -> vol.getName().equals(KafkaMirrorMaker2Cluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(0));
}
 

@Test
public void testExternalConfigurationSecretVolumes() {
    ExternalConfigurationVolumeSource volume = new ExternalConfigurationVolumeSourceBuilder()
            .withName("my-volume")
            .withSecret(new SecretVolumeSourceBuilder().withSecretName("my-secret").build())
            .build();

    KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource)
            .editSpec()
                .withNewExternalConfiguration()
                    .withVolumes(volume)
                .endExternalConfiguration()
            .endSpec()
            .build();
    KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS);

    // Check DeploymentConfig
    DeploymentConfig dep = kc.generateDeploymentConfig(Collections.EMPTY_MAP, true, null, null);
    List<Volume> volumes = dep.getSpec().getTemplate().getSpec().getVolumes();
    List<Volume> selected = volumes.stream().filter(vol -> vol.getName().equals(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(1));
    assertThat(selected.get(0).getName(), is(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume"));
    assertThat(selected.get(0).getSecret(), is(volume.getSecret()));

    List<VolumeMount> volumeMounths = dep.getSpec().getTemplate().getSpec().getContainers().get(0).getVolumeMounts();
    List<VolumeMount> selectedVolumeMounths = volumeMounths.stream().filter(vol -> vol.getName().equals(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(1));
    assertThat(selectedVolumeMounths.get(0).getName(), is(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume"));
    assertThat(selectedVolumeMounths.get(0).getMountPath(), is(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_MOUNT_BASE_PATH + "my-volume"));
}
 

@Test
public void testExternalConfigurationInvalidVolumes() {
    ExternalConfigurationVolumeSource volume = new ExternalConfigurationVolumeSourceBuilder()
            .withName("my-volume")
            .withConfigMap(new ConfigMapVolumeSourceBuilder().withName("my-map").build())
            .withSecret(new SecretVolumeSourceBuilder().withSecretName("my-secret").build())
            .build();

    KafkaConnectS2I resource = new KafkaConnectS2IBuilder(this.resource)
            .editSpec()
                .withNewExternalConfiguration()
                    .withVolumes(volume)
                .endExternalConfiguration()
            .endSpec()
            .build();
    KafkaConnectS2ICluster kc = KafkaConnectS2ICluster.fromCrd(resource, VERSIONS);

    // Check Deployment
    DeploymentConfig dep = kc.generateDeploymentConfig(Collections.EMPTY_MAP, true, null, null);
    List<Volume> volumes = dep.getSpec().getTemplate().getSpec().getVolumes();
    List<Volume> selected = volumes.stream().filter(vol -> vol.getName().equals(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(0));

    List<VolumeMount> volumeMounths = dep.getSpec().getTemplate().getSpec().getContainers().get(0).getVolumeMounts();
    List<VolumeMount> selectedVolumeMounths = volumeMounths.stream().filter(vol -> vol.getName().equals(KafkaConnectCluster.EXTERNAL_CONFIGURATION_VOLUME_NAME_PREFIX + "my-volume")).collect(Collectors.toList());
    assertThat(selected.size(), is(0));
}
 

private static Deployment getProxyApiAppDeploymentResource() {
    return new DeploymentBuilder()
            .withNewMetadata()
            .withName(API_PROXY)
            .addToLabels("app", API_PROXY)
            .endMetadata()
            .withNewSpec()
            .withNewSelector()
            .addToMatchLabels("app", API_PROXY)
            .endSelector()
            .withReplicas(1)
            .withNewTemplate()
            .withNewMetadata()
            .addToLabels("app", API_PROXY)
            .endMetadata()
            .withNewSpec()
            .addNewContainer()
            .withName(API_PROXY)
            .withImage("quay.io/enmasse/api-proxy:latest")
            .withPorts(new ContainerPortBuilder().withContainerPort(8443).withName("https").withProtocol("TCP").build())
            .withVolumeMounts(new VolumeMountBuilder().withMountPath("/etc/tls/private").withName("api-proxy-tls").withReadOnly(true).build())
            .endContainer()
            .withVolumes(Collections.singletonList(new VolumeBuilder().withName("api-proxy-tls").withSecret(new SecretVolumeSourceBuilder().withDefaultMode(420).withSecretName("api-proxy-cert").build()).build()))
            .endSpec()
            .endTemplate()
            .endSpec()
            .build();
}
 

private void mountSshKeySecret(String secretName, PodSpec podSpec, boolean addVolume) {
  if (addVolume) {
    podSpec
        .getVolumes()
        .add(
            new VolumeBuilder()
                .withName(secretName)
                .withSecret(
                    new SecretVolumeSourceBuilder()
                        .withSecretName(secretName)
                        .withDefaultMode(0600)
                        .build())
                .build());
  }

  List<Container> containers = podSpec.getContainers();
  containers.forEach(
      container -> {
        VolumeMount volumeMount =
            new VolumeMountBuilder()
                .withName(secretName)
                .withNewReadOnly(true)
                .withReadOnly(true)
                .withMountPath(SSH_PRIVATE_KEYS_PATH)
                .build();
        container.getVolumeMounts().add(volumeMount);
      });
}
 

@VisibleForTesting
static Pod createPod(WorkflowInstance workflowInstance,
                     RunSpec runSpec,
                     KubernetesSecretSpec secretSpec,
                     String styxEnvironment) {
  final String imageWithTag = runSpec.imageName().contains(":")
      ? runSpec.imageName()
      : runSpec.imageName() + ":latest";

  final String executionId = runSpec.executionId();
  final PodBuilder podBuilder = new PodBuilder()
      .withNewMetadata()
      .withName(executionId)
      .addToAnnotations(STYX_WORKFLOW_INSTANCE_ANNOTATION, workflowInstance.toKey())
      .addToAnnotations(DOCKER_TERMINATION_LOGGING_ANNOTATION,
                        String.valueOf(runSpec.terminationLogging()))
      .endMetadata();

  final PodSpecBuilder specBuilder = new PodSpecBuilder()
      .withRestartPolicy("Never");

  final ResourceRequirementsBuilder resourceRequirements = new ResourceRequirementsBuilder();
  runSpec.memRequest().ifPresent(s -> resourceRequirements.addToRequests("memory", new Quantity(s)));
  runSpec.memLimit().ifPresent(s -> resourceRequirements.addToLimits("memory", new Quantity(s)));

  final ContainerBuilder mainContainerBuilder = new ContainerBuilder()
      .withName(MAIN_CONTAINER_NAME)
      .withImage(imageWithTag)
      .withArgs(runSpec.args())
      .withEnv(buildEnv(workflowInstance, runSpec, styxEnvironment))
      .withResources(resourceRequirements.build());

  secretSpec.serviceAccountSecret().ifPresent(serviceAccountSecret -> {
    final SecretVolumeSource saVolumeSource = new SecretVolumeSourceBuilder()
        .withSecretName(serviceAccountSecret)
        .build();
    final Volume saVolume = new VolumeBuilder()
        .withName(STYX_WORKFLOW_SA_SECRET_NAME)
        .withSecret(saVolumeSource)
        .build();
    specBuilder.addToVolumes(saVolume);

    final VolumeMount saMount = new VolumeMountBuilder()
        .withMountPath(STYX_WORKFLOW_SA_SECRET_MOUNT_PATH)
        .withName(saVolume.getName())
        .withReadOnly(true)
        .build();
    mainContainerBuilder.addToVolumeMounts(saMount);
    mainContainerBuilder.addToEnv(envVar(STYX_WORKFLOW_SA_ENV_VARIABLE,
                                     saMount.getMountPath() + STYX_WORKFLOW_SA_JSON_KEY));
  });

  secretSpec.customSecret().ifPresent(secret -> {
    final SecretVolumeSource secretVolumeSource = new SecretVolumeSourceBuilder()
        .withSecretName(secret.name())
        .build();
    final Volume secretVolume = new VolumeBuilder()
        .withName(secret.name())
        .withSecret(secretVolumeSource)
        .build();
    specBuilder.addToVolumes(secretVolume);

    final VolumeMount secretMount = new VolumeMountBuilder()
        .withMountPath(secret.mountPath())
        .withName(secretVolume.getName())
        .withReadOnly(true)
        .build();
    mainContainerBuilder.addToVolumeMounts(secretMount);
  });

  specBuilder.addToContainers(mainContainerBuilder.build());
  specBuilder.addToContainers(keepaliveContainer());
  podBuilder.withSpec(specBuilder.build());

  return podBuilder.build();
}
 

private Volume buildCertSecretVolume(String secretName) {
  return new VolumeBuilder()
      .withName(CHE_SELF_SIGNED_CERT_VOLUME)
      .withSecret(new SecretVolumeSourceBuilder().withSecretName(secretName).build())
      .build();
}
 

/**
 * Applies secret as file into workspace containers, respecting automount attribute and optional
 * devfile automount property and/or mount path override.
 *
 * @param env kubernetes environment with workspace containers configuration
 * @param runtimeIdentity identity of current runtime
 * @param secret source secret to apply
 * @throws InfrastructureException on misconfigured secrets or other apply error
 */
@Override
public void applySecret(KubernetesEnvironment env, RuntimeIdentity runtimeIdentity, Secret secret)
    throws InfrastructureException {
  final String secretMountPath = secret.getMetadata().getAnnotations().get(ANNOTATION_MOUNT_PATH);
  boolean secretAutomount =
      Boolean.parseBoolean(secret.getMetadata().getAnnotations().get(ANNOTATION_AUTOMOUNT));
  if (secretMountPath == null) {
    throw new InfrastructureException(
        format(
            "Unable to mount secret '%s': It is configured to be mounted as a file but the mount path was not specified. Please define the '%s' annotation on the secret to specify it.",
            secret.getMetadata().getName(), ANNOTATION_MOUNT_PATH));
  }

  Volume volumeFromSecret =
      new VolumeBuilder()
          .withName(secret.getMetadata().getName())
          .withSecret(
              new SecretVolumeSourceBuilder()
                  .withNewSecretName(secret.getMetadata().getName())
                  .build())
          .build();

  for (PodData podData : env.getPodsData().values()) {
    if (!podData.getRole().equals(PodRole.DEPLOYMENT)) {
      continue;
    }
    if (podData
        .getSpec()
        .getVolumes()
        .stream()
        .anyMatch(v -> v.getName().equals(volumeFromSecret.getName()))) {
      volumeFromSecret.setName(volumeFromSecret.getName() + "_" + NameGenerator.generate("", 6));
    }

    podData.getSpec().getVolumes().add(volumeFromSecret);

    for (Container container : podData.getSpec().getContainers()) {
      Optional<ComponentImpl> component = getComponent(env, container.getName());
      // skip components that explicitly disable automount
      if (component.isPresent() && isComponentAutomountFalse(component.get())) {
        continue;
      }
      // if automount disabled globally and not overridden in component
      if (!secretAutomount
          && (!component.isPresent() || !isComponentAutomountTrue(component.get()))) {
        continue;
      }
      // find path override if any
      Optional<String> overridePathOptional = Optional.empty();
      if (component.isPresent()) {
        overridePathOptional =
            getOverridenComponentPath(component.get(), secret.getMetadata().getName());
      }
      final String componentMountPath = overridePathOptional.orElse(secretMountPath);
      container
          .getVolumeMounts()
          .removeIf(vm -> Paths.get(vm.getMountPath()).equals(Paths.get(componentMountPath)));

      secret
          .getData()
          .keySet()
          .forEach(
              secretFile ->
                  container
                      .getVolumeMounts()
                      .add(
                          new VolumeMountBuilder()
                              .withName(volumeFromSecret.getName())
                              .withMountPath(componentMountPath + "/" + secretFile)
                              .withSubPath(secretFile)
                              .withReadOnly(true)
                              .build()));
    }
  }
}
 

@Test
public void shouldNotReplaceNonPVCVolumes() {
  // given
  podData
      .getSpec()
      .getInitContainers()
      .add(
          new ContainerBuilder()
              .withName("userInitContainer")
              .withVolumeMounts(new VolumeMountBuilder().withName("configMap").build())
              .build());

  podData
      .getSpec()
      .getContainers()
      .get(0)
      .getVolumeMounts()
      .add(new VolumeMountBuilder().withName("secret").withSubPath("/home/user/data").build());

  podData
      .getSpec()
      .getVolumes()
      .add(
          new VolumeBuilder()
              .withName("configMap")
              .withConfigMap(new ConfigMapVolumeSourceBuilder().withName("configMap").build())
              .build());
  podData
      .getSpec()
      .getVolumes()
      .add(
          new VolumeBuilder()
              .withName("secret")
              .withSecret(new SecretVolumeSourceBuilder().withSecretName("secret").build())
              .build());

  // when
  podsVolumes.replacePVCVolumesWithCommon(ImmutableMap.of("pod", podData), "commonPVC");

  // then
  assertEquals(podData.getSpec().getVolumes().size(), 2);
  assertNotNull(podData.getSpec().getVolumes().get(0).getConfigMap());
  assertNull(podData.getSpec().getVolumes().get(0).getPersistentVolumeClaim());

  assertNotNull(podData.getSpec().getVolumes().get(1).getSecret());
  assertNull(podData.getSpec().getVolumes().get(1).getPersistentVolumeClaim());

  assertEquals(
      podData.getSpec().getInitContainers().get(0).getVolumeMounts().get(0).getName(),
      "configMap");
  assertEquals(
      podData.getSpec().getContainers().get(0).getVolumeMounts().get(0).getName(), "secret");
}