Skip to content

Commit

Permalink
[FRS-66] Introduce shuffleworker deployed by statefulset
Browse files Browse the repository at this point in the history
  • Loading branch information
Aitozi committed Jun 12, 2022
1 parent 9f6d4e0 commit 9da4f23
Show file tree
Hide file tree
Showing 9 changed files with 530 additions and 24 deletions.
54 changes: 54 additions & 0 deletions shuffle-core/src/main/java/com/alibaba/flink/shuffle/core/config/KubernetesOptions.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,13 +18,18 @@

import com.alibaba.flink.shuffle.common.config.ConfigOption;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.List;
import java.util.Map;

/** This class holds configuration constants used by the remote shuffle deployment. */
public class KubernetesOptions {

private static final Logger LOG = LoggerFactory.getLogger(KubernetesOptions.class);

// --------------------------------------------------------------------------------------------
// Common configurations.
// --------------------------------------------------------------------------------------------
Expand Down Expand Up @@ -277,6 +282,41 @@ public class KubernetesOptions {
+ "value:value1,effect:NoSchedule;key:key2,operator:Exists,"
+ "effect:NoExecute,tolerationSeconds:6000.");

/** The number of shuffler worker. */
public static final ConfigOption<Integer> SHUFFLE_WORKER_REPLICAS =
new ConfigOption<Integer>("remote-shuffle.kubernetes.worker.replicas")
.defaultValue(1)
.description("The number of shuffle worker");

/** The access mode of the shuffle worker's pvc. */
public static final ConfigOption<List<String>> SHUFFLE_WORKER_PVC_ACCESS_MODE =
new ConfigOption<List<String>>("remote-shuffle.kubernetes.worker.pvc.access-mode")
.defaultValue(Collections.singletonList("ReadWriteOnce"))
.description("The access mode of the shuffle worker's pvc");

/** The storage class of the pvc. */
public static final ConfigOption<String> SHUFFLE_WORKER_PVC_STORAGE_CLASS =
new ConfigOption<String>("remote-shuffle.kubernetes.worker.pvc.storage-class")
.defaultValue("alibabacloud-filesystem-ssd")
.description("The storage class of the pvc");

/** The storage size of each pvc. */
public static final ConfigOption<String> SHUFFLE_WORKER_PVC_STORAGE_SIZE =
new ConfigOption<String>("remote-shuffle.kubernetes.worker.pvc.storage.size")
.defaultValue("100Gi")
.description("The storage size of each pvc.");

/** The mount path of the worker pvc. */
public static final ConfigOption<String> SHUFFLE_WORKER_PVC_MOUNT_PATH =
new ConfigOption<String>("remote-shuffle.kubernetes.worker.pvc.mount.path")
.defaultValue("/shuffle-data-pvc")
.description("The mount path of the worker pvc");

public static final ConfigOption<String> SHUFFLE_WORKER_DEPLOY_MODE =
new ConfigOption<String>("remote-shuffle.kubernetes.worker.deploy.mode")
.defaultValue(WorkerMode.DAEMON_SETS.name())
.description("The deploy mode of the shuffle worker");

/**
* The prefix of Kubernetes resource limit factor. It should not be less than 1. The resource
* could be cpu, memory, ephemeral-storage and all other types supported by Kubernetes.
Expand All @@ -291,6 +331,20 @@ public class KubernetesOptions {
public static final String SHUFFLE_WORKER_RESOURCE_LIMIT_FACTOR_PREFIX =
"remote-shuffle.kubernetes.worker.limit-factor.";

/** The mode of the shuffle worker. */
public enum WorkerMode {
STATEFUL_SETS,
DAEMON_SETS;

public static WorkerMode fromString(String value) {
try {
return WorkerMode.valueOf(value);
} catch (Exception e) {
LOG.warn("Failed to parse the worker mode: " + value, e);
return DAEMON_SETS;
}
}
}
// ------------------------------------------------------------------------

/** Not intended to be instantiated. */
Expand Down
36 changes: 35 additions & 1 deletion ...ibaba/flink/shuffle/kubernetes/operator/parameters/KubernetesShuffleWorkerParameters.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import com.alibaba.flink.shuffle.kubernetes.operator.util.Constants;
import com.alibaba.flink.shuffle.kubernetes.operator.util.KubernetesUtils;

import io.fabric8.kubernetes.api.model.Quantity;
import io.fabric8.kubernetes.api.model.apps.DaemonSet;

import java.util.ArrayList;
Expand All @@ -41,7 +42,7 @@
* ShuffleWorkers configuration to kubernetes {@link DaemonSet} configuration.
*/
public class KubernetesShuffleWorkerParameters extends AbstractKubernetesParameters
implements KubernetesDaemonSetParameters {
implements KubernetesDaemonSetParameters, KubernetesStatefulSetParameters {

private final ShuffleWorkerProcessSpec shuffleWorkerProcessSpec;

Expand Down Expand Up @@ -189,8 +190,41 @@ public String getDaemonSetName() {
return KubernetesUtils.getShuffleWorkersNameWithClusterId(getClusterId());
}

@Override
public String getStatefulSetName() {
return KubernetesUtils.getShuffleWorkersNameWithClusterId(getClusterId());
}

@Override
public KubernetesPodParameters getPodTemplateParameters() {
return this;
}

@Override
public int getReplicas() {
return conf.getInteger(KubernetesOptions.SHUFFLE_WORKER_REPLICAS);
}

@Override
public Map<String, Quantity> getStorageResource() {
String size = conf.getString(KubernetesOptions.SHUFFLE_WORKER_PVC_STORAGE_SIZE);
Map<String, Quantity> resource = new HashMap<>();
resource.put(Constants.RESOURCE_NAME_STORAGE, new Quantity(size));
return resource;
}

@Override
public String getStorageClass() {
return conf.getString(KubernetesOptions.SHUFFLE_WORKER_PVC_STORAGE_CLASS);
}

@Override
public List<String> getAccessMode() {
return conf.getList(KubernetesOptions.SHUFFLE_WORKER_PVC_ACCESS_MODE, String.class);
}

@Override
public String getMountPath() {
return conf.getString(KubernetesOptions.SHUFFLE_WORKER_PVC_MOUNT_PATH);
}
}
40 changes: 40 additions & 0 deletions ...alibaba/flink/shuffle/kubernetes/operator/parameters/KubernetesStatefulSetParameters.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
/*
* Copyright 2021 The Flink Remote Shuffle Project
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package com.alibaba.flink.shuffle.kubernetes.operator.parameters;

import io.fabric8.kubernetes.api.model.Quantity;

import java.util.List;
import java.util.Map;

/** The parameters for StatefulSet. */
public interface KubernetesStatefulSetParameters extends KubernetesCommonParameters {

String getStatefulSetName();

KubernetesPodParameters getPodTemplateParameters();

int getReplicas();

Map<String, Quantity> getStorageResource();

String getStorageClass();

List<String> getAccessMode();

String getMountPath();
}
77 changes: 55 additions & 22 deletions ...baba/flink/shuffle/kubernetes/operator/reconciler/RemoteShuffleApplicationReconciler.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,15 +19,18 @@
import com.alibaba.flink.shuffle.common.config.Configuration;
import com.alibaba.flink.shuffle.common.functions.RunnableWithException;
import com.alibaba.flink.shuffle.core.config.ClusterOptions;
import com.alibaba.flink.shuffle.core.config.KubernetesOptions;
import com.alibaba.flink.shuffle.kubernetes.operator.crd.RemoteShuffleApplication;
import com.alibaba.flink.shuffle.kubernetes.operator.parameters.K8sRemoteShuffleFileConfigsParameters;
import com.alibaba.flink.shuffle.kubernetes.operator.parameters.KubernetesConfigMapParameters;
import com.alibaba.flink.shuffle.kubernetes.operator.parameters.KubernetesDaemonSetParameters;
import com.alibaba.flink.shuffle.kubernetes.operator.parameters.KubernetesShuffleManagerParameters;
import com.alibaba.flink.shuffle.kubernetes.operator.parameters.KubernetesShuffleWorkerParameters;
import com.alibaba.flink.shuffle.kubernetes.operator.parameters.KubernetesStatefulSetParameters;
import com.alibaba.flink.shuffle.kubernetes.operator.resources.KubernetesConfigMapBuilder;
import com.alibaba.flink.shuffle.kubernetes.operator.resources.KubernetesDaemonSetBuilder;
import com.alibaba.flink.shuffle.kubernetes.operator.resources.KubernetesDeploymentBuilder;
import com.alibaba.flink.shuffle.kubernetes.operator.resources.KubernetesStatefulSetBuilder;
import com.alibaba.flink.shuffle.kubernetes.operator.util.Constants;
import com.alibaba.flink.shuffle.kubernetes.operator.util.KubernetesInternalOptions;
import com.alibaba.flink.shuffle.kubernetes.operator.util.KubernetesUtils;
Expand All @@ -36,6 +39,7 @@
import io.fabric8.kubernetes.api.model.HasMetadata;
import io.fabric8.kubernetes.api.model.apps.DaemonSet;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.api.model.apps.StatefulSet;
import io.fabric8.kubernetes.client.KubernetesClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand All @@ -48,7 +52,7 @@

/**
* The {@link RemoteShuffleApplicationReconciler} is responsible for reconciling {@link
* RemoteShuffleApplication} to the desired state which is described by {@link
* RemoteShuffleApplication} to the desired state which is described by {@code
* RemoteShuffleApplication#spec}.
*/
public class RemoteShuffleApplicationReconciler {
Expand Down Expand Up @@ -130,27 +134,56 @@ private void reconcileShuffleManager(Configuration configuration, HasMetadata ow
*/
private void reconcileShuffleWorkers(Configuration configuration, HasMetadata owner) {

KubernetesDaemonSetParameters shuffleWorkerParameters =
new KubernetesShuffleWorkerParameters(configuration);
DaemonSet daemonSet =
new KubernetesDaemonSetBuilder()
.buildKubernetesResourceFrom(shuffleWorkerParameters);
KubernetesUtils.setOwnerReference(daemonSet, owner);

LOG.info("Reconcile shuffle workers {}.", daemonSet.getMetadata().getName());
executeReconcileWithRetry(
() -> {
LOG.debug("Try to create or update DaemonSet {}.", daemonSet.toString());
this.kubeClient
.apps()
.daemonSets()
.inNamespace(
checkNotNull(
configuration.getString(
KubernetesInternalOptions.NAMESPACE)))
.createOrReplace(daemonSet);
},
daemonSet.getMetadata().getName());
KubernetesOptions.WorkerMode mode =
KubernetesOptions.WorkerMode.fromString(
configuration.getString(KubernetesOptions.SHUFFLE_WORKER_DEPLOY_MODE));

switch (mode) {
case STATEFUL_SETS:
KubernetesStatefulSetParameters statefulSetParameters =
new KubernetesShuffleWorkerParameters(configuration);
StatefulSet statefulSet =
KubernetesStatefulSetBuilder.INSTANCE.buildKubernetesResourceFrom(
statefulSetParameters);
KubernetesUtils.setOwnerReference(statefulSet, owner);
LOG.info("Reconcile shuffle workers {}.", statefulSet.getMetadata().getName());
executeReconcileWithRetry(
() -> {
LOG.debug("Try to create or update StatefulSet {}.", statefulSet);
this.kubeClient
.apps()
.statefulSets()
.inNamespace(
checkNotNull(
configuration.getString(
KubernetesInternalOptions.NAMESPACE)))
.createOrReplace(statefulSet);
},
statefulSet.getMetadata().getName());
break;
case DAEMON_SETS:
KubernetesDaemonSetParameters shuffleWorkerParameters =
new KubernetesShuffleWorkerParameters(configuration);
DaemonSet daemonSet =
new KubernetesDaemonSetBuilder()
.buildKubernetesResourceFrom(shuffleWorkerParameters);
KubernetesUtils.setOwnerReference(daemonSet, owner);
LOG.info("Reconcile shuffle workers {}.", daemonSet.getMetadata().getName());
executeReconcileWithRetry(
() -> {
LOG.debug(
"Try to create or update DaemonSet {}.", daemonSet.toString());
this.kubeClient
.apps()
.daemonSets()
.inNamespace(
checkNotNull(
configuration.getString(
KubernetesInternalOptions.NAMESPACE)))
.createOrReplace(daemonSet);
},
daemonSet.getMetadata().getName());
}
}

/**
Expand Down
Loading

0 comments on commit 9da4f23

Please sign in to comment.