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

Commit

Permalink
Merge branch '3723-PersistentVolumeClaim-dev' into 3723-PersistentVol…
Browse files Browse the repository at this point in the history 
…umeClaim
  • Loading branch information
@surahman
surahman committed Nov 15, 2021
2 parents d454585 + 650c282 commit 6a449b1
Show file tree
Hide file tree
Showing 6 changed files with 208 additions and 83 deletions.
5 changes: 3 additions & 2 deletions heron/schedulers/src/java/org/apache/heron/scheduler/kubernetes/KubernetesConstants.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -112,11 +112,12 @@ private KubernetesConstants() {
);

enum VolumeClaimTemplateConfigKeys {
claimName,
storageClassName,
sizeLimit,
accessModes,
volumeMode,
path, // Added to container.
subPath, // Added to container.
path, // Added to container.
subPath, // Added to container.
}
}
19 changes: 15 additions & 4 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 @@ -268,11 +268,22 @@ public static boolean getPersistentVolumeClaimDisabled(Config config) {
volumes.put(volumeName, volume);
}

if (KubernetesConstants.VolumeClaimTemplateConfigKeys.storageClassName.equals(key)
&& !matcher.reset(value).matches()) {
/* Validate Claim and Storage Class names.
[1] `claimNameNotOnDemand`: checks for a `claimName` which is not `OnDemand`.
[2] `storageClassName`: Check if it is the provided `option`.
Conditions [1] OR [2] are True, then...
[3] Check for a valid lowercase RFC-1123 pattern.
*/
boolean claimNameNotOnDemand =
KubernetesConstants.VolumeClaimTemplateConfigKeys.claimName.equals(key)
&& !KubernetesConstants.LABEL_ON_DEMAND.equalsIgnoreCase(value);
if ((claimNameNotOnDemand // [1]
||
KubernetesConstants.VolumeClaimTemplateConfigKeys.storageClassName.equals(key)) // [2]
&& !matcher.reset(value).matches()) { // [3]
throw new TopologySubmissionException(
String.format("Storage Class name `%s` does not match lowercase RFC-1123 pattern",
value));
String.format("Option `%s` value `%s` does not match lowercase RFC-1123 pattern",
key, value));
}

volume.put(key, value);
Expand Down
60 changes: 46 additions & 14 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 @@ -78,6 +78,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;
Expand Down Expand Up @@ -519,7 +520,7 @@ private void configurePodSpec(final V1PodTemplateSpec podTemplateSpec,
}

if (!persistentVolumeClaimConfigs.isEmpty()) {
configurePodWithPersistentVolumeClaimMounts(executorContainer);
configurePodWithPersistentVolumeClaimVolumesAndMounts(podSpec, executorContainer);
}

configureExecutorContainer(executorCommand, resource, numberOfInstances, executorContainer);
Expand Down Expand Up @@ -884,6 +885,13 @@ protected List<V1PersistentVolumeClaim> createPersistentVolumeClaims(
for (Map.Entry<String, Map<KubernetesConstants.VolumeClaimTemplateConfigKeys, String>> pvc
: mapOfOpts.entrySet()) {

// Only create claims for `OnDemand` volumes.
final String claimName = pvc.getValue()
.get(KubernetesConstants.VolumeClaimTemplateConfigKeys.claimName);
if (claimName != null && !KubernetesConstants.LABEL_ON_DEMAND.equalsIgnoreCase(claimName)) {
continue;
}

V1PersistentVolumeClaim claim = new V1PersistentVolumeClaimBuilder()
.withNewMetadata()
.withName(pvc.getKey())
Expand Down Expand Up @@ -913,7 +921,7 @@ protected List<V1PersistentVolumeClaim> createPersistentVolumeClaims(
claim.getSpec().setVolumeMode(optionValue);
break;
// Valid ignored options not used in a PVC.
case path: case subPath:
case path: case subPath: case claimName:
break;
default:
throw new TopologySubmissionException(
Expand All @@ -927,13 +935,14 @@ protected List<V1PersistentVolumeClaim> createPersistentVolumeClaims(
}

/**
* Generates the <code>Volume Mounts</code> to be placed in the <code>executor container</code>.
* Generates the <code>Volume</code> and <code>Volume Mounts</code> to be placed in the <code>executor container</code>.
* @param mapConfig Mapping of <code>Volumes</code> to <code>key-value</code> configuration pairs.
* @return A list of configured <code>V1VolumeMount</code>.
* @return A pair of configured lists of <code>V1Volume</code> and <code>V1VolumeMount</code>.
*/
@VisibleForTesting
protected List<V1VolumeMount> createPersistentVolumeClaimVolumeMounts(
protected Pair<List<V1Volume>, List<V1VolumeMount>> createPersistentVolumeClaimVolumesAndMounts(
final Map<String, Map<KubernetesConstants.VolumeClaimTemplateConfigKeys, String>> mapConfig) {
List<V1Volume> volumeList = new LinkedList<>();
List<V1VolumeMount> mountList = new LinkedList<>();
for (Map.Entry<String, Map<KubernetesConstants.VolumeClaimTemplateConfigKeys, String>> configs
: mapConfig.entrySet()) {
Expand All @@ -948,6 +957,19 @@ protected List<V1VolumeMount> createPersistentVolumeClaimVolumeMounts(
String.format("A mount path is required and missing from '%s'", volumeName));
}

// Do not create Volumes for `OnDemand`.
final String claimName = configs.getValue()
.get(KubernetesConstants.VolumeClaimTemplateConfigKeys.claimName);
if (claimName != null && !KubernetesConstants.LABEL_ON_DEMAND.equalsIgnoreCase(claimName)) {
final V1Volume volume = new V1VolumeBuilder()
.withName(volumeName)
.withNewPersistentVolumeClaim()
.withClaimName(claimName)
.endPersistentVolumeClaim()
.build();
volumeList.add(volume);
}

final V1VolumeMountBuilder volumeMount = new V1VolumeMountBuilder()
.withName(volumeName)
.withMountPath(path);
Expand All @@ -956,25 +978,35 @@ protected List<V1VolumeMount> createPersistentVolumeClaimVolumeMounts(
}
mountList.add(volumeMount.build());
}
return mountList;
return new Pair<>(volumeList, mountList);
}

/**
* Makes a call to generate <code>Volume Mounts</code> and then inserts them into the <code>Container</code>.
* @param container All generated <code>V1VolumeMount</code> will be placed in the <code>Container</code>.
* Makes a call to generate <code>Volumes</code> and <code>Volume Mounts</code> and then inserts them.
* @param podSpec All generated <code>V1Volume</code> will be placed in the <code>Pod Spec</code>.
* @param executor All generated <code>V1VolumeMount</code> will be placed in the <code>Container</code>.
*/
@VisibleForTesting
protected void configurePodWithPersistentVolumeClaimMounts(final V1Container container) {
List<V1VolumeMount> volumeMounts =
createPersistentVolumeClaimVolumeMounts(persistentVolumeClaimConfigs);
protected void configurePodWithPersistentVolumeClaimVolumesAndMounts(final V1PodSpec podSpec,
final V1Container executor) {
Pair<List<V1Volume>, List<V1VolumeMount>> volumesAndMounts =
createPersistentVolumeClaimVolumesAndMounts(persistentVolumeClaimConfigs);

// Deduplicate on Names with Persistent Volume Claims taking precedence.

// Deduplicate on Volume Mount names with Persistent Volume Claims taking precedence.
KubernetesUtils.V1ControllerUtils<V1VolumeMount> utilsMounts =
new KubernetesUtils.V1ControllerUtils<>();
container.setVolumeMounts(
utilsMounts.mergeListsDedupe(volumeMounts, container.getVolumeMounts(),
executor.setVolumeMounts(
utilsMounts.mergeListsDedupe(volumesAndMounts.second, executor.getVolumeMounts(),
Comparator.comparing(V1VolumeMount::getName),
"Executor and Persistent Volume Claim Volume Mounts"));

KubernetesUtils.V1ControllerUtils<V1Volume> utilsVolumes =
new KubernetesUtils.V1ControllerUtils<>();
podSpec.setVolumes(
utilsVolumes.mergeListsDedupe(volumesAndMounts.first, podSpec.getVolumes(),
Comparator.comparing(V1Volume::getName),
"Pod and Persistent Volume Claim Volumes"));
}

/**
Expand Down
24 changes: 19 additions & 5 deletions heron/schedulers/tests/java/org/apache/heron/scheduler/kubernetes/KubernetesContextTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -89,30 +89,37 @@ public void testPersistentVolumeClaimDisabled() {
public void testGetVolumeClaimTemplates() {
final String volumeNameOne = "volume-name-one";
final String volumeNameTwo = "volume-name-two";
final String claimName = "OnDeMaNd";
final String keyPattern = KubernetesContext.KUBERNETES_VOLUME_CLAIM_PREFIX + "%s.%s";

final String storageClassField = VolumeClaimTemplateConfigKeys.storageClassName.toString();
final String pathField = VolumeClaimTemplateConfigKeys.path.toString();
final String storageClassField = VolumeClaimTemplateConfigKeys.storageClassName.name();
final String pathField = VolumeClaimTemplateConfigKeys.path.name();
final String claimNameField = VolumeClaimTemplateConfigKeys.claimName.name();
final String expectedStorageClass = "expected-storage-class";
final String storageClassKeyOne = String.format(keyPattern, volumeNameOne, storageClassField);
final String storageClassKeyTwo = String.format(keyPattern, volumeNameTwo, storageClassField);
final String expectedPath = "/path/for/volume/expected";
final String pathKeyOne = String.format(keyPattern, volumeNameOne, pathField);
final String pathKeyTwo = String.format(keyPattern, volumeNameTwo, pathField);
final String claimNameKeyOne = String.format(keyPattern, volumeNameOne, claimNameField);
final String claimNameKeyTwo = String.format(keyPattern, volumeNameTwo, claimNameField);

final Config configPVC = Config.newBuilder()
.put(pathKeyOne, expectedPath)
.put(pathKeyTwo, expectedPath)
.put(claimNameKeyOne, claimName)
.put(claimNameKeyTwo, claimName)
.put(storageClassKeyOne, expectedStorageClass)
.put(storageClassKeyTwo, expectedStorageClass)
.build();

final List<String> expectedKeys = Arrays.asList(volumeNameOne, volumeNameTwo);
final List<VolumeClaimTemplateConfigKeys> expectedOptionsKeys =
Arrays.asList(VolumeClaimTemplateConfigKeys.path,
VolumeClaimTemplateConfigKeys.storageClassName);
VolumeClaimTemplateConfigKeys.storageClassName,
VolumeClaimTemplateConfigKeys.claimName);
final List<String> expectedOptionsValues =
Arrays.asList(expectedPath, expectedStorageClass);
Arrays.asList(expectedPath, expectedStorageClass, claimName);

// List of provided PVC options.
final Map<String, Map<VolumeClaimTemplateConfigKeys, String>> mapOfPVC =
Expand Down Expand Up @@ -164,12 +171,19 @@ public void testGetPersistentVolumeClaimsErrors() {
testCases.add(new TestTuple<>("Invalid Volume Name should trigger exception",
configInvalidVolumeName, "lowercase RFC-1123"));

// Invalid Claim Name.
final Config configInvalidClaimName = Config.newBuilder()
.put(String.format(keyPattern, volumeNameValid, "claimName"), failureValue)
.build();
testCases.add(new TestTuple<>("Invalid Claim Name should trigger exception",
configInvalidClaimName, "Option `claimName`"));

// Invalid Storage Class Name.
final Config configInvalidStorageClassName = Config.newBuilder()
.put(String.format(keyPattern, volumeNameValid, "storageClassName"), failureValue)
.build();
testCases.add(new TestTuple<>("Invalid Storage Class Name should trigger exception",
configInvalidStorageClassName, "Storage Class name"));
configInvalidStorageClassName, "Option `storageClassName`"));

// Testing loop.
for (TestTuple<Config, String> testCase : testCases) {
Expand Down

0 comments on commit 6a449b1

Please sign in to comment.