You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ab...@apache.org on 2019/05/07 21:58:42 UTC
[lucene-solr] 01/05: SOLR-13440: Initial patch.
This is an automated email from the ASF dual-hosted git repository.
ab pushed a commit to branch jira/solr-13440
in repository https://gitbox.apache.org/repos/asf/lucene-solr.git
commit 392e36b9ca8f52c21565ca0309246e4824113c78
Author: Andrzej Bialecki <ab...@apache.org>
AuthorDate: Fri May 3 00:27:00 2019 +0200
SOLR-13440: Initial patch.
---
.../sim/NoopDistributedQueueFactory.java | 94 +++++++++++++
.../autoscaling/sim/SnapshotCloudManager.java | 131 ++++++++++++++++++
.../sim/SnapshotClusterStateProvider.java | 130 ++++++++++++++++++
.../sim/SnapshotDistribStateManager.java | 146 +++++++++++++++++++++
.../autoscaling/sim/SnapshotNodeStateProvider.java | 131 ++++++++++++++++++
.../src/java/org/apache/solr/util/SolrCLI.java | 16 +--
.../autoscaling/sim/TestSnapshotCloudManager.java | 122 +++++++++++++++++
.../solrj/cloud/autoscaling/ReplicaInfo.java | 26 +++-
.../solrj/cloud/autoscaling/VersionedData.java | 20 ++-
.../org/apache/solr/common/util/TimeSource.java | 6 +-
10 files changed, 803 insertions(+), 19 deletions(-)
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/NoopDistributedQueueFactory.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/NoopDistributedQueueFactory.java
new file mode 100644
index 0000000..b04d38e
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/NoopDistributedQueueFactory.java
@@ -0,0 +1,94 @@
+/*
+ * 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.cloud.autoscaling.sim;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Predicate;
+
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.DistributedQueueFactory;
+import org.apache.solr.common.util.Pair;
+
+/**
+ * A queue factory implementation that does nothing.
+ */
+public class NoopDistributedQueueFactory implements DistributedQueueFactory {
+
+ public static final DistributedQueueFactory INSTANCE = new NoopDistributedQueueFactory();
+
+ @Override
+ public DistributedQueue makeQueue(String path) throws IOException {
+ return NoopDistributedQueue.INSTANCE;
+ }
+
+ @Override
+ public void removeQueue(String path) throws IOException {
+
+ }
+
+ private static final class NoopDistributedQueue implements DistributedQueue {
+ static final DistributedQueue INSTANCE = new NoopDistributedQueue();
+
+ @Override
+ public byte[] peek() throws Exception {
+ return new byte[0];
+ }
+
+ @Override
+ public byte[] peek(boolean block) throws Exception {
+ return new byte[0];
+ }
+
+ @Override
+ public byte[] peek(long wait) throws Exception {
+ return new byte[0];
+ }
+
+ @Override
+ public byte[] poll() throws Exception {
+ return new byte[0];
+ }
+
+ @Override
+ public byte[] remove() throws Exception {
+ return new byte[0];
+ }
+
+ @Override
+ public byte[] take() throws Exception {
+ return new byte[0];
+ }
+
+ @Override
+ public void offer(byte[] data) throws Exception {
+
+ }
+
+ @Override
+ public Map<String, Object> getStats() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public Collection<Pair<String, byte[]>> peekElements(int max, long waitMillis, Predicate<String> acceptFilter) throws Exception {
+ return Collections.emptyList();
+ }
+ }
+}
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotCloudManager.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotCloudManager.java
new file mode 100644
index 0000000..cf3b44b
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotCloudManager.java
@@ -0,0 +1,131 @@
+/*
+ * 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.cloud.autoscaling.sim;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.SolrResponse;
+import org.apache.solr.client.solrj.cloud.DistribStateManager;
+import org.apache.solr.client.solrj.cloud.DistributedQueueFactory;
+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.common.util.ObjectCache;
+import org.apache.solr.common.util.TimeSource;
+
+/**
+ * Read-only snapshot of another {@link SolrCloudManager}.
+ */
+public class SnapshotCloudManager implements SolrCloudManager {
+ private ObjectCache objectCache = new ObjectCache();
+ private SnapshotClusterStateProvider clusterStateProvider;
+ private SnapshotNodeStateProvider nodeStateProvider;
+ private SnapshotDistribStateManager distribStateManager;
+ private TimeSource timeSource;
+
+ public SnapshotCloudManager(SolrCloudManager other) throws Exception {
+ this.timeSource = other.getTimeSource();
+ this.clusterStateProvider = new SnapshotClusterStateProvider(other.getClusterStateProvider());
+ this.nodeStateProvider = new SnapshotNodeStateProvider(other);
+ this.distribStateManager = new SnapshotDistribStateManager(other.getDistribStateManager());
+ }
+
+ public SnapshotCloudManager(Map<String, Object> snapshot) {
+ Objects.requireNonNull(snapshot);
+ init(
+ (Map<String, Object>)snapshot.getOrDefault("managerState", Collections.emptyMap()),
+ (Map<String, Object>)snapshot.getOrDefault("clusterState", Collections.emptyMap()),
+ (Map<String, Object>)snapshot.getOrDefault("nodeState", Collections.emptyMap()),
+ (Map<String, Object>)snapshot.getOrDefault("distribState", Collections.emptyMap())
+ );
+ }
+
+ private void init(Map<String, Object> managerState, Map<String, Object> clusterState, Map<String, Object> nodeState, Map<String, Object> distribState) {
+ Objects.requireNonNull(managerState);
+ Objects.requireNonNull(clusterState);
+ Objects.requireNonNull(nodeState);
+ Objects.requireNonNull(distribState);
+ this.timeSource = TimeSource.get((String)managerState.getOrDefault("timeSource", "simTime:50"));
+ this.clusterStateProvider = new SnapshotClusterStateProvider(clusterState);
+ this.nodeStateProvider = new SnapshotNodeStateProvider(nodeState);
+ this.distribStateManager = new SnapshotDistribStateManager(distribState);
+ }
+
+ public Map<String, Object> getSnapshot() {
+ Map<String, Object> snapshot = new LinkedHashMap<>(4);
+ Map<String, Object> managerState = new HashMap<>();
+ managerState.put("timeSource", timeSource.toString());
+ snapshot.put("managerState", managerState);
+
+ snapshot.put("clusterState", clusterStateProvider.getSnapshot());
+ snapshot.put("nodeState", nodeStateProvider.getSnapshot());
+ snapshot.put("distribState", distribStateManager.getSnapshot());
+ return snapshot;
+ }
+
+ @Override
+ public ClusterStateProvider getClusterStateProvider() {
+ return clusterStateProvider;
+ }
+
+ @Override
+ public NodeStateProvider getNodeStateProvider() {
+ return nodeStateProvider;
+ }
+
+ @Override
+ public DistribStateManager getDistribStateManager() {
+ return distribStateManager;
+ }
+
+ @Override
+ public DistributedQueueFactory getDistributedQueueFactory() {
+ return NoopDistributedQueueFactory.INSTANCE;
+ }
+
+ @Override
+ public ObjectCache getObjectCache() {
+ return objectCache;
+ }
+
+ @Override
+ public TimeSource getTimeSource() {
+ return timeSource;
+ }
+
+ @Override
+ public SolrResponse request(SolrRequest req) throws IOException {
+ throw new UnsupportedOperationException("request");
+ }
+
+ @Override
+ public byte[] httpRequest(String url, SolrRequest.METHOD method, Map<String, String> headers, String payload, int timeout, boolean followRedirects) throws IOException {
+ throw new UnsupportedOperationException("httpRequest");
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+}
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotClusterStateProvider.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotClusterStateProvider.java
new file mode 100644
index 0000000..7589268
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotClusterStateProvider.java
@@ -0,0 +1,130 @@
+/*
+ * 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.cloud.autoscaling.sim;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.apache.solr.client.solrj.impl.ClusterStateProvider;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.util.Utils;
+import org.noggit.CharArr;
+import org.noggit.JSONWriter;
+
+/**
+ * Read-only snapshot of another {@link ClusterStateProvider}.
+ */
+public class SnapshotClusterStateProvider implements ClusterStateProvider {
+
+ final Set<String> liveNodes;
+ final ClusterState clusterState;
+ final Map<String, Object> clusterProperties;
+
+ public SnapshotClusterStateProvider(ClusterStateProvider other) throws Exception {
+ liveNodes = Collections.unmodifiableSet(new HashSet<>(other.getLiveNodes()));
+ ClusterState otherState = other.getClusterState();
+ clusterState = new ClusterState(otherState.getZNodeVersion(), liveNodes, otherState.getCollectionsMap());
+ clusterProperties = new HashMap<>(other.getClusterProperties());
+ }
+
+ public SnapshotClusterStateProvider(Map<String, Object> snapshot) {
+ Objects.requireNonNull(snapshot);
+ liveNodes = Collections.unmodifiableSet((Set<String>)snapshot.getOrDefault("liveNodes", Collections.emptySet()));
+ clusterProperties = (Map<String, Object>)snapshot.getOrDefault("clusterProperties", Collections.emptyMap());
+ Map<String, Object> stateMap = new HashMap<>((Map<String, Object>)snapshot.getOrDefault("clusterState", Collections.emptyMap()));
+ Integer version = (Integer)stateMap.remove("version");
+ clusterState = ClusterState.load(version, stateMap, liveNodes, ZkStateReader.CLUSTER_STATE);
+ }
+
+ public Map<String, Object> getSnapshot() {
+ Map<String, Object> snapshot = new HashMap<>();
+ snapshot.put("liveNodes", liveNodes);
+ if (clusterProperties != null) {
+ snapshot.put("clusterProperties", clusterProperties);
+ }
+ Map<String, Object> stateMap = new HashMap<>();
+ snapshot.put("clusterState", stateMap);
+ stateMap.put("version", clusterState.getZNodeVersion());
+ clusterState.forEachCollection(coll -> {
+ CharArr out = new CharArr();
+ JSONWriter writer = new JSONWriter(out, 2);
+ coll.write(writer);
+ String json = out.toString();
+ try {
+ stateMap.put(coll.getName(), Utils.fromJSON(json.getBytes("UTF-8")));
+ } catch (UnsupportedEncodingException e) {
+ throw new RuntimeException("should not happen!", e);
+ }
+ });
+ return snapshot;
+ }
+
+ @Override
+ public ClusterState.CollectionRef getState(String collection) {
+ return clusterState.getCollectionRef(collection);
+ }
+
+ @Override
+ public Set<String> getLiveNodes() {
+ return liveNodes;
+ }
+
+ @Override
+ public List<String> resolveAlias(String alias) {
+ throw new UnsupportedOperationException("resolveAlias");
+ }
+
+ @Override
+ public Map<String, String> getAliasProperties(String alias) {
+ throw new UnsupportedOperationException("getAliasProperties");
+ }
+
+ @Override
+ public ClusterState getClusterState() throws IOException {
+ return clusterState;
+ }
+
+ @Override
+ public Map<String, Object> getClusterProperties() {
+ return clusterProperties;
+ }
+
+ @Override
+ public String getPolicyNameByCollection(String coll) {
+ DocCollection collection = clusterState.getCollectionOrNull(coll);
+ return collection == null ? null : (String)collection.getProperties().get("policy");
+ }
+
+ @Override
+ public void connect() {
+
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+}
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotDistribStateManager.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotDistribStateManager.java
new file mode 100644
index 0000000..4a7fde3
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotDistribStateManager.java
@@ -0,0 +1,146 @@
+package org.apache.solr.cloud.autoscaling.sim;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.stream.Collectors;
+
+import org.apache.solr.client.solrj.cloud.DistribStateManager;
+import org.apache.solr.client.solrj.cloud.autoscaling.AlreadyExistsException;
+import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
+import org.apache.solr.client.solrj.cloud.autoscaling.BadVersionException;
+import org.apache.solr.client.solrj.cloud.autoscaling.NotEmptyException;
+import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.AutoScalingParams;
+import org.apache.solr.common.util.Base64;
+import org.apache.solr.common.util.Utils;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.OpResult;
+import org.apache.zookeeper.Watcher;
+
+/**
+ * Read-only snapshot of another {@link DistribStateManager}
+ */
+public class SnapshotDistribStateManager implements DistribStateManager {
+
+ LinkedHashMap<String, VersionedData> dataMap = new LinkedHashMap<>();
+
+ public SnapshotDistribStateManager(DistribStateManager other) throws Exception {
+ List<String> tree = other.listTree("/");
+ for (String path : tree) {
+ dataMap.put(path, other.getData(path));
+ }
+ }
+
+ public SnapshotDistribStateManager(Map<String, Object> snapshot) {
+ snapshot.forEach((path, value) -> {
+ Map<String, Object> map = (Map<String, Object>)value;
+ int version = (Integer)map.getOrDefault("version", 0);
+ String owner = (String)map.get("owner");
+ String modeStr = (String)map.getOrDefault("mode", CreateMode.PERSISTENT.toString());
+ CreateMode mode = CreateMode.valueOf(modeStr);
+ byte[] bytes = null;
+ if (map.containsKey("data")) {
+ bytes = Base64.base64ToByteArray((String)map.get("data"));
+ }
+ dataMap.put(path, new VersionedData(version, bytes, mode, owner));
+ });
+ }
+
+ public Map<String, Object> getSnapshot() {
+ Map<String, Object> snapshot = new LinkedHashMap<>();
+ dataMap.forEach((path, vd) -> {
+ Map<String, Object> data = new HashMap<>();
+ vd.toMap(data);
+ snapshot.put(path, data);
+ });
+ return snapshot;
+ }
+
+ @Override
+ public boolean hasData(String path) throws IOException, KeeperException, InterruptedException {
+ return dataMap.containsKey(path);
+ }
+
+ @Override
+ public List<String> listData(String path) throws NoSuchElementException, IOException, KeeperException, InterruptedException {
+ return listData(path, null);
+ }
+
+ @Override
+ public List<String> listData(String path, Watcher watcher) throws NoSuchElementException, IOException, KeeperException, InterruptedException {
+ final String prefix = path + "/";
+ return dataMap.entrySet().stream()
+ .filter(e -> e.getKey().startsWith(prefix))
+ .map(e -> {
+ String suffix = e.getKey().substring(prefix.length());
+ int idx = suffix.indexOf('/');
+ if (idx == -1) {
+ return suffix;
+ } else {
+ return suffix.substring(0, idx);
+ }
+ })
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public VersionedData getData(String path, Watcher watcher) throws NoSuchElementException, IOException, KeeperException, InterruptedException {
+ if (!dataMap.containsKey(path)) {
+ throw new NoSuchElementException(path);
+ }
+ return dataMap.get(path);
+ }
+
+ @Override
+ public void makePath(String path) throws AlreadyExistsException, IOException, KeeperException, InterruptedException {
+ throw new UnsupportedOperationException("makePath");
+ }
+
+ @Override
+ public void makePath(String path, byte[] data, CreateMode createMode, boolean failOnExists) throws AlreadyExistsException, IOException, KeeperException, InterruptedException {
+ throw new UnsupportedOperationException("makePath");
+ }
+
+ @Override
+ public String createData(String path, byte[] data, CreateMode mode) throws AlreadyExistsException, IOException, KeeperException, InterruptedException {
+ throw new UnsupportedOperationException("createData");
+ }
+
+ @Override
+ public void removeData(String path, int version) throws NoSuchElementException, IOException, NotEmptyException, KeeperException, InterruptedException, BadVersionException {
+ throw new UnsupportedOperationException("removeData");
+ }
+
+ @Override
+ public void setData(String path, byte[] data, int version) throws BadVersionException, NoSuchElementException, IOException, KeeperException, InterruptedException {
+ throw new UnsupportedOperationException("setData");
+ }
+
+ @Override
+ public List<OpResult> multi(Iterable<Op> ops) throws BadVersionException, NoSuchElementException, AlreadyExistsException, IOException, KeeperException, InterruptedException {
+ throw new UnsupportedOperationException("multi");
+ }
+
+ @Override
+ public AutoScalingConfig getAutoScalingConfig(Watcher watcher) throws InterruptedException, IOException {
+ VersionedData vd = dataMap.get(ZkStateReader.SOLR_AUTOSCALING_CONF_PATH);
+ Map<String, Object> map = new HashMap<>();
+ if (vd != null && vd.getData() != null && vd.getData().length > 0) {
+ map = (Map<String, Object>) Utils.fromJSON(vd.getData());
+ map.put(AutoScalingParams.ZK_VERSION, vd.getVersion());
+ }
+ return new AutoScalingConfig(map);
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+}
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotNodeStateProvider.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotNodeStateProvider.java
new file mode 100644
index 0000000..90460db
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/sim/SnapshotNodeStateProvider.java
@@ -0,0 +1,131 @@
+/*
+ * 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.cloud.autoscaling.sim;
+
+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.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+import org.apache.solr.client.solrj.cloud.NodeStateProvider;
+import org.apache.solr.client.solrj.cloud.SolrCloudManager;
+import org.apache.solr.client.solrj.cloud.autoscaling.ReplicaInfo;
+import org.apache.solr.client.solrj.cloud.autoscaling.Variable;
+
+/**
+ * Read-only snapshot of another {@link NodeStateProvider}.
+ */
+public class SnapshotNodeStateProvider implements NodeStateProvider {
+ private Map<String, Map<String, Object>> nodeValues = new HashMap<>();
+ private Map<String, Map<String, Map<String, List<ReplicaInfo>>>> replicaInfos = new HashMap<>();
+
+ public static final List<String> REPLICA_TAGS = Arrays.asList(
+ Variable.Type.CORE_IDX.metricsAttribute,
+ "QUERY./select.requests",
+ "UPDATE./update.requests"
+ );
+ public static final Set<String> NODE_TAGS = SimCloudManager.createNodeValues("unused:1234_solr").keySet();
+
+
+ public SnapshotNodeStateProvider(SolrCloudManager other) {
+ for (String node : other.getClusterStateProvider().getLiveNodes()) {
+ nodeValues.put(node, other.getNodeStateProvider().getNodeValues(node, NODE_TAGS));
+ Map<String, Map<String, List<ReplicaInfo>>> infos = other.getNodeStateProvider().getReplicaInfo(node, REPLICA_TAGS);
+ infos.forEach((collection, shards) -> {
+ shards.forEach((shard, replicas) -> {
+ replicas.forEach(r -> {
+ List<ReplicaInfo> myReplicas = replicaInfos
+ .computeIfAbsent(node, n -> new HashMap<>())
+ .computeIfAbsent(collection, c -> new HashMap<>())
+ .computeIfAbsent(shard, s -> new ArrayList<>());
+ Map<String, Object> rMap = new HashMap<>();
+ r.toMap(rMap);
+ if (r.isLeader) { // ReplicaInfo.toMap doesn't write this!!!
+ ((Map<String, Object>)rMap.values().iterator().next()).put("leader", "true");
+ }
+ myReplicas.add(new ReplicaInfo(rMap));
+ });
+ });
+ });
+ }
+ }
+
+ public SnapshotNodeStateProvider(Map<String, Object> snapshot) {
+ Objects.requireNonNull(snapshot);
+ nodeValues = (Map<String, Map<String, Object>>)snapshot.getOrDefault("nodeValues", Collections.emptyMap());
+ ((Map<String, Object>)snapshot.getOrDefault("replicaInfos", Collections.emptyMap())).forEach((node, v) -> {
+ Map<String, Map<String, List<ReplicaInfo>>> perNode = replicaInfos.computeIfAbsent(node, n -> new HashMap<>());
+ ((Map<String, Object>)v).forEach((collection, shards) -> {
+ Map<String, List<ReplicaInfo>> perColl = perNode.computeIfAbsent(collection, c -> new HashMap<>());
+ ((Map<String, Object>)shards).forEach((shard, replicas) -> {
+ List<ReplicaInfo> infos = perColl.computeIfAbsent(shard, s -> new ArrayList<>());
+ ((List<Map<String, Object>>)replicas).forEach(replicaMap -> {
+ ReplicaInfo ri = new ReplicaInfo(new HashMap<>(replicaMap)); // constructor modifies this map
+ infos.add(ri);
+ });
+ });
+ });
+ });
+ }
+
+ public Map<String, Object> getSnapshot() {
+ Map<String, Object> snapshot = new HashMap<>();
+ snapshot.put("nodeValues", nodeValues);
+ Map<String, Map<String, Map<String, List<Map<String, Object>>>>> replicaInfosMap = new HashMap<>();
+ snapshot.put("replicaInfos", replicaInfosMap);
+ replicaInfos.forEach((node, perNode) -> {
+ perNode.forEach((collection, shards) -> {
+ shards.forEach((shard, replicas) -> {
+ replicas.forEach(r -> {
+ List<Map<String, Object>> myReplicas = replicaInfosMap
+ .computeIfAbsent(node, n -> new HashMap<>())
+ .computeIfAbsent(collection, c -> new HashMap<>())
+ .computeIfAbsent(shard, s -> new ArrayList<>());
+ Map<String, Object> rMap = new HashMap<>();
+ r.toMap(rMap);
+ if (r.isLeader) { // ReplicaInfo.toMap doesn't write this!!!
+ ((Map<String, Object>)rMap.values().iterator().next()).put("leader", "true");
+ }
+ myReplicas.add(rMap);
+ });
+ });
+ });
+ });
+ return snapshot;
+ }
+
+ @Override
+ public Map<String, Object> getNodeValues(String node, Collection<String> tags) {
+ return nodeValues.getOrDefault(node, Collections.emptyMap());
+ }
+
+ @Override
+ public Map<String, Map<String, List<ReplicaInfo>>> getReplicaInfo(String node, Collection<String> keys) {
+ return replicaInfos.getOrDefault(node, Collections.emptyMap());
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+}
diff --git a/solr/core/src/java/org/apache/solr/util/SolrCLI.java b/solr/core/src/java/org/apache/solr/util/SolrCLI.java
index b383ec8..0e448e9 100755
--- a/solr/core/src/java/org/apache/solr/util/SolrCLI.java
+++ b/solr/core/src/java/org/apache/solr/util/SolrCLI.java
@@ -99,8 +99,6 @@ import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrServerException;
-import org.apache.solr.client.solrj.cloud.DistributedQueue;
-import org.apache.solr.client.solrj.cloud.DistributedQueueFactory;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
@@ -118,6 +116,7 @@ import org.apache.solr.client.solrj.request.CollectionAdminRequest;
import org.apache.solr.client.solrj.request.ContentStreamUpdateRequest;
import org.apache.solr.client.solrj.response.CollectionAdminResponse;
import org.apache.solr.client.solrj.response.QueryResponse;
+import org.apache.solr.cloud.autoscaling.sim.NoopDistributedQueueFactory;
import org.apache.solr.cloud.autoscaling.sim.SimCloudManager;
import org.apache.solr.common.MapWriter;
import org.apache.solr.common.SolrException;
@@ -934,18 +933,7 @@ public class SolrCLI implements CLIO {
@Override
protected void runCloudTool(CloudSolrClient cloudSolrClient, CommandLine cli) throws Exception {
- DistributedQueueFactory dummmyFactory = new DistributedQueueFactory() {
- @Override
- public DistributedQueue makeQueue(String path) throws IOException {
- throw new UnsupportedOperationException("makeQueue");
- }
-
- @Override
- public void removeQueue(String path) throws IOException {
- throw new UnsupportedOperationException("removeQueue");
- }
- };
- try (SolrClientCloudManager realCloudManager = new SolrClientCloudManager(dummmyFactory, cloudSolrClient)) {
+ try (SolrClientCloudManager realCloudManager = new SolrClientCloudManager(NoopDistributedQueueFactory.INSTANCE, cloudSolrClient)) {
AutoScalingConfig config = null;
HashSet<String> liveNodes = new HashSet<>();
String configFile = cli.getOptionValue("a");
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSnapshotCloudManager.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSnapshotCloudManager.java
new file mode 100644
index 0000000..b0253ea
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSnapshotCloudManager.java
@@ -0,0 +1,122 @@
+package org.apache.solr.cloud.autoscaling.sim;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.apache.solr.client.solrj.cloud.DistribStateManager;
+import org.apache.solr.client.solrj.cloud.NodeStateProvider;
+import org.apache.solr.client.solrj.cloud.SolrCloudManager;
+import org.apache.solr.client.solrj.cloud.autoscaling.ReplicaInfo;
+import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.params.CollectionAdminParams;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ */
+public class TestSnapshotCloudManager extends SolrCloudTestCase {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static int NODE_COUNT = 3;
+
+ private static SolrCloudManager realManager;
+
+ // set up a real cluster as the source of test data
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ configureCluster(NODE_COUNT)
+ .addConfig("conf", configset("cloud-minimal"))
+ .configure();
+ CollectionAdminRequest.createCollection(CollectionAdminParams.SYSTEM_COLL, null, 1, 2, 0, 1)
+ .process(cluster.getSolrClient());
+ realManager = cluster.getJettySolrRunner(cluster.getJettySolrRunners().size() - 1).getCoreContainer()
+ .getZkController().getSolrCloudManager();
+ }
+
+ @Test
+ public void testSnapshots() throws Exception {
+ SnapshotCloudManager snapshotCloudManager = new SnapshotCloudManager(realManager);
+ Map<String, Object> snapshot = snapshotCloudManager.getSnapshot();
+ SnapshotCloudManager snapshotCloudManager1 = new SnapshotCloudManager(snapshot);
+ assertClusterStateEquals(realManager.getClusterStateProvider().getClusterState(), snapshotCloudManager.getClusterStateProvider().getClusterState());
+ assertClusterStateEquals(realManager.getClusterStateProvider().getClusterState(), snapshotCloudManager1.getClusterStateProvider().getClusterState());
+ // this will always fail because the metrics will be already different
+ // assertNodeStateProvider(realManager, snapshotCloudManager);
+ assertNodeStateProvider(snapshotCloudManager, snapshotCloudManager1);
+ assertDistribStateManager(snapshotCloudManager.getDistribStateManager(), snapshotCloudManager1.getDistribStateManager());
+ }
+
+ private static void assertNodeStateProvider(SolrCloudManager oneMgr, SolrCloudManager twoMgr) throws Exception {
+ NodeStateProvider one = oneMgr.getNodeStateProvider();
+ NodeStateProvider two = twoMgr.getNodeStateProvider();
+ for (String node : oneMgr.getClusterStateProvider().getLiveNodes()) {
+ Map<String, Object> oneVals = one.getNodeValues(node, SnapshotNodeStateProvider.NODE_TAGS);
+ Map<String, Object> twoVals = two.getNodeValues(node, SnapshotNodeStateProvider.NODE_TAGS);
+ assertEquals(oneVals, twoVals);
+ Map<String, Map<String, List<ReplicaInfo>>> oneInfos = one.getReplicaInfo(node, SnapshotNodeStateProvider.REPLICA_TAGS);
+ Map<String, Map<String, List<ReplicaInfo>>> twoInfos = two.getReplicaInfo(node, SnapshotNodeStateProvider.REPLICA_TAGS);
+ assertEquals(oneInfos, twoInfos);
+ }
+ }
+
+ private static void assertDistribStateManager(DistribStateManager one, DistribStateManager two) throws Exception {
+ List<String> treeOne = new ArrayList<>(one.listTree("/"));
+ List<String> treeTwo = new ArrayList<>(two.listTree("/"));
+ Collections.sort(treeOne);
+ Collections.sort(treeTwo);
+ assertEquals(treeOne, treeTwo);
+ for (String path : treeOne) {
+ VersionedData vd1 = one.getData(path);
+ VersionedData vd2 = two.getData(path);
+ assertEquals(vd1, vd2);
+ }
+ }
+
+ private static void assertClusterStateEquals(ClusterState one, ClusterState two) {
+ assertEquals(one.getLiveNodes(), two.getLiveNodes());
+ assertEquals(one.getCollectionsMap().keySet(), two.getCollectionsMap().keySet());
+ one.forEachCollection(oneColl -> {
+ DocCollection twoColl = two.getCollection(oneColl.getName());
+ Map<String, Slice> oneSlices = oneColl.getSlicesMap();
+ Map<String, Slice> twoSlices = twoColl.getSlicesMap();
+ assertEquals(oneSlices.keySet(), twoSlices.keySet());
+ oneSlices.forEach((s, slice) -> {
+ Slice sTwo = twoSlices.get(s);
+ for (Replica oneReplica : slice.getReplicas()) {
+ Replica twoReplica = sTwo.getReplica(oneReplica.getName());
+ assertNotNull(twoReplica);
+ assertReplicaEquals(oneReplica, twoReplica);
+ }
+ });
+ });
+ }
+
+ private static void assertReplicaEquals(Replica one, Replica two) {
+ assertEquals(one.getName(), two.getName());
+ assertEquals(one.getNodeName(), two.getNodeName());
+ assertEquals(one.getState(), two.getState());
+ assertEquals(one.getType(), two.getType());
+ Map<String, Object> filteredPropsOne = one.getProperties().entrySet().stream()
+ .filter(e -> !(e.getKey().startsWith("INDEX") || e.getKey().startsWith("QUERY") || e.getKey().startsWith("UPDATE")))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ Map<String, Object> filteredPropsTwo = two.getProperties().entrySet().stream()
+ .filter(e -> !(e.getKey().startsWith("INDEX") || e.getKey().startsWith("QUERY") || e.getKey().startsWith("UPDATE")))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ assertEquals(filteredPropsOne, filteredPropsTwo);
+ }
+
+}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaInfo.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaInfo.java
index 9c5cf31..19bd161 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaInfo.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ReplicaInfo.java
@@ -72,7 +72,7 @@ public class ReplicaInfo implements MapWriter {
this.node = node;
}
- ReplicaInfo(Map<String, Object> map) {
+ public ReplicaInfo(Map<String, Object> map) {
this.name = map.keySet().iterator().next();
Map details = (Map) map.get(name);
details = Utils.getDeepCopy(details, 4);
@@ -172,6 +172,30 @@ public class ReplicaInfo implements MapWriter {
}
@Override
+ public boolean equals(Object o) {
+ if (o == null) {
+ return false;
+ }
+ if (!(o instanceof ReplicaInfo)) {
+ return false;
+ }
+ ReplicaInfo other = (ReplicaInfo)o;
+ if (
+ name.equals(other.name) &&
+ collection.equals(other.collection) &&
+ core.equals(other.core) &&
+ isLeader == other.isLeader &&
+ node.equals(other.node) &&
+ shard.equals(other.shard) &&
+ type == other.type &&
+ variables.equals(other.variables)) {
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ @Override
public String toString() {
return Utils.toJSONString(this);
}
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/VersionedData.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/VersionedData.java
index e81d50b..692f041 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/VersionedData.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/VersionedData.java
@@ -16,12 +16,16 @@
*/
package org.apache.solr.client.solrj.cloud.autoscaling;
+import java.io.IOException;
+
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.util.Base64;
import org.apache.zookeeper.CreateMode;
/**
* Immutable representation of binary data with version.
*/
-public class VersionedData {
+public class VersionedData implements MapWriter {
private final int version;
private final byte[] data;
private final String owner;
@@ -56,4 +60,18 @@ public class VersionedData {
public String getOwner() {
return owner;
}
+
+ @Override
+ public void writeMap(EntryWriter ew) throws IOException {
+ ew.put("version", version);
+ if (owner != null) {
+ ew.put("owner", owner);
+ }
+ ew.put("mode", mode.toString());
+ if (data != null) {
+ ew.put("data", Base64.byteArrayToBase64(data));
+ }
+ }
+
+
}
diff --git a/solr/solrj/src/java/org/apache/solr/common/util/TimeSource.java b/solr/solrj/src/java/org/apache/solr/common/util/TimeSource.java
index ac8749d..2e5cd2a 100644
--- a/solr/solrj/src/java/org/apache/solr/common/util/TimeSource.java
+++ b/solr/solrj/src/java/org/apache/solr/common/util/TimeSource.java
@@ -179,11 +179,11 @@ public abstract class TimeSource {
public static TimeSource get(String type) {
if (type == null) {
return NANO_TIME;
- } else if (type.equals("currentTime")) {
+ } else if (type.equals("currentTime") || type.equals(CurrentTimeSource.class.getSimpleName())) {
return CURRENT_TIME;
- } else if (type.equals("nanoTime")) {
+ } else if (type.equals("nanoTime") || type.equals(NanoTimeSource.class.getSimpleName())) {
return NANO_TIME;
- } else if (type.startsWith("simTime")) {
+ } else if (type.startsWith("simTime") || type.startsWith(SimTimeSource.class.getSimpleName())) {
return simTimeSources.computeIfAbsent(type, t -> {
String[] parts = t.split(":");
double mul = 1.0;