You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by sh...@apache.org on 2017/10/09 15:30:55 UTC
[03/19] lucene-solr:branch_7x: Merge branch 'feature/autoscaling' of
https://git-wip-us.apache.org/repos/asf/lucene-solr
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solr-ref-guide/src/solrcloud-autoscaling-api.adoc
----------------------------------------------------------------------
diff --git a/solr/solr-ref-guide/src/solrcloud-autoscaling-api.adoc b/solr/solr-ref-guide/src/solrcloud-autoscaling-api.adoc
index 2bb3d2a..43fdfd7 100644
--- a/solr/solr-ref-guide/src/solrcloud-autoscaling-api.adoc
+++ b/solr/solr-ref-guide/src/solrcloud-autoscaling-api.adoc
@@ -324,3 +324,31 @@ The `remove-policy` command accepts a policy name to be removed from Solr. The p
----
If you attempt to remove a policy that is being used by a collection, this command will fail to delete the policy until the collection itself is deleted.
+
+== History API
+History of autoscaling events is available at `/admin/autoscaling/history`. It returs information
+about past autoscaling events and details about their processing. This history is kept in
+the `.system` collection, and populated by a trigger listener `SystemLogListener` - by default this
+listener is added to all new triggers.
+
+History events are regular Solr documents so they can be also accessed directly by
+searching on the `.system` collection. History handler acts as a regular search handler, so all
+query parameters supported by `/select` handler for that collection are supported here too.
+However, the history handler makes this
+process easier by offering a simpler syntax and knowledge of field names
+used by `SystemLogListener` for serialization of event data.
+
+History documents contain also the action context, if it was available, which gives
+further insight into eg. exact operations that were computed and/or executed.
+
+Specifically, the following query parameters can be used (they are turned into
+filter queries, so an implicit AND is applied):
+
+* `trigger` - trigger name
+* `eventType` - event type / trigger type (eg. `nodeAdded`)
+* `collection` - collection name involved in event processing
+* `stage` - event processing stage
+* `action` - trigger action
+* `node` - node name that the event refers to
+* `beforeAction` - beforeAction stage
+* `afterAction` - afterAction stage
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AddReplicaSuggester.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AddReplicaSuggester.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AddReplicaSuggester.java
index 01149f3..f96c66d 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AddReplicaSuggester.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AddReplicaSuggester.java
@@ -17,11 +17,15 @@
package org.apache.solr.client.solrj.cloud.autoscaling;
+import java.util.Collections;
import java.util.List;
+import java.util.Set;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy.Suggester;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.util.Pair;
class AddReplicaSuggester extends Suggester {
@@ -32,34 +36,37 @@ class AddReplicaSuggester extends Suggester {
}
SolrRequest tryEachNode(boolean strict) {
- String coll = (String) hints.get(Hint.COLL);
- String shard = (String) hints.get(Hint.SHARD);
- if (coll == null || shard == null)
+ Set<Pair<String, String>> shards = (Set<Pair<String, String>>) hints.getOrDefault(Hint.COLL_SHARD, Collections.emptySet());
+ if (shards.isEmpty()) {
throw new RuntimeException("add-replica requires 'collection' and 'shard'");
- //iterate through elements and identify the least loaded
-
- List<Clause.Violation> leastSeriousViolation = null;
- Integer targetNodeIndex = null;
- for (int i = getMatrix().size() - 1; i >= 0; i--) {
- Row row = getMatrix().get(i);
- if (!isAllowed(row.node, Hint.TARGET_NODE)) continue;
- Row tmpRow = row.addReplica(coll, shard);
- tmpRow.violations.clear();
+ }
+ for (Pair<String,String> shard : shards) {
+ Replica.Type type = Replica.Type.get((String) hints.get(Hint.REPLICATYPE));
+ //iterate through elements and identify the least loaded
+ List<Clause.Violation> leastSeriousViolation = null;
+ Integer targetNodeIndex = null;
+ for (int i = getMatrix().size() - 1; i >= 0; i--) {
+ Row row = getMatrix().get(i);
+ if (!row.isLive) continue;
+ if (!isAllowed(row.node, Hint.TARGET_NODE)) continue;
+ Row tmpRow = row.addReplica(shard.first(), shard.second(), type);
- List<Clause.Violation> errs = testChangedMatrix(strict, getModifiedMatrix(getMatrix(), tmpRow, i));
- if(!containsNewErrors(errs)) {
- if(isLessSerious(errs, leastSeriousViolation)){
- leastSeriousViolation = errs;
- targetNodeIndex = i;
+ List<Clause.Violation> errs = testChangedMatrix(strict, getModifiedMatrix(getMatrix(), tmpRow, i));
+ if (!containsNewErrors(errs)) {
+ if (isLessSerious(errs, leastSeriousViolation)) {
+ leastSeriousViolation = errs;
+ targetNodeIndex = i;
+ }
}
}
- }
- if (targetNodeIndex != null) {// there are no rule violations
- getMatrix().set(targetNodeIndex, getMatrix().get(targetNodeIndex).addReplica(coll, shard));
- return CollectionAdminRequest
- .addReplicaToShard(coll, shard)
- .setNode(getMatrix().get(targetNodeIndex).node);
+ if (targetNodeIndex != null) {// there are no rule violations
+ getMatrix().set(targetNodeIndex, getMatrix().get(targetNodeIndex).addReplica(shard.first(), shard.second(), type));
+ return CollectionAdminRequest
+ .addReplicaToShard(shard.first(), shard.second())
+ .setType(type)
+ .setNode(getMatrix().get(targetNodeIndex).node);
+ }
}
return null;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AutoScalingConfig.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AutoScalingConfig.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AutoScalingConfig.java
new file mode 100644
index 0000000..51e8f42
--- /dev/null
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AutoScalingConfig.java
@@ -0,0 +1,531 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.solr.client.solrj.cloud.autoscaling;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.params.AutoScalingParams;
+import org.apache.solr.common.util.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static java.util.stream.Collectors.collectingAndThen;
+import static java.util.stream.Collectors.toList;
+
+/**
+ * Bean representation of <code>autoscaling.json</code>, which parses data
+ * lazily.
+ */
+public class AutoScalingConfig implements MapWriter {
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private final Map<String, Object> jsonMap;
+ private final boolean empty;
+
+ private Policy policy;
+ private Map<String, TriggerConfig> triggers;
+ private Map<String, TriggerListenerConfig> listeners;
+
+ private final int zkVersion;
+
+ /**
+ * Bean representation of trigger listener config.
+ */
+ public static class TriggerListenerConfig implements MapWriter {
+ public final String name;
+ public final String trigger;
+ public final EnumSet<TriggerEventProcessorStage> stages = EnumSet.noneOf(TriggerEventProcessorStage.class);
+ public final String listenerClass;
+ public final Set<String> beforeActions;
+ public final Set<String> afterActions;
+ public final Map<String, Object> properties;
+
+ public TriggerListenerConfig(String name, Map<String, Object> properties) {
+ this.name = name;
+ if (properties == null) {
+ this.properties = Collections.emptyMap();
+ } else {
+ this.properties = Collections.unmodifiableMap(new HashMap<>(properties));
+ }
+ trigger = (String)this.properties.get(AutoScalingParams.TRIGGER);
+ List<Object> stageNames = getList(AutoScalingParams.STAGE, this.properties);
+ for (Object stageName : stageNames) {
+ try {
+ TriggerEventProcessorStage stage = TriggerEventProcessorStage.valueOf(String.valueOf(stageName).toUpperCase(Locale.ROOT));
+ stages.add(stage);
+ } catch (Exception e) {
+ LOG.warn("Invalid stage name '" + name + "' in listener config, skipping: " + properties);
+ }
+ }
+ listenerClass = (String)this.properties.get(AutoScalingParams.CLASS);
+ Set<String> bActions = new HashSet<>();
+ getList(AutoScalingParams.BEFORE_ACTION, this.properties).forEach(o -> bActions.add(String.valueOf(o)));
+ beforeActions = Collections.unmodifiableSet(bActions);
+ Set<String> aActions = new HashSet<>();
+ getList(AutoScalingParams.AFTER_ACTION, this.properties).forEach(o -> aActions.add(String.valueOf(o)));
+ afterActions = Collections.unmodifiableSet(aActions);
+ }
+
+ @Override
+ public void writeMap(EntryWriter ew) throws IOException {
+ // don't write duplicate entries - skip explicit fields if their values
+ // are already contained in properties
+// if (!properties.containsKey(AutoScalingParams.NAME)) {
+// ew.put(AutoScalingParams.NAME, name);
+// }
+ if (!properties.containsKey(AutoScalingParams.CLASS)) {
+ ew.put(AutoScalingParams.CLASS, listenerClass);
+ }
+ if (!properties.containsKey(AutoScalingParams.TRIGGER)) {
+ ew.put(AutoScalingParams.TRIGGER, trigger);
+ }
+ if (!properties.containsKey(AutoScalingParams.STAGE)) {
+ ew.put(AutoScalingParams.STAGE, stages);
+ }
+ if (!properties.containsKey(AutoScalingParams.BEFORE_ACTION)) {
+ ew.put(AutoScalingParams.BEFORE_ACTION, beforeActions);
+ }
+ if (!properties.containsKey(AutoScalingParams.AFTER_ACTION)) {
+ ew.put(AutoScalingParams.AFTER_ACTION, afterActions);
+ }
+ // forEach doesn't allow throwing exceptions...
+ for (Map.Entry<String, Object> entry : properties.entrySet()) {
+ ew.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ TriggerListenerConfig that = (TriggerListenerConfig) o;
+
+ if (name != null ? !name.equals(that.name) : that.name != null) return false;
+ if (trigger != null ? !trigger.equals(that.trigger) : that.trigger != null) return false;
+ if (!stages.equals(that.stages)) return false;
+ if (listenerClass != null ? !listenerClass.equals(that.listenerClass) : that.listenerClass != null) return false;
+ if (!beforeActions.equals(that.beforeActions)) return false;
+ if (!afterActions.equals(that.afterActions)) return false;
+ return properties.equals(that.properties);
+ }
+ }
+
+ /**
+ * Bean representation of trigger config.
+ */
+ public static class TriggerConfig implements MapWriter {
+ /** Trigger name. */
+ public final String name;
+ /** Trigger event type. */
+ public final TriggerEventType event;
+ /** Enabled flag. */
+ public final boolean enabled;
+ /** List of configured actions, never null. */
+ public final List<ActionConfig> actions;
+ /** Map of additional trigger properties, never null. */
+ public final Map<String, Object> properties;
+
+ public TriggerConfig(String name, Map<String, Object> properties) {
+ this.name = name;
+ if (properties != null) {
+ this.properties = Collections.unmodifiableMap(new HashMap<>(properties));
+ } else {
+ this.properties = Collections.emptyMap();
+ }
+ String event = (String) this.properties.get(AutoScalingParams.EVENT);
+ if (event != null) {
+ TriggerEventType type = null;
+ try {
+ type = TriggerEventType.valueOf(event.toUpperCase(Locale.ROOT));
+ } catch (Exception e) {
+ }
+ if (type == null) {
+ this.event = TriggerEventType.INVALID;
+ } else {
+ this.event = type;
+ }
+ } else {
+ this.event = TriggerEventType.INVALID;
+ }
+ enabled = Boolean.parseBoolean(String.valueOf(this.properties.getOrDefault("enabled", "true")));
+
+ List<Map<String, Object>> newActions = (List<Map<String, Object>>)this.properties.get("actions");
+ if (newActions != null) {
+ this.actions = newActions.stream().map(ActionConfig::new).collect(collectingAndThen(toList(), Collections::unmodifiableList));
+ } else {
+ this.actions = Collections.emptyList();
+ }
+ }
+
+ /**
+ * Create a copy of this config with specified enabled flag.
+ * @param enabled true when enabled, false otherwise.
+ * @return modified copy of the configuration
+ */
+ public TriggerConfig withEnabled(boolean enabled) {
+ Map<String, Object> props = new HashMap<>(properties);
+ props.put(AutoScalingParams.ENABLED, String.valueOf(enabled));
+ return new TriggerConfig(name, props);
+ }
+
+ /**
+ * Create a copy of this config with specified property.
+ * @param key property name
+ * @param value property value
+ * @return modified copy of the configuration
+ */
+ public TriggerConfig withProperty(String key, Object value) {
+ Map<String, Object> props = new HashMap<>(properties);
+ props.put(key, String.valueOf(value));
+ return new TriggerConfig(name, props);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ TriggerConfig that = (TriggerConfig) o;
+
+ if (name != null ? !name.equals(that.name) : that.name != null) return false;
+ if (event != that.event) return false;
+ return properties.equals(that.properties);
+ }
+
+ @Override
+ public void writeMap(EntryWriter ew) throws IOException {
+// if (!properties.containsKey(AutoScalingParams.NAME)) {
+// ew.put(AutoScalingParams.NAME, name);
+// }
+ if (!properties.containsKey(AutoScalingParams.EVENT)) {
+ ew.put(AutoScalingParams.EVENT, event.toString());
+ }
+ // forEach doesn't allow throwing exceptions...
+ for (Map.Entry<String, Object> entry : properties.entrySet()) {
+ ew.put(entry.getKey(), entry.getValue());
+ }
+ }
+ }
+
+ /**
+ * Bean representation of trigger action configuration.
+ */
+ public static class ActionConfig implements MapWriter {
+ /** Action name. */
+ public final String name;
+ /** Class name of action implementtion. */
+ public final String actionClass;
+ /** Additional action properties. */
+ public final Map<String, Object> properties;
+
+ /**
+ * Construct from a JSON map.
+ * @param properties JSON map with properties - selected properties will be
+ * used for setting the values of <code>name</code> and
+ * <code>actionClass</code>.
+ */
+ public ActionConfig(Map<String, Object> properties) {
+ if (properties != null) {
+ this.properties = Collections.unmodifiableMap(new HashMap<>(properties));
+ } else {
+ this.properties = Collections.emptyMap();
+ }
+ this.name = (String)this.properties.get(AutoScalingParams.NAME);
+ this.actionClass = (String)this.properties.get(AutoScalingParams.CLASS);
+ }
+
+ @Override
+ public void writeMap(EntryWriter ew) throws IOException {
+ // forEach doesn't allow throwing exceptions...
+ for (Map.Entry<String, Object> entry : properties.entrySet()) {
+ ew.put(entry.getKey(), entry.getValue());
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ ActionConfig that = (ActionConfig) o;
+
+ return properties.equals(that.properties);
+ }
+ }
+
+ /**
+ * Construct from bytes that represent a UTF-8 JSON string.
+ * @param utf8 config data
+ */
+ public AutoScalingConfig(byte[] utf8) {
+ this(utf8 != null && utf8.length > 0 ? (Map<String, Object>)Utils.fromJSON(utf8) : Collections.emptyMap());
+ }
+
+ /**
+ * Construct from a JSON map representation.
+ * @param jsonMap JSON map representation of the config. Note that this map is evaluated lazily, and
+ * outside modifications may cause unpredictable behavior.
+ */
+ public AutoScalingConfig(Map<String, Object> jsonMap) {
+ this.jsonMap = jsonMap;
+ int version = -1;
+ if (jsonMap.containsKey(AutoScalingParams.ZK_VERSION)) {
+ try {
+ version = (Integer)jsonMap.get(AutoScalingParams.ZK_VERSION);
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+ zkVersion = version;
+ jsonMap.remove(AutoScalingParams.ZK_VERSION);
+ empty = jsonMap.isEmpty();
+ }
+
+ private AutoScalingConfig(Policy policy, Map<String, TriggerConfig> triggerConfigs, Map<String,
+ TriggerListenerConfig> listenerConfigs, int zkVersion) {
+ this.policy = policy;
+ this.triggers = triggerConfigs != null ? Collections.unmodifiableMap(triggerConfigs) : null;
+ this.listeners = listenerConfigs != null ? Collections.unmodifiableMap(listenerConfigs) : null;
+ this.jsonMap = null;
+ this.zkVersion = zkVersion;
+ this.empty = policy == null &&
+ (triggerConfigs == null || triggerConfigs.isEmpty()) &&
+ (listenerConfigs == null || listenerConfigs.isEmpty());
+ }
+
+ /**
+ * Return true if the source <code>autoscaling.json</code> was empty, false otherwise.
+ */
+ public boolean isEmpty() {
+ return empty;
+ }
+
+ /**
+ * Get {@link Policy} configuration.
+ */
+ public Policy getPolicy() {
+ if (policy == null) {
+ if (jsonMap != null) {
+ policy = new Policy(jsonMap);
+ } else {
+ policy = new Policy();
+ }
+ }
+ return policy;
+ }
+
+ /**
+ * Get trigger configurations.
+ */
+ public Map<String, TriggerConfig> getTriggerConfigs() {
+ if (triggers == null) {
+ if (jsonMap != null) {
+ Map<String, Object> trigMap = (Map<String, Object>)jsonMap.get("triggers");
+ if (trigMap == null) {
+ triggers = Collections.emptyMap();
+ } else {
+ HashMap<String, TriggerConfig> newTriggers = new HashMap<>(trigMap.size());
+ for (Map.Entry<String, Object> entry : trigMap.entrySet()) {
+ newTriggers.put(entry.getKey(), new TriggerConfig(entry.getKey(), (Map<String, Object>)entry.getValue()));
+ }
+ triggers = Collections.unmodifiableMap(newTriggers);
+ }
+ } else {
+ triggers = Collections.emptyMap();
+ }
+ }
+ return triggers;
+ }
+
+ /**
+ * Check whether triggers for specific event type exist.
+ * @param types list of event types
+ * @return true if there's at least one trigger matching at least one event type,
+ * false otherwise,
+ */
+ public boolean hasTriggerForEvents(TriggerEventType... types) {
+ if (types == null || types.length == 0) {
+ return false;
+ }
+ for (TriggerConfig config : getTriggerConfigs().values()) {
+ for (TriggerEventType type : types) {
+ if (config.event.equals(type)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Get listener configurations.
+ */
+ public Map<String, TriggerListenerConfig> getTriggerListenerConfigs() {
+ if (listeners == null) {
+ if (jsonMap != null) {
+ Map<String, Object> map = (Map<String, Object>)jsonMap.get("listeners");
+ if (map == null) {
+ listeners = Collections.emptyMap();
+ } else {
+ HashMap<String, TriggerListenerConfig> newListeners = new HashMap<>(map.size());
+ for (Map.Entry<String, Object> entry : map.entrySet()) {
+ newListeners.put(entry.getKey(), new TriggerListenerConfig(entry.getKey(), (Map<String, Object>)entry.getValue()));
+ }
+ this.listeners = Collections.unmodifiableMap(newListeners);
+ }
+ } else {
+ listeners = Collections.emptyMap();
+ }
+ }
+ return listeners;
+ }
+
+ /**
+ * Create a copy of the config with replaced policy.
+ * @param policy new policy
+ * @return modified copy of the configuration
+ */
+ public AutoScalingConfig withPolicy(Policy policy) {
+ return new AutoScalingConfig(policy, getTriggerConfigs(), getTriggerListenerConfigs(), zkVersion);
+ }
+
+ /**
+ * Create a copy of the config with replaced trigger configurations.
+ * @param configs new trigger configurations
+ * @return modified copy of the configuration
+ */
+ public AutoScalingConfig withTriggerConfigs(Map<String, TriggerConfig> configs) {
+ return new AutoScalingConfig(getPolicy(), configs, getTriggerListenerConfigs(), zkVersion);
+ }
+
+ /**
+ * Create a copy of the config with replaced trigger configuration
+ * @param config new trigger configuration
+ * @return modified copy of the configuration
+ */
+ public AutoScalingConfig withTriggerConfig(TriggerConfig config) {
+ Map<String, TriggerConfig> configs = new HashMap<>(getTriggerConfigs());
+ configs.put(config.name, config);
+ return withTriggerConfigs(configs);
+ }
+
+ /**
+ * Create a copy of the config without a trigger configuration.
+ * @param name trigger configuration name
+ * @return modified copy of the configuration, even if the specified config name didn't exist.
+ */
+ public AutoScalingConfig withoutTriggerConfig(String name) {
+ Map<String, TriggerConfig> configs = new HashMap<>(getTriggerConfigs());
+ configs.remove(name);
+ return withTriggerConfigs(configs);
+ }
+
+ /**
+ * Create a copy of the config with replaced trigger listener configurations.
+ * @param configs new trigger listener configurations
+ * @return modified copy of the configuration
+ */
+ public AutoScalingConfig withTriggerListenerConfigs(Map<String, TriggerListenerConfig> configs) {
+ return new AutoScalingConfig(getPolicy(), getTriggerConfigs(), configs, zkVersion);
+ }
+
+ /**
+ * Create a copy of the config with replaced trigger listener configuration.
+ * @param config new trigger listener configuration
+ * @return modified copy of the configuration
+ */
+ public AutoScalingConfig withTriggerListenerConfig(TriggerListenerConfig config) {
+ Map<String, TriggerListenerConfig> configs = new HashMap<>(getTriggerListenerConfigs());
+ configs.put(config.name, config);
+ return withTriggerListenerConfigs(configs);
+ }
+
+ /**
+ * Create a copy of the config without a trigger listener configuration.
+ * @param name trigger listener configuration name
+ * @return modified copy of the configuration, even if the specified config name didn't exist.
+ */
+ public AutoScalingConfig withoutTriggerListenerConfig(String name) {
+ Map<String, TriggerListenerConfig> configs = new HashMap<>(getTriggerListenerConfigs());
+ configs.remove(name);
+ return withTriggerListenerConfigs(configs);
+ }
+
+ /**
+ * Return the znode version that was used to create this configuration.
+ */
+ public int getZkVersion() {
+ return zkVersion;
+ }
+
+ @Override
+ public void writeMap(EntryWriter ew) throws IOException {
+ Policy policy = getPolicy();
+ // properties of Policy are expected at top level
+ policy.writeMap(ew);
+
+ ew.put("triggers", getTriggerConfigs());
+ ew.put("listeners", getTriggerListenerConfigs());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ AutoScalingConfig that = (AutoScalingConfig) o;
+
+ if (!getPolicy().equals(that.getPolicy())) return false;
+ if (!triggers.equals(that.triggers)) return false;
+ return listeners.equals(that.listeners);
+ }
+
+ private static List<Object> getList(String key, Map<String, Object> properties) {
+ return getList(key, properties, null);
+ }
+
+ private static List<Object> getList(String key, Map<String, Object> properties, List<Object> defaultList) {
+ if (defaultList == null) {
+ defaultList = Collections.emptyList();
+ }
+ Object o = properties.get(key);
+ if (o == null) {
+ return defaultList;
+ }
+ if (o instanceof List) {
+ return (List)o;
+ } else if (o instanceof Collection) {
+ return new ArrayList<>((Collection) o);
+ } else {
+ return Collections.singletonList(String.valueOf(o));
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Clause.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Clause.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Clause.java
index ff56627..ec6f73c 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Clause.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Clause.java
@@ -20,18 +20,20 @@ package org.apache.solr.client.solrj.cloud.autoscaling;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.solr.client.solrj.cloud.autoscaling.Policy.ReplicaInfo;
import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.rule.ImplicitSnitch;
import org.apache.solr.common.util.StrUtils;
import org.apache.solr.common.util.Utils;
@@ -50,13 +52,16 @@ import static org.apache.solr.common.params.CoreAdminParams.SHARD;
// a set of conditions in a policy
public class Clause implements MapWriter, Comparable<Clause> {
- public Map<String, Object> original;
- public Condition collection, shard, replica, tag, globalTag;
+ final Map<String, Object> original;
+ Condition collection, shard, replica, tag, globalTag;
+ final Replica.Type type;
boolean strict = true;
public Clause(Map<String, Object> m) {
- this.original = m;
+ this.original = Utils.getDeepCopy(m, 10);
+ String type = (String) m.get("type");
+ this.type = type == null || ANY.equals(type) ? null : Replica.Type.valueOf(type.toUpperCase(Locale.ROOT));
strict = Boolean.parseBoolean(String.valueOf(m.getOrDefault("strict", "true")));
Optional<String> globalTagName = m.keySet().stream().filter(Policy.GLOBAL_ONLY_TAGS::contains).findFirst();
if (globalTagName.isPresent()) {
@@ -70,7 +75,7 @@ public class Clause implements MapWriter, Comparable<Clause> {
} else {
collection = parse(COLLECTION, m);
shard = parse(SHARD, m);
- if(m.get(REPLICA) == null){
+ if (m.get(REPLICA) == null) {
throw new RuntimeException(StrUtils.formatString("'replica' is required in {0}", Utils.toJSONString(m)));
}
this.replica = parse(REPLICA, m);
@@ -79,6 +84,12 @@ public class Clause implements MapWriter, Comparable<Clause> {
}
if (tag == null)
throw new RuntimeException("Invalid op, must have one and only one tag other than collection, shard,replica " + Utils.toJSONString(m));
+ if (tag.name.startsWith(Clause.METRICS_PREFIX)) {
+ List<String> ss = StrUtils.splitSmart(tag.name, ':');
+ if (ss.size() < 3 || ss.size() > 4) {
+ throw new RuntimeException("Invalid metrics: param in " + Utils.toJSONString(m) + " must have at 2 or 3 segments after 'metrics:' separated by ':'");
+ }
+ }
}
@@ -100,32 +111,44 @@ public class Clause implements MapWriter, Comparable<Clause> {
tag = parse(s, singletonMap(s, o));
}
+ private int compareTypes(Replica.Type t1, Replica.Type t2) {
+ if (t1 == null && t2 == null) return 0;
+ if (t1 != null && t2 == null) return -1;
+ if (t1 == null) return 1;
+ return 0;
+ }
+
@Override
public int compareTo(Clause that) {
- try {
- int v = Integer.compare(this.tag.op.priority, that.tag.op.priority);
- if (v != 0) return v;
- if (this.isPerCollectiontag() && that.isPerCollectiontag()) {
- v = Integer.compare(this.replica.op.priority, that.replica.op.priority);
- if (v == 0) {
- v = Long.compare((Long) this.replica.val, (Long) that.replica.val);
- v = this.replica.op == LESS_THAN ? v : v * -1;
- }
- return v;
- } else {
- return 0;
+ int v = Integer.compare(this.tag.op.priority, that.tag.op.priority);
+ if (v != 0) return v;
+ if (this.isPerCollectiontag() && that.isPerCollectiontag()) {
+ v = Integer.compare(this.replica.op.priority, that.replica.op.priority);
+ if (v == 0) {// higher the number of replicas , harder to satisfy
+ v = Long.compare((Long) this.replica.val, (Long) that.replica.val);
+ v = this.replica.op == LESS_THAN ? v : v * -1;
}
- } catch (NullPointerException e) {
- throw e;
+ if (v == 0) v = compareTypes(this.type, that.type);
+ return v;
+ } else {
+ return 0;
}
}
- void addTags(List<String> params) {
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ Clause that = (Clause)o;
+ return compareTo(that) == 0;
+ }
+
+ void addTags(Collection<String> params) {
if (globalTag != null && !params.contains(globalTag.name)) params.add(globalTag.name);
if (tag != null && !params.contains(tag.name)) params.add(tag.name);
}
- public static class Condition {
+ class Condition {
final String name;
final Object val;
final Operand op;
@@ -136,19 +159,18 @@ public class Clause implements MapWriter, Comparable<Clause> {
this.op = op;
}
- TestStatus match(Row row) {
- return op.match(val, row.getVal(name));
- }
-
- TestStatus match(Object testVal) {
- return op.match(this.val, testVal);
+ boolean isPass(Object inputVal) {
+ if (inputVal instanceof ReplicaCount) inputVal = ((ReplicaCount) inputVal).getVal(type);
+ ValidateInfo validator = getValidator(name);
+ if (validator == LazyValidator.INST) { // we don't know the type
+ return op.match(parseString(val), parseString(inputVal)) == PASS;
+ } else {
+ return op.match(val, validate(name, inputVal, false)) == PASS;
+ }
}
- public boolean isPass(Object inputVal) {
- return op.match(val, validate(name, inputVal, false)) == PASS;
- }
- public boolean isPass(Row row) {
+ boolean isPass(Row row) {
return op.match(val, row.getVal(name)) == PASS;
}
@@ -161,8 +183,14 @@ public class Clause implements MapWriter, Comparable<Clause> {
return false;
}
- public Integer delta(Object val) {
- return op.delta(this.val, val);
+ public Long delta(Object val) {
+ if (this.val instanceof String) {
+ if (op == LESS_THAN || op == GREATER_THAN) {
+ return op.delta(Clause.parseDouble(name, this.val), Clause.parseDouble(name, val));
+ } else {
+ return 0l;
+ }
+ } else return op.delta(this.val, val);
}
public String getName() {
@@ -172,13 +200,13 @@ public class Clause implements MapWriter, Comparable<Clause> {
public Object getValue() {
return val;
}
-
+
public Operand getOperand() {
return op;
}
}
- static Condition parse(String s, Map m) {
+ Condition parse(String s, Map m) {
Object expectedVal = null;
Object val = m.get(s);
try {
@@ -209,12 +237,12 @@ public class Clause implements MapWriter, Comparable<Clause> {
public class Violation implements MapWriter {
final String shard, coll, node;
final Object actualVal;
- final Integer delta;//how far is the actual value from the expected value
+ final Long delta;//how far is the actual value from the expected value
final Object tagKey;
private final int hash;
- private Violation(String coll, String shard, String node, Object actualVal, Integer delta, Object tagKey) {
+ private Violation(String coll, String shard, String node, Object actualVal, Long delta, Object tagKey) {
this.shard = shard;
this.coll = coll;
this.node = node;
@@ -252,14 +280,19 @@ public class Clause implements MapWriter, Comparable<Clause> {
}
@Override
+ public String toString() {
+ return Utils.toJSONString(Utils.getDeepCopy(toMap(new LinkedHashMap<>()), 5));
+ }
+
+ @Override
public void writeMap(EntryWriter ew) throws IOException {
ew.putIfNotNull("collection", coll);
ew.putIfNotNull("shard", shard);
ew.putIfNotNull("node", node);
ew.putIfNotNull("tagKey", String.valueOf(tagKey));
ew.putIfNotNull("violation", (MapWriter) ew1 -> {
- ew1.put(getClause().isPerCollectiontag() ? "replica" : tag.name,
- String.valueOf(actualVal));
+ if (getClause().isPerCollectiontag()) ew1.put("replica", actualVal);
+ else ew1.put(tag.name, String.valueOf(actualVal));
ew1.putIfNotNull("delta", delta);
});
ew.put("clause", getClause());
@@ -270,12 +303,12 @@ public class Clause implements MapWriter, Comparable<Clause> {
public List<Violation> test(List<Row> allRows) {
List<Violation> violations = new ArrayList<>();
if (isPerCollectiontag()) {
- Map<String, Map<String, Map<String, AtomicInteger>>> replicaCount = computeReplicaCounts(allRows);
- for (Map.Entry<String, Map<String, Map<String, AtomicInteger>>> e : replicaCount.entrySet()) {
+ Map<String, Map<String, Map<String, ReplicaCount>>> replicaCount = computeReplicaCounts(allRows);
+ for (Map.Entry<String, Map<String, Map<String, ReplicaCount>>> e : replicaCount.entrySet()) {
if (!collection.isPass(e.getKey())) continue;
- for (Map.Entry<String, Map<String, AtomicInteger>> shardVsCount : e.getValue().entrySet()) {
+ for (Map.Entry<String, Map<String, ReplicaCount>> shardVsCount : e.getValue().entrySet()) {
if (!shard.isPass(shardVsCount.getKey())) continue;
- for (Map.Entry<String, AtomicInteger> counts : shardVsCount.getValue().entrySet()) {
+ for (Map.Entry<String, ReplicaCount> counts : shardVsCount.getValue().entrySet()) {
if (!replica.isPass(counts.getValue())) {
violations.add(new Violation(
e.getKey(),
@@ -301,24 +334,23 @@ public class Clause implements MapWriter, Comparable<Clause> {
}
- private Map<String, Map<String, Map<String, AtomicInteger>>> computeReplicaCounts(List<Row> allRows) {
- Map<String, Map<String, Map<String, AtomicInteger>>> collVsShardVsTagVsCount = new HashMap<>();
- for (Row row : allRows)
+ private Map<String, Map<String, Map<String, ReplicaCount>>> computeReplicaCounts(List<Row> allRows) {
+ Map<String, Map<String, Map<String, ReplicaCount>>> collVsShardVsTagVsCount = new HashMap<>();
+ for (Row row : allRows) {
for (Map.Entry<String, Map<String, List<ReplicaInfo>>> colls : row.collectionVsShardVsReplicas.entrySet()) {
String collectionName = colls.getKey();
if (!collection.isPass(collectionName)) continue;
- collVsShardVsTagVsCount.putIfAbsent(collectionName, new HashMap<>());
- Map<String, Map<String, AtomicInteger>> collMap = collVsShardVsTagVsCount.get(collectionName);
+ Map<String, Map<String, ReplicaCount>> collMap = collVsShardVsTagVsCount.computeIfAbsent(collectionName, s -> new HashMap<>());
for (Map.Entry<String, List<ReplicaInfo>> shards : colls.getValue().entrySet()) {
String shardName = shards.getKey();
if (ANY.equals(shard.val)) shardName = ANY;
if (!shard.isPass(shardName)) break;
- collMap.putIfAbsent(shardName, new HashMap<>());
- Map<String, AtomicInteger> tagVsCount = collMap.get(shardName);
+ Map<String, ReplicaCount> tagVsCount = collMap.computeIfAbsent(shardName, s -> new HashMap<>());
Object tagVal = row.getVal(tag.name);
- tagVsCount.putIfAbsent(tag.isPass(tagVal) ? String.valueOf(tagVal) : "", new AtomicInteger());
+ tagVsCount.computeIfAbsent(tag.isPass(tagVal) ? String.valueOf(tagVal) : "", s -> new ReplicaCount());
if (tag.isPass(tagVal)) {
- tagVsCount.get(String.valueOf(tagVal)).addAndGet(shards.getValue().size());
+ tagVsCount.get(String.valueOf(tagVal)).increment(shards.getValue());
+ }
}
}
}
@@ -343,7 +375,7 @@ public class Clause implements MapWriter, Comparable<Clause> {
NOT_APPLICABLE, FAIL, PASS
}
- private static final Set<String> IGNORE_TAGS = new HashSet<>(Arrays.asList(REPLICA, COLLECTION, SHARD, "strict"));
+ private static final Set<String> IGNORE_TAGS = new HashSet<>(Arrays.asList(REPLICA, COLLECTION, SHARD, "strict", "type"));
static class ValidateInfo {
final Class type;
@@ -356,54 +388,82 @@ public class Clause implements MapWriter, Comparable<Clause> {
this.type = type;
this.vals = vals;
this.min = min;
- if(min != null && !type.isInstance(min)) throw new RuntimeException("wrong min value type, expected: " + type.getName() + " actual: " + min.getClass().getName());
+ if (min != null && !type.isInstance(min))
+ throw new RuntimeException("wrong min value type, expected: " + type.getName() + " actual: " + min.getClass().getName());
this.max = max;
- if(max != null && !type.isInstance(max)) throw new RuntimeException("wrong max value type, expected: " + type.getName() + " actual: " + max.getClass().getName());
+ if (max != null && !type.isInstance(max))
+ throw new RuntimeException("wrong max value type, expected: " + type.getName() + " actual: " + max.getClass().getName());
+ }
+
+ public Object validate(String name, Object val, boolean isRuleVal) {
+ if (type == Double.class) {
+ Double num = parseDouble(name, val);
+ if (isRuleVal) {
+ if (min != null)
+ if (Double.compare(num, (Double) min) == -1)
+ throw new RuntimeException(name + ": " + val + " must be greater than " + min);
+ if (max != null)
+ if (Double.compare(num, (Double) max) == 1)
+ throw new RuntimeException(name + ": " + val + " must be less than " + max);
+ }
+ return num;
+ } else if (type == Long.class) {
+ Long num = parseLong(name, val);
+ if (isRuleVal) {
+ if (min != null)
+ if (num < min.longValue())
+ throw new RuntimeException(name + ": " + val + " must be greater than " + min);
+ if (max != null)
+ if (num > max.longValue())
+ throw new RuntimeException(name + ": " + val + " must be less than " + max);
+ }
+ return num;
+ } else if (type == String.class) {
+ if (isRuleVal && vals != null && !vals.contains(val))
+ throw new RuntimeException(name + ": " + val + " must be one of " + StrUtils.join(vals, ','));
+ return val;
+ } else {
+ throw new RuntimeException("Invalid type ");
+ }
+
}
}
+ static class LazyValidator extends ValidateInfo {
+ static final LazyValidator INST = new LazyValidator();
+
+ LazyValidator() {
+ super(null, null, null, null);
+ }
+
+ @Override
+ public Object validate(String name, Object val, boolean isRuleVal) {
+ return parseString(val);
+ }
+ }
+
+ public static String parseString(Object val) {
+ return val == null ? null : String.valueOf(val);
+ }
/**
- *
- * @param name name of the condition
- * @param val value of the condition
+ * @param name name of the condition
+ * @param val value of the condition
* @param isRuleVal is this provided in the rule
* @return actual validated value
*/
- public static Object validate(String name, Object val, boolean isRuleVal) {
+ public static Object validate(String name, Object val, boolean isRuleVal) {
if (val == null) return null;
+ ValidateInfo info = getValidator(name);
+ if (info == null) throw new RuntimeException("Unknown type :" + name);
+ return info.validate(name, val, isRuleVal);
+ }
+
+ private static ValidateInfo getValidator(String name) {
ValidateInfo info = validatetypes.get(name);
if (info == null && name.startsWith(ImplicitSnitch.SYSPROP)) info = validatetypes.get("STRING");
- if (info == null) throw new RuntimeException("Unknown type :" + name);
- if (info.type == Double.class) {
- Double num = parseDouble(name, val);
- if (isRuleVal) {
- if (info.min != null)
- if (Double.compare(num, (Double) info.min) == -1)
- throw new RuntimeException(name + ": " + val + " must be greater than " + info.min);
- if (info.max != null)
- if (Double.compare(num, (Double) info.max) == 1)
- throw new RuntimeException(name + ": " + val + " must be less than " + info.max);
- }
- return num;
- } else if (info.type == Long.class) {
- Long num = parseLong(name, val);
- if (isRuleVal) {
- if (info.min != null)
- if (num < info.min.longValue())
- throw new RuntimeException(name + ": " + val + " must be greater than " + info.min);
- if (info.max != null)
- if (num > info.max.longValue())
- throw new RuntimeException(name + ": " + val + " must be less than " + info.max);
- }
- return num;
- } else if (info.type == String.class) {
- if (isRuleVal && info.vals != null && !info.vals.contains(val))
- throw new RuntimeException(name + ": " + val + " must be one of " + StrUtils.join(info.vals, ','));
- return val;
- } else {
- throw new RuntimeException("Invalid type ");
- }
+ if (info == null && name.startsWith(METRICS_PREFIX)) info = LazyValidator.INST;
+ return info;
}
public static Long parseLong(String name, Object val) {
@@ -425,7 +485,7 @@ public class Clause implements MapWriter, Comparable<Clause> {
num = (Number) val;
}
- if (num != null) {
+ if (num != null) {
return num.longValue();
}
throw new RuntimeException(name + ": " + val + "not a valid number");
@@ -446,12 +506,14 @@ public class Clause implements MapWriter, Comparable<Clause> {
num = (Number) val;
}
- if (num != null) {
+ if (num != null) {
return num.doubleValue();
}
throw new RuntimeException(name + ": " + val + "not a valid number");
}
+ public static final String METRICS_PREFIX = "metrics:";
+
private static final Map<String, ValidateInfo> validatetypes = new HashMap<>();
static {
@@ -467,6 +529,7 @@ public class Clause implements MapWriter, Comparable<Clause> {
validatetypes.put("NUMBER", new ValidateInfo(Long.class, null, 0L, Long.MAX_VALUE));//generic number validation
validatetypes.put("STRING", new ValidateInfo(String.class, null, null, null));//generic string validation
validatetypes.put("node", new ValidateInfo(String.class, null, null, null));
+ validatetypes.put("LAZY", LazyValidator.INST);
for (String ip : ImplicitSnitch.IP_SNITCHES) validatetypes.put(ip, new ValidateInfo(Long.class, null, 0L, 255L));
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
index e873625..58972af 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
@@ -38,7 +38,7 @@ public interface ClusterDataProvider extends Closeable {
* <p>
* the format is {collection:shard :[{replicadetails}]}
*/
- Map<String, Map<String, List<Policy.ReplicaInfo>>> getReplicaInfo(String node, Collection<String> keys);
+ Map<String, Map<String, List<ReplicaInfo>>> getReplicaInfo(String node, Collection<String> keys);
Collection<String> getNodes();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/MoveReplicaSuggester.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/MoveReplicaSuggester.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/MoveReplicaSuggester.java
index e97cb0e..b32fc98 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/MoveReplicaSuggester.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/MoveReplicaSuggester.java
@@ -21,7 +21,6 @@ import java.util.List;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.cloud.autoscaling.Clause.Violation;
-import org.apache.solr.client.solrj.cloud.autoscaling.Policy.ReplicaInfo;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy.Suggester;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.common.util.Pair;
@@ -40,41 +39,41 @@ public class MoveReplicaSuggester extends Suggester {
List<Clause.Violation> leastSeriousViolation = null;
Integer targetNodeIndex = null;
Integer sourceNodeIndex = null;
- ReplicaInfo fromReplicaInfo = null;
+ ReplicaInfo sourceReplicaInfo = null;
for (Pair<ReplicaInfo, Row> fromReplica : getValidReplicas(true, true, -1)) {
Row fromRow = fromReplica.second();
ReplicaInfo replicaInfo = fromReplica.first();
String coll = replicaInfo.collection;
String shard = replicaInfo.shard;
- Pair<Row, ReplicaInfo> pair = fromRow.removeReplica(coll, shard);
- Row tmpRow = pair.first();
- if (tmpRow == null) {
+ Pair<Row, ReplicaInfo> pair = fromRow.removeReplica(coll, shard, replicaInfo.type);
+ Row srcTmpRow = pair.first();
+ if (srcTmpRow == null) {
//no such replica available
continue;
}
- tmpRow.violations.clear();
final int i = getMatrix().indexOf(fromRow);
for (int j = getMatrix().size() - 1; j > i; j--) {
Row targetRow = getMatrix().get(j);
+ if(!targetRow.isLive) continue;
if (!isAllowed(targetRow.node, Hint.TARGET_NODE)) continue;
- targetRow = targetRow.addReplica(coll, shard);
- targetRow.violations.clear();
- List<Violation> errs = testChangedMatrix(strict, getModifiedMatrix(getModifiedMatrix(getMatrix(), tmpRow, i), targetRow, j));
- if (!containsNewErrors(errs) && isLessSerious(errs, leastSeriousViolation)) {
+ targetRow = targetRow.addReplica(coll, shard, replicaInfo.type);
+ List<Violation> errs = testChangedMatrix(strict, getModifiedMatrix(getModifiedMatrix(getMatrix(), srcTmpRow, i), targetRow, j));
+ if (!containsNewErrors(errs) && isLessSerious(errs, leastSeriousViolation) &&
+ Policy.compareRows(srcTmpRow, targetRow, session.getPolicy()) < 1) {
leastSeriousViolation = errs;
targetNodeIndex = j;
sourceNodeIndex = i;
- fromReplicaInfo = replicaInfo;
+ sourceReplicaInfo = replicaInfo;
}
}
}
if (targetNodeIndex != null && sourceNodeIndex != null) {
- getMatrix().set(sourceNodeIndex, getMatrix().get(sourceNodeIndex).removeReplica(fromReplicaInfo.collection, fromReplicaInfo.shard).first());
- getMatrix().set(targetNodeIndex, getMatrix().get(targetNodeIndex).addReplica(fromReplicaInfo.collection, fromReplicaInfo.shard));
+ getMatrix().set(sourceNodeIndex, getMatrix().get(sourceNodeIndex).removeReplica(sourceReplicaInfo.collection, sourceReplicaInfo.shard, sourceReplicaInfo.type).first());
+ getMatrix().set(targetNodeIndex, getMatrix().get(targetNodeIndex).addReplica(sourceReplicaInfo.collection, sourceReplicaInfo.shard, sourceReplicaInfo.type));
return new CollectionAdminRequest.MoveReplica(
- fromReplicaInfo.collection,
- fromReplicaInfo.name,
+ sourceReplicaInfo.collection,
+ sourceReplicaInfo.name,
getMatrix().get(targetNodeIndex).node);
}
return null;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/NoneSuggester.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/NoneSuggester.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/NoneSuggester.java
new file mode 100644
index 0000000..f686014
--- /dev/null
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/NoneSuggester.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.client.solrj.cloud.autoscaling;
+
+import org.apache.solr.client.solrj.SolrRequest;
+
+public class NoneSuggester extends Policy.Suggester{
+ @Override
+ SolrRequest init() {
+ return null;
+ }
+
+ @Override
+ public SolrRequest getOperation() {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Operand.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Operand.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Operand.java
index e012718..13738c6 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Operand.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Operand.java
@@ -25,6 +25,7 @@ import static org.apache.solr.client.solrj.cloud.autoscaling.Policy.ANY;
import java.util.Objects;
import org.apache.solr.client.solrj.cloud.autoscaling.Clause.TestStatus;
+import org.apache.solr.common.params.CoreAdminParams;
public enum Operand {
@@ -37,7 +38,7 @@ public enum Operand {
},
EQUAL("", 0) {
@Override
- public int _delta(int expected, int actual) {
+ public long _delta(long expected, long actual) {
return expected - actual;
}
},
@@ -48,7 +49,7 @@ public enum Operand {
}
@Override
- public int _delta(int expected, int actual) {
+ public long _delta(long expected, long actual) {
return expected - actual;
}
@@ -57,6 +58,7 @@ public enum Operand {
@Override
public TestStatus match(Object ruleVal, Object testVal) {
if (testVal == null) return NOT_APPLICABLE;
+ if (ruleVal instanceof String) ruleVal = Clause.parseDouble("", ruleVal);
if (ruleVal instanceof Double) {
return Double.compare(Clause.parseDouble("", testVal), (Double) ruleVal) == 1 ? PASS : FAIL;
}
@@ -64,7 +66,7 @@ public enum Operand {
}
@Override
- protected int _delta(int expected, int actual) {
+ protected long _delta(long expected, long actual) {
return actual > expected ? 0 : (expected + 1) - actual;
}
},
@@ -72,6 +74,7 @@ public enum Operand {
@Override
public TestStatus match(Object ruleVal, Object testVal) {
if (testVal == null) return NOT_APPLICABLE;
+ if (ruleVal instanceof String) ruleVal = Clause.parseDouble("", ruleVal);
if (ruleVal instanceof Double) {
return Double.compare(Clause.parseDouble("", testVal), (Double) ruleVal) == -1 ? PASS : FAIL;
}
@@ -79,7 +82,7 @@ public enum Operand {
}
@Override
- protected int _delta(int expected, int actual) {
+ protected long _delta(long expected, long actual) {
return actual < expected ? 0 : (expected ) - actual;
}
@@ -92,10 +95,6 @@ public enum Operand {
this.priority = priority;
}
- public String toStr(Object expectedVal) {
- return operand + expectedVal.toString();
- }
-
public TestStatus match(Object ruleVal, Object testVal) {
return Objects.equals(ruleVal, testVal) ? PASS : FAIL;
}
@@ -107,17 +106,17 @@ public enum Operand {
}
- public Integer delta(Object expected, Object actual) {
+ public Long delta(Object expected, Object actual) {
try {
- Integer expectedInt = Integer.parseInt(String.valueOf(expected));
- Integer actualInt = Integer.parseInt(String.valueOf(actual));
+ Long expectedInt = (Long) Clause.validate(CoreAdminParams.REPLICA, expected, false);
+ Long actualInt = (Long) Clause.validate(CoreAdminParams.REPLICA, actual, false);
return _delta(expectedInt, actualInt);
} catch (Exception e) {
return null;
}
}
- protected int _delta(int expected, int actual) {
+ protected long _delta(long expected, long actual) {
return 0;
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
index ccb0dee..12024e4 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Policy.java
@@ -32,6 +32,7 @@ import java.util.Objects;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
+import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -39,6 +40,7 @@ import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.cloud.autoscaling.Clause.Violation;
import org.apache.solr.common.IteratorWriter;
import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.params.CollectionParams.CollectionAction;
import org.apache.solr.common.util.Pair;
import org.apache.solr.common.util.StrUtils;
@@ -46,6 +48,7 @@ import org.apache.solr.common.util.Utils;
import static java.util.Collections.emptyList;
import static java.util.Collections.emptyMap;
+import static java.util.stream.Collectors.collectingAndThen;
import static java.util.stream.Collectors.toList;
/*The class that reads, parses and applies policies specified in
@@ -61,55 +64,78 @@ public class Policy implements MapWriter {
public static final String POLICY = "policy";
public static final String EACH = "#EACH";
public static final String ANY = "#ANY";
+ public static final String POLICIES = "policies";
public static final String CLUSTER_POLICY = "cluster-policy";
- public static final String CLUSTER_PREFERENCE = "cluster-preferences";
+ public static final String CLUSTER_PREFERENCES = "cluster-preferences";
public static final Set<String> GLOBAL_ONLY_TAGS = Collections.singleton("cores");
- final Map<String, List<Clause>> policies = new HashMap<>();
+ public static final Preference DEFAULT_PREFERENCE = new Preference((Map<String, Object>) Utils.fromJSONString("{minimize : cores, precision:1}"));
+ final Map<String, List<Clause>> policies;
final List<Clause> clusterPolicy;
final List<Preference> clusterPreferences;
final List<String> params;
+ public Policy() {
+ this(Collections.emptyMap());
+ }
public Policy(Map<String, Object> jsonMap) {
-
- clusterPreferences = ((List<Map<String, Object>>) jsonMap.getOrDefault(CLUSTER_PREFERENCE, emptyList())).stream()
- .map(Preference::new)
+ int[] idx = new int[1];
+ List<Preference> initialClusterPreferences = ((List<Map<String, Object>>) jsonMap.getOrDefault(CLUSTER_PREFERENCES, emptyList())).stream()
+ .map(m -> new Preference(m, idx[0]++))
.collect(toList());
- for (int i = 0; i < clusterPreferences.size() - 1; i++) {
- Preference preference = clusterPreferences.get(i);
- preference.next = clusterPreferences.get(i + 1);
+ for (int i = 0; i < initialClusterPreferences.size() - 1; i++) {
+ Preference preference = initialClusterPreferences.get(i);
+ preference.next = initialClusterPreferences.get(i + 1);
}
- if (clusterPreferences.isEmpty()) {
- clusterPreferences.add(new Preference((Map<String, Object>) Utils.fromJSONString("{minimize : cores, precision:1}")));
+ if (initialClusterPreferences.isEmpty()) {
+ initialClusterPreferences.add(DEFAULT_PREFERENCE);
}
- SortedSet<String> paramsOfInterest = new TreeSet<>();
+ this.clusterPreferences = Collections.unmodifiableList(initialClusterPreferences);
+ final SortedSet<String> paramsOfInterest = new TreeSet<>();
for (Preference preference : clusterPreferences) {
if (paramsOfInterest.contains(preference.name.name())) {
throw new RuntimeException(preference.name + " is repeated");
}
paramsOfInterest.add(preference.name.toString());
}
- this.params = new ArrayList<>(paramsOfInterest);
-
+ List<String> newParams = new ArrayList<>(paramsOfInterest);
clusterPolicy = ((List<Map<String, Object>>) jsonMap.getOrDefault(CLUSTER_POLICY, emptyList())).stream()
.map(Clause::new)
.filter(clause -> {
- clause.addTags(params);
+ clause.addTags(newParams);
return true;
})
- .collect(Collectors.toList());
+ .collect(collectingAndThen(toList(), Collections::unmodifiableList));
+
+ this.policies = Collections.unmodifiableMap(
+ policiesFromMap((Map<String, List<Map<String, Object>>>)jsonMap.getOrDefault(POLICIES, emptyMap()), newParams));
+ this.params = Collections.unmodifiableList(newParams);
- ((Map<String, List<Map<String, Object>>>) jsonMap.getOrDefault("policies", emptyMap())).forEach((s, l1) ->
- this.policies.put(s, l1.stream()
- .map(Clause::new)
- .filter(clause -> {
- if (!clause.isPerCollectiontag())
- throw new RuntimeException(clause.globalTag.name + " is only allowed in 'cluster-policy'");
- clause.addTags(params);
- return true;
- })
- .sorted()
- .collect(toList())));
+ }
+
+ private Policy(Map<String, List<Clause>> policies, List<Clause> clusterPolicy, List<Preference> clusterPreferences,
+ List<String> params) {
+ this.policies = policies != null ? Collections.unmodifiableMap(policies) : Collections.emptyMap();
+ this.clusterPolicy = clusterPolicy != null ? Collections.unmodifiableList(clusterPolicy) : Collections.emptyList();
+ this.clusterPreferences = clusterPreferences != null ? Collections.unmodifiableList(clusterPreferences) :
+ Collections.singletonList(DEFAULT_PREFERENCE);
+ this.params = params != null ? Collections.unmodifiableList(params) : Collections.emptyList();
+ }
+
+ public Policy withPolicies(Map<String, List<Clause>> policies) {
+ return new Policy(policies, clusterPolicy, clusterPreferences, params);
+ }
+
+ public Policy withClusterPreferences(List<Preference> clusterPreferences) {
+ return new Policy(policies, clusterPolicy, clusterPreferences, params);
+ }
+
+ public Policy withClusterPolicy(List<Clause> clusterPolicy) {
+ return new Policy(policies, clusterPolicy, clusterPreferences, params);
+ }
+
+ public Policy withParams(List<String> params) {
+ return new Policy(policies, clusterPolicy, clusterPreferences, params);
}
public List<Clause> getClusterPolicy() {
@@ -123,24 +149,43 @@ public class Policy implements MapWriter {
@Override
public void writeMap(EntryWriter ew) throws IOException {
if (!policies.isEmpty()) {
- ew.put("policies", (MapWriter) ew1 -> {
+ ew.put(POLICIES, (MapWriter) ew1 -> {
for (Map.Entry<String, List<Clause>> e : policies.entrySet()) {
ew1.put(e.getKey(), e.getValue());
}
});
}
if (!clusterPreferences.isEmpty()) {
- ew.put("preferences", (IteratorWriter) iw -> {
+ ew.put(CLUSTER_PREFERENCES, (IteratorWriter) iw -> {
for (Preference p : clusterPreferences) iw.add(p);
});
}
+ if (!clusterPolicy.isEmpty()) {
+ ew.put(CLUSTER_POLICY, (IteratorWriter) iw -> {
+ for (Clause c : clusterPolicy) {
+ iw.add(c);
+ }
+ });
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ Policy policy = (Policy) o;
+ if (!getPolicies().equals(policy.getPolicies())) return false;
+ if (!getClusterPolicy().equals(policy.getClusterPolicy())) return false;
+ if (!getClusterPreferences().equals(policy.getClusterPreferences())) return false;
+ return params.equals(policy.params);
}
/*This stores the logical state of the system, given a policy and
- * a cluster state.
- *
- */
+ * a cluster state.
+ *
+ */
public class Session implements MapWriter {
final List<String> nodes;
final ClusterDataProvider dataProvider;
@@ -208,20 +253,7 @@ public class Policy implements MapWriter {
* Apply the preferences and conditions
*/
private void applyRules() {
- if (!clusterPreferences.isEmpty()) {
- //this is to set the approximate value according to the precision
- ArrayList<Row> tmpMatrix = new ArrayList<>(matrix);
- for (Preference p : clusterPreferences) {
- Collections.sort(tmpMatrix, (r1, r2) -> p.compare(r1, r2, false));
- p.setApproxVal(tmpMatrix);
- }
- //approximate values are set now. Let's do recursive sorting
- Collections.sort(matrix, (Row r1, Row r2) -> {
- int result = clusterPreferences.get(0).compare(r1, r2, true);
- if (result == 0) result = clusterPreferences.get(0).compare(r1, r2, false);
- return result;
- });
- }
+ setApproxValuesAndSortNodes(clusterPreferences, matrix);
for (Clause clause : expandedClauses) {
List<Violation> errs = clause.test(matrix);
@@ -229,6 +261,8 @@ public class Policy implements MapWriter {
}
}
+
+
public List<Violation> getViolations() {
return violations;
}
@@ -258,6 +292,22 @@ public class Policy implements MapWriter {
}
}
+ static void setApproxValuesAndSortNodes(List<Preference> clusterPreferences, List<Row> matrix) {
+ if (!clusterPreferences.isEmpty()) {
+ //this is to set the approximate value according to the precision
+ ArrayList<Row> tmpMatrix = new ArrayList<>(matrix);
+ for (Preference p : clusterPreferences) {
+ Collections.sort(tmpMatrix, (r1, r2) -> p.compare(r1, r2, false));
+ p.setApproxVal(tmpMatrix);
+ }
+ //approximate values are set now. Let's do recursive sorting
+ Collections.sort(matrix, (Row r1, Row r2) -> {
+ int result = clusterPreferences.get(0).compare(r1, r2, true);
+ if (result == 0) result = clusterPreferences.get(0).compare(r1, r2, false);
+ return result;
+ });
+ }
+ }
public Session createSession(ClusterDataProvider dataProvider) {
return new Session(dataProvider);
@@ -298,37 +348,6 @@ public class Policy implements MapWriter {
}
- public static class ReplicaInfo implements MapWriter {
- final String name;
- String core, collection, shard;
- Map<String, Object> variables;
-
- public ReplicaInfo(String name, String coll, String shard, Map<String, Object> vals) {
- this.name = name;
- this.variables = vals;
- this.collection = coll;
- this.shard = shard;
- }
-
- @Override
- public void writeMap(EntryWriter ew) throws IOException {
- ew.put(name, variables);
- }
-
- public String getCore() {
- return core;
- }
-
- public String getCollection() {
- return collection;
- }
-
- public String getShard() {
- return shard;
- }
- }
-
-
/* A suggester is capable of suggesting a collection operation
* given a particular session. Before it suggests a new operation,
* it ensures that ,
@@ -348,10 +367,12 @@ public class Policy implements MapWriter {
}
public Suggester hint(Hint hint, Object value) {
- if (hint == Hint.TARGET_NODE || hint == Hint.SRC_NODE) {
- ((Set) hints.computeIfAbsent(hint, h -> new HashSet<>())).add(value);
+ hint.validator.accept(value);
+ if (hint.multiValued) {
+ Collection<?> values = value instanceof Collection ? (Collection)value : Collections.singletonList(value);
+ ((Set) hints.computeIfAbsent(hint, h -> new HashSet<>())).addAll(values);
} else {
- hints.put(hint, value);
+ hints.put(hint, value == null ? null : String.valueOf(value));
}
return this;
}
@@ -361,21 +382,30 @@ public class Policy implements MapWriter {
public SolrRequest getOperation() {
if (!isInitialized) {
- String coll = (String) hints.get(Hint.COLL);
- String shard = (String) hints.get(Hint.SHARD);
- // if this is not a known collection from the existing clusterstate,
- // then add it
- if (session.matrix.stream().noneMatch(row -> row.collectionVsShardVsReplicas.containsKey(coll))) {
- session.addClausesForCollection(session.dataProvider, coll);
+ Set<String> collections = (Set<String>) hints.getOrDefault(Hint.COLL, Collections.emptySet());
+ Set<Pair<String, String>> s = (Set<Pair<String, String>>) hints.getOrDefault(Hint.COLL_SHARD, Collections.emptySet());
+ if (!collections.isEmpty() || !s.isEmpty()) {
+ HashSet<Pair<String, String>> shards = new HashSet<>(s);
+ collections.stream().forEach(c -> shards.add(new Pair<>(c, null)));
+ for (Pair<String, String> shard : shards) {
+ // if this is not a known collection from the existing clusterstate,
+ // then add it
+ if (session.matrix.stream().noneMatch(row -> row.collectionVsShardVsReplicas.containsKey(shard.first()))) {
+ session.addClausesForCollection(session.dataProvider, shard.first());
+ }
+ for (Row row : session.matrix) {
+ Map<String, List<ReplicaInfo>> shardInfo = row.collectionVsShardVsReplicas.computeIfAbsent(shard.first(), it -> new HashMap<>());
+ if (shard.second() != null) shardInfo.computeIfAbsent(shard.second(), it -> new ArrayList<>());
+ }
+ }
Collections.sort(session.expandedClauses);
}
- if (coll != null) {
- for (Row row : session.matrix) {
- if (!row.collectionVsShardVsReplicas.containsKey(coll)) row.collectionVsShardVsReplicas.put(coll, new HashMap<>());
- if (shard != null) {
- Map<String, List<ReplicaInfo>> shardInfo = row.collectionVsShardVsReplicas.get(coll);
- if (!shardInfo.containsKey(shard)) shardInfo.put(shard, new ArrayList<>());
- }
+ Set<String> srcNodes = (Set<String>) hints.get(Hint.SRC_NODE);
+ if (srcNodes != null && !srcNodes.isEmpty()) {
+ // the source node is dead so live nodes may not have it
+ for (String srcNode : srcNodes) {
+ if(session.matrix.stream().noneMatch(row -> row.node.equals(srcNode)))
+ session.matrix.add(new Row(srcNode, session.getPolicy().params, session.dataProvider));
}
}
session.applyRules();
@@ -420,7 +450,7 @@ public class Policy implements MapWriter {
}
List<Pair<ReplicaInfo, Row>> getValidReplicas(boolean sortDesc, boolean isSource, int until) {
- List<Pair<Policy.ReplicaInfo, Row>> allPossibleReplicas = new ArrayList<>();
+ List<Pair<ReplicaInfo, Row>> allPossibleReplicas = new ArrayList<>();
if (sortDesc) {
if (until == -1) until = getMatrix().size();
@@ -433,18 +463,20 @@ public class Policy implements MapWriter {
return allPossibleReplicas;
}
- void addReplicaToList(Row r, boolean isSource, List<Pair<Policy.ReplicaInfo, Row>> replicaList) {
+ void addReplicaToList(Row r, boolean isSource, List<Pair<ReplicaInfo, Row>> replicaList) {
if (!isAllowed(r.node, isSource ? Hint.SRC_NODE : Hint.TARGET_NODE)) return;
- for (Map.Entry<String, Map<String, List<Policy.ReplicaInfo>>> e : r.collectionVsShardVsReplicas.entrySet()) {
+ for (Map.Entry<String, Map<String, List<ReplicaInfo>>> e : r.collectionVsShardVsReplicas.entrySet()) {
if (!isAllowed(e.getKey(), Hint.COLL)) continue;
- for (Map.Entry<String, List<Policy.ReplicaInfo>> shard : e.getValue().entrySet()) {
- if (!isAllowed(e.getKey(), Hint.SHARD)) continue;
+ for (Map.Entry<String, List<ReplicaInfo>> shard : e.getValue().entrySet()) {
+ if (!isAllowed(new Pair<>(e.getKey(), shard.getKey()), Hint.COLL_SHARD)) continue;//todo fix
+ if(shard.getValue() == null || shard.getValue().isEmpty()) continue;
replicaList.add(new Pair<>(shard.getValue().get(0), r));
}
}
}
- protected List<Violation> testChangedMatrix(boolean strict, List<Row> rows) {
+ List<Violation> testChangedMatrix(boolean strict, List<Row> rows) {
+ setApproxValuesAndSortNodes(session.getPolicy().clusterPreferences,rows);
List<Violation> errors = new ArrayList<>();
for (Clause clause : session.expandedClauses) {
if (strict || clause.strict) {
@@ -465,7 +497,7 @@ public class Policy implements MapWriter {
protected boolean isAllowed(Object v, Hint hint) {
Object hintVal = hints.get(hint);
- if (hint == Hint.TARGET_NODE || hint == Hint.SRC_NODE) {
+ if (hint.multiValued) {
Set set = (Set) hintVal;
return set == null || set.contains(v);
} else {
@@ -474,12 +506,68 @@ public class Policy implements MapWriter {
}
public enum Hint {
- COLL, SHARD, SRC_NODE, TARGET_NODE
+ COLL(true),
+ // collection shard pair
+ // this should be a Pair<String, String> , (collection,shard)
+ COLL_SHARD(true, v -> {
+ Collection c = v instanceof Collection ? (Collection) v : Collections.singleton(v);
+ for (Object o : c) {
+ if (!(o instanceof Pair)) {
+ throw new RuntimeException("SHARD hint must use a Pair");
+ }
+ Pair p = (Pair) o;
+ if (p.first() == null || p.second() == null) {
+ throw new RuntimeException("Both collection and shard must not be null");
+ }
+ }
+
+ }),
+ SRC_NODE(true),
+ TARGET_NODE(true),
+ REPLICATYPE(false, o -> {
+ if (!(o instanceof Replica.Type)) {
+ throw new RuntimeException("REPLICATYPE hint must use a ReplicaType");
+ }
+ });
+
+ public final boolean multiValued;
+ public final Consumer<Object> validator;
+
+ Hint(boolean multiValued) {
+ this(multiValued, v -> {
+ Collection c = v instanceof Collection ? (Collection) v : Collections.singleton(v);
+ for (Object o : c) {
+ if (!(o instanceof String)) throw new RuntimeException("hint must be of type String");
+ }
+ });
+ }
+
+ Hint(boolean multiValued, Consumer<Object> c) {
+ this.multiValued = multiValued;
+ this.validator = c;
+ }
+
}
}
+ public static Map<String, List<Clause>> policiesFromMap(Map<String, List<Map<String, Object>>> map, List<String> newParams) {
+ Map<String, List<Clause>> newPolicies = new HashMap<>();
+ map.forEach((s, l1) ->
+ newPolicies.put(s, l1.stream()
+ .map(Clause::new)
+ .filter(clause -> {
+ if (!clause.isPerCollectiontag())
+ throw new RuntimeException(clause.globalTag.name + " is only allowed in 'cluster-policy'");
+ clause.addTags(newParams);
+ return true;
+ })
+ .sorted()
+ .collect(collectingAndThen(toList(), Collections::unmodifiableList))));
+ return newPolicies;
+ }
+
public static List<Clause> mergePolicies(String coll,
List<Clause> collPolicy,
List<Clause> globalPolicy) {
@@ -518,4 +606,21 @@ public class Policy implements MapWriter {
public Map<String, List<Clause>> getPolicies() {
return policies;
}
+
+ public List<String> getParams() {
+ return params;
+ }
+
+ /**
+ * Compares two {@link Row} loads according to a policy.
+ *
+ * @param r1 the first {@link Row} to compare
+ * @param r2 the second {@link Row} to compare
+ * @return the value {@code 0} if r1 and r2 are equally loaded
+ * a value {@code -1} if r1 is more loaded than r2
+ * a value {@code 1} if r1 is less loaded than r2
+ */
+ static int compareRows(Row r1, Row r2, Policy policy) {
+ return policy.clusterPreferences.get(0).compare(r1, r2, true);
+ }
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
index deea175..bc2a5d7 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
@@ -20,28 +20,35 @@ package org.apache.solr.client.solrj.cloud.autoscaling;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.HashMap;
+import java.util.EnumMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy.Suggester.Hint;
import org.apache.solr.common.SolrException;
-import org.apache.solr.common.params.CoreAdminParams;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.ReplicaPosition;
+import org.apache.solr.common.util.Pair;
import org.apache.solr.common.util.Utils;
import static org.apache.solr.common.params.CollectionParams.CollectionAction.ADDREPLICA;
+import static org.apache.solr.common.params.CoreAdminParams.NODE;
public class PolicyHelper {
- public static Map<String, List<String>> getReplicaLocations(String collName, Map<String, Object> autoScalingJson,
- ClusterDataProvider cdp,
- Map<String, String> optionalPolicyMapping,
- List<String> shardNames,
- int repFactor,
- List<String> nodesList) {
- Map<String, List<String>> positionMapping = new HashMap<>();
- for (String shardName : shardNames) positionMapping.put(shardName, new ArrayList<>(repFactor));
- if (optionalPolicyMapping != null) {
+ private static ThreadLocal<Map<String, String>> policyMapping = new ThreadLocal<>();
+ public static List<ReplicaPosition> getReplicaLocations(String collName, AutoScalingConfig autoScalingConfig,
+ ClusterDataProvider cdp,
+ Map<String, String> optionalPolicyMapping,
+ List<String> shardNames,
+ int nrtReplicas,
+ int tlogReplicas,
+ int pullReplicas,
+ List<String> nodesList) {
+ List<ReplicaPosition> positions = new ArrayList<>();
final ClusterDataProvider delegate = cdp;
cdp = new ClusterDataProvider() {
@Override
@@ -50,7 +57,7 @@ public class PolicyHelper {
}
@Override
- public Map<String, Map<String, List<Policy.ReplicaInfo>>> getReplicaInfo(String node, Collection<String> keys) {
+ public Map<String, Map<String, List<ReplicaInfo>>> getReplicaInfo(String node, Collection<String> keys) {
return delegate.getReplicaInfo(node, keys);
}
@@ -61,36 +68,126 @@ public class PolicyHelper {
@Override
public String getPolicyNameByCollection(String coll) {
- return optionalPolicyMapping.containsKey(coll) ?
+ return policyMapping.get() != null && policyMapping.get().containsKey(coll) ?
optionalPolicyMapping.get(coll) :
delegate.getPolicyNameByCollection(coll);
}
};
+ policyMapping.set(optionalPolicyMapping);
+ Policy.Session session = null;
+ try {
+ session = SESSION_REF.get() != null ?
+ SESSION_REF.get().initOrGet(cdp, autoScalingConfig.getPolicy()) :
+ autoScalingConfig.getPolicy().createSession(cdp);
+
+ Map<Replica.Type, Integer> typeVsCount = new EnumMap<>(Replica.Type.class);
+ typeVsCount.put(Replica.Type.NRT, nrtReplicas);
+ typeVsCount.put(Replica.Type.TLOG, tlogReplicas);
+ typeVsCount.put(Replica.Type.PULL, pullReplicas);
+ for (String shardName : shardNames) {
+ int idx = 0;
+ for (Map.Entry<Replica.Type, Integer> e : typeVsCount.entrySet()) {
+ for (int i = 0; i < e.getValue(); i++) {
+ Policy.Suggester suggester = session.getSuggester(ADDREPLICA)
+ .hint(Hint.REPLICATYPE, e.getKey())
+ .hint(Hint.COLL_SHARD, new Pair<>(collName, shardName));
+ if (nodesList != null) {
+ for (String nodeName : nodesList) {
+ suggester = suggester.hint(Hint.TARGET_NODE, nodeName);
+ }
+ }
+ SolrRequest op = suggester.getOperation();
+ if (op == null) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "No node can satisfy the rules " +
+ Utils.toJSONString(Utils.getDeepCopy(session.expandedClauses, 4, true)));
+ }
+ session = suggester.getSession();
+ positions.add(new ReplicaPosition(shardName, ++idx, e.getKey(), op.getParams().get(NODE)));
+ }
+ }
+ }
+ } finally {
+ if (session != null && SESSION_REF.get() != null) SESSION_REF.get().updateSession(session);
+ policyMapping.remove();
}
+ return positions;
+ }
- Policy policy = new Policy(autoScalingJson);
- Policy.Session session = policy.createSession(cdp);
- for (String shardName : shardNames) {
- for (int i = 0; i < repFactor; i++) {
- Policy.Suggester suggester = session.getSuggester(ADDREPLICA)
- .hint(Hint.COLL, collName)
- .hint(Hint.SHARD, shardName);
- if (nodesList != null) {
- for (String nodeName : nodesList) {
- suggester = suggester.hint(Hint.TARGET_NODE, nodeName);
- }
+ public static final int SESSION_EXPIRY = 180;//3 seconds
+ public static ThreadLocal<Long> REF_VERSION = new ThreadLocal<>();
+
+ public static class SessionRef {
+ private final AtomicLong myVersion = new AtomicLong(0);
+ AtomicInteger refCount = new AtomicInteger();
+ private Policy.Session session;
+ long lastUsedTime;
+
+ public SessionRef() {
+ }
+
+ public long getRefVersion(){
+ return myVersion.get();
+ }
+
+
+ public void decref(long version) {
+ synchronized (SessionRef.class) {
+ if (session == null) return;
+ if(myVersion.get() != version) return;
+ if (refCount.decrementAndGet() <= 0) {
+ session = null;
+ lastUsedTime = 0;
}
- SolrRequest op = suggester.getOperation();
- if (op == null) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "No node can satisfy the rules "+ Utils.toJSONString(Utils.getDeepCopy(session.expandedClauses, 4, true)));
+ }
+ }
+
+ public int getRefCount() {
+ return refCount.get();
+ }
+
+ public Policy.Session get() {
+ synchronized (SessionRef.class) {
+ if (session == null) return null;
+ if (TimeUnit.SECONDS.convert(System.nanoTime() - lastUsedTime, TimeUnit.NANOSECONDS) > SESSION_EXPIRY) {
+ session = null;
+ return null;
+ } else {
+ REF_VERSION.set(myVersion.get());
+ refCount.incrementAndGet();
+ return session;
}
- session = suggester.getSession();
- positionMapping.get(shardName).add(op.getParams().get(CoreAdminParams.NODE));
}
}
- return positionMapping;
+ public Policy.Session initOrGet(ClusterDataProvider cdp, Policy policy) {
+ synchronized (SessionRef.class) {
+ Policy.Session session = get();
+ if (session != null) return session;
+ this.session = policy.createSession(cdp);
+ myVersion.incrementAndGet();
+ lastUsedTime = System.nanoTime();
+ REF_VERSION.set(myVersion.get());
+ refCount.set(1);
+ return this.session;
+ }
+ }
+
+
+ private void updateSession(Policy.Session session) {
+ this.session = session;
+ lastUsedTime = System.nanoTime();
+ }
+ }
+
+ public static void clearFlagAndDecref(SessionRef policySessionRef) {
+ Long refVersion = REF_VERSION.get();
+ if (refVersion != null) policySessionRef.decref(refVersion);
+ REF_VERSION.remove();
}
+
+ public static ThreadLocal<SessionRef> SESSION_REF = new ThreadLocal<>();
+
+
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/ccf1bf72/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Preference.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Preference.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Preference.java
index bb45628..0142107 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Preference.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Preference.java
@@ -18,22 +18,32 @@
package org.apache.solr.client.solrj.cloud.autoscaling;
import java.io.IOException;
+import java.lang.invoke.MethodHandles;
import java.util.List;
import java.util.Map;
import org.apache.solr.common.MapWriter;
import org.apache.solr.common.util.StrUtils;
import org.apache.solr.common.util.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class Preference implements MapWriter {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
final Policy.SortParam name;
Integer precision;
final Policy.Sort sort;
Preference next;
- public int idx;
+ final int idx;
private final Map original;
- Preference(Map<String, Object> m) {
+ public Preference(Map<String, Object> m) {
+ this(m, 0);
+ }
+
+ public Preference(Map<String, Object> m, int idx) {
+ this.idx = idx;
this.original = Utils.getDeepCopy(m,3);
sort = Policy.Sort.get(m);
name = Policy.SortParam.get(m.get(sort.name()).toString());
@@ -43,7 +53,7 @@ public class Preference implements MapWriter {
throw new RuntimeException("precision must be a positive value ");
}
if(precision< name.min || precision> name.max){
- throw new RuntimeException(StrUtils.formatString("invalid precision value {0} must lie between {1} and {1}",
+ throw new RuntimeException(StrUtils.formatString("invalid precision value {0} , must lie between {1} and {2}",
precision, name.min, name.max ) );
}
@@ -53,6 +63,9 @@ public class Preference implements MapWriter {
// recursive, it uses the precision to tie & when there is a tie use the next preference to compare
// in non-recursive mode, precision is not taken into consideration and sort is done on actual value
int compare(Row r1, Row r2, boolean useApprox) {
+ if (!r1.isLive && !r2.isLive) return 0;
+ if (!r1.isLive) return -1;
+ if (!r2.isLive) return 1;
Object o1 = useApprox ? r1.cells[idx].approxVal : r1.cells[idx].val;
Object o2 = useApprox ? r2.cells[idx].approxVal : r2.cells[idx].val;
int result = 0;
@@ -68,6 +81,9 @@ public class Preference implements MapWriter {
void setApproxVal(List<Row> tmpMatrix) {
Object prevVal = null;
for (Row row : tmpMatrix) {
+ if (!row.isLive) {
+ continue;
+ }
prevVal = row.cells[idx].approxVal =
(prevVal == null || Double.compare(Math.abs(((Number) prevVal).doubleValue() - ((Number) row.cells[idx].val).doubleValue()), precision) > 0) ?
row.cells[idx].val :
@@ -83,7 +99,27 @@ public class Preference implements MapWriter {
}
}
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ Preference that = (Preference) o;
+
+ if (idx != that.idx) return false;
+ if (getName() != that.getName()) return false;
+ if (precision != null ? !precision.equals(that.precision) : that.precision != null) return false;
+ if (sort != that.sort) return false;
+ if (next != null ? !next.equals(that.next) : that.next != null) return false;
+ return original.equals(that.original);
+ }
+
public Policy.SortParam getName() {
return name;
}
+
+ @Override
+ public String toString() {
+ return Utils.toJSONString(this);
+ }
}