You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2017/04/28 07:27:53 UTC
[2/2] flink git commit: [FLINK-6336] [mesos] Add mesos placement
constraints
[FLINK-6336] [mesos] Add mesos placement constraints
This adds a HostAttrValueConstraint evaluator for fenzo, allowing tasks
to be placed on agents with matching attributes.
Signed-off-by: Stephen Gran <st...@piksel.com>
This closes #3744.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8ef6926f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8ef6926f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8ef6926f
Branch: refs/heads/master
Commit: 8ef6926f3c735fb4e111379fb1edab5184bc7458
Parents: 4ef9672
Author: Stephen Gran <st...@piksel.com>
Authored: Thu Apr 20 10:22:08 2017 +0100
Committer: Till Rohrmann <tr...@apache.org>
Committed: Fri Apr 28 09:24:39 2017 +0200
----------------------------------------------------------------------
docs/setup/config.md | 2 +
docs/setup/mesos.md | 4 ++
.../clusterframework/LaunchableMesosWorker.java | 2 +-
.../MesosTaskManagerParameters.java | 59 +++++++++++++++++-
.../MesosFlinkResourceManagerTest.java | 3 +-
.../MesosTaskManagerParametersTest.java | 65 ++++++++++++++++++++
6 files changed, 130 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/8ef6926f/docs/setup/config.md
----------------------------------------------------------------------
diff --git a/docs/setup/config.md b/docs/setup/config.md
index d8a7082..250d92f 100644
--- a/docs/setup/config.md
+++ b/docs/setup/config.md
@@ -450,6 +450,8 @@ use the `env.java.opts` setting, which is the `%jvmopts%` variable in the String
- `mesos.initial-tasks`: The initial workers to bring up when the master starts (**DEFAULT**: The number of workers specified at cluster startup).
+- `mesos.constraints.hard.hostattribute`: Constraints for task placement on mesos (**DEFAULT**: None).
+
- `mesos.maximum-failed-tasks`: The maximum number of failed workers before the cluster fails (**DEFAULT**: Number of initial workers).
May be set to -1 to disable this feature.
http://git-wip-us.apache.org/repos/asf/flink/blob/8ef6926f/docs/setup/mesos.md
----------------------------------------------------------------------
diff --git a/docs/setup/mesos.md b/docs/setup/mesos.md
index 62168e7..4c72fbc 100644
--- a/docs/setup/mesos.md
+++ b/docs/setup/mesos.md
@@ -222,6 +222,10 @@ When running Flink with Marathon, the whole Flink cluster including the job mana
`mesos.initial-tasks`: The initial workers to bring up when the master starts (**DEFAULT**: The number of workers specified at cluster startup).
+`mesos.constraints.hard.hostattribute`: Constraints for task placement on mesos based on agent attributes (**DEFAULT**: None).
+Takes a comma-separated list of key:value pairs corresponding to the attributes exposed by the target
+mesos agents. Example: `az:eu-west-1a,series:t2`
+
`mesos.maximum-failed-tasks`: The maximum number of failed workers before the cluster fails (**DEFAULT**: Number of initial workers).
May be set to -1 to disable this feature.
http://git-wip-us.apache.org/repos/asf/flink/blob/8ef6926f/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java
index ff5472e..3d5350a 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/LaunchableMesosWorker.java
@@ -138,7 +138,7 @@ public class LaunchableMesosWorker implements LaunchableTask {
@Override
public List<? extends ConstraintEvaluator> getHardConstraints() {
- return null;
+ return params.constraints();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/8ef6926f/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
index 6064c46..aaab027 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParameters.java
@@ -18,6 +18,9 @@
package org.apache.flink.mesos.runtime.clusterframework;
+import com.netflix.fenzo.ConstraintEvaluator;
+import com.netflix.fenzo.functions.Func1;
+import com.netflix.fenzo.plugins.HostAttrValueConstraint;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
@@ -63,7 +66,12 @@ public class MesosTaskManagerParameters {
public static final ConfigOption<String> MESOS_RM_CONTAINER_VOLUMES =
key("mesos.resourcemanager.tasks.container.volumes")
+ .noDefaultValue();
+
+ public static final ConfigOption<String> MESOS_CONSTRAINTS_HARD_HOSTATTR =
+ key("mesos.constraints.hard.hostattribute")
.noDefaultValue();
+
/**
* Value for {@code MESOS_RESOURCEMANAGER_TASKS_CONTAINER_TYPE} setting. Tells to use the Mesos containerizer.
*/
@@ -82,19 +90,23 @@ public class MesosTaskManagerParameters {
private final ContaineredTaskManagerParameters containeredParameters;
private final List<Protos.Volume> containerVolumes;
+
+ private final List<ConstraintEvaluator> constraints;
public MesosTaskManagerParameters(
double cpus,
ContainerType containerType,
Option<String> containerImageName,
ContaineredTaskManagerParameters containeredParameters,
- List<Protos.Volume> containerVolumes) {
+ List<Protos.Volume> containerVolumes,
+ List<ConstraintEvaluator> constraints) {
this.cpus = cpus;
this.containerType = Preconditions.checkNotNull(containerType);
this.containerImageName = Preconditions.checkNotNull(containerImageName);
this.containeredParameters = Preconditions.checkNotNull(containeredParameters);
this.containerVolumes = Preconditions.checkNotNull(containerVolumes);
+ this.constraints = Preconditions.checkNotNull(constraints);
}
@@ -135,6 +147,13 @@ public class MesosTaskManagerParameters {
return containerVolumes;
}
+ /**
+ * Get the placement constraints
+ */
+ public List<ConstraintEvaluator> constraints() {
+ return constraints;
+ }
+
@Override
public String toString() {
return "MesosTaskManagerParameters{" +
@@ -143,6 +162,7 @@ public class MesosTaskManagerParameters {
", containerImageName=" + containerImageName +
", containeredParameters=" + containeredParameters +
", containerVolumes=" + containerVolumes +
+ ", constraints=" + constraints +
'}';
}
@@ -152,6 +172,7 @@ public class MesosTaskManagerParameters {
*/
public static MesosTaskManagerParameters create(Configuration flinkConfig) {
+ List<ConstraintEvaluator> constraints = parseConstraints(flinkConfig.getString(MESOS_CONSTRAINTS_HARD_HOSTATTR));
// parse the common parameters
ContaineredTaskManagerParameters containeredParameters = ContaineredTaskManagerParameters.create(
flinkConfig,
@@ -191,10 +212,42 @@ public class MesosTaskManagerParameters {
cpus,
containerType,
Option.apply(imageName),
- containeredParameters,
- containerVolumes);
+ containeredParameters,
+ containerVolumes,
+ constraints);
+ }
+
+ private static List<ConstraintEvaluator> parseConstraints(String mesosConstraints) {
+
+ if (mesosConstraints == null || mesosConstraints.isEmpty()) {
+ return Collections.emptyList();
+ } else {
+ List<ConstraintEvaluator> constraints = new ArrayList<>();
+
+ for (String constraint : mesosConstraints.split(",")) {
+ if (constraint.isEmpty()) {
+ continue;
+ }
+ final String[] constraintList = constraint.split(":");
+ if (constraintList.length != 2) {
+ continue;
+ }
+ addHostAttrValueConstraint(constraints, constraintList[0], constraintList[1]);
+ }
+
+ return constraints;
+ }
}
+ private static void addHostAttrValueConstraint(List<ConstraintEvaluator> constraints, String constraintKey, final String constraintValue) {
+ constraints.add(new HostAttrValueConstraint(constraintKey, new Func1<String, String>() {
+ @Override
+ public String call(String s) {
+ return constraintValue;
+ }
+ }));
+ }
+
/**
* Used to build volume specs for mesos. This allows for mounting additional volumes into a container
*
http://git-wip-us.apache.org/repos/asf/flink/blob/8ef6926f/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosFlinkResourceManagerTest.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosFlinkResourceManagerTest.java b/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosFlinkResourceManagerTest.java
index 745d35f..7915e8a 100644
--- a/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosFlinkResourceManagerTest.java
+++ b/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosFlinkResourceManagerTest.java
@@ -24,6 +24,7 @@ import akka.actor.ActorSystem;
import akka.testkit.JavaTestKit;
import akka.testkit.TestActorRef;
import akka.testkit.TestProbe;
+import com.netflix.fenzo.ConstraintEvaluator;
import junit.framework.AssertionFailedError;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
@@ -200,7 +201,7 @@ public class MesosFlinkResourceManagerTest extends TestLogger {
ContaineredTaskManagerParameters containeredParams =
new ContaineredTaskManagerParameters(1024, 768, 256, 4, new HashMap<String, String>());
MesosTaskManagerParameters tmParams = new MesosTaskManagerParameters(
- 1.0, MesosTaskManagerParameters.ContainerType.MESOS, Option.<String>empty(), containeredParams, new ArrayList<Protos.Volume>());
+ 1.0, MesosTaskManagerParameters.ContainerType.MESOS, Option.<String>empty(), containeredParams, Collections.<Protos.Volume>emptyList(), Collections.<ConstraintEvaluator>emptyList());
TestActorRef<TestingMesosFlinkResourceManager> resourceManagerRef =
TestActorRef.create(system, MesosFlinkResourceManager.createActorProps(
http://git-wip-us.apache.org/repos/asf/flink/blob/8ef6926f/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParametersTest.java
----------------------------------------------------------------------
diff --git a/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParametersTest.java b/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParametersTest.java
index c4a7bcb..bc303e9 100644
--- a/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParametersTest.java
+++ b/flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosTaskManagerParametersTest.java
@@ -18,6 +18,9 @@
package org.apache.flink.mesos.runtime.clusterframework;
+import com.netflix.fenzo.ConstraintEvaluator;
+import com.netflix.fenzo.functions.Func1;
+import com.netflix.fenzo.plugins.HostAttrValueConstraint;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.TestLogger;
import org.apache.mesos.Protos;
@@ -26,6 +29,7 @@ import scala.Option;
import java.util.List;
+import static org.hamcrest.core.Is.is;
import static org.junit.Assert.*;
public class MesosTaskManagerParametersTest extends TestLogger {
@@ -73,5 +77,66 @@ public class MesosTaskManagerParametersTest extends TestLogger {
assertEquals("/host/path", params.containerVolumes().get(0).getHostPath());
assertEquals(Protos.Volume.Mode.RO, params.containerVolumes().get(0).getMode());
}
+
+ @Test
+ public void givenTwoConstraintsInConfigShouldBeParsed() throws Exception {
+
+ MesosTaskManagerParameters mesosTaskManagerParameters = MesosTaskManagerParameters.create(withHardHostAttrConstraintConfiguration("cluster:foo,az:eu-west-1"));
+ assertThat(mesosTaskManagerParameters.constraints().size(), is(2));
+ ConstraintEvaluator firstConstraintEvaluator = new HostAttrValueConstraint("cluster", new Func1<String, String>() {
+ @Override
+ public String call(String s) {
+ return "foo";
+ }
+ });
+ ConstraintEvaluator secondConstraintEvaluator = new HostAttrValueConstraint("az", new Func1<String, String>() {
+ @Override
+ public String call(String s) {
+ return "foo";
+ }
+ });
+ assertThat(mesosTaskManagerParameters.constraints().get(0).getName(), is(firstConstraintEvaluator.getName()));
+ assertThat(mesosTaskManagerParameters.constraints().get(1).getName(), is(secondConstraintEvaluator.getName()));
+
+ }
+
+ @Test
+ public void givenOneConstraintInConfigShouldBeParsed() throws Exception {
+
+ MesosTaskManagerParameters mesosTaskManagerParameters = MesosTaskManagerParameters.create(withHardHostAttrConstraintConfiguration("cluster:foo"));
+ assertThat(mesosTaskManagerParameters.constraints().size(), is(1));
+ ConstraintEvaluator firstConstraintEvaluator = new HostAttrValueConstraint("cluster", new Func1<String, String>() {
+ @Override
+ public String call(String s) {
+ return "foo";
+ }
+ });
+ assertThat(mesosTaskManagerParameters.constraints().get(0).getName(), is(firstConstraintEvaluator.getName()));
+ }
+
+ @Test
+ public void givenEmptyConstraintInConfigShouldBeParsed() throws Exception {
+
+ MesosTaskManagerParameters mesosTaskManagerParameters = MesosTaskManagerParameters.create(withHardHostAttrConstraintConfiguration(""));
+ assertThat(mesosTaskManagerParameters.constraints().size(), is(0));
+ }
+
+ @Test
+ public void givenInvalidConstraintInConfigShouldBeParsed() throws Exception {
+
+ MesosTaskManagerParameters mesosTaskManagerParameters = MesosTaskManagerParameters.create(withHardHostAttrConstraintConfiguration(",:,"));
+ assertThat(mesosTaskManagerParameters.constraints().size(), is(0));
+ }
+
+
+ private static Configuration withHardHostAttrConstraintConfiguration(final String configuration) {
+ return new Configuration() {
+ private static final long serialVersionUID = -3249384117909445760L;
+
+ {
+ setString(MesosTaskManagerParameters.MESOS_CONSTRAINTS_HARD_HOSTATTR, configuration);
+ }
+ };
+ }
}