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 2017/05/23 09:58:35 UTC
[1/2] lucene-solr:jira/solr-10515: SOLR-10714: OverseerTriggerThread
does not start triggers on overseer start until autoscaling config watcher is
fired
Repository: lucene-solr
Updated Branches:
refs/heads/jira/solr-10515 e15ce8098 -> 5aed0bfa9
SOLR-10714: OverseerTriggerThread does not start triggers on overseer start until autoscaling config watcher is fired
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/cbe1e778
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/cbe1e778
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/cbe1e778
Branch: refs/heads/jira/solr-10515
Commit: cbe1e778a00b34ebfbe2b5b2e337f0f62fafc432
Parents: e15ce80
Author: Shalin Shekhar Mangar <sh...@apache.org>
Authored: Fri May 19 17:51:18 2017 +0530
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Sun May 21 21:42:27 2017 +0200
----------------------------------------------------------------------
solr/CHANGES.txt | 3 +
.../solr/cloud/autoscaling/AutoScaling.java | 2 +
.../autoscaling/OverseerTriggerThread.java | 162 ++++++++++---------
.../autoscaling/TriggerIntegrationTest.java | 56 ++++++-
4 files changed, 150 insertions(+), 73 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/cbe1e778/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index 211ed14..03f2f39 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -101,6 +101,9 @@ Bug Fixes
* SOLR-10602: Triggers should be able to restore state from old instances when taking over. (shalin)
+* SOLR-10714: OverseerTriggerThread does not start triggers on overseer start until autoscaling
+ config watcher is fired. (shalin)
+
Optimizations
----------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/cbe1e778/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
index e8d9b5e..e6f65af 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
@@ -23,6 +23,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import com.google.common.base.Preconditions;
import org.apache.lucene.store.AlreadyClosedException;
import org.apache.solr.common.MapWriter;
import org.apache.solr.core.CoreContainer;
@@ -129,6 +130,7 @@ public class AutoScaling {
private boolean isClosed = false;
public TriggerFactory(CoreContainer coreContainer) {
+ Preconditions.checkNotNull(coreContainer);
this.coreContainer = coreContainer;
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/cbe1e778/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
index dcc76b4..67d411e 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
@@ -54,6 +54,8 @@ public class OverseerTriggerThread implements Runnable, Closeable {
private final ZkStateReader zkStateReader;
+ private final SolrZkClient zkClient;
+
private final ScheduledTriggers scheduledTriggers;
private final AutoScaling.TriggerFactory triggerFactory;
@@ -74,7 +76,8 @@ public class OverseerTriggerThread implements Runnable, Closeable {
public OverseerTriggerThread(ZkController zkController) {
this.zkController = zkController;
zkStateReader = zkController.getZkStateReader();
- scheduledTriggers = new ScheduledTriggers(zkController.getZkClient());
+ zkClient = zkController.getZkClient();
+ scheduledTriggers = new ScheduledTriggers(zkClient);
triggerFactory = new AutoScaling.TriggerFactory(zkController.getCoreContainer());
}
@@ -90,13 +93,27 @@ public class OverseerTriggerThread implements Runnable, Closeable {
}
IOUtils.closeQuietly(triggerFactory);
IOUtils.closeQuietly(scheduledTriggers);
+ log.debug("OverseerTriggerThread has been closed explicitly");
}
@Override
public void run() {
int lastZnodeVersion = znodeVersion;
- SolrZkClient zkClient = zkStateReader.getZkClient();
- createWatcher(zkClient);
+
+ try {
+ refreshAutoScalingConf(new AutoScalingWatcher());
+ } catch (KeeperException.ConnectionLossException | KeeperException.SessionExpiredException e) {
+ log.warn("ZooKeeper watch triggered for autoscaling conf, but Solr cannot talk to ZK: [{}]", e.getMessage());
+ } catch (KeeperException e) {
+ log.error("A ZK error has occurred", e);
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "A ZK error has occurred", e);
+ } catch (InterruptedException e) {
+ // Restore the interrupted status
+ Thread.currentThread().interrupt();
+ log.warn("Interrupted", e);
+ } catch (Exception e) {
+ log.error("Unexpected exception", e);
+ }
while (!isClosed) {
Map<String, AutoScaling.Trigger> copy = null;
@@ -104,18 +121,31 @@ public class OverseerTriggerThread implements Runnable, Closeable {
// this can throw InterruptedException and we don't want to unlock if it did, so we keep this outside
// of the try/finally block
updateLock.lockInterruptibly();
+
+ // must check for close here before we await on the condition otherwise we can only be woken up on interruption
+ if (isClosed) {
+ log.warn("OverseerTriggerThread has been closed, exiting.");
+ break;
+ }
+
+ log.debug("Current znodeVersion {}, lastZnodeVersion {}", znodeVersion, lastZnodeVersion);
+
try {
if (znodeVersion == lastZnodeVersion) {
updated.await();
// are we closed?
- if (isClosed) break;
+ if (isClosed) {
+ log.warn("OverseerTriggerThread woken up but we are closed, exiting.");
+ break;
+ }
// spurious wakeup?
if (znodeVersion == lastZnodeVersion) continue;
- lastZnodeVersion = znodeVersion;
}
copy = new HashMap<>(activeTriggers);
+ lastZnodeVersion = znodeVersion;
+ log.debug("Processed trigger updates upto znodeVersion {}", znodeVersion);
} catch (InterruptedException e) {
// Restore the interrupted status
Thread.currentThread().interrupt();
@@ -145,77 +175,65 @@ public class OverseerTriggerThread implements Runnable, Closeable {
}
}
- private void createWatcher(SolrZkClient zkClient) {
+ class AutoScalingWatcher implements Watcher {
+ @Override
+ public void process(WatchedEvent watchedEvent) {
+ // session events are not change events, and do not remove the watcher
+ if (Event.EventType.None.equals(watchedEvent.getType())) {
+ return;
+ }
+
+ try {
+ refreshAutoScalingConf(this);
+ } catch (KeeperException.ConnectionLossException | KeeperException.SessionExpiredException e) {
+ log.warn("ZooKeeper watch triggered for autoscaling conf, but Solr cannot talk to ZK: [{}]", e.getMessage());
+ } catch (KeeperException e) {
+ log.error("A ZK error has occurred", e);
+ throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "A ZK error has occurred", e);
+ } catch (InterruptedException e) {
+ // Restore the interrupted status
+ Thread.currentThread().interrupt();
+ log.warn("Interrupted", e);
+ } catch (Exception e) {
+ log.error("Unexpected exception", e);
+ }
+ }
+
+ }
+
+ private void refreshAutoScalingConf(Watcher watcher) throws KeeperException, InterruptedException {
+ updateLock.lock();
try {
- zkClient.exists(ZkStateReader.SOLR_AUTOSCALING_CONF_PATH, new Watcher() {
- @Override
- public void process(WatchedEvent watchedEvent) {
- // session events are not change events, and do not remove the watcher
- if (Event.EventType.None.equals(watchedEvent.getType())) {
- return;
- }
+ if (isClosed) {
+ return;
+ }
+ final Stat stat = new Stat();
+ final byte[] data = zkClient.getData(ZkStateReader.SOLR_AUTOSCALING_CONF_PATH, watcher, stat, true);
+ log.debug("Refreshing {} with znode version {}", ZkStateReader.SOLR_AUTOSCALING_CONF_PATH, stat.getVersion());
+ if (znodeVersion >= stat.getVersion()) {
+ // protect against reordered watcher fires by ensuring that we only move forward
+ return;
+ }
+ znodeVersion = stat.getVersion();
+ Map<String, AutoScaling.Trigger> triggerMap = loadTriggers(triggerFactory, data);
- try {
- refreshAndWatch();
- } catch (KeeperException.ConnectionLossException | KeeperException.SessionExpiredException e) {
- log.warn("ZooKeeper watch triggered for autoscaling conf, but Solr cannot talk to ZK: [{}]", e.getMessage());
- } catch (KeeperException e) {
- log.error("A ZK error has occurred", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "A ZK error has occurred", e);
- } catch (InterruptedException e) {
- // Restore the interrupted status
- Thread.currentThread().interrupt();
- log.warn("Interrupted", e);
- } catch (Exception e) {
- log.error("Unexpected exception", e);
- }
- }
+ // remove all active triggers that have been removed from ZK
+ Set<String> trackingKeySet = activeTriggers.keySet();
+ trackingKeySet.retainAll(triggerMap.keySet());
- private void refreshAndWatch() throws KeeperException, InterruptedException {
- updateLock.lock();
- try {
- if (isClosed) {
- return;
- }
- final Stat stat = new Stat();
- final byte[] data = zkClient.getData(ZkStateReader.SOLR_AUTOSCALING_CONF_PATH, this, stat, true);
- log.debug("{} watcher fired with znode version {}", ZkStateReader.SOLR_AUTOSCALING_CONF_PATH, stat.getVersion());
- if (znodeVersion >= stat.getVersion()) {
- // protect against reordered watcher fires by ensuring that we only move forward
- return;
- }
- znodeVersion = stat.getVersion();
- Map<String, AutoScaling.Trigger> triggerMap = loadTriggers(triggerFactory, data);
-
- // remove all active triggers that have been removed from ZK
- Set<String> trackingKeySet = activeTriggers.keySet();
- trackingKeySet.retainAll(triggerMap.keySet());
-
- // now lets add or remove triggers which have been enabled or disabled respectively
- for (Map.Entry<String, AutoScaling.Trigger> entry : triggerMap.entrySet()) {
- String triggerName = entry.getKey();
- AutoScaling.Trigger trigger = entry.getValue();
- if (trigger.isEnabled()) {
- activeTriggers.put(triggerName, trigger);
- } else {
- activeTriggers.remove(triggerName);
- }
- }
- updated.signalAll();
- } finally {
- updateLock.unlock();
- }
+ // now lets add or remove triggers which have been enabled or disabled respectively
+ for (Map.Entry<String, AutoScaling.Trigger> entry : triggerMap.entrySet()) {
+ String triggerName = entry.getKey();
+ AutoScaling.Trigger trigger = entry.getValue();
+ if (trigger.isEnabled()) {
+ activeTriggers.put(triggerName, trigger);
+ } else {
+ activeTriggers.remove(triggerName);
}
- }, true);
- } catch (KeeperException.ConnectionLossException | KeeperException.SessionExpiredException e) {
- log.error("OverseerTriggerThread could not talk to ZooKeeper", e);
- } catch (KeeperException e) {
- log.error("Exception in OverseerTriggerThread", e);
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
- } catch (InterruptedException e) {
- // Restore the interrupted status
- Thread.currentThread().interrupt();
- log.error("OverseerTriggerThread interrupted", e);
+ }
+ updated.signalAll();
+ } finally {
+ updateLock.unlock();
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/cbe1e778/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
index 75cf320..7fb8da3 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
@@ -35,6 +35,7 @@ import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.response.CollectionAdminResponse;
import org.apache.solr.cloud.SolrCloudTestCase;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.util.NamedList;
@@ -76,7 +77,7 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
}
@Before
- public void setupTest() throws KeeperException, InterruptedException, IOException, SolrServerException {
+ public void setupTest() throws Exception {
waitForSeconds = 1 + random().nextInt(3);
actionCreated = new CountDownLatch(1);
triggerFiredLatch = new CountDownLatch(1);
@@ -89,6 +90,11 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
// todo nocommit -- add testing for the v2 path
// String path = random().nextBoolean() ? "/admin/autoscaling" : "/v2/cluster/autoscaling";
this.path = "/admin/autoscaling";
+ while (cluster.getJettySolrRunners().size() < 2) {
+ // perhaps a test stopped a node but didn't start it back
+ // lets start a node
+ cluster.startJettySolrRunner();
+ }
}
@Test
@@ -420,6 +426,54 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
lostNodeName, nodeLostEvent.getProperty(NodeLostTrigger.NodeLostEvent.NODE_NAME));
}
+ @Test
+ public void testContinueTriggersOnOverseerRestart() throws Exception {
+ CollectionAdminRequest.OverseerStatus status = new CollectionAdminRequest.OverseerStatus();
+ CloudSolrClient solrClient = cluster.getSolrClient();
+ CollectionAdminResponse adminResponse = status.process(solrClient);
+ NamedList<Object> response = adminResponse.getResponse();
+ String leader = (String) response.get("leader");
+ JettySolrRunner overseerNode = null;
+ int index = -1;
+ List<JettySolrRunner> jettySolrRunners = cluster.getJettySolrRunners();
+ for (int i = 0; i < jettySolrRunners.size(); i++) {
+ JettySolrRunner runner = jettySolrRunners.get(i);
+ if (runner.getNodeName().equals(leader)) {
+ overseerNode = runner;
+ index = i;
+ break;
+ }
+ }
+ assertNotNull(overseerNode);
+
+ String setTriggerCommand = "{" +
+ "'set-trigger' : {" +
+ "'name' : 'node_added_trigger'," +
+ "'event' : 'nodeAdded'," +
+ "'waitFor' : '" + waitForSeconds + "s'," +
+ "'enabled' : true," +
+ "'actions' : [{'name':'test','class':'" + TestTriggerAction.class.getName() + "'}]" +
+ "}}";
+ SolrRequest req = new AutoScalingHandlerTest.AutoScalingRequest(SolrRequest.METHOD.POST, path, setTriggerCommand);
+ response = solrClient.request(req);
+ assertEquals(response.get("result").toString(), "success");
+ if (!actionCreated.await(3, TimeUnit.SECONDS)) {
+ fail("The TriggerAction should have been created by now");
+ }
+
+ // stop the overseer, somebody else will take over as the overseer
+ cluster.stopJettySolrRunner(index);
+
+ JettySolrRunner newNode = cluster.startJettySolrRunner();
+ boolean await = triggerFiredLatch.await(20, TimeUnit.SECONDS);
+ assertTrue("The trigger did not fire at all", await);
+ assertTrue(triggerFired.get());
+ NodeAddedTrigger.NodeAddedEvent nodeAddedEvent = (NodeAddedTrigger.NodeAddedEvent) eventRef.get();
+ assertNotNull(nodeAddedEvent);
+ assertEquals("The node added trigger was fired but for a different node",
+ newNode.getNodeName(), nodeAddedEvent.getProperty(TriggerEventBase.NODE_NAME));
+ }
+
public static class TestTriggerAction implements TriggerAction {
public TestTriggerAction() {
[2/2] lucene-solr:jira/solr-10515: SOLR-10515 Save trigger state
after each run, and restore it when restarted. Use time-based event id-s.
Posted by ab...@apache.org.
SOLR-10515 Save trigger state after each run, and restore it when restarted.
Use time-based event id-s.
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/5aed0bfa
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/5aed0bfa
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/5aed0bfa
Branch: refs/heads/jira/solr-10515
Commit: 5aed0bfa90be94f1dfba295351604e7dc37dd4b6
Parents: cbe1e77
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Tue May 23 11:57:00 2017 +0200
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Tue May 23 11:57:00 2017 +0200
----------------------------------------------------------------------
.../solr/cloud/autoscaling/AutoScaling.java | 39 +++++-
.../cloud/autoscaling/NodeAddedTrigger.java | 44 +++++--
.../solr/cloud/autoscaling/NodeLostTrigger.java | 38 +++++-
.../autoscaling/OverseerTriggerThread.java | 10 ++
.../cloud/autoscaling/ScheduledTriggers.java | 130 ++++++++++++-------
.../solr/cloud/autoscaling/TriggerBase.java | 107 +++++++++++++++
.../cloud/autoscaling/TriggerEventBase.java | 56 ++++++--
.../cloud/autoscaling/TriggerEventQueue.java | 44 +++++--
.../src/java/org/apache/solr/util/IdUtils.java | 37 ++++++
.../cloud/autoscaling/NodeAddedTriggerTest.java | 6 +-
.../cloud/autoscaling/NodeLostTriggerTest.java | 6 +-
.../autoscaling/TriggerIntegrationTest.java | 29 ++++-
.../apache/solr/common/cloud/ZkStateReader.java | 1 +
13 files changed, 453 insertions(+), 94 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
index e6f65af..0d6da79 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
@@ -51,16 +51,33 @@ public class AutoScaling {
}
public interface TriggerEvent extends MapWriter {
+ /** Unique event id. */
+ String getId();
+
+ /** Event type. */
EventType getEventType();
+ /** Name of the trigger that fired the event. */
String getSource();
- long getEventNanoTime();
+ /** Timestamp of the actual event.
+ * NOTE: this is NOT the timestamp when the event was fired - events may be fired
+ * much later than the actual condition that generated the event, due to the "waitFor" limit.
+ */
+ long getEventTime();
+ /**
+ * Set event properties.
+ * @param properties may be null. A shallow copy of this parameter is used.
+ */
void setProperties(Map<String, Object> properties);
+ /**
+ * Get event properties (modifiable).
+ */
Map<String, Object> getProperties();
+ /** Get a named event property or null if missing. */
Object getProperty(String name);
}
@@ -102,25 +119,45 @@ public class AutoScaling {
* @param <E> the {@link TriggerEvent} which is handled by this Trigger
*/
public interface Trigger<E extends TriggerEvent> extends Closeable, Runnable {
+ /**
+ * Trigger name.
+ */
String getName();
+ /**
+ * Event type generated by this trigger.
+ */
EventType getEventType();
+ /** Returns true if this trigger is enabled. */
boolean isEnabled();
+ /** Trigger properties. */
Map<String, Object> getProperties();
+ /** Number of seconds to wait between fired events ("waitFor" property). */
int getWaitForSecond();
+ /** Actions to execute when event is fired. */
List<TriggerAction> getActions();
+ /** Set event listener to call when event is fired. */
void setListener(TriggerListener<E> listener);
+ /** Get event listener. */
TriggerListener<E> getListener();
+ /** Return true when this trigger is closed and cannot be used. */
boolean isClosed();
+ /** Set internal state of this trigger from another instance. */
void restoreState(Trigger<E> old);
+
+ /** Save internal state of this trigger in ZooKeeper. */
+ void saveState();
+
+ /** Restore internal state of this trigger from ZooKeeper. */
+ void restoreState();
}
public static class TriggerFactory implements Closeable {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
index dac40b5..d783f86 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
@@ -20,6 +20,7 @@ package org.apache.solr.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.HashMap;
import java.util.HashSet;
@@ -41,7 +42,7 @@ import org.slf4j.LoggerFactory;
/**
* Trigger for the {@link org.apache.solr.cloud.autoscaling.AutoScaling.EventType#NODEADDED} event
*/
-public class NodeAddedTrigger implements AutoScaling.Trigger<NodeAddedTrigger.NodeAddedEvent> {
+public class NodeAddedTrigger extends TriggerBase<NodeAddedTrigger.NodeAddedEvent> {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private final String name;
@@ -61,6 +62,7 @@ public class NodeAddedTrigger implements AutoScaling.Trigger<NodeAddedTrigger.No
public NodeAddedTrigger(String name, Map<String, Object> properties,
CoreContainer container) {
+ super(container.getZkController().getZkClient());
this.name = name;
this.properties = properties;
this.container = container;
@@ -76,6 +78,8 @@ public class NodeAddedTrigger implements AutoScaling.Trigger<NodeAddedTrigger.No
} else {
actions = Collections.emptyList();
}
+ lastLiveNodes = container.getZkController().getZkStateReader().getClusterState().getLiveNodes();
+ log.debug("Initial livenodes: {}", lastLiveNodes);
this.enabled = (boolean) properties.getOrDefault("enabled", true);
this.waitForSecond = ((Long) properties.getOrDefault("waitFor", -1L)).intValue();
this.eventType = AutoScaling.EventType.valueOf(properties.get("event").toString().toUpperCase(Locale.ROOT));
@@ -160,6 +164,28 @@ public class NodeAddedTrigger implements AutoScaling.Trigger<NodeAddedTrigger.No
}
@Override
+ protected Map<String, Object> getState() {
+ Map<String,Object> state = new HashMap<>();
+ state.put("lastLiveNodes", lastLiveNodes);
+ state.put("nodeNameVsTimeAdded", nodeNameVsTimeAdded);
+ return state;
+ }
+
+ @Override
+ protected void setState(Map<String, Object> state) {
+ this.lastLiveNodes.clear();
+ this.nodeNameVsTimeAdded.clear();
+ Collection<String> lastLiveNodes = (Collection<String>)state.get("lastLiveNodes");
+ if (lastLiveNodes != null) {
+ this.lastLiveNodes.addAll(lastLiveNodes);
+ }
+ Map<String,Long> nodeNameVsTimeAdded = (Map<String,Long>)state.get("nodeNameVsTimeAdded");
+ if (nodeNameVsTimeAdded != null) {
+ this.nodeNameVsTimeAdded.putAll(nodeNameVsTimeAdded);
+ }
+ }
+
+ @Override
public void run() {
try {
synchronized (this) {
@@ -187,21 +213,21 @@ public class NodeAddedTrigger implements AutoScaling.Trigger<NodeAddedTrigger.No
Set<String> copyOfNew = new HashSet<>(newLiveNodes);
copyOfNew.removeAll(lastLiveNodes);
copyOfNew.forEach(n -> {
- long nanoTime = System.nanoTime();
- nodeNameVsTimeAdded.put(n, nanoTime);
- log.info("Tracking new node: {} at {} nanotime", n, nanoTime);
+ long eventTime = System.currentTimeMillis();
+ nodeNameVsTimeAdded.put(n, eventTime);
+ log.debug("Tracking new node: {} at time {}", n, eventTime);
});
// has enough time expired to trigger events for a node?
for (Map.Entry<String, Long> entry : nodeNameVsTimeAdded.entrySet()) {
String nodeName = entry.getKey();
Long timeAdded = entry.getValue();
- long now = System.nanoTime();
- if (TimeUnit.SECONDS.convert(now - timeAdded, TimeUnit.NANOSECONDS) >= getWaitForSecond()) {
+ long now = System.currentTimeMillis();
+ if (TimeUnit.SECONDS.convert(now - timeAdded, TimeUnit.MILLISECONDS) >= getWaitForSecond()) {
// fire!
AutoScaling.TriggerListener<NodeAddedEvent> listener = listenerRef.get();
if (listener != null) {
- log.info("NodeAddedTrigger {} firing registered listener for node: {} added at {} nanotime, now: {} nanotime", name, nodeName, timeAdded, now);
+ log.debug("NodeAddedTrigger {} firing registered listener for node: {} added at time {} , now: {}", name, nodeName, timeAdded, now);
if (listener.triggerFired(new NodeAddedEvent(getEventType(), getName(), timeAdded, nodeName))) {
// remove from tracking set only if the fire was accepted
trackingKeySet.remove(nodeName);
@@ -227,8 +253,8 @@ public class NodeAddedTrigger implements AutoScaling.Trigger<NodeAddedTrigger.No
public static class NodeAddedEvent extends TriggerEventBase {
- public NodeAddedEvent(AutoScaling.EventType eventType, String source, long nodeAddedNanoTime, String nodeAdded) {
- super(eventType, source, nodeAddedNanoTime, Collections.singletonMap(NODE_NAME, nodeAdded));
+ public NodeAddedEvent(AutoScaling.EventType eventType, String source, long nodeAddedTime, String nodeAdded) {
+ super(eventType, source, nodeAddedTime, Collections.singletonMap(NODE_NAME, nodeAdded));
}
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
index 4eb71fe..2024a30 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
@@ -20,6 +20,7 @@ package org.apache.solr.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.HashMap;
import java.util.HashSet;
@@ -41,7 +42,7 @@ import org.slf4j.LoggerFactory;
/**
* Trigger for the {@link AutoScaling.EventType#NODELOST} event
*/
-public class NodeLostTrigger implements AutoScaling.Trigger<NodeLostTrigger.NodeLostEvent> {
+public class NodeLostTrigger extends TriggerBase<NodeLostTrigger.NodeLostEvent> {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private final String name;
@@ -61,6 +62,7 @@ public class NodeLostTrigger implements AutoScaling.Trigger<NodeLostTrigger.Node
public NodeLostTrigger(String name, Map<String, Object> properties,
CoreContainer container) {
+ super(container.getZkController().getZkClient());
this.name = name;
this.properties = properties;
this.container = container;
@@ -161,6 +163,28 @@ public class NodeLostTrigger implements AutoScaling.Trigger<NodeLostTrigger.Node
}
@Override
+ protected Map<String, Object> getState() {
+ Map<String,Object> state = new HashMap<>();
+ state.put("lastLiveNodes", lastLiveNodes);
+ state.put("nodeNameVsTimeRemoved", nodeNameVsTimeRemoved);
+ return state;
+ }
+
+ @Override
+ protected void setState(Map<String, Object> state) {
+ this.lastLiveNodes.clear();
+ this.nodeNameVsTimeRemoved.clear();
+ Collection<String> lastLiveNodes = (Collection<String>)state.get("lastLiveNodes");
+ if (lastLiveNodes != null) {
+ this.lastLiveNodes.addAll(lastLiveNodes);
+ }
+ Map<String,Long> nodeNameVsTimeRemoved = (Map<String,Long>)state.get("nodeNameVsTimeRemoved");
+ if (nodeNameVsTimeRemoved != null) {
+ this.nodeNameVsTimeRemoved.putAll(nodeNameVsTimeRemoved);
+ }
+ }
+
+ @Override
public void run() {
try {
synchronized (this) {
@@ -184,19 +208,19 @@ public class NodeLostTrigger implements AutoScaling.Trigger<NodeLostTrigger.Node
Set<String> copyOfLastLiveNodes = new HashSet<>(lastLiveNodes);
copyOfLastLiveNodes.removeAll(newLiveNodes);
copyOfLastLiveNodes.forEach(n -> {
- log.info("Tracking lost node: {}", n);
- nodeNameVsTimeRemoved.put(n, System.nanoTime());
+ log.debug("Tracking lost node: {}", n);
+ nodeNameVsTimeRemoved.put(n, System.currentTimeMillis());
});
// has enough time expired to trigger events for a node?
for (Map.Entry<String, Long> entry : nodeNameVsTimeRemoved.entrySet()) {
String nodeName = entry.getKey();
Long timeRemoved = entry.getValue();
- if (TimeUnit.SECONDS.convert(System.nanoTime() - timeRemoved, TimeUnit.NANOSECONDS) >= getWaitForSecond()) {
+ if (TimeUnit.SECONDS.convert(System.currentTimeMillis() - timeRemoved, TimeUnit.MILLISECONDS) >= getWaitForSecond()) {
// fire!
AutoScaling.TriggerListener<NodeLostEvent> listener = listenerRef.get();
if (listener != null) {
- log.info("NodeLostTrigger firing registered listener");
+ log.debug("NodeLostTrigger firing registered listener");
if (listener.triggerFired(new NodeLostEvent(getEventType(), getName(), timeRemoved, nodeName))) {
trackingKeySet.remove(nodeName);
}
@@ -221,8 +245,8 @@ public class NodeLostTrigger implements AutoScaling.Trigger<NodeLostTrigger.Node
public static class NodeLostEvent extends TriggerEventBase {
- public NodeLostEvent(AutoScaling.EventType eventType, String source, long nodeLostNanoTime, String nodeRemoved) {
- super(eventType, source, nodeLostNanoTime, Collections.singletonMap(NODE_NAME, nodeRemoved));
+ public NodeLostEvent(AutoScaling.EventType eventType, String source, long nodeLostTime, String nodeRemoved) {
+ super(eventType, source, nodeLostTime, Collections.singletonMap(NODE_NAME, nodeRemoved));
}
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
index 67d411e..f0357cc 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
@@ -231,6 +232,15 @@ public class OverseerTriggerThread implements Runnable, Closeable {
activeTriggers.remove(triggerName);
}
}
+ // remove state for all triggers that are not present in the current conf
+ if (zkClient.exists(ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH, true)) {
+ List<String> states = zkClient.getChildren(ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH, null, true);
+ for (String state : states) {
+ if (!triggerMap.containsKey(state)) {
+ zkClient.delete(ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH + "/" + state, -1, true);
+ }
+ }
+ }
updated.signalAll();
} finally {
updateLock.unlock();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
index 74b61a1..f14bcb7 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
@@ -127,48 +127,49 @@ public class ScheduledTriggers implements Closeable {
return true;
}
boolean replaying = event.getProperty(TriggerEventBase.REPLAYING) != null ? (Boolean)event.getProperty(TriggerEventBase.REPLAYING) : false;
- boolean enqueued = false;
- try {
- if (!replaying) {
- enqueued = scheduledTrigger.enqueue(event);
- }
- AutoScaling.Trigger source = scheduledSource.trigger;
- if (source.isClosed()) {
- log.warn("Ignoring autoscaling event " + event + " because the source trigger: " + source + " has already been closed");
- // we do not want to lose this event just because the trigger were closed, perhaps a replacement will need it
- return false;
- }
- if (hasPendingActions.compareAndSet(false, true)) {
- List<TriggerAction> actions = source.getActions();
- if (actions != null) {
- actionExecutor.submit(() -> {
- assert hasPendingActions.get();
- try {
- // let the action executor thread wait instead of the trigger thread so we use the throttle here
- actionThrottle.minimumWaitBetweenActions();
- actionThrottle.markAttemptingAction();
- for (TriggerAction action : actions) {
- try {
- action.process(event);
- } catch (Exception e) {
- log.error("Error executing action: " + action.getName() + " for trigger event: " + event, e);
- throw e;
- }
+ final boolean enqueued;
+ if (replaying) {
+ enqueued = false;
+ } else {
+ enqueued = scheduledTrigger.enqueue(event);
+ }
+ AutoScaling.Trigger source = scheduledSource.trigger;
+ if (source.isClosed()) {
+ log.warn("Ignoring autoscaling event " + event + " because the source trigger: " + source + " has already been closed");
+ // we do not want to lose this event just because the trigger were closed, perhaps a replacement will need it
+ return false;
+ }
+ if (hasPendingActions.compareAndSet(false, true)) {
+ List<TriggerAction> actions = source.getActions();
+ if (actions != null) {
+ actionExecutor.submit(() -> {
+ assert hasPendingActions.get();
+ log.debug("-- processing actions for " + event);
+ try {
+ // let the action executor thread wait instead of the trigger thread so we use the throttle here
+ actionThrottle.minimumWaitBetweenActions();
+ actionThrottle.markAttemptingAction();
+ for (TriggerAction action : actions) {
+ try {
+ action.process(event);
+ } catch (Exception e) {
+ log.error("Error executing action: " + action.getName() + " for trigger event: " + event, e);
+ throw e;
}
- } finally {
- hasPendingActions.set(false);
}
- });
- }
- return true;
- } else {
- // there is an action in the queue and we don't want to enqueue another until it is complete
- return false;
- }
- } finally {
- if (enqueued) {
- scheduledTrigger.dequeue();
+ if (enqueued) {
+ AutoScaling.TriggerEvent ev = scheduledTrigger.dequeue();
+ assert ev.getId().equals(event.getId());
+ }
+ } finally {
+ hasPendingActions.set(false);
+ }
+ });
}
+ return true;
+ } else {
+ // there is an action in the queue and we don't want to enqueue another until it is complete
+ return false;
}
});
scheduledTrigger.scheduledFuture = scheduledThreadPoolExecutor.scheduleWithFixedDelay(scheduledTrigger, 0, DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS, TimeUnit.SECONDS);
@@ -201,35 +202,66 @@ public class ScheduledTriggers implements Closeable {
}
scheduledTriggers.clear();
}
- ExecutorUtil.shutdownAndAwaitTermination(scheduledThreadPoolExecutor);
- ExecutorUtil.shutdownAndAwaitTermination(actionExecutor);
+ // shutdown and interrupt all running tasks because there's no longer any
+ // guarantee about cluster state
+ scheduledThreadPoolExecutor.shutdownNow();
+ actionExecutor.shutdownNow();
}
private class ScheduledTrigger implements Runnable, Closeable {
AutoScaling.Trigger trigger;
ScheduledFuture<?> scheduledFuture;
TriggerEventQueue queue;
+ boolean replay;
+ boolean isClosed;
ScheduledTrigger(AutoScaling.Trigger trigger, SolrZkClient zkClient, Overseer.Stats stats) {
this.trigger = trigger;
this.queue = new TriggerEventQueue(zkClient, trigger.getName(), stats);
+ this.replay = true;
+ this.isClosed = false;
}
public boolean enqueue(AutoScaling.TriggerEvent event) {
+ if (isClosed) {
+ throw new AlreadyClosedException("ScheduledTrigger " + trigger.getName() + " has been closed.");
+ }
+ log.debug("--Enqueue " + event);
return queue.offerEvent(event);
}
public AutoScaling.TriggerEvent dequeue() {
- return queue.pollEvent();
+ if (isClosed) {
+ throw new AlreadyClosedException("ScheduledTrigger " + trigger.getName() + " has been closed.");
+ }
+ AutoScaling.TriggerEvent event = queue.pollEvent();
+ log.debug("--Dequeue " + event);
+ return event;
}
@Override
public void run() {
- log.info("--running " + trigger.getName());
- // replay accumulated events first, if any
- AutoScaling.TriggerEvent event;
- while ((event = queue.pollEvent()) != null) {
- trigger.getListener().triggerFired(event);
+ if (isClosed) {
+ throw new AlreadyClosedException("ScheduledTrigger " + trigger.getName() + " has been closed.");
+ }
+ log.debug("-- run scheduled trigger " + trigger.getName());
+ // replay accumulated events on first run, if any
+ if (replay) {
+ log.debug("--replaying...");
+ AutoScaling.TriggerEvent event;
+ // peek first without removing - we may crash before calling the listener
+ while ((event = queue.peekEvent()) != null) {
+ // override REPLAYING=true
+ event.getProperties().put(TriggerEventBase.REPLAYING, true);
+ if (! trigger.getListener().triggerFired(event)) {
+ log.error("Failed to re-play event, discarding: " + event);
+ }
+ log.debug("--replayed event: " + event);
+ queue.pollEvent(); // always remove it from queue
+ }
+ // now restore saved state to possibly generate new events from old state on the first run
+ trigger.restoreState();
+ replay = false;
}
// fire a trigger only if an action is not pending
// note this is not fool proof e.g. it does not prevent an action being executed while a trigger
@@ -241,12 +273,16 @@ public class ScheduledTriggers implements Closeable {
// log but do not propagate exception because an exception thrown from a scheduled operation
// will suppress future executions
log.error("Unexpected execution from trigger: " + trigger.getName(), e);
+ } finally {
+ // checkpoint after each run
+ trigger.saveState();
}
}
}
@Override
public void close() throws IOException {
+ isClosed = true;
if (scheduledFuture != null) {
scheduledFuture.cancel(true);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerBase.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerBase.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerBase.java
new file mode 100644
index 0000000..b94824c
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerBase.java
@@ -0,0 +1,107 @@
+/*
+ * 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;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Map;
+
+import org.apache.solr.common.cloud.SolrZkClient;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.util.Utils;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Base class for {@link org.apache.solr.cloud.autoscaling.AutoScaling.Trigger} implementations.
+ * It handles state snapshot / restore in ZK.
+ */
+public abstract class TriggerBase<E extends AutoScaling.TriggerEvent> implements AutoScaling.Trigger<E> {
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ protected SolrZkClient zkClient;
+ protected byte[] lastState;
+
+
+ protected TriggerBase(SolrZkClient zkClient) {
+ this.zkClient = zkClient;
+ try {
+ zkClient.makePath(ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH, false, true);
+ } catch (KeeperException | InterruptedException e) {
+ LOG.warn("Exception checking ZK path " + ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH, e);
+ }
+ }
+
+ /**
+ * Prepare and return internal state of this trigger in a format suitable for persisting in ZK.
+ * @return map of internal state properties. Note: values must be supported by {@link Utils#toJSON(Object)}.
+ */
+ protected abstract Map<String,Object> getState();
+
+ /**
+ * Restore internal state of this trigger from properties retrieved from ZK.
+ * @param state never null but may be empty.
+ */
+ protected abstract void setState(Map<String,Object> state);
+
+ @Override
+ public void saveState() {
+ Map<String,Object> state = getState();
+ byte[] data = Utils.toJSON(state);
+ // skip saving if identical
+ if (lastState != null && lastState.equals(data)) {
+ LOG.debug("--skip saving " + getName());
+ return;
+ }
+ String path = ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH + "/" + getName();
+ LOG.info("--save state: " + path + ": " + state);
+ try {
+ if (zkClient.exists(path, true)) {
+ // update
+ zkClient.setData(path, data, -1, true);
+ } else {
+ // create
+ zkClient.create(path, data, CreateMode.PERSISTENT, true);
+ }
+ lastState = data;
+ } catch (KeeperException | InterruptedException e) {
+ LOG.warn("Exception updating trigger state '" + path + "'", e);
+ }
+ }
+
+ @Override
+ public void restoreState() {
+ byte[] data = null;
+ String path = ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH + "/" + getName();
+ LOG.info("--restore state: " + path);
+ try {
+ if (zkClient.exists(path, true)) {
+ data = zkClient.getData(path, null, new Stat(), true);
+ }
+ } catch (KeeperException | InterruptedException e) {
+ LOG.warn("Exception getting trigger state '" + path + "'", e);
+ }
+ if (data != null) {
+ Map<String, Object> state = (Map<String, Object>) Utils.fromJSON(data);
+ LOG.info(" -- state: " + state);
+ setState(state);
+ lastState = data;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventBase.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventBase.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventBase.java
index 168dfaa..13fd611 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventBase.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventBase.java
@@ -20,7 +20,7 @@ import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
-import org.apache.solr.common.MapWriter;
+import org.apache.solr.util.IdUtils;
/**
* Base class for event implementations.
@@ -29,29 +29,41 @@ public abstract class TriggerEventBase implements AutoScaling.TriggerEvent {
public static final String REPLAYING = "replaying";
public static final String NODE_NAME = "nodeName";
+ protected final String id;
protected final String source;
- protected final long eventNanoTime;
+ protected final long eventTime;
protected final AutoScaling.EventType eventType;
protected final Map<String, Object> properties = new HashMap<>();
- protected TriggerEventBase(AutoScaling.EventType eventType, String source, long eventNanoTime,
+ protected TriggerEventBase(AutoScaling.EventType eventType, String source, long eventTime,
Map<String, Object> properties) {
+ this(IdUtils.timeRandomId(eventTime), eventType, source, eventTime, properties);
+ }
+
+ protected TriggerEventBase(String id, AutoScaling.EventType eventType, String source, long eventTime,
+ Map<String, Object> properties) {
+ this.id = id;
this.eventType = eventType;
this.source = source;
- this.eventNanoTime = eventNanoTime;
+ this.eventTime = eventTime;
if (properties != null) {
this.properties.putAll(properties);
}
}
@Override
+ public String getId() {
+ return id;
+ }
+
+ @Override
public String getSource() {
return source;
}
@Override
- public long getEventNanoTime() {
- return eventNanoTime;
+ public long getEventTime() {
+ return eventTime;
}
@Override
@@ -79,17 +91,43 @@ public abstract class TriggerEventBase implements AutoScaling.TriggerEvent {
@Override
public void writeMap(EntryWriter ew) throws IOException {
+ ew.put("id", id);
ew.put("source", source);
- ew.put("eventNanoTime", eventNanoTime);
+ ew.put("eventTime", eventTime);
ew.put("eventType", eventType.toString());
ew.put("properties", properties);
}
@Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ TriggerEventBase that = (TriggerEventBase) o;
+
+ if (eventTime != that.eventTime) return false;
+ if (!id.equals(that.id)) return false;
+ if (!source.equals(that.source)) return false;
+ if (eventType != that.eventType) return false;
+ return properties.equals(that.properties);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = id.hashCode();
+ result = 31 * result + source.hashCode();
+ result = 31 * result + (int) (eventTime ^ (eventTime >>> 32));
+ result = 31 * result + eventType.hashCode();
+ result = 31 * result + properties.hashCode();
+ return result;
+ }
+
+ @Override
public String toString() {
return this.getClass().getSimpleName() + "{" +
- "source='" + source + '\'' +
- ", eventNanoTime=" + eventNanoTime +
+ "id='" + id + '\'' +
+ ", source='" + source + '\'' +
+ ", eventTime=" + eventTime +
", properties=" + properties +
'}';
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventQueue.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventQueue.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventQueue.java
index 83b30fc..df1b2c4 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventQueue.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/TriggerEventQueue.java
@@ -1,7 +1,6 @@
package org.apache.solr.cloud.autoscaling;
import java.lang.invoke.MethodHandles;
-import java.util.HashMap;
import java.util.Map;
import org.apache.solr.cloud.DistributedQueue;
@@ -19,11 +18,10 @@ import org.slf4j.LoggerFactory;
public class TriggerEventQueue extends DistributedQueue {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- public static class ReplayingEvent extends TriggerEventBase {
+ public static class QueuedEvent extends TriggerEventBase {
- public ReplayingEvent(AutoScaling.EventType eventType, String source, long eventNanoTime, Map<String, Object> properties) {
- super(eventType, source, eventNanoTime, properties);
- this.properties.put(REPLAYING, true);
+ public QueuedEvent(String id, AutoScaling.EventType eventType, String source, long eventTime, Map<String, Object> properties) {
+ super(id, eventType, source, eventTime, properties);
}
}
@@ -45,6 +43,28 @@ public class TriggerEventQueue extends DistributedQueue {
}
}
+ public AutoScaling.TriggerEvent peekEvent() {
+ byte[] data;
+ try {
+ while ((data = peek()) != null) {
+ if (data.length == 0) {
+ LOG.warn("ignoring empty data...");
+ continue;
+ }
+ try {
+ Map<String, Object> map = (Map<String, Object>) Utils.fromJSON(data);
+ return fromMap(map);
+ } catch (Exception e) {
+ LOG.warn("Invalid event data, ignoring: " + new String(data));
+ continue;
+ }
+ }
+ } catch (KeeperException | InterruptedException e) {
+ LOG.warn("Exception peeking queue of trigger " + triggerName, e);
+ }
+ return null;
+ }
+
public AutoScaling.TriggerEvent pollEvent() {
byte[] data;
try {
@@ -55,11 +75,7 @@ public class TriggerEventQueue extends DistributedQueue {
}
try {
Map<String, Object> map = (Map<String, Object>) Utils.fromJSON(data);
- String source = (String)map.get("source");
- long eventNanoTime = ((Number)map.get("eventNanoTime")).longValue();
- AutoScaling.EventType eventType = AutoScaling.EventType.valueOf((String)map.get("eventType"));
- Map<String, Object> properties = (Map<String, Object>)map.get("properties");
- return new ReplayingEvent(eventType, source, eventNanoTime, properties);
+ return fromMap(map);
} catch (Exception e) {
LOG.warn("Invalid event data, ignoring: " + new String(data));
continue;
@@ -71,4 +87,12 @@ public class TriggerEventQueue extends DistributedQueue {
return null;
}
+ private static QueuedEvent fromMap(Map<String, Object> map) {
+ String id = (String)map.get("id");
+ String source = (String)map.get("source");
+ long eventTime = ((Number)map.get("eventTime")).longValue();
+ AutoScaling.EventType eventType = AutoScaling.EventType.valueOf((String)map.get("eventType"));
+ Map<String, Object> properties = (Map<String, Object>)map.get("properties");
+ return new QueuedEvent(id, eventType, source, eventTime, properties);
+ }
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/java/org/apache/solr/util/IdUtils.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/util/IdUtils.java b/solr/core/src/java/org/apache/solr/util/IdUtils.java
new file mode 100644
index 0000000..982e691
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/util/IdUtils.java
@@ -0,0 +1,37 @@
+package org.apache.solr.util;
+
+import org.apache.lucene.util.StringHelper;
+
+/**
+ *
+ */
+public class IdUtils {
+
+ /**
+ * Generate a short random id (see {@link StringHelper#randomId()}).
+ */
+ public static final String randomId() {
+ return StringHelper.idToString(StringHelper.randomId());
+ }
+
+ /**
+ * Generate a random id with a timestamp, in the format:
+ * <code>hex(timestamp) + 'T' + randomId</code>. This method
+ * uses {@link System#currentTimeMillis()} as a timestamp.
+ */
+ public static final String timeRandomId() {
+ return timeRandomId(System.currentTimeMillis());
+ }
+
+ /**
+ * Generate a random id with a timestamp, in the format:
+ * <code>hex(timestamp) + 'T' + randomId</code>.
+ * @param time value representing timestamp
+ */
+ public static final String timeRandomId(long time) {
+ StringBuilder sb = new StringBuilder(Long.toHexString(time));
+ sb.append('T');
+ sb.append(randomId());
+ return sb.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeAddedTriggerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeAddedTriggerTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeAddedTriggerTest.java
index f916382..ef8ccf1 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeAddedTriggerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeAddedTriggerTest.java
@@ -65,7 +65,7 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
trigger.setListener(event -> {
if (fired.compareAndSet(false, true)) {
eventRef.set(event);
- if (System.nanoTime() - event.getEventNanoTime() <= TimeUnit.NANOSECONDS.convert(waitForSeconds, TimeUnit.SECONDS)) {
+ if (System.nanoTime() - event.getEventTime() <= TimeUnit.NANOSECONDS.convert(waitForSeconds, TimeUnit.SECONDS)) {
fail("NodeAddedListener was fired before the configured waitFor period");
}
} else {
@@ -99,7 +99,7 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
AtomicBoolean fired = new AtomicBoolean(false);
trigger.setListener(event -> {
if (fired.compareAndSet(false, true)) {
- if (System.nanoTime() - event.getEventNanoTime() <= TimeUnit.NANOSECONDS.convert(waitTime, TimeUnit.SECONDS)) {
+ if (System.nanoTime() - event.getEventTime() <= TimeUnit.NANOSECONDS.convert(waitTime, TimeUnit.SECONDS)) {
fail("NodeAddedListener was fired before the configured waitFor period");
}
} else {
@@ -188,7 +188,7 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
newTrigger.setListener(event -> {
if (fired.compareAndSet(false, true)) {
eventRef.set(event);
- if (System.nanoTime() - event.getEventNanoTime() <= TimeUnit.NANOSECONDS.convert(waitTime, TimeUnit.SECONDS)) {
+ if (System.nanoTime() - event.getEventTime() <= TimeUnit.NANOSECONDS.convert(waitTime, TimeUnit.SECONDS)) {
fail("NodeAddedListener was fired before the configured waitFor period");
}
} else {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java
index 6d2e4a7..ec110dc 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java
@@ -66,7 +66,7 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
trigger.setListener(event -> {
if (fired.compareAndSet(false, true)) {
eventRef.set(event);
- if (System.nanoTime() - event.getEventNanoTime() <= TimeUnit.NANOSECONDS.convert(waitForSeconds, TimeUnit.SECONDS)) {
+ if (System.nanoTime() - event.getEventTime() <= TimeUnit.NANOSECONDS.convert(waitForSeconds, TimeUnit.SECONDS)) {
fail("NodeLostListener was fired before the configured waitFor period");
}
} else {
@@ -102,7 +102,7 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
AtomicBoolean fired = new AtomicBoolean(false);
trigger.setListener(event -> {
if (fired.compareAndSet(false, true)) {
- if (System.nanoTime() - event.getEventNanoTime() <= TimeUnit.NANOSECONDS.convert(waitTime, TimeUnit.SECONDS)) {
+ if (System.nanoTime() - event.getEventTime() <= TimeUnit.NANOSECONDS.convert(waitTime, TimeUnit.SECONDS)) {
fail("NodeLostListener was fired before the configured waitFor period");
}
} else {
@@ -215,7 +215,7 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
newTrigger.setListener(event -> {
if (fired.compareAndSet(false, true)) {
eventRef.set(event);
- if (System.nanoTime() - event.getEventNanoTime() <= TimeUnit.NANOSECONDS.convert(waitForSeconds, TimeUnit.SECONDS)) {
+ if (System.nanoTime() - event.getEventTime() <= TimeUnit.NANOSECONDS.convert(waitForSeconds, TimeUnit.SECONDS)) {
fail("NodeLostListener was fired before the configured waitFor period");
}
} else {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
index 7fb8da3..235d748 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/TriggerIntegrationTest.java
@@ -31,7 +31,6 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.solr.client.solrj.SolrRequest;
-import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.embedded.JettySolrRunner;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
@@ -42,7 +41,6 @@ import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.Utils;
import org.apache.solr.util.LogLevel;
import org.apache.solr.util.TimeOut;
-import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -64,6 +62,8 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
private static CountDownLatch triggerFiredLatch;
private static int waitForSeconds = 1;
private static CyclicBarrier actionStarted;
+ private static CyclicBarrier actionInterrupted;
+ private static CyclicBarrier actionCompleted;
private static AtomicBoolean triggerFired;
private static AtomicReference<AutoScaling.TriggerEvent> eventRef;
@@ -83,6 +83,8 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
triggerFiredLatch = new CountDownLatch(1);
triggerFired = new AtomicBoolean(false);
actionStarted = new CyclicBarrier(2);
+ actionInterrupted = new CyclicBarrier(2);
+ actionCompleted = new CyclicBarrier(2);
eventRef = new AtomicReference<>();
// clear any persisted auto scaling configuration
Stat stat = zkClient().setData(SOLR_AUTOSCALING_CONF_PATH, Utils.toJSON(new ZkNodeProps()), true);
@@ -495,7 +497,7 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
public void process(AutoScaling.TriggerEvent event) {
if (triggerFired.compareAndSet(false, true)) {
eventRef.set(event);
- if (System.nanoTime() - event.getEventNanoTime() <= TimeUnit.NANOSECONDS.convert(waitForSeconds, TimeUnit.SECONDS)) {
+ if (System.nanoTime() - event.getEventTime() <= TimeUnit.NANOSECONDS.convert(waitForSeconds, TimeUnit.SECONDS)) {
fail("NodeAddedListener was fired before the configured waitFor period");
}
triggerFiredLatch.countDown();
@@ -543,7 +545,17 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
try {
Thread.sleep(5000);
triggerFired.compareAndSet(false, true);
+ try {
+ actionCompleted.await();
+ } catch (InterruptedException | BrokenBarrierException e) {
+ throw new RuntimeException("broken barrier", e);
+ }
} catch (InterruptedException e) {
+ try {
+ actionInterrupted.await();
+ } catch (InterruptedException | BrokenBarrierException e1) {
+ throw new RuntimeException("broken barrier", e1);
+ }
return;
}
}
@@ -590,8 +602,9 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
// add node to generate the event
JettySolrRunner newNode = cluster.startJettySolrRunner();
+ // we are the first party so we expect index 1
int await = actionStarted.await(60, TimeUnit.SECONDS);
- assertEquals("The trigger did not fire at all", 1, await);
+ assertEquals("action started too early", 1, await);
// event should be there
NodeAddedTrigger.NodeAddedEvent nodeAddedEvent = (NodeAddedTrigger.NodeAddedEvent) eventRef.get();
assertNotNull(nodeAddedEvent);
@@ -601,10 +614,16 @@ public class TriggerIntegrationTest extends SolrCloudTestCase {
// kill overseer leader
cluster.stopJettySolrRunner(overseerLeaderIndex);
Thread.sleep(5000);
+ // we are the last party, so we expect index 0
+ await = actionInterrupted.await(3, TimeUnit.SECONDS);
+ assertEquals("action wasn't interrupted", 0, await);
// new overseer leader should be elected and run triggers
newNode = cluster.startJettySolrRunner();
// it should fire again but not complete yet
- await = actionStarted.await(600, TimeUnit.SECONDS);
+ await = actionStarted.await(60, TimeUnit.SECONDS);
AutoScaling.TriggerEvent replayedEvent = eventRef.get();
+ assertTrue(replayedEvent instanceof TriggerEventQueue.QueuedEvent);
+ await = actionCompleted.await(10, TimeUnit.SECONDS);
+ assertTrue(triggerFired.get());
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/5aed0bfa/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 45995b8..f23b0b5 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
@@ -93,6 +93,7 @@ public class ZkStateReader implements Closeable {
public static final String SOLR_SECURITY_CONF_PATH = "/security.json";
public static final String SOLR_AUTOSCALING_CONF_PATH = "/autoscaling.json";
public static final String SOLR_AUTOSCALING_EVENTS_PATH = "/autoscaling/events";
+ public static final String SOLR_AUTOSCALING_TRIGGER_STATE_PATH = "/autoscaling/triggerState";
public static final String REPLICATION_FACTOR = "replicationFactor";
public static final String MAX_SHARDS_PER_NODE = "maxShardsPerNode";