You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by GitBox <gi...@apache.org> on 2020/02/25 02:37:39 UTC
[GitHub] [storm] bipinprasad opened a new pull request #3215: Storm3585 -
New compact Constraint config including maxCoLocationCnt
bipinprasad opened a new pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215
Config.TOPOLOGY_RAS_CONSTRAINTS for use by ConstraintSolverStrategy can be specified in a new
compact format that will allow specification of MaxCoLocationCnt.
Config.TOPOLOGY_SPREAD_COMPONENTS is deprecated (and is equivalent to specifying 1 as the MaxCoLocationCnt for the components.
New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other
incompatible components and an optional number that specifies the maximum co-location count for the component on a node, example:
<pr/>
{ "comp-1": [ 2, "comp-2", "comp-3" ], # comp-1 cannot exist on same node as comp-2 or comp-3, and at most 2 comp-1 same node
"comp-2": [ "comp-4" ] # comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389802244
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,80 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads Map of component and its maxCoLocationCnt
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, Map<String, Integer> spreads, Map<String, Object> config) {
+ if (consolidatedConfigFlag) {
+ // single configuration for each component
+ Map<String, Map<String,Object>> modifiedConstraints = new HashMap<>();
+ for (List<String> constraint: constraints) {
+ if (constraint.size() < 2) {
+ continue;
+ }
+ String comp = constraint.get(0);
+ List<String> others = constraint.subList(1, constraint.size());
+ List<Object> incompatibleComponents = (List<Object>)modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>())
+ .computeIfAbsent(ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, k -> new ArrayList<Object>());
+ incompatibleComponents.addAll(others);
+ }
+ for (String comp: spreads.keySet()) {
+ modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>()).put(ConstraintSolverStrategy.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, ""+spreads.get(comp));
+ }
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, modifiedConstraints);
+ } else {
+ // constraint and MaxCoLocationCnts are separate - no maxCoLocationCnt implied as 1
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, constraints);
+ config.put(Config.TOPOLOGY_SPREAD_COMPONENTS, spreads.entrySet().stream().map(e -> Arrays.asList()));
+ }
+ }
+
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidated is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidated is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads List of components that can have only one Executor on a node (i.e. their maxCoLocationCnt = 1)
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, List<String> spreads, Map<String, Object> config) {
Review comment:
Somehow this "should' be used otherwise an edge test case will not be properly handled. Removing it for now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] kishorvpatil commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
kishorvpatil commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r384639660
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -506,20 +581,21 @@ public SchedulingResult asSchedulingResult() {
return SchedulingResult.success("Fully Scheduled by ConstraintSolverStrategy (" + statesSearched
+ " states traversed in " + timeTakenMillis + "ms, backtracked " + backtracked + " times)");
}
+ state.logNodeCompAssignments();
return SchedulingResult.failure(SchedulingStatus.FAIL_NOT_ENOUGH_RESOURCES,
"Cannot find scheduling that satisfies all constraints (" + statesSearched
+ " states traversed in " + timeTakenMillis + "ms, backtracked " + backtracked + " times)");
}
}
- protected static class SearcherState {
+ protected static final class SearcherState {
final long startTimeMillis;
private final long maxEndTimeMs;
// A map of the worker to the components in the worker to be able to enforce constraints.
- private final Map<WorkerSlot, Set<String>> workerCompAssignment;
+ private final Map<WorkerSlot, Map<String, Integer>> workerCompAssignment;
Review comment:
rename?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387448124
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ TestConfig config = new TestConfig();
+ Collection<Object> passCases = new LinkedList<Object>();
+ Collection<Object> failCases = new LinkedList<Object>();
+
+ List<Object> passCaseListOfList = new ArrayList<>();
+ passCaseListOfList.add(Arrays.asList("comp1", "comp2"));
+ passCaseListOfList.add(Arrays.asList("comp1", "comp3"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp4"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp5"));
+
+ Map<Object, Object> passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)10 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp2", "comp3")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp4", "comp5")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCases.add(passCaseMapOfMap);
+
+ passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp2", "comp3")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
Review comment:
(Integer) can be removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387938501
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -613,14 +743,51 @@ public void backtrack(Map<ExecutorDetails, String> execToComp, RasNode node, Wor
String comp = execToComp.get(exec);
LOG.trace("Backtracking {} {} from {}", exec, comp, workerSlot);
if (okToRemoveFromWorker[execIndex]) {
- workerCompAssignment.get(workerSlot).remove(comp);
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.get(workerSlot);
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) - 1); // decrement assignment count
okToRemoveFromWorker[execIndex] = false;
}
if (okToRemoveFromNode[execIndex]) {
- nodeCompAssignment.get(node).remove(comp);
+ Map<String, Integer> oneMap = nodeCompAssignmentCnts.get(node);
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) - 1); // decrement assignment count
okToRemoveFromNode[execIndex] = false;
}
node.freeSingleExecutor(exec, td);
}
+
+ /**
+ * Use this method to log the current component assignments on the Node.
+ * Useful for debugging and tests.
+ */
+ public void logNodeCompAssignments() {
+ if (nodeCompAssignmentCnts == null || nodeCompAssignmentCnts.isEmpty()) {
+ LOG.info("NodeCompAssignment is empty");
+ return;
+ }
+ StringBuffer sb = new StringBuffer();
+ int cntAllNodes = 0;
+ int cntFilledNodes = 0;
+ for (RasNode node: new TreeSet<RasNode>(nodeCompAssignmentCnts.keySet())) {
Review comment:
fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388515317
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,80 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
Review comment:
typo: `when` --> `is`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r384869964
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -447,22 +518,24 @@ public boolean isExecAssignmentToWorkerValid(WorkerSlot worker, SearcherState st
return retMap;
}
- private ArrayList<ExecutorDetails> getSortedExecs(HashSet<String> spreadComps, Map<String, Map<String, Integer>> constraintMatrix,
+ private ArrayList<ExecutorDetails> getSortedExecs(Map<String, Integer> spreadCompCnts,
+ Map<String, Set<String>> constraintMatrix,
Map<String, Set<ExecutorDetails>> compToExecs) {
ArrayList<ExecutorDetails> retList = new ArrayList<>();
//find number of constraints per component
//Key->Comp Value-># of constraints
- Map<String, Integer> compConstraintCountMap = new HashMap<>();
+ Map<String, Double> compConstraintCountMap = new HashMap<>();
constraintMatrix.forEach((comp, subMatrix) -> {
- int count = subMatrix.values().stream().mapToInt(Number::intValue).sum();
- //check component is declared for spreading
- if (spreadComps.contains(comp)) {
- count++;
+ double count = subMatrix.size();
+ // check if component is declared for spreading
+ if (spreadCompCnts.containsKey(comp)) {
+ // lower (1 and above only) value is most constrained should have higher count
+ count += (1_000.0 / spreadCompCnts.get(comp));
Review comment:
Replaced 1000 with component count.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387886374
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
+ for (String comp2: list) {
+ if (!comps.contains(comp2)) {
+ LOG.warn("{} {} declared for Comp {} is not a valid component", ctype, comp2, comp1);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ if (!maxCoLocationCnts.containsKey(comp1)) {
Review comment:
Fixed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389758936
##########
File path: storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
##########
@@ -850,6 +852,128 @@ public void validateField(String name, Object o) {
}
}
+ public static class CustomIsExactlyOneOfValidators extends Validator {
+ private Class<?>[] subValidators;
+ private List<String> validatorClassNames;
+
+ public CustomIsExactlyOneOfValidators(Map<String, Object> params) {
+ this.subValidators = (Class<?>[]) params.get(ConfigValidationAnnotations.ValidatorParams.VALUE_VALIDATOR_CLASSES);
+ this.validatorClassNames = Arrays.asList(subValidators).stream().map(x -> x.getName()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ return;
+ }
+
+ HashMap<String, Exception> validatorExceptions = new HashMap<>();
+ Set<String> selectedValidators = new HashSet<>();
+ for (Class<?> vv : subValidators) {
+ Object valueValidator;
+ try {
+ valueValidator = vv.getConstructor().newInstance();
+ } catch (Exception ex) {
+ throw new IllegalArgumentException(vv.getName() + " instantiation failure", ex);
+ }
+ if (valueValidator instanceof Validator) {
+ try {
+ ((Validator) valueValidator).validateField(name + " " + vv.getSimpleName() + " value", o);
+ selectedValidators.add(vv.getName());
+ } catch (Exception ex) {
+ // only one will pass, so ignore all validation errors - stored for future use
+ validatorExceptions.put(vv.getName(), ex);
+ }
+ } else {
+ String err = String.format("validator: %s cannot be used in CustomExactlyOneOfValidators to validate values. "
+ + "Individual entry validators must a instance of Validator class", vv.getName());
+ LOG.warn(err);
+ }
+ }
+ // check if one and only one validation succeeded
+ if (selectedValidators.isEmpty()) {
+ String parseErrs = String.join(";\n\t", validatorExceptions.entrySet().stream()
+ .map(e -> String.format("%s:%s", e.getKey(), e.getValue())).collect(Collectors.toList()));
+ String err = String.format("Field %s must be one of %s; parse errors are \n\t%s", name,
+ String.join(", ", validatorClassNames), parseErrs);
+ throw new IllegalArgumentException(err);
+ }
+ if (selectedValidators.size() > 1) {
+ throw new IllegalArgumentException("Field " + name + " must match exactly one of " + String.join(", ", selectedValidators));
+ }
+ }
+ }
+
+ public static class RasConstraintsTypeValidator extends Validator {
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
+
+ public static class RasConstraint {
+ int maxNodeCoLocationCnt = -1;
+ Set<String> incompatibleComponents = new HashSet<>();
+ }
+
+ public Map<String, RasConstraint> rasConstraints = new HashMap<>(); // parsedConstraints
Review comment:
Originally I thought this could be used in testing and parsing. But due to package structure it cant. Removed this class. Check for validity only (also added test for the String in list check.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389810185
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -146,7 +267,43 @@ public void basicUnitTestWithKillAndRecover(ConstraintSolverStrategy cs, int bol
Assert.assertTrue("Assert scheduling topology success " + result, result.isSuccess());
Assert.assertEquals("topo all executors scheduled?", 0, cluster.getUnassignedExecutors(topo).size());
- Assert.assertTrue("Valid Scheduling?", ConstraintSolverStrategy.validateSolution(cluster, topo));
+ Assert.assertTrue("Valid Scheduling?", ConstraintSolverStrategy.validateSolution(cluster, topo, null));
+ }
+
+ @Test
+ public void testNewConstraintFormat() {
+ String s = String.format(
+ "{ \"comp-1\": "
+ + " { \"%s\": 2, "
+ + " \"%s\": [\"comp-2\", \"comp-3\" ] }, "
+ + " \"comp-2\": "
+ + " { \"%s\": [ \"comp-4\" ] }, "
+ + " \"comp-3\": "
+ + " { \"%s\": \"comp-5\" } "
+ + "}",
+ ConstraintSolverStrategy.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT,
+ ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS,
+ ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS,
+ ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS
+ );
+ Object jsonValue = JSONValue.parse(s);
+ Map<String, Object> config = Utils.readDefaultConfig();
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, jsonValue);
+ Set<String> allComps = new HashSet<>();
+ allComps.addAll(Arrays.asList("comp-1", "comp-2", "comp-3", "comp-4", "comp-5"));
+ ConstraintSolverStrategy.ConstraintConfig constraintConfig = new ConstraintSolverStrategy.ConstraintConfig(config, allComps);
+
+ Set<String> expectedSetComp1 = new HashSet<>();
+ expectedSetComp1.addAll(Arrays.asList("comp-2", "comp-3"));
+ Set<String> expectedSetComp2 = new HashSet<>();
+ expectedSetComp2.addAll(Arrays.asList("comp-1", "comp-4"));
+ Set<String> expectedSetComp3 = new HashSet<>();
+ expectedSetComp3.addAll(Arrays.asList("comp-1", "comp-5"));
+ Assert.assertEquals("comp-1 incompatible components", expectedSetComp1, constraintConfig.getIncompatibleComponents().get("comp-1"));
+ Assert.assertEquals("comp-2 incompatible components", expectedSetComp2, constraintConfig.getIncompatibleComponents().get("comp-2"));
+ Assert.assertEquals("comp-3 incompatible components", expectedSetComp3, constraintConfig.getIncompatibleComponents().get("comp-3"));
+ Assert.assertEquals("comp-1 maxNodeCoLocationCnt", (int)2, (int)constraintConfig.getMaxCoLocationCnts().getOrDefault("comp-1", -1));
Review comment:
fixed cast
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387874239
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ TestConfig config = new TestConfig();
+ Collection<Object> passCases = new LinkedList<Object>();
+ Collection<Object> failCases = new LinkedList<Object>();
+
+ List<Object> passCaseListOfList = new ArrayList<>();
+ passCaseListOfList.add(Arrays.asList("comp1", "comp2"));
+ passCaseListOfList.add(Arrays.asList("comp1", "comp3"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp4"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp5"));
+
+ Map<Object, Object> passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)10 },
Review comment:
Removed superflous Integer cast
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388539976
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -239,42 +435,50 @@ public void testScheduleLargeExecutorConstraintCount(int parallelismMultiplier)
scheduler.schedule(topologies, cluster);
boolean scheduleSuccess = isStatusSuccess(cluster.getStatus(topo.getId()));
- LOG.info("testScheduleLargeExecutorCount scheduling {} with {}x executor multiplier", scheduleSuccess ? "succeeds" : "fails",
- parallelismMultiplier);
+ LOG.info("testScheduleLargeExecutorCount scheduling {} with {}x executor multiplier, consolidatedConfigFlag={}",
+ scheduleSuccess ? "succeeds" : "fails", parallelismMultiplier, consolidatedConfigFlag);
Assert.assertTrue(scheduleSuccess);
}
@Test
public void testIntegrationWithRAS() {
- List<List<String>> constraints = new LinkedList<>();
- addContraints("spout-0", "bolt-0", constraints);
- addContraints("bolt-1", "bolt-1", constraints);
- addContraints("bolt-1", "bolt-2", constraints);
- List<String> spread = new LinkedList<>();
- spread.add("spout-0");
+ if (!consolidatedConfigFlag) {
+ LOG.info("Skipping test since bolt-1 maxCoLocationCnt=10 requires consolidatedConfigFlag=true, current={}", consolidatedConfigFlag);
+ return;
+ }
Map<String, Object> config = Utils.readDefaultConfig();
config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, ConstraintSolverStrategy.class.getName());
- config.put(Config.TOPOLOGY_SPREAD_COMPONENTS, spread);
- config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, constraints);
config.put(Config.TOPOLOGY_RAS_CONSTRAINT_MAX_STATE_SEARCH, MAX_TRAVERSAL_DEPTH);
config.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, 100_000);
config.put(Config.TOPOLOGY_PRIORITY, 1);
config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 10);
config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 100);
config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 0.0);
+ List<List<String>> constraints = new LinkedList<>();
+ addConstraints("spout-0", "bolt-0", constraints);
+ // commented out unsatisfiable constraint since there are 300 executors and cannot fit on 30 nodes, added as spread
+ // addConstraints("bolt-1", "bolt-1", constraints);
Review comment:
Remove this if it is not needed?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r384870087
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -506,20 +581,21 @@ public SchedulingResult asSchedulingResult() {
return SchedulingResult.success("Fully Scheduled by ConstraintSolverStrategy (" + statesSearched
+ " states traversed in " + timeTakenMillis + "ms, backtracked " + backtracked + " times)");
}
+ state.logNodeCompAssignments();
return SchedulingResult.failure(SchedulingStatus.FAIL_NOT_ENOUGH_RESOURCES,
"Cannot find scheduling that satisfies all constraints (" + statesSearched
+ " states traversed in " + timeTakenMillis + "ms, backtracked " + backtracked + " times)");
}
}
- protected static class SearcherState {
+ protected static final class SearcherState {
final long startTimeMillis;
private final long maxEndTimeMs;
// A map of the worker to the components in the worker to be able to enforce constraints.
- private final Map<WorkerSlot, Set<String>> workerCompAssignment;
+ private final Map<WorkerSlot, Map<String, Integer>> workerCompAssignment;
Review comment:
Changed names to end in Cnts.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387880032
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
Review comment:
constraint variable here may not be a String, i.e. could be an Integer, depends on how quoted/parsed as a Json string.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388454006
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +42,182 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ // getExecutorToComponent().values() also contains system components
+ this(topo.getConf(), Sets.union(topo.getComponents().keySet(), new HashSet(topo.getExecutorToComponent().values())));
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
Review comment:
`incompatibleComponents` means the component cannot co-exist on the same worker. It does not necessary mean that there can only be at most one of this component on the same node.
Here we don't need to (or shouldn't) differentiate whether `comp1.equals(comp2)` or not and we can deal with it just like others.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r388000583
##########
File path: storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
##########
@@ -850,6 +852,128 @@ public void validateField(String name, Object o) {
}
}
+ public static class CustomIsExactlyOneOfValidators extends Validator {
+ private Class<?>[] subValidators;
+ private List<String> validatorClassNames;
+
+ public CustomIsExactlyOneOfValidators(Map<String, Object> params) {
+ this.subValidators = (Class<?>[]) params.get(ConfigValidationAnnotations.ValidatorParams.VALUE_VALIDATOR_CLASSES);
+ this.validatorClassNames = Arrays.asList(subValidators).stream().map(x -> x.getName()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
Review comment:
Allow null in RasConstraintsTypeValidator and CustomIsExactlyOneOfValidators.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387927220
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -262,28 +367,27 @@ public SchedulingResult schedule(Cluster cluster, TopologyDetails td) {
//get mapping of components to executors
compToExecs = getCompToExecs(execToComp);
- //get topology constraints
- constraintMatrix = getConstraintMap(td, compToExecs.keySet());
Review comment:
Fixed - added system components so they can also be specified in constraints.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r386759562
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -45,64 +45,142 @@
//hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ * { "comp-1": [ 2, "comp-2", "comp-3" ], # comp-1 cannot exist on same node as comp-2 or comp-3, and at most 2 comp-1 same node
Review comment:
Updated Config.java, ConfigValidationAnnotations.java, ConfigValidations.java, ConstraintSolverStrategy and TestConstraintSolverStrategy.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388527286
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,80 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads Map of component and its maxCoLocationCnt
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, Map<String, Integer> spreads, Map<String, Object> config) {
+ if (consolidatedConfigFlag) {
+ // single configuration for each component
+ Map<String, Map<String,Object>> modifiedConstraints = new HashMap<>();
+ for (List<String> constraint: constraints) {
+ if (constraint.size() < 2) {
+ continue;
+ }
+ String comp = constraint.get(0);
+ List<String> others = constraint.subList(1, constraint.size());
+ List<Object> incompatibleComponents = (List<Object>)modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>())
+ .computeIfAbsent(ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, k -> new ArrayList<Object>());
+ incompatibleComponents.addAll(others);
+ }
+ for (String comp: spreads.keySet()) {
+ modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>()).put(ConstraintSolverStrategy.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, ""+spreads.get(comp));
+ }
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, modifiedConstraints);
+ } else {
+ // constraint and MaxCoLocationCnts are separate - no maxCoLocationCnt implied as 1
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, constraints);
+ config.put(Config.TOPOLOGY_SPREAD_COMPONENTS, spreads.entrySet().stream().map(e -> Arrays.asList()));
Review comment:
Will this include the value, which are integers into the array? maybe `new ArrayList<>(spreads.keySet());` is what should be here
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388450899
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +42,182 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
Review comment:
typo: node -> worker
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389921217
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -593,13 +710,26 @@ public ExecutorDetails currentExec() {
return execs.get(execIndex);
}
+ /**
+ * Assign executor to worker and node.
+ * TODO: tryToSchedule is a misnomer, since it always schedules.
+ * Assignment validity check is done before the call to tryToSchedule().
+ *
+ * @param execToComp Mapping from executor to component name.
+ * @param node RasNode on which to schedule.
+ * @param workerSlot WorkerSlot on which to schedule.
+ */
public void tryToSchedule(Map<ExecutorDetails, String> execToComp, RasNode node, WorkerSlot workerSlot) {
ExecutorDetails exec = currentExec();
String comp = execToComp.get(exec);
LOG.trace("Trying assignment of {} {} to {}", exec, comp, workerSlot);
- //It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
- okToRemoveFromWorker[execIndex] = workerCompAssignment.computeIfAbsent(workerSlot, (k) -> new HashSet<>()).add(comp);
- okToRemoveFromNode[execIndex] = nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ // It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.computeIfAbsent(workerSlot, (k) -> new HashMap<>());
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) + 1); // increment assignment count
+ okToRemoveFromWorker[execIndex] = true;
Review comment:
Let me explain my understanding of the original code and please let me know what I am missing.
Let's say we are trying to assign `exec1` (whose component is `comp1`) to `node1`. And let's discuss only `nodeCompAssignment` for simplicity.
First of all, we have `isExecAssignmentToWorkerValid` check before `tryToSchedule`, so there won't be any cases where `comp1` is already in on `node1` in `nodeCompAssignment` map, so `okToRemoveFromNoder[exec1]` is true. When we backtrack, we find it `true` and set it `false`. It looks `okToRemoveFromNoder` is not needed at all.
Secondly, if we don't check `isExecAssignmentToWorkerValid` before `tryToSchedule`, let's say we already have `exec0` (whose component is `comp1`) scheduled on `node1` and now we want to schedule `exec1`.
Now we will have `okToRemoveFromNoder[exec1]=false` since `comp1` is already in `nodeCompAssignment` map. And when we backtrack, since `okToRemoveFromNoder[exec1]=false`, we will not remove `comp1` from `nodeCompAssignment` map since there is a `exec0` (also a `comp1`) scheduled on this `node1`. This is also a correct behavior.
So I don't see a bug in the original code. Please elaborate the bug you observed (asking this just for my knowledge). Thanks.
---
If we change `workerCompAssignment` and `nodeCompAssignment` from `<Map, Set>` to `<Map, Map>`, I believe `okToRemoveFromWorker` and `okToRemoveFromNode` are not needed anymore.
Another solution is to change `okToRemoveFromNode` from `boolean[]` to `int[]` to keep track how many of the same component is on this node so far.
I can't say which solution is better. I believe `okToRemoveFromNode` was introduced for performance but I don't know how much the performance gain is.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387448146
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ TestConfig config = new TestConfig();
+ Collection<Object> passCases = new LinkedList<Object>();
+ Collection<Object> failCases = new LinkedList<Object>();
+
+ List<Object> passCaseListOfList = new ArrayList<>();
+ passCaseListOfList.add(Arrays.asList("comp1", "comp2"));
+ passCaseListOfList.add(Arrays.asList("comp1", "comp3"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp4"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp5"));
+
+ Map<Object, Object> passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)10 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp2", "comp3")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp4", "comp5")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCases.add(passCaseMapOfMap);
+
+ passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp2", "comp3")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp4", "comp5")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCases.add(passCaseMapOfMap);
+
+ passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, "comp2"},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
Review comment:
(Integer) can be removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388510854
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -447,22 +552,24 @@ public boolean isExecAssignmentToWorkerValid(WorkerSlot worker, SearcherState st
return retMap;
}
- private ArrayList<ExecutorDetails> getSortedExecs(HashSet<String> spreadComps, Map<String, Map<String, Integer>> constraintMatrix,
+ private ArrayList<ExecutorDetails> getSortedExecs(Map<String, Integer> spreadCompCnts,
+ Map<String, Set<String>> constraintMatrix,
Map<String, Set<ExecutorDetails>> compToExecs) {
ArrayList<ExecutorDetails> retList = new ArrayList<>();
//find number of constraints per component
//Key->Comp Value-># of constraints
- Map<String, Integer> compConstraintCountMap = new HashMap<>();
+ Map<String, Double> compConstraintCountMap = new HashMap<>();
constraintMatrix.forEach((comp, subMatrix) -> {
- int count = subMatrix.values().stream().mapToInt(Number::intValue).sum();
- //check component is declared for spreading
- if (spreadComps.contains(comp)) {
- count++;
+ double count = subMatrix.size();
+ // check if component is declared for spreading
+ if (spreadCompCnts.containsKey(comp)) {
+ // lower (1 and above only) value is most constrained should have higher count
Review comment:
typo: `most` -> `more`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387930157
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -420,21 +528,24 @@ public boolean isExecAssignmentToWorkerValid(WorkerSlot worker, SearcherState st
//check if exec can be on worker based on user defined component exclusions
String execComp = execToComp.get(exec);
- Set<String> components = state.workerCompAssignment.get(worker);
- if (components != null) {
- Map<String, Integer> subMatrix = constraintMatrix.get(execComp);
- for (String comp : components) {
- if (subMatrix.get(comp) != 0) {
+ Map<String, Integer> compAssignmentCnts = state.workerCompAssignmentCnts.get(worker);
+ if (compAssignmentCnts != null && constraintConfig.incompatibleComponents.containsKey(execComp)) {
+ Set<String> subMatrix = constraintConfig.incompatibleComponents.get(execComp);
+ for (String comp : compAssignmentCnts.keySet()) {
+ if (subMatrix.contains(comp)) {
LOG.trace("{} found {} constraint violation {} on {}", exec, execComp, comp, worker);
return false;
}
}
}
//check if exec satisfy spread
- if (spreadComps.contains(execComp)) {
- if (state.nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).contains(execComp)) {
- LOG.trace("{} Found spread violation {} on node {}", exec, execComp, node.getId());
+ if (constraintConfig.maxCoLocationCnts.containsKey(execComp)) {
+ int coLocationMaxCnt = constraintConfig.maxCoLocationCnts.get(execComp);
+ if (state.nodeCompAssignmentCnts.containsKey(node)
+ && state.nodeCompAssignmentCnts.get(node).getOrDefault(execComp, 0) >= coLocationMaxCnt) {
Review comment:
This check is before assignment.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387902678
##########
File path: storm-client/src/jvm/org/apache/storm/Config.java
##########
@@ -304,11 +307,13 @@
// an error will be thrown by nimbus on topology submission and not by the client prior to submitting
// the topology.
public static final String TOPOLOGY_SCHEDULER_STRATEGY = "topology.scheduler.strategy";
+
/**
- * Declare scheduling constraints for a topology used by the constraint solver strategy. A List of pairs (also a list) of components
- * that cannot coexist in the same worker.
+ * Declare scheduling constraints for a topology used by the constraint solver strategy. The format can be either
Review comment:
Added comments on the config format.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387448105
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ TestConfig config = new TestConfig();
+ Collection<Object> passCases = new LinkedList<Object>();
+ Collection<Object> failCases = new LinkedList<Object>();
+
+ List<Object> passCaseListOfList = new ArrayList<>();
+ passCaseListOfList.add(Arrays.asList("comp1", "comp2"));
+ passCaseListOfList.add(Arrays.asList("comp1", "comp3"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp4"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp5"));
+
+ Map<Object, Object> passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)10 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp2", "comp3")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
Review comment:
(Integer) can be removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388531090
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -146,7 +267,43 @@ public void basicUnitTestWithKillAndRecover(ConstraintSolverStrategy cs, int bol
Assert.assertTrue("Assert scheduling topology success " + result, result.isSuccess());
Assert.assertEquals("topo all executors scheduled?", 0, cluster.getUnassignedExecutors(topo).size());
- Assert.assertTrue("Valid Scheduling?", ConstraintSolverStrategy.validateSolution(cluster, topo));
+ Assert.assertTrue("Valid Scheduling?", ConstraintSolverStrategy.validateSolution(cluster, topo, null));
+ }
+
+ @Test
+ public void testNewConstraintFormat() {
+ String s = String.format(
+ "{ \"comp-1\": "
+ + " { \"%s\": 2, "
+ + " \"%s\": [\"comp-2\", \"comp-3\" ] }, "
+ + " \"comp-2\": "
+ + " { \"%s\": [ \"comp-4\" ] }, "
+ + " \"comp-3\": "
+ + " { \"%s\": \"comp-5\" } "
+ + "}",
+ ConstraintSolverStrategy.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT,
+ ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS,
+ ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS,
+ ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS
+ );
+ Object jsonValue = JSONValue.parse(s);
+ Map<String, Object> config = Utils.readDefaultConfig();
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, jsonValue);
+ Set<String> allComps = new HashSet<>();
+ allComps.addAll(Arrays.asList("comp-1", "comp-2", "comp-3", "comp-4", "comp-5"));
+ ConstraintSolverStrategy.ConstraintConfig constraintConfig = new ConstraintSolverStrategy.ConstraintConfig(config, allComps);
+
+ Set<String> expectedSetComp1 = new HashSet<>();
+ expectedSetComp1.addAll(Arrays.asList("comp-2", "comp-3"));
+ Set<String> expectedSetComp2 = new HashSet<>();
+ expectedSetComp2.addAll(Arrays.asList("comp-1", "comp-4"));
+ Set<String> expectedSetComp3 = new HashSet<>();
+ expectedSetComp3.addAll(Arrays.asList("comp-1", "comp-5"));
+ Assert.assertEquals("comp-1 incompatible components", expectedSetComp1, constraintConfig.getIncompatibleComponents().get("comp-1"));
+ Assert.assertEquals("comp-2 incompatible components", expectedSetComp2, constraintConfig.getIncompatibleComponents().get("comp-2"));
+ Assert.assertEquals("comp-3 incompatible components", expectedSetComp3, constraintConfig.getIncompatibleComponents().get("comp-3"));
+ Assert.assertEquals("comp-1 maxNodeCoLocationCnt", (int)2, (int)constraintConfig.getMaxCoLocationCnts().getOrDefault("comp-1", -1));
Review comment:
nit: space after `(int)`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388516152
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,80 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads Map of component and its maxCoLocationCnt
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, Map<String, Integer> spreads, Map<String, Object> config) {
+ if (consolidatedConfigFlag) {
+ // single configuration for each component
+ Map<String, Map<String,Object>> modifiedConstraints = new HashMap<>();
+ for (List<String> constraint: constraints) {
+ if (constraint.size() < 2) {
+ continue;
+ }
+ String comp = constraint.get(0);
+ List<String> others = constraint.subList(1, constraint.size());
+ List<Object> incompatibleComponents = (List<Object>)modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>())
Review comment:
nit: space after `(List<Object>)`
`String,Object` can be deleted
Object in `new ArrayList<Object>` can be deleted (Line157)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387447935
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ TestConfig config = new TestConfig();
+ Collection<Object> passCases = new LinkedList<Object>();
+ Collection<Object> failCases = new LinkedList<Object>();
+
+ List<Object> passCaseListOfList = new ArrayList<>();
+ passCaseListOfList.add(Arrays.asList("comp1", "comp2"));
+ passCaseListOfList.add(Arrays.asList("comp1", "comp3"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp4"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp5"));
+
+ Map<Object, Object> passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)10 },
Review comment:
(Integer) can be removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389644784
##########
File path: storm-client/src/jvm/org/apache/storm/Config.java
##########
@@ -304,15 +307,38 @@
// an error will be thrown by nimbus on topology submission and not by the client prior to submitting
// the topology.
public static final String TOPOLOGY_SCHEDULER_STRATEGY = "topology.scheduler.strategy";
+
/**
- * Declare scheduling constraints for a topology used by the constraint solver strategy. A List of pairs (also a list) of components
- * that cannot coexist in the same worker.
+ * Declare scheduling constraints for a topology used by the constraint solver strategy. The format can be either
+ * old style (validated by ListOfListOfStringValidator.class or the newer style, which is a list of specific type of
+ * Maps (validated by RasConstraintsTypeValidator.class). The value must be in one or the other format.
+ *
+ * <p>
+ * Old style Config.TOPOLOGY_RAS_CONSTRAINTS (ListOfListOfString) specified a list of components that cannot
+ * co-exist on the same Worker.
+ * </p>
+ *
+ * <p>
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * (which serves the same function as the old style configuration) and optional number that specifies
+ * the maximum co-location count for the component on a node.
+ * </p>
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
Review comment:
fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r391234937
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -593,13 +710,26 @@ public ExecutorDetails currentExec() {
return execs.get(execIndex);
}
+ /**
+ * Assign executor to worker and node.
+ * TODO: tryToSchedule is a misnomer, since it always schedules.
+ * Assignment validity check is done before the call to tryToSchedule().
+ *
+ * @param execToComp Mapping from executor to component name.
+ * @param node RasNode on which to schedule.
+ * @param workerSlot WorkerSlot on which to schedule.
+ */
public void tryToSchedule(Map<ExecutorDetails, String> execToComp, RasNode node, WorkerSlot workerSlot) {
ExecutorDetails exec = currentExec();
String comp = execToComp.get(exec);
LOG.trace("Trying assignment of {} {} to {}", exec, comp, workerSlot);
- //It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
- okToRemoveFromWorker[execIndex] = workerCompAssignment.computeIfAbsent(workerSlot, (k) -> new HashSet<>()).add(comp);
- okToRemoveFromNode[execIndex] = nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ // It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.computeIfAbsent(workerSlot, (k) -> new HashMap<>());
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) + 1); // increment assignment count
+ okToRemoveFromWorker[execIndex] = true;
Review comment:
As we discussed offline, the previous code doesn't have a bug.
okToRemoveFromWorker and okToRemoveFromNode are kept here just for safety.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389870997
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -159,12 +316,30 @@ public void testConstraintSolverForceBacktrack() {
return super.sortByValues(map).descendingMap();
}
};
- basicUnitTestWithKillAndRecover(cs, BACKTRACK_BOLT_PARALLEL);
+ basicUnitTestWithKillAndRecover(cs, BACKTRACK_BOLT_PARALLEL, 2);
+ }
+
+ @Test
+ public void testConstraintSolverForceBacktrackWithSpreadCoLocation() {
Review comment:
Also added Assertion to check for invalid test case (i.e. when maxCoLocationCnt is set to > 1 and testing under old config scenario.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389870011
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -239,42 +435,50 @@ public void testScheduleLargeExecutorConstraintCount(int parallelismMultiplier)
scheduler.schedule(topologies, cluster);
boolean scheduleSuccess = isStatusSuccess(cluster.getStatus(topo.getId()));
- LOG.info("testScheduleLargeExecutorCount scheduling {} with {}x executor multiplier", scheduleSuccess ? "succeeds" : "fails",
- parallelismMultiplier);
+ LOG.info("testScheduleLargeExecutorCount scheduling {} with {}x executor multiplier, consolidatedConfigFlag={}",
+ scheduleSuccess ? "succeeds" : "fails", parallelismMultiplier, consolidatedConfigFlag);
Assert.assertTrue(scheduleSuccess);
}
@Test
public void testIntegrationWithRAS() {
- List<List<String>> constraints = new LinkedList<>();
- addContraints("spout-0", "bolt-0", constraints);
- addContraints("bolt-1", "bolt-1", constraints);
- addContraints("bolt-1", "bolt-2", constraints);
- List<String> spread = new LinkedList<>();
- spread.add("spout-0");
+ if (!consolidatedConfigFlag) {
+ LOG.info("Skipping test since bolt-1 maxCoLocationCnt=10 requires consolidatedConfigFlag=true, current={}", consolidatedConfigFlag);
+ return;
+ }
Map<String, Object> config = Utils.readDefaultConfig();
config.put(Config.TOPOLOGY_SCHEDULER_STRATEGY, ConstraintSolverStrategy.class.getName());
- config.put(Config.TOPOLOGY_SPREAD_COMPONENTS, spread);
- config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, constraints);
config.put(Config.TOPOLOGY_RAS_CONSTRAINT_MAX_STATE_SEARCH, MAX_TRAVERSAL_DEPTH);
config.put(Config.TOPOLOGY_WORKER_MAX_HEAP_SIZE_MB, 100_000);
config.put(Config.TOPOLOGY_PRIORITY, 1);
config.put(Config.TOPOLOGY_COMPONENT_CPU_PCORE_PERCENT, 10);
config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_ONHEAP_MEMORY_MB, 100);
config.put(Config.TOPOLOGY_COMPONENT_RESOURCES_OFFHEAP_MEMORY_MB, 0.0);
+ List<List<String>> constraints = new LinkedList<>();
+ addConstraints("spout-0", "bolt-0", constraints);
+ // commented out unsatisfiable constraint since there are 300 executors and cannot fit on 30 nodes, added as spread
+ // addConstraints("bolt-1", "bolt-1", constraints);
Review comment:
Added back. Constraint is satifisfiable when bolt-1 -> bolt-1 is not interpreted as maxNodeCoLocationCnt (fixed in ConstraintSolverStrategy)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389793701
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +42,182 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ // getExecutorToComponent().values() also contains system components
+ this(topo.getConf(), Sets.union(topo.getComponents().keySet(), new HashSet(topo.getExecutorToComponent().values())));
Review comment:
There is no guarantee in TopologyDetails.class that latter is necessarily a superset. Such assumption should not be made in ConstraintConfig - even if true in most instances.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387454324
##########
File path: storm-client/src/jvm/org/apache/storm/Config.java
##########
@@ -304,11 +307,13 @@
// an error will be thrown by nimbus on topology submission and not by the client prior to submitting
// the topology.
public static final String TOPOLOGY_SCHEDULER_STRATEGY = "topology.scheduler.strategy";
+
/**
- * Declare scheduling constraints for a topology used by the constraint solver strategy. A List of pairs (also a list) of components
- * that cannot coexist in the same worker.
+ * Declare scheduling constraints for a topology used by the constraint solver strategy. The format can be either
Review comment:
We might need more detailed explanations on what this config does.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387880639
##########
File path: storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
##########
@@ -850,6 +852,128 @@ public void validateField(String name, Object o) {
}
}
+ public static class CustomIsExactlyOneOfValidators extends Validator {
+ private Class<?>[] subValidators;
+ private List<String> validatorClassNames;
+
+ public CustomIsExactlyOneOfValidators(Map<String, Object> params) {
+ this.subValidators = (Class<?>[]) params.get(ConfigValidationAnnotations.ValidatorParams.VALUE_VALIDATOR_CLASSES);
+ this.validatorClassNames = Arrays.asList(subValidators).stream().map(x -> x.getName()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
+ }
+
+ HashMap<String, Exception> validatorExceptions = new HashMap<>();
+ Set<String> selectedValidators = new HashSet<>();
+ for (Class<?> vv : subValidators) {
+ Object valueValidator;
+ try {
+ valueValidator = vv.getConstructor().newInstance();
+ } catch (Exception ex) {
+ throw new IllegalArgumentException(vv.getName() + " instantiation failure", ex);
+ }
+ if (valueValidator instanceof Validator) {
+ try {
+ ((Validator) valueValidator).validateField(name + " " + vv.getSimpleName() + " value", o);
+ selectedValidators.add(vv.getName());
+ } catch (Exception ex) {
+ // only one will pass, so ignore all validation errors - stored for future use
+ validatorExceptions.put(vv.getName(), ex);
+ }
+ } else {
+ String err = String.format("validator: %s cannot be used in CustomExactlyOneOfValidators to validate values. "
+ + "Individual entry validators must a instance of Validator class", vv.getName());
+ LOG.warn(err);
+ }
+ }
+ // check if one and only one validation succeeded
+ if (selectedValidators.isEmpty()) {
+ String parseErrs = String.join(";\n\t", validatorExceptions.entrySet().stream()
+ .map(e -> String.format("%s:%s", e.getKey(), e.getValue())).collect(Collectors.toList()));
+ String err = String.format("Field %s must be one of %s; parse errors are \n\t%s", name,
+ String.join(", ", validatorClassNames), parseErrs);
+ throw new IllegalArgumentException(err);
+ }
+ if (selectedValidators.size() > 1) {
+ throw new IllegalArgumentException("Field " + name + " must match exactly one of " + String.join(", ", selectedValidators));
+ }
+ }
+ }
+
+ public static class RasConstraintsTypeValidator extends Validator {
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
+
+ public static class RasConstraint {
+ int maxNodeCoLocationCnt = -1;
+ Set<String> incompatibleComponents = new HashSet<>();
+ }
+
+ public Map<String, RasConstraint> rasConstraints = new HashMap<>(); // parsedConstraints
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
+ }
+ if (!(o instanceof Map)) {
+ throw new IllegalArgumentException(
+ "Field " + name + " must be an Iterable containing only Map of Maps");
+ }
+ Map<String, Object> map1 = (Map<String, Object>)o;
+ for (Map.Entry<String, Object> entry1: map1.entrySet()) {
+ String comp1 = entry1.getKey();
+ Object o2 = entry1.getValue();
+ RasConstraint rasConstraint = new RasConstraint();
+ rasConstraints.put(comp1, rasConstraint);
+ if (!(o2 instanceof Map)) {
+ String err = String.format("Field %s, component %s, expecting constraints Map with keys [\"%s\", \"%s\"], in \"%s\"",
+ name, comp1, CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, o);
+ throw new IllegalArgumentException(err);
+ }
+ Map<String, Object> map2 = (Map<String, Object>)o2;
+ for (Map.Entry<String, Object> entry2: map2.entrySet()) {
+ String constraintType = entry2.getKey();
+ Object o3 = entry2.getValue();
+ switch (constraintType) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int intVal = Integer.parseInt("" + o3);
Review comment:
Thanks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387881828
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
Review comment:
Standard style is not to have a space between cast and variable. Is this not correct?
Changed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389812308
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -159,12 +316,30 @@ public void testConstraintSolverForceBacktrack() {
return super.sortByValues(map).descendingMap();
}
};
- basicUnitTestWithKillAndRecover(cs, BACKTRACK_BOLT_PARALLEL);
+ basicUnitTestWithKillAndRecover(cs, BACKTRACK_BOLT_PARALLEL, 2);
+ }
+
+ @Test
+ public void testConstraintSolverForceBacktrackWithSpreadCoLocation() {
Review comment:
None. Removed testConstraintSolverForceBacktrack().
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389795994
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +42,182 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ // getExecutorToComponent().values() also contains system components
+ this(topo.getConf(), Sets.union(topo.getComponents().keySet(), new HashSet(topo.getExecutorToComponent().values())));
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
Review comment:
Fixed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387881742
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
Review comment:
They should be removed since they are not needed. But we don't have to remove them from other tests in this PR since they are old code and not harmful. They can be done in a separate PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387469225
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -613,14 +743,51 @@ public void backtrack(Map<ExecutorDetails, String> execToComp, RasNode node, Wor
String comp = execToComp.get(exec);
LOG.trace("Backtracking {} {} from {}", exec, comp, workerSlot);
if (okToRemoveFromWorker[execIndex]) {
- workerCompAssignment.get(workerSlot).remove(comp);
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.get(workerSlot);
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) - 1); // decrement assignment count
okToRemoveFromWorker[execIndex] = false;
}
if (okToRemoveFromNode[execIndex]) {
- nodeCompAssignment.get(node).remove(comp);
+ Map<String, Integer> oneMap = nodeCompAssignmentCnts.get(node);
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) - 1); // decrement assignment count
okToRemoveFromNode[execIndex] = false;
}
node.freeSingleExecutor(exec, td);
}
+
+ /**
+ * Use this method to log the current component assignments on the Node.
+ * Useful for debugging and tests.
+ */
+ public void logNodeCompAssignments() {
+ if (nodeCompAssignmentCnts == null || nodeCompAssignmentCnts.isEmpty()) {
+ LOG.info("NodeCompAssignment is empty");
+ return;
+ }
+ StringBuffer sb = new StringBuffer();
+ int cntAllNodes = 0;
+ int cntFilledNodes = 0;
+ for (RasNode node: new TreeSet<RasNode>(nodeCompAssignmentCnts.keySet())) {
Review comment:
`new TreeSet<RasNode>` can be `new TreeSet<>`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387940338
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -593,13 +710,26 @@ public ExecutorDetails currentExec() {
return execs.get(execIndex);
}
+ /**
+ * Assign executor to worker and node.
+ * TODO: tryToSchedule is a misnomer, since it always schedules.
+ * Assignment validity check is done before the call to tryToSchedule().
+ *
+ * @param execToComp Mapping from executor to component name.
+ * @param node RasNode on which to schedule.
+ * @param workerSlot WorkerSlot on which to schedule.
+ */
public void tryToSchedule(Map<ExecutorDetails, String> execToComp, RasNode node, WorkerSlot workerSlot) {
ExecutorDetails exec = currentExec();
String comp = execToComp.get(exec);
LOG.trace("Trying assignment of {} {} to {}", exec, comp, workerSlot);
- //It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
- okToRemoveFromWorker[execIndex] = workerCompAssignment.computeIfAbsent(workerSlot, (k) -> new HashSet<>()).add(comp);
- okToRemoveFromNode[execIndex] = nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ // It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.computeIfAbsent(workerSlot, (k) -> new HashMap<>());
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) + 1); // increment assignment count
+ okToRemoveFromWorker[execIndex] = true;
Review comment:
Yes
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387438072
##########
File path: storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
##########
@@ -850,6 +852,128 @@ public void validateField(String name, Object o) {
}
}
+ public static class CustomIsExactlyOneOfValidators extends Validator {
+ private Class<?>[] subValidators;
+ private List<String> validatorClassNames;
+
+ public CustomIsExactlyOneOfValidators(Map<String, Object> params) {
+ this.subValidators = (Class<?>[]) params.get(ConfigValidationAnnotations.ValidatorParams.VALUE_VALIDATOR_CLASSES);
+ this.validatorClassNames = Arrays.asList(subValidators).stream().map(x -> x.getName()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
+ }
+
+ HashMap<String, Exception> validatorExceptions = new HashMap<>();
+ Set<String> selectedValidators = new HashSet<>();
+ for (Class<?> vv : subValidators) {
+ Object valueValidator;
+ try {
+ valueValidator = vv.getConstructor().newInstance();
+ } catch (Exception ex) {
+ throw new IllegalArgumentException(vv.getName() + " instantiation failure", ex);
+ }
+ if (valueValidator instanceof Validator) {
+ try {
+ ((Validator) valueValidator).validateField(name + " " + vv.getSimpleName() + " value", o);
+ selectedValidators.add(vv.getName());
+ } catch (Exception ex) {
+ // only one will pass, so ignore all validation errors - stored for future use
+ validatorExceptions.put(vv.getName(), ex);
+ }
+ } else {
+ String err = String.format("validator: %s cannot be used in CustomExactlyOneOfValidators to validate values. "
+ + "Individual entry validators must a instance of Validator class", vv.getName());
Review comment:
typo: missing 'be" in "must a instance"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r390723935
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,80 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads Map of component and its maxCoLocationCnt
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, Map<String, Integer> spreads, Map<String, Object> config) {
+ if (consolidatedConfigFlag) {
+ // single configuration for each component
+ Map<String, Map<String,Object>> modifiedConstraints = new HashMap<>();
+ for (List<String> constraint: constraints) {
+ if (constraint.size() < 2) {
+ continue;
+ }
+ String comp = constraint.get(0);
+ List<String> others = constraint.subList(1, constraint.size());
+ List<Object> incompatibleComponents = (List<Object>)modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>())
+ .computeIfAbsent(ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, k -> new ArrayList<Object>());
+ incompatibleComponents.addAll(others);
+ }
+ for (String comp: spreads.keySet()) {
+ modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>()).put(ConstraintSolverStrategy.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, ""+spreads.get(comp));
Review comment:
yes on first
sending in String into Map only.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387450633
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
Review comment:
`(String) constraint` will be better?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387780751
##########
File path: storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
##########
@@ -850,6 +852,128 @@ public void validateField(String name, Object o) {
}
}
+ public static class CustomIsExactlyOneOfValidators extends Validator {
+ private Class<?>[] subValidators;
+ private List<String> validatorClassNames;
+
+ public CustomIsExactlyOneOfValidators(Map<String, Object> params) {
+ this.subValidators = (Class<?>[]) params.get(ConfigValidationAnnotations.ValidatorParams.VALUE_VALIDATOR_CLASSES);
+ this.validatorClassNames = Arrays.asList(subValidators).stream().map(x -> x.getName()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
+ }
+
+ HashMap<String, Exception> validatorExceptions = new HashMap<>();
+ Set<String> selectedValidators = new HashSet<>();
+ for (Class<?> vv : subValidators) {
+ Object valueValidator;
+ try {
+ valueValidator = vv.getConstructor().newInstance();
+ } catch (Exception ex) {
+ throw new IllegalArgumentException(vv.getName() + " instantiation failure", ex);
+ }
+ if (valueValidator instanceof Validator) {
+ try {
+ ((Validator) valueValidator).validateField(name + " " + vv.getSimpleName() + " value", o);
+ selectedValidators.add(vv.getName());
+ } catch (Exception ex) {
+ // only one will pass, so ignore all validation errors - stored for future use
+ validatorExceptions.put(vv.getName(), ex);
+ }
+ } else {
+ String err = String.format("validator: %s cannot be used in CustomExactlyOneOfValidators to validate values. "
+ + "Individual entry validators must a instance of Validator class", vv.getName());
+ LOG.warn(err);
+ }
+ }
+ // check if one and only one validation succeeded
+ if (selectedValidators.isEmpty()) {
+ String parseErrs = String.join(";\n\t", validatorExceptions.entrySet().stream()
+ .map(e -> String.format("%s:%s", e.getKey(), e.getValue())).collect(Collectors.toList()));
+ String err = String.format("Field %s must be one of %s; parse errors are \n\t%s", name,
+ String.join(", ", validatorClassNames), parseErrs);
+ throw new IllegalArgumentException(err);
+ }
+ if (selectedValidators.size() > 1) {
+ throw new IllegalArgumentException("Field " + name + " must match exactly one of " + String.join(", ", selectedValidators));
+ }
+ }
+ }
+
+ public static class RasConstraintsTypeValidator extends Validator {
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
+
+ public static class RasConstraint {
+ int maxNodeCoLocationCnt = -1;
+ Set<String> incompatibleComponents = new HashSet<>();
+ }
+
+ public Map<String, RasConstraint> rasConstraints = new HashMap<>(); // parsedConstraints
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
+ }
+ if (!(o instanceof Map)) {
+ throw new IllegalArgumentException(
+ "Field " + name + " must be an Iterable containing only Map of Maps");
+ }
+ Map<String, Object> map1 = (Map<String, Object>)o;
+ for (Map.Entry<String, Object> entry1: map1.entrySet()) {
+ String comp1 = entry1.getKey();
+ Object o2 = entry1.getValue();
+ RasConstraint rasConstraint = new RasConstraint();
+ rasConstraints.put(comp1, rasConstraint);
+ if (!(o2 instanceof Map)) {
+ String err = String.format("Field %s, component %s, expecting constraints Map with keys [\"%s\", \"%s\"], in \"%s\"",
+ name, comp1, CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, o);
+ throw new IllegalArgumentException(err);
+ }
+ Map<String, Object> map2 = (Map<String, Object>)o2;
+ for (Map.Entry<String, Object> entry2: map2.entrySet()) {
+ String constraintType = entry2.getKey();
+ Object o3 = entry2.getValue();
+ switch (constraintType) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int intVal = Integer.parseInt("" + o3);
Review comment:
o3 is not necessarily a String. May come in as an Integer. In any case, casting is not good practice.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r390722657
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,80 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads Map of component and its maxCoLocationCnt
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, Map<String, Integer> spreads, Map<String, Object> config) {
+ if (consolidatedConfigFlag) {
+ // single configuration for each component
+ Map<String, Map<String,Object>> modifiedConstraints = new HashMap<>();
+ for (List<String> constraint: constraints) {
+ if (constraint.size() < 2) {
+ continue;
+ }
+ String comp = constraint.get(0);
+ List<String> others = constraint.subList(1, constraint.size());
+ List<Object> incompatibleComponents = (List<Object>)modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>())
Review comment:
fix
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387440555
##########
File path: storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
##########
@@ -850,6 +852,128 @@ public void validateField(String name, Object o) {
}
}
+ public static class CustomIsExactlyOneOfValidators extends Validator {
+ private Class<?>[] subValidators;
+ private List<String> validatorClassNames;
+
+ public CustomIsExactlyOneOfValidators(Map<String, Object> params) {
+ this.subValidators = (Class<?>[]) params.get(ConfigValidationAnnotations.ValidatorParams.VALUE_VALIDATOR_CLASSES);
+ this.validatorClassNames = Arrays.asList(subValidators).stream().map(x -> x.getName()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
+ }
+
+ HashMap<String, Exception> validatorExceptions = new HashMap<>();
+ Set<String> selectedValidators = new HashSet<>();
+ for (Class<?> vv : subValidators) {
+ Object valueValidator;
+ try {
+ valueValidator = vv.getConstructor().newInstance();
+ } catch (Exception ex) {
+ throw new IllegalArgumentException(vv.getName() + " instantiation failure", ex);
+ }
+ if (valueValidator instanceof Validator) {
+ try {
+ ((Validator) valueValidator).validateField(name + " " + vv.getSimpleName() + " value", o);
+ selectedValidators.add(vv.getName());
+ } catch (Exception ex) {
+ // only one will pass, so ignore all validation errors - stored for future use
+ validatorExceptions.put(vv.getName(), ex);
+ }
+ } else {
+ String err = String.format("validator: %s cannot be used in CustomExactlyOneOfValidators to validate values. "
+ + "Individual entry validators must a instance of Validator class", vv.getName());
+ LOG.warn(err);
+ }
+ }
+ // check if one and only one validation succeeded
+ if (selectedValidators.isEmpty()) {
+ String parseErrs = String.join(";\n\t", validatorExceptions.entrySet().stream()
+ .map(e -> String.format("%s:%s", e.getKey(), e.getValue())).collect(Collectors.toList()));
+ String err = String.format("Field %s must be one of %s; parse errors are \n\t%s", name,
+ String.join(", ", validatorClassNames), parseErrs);
+ throw new IllegalArgumentException(err);
+ }
+ if (selectedValidators.size() > 1) {
+ throw new IllegalArgumentException("Field " + name + " must match exactly one of " + String.join(", ", selectedValidators));
+ }
+ }
+ }
+
+ public static class RasConstraintsTypeValidator extends Validator {
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
+
+ public static class RasConstraint {
+ int maxNodeCoLocationCnt = -1;
+ Set<String> incompatibleComponents = new HashSet<>();
+ }
+
+ public Map<String, RasConstraint> rasConstraints = new HashMap<>(); // parsedConstraints
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
+ }
+ if (!(o instanceof Map)) {
+ throw new IllegalArgumentException(
+ "Field " + name + " must be an Iterable containing only Map of Maps");
+ }
+ Map<String, Object> map1 = (Map<String, Object>)o;
+ for (Map.Entry<String, Object> entry1: map1.entrySet()) {
+ String comp1 = entry1.getKey();
+ Object o2 = entry1.getValue();
+ RasConstraint rasConstraint = new RasConstraint();
+ rasConstraints.put(comp1, rasConstraint);
+ if (!(o2 instanceof Map)) {
+ String err = String.format("Field %s, component %s, expecting constraints Map with keys [\"%s\", \"%s\"], in \"%s\"",
+ name, comp1, CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, o);
+ throw new IllegalArgumentException(err);
+ }
+ Map<String, Object> map2 = (Map<String, Object>)o2;
+ for (Map.Entry<String, Object> entry2: map2.entrySet()) {
+ String constraintType = entry2.getKey();
+ Object o3 = entry2.getValue();
+ switch (constraintType) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int intVal = Integer.parseInt("" + o3);
Review comment:
maybe `(String) o3` is clearer than `"" + o3`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387453648
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
+ for (String comp2: list) {
+ if (!comps.contains(comp2)) {
+ LOG.warn("{} {} declared for Comp {} is not a valid component", ctype, comp2, comp1);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ if (!maxCoLocationCnts.containsKey(comp1)) {
+ maxCoLocationCnts.put(comp1, 1);
+ }
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
+ }
+ break;
+
+ default:
+ LOG.warn("ConstraintType={} invalid for component={}, valid values are {} and {}, ignoring value={}",
+ ctype, comp1, CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT,
+ CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, constraint);
+ break;
+ }
+ });
+ }
+ });
+ }
+
+ // process Config.TOPOLOGY_SPREAD_COMPONENTS - old style
+ Object obj = conf.get(Config.TOPOLOGY_SPREAD_COMPONENTS);
Review comment:
This means we want `TOPOLOGY_SPREAD_COMPONENTS` to override the `maxCoLocationCnts` from `TOPOLOGY_RAS_CONSTRAINTS`. But it's not very obvious to users unless we have enough comments/docs to explain this.
But having two configs doing similar work is always confusing.
I would suggest
1. deprecate this config
2. for backwards compatibility, when this config is provided by the topology, only use it when `maxCoLocationCnts` for the components is not defined in `TOPOLOGY_RAS_CONSTRAINTS`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] kishorvpatil commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
kishorvpatil commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387757114
##########
File path: storm-client/src/jvm/org/apache/storm/Config.java
##########
@@ -304,11 +307,13 @@
// an error will be thrown by nimbus on topology submission and not by the client prior to submitting
// the topology.
public static final String TOPOLOGY_SCHEDULER_STRATEGY = "topology.scheduler.strategy";
+
/**
- * Declare scheduling constraints for a topology used by the constraint solver strategy. A List of pairs (also a list) of components
- * that cannot coexist in the same worker.
+ * Declare scheduling constraints for a topology used by the constraint solver strategy. The format can be either
+ * old style (validated by ListOfListOfStringValidator.class or the newer style, which is a list of specific type of
+ * Maps (validated by RasConstraintsTypeValidator.class). The value must be in one or the other format.
Review comment:
Some examples of old style and new style configurations might help.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388532141
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -159,12 +316,30 @@ public void testConstraintSolverForceBacktrack() {
return super.sortByValues(map).descendingMap();
}
};
- basicUnitTestWithKillAndRecover(cs, BACKTRACK_BOLT_PARALLEL);
+ basicUnitTestWithKillAndRecover(cs, BACKTRACK_BOLT_PARALLEL, 2);
+ }
+
+ @Test
+ public void testConstraintSolverForceBacktrackWithSpreadCoLocation() {
Review comment:
What's the different between this and `testConstraintSolverForceBacktrack `
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388451458
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +42,182 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ // getExecutorToComponent().values() also contains system components
+ this(topo.getConf(), Sets.union(topo.getComponents().keySet(), new HashSet(topo.getExecutorToComponent().values())));
Review comment:
`Sets.union` doesn't seem necessary since the former is a subset of the latter
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388509017
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -420,21 +528,24 @@ public boolean isExecAssignmentToWorkerValid(WorkerSlot worker, SearcherState st
//check if exec can be on worker based on user defined component exclusions
String execComp = execToComp.get(exec);
- Set<String> components = state.workerCompAssignment.get(worker);
- if (components != null) {
- Map<String, Integer> subMatrix = constraintMatrix.get(execComp);
- for (String comp : components) {
- if (subMatrix.get(comp) != 0) {
+ Map<String, Integer> compAssignmentCnts = state.workerCompAssignmentCnts.get(worker);
+ if (compAssignmentCnts != null && constraintConfig.incompatibleComponents.containsKey(execComp)) {
+ Set<String> subMatrix = constraintConfig.incompatibleComponents.get(execComp);
+ for (String comp : compAssignmentCnts.keySet()) {
+ if (subMatrix.contains(comp)) {
LOG.trace("{} found {} constraint violation {} on {}", exec, execComp, comp, worker);
return false;
}
}
}
//check if exec satisfy spread
- if (spreadComps.contains(execComp)) {
- if (state.nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).contains(execComp)) {
- LOG.trace("{} Found spread violation {} on node {}", exec, execComp, node.getId());
+ if (constraintConfig.maxCoLocationCnts.containsKey(execComp)) {
+ int coLocationMaxCnt = constraintConfig.maxCoLocationCnts.get(execComp);
+ if (state.nodeCompAssignmentCnts.containsKey(node)
+ && state.nodeCompAssignmentCnts.get(node).getOrDefault(execComp, 0) >= coLocationMaxCnt) {
Review comment:
You are right. Thanks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387895854
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
+ for (String comp2: list) {
+ if (!comps.contains(comp2)) {
+ LOG.warn("{} {} declared for Comp {} is not a valid component", ctype, comp2, comp1);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ if (!maxCoLocationCnts.containsKey(comp1)) {
+ maxCoLocationCnts.put(comp1, 1);
+ }
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
+ }
+ break;
+
+ default:
+ LOG.warn("ConstraintType={} invalid for component={}, valid values are {} and {}, ignoring value={}",
+ ctype, comp1, CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT,
+ CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, constraint);
+ break;
+ }
+ });
+ }
+ });
+ }
+
+ // process Config.TOPOLOGY_SPREAD_COMPONENTS - old style
+ Object obj = conf.get(Config.TOPOLOGY_SPREAD_COMPONENTS);
Review comment:
Changed code to use only if maxNodeCoLocationCnt is not already defined in TOPOLOGY_RAS_CONSTRAINTS., warn otherwise.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] kishorvpatil commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
kishorvpatil commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r385330209
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -45,64 +45,142 @@
//hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ * { "comp-1": [ 2, "comp-2", "comp-3" ], # comp-1 cannot exist on same node as comp-2 or comp-3, and at most 2 comp-1 same node
Review comment:
Please correct the comment, @Ethanlm thinks the component exclusion is at worker level
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] kishorvpatil commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
kishorvpatil commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r384640010
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -54,27 +56,76 @@
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.scheduler.resource.normalization.ResourceMetrics;
+@RunWith(Parameterized.class)
public class TestConstraintSolverStrategy {
+ @Parameters
+ public static Object[] data() {
+ return new Object[] { false, true };
+ }
+ //public static Collection<Object[]> data(){
Review comment:
Remove commented code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387455490
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -134,48 +252,51 @@ private static boolean checkConstraintsSatisfied(Cluster cluster, TopologyDetail
return workerToNodes;
}
- private static boolean checkSpreadSchedulingValid(Cluster cluster, TopologyDetails topo) {
+ private static boolean checkSpreadSchedulingValid(Cluster cluster, TopologyDetails topo, ConstraintConfig constraintConfig) {
LOG.info("Checking for a valid scheduling...");
assert (cluster.getAssignmentById(topo.getId()) != null);
- Map<ExecutorDetails, WorkerSlot> result = cluster.getAssignmentById(topo.getId()).getExecutorToSlot();
+ if (constraintConfig == null) {
+ constraintConfig = new ConstraintConfig(topo);
+ }
Map<ExecutorDetails, String> execToComp = topo.getExecutorToComponent();
- Map<WorkerSlot, HashSet<ExecutorDetails>> workerExecMap = new HashMap<>();
- Map<WorkerSlot, HashSet<String>> workerCompMap = new HashMap<>();
- Map<RasNode, HashSet<String>> nodeCompMap = new HashMap<>();
+ Map<String, Map<String, Integer>> nodeCompMap = new HashMap<>(); // this is the critical count
Map<WorkerSlot, RasNode> workerToNodes = workerToNodes(cluster);
boolean ret = true;
- HashSet<String> spreadComps = getSpreadComps(topo);
- for (Map.Entry<ExecutorDetails, WorkerSlot> entry : result.entrySet()) {
+ Map<String, Integer> spreadCompCnts = constraintConfig.maxCoLocationCnts;
+ for (Map.Entry<ExecutorDetails, WorkerSlot> entry : cluster.getAssignmentById(topo.getId()).getExecutorToSlot().entrySet()) {
ExecutorDetails exec = entry.getKey();
+ String comp = execToComp.get(exec);
WorkerSlot worker = entry.getValue();
RasNode node = workerToNodes.get(worker);
-
- if (workerExecMap.computeIfAbsent(worker, (k) -> new HashSet<>()).contains(exec)) {
- LOG.error("Incorrect Scheduling: Found duplicate in scheduling");
- return false;
- }
- workerExecMap.get(worker).add(exec);
- String comp = execToComp.get(exec);
- workerCompMap.computeIfAbsent(worker, (k) -> new HashSet<>()).add(comp);
- if (spreadComps.contains(comp)) {
- if (nodeCompMap.computeIfAbsent(node, (k) -> new HashSet<>()).contains(comp)) {
- LOG.error("Incorrect Scheduling: Spread for Component: {} {} on node {} not satisfied {}",
- comp, exec, node.getId(), nodeCompMap.get(node));
+ String nodeId = node.getId();
+
+ if (spreadCompCnts.containsKey(comp)) {
+ int allowedColocationMaxCnt = spreadCompCnts.get(comp);
+ Map<String, Integer> oneNodeCompMap = nodeCompMap.computeIfAbsent(nodeId, (k) -> new HashMap<>());
+ oneNodeCompMap.put(comp, oneNodeCompMap.getOrDefault(comp, 0) + 1);
+ if (allowedColocationMaxCnt < oneNodeCompMap.get(comp)) {
+ LOG.error("Incorrect Scheduling: MaxCoLocationCnt for Component: {} {} on node {} not satisfied, cnt {} > allowed {}",
+ comp, exec, nodeId, oneNodeCompMap.get(comp), allowedColocationMaxCnt);
ret = false;
}
}
- nodeCompMap.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ }
+ if (!ret) {
+ LOG.error("Incorrect MaxCoLocationCnts: Node-Component-Cnt {}", nodeCompMap);
}
return ret;
}
/**
* Check if resource constraints satisfied.
*/
- private static boolean checkResourcesCorrect(Cluster cluster, TopologyDetails topo) {
+ private static boolean checkResourcesCorrect(Cluster cluster, TopologyDetails topo, ConstraintConfig constraintConfig) {
Review comment:
`constraintConfig` doesn't seem necessary here
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387448197
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ TestConfig config = new TestConfig();
+ Collection<Object> passCases = new LinkedList<Object>();
+ Collection<Object> failCases = new LinkedList<Object>();
+
+ List<Object> passCaseListOfList = new ArrayList<>();
+ passCaseListOfList.add(Arrays.asList("comp1", "comp2"));
+ passCaseListOfList.add(Arrays.asList("comp1", "comp3"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp4"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp5"));
+
+ Map<Object, Object> passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)10 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp2", "comp3")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp4", "comp5")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCases.add(passCaseMapOfMap);
+
+ passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp2", "comp3")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp4", "comp5")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCases.add(passCaseMapOfMap);
+
+ passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, "comp2"},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, "comp4"},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCases.add(passCaseMapOfMap);
+
+ for (Object value : passCases) {
+ config.put(TestConfig.TEST_MAP_CONFIG_9, value);
+ ConfigValidation.validateFields(config, Arrays.asList(TestConfig.class));
+ }
+
+ List<Object> failCaseList = new ArrayList<>();
+ failCaseList.add(Arrays.asList("comp1", Arrays.asList("comp2", "comp3")));
+ failCaseList.add(Arrays.asList("comp3", Arrays.asList("comp4", "comp5")));
+ failCases.add(failCaseList);
+
+ Map<String, Object> failCaseMapOfMap = new HashMap<>();
+ failCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)10 },
Review comment:
(Integer) can be removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387464569
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -420,21 +528,24 @@ public boolean isExecAssignmentToWorkerValid(WorkerSlot worker, SearcherState st
//check if exec can be on worker based on user defined component exclusions
String execComp = execToComp.get(exec);
- Set<String> components = state.workerCompAssignment.get(worker);
- if (components != null) {
- Map<String, Integer> subMatrix = constraintMatrix.get(execComp);
- for (String comp : components) {
- if (subMatrix.get(comp) != 0) {
+ Map<String, Integer> compAssignmentCnts = state.workerCompAssignmentCnts.get(worker);
+ if (compAssignmentCnts != null && constraintConfig.incompatibleComponents.containsKey(execComp)) {
+ Set<String> subMatrix = constraintConfig.incompatibleComponents.get(execComp);
+ for (String comp : compAssignmentCnts.keySet()) {
+ if (subMatrix.contains(comp)) {
LOG.trace("{} found {} constraint violation {} on {}", exec, execComp, comp, worker);
return false;
}
}
}
//check if exec satisfy spread
- if (spreadComps.contains(execComp)) {
- if (state.nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).contains(execComp)) {
- LOG.trace("{} Found spread violation {} on node {}", exec, execComp, node.getId());
+ if (constraintConfig.maxCoLocationCnts.containsKey(execComp)) {
+ int coLocationMaxCnt = constraintConfig.maxCoLocationCnts.get(execComp);
+ if (state.nodeCompAssignmentCnts.containsKey(node)
+ && state.nodeCompAssignmentCnts.get(node).getOrDefault(execComp, 0) >= coLocationMaxCnt) {
Review comment:
Maybe we should change `>=` to `>`?
Here `coLocationMaxCnt` is exclusive. And `coLocationMaxCnt=1` would mean 0 of this component can co-exist on the same node, which doesn't seem right to me
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387778133
##########
File path: storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
##########
@@ -850,6 +852,128 @@ public void validateField(String name, Object o) {
}
}
+ public static class CustomIsExactlyOneOfValidators extends Validator {
+ private Class<?>[] subValidators;
+ private List<String> validatorClassNames;
+
+ public CustomIsExactlyOneOfValidators(Map<String, Object> params) {
+ this.subValidators = (Class<?>[]) params.get(ConfigValidationAnnotations.ValidatorParams.VALUE_VALIDATOR_CLASSES);
+ this.validatorClassNames = Arrays.asList(subValidators).stream().map(x -> x.getName()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
+ }
+
+ HashMap<String, Exception> validatorExceptions = new HashMap<>();
+ Set<String> selectedValidators = new HashSet<>();
+ for (Class<?> vv : subValidators) {
+ Object valueValidator;
+ try {
+ valueValidator = vv.getConstructor().newInstance();
+ } catch (Exception ex) {
+ throw new IllegalArgumentException(vv.getName() + " instantiation failure", ex);
+ }
+ if (valueValidator instanceof Validator) {
+ try {
+ ((Validator) valueValidator).validateField(name + " " + vv.getSimpleName() + " value", o);
+ selectedValidators.add(vv.getName());
+ } catch (Exception ex) {
+ // only one will pass, so ignore all validation errors - stored for future use
+ validatorExceptions.put(vv.getName(), ex);
+ }
+ } else {
+ String err = String.format("validator: %s cannot be used in CustomExactlyOneOfValidators to validate values. "
+ + "Individual entry validators must a instance of Validator class", vv.getName());
Review comment:
Good catch.This was copied message - changed here and two other places in original code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] kishorvpatil commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
kishorvpatil commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r385335111
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -45,64 +45,142 @@
//hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ * { "comp-1": [ 2, "comp-2", "comp-3" ], # comp-1 cannot exist on same node as comp-2 or comp-3, and at most 2 comp-1 same node
Review comment:
The description can be modified to https://github.com/apache/storm/blob/master/storm-client/src/jvm/org/apache/storm/Config.java#L307-L312
Also, validator needs to allow for list or map.
The configuration could be more explicit in mentioning node level collation and component isolation.
`{"comp-1": { "maxNodeCollations" : 2, "incompatibleComponents": [ "comp-2", "comp-3" ]} }`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388522827
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,80 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads Map of component and its maxCoLocationCnt
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, Map<String, Integer> spreads, Map<String, Object> config) {
+ if (consolidatedConfigFlag) {
+ // single configuration for each component
+ Map<String, Map<String,Object>> modifiedConstraints = new HashMap<>();
+ for (List<String> constraint: constraints) {
+ if (constraint.size() < 2) {
+ continue;
+ }
+ String comp = constraint.get(0);
+ List<String> others = constraint.subList(1, constraint.size());
+ List<Object> incompatibleComponents = (List<Object>)modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>())
+ .computeIfAbsent(ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, k -> new ArrayList<Object>());
+ incompatibleComponents.addAll(others);
+ }
+ for (String comp: spreads.keySet()) {
+ modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>()).put(ConstraintSolverStrategy.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, ""+spreads.get(comp));
+ }
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, modifiedConstraints);
+ } else {
+ // constraint and MaxCoLocationCnts are separate - no maxCoLocationCnt implied as 1
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, constraints);
+ config.put(Config.TOPOLOGY_SPREAD_COMPONENTS, spreads.entrySet().stream().map(e -> Arrays.asList()));
+ }
+ }
+
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidated is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidated is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads List of components that can have only one Executor on a node (i.e. their maxCoLocationCnt = 1)
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, List<String> spreads, Map<String, Object> config) {
Review comment:
This method is not used by anyone and can be removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387439682
##########
File path: storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
##########
@@ -850,6 +852,128 @@ public void validateField(String name, Object o) {
}
}
+ public static class CustomIsExactlyOneOfValidators extends Validator {
+ private Class<?>[] subValidators;
+ private List<String> validatorClassNames;
+
+ public CustomIsExactlyOneOfValidators(Map<String, Object> params) {
+ this.subValidators = (Class<?>[]) params.get(ConfigValidationAnnotations.ValidatorParams.VALUE_VALIDATOR_CLASSES);
+ this.validatorClassNames = Arrays.asList(subValidators).stream().map(x -> x.getName()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
Review comment:
Originally when the validation class for `topology.ras.constraints` with ListOfListOfStringValidator, a value of `null` is accpetable.
The null check here will break backwards compatibility. So the question here is: do we want to maintain the backwards compatibility?
One thing I notice is that other validation classes accept `null`. And we rely on `NotNull` annotation if we don't accept null as the value.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r390704047
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,54 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads Map of component and its maxCoLocationCnt
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, Map<String, Integer> spreads, Map<String, Object> config) {
+ if (consolidatedConfigFlag) {
+ // single configuration for each component
+ Map<String, Map<String,Object>> modifiedConstraints = new HashMap<>();
+ for (List<String> constraint: constraints) {
+ if (constraint.size() < 2) {
+ continue;
+ }
+ String comp = constraint.get(0);
+ List<String> others = constraint.subList(1, constraint.size());
+ List<Object> incompatibleComponents = (List<Object>)modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>())
+ .computeIfAbsent(ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, k -> new ArrayList<Object>());
+ incompatibleComponents.addAll(others);
+ }
+ for (String comp: spreads.keySet()) {
+ modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>()).put(ConstraintSolverStrategy.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, ""+spreads.get(comp));
+ }
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, modifiedConstraints);
+ } else {
+ // constraint and MaxCoLocationCnts are separate - no maxCoLocationCnt implied as 1
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, constraints);
+ for (Map.Entry<String, Integer> e: spreads.entrySet()) {
+ if (e.getValue() > 1) {
Review comment:
Maybe this should be != 1? There are a few places like these should be changed if you think this makes sense
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r384870166
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -54,27 +56,76 @@
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.scheduler.resource.normalization.ResourceMetrics;
+@RunWith(Parameterized.class)
public class TestConstraintSolverStrategy {
+ @Parameters
+ public static Object[] data() {
+ return new Object[] { false, true };
+ }
+ //public static Collection<Object[]> data(){
Review comment:
Removed comment lines
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387450414
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
+ for (String comp2: list) {
+ if (!comps.contains(comp2)) {
+ LOG.warn("{} {} declared for Comp {} is not a valid component", ctype, comp2, comp1);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ if (!maxCoLocationCnts.containsKey(comp1)) {
+ maxCoLocationCnts.put(comp1, 1);
+ }
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
+ }
+ break;
+
+ default:
+ LOG.warn("ConstraintType={} invalid for component={}, valid values are {} and {}, ignoring value={}",
+ ctype, comp1, CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT,
+ CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, constraint);
+ break;
+ }
+ });
+ }
Review comment:
When `!comps.contains(comp1)`, it would better to log some warning.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r387968031
##########
File path: storm-client/src/jvm/org/apache/storm/Config.java
##########
@@ -304,15 +307,38 @@
// an error will be thrown by nimbus on topology submission and not by the client prior to submitting
// the topology.
public static final String TOPOLOGY_SCHEDULER_STRATEGY = "topology.scheduler.strategy";
+
/**
- * Declare scheduling constraints for a topology used by the constraint solver strategy. A List of pairs (also a list) of components
- * that cannot coexist in the same worker.
+ * Declare scheduling constraints for a topology used by the constraint solver strategy. The format can be either
+ * old style (validated by ListOfListOfStringValidator.class or the newer style, which is a list of specific type of
+ * Maps (validated by RasConstraintsTypeValidator.class). The value must be in one or the other format.
+ *
+ * <p>
+ * Old style Config.TOPOLOGY_RAS_CONSTRAINTS (ListOfListOfString) specified a list of components that cannot
+ * co-exist on the same Worker.
+ * </p>
+ *
+ * <p>
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * (which serves the same function as the old style configuration) and optional number that specifies
+ * the maximum co-location count for the component on a node.
+ * </p>
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
Review comment:
typo: "node" -> "worker"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r391234937
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -593,13 +710,26 @@ public ExecutorDetails currentExec() {
return execs.get(execIndex);
}
+ /**
+ * Assign executor to worker and node.
+ * TODO: tryToSchedule is a misnomer, since it always schedules.
+ * Assignment validity check is done before the call to tryToSchedule().
+ *
+ * @param execToComp Mapping from executor to component name.
+ * @param node RasNode on which to schedule.
+ * @param workerSlot WorkerSlot on which to schedule.
+ */
public void tryToSchedule(Map<ExecutorDetails, String> execToComp, RasNode node, WorkerSlot workerSlot) {
ExecutorDetails exec = currentExec();
String comp = execToComp.get(exec);
LOG.trace("Trying assignment of {} {} to {}", exec, comp, workerSlot);
- //It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
- okToRemoveFromWorker[execIndex] = workerCompAssignment.computeIfAbsent(workerSlot, (k) -> new HashSet<>()).add(comp);
- okToRemoveFromNode[execIndex] = nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ // It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.computeIfAbsent(workerSlot, (k) -> new HashMap<>());
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) + 1); // increment assignment count
+ okToRemoveFromWorker[execIndex] = true;
Review comment:
As we discussed offline, the previous code doesn't have a bug.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm merged pull request #3215: [STORM-3585] New compact
Constraint config including maxNodeCoLocationCnt and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm merged pull request #3215: [STORM-3585] New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387872435
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
Review comment:
All the tests have it. For consistency, they should all be removed or leave them as is?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r388018463
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
+ for (String comp2: list) {
+ if (!comps.contains(comp2)) {
+ LOG.warn("{} {} declared for Comp {} is not a valid component", ctype, comp2, comp1);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ if (!maxCoLocationCnts.containsKey(comp1)) {
+ maxCoLocationCnts.put(comp1, 1);
+ }
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
+ }
+ break;
+
+ default:
+ LOG.warn("ConstraintType={} invalid for component={}, valid values are {} and {}, ignoring value={}",
+ ctype, comp1, CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT,
+ CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, constraint);
+ break;
+ }
+ });
+ }
+ });
+ }
+
+ // process Config.TOPOLOGY_SPREAD_COMPONENTS - old style
+ Object obj = conf.get(Config.TOPOLOGY_SPREAD_COMPONENTS);
Review comment:
TOPOLOGY_SPREAD_COMPONENTS marked as @Deprecated.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387468864
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -593,13 +710,26 @@ public ExecutorDetails currentExec() {
return execs.get(execIndex);
}
+ /**
+ * Assign executor to worker and node.
+ * TODO: tryToSchedule is a misnomer, since it always schedules.
+ * Assignment validity check is done before the call to tryToSchedule().
+ *
+ * @param execToComp Mapping from executor to component name.
+ * @param node RasNode on which to schedule.
+ * @param workerSlot WorkerSlot on which to schedule.
+ */
public void tryToSchedule(Map<ExecutorDetails, String> execToComp, RasNode node, WorkerSlot workerSlot) {
ExecutorDetails exec = currentExec();
String comp = execToComp.get(exec);
LOG.trace("Trying assignment of {} {} to {}", exec, comp, workerSlot);
- //It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
- okToRemoveFromWorker[execIndex] = workerCompAssignment.computeIfAbsent(workerSlot, (k) -> new HashSet<>()).add(comp);
- okToRemoveFromNode[execIndex] = nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ // It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.computeIfAbsent(workerSlot, (k) -> new HashMap<>());
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) + 1); // increment assignment count
+ okToRemoveFromWorker[execIndex] = true;
Review comment:
Is this fixing some kind of bug?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388456448
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +42,182 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ // getExecutorToComponent().values() also contains system components
+ this(topo.getConf(), Sets.union(topo.getComponents().keySet(), new HashSet(topo.getExecutorToComponent().values())));
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
Review comment:
The meaning of this comment is not very clear to me
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387879241
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
+ for (String comp2: list) {
+ if (!comps.contains(comp2)) {
+ LOG.warn("{} {} declared for Comp {} is not a valid component", ctype, comp2, comp1);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ if (!maxCoLocationCnts.containsKey(comp1)) {
+ maxCoLocationCnts.put(comp1, 1);
+ }
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
+ }
+ break;
+
+ default:
+ LOG.warn("ConstraintType={} invalid for component={}, valid values are {} and {}, ignoring value={}",
+ ctype, comp1, CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT,
+ CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, constraint);
+ break;
+ }
+ });
+ }
Review comment:
Added warning log.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389808363
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,80 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads Map of component and its maxCoLocationCnt
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, Map<String, Integer> spreads, Map<String, Object> config) {
+ if (consolidatedConfigFlag) {
+ // single configuration for each component
+ Map<String, Map<String,Object>> modifiedConstraints = new HashMap<>();
+ for (List<String> constraint: constraints) {
+ if (constraint.size() < 2) {
+ continue;
+ }
+ String comp = constraint.get(0);
+ List<String> others = constraint.subList(1, constraint.size());
+ List<Object> incompatibleComponents = (List<Object>)modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>())
+ .computeIfAbsent(ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, k -> new ArrayList<Object>());
+ incompatibleComponents.addAll(others);
+ }
+ for (String comp: spreads.keySet()) {
+ modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>()).put(ConstraintSolverStrategy.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, ""+spreads.get(comp));
+ }
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, modifiedConstraints);
+ } else {
+ // constraint and MaxCoLocationCnts are separate - no maxCoLocationCnt implied as 1
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, constraints);
+ config.put(Config.TOPOLOGY_SPREAD_COMPONENTS, spreads.entrySet().stream().map(e -> Arrays.asList()));
Review comment:
fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387452475
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
+ for (String comp2: list) {
+ if (!comps.contains(comp2)) {
+ LOG.warn("{} {} declared for Comp {} is not a valid component", ctype, comp2, comp1);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ if (!maxCoLocationCnts.containsKey(comp1)) {
Review comment:
Consider this case:
`{"comp-1", {"incompatibleComponents", "comp-1"}, {"maxNodeCoLocationCnt" , 2}} `
This means we can have at most 2 `comp-1` on the same node, but no more than 1 `comp-1` on the same worker.
The code here mixes worker-level and node-level constraints and the worker-level constraint will be ignored
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387904802
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -134,48 +252,51 @@ private static boolean checkConstraintsSatisfied(Cluster cluster, TopologyDetail
return workerToNodes;
}
- private static boolean checkSpreadSchedulingValid(Cluster cluster, TopologyDetails topo) {
+ private static boolean checkSpreadSchedulingValid(Cluster cluster, TopologyDetails topo, ConstraintConfig constraintConfig) {
LOG.info("Checking for a valid scheduling...");
assert (cluster.getAssignmentById(topo.getId()) != null);
- Map<ExecutorDetails, WorkerSlot> result = cluster.getAssignmentById(topo.getId()).getExecutorToSlot();
+ if (constraintConfig == null) {
+ constraintConfig = new ConstraintConfig(topo);
+ }
Map<ExecutorDetails, String> execToComp = topo.getExecutorToComponent();
- Map<WorkerSlot, HashSet<ExecutorDetails>> workerExecMap = new HashMap<>();
- Map<WorkerSlot, HashSet<String>> workerCompMap = new HashMap<>();
- Map<RasNode, HashSet<String>> nodeCompMap = new HashMap<>();
+ Map<String, Map<String, Integer>> nodeCompMap = new HashMap<>(); // this is the critical count
Map<WorkerSlot, RasNode> workerToNodes = workerToNodes(cluster);
boolean ret = true;
- HashSet<String> spreadComps = getSpreadComps(topo);
- for (Map.Entry<ExecutorDetails, WorkerSlot> entry : result.entrySet()) {
+ Map<String, Integer> spreadCompCnts = constraintConfig.maxCoLocationCnts;
+ for (Map.Entry<ExecutorDetails, WorkerSlot> entry : cluster.getAssignmentById(topo.getId()).getExecutorToSlot().entrySet()) {
ExecutorDetails exec = entry.getKey();
+ String comp = execToComp.get(exec);
WorkerSlot worker = entry.getValue();
RasNode node = workerToNodes.get(worker);
-
- if (workerExecMap.computeIfAbsent(worker, (k) -> new HashSet<>()).contains(exec)) {
- LOG.error("Incorrect Scheduling: Found duplicate in scheduling");
- return false;
- }
- workerExecMap.get(worker).add(exec);
- String comp = execToComp.get(exec);
- workerCompMap.computeIfAbsent(worker, (k) -> new HashSet<>()).add(comp);
- if (spreadComps.contains(comp)) {
- if (nodeCompMap.computeIfAbsent(node, (k) -> new HashSet<>()).contains(comp)) {
- LOG.error("Incorrect Scheduling: Spread for Component: {} {} on node {} not satisfied {}",
- comp, exec, node.getId(), nodeCompMap.get(node));
+ String nodeId = node.getId();
+
+ if (spreadCompCnts.containsKey(comp)) {
+ int allowedColocationMaxCnt = spreadCompCnts.get(comp);
+ Map<String, Integer> oneNodeCompMap = nodeCompMap.computeIfAbsent(nodeId, (k) -> new HashMap<>());
+ oneNodeCompMap.put(comp, oneNodeCompMap.getOrDefault(comp, 0) + 1);
+ if (allowedColocationMaxCnt < oneNodeCompMap.get(comp)) {
+ LOG.error("Incorrect Scheduling: MaxCoLocationCnt for Component: {} {} on node {} not satisfied, cnt {} > allowed {}",
+ comp, exec, nodeId, oneNodeCompMap.get(comp), allowedColocationMaxCnt);
ret = false;
}
}
- nodeCompMap.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ }
+ if (!ret) {
+ LOG.error("Incorrect MaxCoLocationCnts: Node-Component-Cnt {}", nodeCompMap);
}
return ret;
}
/**
* Check if resource constraints satisfied.
*/
- private static boolean checkResourcesCorrect(Cluster cluster, TopologyDetails topo) {
+ private static boolean checkResourcesCorrect(Cluster cluster, TopologyDetails topo, ConstraintConfig constraintConfig) {
Review comment:
removed unused param
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387461422
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -262,28 +367,27 @@ public SchedulingResult schedule(Cluster cluster, TopologyDetails td) {
//get mapping of components to executors
compToExecs = getCompToExecs(execToComp);
- //get topology constraints
- constraintMatrix = getConstraintMap(td, compToExecs.keySet());
Review comment:
This constraintMatrix includes system components, like `Acker` because it gets components from execToComp
But `topo.getComponents().keySet()` in `new ConstraintConfig(td);` only includes non-system components.
I have to say the storm code is confusing. But just want to point out the different between these two that will cause some behavior changes because of the code change here
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388447661
##########
File path: storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
##########
@@ -850,6 +852,128 @@ public void validateField(String name, Object o) {
}
}
+ public static class CustomIsExactlyOneOfValidators extends Validator {
+ private Class<?>[] subValidators;
+ private List<String> validatorClassNames;
+
+ public CustomIsExactlyOneOfValidators(Map<String, Object> params) {
+ this.subValidators = (Class<?>[]) params.get(ConfigValidationAnnotations.ValidatorParams.VALUE_VALIDATOR_CLASSES);
+ this.validatorClassNames = Arrays.asList(subValidators).stream().map(x -> x.getName()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ return;
+ }
+
+ HashMap<String, Exception> validatorExceptions = new HashMap<>();
+ Set<String> selectedValidators = new HashSet<>();
+ for (Class<?> vv : subValidators) {
+ Object valueValidator;
+ try {
+ valueValidator = vv.getConstructor().newInstance();
+ } catch (Exception ex) {
+ throw new IllegalArgumentException(vv.getName() + " instantiation failure", ex);
+ }
+ if (valueValidator instanceof Validator) {
+ try {
+ ((Validator) valueValidator).validateField(name + " " + vv.getSimpleName() + " value", o);
+ selectedValidators.add(vv.getName());
+ } catch (Exception ex) {
+ // only one will pass, so ignore all validation errors - stored for future use
+ validatorExceptions.put(vv.getName(), ex);
+ }
+ } else {
+ String err = String.format("validator: %s cannot be used in CustomExactlyOneOfValidators to validate values. "
+ + "Individual entry validators must a instance of Validator class", vv.getName());
+ LOG.warn(err);
+ }
+ }
+ // check if one and only one validation succeeded
+ if (selectedValidators.isEmpty()) {
+ String parseErrs = String.join(";\n\t", validatorExceptions.entrySet().stream()
+ .map(e -> String.format("%s:%s", e.getKey(), e.getValue())).collect(Collectors.toList()));
+ String err = String.format("Field %s must be one of %s; parse errors are \n\t%s", name,
+ String.join(", ", validatorClassNames), parseErrs);
+ throw new IllegalArgumentException(err);
+ }
+ if (selectedValidators.size() > 1) {
+ throw new IllegalArgumentException("Field " + name + " must match exactly one of " + String.join(", ", selectedValidators));
+ }
+ }
+ }
+
+ public static class RasConstraintsTypeValidator extends Validator {
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
+
+ public static class RasConstraint {
+ int maxNodeCoLocationCnt = -1;
+ Set<String> incompatibleComponents = new HashSet<>();
+ }
+
+ public Map<String, RasConstraint> rasConstraints = new HashMap<>(); // parsedConstraints
Review comment:
Line 911-916 doesn't seem necessary? Or at least it should be private
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389789482
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +42,182 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
Review comment:
Fixed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] kishorvpatil commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
kishorvpatil commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r384636994
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -447,22 +518,24 @@ public boolean isExecAssignmentToWorkerValid(WorkerSlot worker, SearcherState st
return retMap;
}
- private ArrayList<ExecutorDetails> getSortedExecs(HashSet<String> spreadComps, Map<String, Map<String, Integer>> constraintMatrix,
+ private ArrayList<ExecutorDetails> getSortedExecs(Map<String, Integer> spreadCompCnts,
+ Map<String, Set<String>> constraintMatrix,
Map<String, Set<ExecutorDetails>> compToExecs) {
ArrayList<ExecutorDetails> retList = new ArrayList<>();
//find number of constraints per component
//Key->Comp Value-># of constraints
- Map<String, Integer> compConstraintCountMap = new HashMap<>();
+ Map<String, Double> compConstraintCountMap = new HashMap<>();
constraintMatrix.forEach((comp, subMatrix) -> {
- int count = subMatrix.values().stream().mapToInt(Number::intValue).sum();
- //check component is declared for spreading
- if (spreadComps.contains(comp)) {
- count++;
+ double count = subMatrix.size();
+ // check if component is declared for spreading
+ if (spreadCompCnts.containsKey(comp)) {
+ // lower (1 and above only) value is most constrained should have higher count
+ count += (1_000.0 / spreadCompCnts.get(comp));
Review comment:
Instead of using large number this should be function on total executors requested by topology.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387446344
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
Review comment:
All the exceptions in the signature can be removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r390718868
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,54 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads Map of component and its maxCoLocationCnt
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, Map<String, Integer> spreads, Map<String, Object> config) {
+ if (consolidatedConfigFlag) {
+ // single configuration for each component
+ Map<String, Map<String,Object>> modifiedConstraints = new HashMap<>();
+ for (List<String> constraint: constraints) {
+ if (constraint.size() < 2) {
+ continue;
+ }
+ String comp = constraint.get(0);
+ List<String> others = constraint.subList(1, constraint.size());
+ List<Object> incompatibleComponents = (List<Object>)modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>())
+ .computeIfAbsent(ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, k -> new ArrayList<Object>());
+ incompatibleComponents.addAll(others);
+ }
+ for (String comp: spreads.keySet()) {
+ modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>()).put(ConstraintSolverStrategy.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, ""+spreads.get(comp));
+ }
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, modifiedConstraints);
+ } else {
+ // constraint and MaxCoLocationCnts are separate - no maxCoLocationCnt implied as 1
+ config.put(Config.TOPOLOGY_RAS_CONSTRAINTS, constraints);
+ for (Map.Entry<String, Integer> e: spreads.entrySet()) {
+ if (e.getValue() > 1) {
Review comment:
maxNodeCoLocationCnt is set to 1 or higher only. != 1 and >1 mean the same here. No change required.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388513068
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -54,27 +58,70 @@
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.scheduler.resource.normalization.ResourceMetrics;
+@RunWith(Parameterized.class)
public class TestConstraintSolverStrategy {
+ @Parameters
+ public static Object[] data() {
+ return new Object[] { false, true };
+ }
+
private static final Logger LOG = LoggerFactory.getLogger(TestConstraintSolverStrategy.class);
private static final int MAX_TRAVERSAL_DEPTH = 2000;
private static final int NORMAL_BOLT_PARALLEL = 11;
//Dropping the parallelism of the bolts to 3 instead of 11 so we can find a solution in a reasonable amount of work when backtracking.
private static final int BACKTRACK_BOLT_PARALLEL = 3;
+ private static final int CO_LOCATION_CNT = 2;
- public Map<String, Object> makeTestTopoConf() {
+ // class members
+ public Boolean consolidatedConfigFlag = Boolean.TRUE;
+
+ public TestConstraintSolverStrategy(boolean consolidatedConfigFlag) {
+ this.consolidatedConfigFlag = consolidatedConfigFlag;
+ LOG.info("Running tests with consolidatedConfigFlag={}", ""+consolidatedConfigFlag);
Review comment:
`""+` doesn't seem necessary.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389768350
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -593,13 +710,26 @@ public ExecutorDetails currentExec() {
return execs.get(execIndex);
}
+ /**
+ * Assign executor to worker and node.
+ * TODO: tryToSchedule is a misnomer, since it always schedules.
+ * Assignment validity check is done before the call to tryToSchedule().
+ *
+ * @param execToComp Mapping from executor to component name.
+ * @param node RasNode on which to schedule.
+ * @param workerSlot WorkerSlot on which to schedule.
+ */
public void tryToSchedule(Map<ExecutorDetails, String> execToComp, RasNode node, WorkerSlot workerSlot) {
ExecutorDetails exec = currentExec();
String comp = execToComp.get(exec);
LOG.trace("Trying assignment of {} {} to {}", exec, comp, workerSlot);
- //It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
- okToRemoveFromWorker[execIndex] = workerCompAssignment.computeIfAbsent(workerSlot, (k) -> new HashSet<>()).add(comp);
- okToRemoveFromNode[execIndex] = nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ // It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.computeIfAbsent(workerSlot, (k) -> new HashMap<>());
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) + 1); // increment assignment count
+ okToRemoveFromWorker[execIndex] = true;
Review comment:
It was a bug - as it should be always removed upon backtracking if assigned - counting is just safety check to make sure it does not go below zero.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r390722236
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,80 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
Review comment:
fix
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387780751
##########
File path: storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
##########
@@ -850,6 +852,128 @@ public void validateField(String name, Object o) {
}
}
+ public static class CustomIsExactlyOneOfValidators extends Validator {
+ private Class<?>[] subValidators;
+ private List<String> validatorClassNames;
+
+ public CustomIsExactlyOneOfValidators(Map<String, Object> params) {
+ this.subValidators = (Class<?>[]) params.get(ConfigValidationAnnotations.ValidatorParams.VALUE_VALIDATOR_CLASSES);
+ this.validatorClassNames = Arrays.asList(subValidators).stream().map(x -> x.getName()).collect(Collectors.toList());
+ }
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
+ }
+
+ HashMap<String, Exception> validatorExceptions = new HashMap<>();
+ Set<String> selectedValidators = new HashSet<>();
+ for (Class<?> vv : subValidators) {
+ Object valueValidator;
+ try {
+ valueValidator = vv.getConstructor().newInstance();
+ } catch (Exception ex) {
+ throw new IllegalArgumentException(vv.getName() + " instantiation failure", ex);
+ }
+ if (valueValidator instanceof Validator) {
+ try {
+ ((Validator) valueValidator).validateField(name + " " + vv.getSimpleName() + " value", o);
+ selectedValidators.add(vv.getName());
+ } catch (Exception ex) {
+ // only one will pass, so ignore all validation errors - stored for future use
+ validatorExceptions.put(vv.getName(), ex);
+ }
+ } else {
+ String err = String.format("validator: %s cannot be used in CustomExactlyOneOfValidators to validate values. "
+ + "Individual entry validators must a instance of Validator class", vv.getName());
+ LOG.warn(err);
+ }
+ }
+ // check if one and only one validation succeeded
+ if (selectedValidators.isEmpty()) {
+ String parseErrs = String.join(";\n\t", validatorExceptions.entrySet().stream()
+ .map(e -> String.format("%s:%s", e.getKey(), e.getValue())).collect(Collectors.toList()));
+ String err = String.format("Field %s must be one of %s; parse errors are \n\t%s", name,
+ String.join(", ", validatorClassNames), parseErrs);
+ throw new IllegalArgumentException(err);
+ }
+ if (selectedValidators.size() > 1) {
+ throw new IllegalArgumentException("Field " + name + " must match exactly one of " + String.join(", ", selectedValidators));
+ }
+ }
+ }
+
+ public static class RasConstraintsTypeValidator extends Validator {
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
+
+ public static class RasConstraint {
+ int maxNodeCoLocationCnt = -1;
+ Set<String> incompatibleComponents = new HashSet<>();
+ }
+
+ public Map<String, RasConstraint> rasConstraints = new HashMap<>(); // parsedConstraints
+
+ @Override
+ public void validateField(String name, Object o) {
+ if (o == null) {
+ throw new IllegalArgumentException("Field " + name + " must be set.");
+ }
+ if (!(o instanceof Map)) {
+ throw new IllegalArgumentException(
+ "Field " + name + " must be an Iterable containing only Map of Maps");
+ }
+ Map<String, Object> map1 = (Map<String, Object>)o;
+ for (Map.Entry<String, Object> entry1: map1.entrySet()) {
+ String comp1 = entry1.getKey();
+ Object o2 = entry1.getValue();
+ RasConstraint rasConstraint = new RasConstraint();
+ rasConstraints.put(comp1, rasConstraint);
+ if (!(o2 instanceof Map)) {
+ String err = String.format("Field %s, component %s, expecting constraints Map with keys [\"%s\", \"%s\"], in \"%s\"",
+ name, comp1, CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, o);
+ throw new IllegalArgumentException(err);
+ }
+ Map<String, Object> map2 = (Map<String, Object>)o2;
+ for (Map.Entry<String, Object> entry2: map2.entrySet()) {
+ String constraintType = entry2.getKey();
+ Object o3 = entry2.getValue();
+ switch (constraintType) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int intVal = Integer.parseInt("" + o3);
Review comment:
o3 is not a String. Casting is not good practice.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387451041
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
Review comment:
nit: space between `(String)` and `constraint`, `(List<String>)` and `constraint`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388505639
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -593,13 +710,26 @@ public ExecutorDetails currentExec() {
return execs.get(execIndex);
}
+ /**
+ * Assign executor to worker and node.
+ * TODO: tryToSchedule is a misnomer, since it always schedules.
+ * Assignment validity check is done before the call to tryToSchedule().
+ *
+ * @param execToComp Mapping from executor to component name.
+ * @param node RasNode on which to schedule.
+ * @param workerSlot WorkerSlot on which to schedule.
+ */
public void tryToSchedule(Map<ExecutorDetails, String> execToComp, RasNode node, WorkerSlot workerSlot) {
ExecutorDetails exec = currentExec();
String comp = execToComp.get(exec);
LOG.trace("Trying assignment of {} {} to {}", exec, comp, workerSlot);
- //It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
- okToRemoveFromWorker[execIndex] = workerCompAssignment.computeIfAbsent(workerSlot, (k) -> new HashSet<>()).add(comp);
- okToRemoveFromNode[execIndex] = nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ // It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.computeIfAbsent(workerSlot, (k) -> new HashMap<>());
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) + 1); // increment assignment count
+ okToRemoveFromWorker[execIndex] = true;
Review comment:
Originally, `okToRemoveFromWorker` is an indicator that whether the component should be removed from `workerCompAssignment` and `nodeCompAssignment` during backtracking since originally they are both `<Map, Set>`. I don't see a bug here. It's more like an optimization.
Since now we change `workerCompAssignment` and `nodeCompAssignment` to `<Map, Map>`, `okToRemoveFromWorker` is actually not needed any more.
Or if we leave `workerCompAssignment` as `<Map, Set>`, then we need to keep `okToRemoveFromWorker`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389956774
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -593,13 +710,26 @@ public ExecutorDetails currentExec() {
return execs.get(execIndex);
}
+ /**
+ * Assign executor to worker and node.
+ * TODO: tryToSchedule is a misnomer, since it always schedules.
+ * Assignment validity check is done before the call to tryToSchedule().
+ *
+ * @param execToComp Mapping from executor to component name.
+ * @param node RasNode on which to schedule.
+ * @param workerSlot WorkerSlot on which to schedule.
+ */
public void tryToSchedule(Map<ExecutorDetails, String> execToComp, RasNode node, WorkerSlot workerSlot) {
ExecutorDetails exec = currentExec();
String comp = execToComp.get(exec);
LOG.trace("Trying assignment of {} {} to {}", exec, comp, workerSlot);
- //It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
- okToRemoveFromWorker[execIndex] = workerCompAssignment.computeIfAbsent(workerSlot, (k) -> new HashSet<>()).add(comp);
- okToRemoveFromNode[execIndex] = nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ // It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.computeIfAbsent(workerSlot, (k) -> new HashMap<>());
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) + 1); // increment assignment count
+ okToRemoveFromWorker[execIndex] = true;
Review comment:
Current (and prior) code always checks before assignment. So let's consider that case only for simplicity - that the executor can be assigned to the selected worker and node.
If the executor is scheduled in tryToSchedule() then the okToRemoveFromNode flag should be flipped to true, so that when backtracking, the component (or component count) is properly decremented).
Earlier code would not not remove a component once added (because the flag would only be flipped to true once and any subsequent assignment backtracking would not remove the component - leading to inconsistent understanding of the current component distribution (and constraint evaluation). So the error will manifest when backtracking.
Let's say exec1 and exec2 are both assigning the same component to the same worker/node and backtracking happens for exec2 assignment. is exec2 component is not properly decremented out of node1.
The assignment (nodeCompAssignment) is a shared variable across executors. So there is no indicator which executor assigned. This flag is only retained in the okToRemove.
Hope this makes sense - unless I am confused!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387938745
##########
File path: storm-client/src/jvm/org/apache/storm/Config.java
##########
@@ -304,11 +307,13 @@
// an error will be thrown by nimbus on topology submission and not by the client prior to submitting
// the topology.
public static final String TOPOLOGY_SCHEDULER_STRATEGY = "topology.scheduler.strategy";
+
/**
- * Declare scheduling constraints for a topology used by the constraint solver strategy. A List of pairs (also a list) of components
- * that cannot coexist in the same worker.
+ * Declare scheduling constraints for a topology used by the constraint solver strategy. The format can be either
+ * old style (validated by ListOfListOfStringValidator.class or the newer style, which is a list of specific type of
+ * Maps (validated by RasConstraintsTypeValidator.class). The value must be in one or the other format.
Review comment:
Added
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387966130
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
Review comment:
Removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387448216
##########
File path: storm-client/test/jvm/org/apache/storm/TestConfigValidate.java
##########
@@ -541,6 +547,127 @@ public void testMapEntryCustomAnnotation() throws InvocationTargetException, NoS
}
}
+ @Test
+ public void testExactlyOneOfCustomAnnotation() throws InvocationTargetException, NoSuchMethodException, NoSuchFieldException,
+ InstantiationException, IllegalAccessException {
+ TestConfig config = new TestConfig();
+ Collection<Object> passCases = new LinkedList<Object>();
+ Collection<Object> failCases = new LinkedList<Object>();
+
+ List<Object> passCaseListOfList = new ArrayList<>();
+ passCaseListOfList.add(Arrays.asList("comp1", "comp2"));
+ passCaseListOfList.add(Arrays.asList("comp1", "comp3"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp4"));
+ passCaseListOfList.add(Arrays.asList("comp2", "comp5"));
+
+ Map<Object, Object> passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)10 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp2", "comp3")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp4", "comp5")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCases.add(passCaseMapOfMap);
+
+ passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp2", "comp3")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList("comp4", "comp5")},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCases.add(passCaseMapOfMap);
+
+ passCaseMapOfMap = new HashMap<>();
+ passCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, "comp2"},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, "comp4"},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ passCases.add(passCaseMapOfMap);
+
+ for (Object value : passCases) {
+ config.put(TestConfig.TEST_MAP_CONFIG_9, value);
+ ConfigValidation.validateFields(config, Arrays.asList(TestConfig.class));
+ }
+
+ List<Object> failCaseList = new ArrayList<>();
+ failCaseList.add(Arrays.asList("comp1", Arrays.asList("comp2", "comp3")));
+ failCaseList.add(Arrays.asList("comp3", Arrays.asList("comp4", "comp5")));
+ failCases.add(failCaseList);
+
+ Map<String, Object> failCaseMapOfMap = new HashMap<>();
+ failCaseMapOfMap.put("comp1",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)10 },
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, Arrays.asList(1, 2, 3)},
+ }).collect(Collectors.toMap(data -> data[0], data -> data[1]))
+ );
+ failCaseMapOfMap.put("comp2",
+ Stream.of(new Object[][] {
+ { RasConstraintsTypeValidator.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, (Integer)2 },
Review comment:
(Integer) can be removed
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388480955
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +42,182 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ // getExecutorToComponent().values() also contains system components
+ this(topo.getConf(), Sets.union(topo.getComponents().keySet(), new HashSet(topo.getExecutorToComponent().values())));
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
Review comment:
This comment is no longer valid. v is a map now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r390722185
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -54,27 +58,70 @@
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.scheduler.resource.normalization.ResourceMetrics;
+@RunWith(Parameterized.class)
public class TestConstraintSolverStrategy {
+ @Parameters
+ public static Object[] data() {
+ return new Object[] { false, true };
+ }
+
private static final Logger LOG = LoggerFactory.getLogger(TestConstraintSolverStrategy.class);
private static final int MAX_TRAVERSAL_DEPTH = 2000;
private static final int NORMAL_BOLT_PARALLEL = 11;
//Dropping the parallelism of the bolts to 3 instead of 11 so we can find a solution in a reasonable amount of work when backtracking.
private static final int BACKTRACK_BOLT_PARALLEL = 3;
+ private static final int CO_LOCATION_CNT = 2;
- public Map<String, Object> makeTestTopoConf() {
+ // class members
+ public Boolean consolidatedConfigFlag = Boolean.TRUE;
+
+ public TestConstraintSolverStrategy(boolean consolidatedConfigFlag) {
+ this.consolidatedConfigFlag = consolidatedConfigFlag;
+ LOG.info("Running tests with consolidatedConfigFlag={}", ""+consolidatedConfigFlag);
Review comment:
fix
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] bipinprasad commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
bipinprasad commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387881828
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
Review comment:
Standard style is not to have a space between cast and variable. Is this not correct?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r388516704
##########
File path: storm-server/src/test/java/org/apache/storm/scheduler/resource/strategies/scheduling/TestConstraintSolverStrategy.java
##########
@@ -85,6 +132,80 @@
return config;
}
+ /**
+ * Set Config.TOPOLOGY_RAS_CONSTRAINTS (when consolidatedConfigFlag is true) or both
+ * Config.TOPOLOGY_RAS_CONSTRAINTS/Config.TOPOLOGY_SPREAD_COMPONENTS (when consolidatedConfigFlag is false).
+ *
+ * When consolidatedConfigFlag when true, use the new more consolidated format to set Config.TOPOLOGY_RAS_CONSTRAINTS.
+ * When false, use the old configuration format for Config.TOPOLOGY_RAS_CONSTRAINTS/TOPOLOGY_SPREAD_COMPONENTS.
+ *
+ * @param constraints List of components, where the first one cannot co-exist with the others in the list
+ * @param spreads Map of component and its maxCoLocationCnt
+ * @param config Configuration to be updated
+ */
+ private void setConstraintConfig(List<List<String>> constraints, Map<String, Integer> spreads, Map<String, Object> config) {
+ if (consolidatedConfigFlag) {
+ // single configuration for each component
+ Map<String, Map<String,Object>> modifiedConstraints = new HashMap<>();
+ for (List<String> constraint: constraints) {
+ if (constraint.size() < 2) {
+ continue;
+ }
+ String comp = constraint.get(0);
+ List<String> others = constraint.subList(1, constraint.size());
+ List<Object> incompatibleComponents = (List<Object>)modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>())
+ .computeIfAbsent(ConstraintSolverStrategy.CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS, k -> new ArrayList<Object>());
+ incompatibleComponents.addAll(others);
+ }
+ for (String comp: spreads.keySet()) {
+ modifiedConstraints.computeIfAbsent(comp, k -> new HashMap<String,Object>()).put(ConstraintSolverStrategy.CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT, ""+spreads.get(comp));
Review comment:
`String,Object` can be removed; `""+` doesn't seem necessary
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r391234937
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -593,13 +710,26 @@ public ExecutorDetails currentExec() {
return execs.get(execIndex);
}
+ /**
+ * Assign executor to worker and node.
+ * TODO: tryToSchedule is a misnomer, since it always schedules.
+ * Assignment validity check is done before the call to tryToSchedule().
+ *
+ * @param execToComp Mapping from executor to component name.
+ * @param node RasNode on which to schedule.
+ * @param workerSlot WorkerSlot on which to schedule.
+ */
public void tryToSchedule(Map<ExecutorDetails, String> execToComp, RasNode node, WorkerSlot workerSlot) {
ExecutorDetails exec = currentExec();
String comp = execToComp.get(exec);
LOG.trace("Trying assignment of {} {} to {}", exec, comp, workerSlot);
- //It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
- okToRemoveFromWorker[execIndex] = workerCompAssignment.computeIfAbsent(workerSlot, (k) -> new HashSet<>()).add(comp);
- okToRemoveFromNode[execIndex] = nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ // It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.computeIfAbsent(workerSlot, (k) -> new HashMap<>());
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) + 1); // increment assignment count
+ okToRemoveFromWorker[execIndex] = true;
Review comment:
As we discussed offline, the previous code doesn't have a bug.
okToRemoveFromWorker and okToRemoveFromNode are kept here just for safety and can be removed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxNodeCoLocationCnt
and incompatibleComponents
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxNodeCoLocationCnt and incompatibleComponents
URL: https://github.com/apache/storm/pull/3215#discussion_r389921217
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -593,13 +710,26 @@ public ExecutorDetails currentExec() {
return execs.get(execIndex);
}
+ /**
+ * Assign executor to worker and node.
+ * TODO: tryToSchedule is a misnomer, since it always schedules.
+ * Assignment validity check is done before the call to tryToSchedule().
+ *
+ * @param execToComp Mapping from executor to component name.
+ * @param node RasNode on which to schedule.
+ * @param workerSlot WorkerSlot on which to schedule.
+ */
public void tryToSchedule(Map<ExecutorDetails, String> execToComp, RasNode node, WorkerSlot workerSlot) {
ExecutorDetails exec = currentExec();
String comp = execToComp.get(exec);
LOG.trace("Trying assignment of {} {} to {}", exec, comp, workerSlot);
- //It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
- okToRemoveFromWorker[execIndex] = workerCompAssignment.computeIfAbsent(workerSlot, (k) -> new HashSet<>()).add(comp);
- okToRemoveFromNode[execIndex] = nodeCompAssignment.computeIfAbsent(node, (k) -> new HashSet<>()).add(comp);
+ // It is possible that this component is already scheduled on this node or worker. If so when we backtrack we cannot remove it
+ Map<String, Integer> oneMap = workerCompAssignmentCnts.computeIfAbsent(workerSlot, (k) -> new HashMap<>());
+ oneMap.put(comp, oneMap.getOrDefault(comp, 0) + 1); // increment assignment count
+ okToRemoveFromWorker[execIndex] = true;
Review comment:
Let me explain my understanding of the original code and please let me know what I am missing.
Let's say we are trying to assign `exec1` (whose component is `comp1`) to `node1`. And let's discuss only `nodeCompAssignment` for simplicity.
First of all, we have `isExecAssignmentToWorkerValid` check before `tryToSchedule`, so there won't be any cases where `comp1` is already in on `node1` in `nodeCompAssignment` map, so `okToRemoveFromNoder[exec1]` is true. When we backtrack, we find it `true` and set it `false`. It looks `okToRemoveFromNoder` is not needed at all.
Secondly, if we don't check `isExecAssignmentToWorkerValid` before `tryToSchedule`, let's say we already have `exec0` (whose component is `comp1`) scheduled on `node1` and now we want to schedule `exec1`.
Now we will have `okToRemoveFromNoder[exec1]=false` since `comp1` is already in `nodeCompAssignment` map. And when we backtrack, since `okToRemoveFromNoder[exec1]=false`, we will not remove `comp1` from `nodeCompAssignment` map since there is a `exec0` (also a `comp1`) scheduled on this `node1`. This is also a correct behavior.
So I don't see a bug in the original code. Please elaborate the bug you observed (asking this just for my knowledge). Thanks.
---
If we change `workerCompAssignment` and `nodeCompAssignment` from `<Map, Set>` to `<Map, Map>` (as in this PR), I believe `okToRemoveFromWorker` and `okToRemoveFromNode` are not needed anymore.
Another solution is to change `okToRemoveFromNode` from `boolean[]` to `int[]` to keep track how many of the same component is on this node so far.
I can't say which solution is better. I believe `okToRemoveFromNode` was introduced for performance but I don't know how much the performance gain is.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387884472
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
Review comment:
Thanks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [storm] Ethanlm commented on a change in pull request #3215:
Storm3585 - New compact Constraint config including maxCoLocationCnt
Posted by GitBox <gi...@apache.org>.
Ethanlm commented on a change in pull request #3215: Storm3585 - New compact Constraint config including maxCoLocationCnt
URL: https://github.com/apache/storm/pull/3215#discussion_r387885841
##########
File path: storm-server/src/main/java/org/apache/storm/scheduler/resource/strategies/scheduling/ConstraintSolverStrategy.java
##########
@@ -38,71 +39,188 @@
import org.apache.storm.shade.com.google.common.annotations.VisibleForTesting;
import org.apache.storm.utils.ObjectReader;
import org.apache.storm.utils.Time;
+import org.apache.storm.validation.ConfigValidation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConstraintSolverStrategy extends BaseResourceAwareStrategy {
- //hard coded max number of states to search
private static final Logger LOG = LoggerFactory.getLogger(ConstraintSolverStrategy.class);
- //constraints and spreads
- private Map<String, Map<String, Integer>> constraintMatrix;
- private HashSet<String> spreadComps = new HashSet<>();
+ public static final String CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT = "maxNodeCoLocationCnt";
+ public static final String CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS = "incompatibleComponents";
- private Map<String, RasNode> nodes;
- private Map<ExecutorDetails, String> execToComp;
- private Map<String, Set<ExecutorDetails>> compToExecs;
- private List<String> favoredNodeIds;
- private List<String> unFavoredNodeIds;
+ /**
+ * Component constraint as derived from configuration.
+ * This is backward compatible and can parse old style Config.TOPOLOGY_RAS_CONSTRAINTS and Config.TOPOLOGY_SPREAD_COMPONENTS.
+ * New style Config.TOPOLOGY_RAS_CONSTRAINTS is map where each component has a list of other incompatible components
+ * and an optional number that specifies the maximum co-location count for the component on a node.
+ *
+ * <p>comp-1 cannot exist on same worker as comp-2 or comp-3, and at most "2" comp-1 on same node</p>
+ * <p>comp-2 and comp-4 cannot be on same node (missing comp-1 is implied from comp-1 constraint)</p>
+ *
+ * <p>
+ * { "comp-1": { "maxNodeCoLocationCnt": 2, "incompatibleComponents": ["comp-2", "comp-3" ] },
+ * "comp-2": { "incompatibleComponents": [ "comp-4" ] }
+ * }
+ * </p>
+ */
+ public static final class ConstraintConfig {
+ private Map<String, Set<String>> incompatibleComponents = new HashMap<>();
+ private Map<String, Integer> maxCoLocationCnts = new HashMap<>(); // maximum node CoLocationCnt for restricted components
- static Map<String, Map<String, Integer>> getConstraintMap(TopologyDetails topo, Set<String> comps) {
- Map<String, Map<String, Integer>> matrix = new HashMap<>();
- for (String comp : comps) {
- matrix.put(comp, new HashMap<>());
- for (String comp2 : comps) {
- matrix.get(comp).put(comp2, 0);
- }
+ ConstraintConfig(TopologyDetails topo) {
+ this(topo.getConf(), topo.getComponents().keySet());
}
- List<List<String>> constraints = (List<List<String>>) topo.getConf().get(Config.TOPOLOGY_RAS_CONSTRAINTS);
- if (constraints != null) {
- for (List<String> constraintPair : constraints) {
- String comp1 = constraintPair.get(0);
- String comp2 = constraintPair.get(1);
- if (!matrix.containsKey(comp1)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
- continue;
+
+ ConstraintConfig(Map<String, Object> conf, Set<String> comps) {
+ // spread constraints, old style may come from Config.TOPOLOGY_RAS_CONSTRAINTS where the target
+ Object rasConstraints = conf.get(Config.TOPOLOGY_RAS_CONSTRAINTS);
+ comps.forEach(k -> incompatibleComponents.computeIfAbsent(k, x -> new HashSet<>()));
+ if (rasConstraints instanceof List) {
+ // old style
+ List<List<String>> constraints = (List<List<String>>) rasConstraints;
+ for (List<String> constraintPair : constraints) {
+ String comp1 = constraintPair.get(0);
+ String comp2 = constraintPair.get(1);
+ if (!comps.contains(comp1)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp1);
+ continue;
+ }
+ if (!comps.contains(comp2)) {
+ LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
+ continue;
+ }
+ if (comp1.equals(comp2)) {
+ maxCoLocationCnts.put(comp1, 1);
+ } else {
+ incompatibleComponents.get(comp1).add(comp2);
+ incompatibleComponents.get(comp2).add(comp1);
+ }
}
- if (!matrix.containsKey(comp2)) {
- LOG.warn("Comp: {} declared in constraints is not valid!", comp2);
- continue;
+ } else {
+ Map<String, Map<String,?>> constraintMap = (Map<String, Map<String,?>>) rasConstraints;
+ constraintMap.forEach((comp1, v) -> {
+ if (comps.contains(comp1)) {
+ // v is a list of other components and an optional number which is a maxCoLocationCnt
+ v.forEach((ctype, constraint) -> {
+ switch (ctype) {
+ case CONSTRAINT_TYPE_MAX_NODE_CO_LOCATION_CNT:
+ try {
+ int numValue = Integer.parseInt("" + constraint);
+ if (numValue < 1) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, numValue, comp1);
+ } else {
+ maxCoLocationCnts.put(comp1, numValue);
+ }
+ } catch (Exception ex) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expected >= 1", ctype, constraint, comp1);
+ }
+ break;
+
+ case CONSTRAINT_TYPE_INCOMPATIBLE_COMPONENTS:
+ if (!(constraint instanceof List || constraint instanceof String)) {
+ LOG.warn("{} {} declared for Comp {} is not valid, expecting a list of components or 1 component",
+ ctype, constraint, comp1);
+ break;
+ }
+ List<String> list;
+ list = (constraint instanceof String) ? Arrays.asList((String)constraint) : (List<String>)constraint;
Review comment:
Standard style is to have a space
https://www.oracle.com/technetwork/java/codeconventions-150003.pdf (search Casts)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services