You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ja...@apache.org on 2021/03/26 16:43:03 UTC
[lucene] tag history/branches/lucene-solr/jira/solr14275 created
(now 243b3b8)
This is an automated email from the ASF dual-hosted git repository.
janhoy pushed a change to tag history/branches/lucene-solr/jira/solr14275
in repository https://gitbox.apache.org/repos/asf/lucene.git.
at 243b3b8 (commit)
This tag includes the following new commits:
new 2b4e88f Optimization for policy rules
new 9783337 Optimized FreeDiskVariable
new 243b3b8 fixed the test
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[lucene] 02/03: Optimized FreeDiskVariable
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
janhoy pushed a commit to tag history/branches/lucene-solr/jira/solr14275
in repository https://gitbox.apache.org/repos/asf/lucene.git
commit 97833376a1de98c46b2de33357e0ab58c624fbc3
Author: noble <no...@apache.org>
AuthorDate: Mon Mar 30 13:22:46 2020 +1100
Optimized FreeDiskVariable
---
.../solr/client/solrj/cloud/autoscaling/FreeDiskVariable.java | 7 +++++++
.../solr/client/solrj/cloud/autoscaling/PerClauseData.java | 1 +
.../org/apache/solr/client/solrj/cloud/autoscaling/Row.java | 10 +++++++++-
3 files changed, 17 insertions(+), 1 deletion(-)
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/FreeDiskVariable.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/FreeDiskVariable.java
index 2193ea3..ddc6e1a 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/FreeDiskVariable.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/FreeDiskVariable.java
@@ -148,6 +148,13 @@ public class FreeDiskVariable extends VariableBase {
@Override
public void projectAddReplica(Cell cell, ReplicaInfo ri, Consumer<Row.OperationInfo> ops, boolean strictMode) {
//go through other replicas of this shard and copy the index size value into this
+ if(cell.getRow().session.perClauseData != null){
+ PerClauseData.ShardDetails sd = cell.getRow().session.perClauseData.getShardDetails(ri.getCollection(), ri.getShard());
+ if(sd.indexSize == null) return;
+ Double currFreeDisk = cell.val == null ? 0.0d : (Double) cell.val;
+ cell.val = currFreeDisk - sd.indexSize;
+ return;
+ }
for (Row row : cell.getRow().session.matrix) {
row.forEachReplica(replicaInfo -> {
if (ri != replicaInfo &&
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PerClauseData.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PerClauseData.java
index 5e152b8..cc815ba 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PerClauseData.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PerClauseData.java
@@ -91,6 +91,7 @@ public class PerClauseData implements ReflectMapWriter, Cloneable {
public static class ShardDetails implements ReflectMapWriter, Cloneable {
final String coll;
final String shard;
+ Double indexSize;
ReplicaCount replicas = new ReplicaCount();
public Map<Clause, Map<String, ReplicaCount>> values = new HashMap<>();
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java
index 9c69c02..a288815 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java
@@ -41,6 +41,7 @@ import org.apache.solr.common.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.solr.client.solrj.cloud.autoscaling.Variable.Type.CORE_IDX;
import static org.apache.solr.common.params.CoreAdminParams.NODE;
/**
@@ -400,7 +401,14 @@ public class Row implements MapWriter {
void initPerClauseData() {
if(session== null || session.perClauseData == null) return;
- forEachReplica(it -> session.perClauseData.getShardDetails(it.getCollection(), it.getShard()).replicas.increment(it.getType()));
+ forEachReplica(it -> {
+ PerClauseData.ShardDetails shardDetails = session.perClauseData.getShardDetails(it.getCollection(), it.getShard());
+ shardDetails.replicas.increment(it.getType());
+ Number idxSize = (Number) it.getVariable(CORE_IDX.tagName);
+ if(idxSize != null){
+ shardDetails.indexSize = idxSize.doubleValue();
+ }
+ });
for (Clause clause : session.expandedClauses) {
if(clause.put == Clause.Put.ON_EACH) continue;
if(clause.dataGrouping == Clause.DataGrouping.SHARD || clause.dataGrouping == Clause.DataGrouping.COLL) {
[lucene] 03/03: fixed the test
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
janhoy pushed a commit to tag history/branches/lucene-solr/jira/solr14275
in repository https://gitbox.apache.org/repos/asf/lucene.git
commit 243b3b88ba913682d71bc55ca06efebffdbf56db
Author: noble <no...@apache.org>
AuthorDate: Tue Mar 31 15:22:06 2020 +1100
fixed the test
---
.../org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy2.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy2.java b/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy2.java
index d4b69b1..95ab6a3 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy2.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy2.java
@@ -561,10 +561,10 @@ public class TestPolicy2 extends SolrTestCaseJ4 {
Suggester suggester = session.getSuggester(CollectionParams.CollectionAction.ADDREPLICA);
suggester.hint(Suggester.Hint.COLL_SHARD, new Pair<>(collName, "shards_" + i));
SolrRequest op = suggester.getSuggestion();
+ session = suggester.getSession();
assertNotNull(op);
}
}
-
System.out.println("Time taken : "+ (System.currentTimeMillis() -start));
}
[lucene] 01/03: Optimization for policy rules
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
janhoy pushed a commit to tag history/branches/lucene-solr/jira/solr14275
in repository https://gitbox.apache.org/repos/asf/lucene.git
commit 2b4e88fb57d3494a29cac1deb8d0348faaa18cfa
Author: noble <no...@apache.org>
AuthorDate: Sat Mar 28 18:20:16 2020 +1100
Optimization for policy rules
---
.../cloud/autoscaling/AddReplicaSuggester.java | 8 +
.../client/solrj/cloud/autoscaling/Clause.java | 29 +++-
.../solrj/cloud/autoscaling/PerClauseData.java | 187 +++++++++++++++++++++
.../client/solrj/cloud/autoscaling/Policy.java | 8 +-
.../solrj/cloud/autoscaling/ReplicaCount.java | 21 +++
.../solrj/cloud/autoscaling/ReplicaVariable.java | 24 ++-
.../solr/client/solrj/cloud/autoscaling/Row.java | 39 +++++
.../client/solrj/cloud/autoscaling/Violation.java | 9 +-
.../solrj/solr/autoscaling/emptydiagnostics.json | 20 +++
.../solrj/cloud/autoscaling/TestPolicy2.java | 67 ++++++++
10 files changed, 397 insertions(+), 15 deletions(-)
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 87b831a..e24d606 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
@@ -47,6 +47,11 @@ class AddReplicaSuggester extends Suggester {
//iterate through nodes and identify the least loaded
List<Violation> leastSeriousViolation = null;
Row bestNode = null;
+ // nocommit
+ // possible optimization? for large clusters compare only up to
+ // that many eligible rows from the top
+ int limitTopRows = getMatrix().size() > 100 ? 3 : getMatrix().size();
+
for (int i = getMatrix().size() - 1; i >= 0; i--) {
Row row = getMatrix().get(i);
if (!isNodeSuitableForReplicaAddition(row, null)) continue;
@@ -58,6 +63,9 @@ class AddReplicaSuggester extends Suggester {
leastSeriousViolation = errs;
bestNode = tmpRow;
}
+ if (bestNode != null && limitTopRows-- < 0) {
+ break;
+ }
}
}
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 f12ecce..a4ff43e 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
@@ -69,6 +69,7 @@ public class Clause implements MapWriter, Comparable<Clause> {
final Replica.Type type;
Put put;
boolean strict;
+ DataGrouping dataGrouping;
protected Clause(Clause clause, Function<Condition, Object> computedValueEvaluator) {
this.original = clause.original;
@@ -84,10 +85,11 @@ public class Clause implements MapWriter, Comparable<Clause> {
this.hasComputedValue = clause.hasComputedValue;
this.strict = clause.strict;
derivedFrom = clause.derivedFrom;
+ this.dataGrouping =clause.dataGrouping;
}
// internal use only
- Clause(Map<String, Object> original, Condition tag, Condition globalTag, boolean isStrict, Put put, boolean nodeSetPresent) {
+ Clause(Map<String, Object> original, Condition tag, Condition globalTag, boolean isStrict, Put put, boolean nodeSetPresent, DataGrouping dataGrouping) {
this.hashCode = original.hashCode();
this.original = original;
this.tag = tag;
@@ -145,6 +147,15 @@ public class Clause implements MapWriter, Comparable<Clause> {
}
doPostValidate(collection, shard, replica, tag, globalTag);
hasComputedValue = hasComputedValue();
+
+ if(globalTag != null || tag.name.equals("node")){
+ dataGrouping = DataGrouping.NODE;
+ } else if(shard.val.toString().equals(ANY)){
+ dataGrouping = DataGrouping.COLL;
+ } else {
+ dataGrouping = DataGrouping.SHARD;
+ }
+
}
private boolean parseNodeset(Map<String, Object> m) {
@@ -660,17 +671,23 @@ public class Clause implements MapWriter, Comparable<Clause> {
public List<Violation> test(Policy.Session session, double[] deviations) {
if (isPerCollectiontag()) {
if(nodeSetPresent) {
- if(put == Put.ON_EACH){
+ if(put == Put.ON_EACH) {
return testPerNode(session, deviations) ;
} else {
- return testGroupNodes(session, deviations);
+ return session.perClauseData.computeViolations(session, this);
+ //return testGroupNodes(session, deviations);
}
}
+ if(tag.varType == Type.NODE){
+ return session.perClauseData.computeViolations(session, this);
+ }
+
+
return tag.varType == Type.NODE ||
(tag.varType.meta.isNodeSpecificVal() && replica.computedType == null) ?
testPerNode(session, deviations) :
- testGroupNodes(session, deviations);
+ testGroupNodes(session, deviations);
} else {
ComputedValueEvaluator computedValueEvaluator = new ComputedValueEvaluator(session);
Violation.Ctx ctx = new Violation.Ctx(this, session.matrix, computedValueEvaluator);
@@ -826,4 +843,8 @@ public class Clause implements MapWriter, Comparable<Clause> {
}
}
+ enum DataGrouping {
+ COLL, SHARD, NODE,GLOBAL;
+ }
+
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PerClauseData.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PerClauseData.java
new file mode 100644
index 0000000..5e152b8
--- /dev/null
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PerClauseData.java
@@ -0,0 +1,187 @@
+/*
+ * 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.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+
+import org.apache.solr.common.annotation.JsonProperty;
+import org.apache.solr.common.util.ReflectMapWriter;
+import org.apache.solr.common.util.Utils;
+
+public class PerClauseData implements ReflectMapWriter, Cloneable {
+ @JsonProperty
+ public Map<String, CollectionDetails> collections = new HashMap<>();
+
+ static Function<String, ReplicaCount> NEW_CLAUSEVAL_FUN = c -> new ReplicaCount();
+
+ ReplicaCount getClauseValue(String coll, String shard, Clause clause, String key) {
+ Map<String, ReplicaCount> countMap = getCountsForClause(coll, shard, clause);
+ return countMap.computeIfAbsent(key,NEW_CLAUSEVAL_FUN);
+ }
+
+ private Map<String, ReplicaCount> getCountsForClause(String coll, String shard, Clause clause) {
+ CollectionDetails cd = collections.get(coll);
+ if (cd == null) collections.put(coll, cd = new CollectionDetails(coll));
+ ShardDetails psd = null;
+ if (shard != null && clause.dataGrouping == Clause.DataGrouping.SHARD) {
+ psd = cd.shards.get(shard);
+ if (psd == null) cd.shards.put(shard, psd = new ShardDetails(coll, shard));
+ }
+
+ Map<Clause, Map<String, ReplicaCount>> map = (psd == null ? cd.clauseValues : psd.values);
+
+ return (Map<String, ReplicaCount>) map.computeIfAbsent(clause, Utils.NEW_HASHMAP_FUN);
+ }
+
+ PerClauseData copy() {
+ PerClauseData result = new PerClauseData();
+ collections.forEach((s, v) -> result.collections.put(s, v.copy()));
+ return result;
+ }
+
+
+
+ ShardDetails getShardDetails(String c, String s) {
+ CollectionDetails cd = collections.get(c);
+ if (cd == null) collections.put(c, cd = new CollectionDetails(c));
+ ShardDetails sd = cd.shards.get(s);
+ if (sd == null) cd.shards.put(c, sd = new ShardDetails(c, s));
+ return sd;
+ }
+
+
+ public static class CollectionDetails implements ReflectMapWriter, Cloneable {
+
+ final String coll;
+ Map<String, ShardDetails> shards = new HashMap<>();
+
+ Map<Clause, Map<String, ReplicaCount>> clauseValues = new HashMap<>();
+
+ CollectionDetails copy() {
+ CollectionDetails result = new CollectionDetails(coll);
+ shards.forEach((k, shardDetails) -> result.shards.put(k, shardDetails.copy()));
+ return result;
+ }
+
+ CollectionDetails(String coll) {
+ this.coll = coll;
+ }
+ }
+
+ public static class ShardDetails implements ReflectMapWriter, Cloneable {
+ final String coll;
+ final String shard;
+ ReplicaCount replicas = new ReplicaCount();
+
+ public Map<Clause, Map<String, ReplicaCount>> values = new HashMap<>();
+
+ ShardDetails(String coll, String shard) {
+ this.coll = coll;
+ this.shard = shard;
+ }
+
+
+ ShardDetails copy() {
+ ShardDetails result = new ShardDetails(coll, shard);
+ values.forEach((clause, clauseVal) -> {
+ HashMap<String,ReplicaCount> m = new HashMap(clauseVal);
+ for (Map.Entry<String, ReplicaCount> e : m.entrySet()) e.setValue(e.getValue().copy());
+ result.values.put(clause, m);
+ });
+ return result;
+ }
+ }
+
+ static class LazyViolation extends Violation {
+ private Policy.Session session;
+
+ LazyViolation(SealedClause clause, String coll, String shard, String node, Object actualVal, Double replicaCountDelta, Object tagKey, Policy.Session session) {
+ super(clause, coll, shard, node, actualVal, replicaCountDelta, tagKey);
+ super.replicaInfoAndErrs = null;
+ this.session = session;
+ }
+
+ @Override
+ public List<ReplicaInfoAndErr> getViolatingReplicas() {
+ if(replicaInfoAndErrs == null){
+ populateReplicas();
+ }
+ return replicaInfoAndErrs;
+ }
+
+ private void populateReplicas() {
+ replicaInfoAndErrs = new ArrayList<>();
+ for (Row row : session.matrix) {
+ if(getClause().getThirdTag().isPass(row)) {
+ row.forEachReplica(coll, ri -> {
+ if(shard == null || Objects.equals(shard, ri.getShard()))
+ replicaInfoAndErrs.add(new ReplicaInfoAndErr(ri));
+ });
+
+ }
+ }
+ }
+ }
+
+ void getViolations( Map<Clause, Map<String, ReplicaCount>> vals ,
+ List<Violation> violations,
+ Clause.ComputedValueEvaluator evaluator,
+ Clause clause){
+
+ Map<String, ReplicaCount> cv = vals.get(clause);
+ if (cv == null || cv.isEmpty()) return;
+ SealedClause sc = clause.getSealedClause(evaluator);
+ cv.forEach((s, replicaCount) -> {
+ if (!sc.replica.isPass(replicaCount)) {
+ Violation v = new LazyViolation(
+ sc,
+ evaluator.collName,
+ evaluator.shardName,
+ null,
+ replicaCount,
+ sc.getReplica().replicaCountDelta(replicaCount),
+ s,
+ evaluator.session);
+ violations.add(v);
+ }
+ });
+
+ }
+
+ List<Violation> computeViolations(Policy.Session session, Clause clause) {
+ Clause.ComputedValueEvaluator evaluator = new Clause.ComputedValueEvaluator(session);
+ List<Violation> result = new ArrayList<>();
+ collections.forEach((coll, cd) -> {
+ evaluator.collName = coll;
+ evaluator.shardName = null;
+ if (clause.dataGrouping == Clause.DataGrouping.COLL) {
+ getViolations(cd.clauseValues, result, evaluator, clause);
+ } else if (clause.dataGrouping == Clause.DataGrouping.SHARD) {
+ cd.shards.forEach((shard, sd) -> {
+ evaluator.shardName = shard;
+ getViolations(sd.values, result, evaluator, clause);
+ });
+ }
+ });
+ return result;
+ }}
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 39237dd..88ec1ca 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
@@ -521,6 +521,7 @@ public class Policy implements MapWriter {
final Policy policy;
List<Clause> expandedClauses;
List<Violation> violations = new ArrayList<>();
+ PerClauseData perClauseData = new PerClauseData();
Transaction transaction;
@@ -554,7 +555,7 @@ public class Policy implements MapWriter {
if (!withCollMap.isEmpty()) {
Clause withCollClause = new Clause((Map<String,Object>)Utils.fromJSONString("{withCollection:'*' , node: '#ANY'}") ,
new Condition(NODE.tagName, "#ANY", Operand.EQUAL, null, null),
- new Condition(WITH_COLLECTION.tagName,"*" , Operand.EQUAL, null, null), true, null, false
+ new Condition(WITH_COLLECTION.tagName,"*" , Operand.EQUAL, null, null), true, null, false, Clause.DataGrouping.NODE
);
expandedClauses.add(withCollClause);
}
@@ -575,7 +576,7 @@ public class Policy implements MapWriter {
private Session(List<String> nodes, SolrCloudManager cloudManager,
List<Row> matrix, List<Clause> expandedClauses, int znodeVersion,
- NodeStateProvider nodeStateProvider, Policy policy, Transaction transaction) {
+ NodeStateProvider nodeStateProvider, Policy policy, Transaction transaction, PerClauseData perClauseData) {
this.transaction = transaction;
this.policy = policy;
this.nodes = nodes;
@@ -584,6 +585,7 @@ public class Policy implements MapWriter {
this.expandedClauses = expandedClauses;
this.znodeVersion = znodeVersion;
this.nodeStateProvider = nodeStateProvider;
+ this.perClauseData = perClauseData;
for (Row row : matrix) row.session = this;
}
@@ -604,7 +606,7 @@ public class Policy implements MapWriter {
}
Session copy() {
- return new Session(nodes, cloudManager, getMatrixCopy(), expandedClauses, znodeVersion, nodeStateProvider, policy, transaction);
+ return new Session(nodes, cloudManager, getMatrixCopy(), expandedClauses, znodeVersion, nodeStateProvider, policy, transaction, perClauseData.copy());
}
public Row getNode(String node) {
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaCount.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaCount.java
index 8f39b64..caa4ad5 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaCount.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaCount.java
@@ -79,6 +79,23 @@ class ReplicaCount implements MapWriter {
tlog += count.tlog;
}
+ void incr(ReplicaInfo ri, int delta){
+ switch (ri.getType()) {
+ case NRT:
+ nrt+=delta;
+ break;
+ case PULL:
+ pull+=delta;
+ break;
+ case TLOG:
+ tlog+=delta;
+ break;
+ default:
+ nrt+=delta;
+ }
+
+ }
+
public void increment(Replica.Type type) {
switch (type) {
@@ -125,4 +142,8 @@ class ReplicaCount implements MapWriter {
if (type == Replica.Type.TLOG) return (int) (tlog - expectedReplicaCount);
throw new RuntimeException("NO type");
}
+
+ public void decrement(ReplicaInfo ri) {
+
+ }
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaVariable.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaVariable.java
index 675382a..4f82e92 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaVariable.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaVariable.java
@@ -21,6 +21,7 @@ import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.function.BiConsumer;
import org.apache.solr.common.util.StrUtils;
@@ -33,11 +34,26 @@ class ReplicaVariable extends VariableBase {
public static final String REPLICASCOUNT = "relevantReplicas";
+ static int getRelevantReplicasCount(Policy.Session session, Condition cv, String collection, String shard) {
+ PerClauseData.CollectionDetails cd = session.perClauseData.collections.get(collection);
+ if (cd != null) {
+ if (shard != null) {
+ PerClauseData.ShardDetails sd = cd.shards.get(shard);
+ if (sd != null) return sd.replicas.getVal(cv.clause.type).intValue();
+ } else {
+ int totalReplicasOfInterest = 0;
- static int getRelevantReplicasCount(Policy.Session session, Condition cv, String collection, String shard) {
- int totalReplicasOfInterest = 0;
- Clause clause = cv.getClause();
+ for (PerClauseData.ShardDetails sd : cd.shards.values()) {
+ totalReplicasOfInterest += sd.replicas.getVal(cv.clause.type).intValue();
+ }
+ return totalReplicasOfInterest;
+ }
+
+ }
+ return 0;
+
+ /* Clause clause = cv.getClause();
for (Row row : session.matrix) {
Integer perShardCount = row.computeCacheIfAbsent(collection, shard, REPLICASCOUNT, cv.clause, o -> {
int[] result = new int[1];
@@ -50,7 +66,7 @@ class ReplicaVariable extends VariableBase {
if (perShardCount != null)
totalReplicasOfInterest += perShardCount;
}
- return totalReplicasOfInterest;
+ return totalReplicasOfInterest;*/
}
@Override
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java
index 2cc48ea..9c69c02 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Row.java
@@ -78,6 +78,7 @@ public class Row implements MapWriter {
this.globalCache = new HashMap();
this.perCollCache = new HashMap();
isAlreadyCopied = true;
+ initPerClauseData();
}
@@ -246,6 +247,8 @@ public class Row implements MapWriter {
}
}
+ modifyPerClauseCount(ri, 1);
+
return row;
}
@@ -343,6 +346,7 @@ public class Row implements MapWriter {
for (Cell cell : row.cells) {
cell.type.projectRemoveReplica(cell, removed, opCollector);
}
+ modifyPerClauseCount(removed, -1);
return row;
}
@@ -376,4 +380,39 @@ public class Row implements MapWriter {
}
}));
}
+
+ void modifyPerClauseCount(ReplicaInfo ri, int delta) {
+ if (session == null || session.perClauseData == null || ri == null) return;
+ session.perClauseData.getShardDetails(ri.getCollection(),ri.getShard()).replicas.incr(ri, delta);
+ for (Clause clause : session.expandedClauses) {
+ if (clause.put == Clause.Put.ON_EACH) continue;
+ if (clause.dataGrouping == Clause.DataGrouping.SHARD || clause.dataGrouping == Clause.DataGrouping.COLL) {
+ if (clause.tag.isPass(this)) {
+ session.perClauseData.getClauseValue(
+ ri.getCollection(),
+ ri.getShard(),
+ clause,
+ String.valueOf(this.getVal(clause.tag.name))).incr(ri, delta);
+ }
+ }
+ }
+ }
+
+ void initPerClauseData() {
+ if(session== null || session.perClauseData == null) return;
+ forEachReplica(it -> session.perClauseData.getShardDetails(it.getCollection(), it.getShard()).replicas.increment(it.getType()));
+ for (Clause clause : session.expandedClauses) {
+ if(clause.put == Clause.Put.ON_EACH) continue;
+ if(clause.dataGrouping == Clause.DataGrouping.SHARD || clause.dataGrouping == Clause.DataGrouping.COLL) {
+ if(clause.tag.isPass(this)) {
+ forEachReplica(it -> session.perClauseData.getClauseValue(
+ it.getCollection(),
+ it.getShard(),
+ clause, String.valueOf(this.getVal(clause.tag.name))
+
+ ).incr(it, 1));
+ }
+ }
+ }
+ }
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Violation.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Violation.java
index e0d2048..64efc2c 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Violation.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/Violation.java
@@ -19,6 +19,7 @@ package org.apache.solr.client.solrj.cloud.autoscaling;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
@@ -36,7 +37,7 @@ public class Violation implements MapWriter {
final Object tagKey;
private final int hash;
private final Clause clause;
- private List<ReplicaInfoAndErr> replicaInfoAndErrs = new ArrayList<>();
+ protected List<ReplicaInfoAndErr> replicaInfoAndErrs = new ArrayList<>();
Violation(SealedClause clause, String coll, String shard, String node, Object actualVal, Double replicaCountDelta, Object tagKey) {
this.clause = clause;
@@ -55,7 +56,7 @@ public class Violation implements MapWriter {
}
public List<ReplicaInfoAndErr> getViolatingReplicas() {
- return replicaInfoAndErrs;
+ return replicaInfoAndErrs == null ? Collections.EMPTY_LIST : replicaInfoAndErrs;
}
public Clause getClause() {
@@ -108,7 +109,7 @@ public class Violation implements MapWriter {
return false;
}
- static class ReplicaInfoAndErr implements MapWriter{
+ static class ReplicaInfoAndErr implements MapWriter {
final ReplicaInfo replicaInfo;
Double delta;
@@ -145,7 +146,7 @@ public class Violation implements MapWriter {
ew1.putIfNotNull("delta", replicaCountDelta);
});
ew.put("clause", getClause());
- if (!replicaInfoAndErrs.isEmpty()) {
+ if (!getViolatingReplicas().isEmpty()) {
ew.put("violatingReplicas", (IteratorWriter) iw -> {
for (ReplicaInfoAndErr replicaInfoAndErr : replicaInfoAndErrs) {
iw.add(replicaInfoAndErr.replicaInfo);
diff --git a/solr/solrj/src/test-files/solrj/solr/autoscaling/emptydiagnostics.json b/solr/solrj/src/test-files/solrj/solr/autoscaling/emptydiagnostics.json
new file mode 100644
index 0000000..f284054
--- /dev/null
+++ b/solr/solrj/src/test-files/solrj/solr/autoscaling/emptydiagnostics.json
@@ -0,0 +1,20 @@
+{
+ "diagnostics": {
+ "sortedNodes": [
+ {
+ "node": "10.0.0.80:8983_solr",
+ "isLive": true,
+ "cores" : 0,
+ "freedisk": 680,
+ "totaldisk": 1037,
+ "replicas": {}
+ }
+ ],
+ "cluster": {
+ "collections": {},
+ "live_nodes": [
+
+ ]
+ }
+ }
+}
diff --git a/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy2.java b/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy2.java
index 63b7da4..d4b69b1 100644
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy2.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy2.java
@@ -36,12 +36,15 @@ import java.util.stream.Collectors;
import com.google.common.collect.ImmutableSet;
import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.cloud.NodeStateProvider;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
import org.apache.solr.client.solrj.impl.ClusterStateProvider;
import org.apache.solr.client.solrj.impl.SolrClientNodeStateProvider;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.ReplicaPosition;
+import org.apache.solr.common.params.CollectionParams;
+import org.apache.solr.common.util.Pair;
import org.apache.solr.common.util.Utils;
import org.junit.Ignore;
import org.slf4j.Logger;
@@ -502,4 +505,68 @@ public class TestPolicy2 extends SolrTestCaseJ4 {
}
}
+ public void test1000Nodes() throws Exception {
+ String collName = "go";
+ int shards = 300;
+ int repFactor = 4;
+ int numNodes = 1000;
+ long start = System.currentTimeMillis();
+
+ Map<String, Object> m = (Map<String, Object>) loadFromResource("emptydiagnostics.json");
+ List<Map> nodes = (List<Map>) Utils.getObjectByPath(m, false, "diagnostics/sortedNodes");
+ List<String> liveNodes = (List<String>) Utils.getObjectByPath(m, false, "diagnostics/cluster/live_nodes");
+ Map aNode = nodes.remove(0);
+ String nodeName = (String) aNode.get("node");
+ for (int i = 0; i < numNodes; i++) {
+ Map copy = Utils.getDeepCopy(aNode, 2);
+ String nodeNameTmp = nodeName + "_" + i;
+ copy.put("node", nodeNameTmp);
+ nodes.add(copy);
+ liveNodes.add(nodeNameTmp);
+
+ }
+
+ SolrCloudManager cloudManagerFromDiagnostics = createCloudManagerFromDiagnostics(m);
+ AutoScalingConfig autoScalingConfig = new AutoScalingConfig((Map<String, Object>) Utils.fromJSONString("{\n" +
+ " 'cluster-preferences': [\n" +
+ " {\n" +
+ " 'minimize': 'cores',\n" +
+ " 'precision': 5\n" +
+ " },\n" +
+ " {\n" +
+ " 'maximize': 'freedisk',\n" +
+ " 'precision': 10\n" +
+ " },\n" +
+ " \n" +
+ " ],\n" +
+ " 'cluster-policy': [\n" +
+ " {\n" +
+ " 'replica': '<2',\n" +
+ " 'shard': '#EACH',\n" +
+ " 'node': '#ANY',\n" +
+ " },\n" +
+ " {\n" +
+ " 'replica': '#EQUAL',\n" +
+ " 'node': '#ANY',\n" +
+ " }\n" +
+ " ]\n" +
+ "}"
+ ));
+
+
+ Policy.Session session = autoScalingConfig.getPolicy().createSession(cloudManagerFromDiagnostics);
+
+ for (int i = 0; i < shards; i++) {
+ for (int j = 0; j < repFactor; j++) {
+ Suggester suggester = session.getSuggester(CollectionParams.CollectionAction.ADDREPLICA);
+ suggester.hint(Suggester.Hint.COLL_SHARD, new Pair<>(collName, "shards_" + i));
+ SolrRequest op = suggester.getSuggestion();
+ assertNotNull(op);
+ }
+ }
+
+ System.out.println("Time taken : "+ (System.currentTimeMillis() -start));
+ }
+
+
}