Skip to content
This repository has been archived by the owner on Mar 3, 2023. It is now read-only.

Commit

Permalink
[3821] Remove deprecated Host Path, NFS, and EBS support for Kubernet…
Browse files Browse the repository at this point in the history 
…es (#3825)

* Removed deprecated loading of EBS, NFS, and Host Path via topology configurations for Kubernetes.

* Refactored creation of NFS, Host Path, Empty Directory, and Persistent Volume Claims (templates and volumes) for Kubernetes to a hybrid-factory pattern for volume generation.
  • Loading branch information
@surahman
surahman authored May 7, 2022
1 parent c18ff1b commit dc2d9f6
Show file tree
Hide file tree
Showing 5 changed files with 531 additions and 468 deletions.
51 changes: 0 additions & 51 deletions heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/KubernetesContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -69,25 +69,6 @@ public enum KubernetesResourceRequestMode {
public static final String KUBERNETES_VOLUME_TYPE = "heron.kubernetes.volume.type";


// HostPath volume keys
// https://kubernetes.io/docs/concepts/storage/volumes/#hostpath
public static final String KUBERNETES_VOLUME_HOSTPATH_PATH =
"heron.kubernetes.volume.hostPath.path";

// nfs volume keys
// https://kubernetes.io/docs/concepts/storage/volumes/#nfs
public static final String KUBERNETES_VOLUME_NFS_PATH =
"heron.kubernetes.volume.nfs.path";
public static final String KUBERNETES_VOLUME_NFS_SERVER =
"heron.kubernetes.volume.nfs.server";

// awsElasticBlockStore volume keys
// https://kubernetes.io/docs/concepts/storage/volumes/#awselasticblockstore
public static final String KUBERNETES_VOLUME_AWS_EBS_VOLUME_ID =
"heron.kubernetes.volume.awsElasticBlockStore.volumeID";
public static final String KUBERNETES_VOLUME_AWS_EBS_FS_TYPE =
"heron.kubernetes.volume.awsElasticBlockStore.fsType";

// Pod Template ConfigMap: heron.kubernetes.[executor | manager].pod.template
public static final String KUBERNETES_POD_TEMPLATE_LOCATION =
"heron.kubernetes.%s.pod.template";
Expand Down Expand Up @@ -166,38 +147,6 @@ public static KubernetesResourceRequestMode getKubernetesRequestMode(Config conf
config.getStringValue(KUBERNETES_RESOURCE_REQUEST_MODE));
}

static String getVolumeType(Config config) {
return config.getStringValue(KUBERNETES_VOLUME_TYPE);
}

static String getVolumeName(Config config) {
return config.getStringValue(KUBERNETES_VOLUME_NAME);
}

static String getHostPathVolumePath(Config config) {
return config.getStringValue(KUBERNETES_VOLUME_HOSTPATH_PATH);
}

static String getNfsVolumePath(Config config) {
return config.getStringValue(KUBERNETES_VOLUME_NFS_PATH);
}

static String getNfsServer(Config config) {
return config.getStringValue(KUBERNETES_VOLUME_NFS_SERVER);
}

static String getAwsEbsVolumeId(Config config) {
return config.getStringValue(KUBERNETES_VOLUME_AWS_EBS_VOLUME_ID);
}

static String getAwsEbsFsType(Config config) {
return config.getStringValue(KUBERNETES_VOLUME_AWS_EBS_FS_TYPE);
}

static boolean hasVolume(Config config) {
return isNotEmpty(getVolumeType(config));
}

static String getContainerVolumeName(Config config) {
return config.getStringValue(KUBERNETES_CONTAINER_VOLUME_MOUNT_NAME);
}
Expand Down
180 changes: 14 additions & 166 deletions heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/V1Controller.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,6 @@
import io.kubernetes.client.openapi.models.V1ObjectFieldSelector;
import io.kubernetes.client.openapi.models.V1ObjectMeta;
import io.kubernetes.client.openapi.models.V1PersistentVolumeClaim;
import io.kubernetes.client.openapi.models.V1PersistentVolumeClaimBuilder;
import io.kubernetes.client.openapi.models.V1PodSpec;
import io.kubernetes.client.openapi.models.V1PodTemplate;
import io.kubernetes.client.openapi.models.V1PodTemplateSpec;
Expand All @@ -78,9 +77,7 @@
import io.kubernetes.client.openapi.models.V1Status;
import io.kubernetes.client.openapi.models.V1Toleration;
import io.kubernetes.client.openapi.models.V1Volume;
import io.kubernetes.client.openapi.models.V1VolumeBuilder;
import io.kubernetes.client.openapi.models.V1VolumeMount;
import io.kubernetes.client.openapi.models.V1VolumeMountBuilder;
import io.kubernetes.client.util.PatchUtils;
import io.kubernetes.client.util.Yaml;
import okhttp3.Response;
Expand Down Expand Up @@ -656,8 +653,6 @@ private void configurePodSpec(final V1PodTemplateSpec podTemplateSpec, Resource

podSpec.setContainers(containers);

addVolumesIfPresent(podSpec);

mountSecretsAsVolumes(podSpec);
}

Expand Down Expand Up @@ -695,28 +690,6 @@ protected static List<V1Toleration> getTolerations() {
return tolerations;
}

/**
* Adds volume to the <code>Pod Spec</code> that Heron requires. Heron's values taking precedence.
* @param spec <code>Pod Spec</code> to be configured.
*/
@VisibleForTesting
protected void addVolumesIfPresent(final V1PodSpec spec) {
final Config config = getConfiguration();
if (KubernetesContext.hasVolume(config)) {
final V1Volume volumeFromConfig = Volumes.get().create(config);
if (volumeFromConfig != null) {
// Merge volumes. Deduplicate using volume's name with Heron defaults taking precedence.
KubernetesUtils.V1ControllerUtils<V1Volume> utils =
new KubernetesUtils.V1ControllerUtils<>();
spec.setVolumes(
utils.mergeListsDedupe(Collections.singletonList(volumeFromConfig), spec.getVolumes(),
Comparator.comparing(V1Volume::getName), "Pod Template Volumes")
);
LOG.fine("Adding volume: " + volumeFromConfig);
}
}
}

/**
* Adds <code>Volume Mounts</code> for <code>Secrets</code> to a pod.
* @param podSpec <code>Pod Spec</code> to add secrets to.
Expand Down Expand Up @@ -1145,77 +1118,13 @@ protected List<V1PersistentVolumeClaim> createPersistentVolumeClaims(
continue;
}

V1PersistentVolumeClaim claim = new V1PersistentVolumeClaimBuilder()
.withNewMetadata()
.withName(pvc.getKey())
.withLabels(getPersistentVolumeClaimLabels(getTopologyName()))
.endMetadata()
.withNewSpec()
.withStorageClassName("")
.endSpec()
.build();

// Populate PVC options.
for (Map.Entry<KubernetesConstants.VolumeConfigKeys, String> option
: pvc.getValue().entrySet()) {
String optionValue = option.getValue();
switch(option.getKey()) {
case storageClassName:
claim.getSpec().setStorageClassName(optionValue);
break;
case sizeLimit:
claim.getSpec().setResources(
new V1ResourceRequirements()
.putRequestsItem("storage", new Quantity(optionValue)));
break;
case accessModes:
claim.getSpec().setAccessModes(Arrays.asList(optionValue.split(",")));
break;
case volumeMode:
claim.getSpec().setVolumeMode(optionValue);
break;
// Valid ignored options not used in a PVC.
default:
break;
}
}
listOfPVCs.add(claim);
listOfPVCs.add(Volumes.get()
.createPersistentVolumeClaim(pvc.getKey(),
getPersistentVolumeClaimLabels(getTopologyName()), pvc.getValue()));
}
return listOfPVCs;
}

/**
* Generates the <code>Volume Mounts</code> to be placed in the <code>Executor</code>
* and <code>Manager</code> from options on the CLI.
* @param volumeName Name of the <code>Volume</code>.
* @param configs Mapping of <code>Volume</code> option <code>key-value</code> configuration pairs.
* @return A configured <code>V1VolumeMount</code>.
*/
@VisibleForTesting
protected V1VolumeMount createVolumeMountsCLI(final String volumeName,
final Map<KubernetesConstants.VolumeConfigKeys, String> configs) {
final V1VolumeMount volumeMount = new V1VolumeMountBuilder()
.withName(volumeName)
.build();
for (Map.Entry<KubernetesConstants.VolumeConfigKeys, String> config : configs.entrySet()) {
switch (config.getKey()) {
case path:
volumeMount.mountPath(config.getValue());
break;
case subPath:
volumeMount.subPath(config.getValue());
break;
case readOnly:
volumeMount.readOnly(Boolean.parseBoolean(config.getValue()));
break;
default:
break;
}
}

return volumeMount;
}

/**
* Generates the <code>Volume</code>s and <code>Volume Mounts</code> for <code>Persistent Volume Claims</code>s
* to be placed in the <code>Executor</code> and <code>Manager</code> from options on the CLI.
Expand All @@ -1235,16 +1144,9 @@ protected void createVolumeAndMountsPersistentVolumeClaimCLI(
final String claimName = configs.getValue()
.get(KubernetesConstants.VolumeConfigKeys.claimName);
if (claimName != null && !KubernetesConstants.LABEL_ON_DEMAND.equalsIgnoreCase(claimName)) {
volumes.add(
new V1VolumeBuilder()
.withName(volumeName)
.withNewPersistentVolumeClaim()
.withClaimName(claimName)
.endPersistentVolumeClaim()
.build()
);
volumes.add(Volumes.get().createPersistentVolumeClaim(claimName, volumeName));
}
volumeMounts.add(createVolumeMountsCLI(volumeName, configs.getValue()));
volumeMounts.add(Volumes.get().createMount(volumeName, configs.getValue()));
}
}

Expand All @@ -1262,27 +1164,10 @@ protected void createVolumeAndMountsEmptyDirCLI(
for (Map.Entry<String, Map<KubernetesConstants.VolumeConfigKeys, String>> configs
: mapOfOpts.entrySet()) {
final String volumeName = configs.getKey();
final V1Volume volume = new V1VolumeBuilder()
.withName(volumeName)
.withNewEmptyDir()
.endEmptyDir()
.build();

for (Map.Entry<KubernetesConstants.VolumeConfigKeys, String> config
: configs.getValue().entrySet()) {
switch(config.getKey()) {
case medium:
volume.getEmptyDir().medium(config.getValue());
break;
case sizeLimit:
volume.getEmptyDir().sizeLimit(new Quantity(config.getValue()));
break;
default:
break;
}
}
final V1Volume volume = Volumes.get()
.createVolume(Volumes.VolumeType.EmptyDir, volumeName, configs.getValue());
volumes.add(volume);
volumeMounts.add(createVolumeMountsCLI(volumeName, configs.getValue()));
volumeMounts.add(Volumes.get().createMount(volumeName, configs.getValue()));
}
}

Expand All @@ -1300,27 +1185,10 @@ protected void createVolumeAndMountsHostPathCLI(
for (Map.Entry<String, Map<KubernetesConstants.VolumeConfigKeys, String>> configs
: mapOfOpts.entrySet()) {
final String volumeName = configs.getKey();
final V1Volume volume = new V1VolumeBuilder()
.withName(volumeName)
.withNewHostPath()
.endHostPath()
.build();

for (Map.Entry<KubernetesConstants.VolumeConfigKeys, String> config
: configs.getValue().entrySet()) {
switch(config.getKey()) {
case type:
volume.getHostPath().setType(config.getValue());
break;
case pathOnHost:
volume.getHostPath().setPath(config.getValue());
break;
default:
break;
}
}
final V1Volume volume = Volumes.get()
.createVolume(Volumes.VolumeType.HostPath, volumeName, configs.getValue());
volumes.add(volume);
volumeMounts.add(createVolumeMountsCLI(volumeName, configs.getValue()));
volumeMounts.add(Volumes.get().createMount(volumeName, configs.getValue()));
}
}

Expand All @@ -1338,30 +1206,10 @@ protected void createVolumeAndMountsNFSCLI(
for (Map.Entry<String, Map<KubernetesConstants.VolumeConfigKeys, String>> configs
: mapOfOpts.entrySet()) {
final String volumeName = configs.getKey();
final V1Volume volume = new V1VolumeBuilder()
.withName(volumeName)
.withNewNfs()
.endNfs()
.build();

for (Map.Entry<KubernetesConstants.VolumeConfigKeys, String> config
: configs.getValue().entrySet()) {
switch(config.getKey()) {
case server:
volume.getNfs().setServer(config.getValue());
break;
case pathOnNFS:
volume.getNfs().setPath(config.getValue());
break;
case readOnly:
volume.getNfs().setReadOnly(Boolean.parseBoolean(config.getValue()));
break;
default:
break;
}
}
final V1Volume volume = Volumes.get()
.createVolume(Volumes.VolumeType.NetworkFileSystem, volumeName, configs.getValue());
volumes.add(volume);
volumeMounts.add(createVolumeMountsCLI(volumeName, configs.getValue()));
volumeMounts.add(Volumes.get().createMount(volumeName, configs.getValue()));
}
}

Expand Down
Loading

0 comments on commit dc2d9f6

Please sign in to comment.