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 2018/02/26 18:54:21 UTC
lucene-solr:jira/solr-11670: Initial patch.
Repository: lucene-solr
Updated Branches:
refs/heads/jira/solr-11670 [created] 760f7a395
Initial patch.
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/760f7a39
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/760f7a39
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/760f7a39
Branch: refs/heads/jira/solr-11670
Commit: 760f7a39529918cc0e2803d86316451431b64a0a
Parents: a2eb7f3
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Mon Feb 26 19:53:47 2018 +0100
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Mon Feb 26 19:53:47 2018 +0100
----------------------------------------------------------------------
.../org/apache/solr/cloud/MaintenanceTask.java | 35 +++++++
.../apache/solr/cloud/MaintenanceThread.java | 104 +++++++++++++++++++
.../java/org/apache/solr/cloud/Overseer.java | 28 +++++
.../cloud/api/collections/SplitShardCmd.java | 72 +++++++++++++
.../solr/cloud/overseer/SliceMutator.java | 1 +
.../cloud/autoscaling/sim/SimCloudManager.java | 1 +
.../autoscaling/AlreadyExistsException.java | 12 +--
.../apache/solr/common/cloud/ZkStateReader.java | 1 +
8 files changed, 248 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/760f7a39/solr/core/src/java/org/apache/solr/cloud/MaintenanceTask.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/MaintenanceTask.java b/solr/core/src/java/org/apache/solr/cloud/MaintenanceTask.java
new file mode 100644
index 0000000..5525c42
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/MaintenanceTask.java
@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ *
+ */
+public interface MaintenanceTask extends Runnable, Closeable {
+
+ default String getName() {
+ return getClass().getSimpleName();
+ }
+
+ default void close() throws IOException {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/760f7a39/solr/core/src/java/org/apache/solr/cloud/MaintenanceThread.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/MaintenanceThread.java b/solr/core/src/java/org/apache/solr/cloud/MaintenanceThread.java
new file mode 100644
index 0000000..216beca
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/MaintenanceThread.java
@@ -0,0 +1,104 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.solr.client.solrj.cloud.autoscaling.AlreadyExistsException;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
+import org.apache.solr.common.SolrCloseable;
+import org.apache.solr.common.util.IOUtils;
+import org.apache.solr.util.DefaultSolrThreadFactory;
+
+/**
+ * Responsible for running periodically some maintenance tasks.
+ */
+public class MaintenanceThread implements SolrCloseable {
+
+ private class Task {
+ MaintenanceTask maintenanceTask;
+ ScheduledFuture<?> scheduledFuture;
+ }
+
+ private final SolrCloudManager cloudManager;
+ private final ScheduledThreadPoolExecutor taskExecutor;
+ private final Map<String, Task> tasks = new ConcurrentHashMap<>();
+
+ private transient boolean isClosed = false;
+
+ public MaintenanceThread(SolrCloudManager cloudManager) {
+ this.cloudManager = cloudManager;
+ this.taskExecutor = (ScheduledThreadPoolExecutor) Executors.newScheduledThreadPool(3, new DefaultSolrThreadFactory("MaintenanceTask"));
+ }
+
+ public void addTask(MaintenanceTask task, int period, TimeUnit timeUnit, boolean replace) throws AlreadyExistsException {
+ if (tasks.containsKey(task.getName())) {
+ if (replace) {
+ Task oldTask = tasks.get(task.getName());
+ IOUtils.closeQuietly(oldTask.maintenanceTask);
+ if (oldTask.scheduledFuture != null) {
+ oldTask.scheduledFuture.cancel(false);
+ }
+ } else {
+ throw new AlreadyExistsException(task.getName());
+ }
+ }
+ Task t = new Task();
+ t.maintenanceTask = task;
+ tasks.put(task.getName(), t);
+ t.scheduledFuture = taskExecutor.scheduleWithFixedDelay(task, 0,
+ cloudManager.getTimeSource().convertDelay(timeUnit, period, TimeUnit.MILLISECONDS),
+ TimeUnit.MILLISECONDS);
+ }
+
+ public void removeTask(String name) {
+ Task t = tasks.get(name);
+ if (t == null) {
+ // ignore
+ return;
+ }
+ IOUtils.closeQuietly(t.maintenanceTask);
+ if (t.scheduledFuture != null) {
+ t.scheduledFuture.cancel(false);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ isClosed = true;
+ for (Task task : tasks.values()) {
+ IOUtils.closeQuietly(task.maintenanceTask);
+ if (task.scheduledFuture != null) {
+ task.scheduledFuture.cancel(false);
+ }
+ }
+ tasks.clear();
+ taskExecutor.shutdown();
+ }
+
+ @Override
+ public boolean isClosed() {
+ return isClosed;
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/760f7a39/solr/core/src/java/org/apache/solr/cloud/Overseer.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/Overseer.java b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
index 608fb20..dd03a99 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Overseer.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
@@ -484,6 +484,8 @@ public class Overseer implements SolrCloseable {
private OverseerThread triggerThread;
+ private OverseerThread maintenanceThread;
+
private final ZkStateReader reader;
private final ShardHandler shardHandler;
@@ -527,6 +529,11 @@ public class Overseer implements SolrCloseable {
updaterThread = new OverseerThread(tg, new ClusterStateUpdater(reader, id, stats), "OverseerStateUpdate-" + id);
updaterThread.setDaemon(true);
+ // create this early so that collection commands may register their tasks
+ ThreadGroup maintenanceThreadGroup = new ThreadGroup("Overseer maintenance process");
+ MaintenanceThread maintenance = new MaintenanceThread(zkController.getSolrCloudManager());
+ maintenanceThread = new OverseerThread(maintenanceThreadGroup, maintenance, "OverseerMaintenanceThread-" + id);
+
ThreadGroup ccTg = new ThreadGroup("Overseer collection creation process.");
OverseerNodePrioritizer overseerPrioritizer = new OverseerNodePrioritizer(reader, adminPath, shardHandler.getShardHandlerFactory());
@@ -542,6 +549,7 @@ public class Overseer implements SolrCloseable {
updaterThread.start();
ccThread.start();
triggerThread.start();
+ maintenanceThread.start();
assert ObjectReleaseTracker.track(this);
}
@@ -579,6 +587,16 @@ public class Overseer implements SolrCloseable {
public synchronized OverseerThread getTriggerThread() {
return triggerThread;
}
+
+ /**
+ * For tests.
+ * @lucene.internal
+ * @return trigger thread
+ */
+ public synchronized OverseerThread getMaintenanceThread() {
+ return maintenanceThread;
+ }
+
public synchronized void close() {
if (closed) return;
@@ -608,6 +626,10 @@ public class Overseer implements SolrCloseable {
IOUtils.closeQuietly(triggerThread);
triggerThread.interrupt();
}
+ if (maintenanceThread != null) {
+ IOUtils.closeQuietly(maintenanceThread);
+ maintenanceThread.interrupt();
+ }
if (updaterThread != null) {
try {
updaterThread.join();
@@ -623,9 +645,15 @@ public class Overseer implements SolrCloseable {
triggerThread.join();
} catch (InterruptedException e) {}
}
+ if (maintenanceThread != null) {
+ try {
+ maintenanceThread.join();
+ } catch (InterruptedException e) {}
+ }
updaterThread = null;
ccThread = null;
triggerThread = null;
+ maintenanceThread = null;
}
/**
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/760f7a39/solr/core/src/java/org/apache/solr/cloud/api/collections/SplitShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/SplitShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/SplitShardCmd.java
index 03e7430..3c3ae4f 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/SplitShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/SplitShardCmd.java
@@ -18,6 +18,7 @@
package org.apache.solr.cloud.api.collections;
+import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collection;
@@ -26,12 +27,16 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.solr.client.solrj.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.autoscaling.AlreadyExistsException;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
import org.apache.solr.client.solrj.request.CoreAdminRequest;
+import org.apache.solr.cloud.MaintenanceTask;
+import org.apache.solr.cloud.MaintenanceThread;
import org.apache.solr.cloud.Overseer;
import org.apache.solr.cloud.overseer.OverseerAction;
import org.apache.solr.common.SolrException;
@@ -69,8 +74,24 @@ public class SplitShardCmd implements OverseerCollectionMessageHandler.Cmd {
private final OverseerCollectionMessageHandler ocmh;
+ public static final String CLEANUP_PERIOD_PROP = "split.shard.cleanup.period";
+ public static final String CLEANUP_TTL_PROP = "split.shard.cleanup.ttl";
+
+ public static final int DEFAULT_CLEANUP_PERIOD_SECONDS = 3600;
+ public static final int DEFAULT_CLEANUP_TTL_SECONDS = 3600 * 24 * 2;
+
public SplitShardCmd(OverseerCollectionMessageHandler ocmh) {
this.ocmh = ocmh;
+ int cleanupPeriod = ocmh.cloudManager.getClusterStateProvider().getClusterProperty(CLEANUP_PERIOD_PROP, DEFAULT_CLEANUP_PERIOD_SECONDS);
+ int cleanupTTL = ocmh.cloudManager.getClusterStateProvider().getClusterProperty(CLEANUP_TTL_PROP, DEFAULT_CLEANUP_TTL_SECONDS);
+ InactiveSliceCleanupTask task = new InactiveSliceCleanupTask(ocmh, cleanupTTL);
+ try {
+ ((MaintenanceThread) ocmh.overseer.getMaintenanceThread().getThread())
+ .addTask(task, cleanupPeriod, TimeUnit.SECONDS, false);
+ } catch (AlreadyExistsException e) { // should not happen
+ log.error("Already registered cleanup task? ", e);
+ throw new RuntimeException(e);
+ }
}
@Override
@@ -537,4 +558,55 @@ public class SplitShardCmd implements OverseerCollectionMessageHandler.Cmd {
}
return rangesStr;
}
+
+ /**
+ * This maintenance task checks whether there are shards that have been inactive for a long
+ * time (which usually means they are left-overs from shard splitting) and removes them after
+ * their cleanupTTL period elapsed.
+ */
+ public static class InactiveSliceCleanupTask implements MaintenanceTask {
+ private OverseerCollectionMessageHandler ocmh;
+ private int cleanupTTL;
+ private DeleteShardCmd cmd;
+
+ public InactiveSliceCleanupTask(OverseerCollectionMessageHandler ocmh, int cleanupTTL) {
+ this.ocmh = ocmh;
+ this.cleanupTTL = cleanupTTL;
+ cmd = (DeleteShardCmd)ocmh.commandMap.get(DELETESHARD);
+ }
+
+ @Override
+ public synchronized void run() {
+ try {
+ ClusterState state = ocmh.cloudManager.getClusterStateProvider().getClusterState();
+ state.forEachCollection(coll -> {
+ coll.getSlices().forEach(s -> {
+ if (Slice.State.INACTIVE.equals(s.getState())) {
+ String tstampStr = s.getStr(ZkStateReader.STATE_TIMESTAMP_PROP);
+ if (tstampStr == null || tstampStr.isEmpty()) {
+ return;
+ }
+ long timestamp = Long.parseLong(tstampStr);
+ long delta = TimeUnit.NANOSECONDS.toSeconds(ocmh.cloudManager.getTimeSource().getTime() - timestamp);
+ if (delta > cleanupTTL) {
+ ZkNodeProps props = new ZkNodeProps(ZkStateReader.COLLECTION_PROP, coll.getName(),
+ ZkStateReader.SHARD_ID_PROP, s.getName());
+ NamedList results = new NamedList();
+ try {
+ cmd.call(state, props, results);
+ if (results.get("failure") != null) {
+ throw new Exception("Failed to delete inactive shard: " + results);
+ }
+ } catch (Exception e) {
+ log.warn("Exception deleting inactive shard " + coll.getName() + "/" + s.getName(), e);
+ }
+ }
+ }
+ });
+ });
+ } catch (IOException e) {
+ log.warn("Error running inactive shard cleanup task", e);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/760f7a39/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
index 87bf481..2025b52 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/SliceMutator.java
@@ -178,6 +178,7 @@ public class SliceMutator {
props.remove("shard_parent_zk_session");
}
props.put(ZkStateReader.STATE_PROP, message.getStr(key));
+ props.put(ZkStateReader.STATE_TIMESTAMP_PROP, String.valueOf(dataProvider.getTimeSource().getTime()));
Slice newSlice = new Slice(slice.getName(), slice.getReplicasCopy(), props);
slicesCopy.put(slice.getName(), newSlice);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/760f7a39/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
index c65a676..5ffc858 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
@@ -496,6 +496,7 @@ public class SimCloudManager implements SolrCloudManager {
SolrQueryResponse queryResponse = new SolrQueryResponse();
autoScalingHandler.handleRequest(queryRequest, queryResponse);
if (queryResponse.getException() != null) {
+ LOG.debug("-- exception handling request", queryResponse.getException());
throw new IOException(queryResponse.getException());
}
SolrResponse rsp = new SolrResponseBase();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/760f7a39/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AlreadyExistsException.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AlreadyExistsException.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AlreadyExistsException.java
index 608b254..e29bd6f 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AlreadyExistsException.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/AlreadyExistsException.java
@@ -21,15 +21,15 @@ package org.apache.solr.client.solrj.cloud.autoscaling;
*/
public class AlreadyExistsException extends Exception {
- private final String path;
+ private final String id;
- public AlreadyExistsException(String path) {
- super("Path already exists: " + path);
- this.path = path;
+ public AlreadyExistsException(String id) {
+ super("Already exists: " + id);
+ this.id = id;
}
- public String getPath() {
- return path;
+ public String getId() {
+ return id;
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/760f7a39/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
index dbda3d8..aecdf89 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
@@ -95,6 +95,7 @@ public class ZkStateReader implements Closeable {
public static final String PROPERTY_VALUE_PROP = "property.value";
public static final String MAX_AT_ONCE_PROP = "maxAtOnce";
public static final String MAX_WAIT_SECONDS_PROP = "maxWaitSeconds";
+ public static final String STATE_TIMESTAMP_PROP = "stateTimestamp";
public static final String COLLECTIONS_ZKNODE = "/collections";
public static final String LIVE_NODES_ZKNODE = "/live_nodes";
public static final String ALIASES = "/aliases.json";