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:
[email protected]
With regards,
Apache Git Services
