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/07/05 19:43:16 UTC
[2/5] lucene-solr:jira/solr-10996: WIP.
WIP.
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/6a35a889
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/6a35a889
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/6a35a889
Branch: refs/heads/jira/solr-10996
Commit: 6a35a8896e9cdc065c7e2dd3c07f009243c981c1
Parents: 9e93a78
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Tue Jul 4 10:01:49 2017 +0200
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Tue Jul 4 10:01:49 2017 +0200
----------------------------------------------------------------------
.../solr/cloud/autoscaling/AutoScaling.java | 22 ++---
.../cloud/autoscaling/AutoScalingConfig.java | 89 +++++++++++++++-----
.../cloud/autoscaling/HttpTriggerListener.java | 38 +++++++++
.../cloud/autoscaling/LogTriggerListener.java | 42 +++++++++
.../autoscaling/AutoScalingHandlerTest.java | 6 +-
.../cloud/autoscaling/NodeAddedTriggerTest.java | 10 +--
.../cloud/autoscaling/NodeLostTriggerTest.java | 10 +--
7 files changed, 172 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a35a889/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 c1d676d..5f80f80 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
@@ -64,21 +64,17 @@ public class AutoScaling {
}
public interface TriggerListener {
+
+ void init(CoreContainer coreContainer);
+
/**
- * This method is executed when a trigger is ready to fire.
- *
- * @param event a subclass of {@link TriggerEvent}
- * @return true if the listener was ready to perform actions on the event, false
- * otherwise. If false was returned then callers should assume the event was discarded.
+ * This method is called when either a particular <code>stage</code> or
+ * <code>actionName</code> is reached during event processing.
+ * @param stage {@link TriggerStage} that this listener was registered for, or null
+ * @param actionName {@link TriggerAction} name that this listener was registered for, or null
+ * @param event current event being processed
*/
- boolean triggerFired(TriggerEvent event);
- }
-
- public static class HttpCallbackListener implements TriggerListener {
- @Override
- public boolean triggerFired(TriggerEvent event) {
- return true;
- }
+ void onEvent(TriggerStage stage, String actionName, TriggerEvent event);
}
/**
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a35a889/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingConfig.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingConfig.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingConfig.java
index 5714ac9..753f6e9 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingConfig.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingConfig.java
@@ -16,6 +16,8 @@
*/
package org.apache.solr.cloud.autoscaling;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -25,12 +27,15 @@ import java.util.Map;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
import org.apache.solr.common.params.AutoScalingParams;
import org.apache.solr.common.util.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Simple bean representation of <code>autoscaling.json</code>, which parses data
* lazily.
*/
public class AutoScalingConfig {
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private final Map<String, Object> jsonMap;
@@ -42,15 +47,26 @@ public class AutoScalingConfig {
* Bean representation of {@link org.apache.solr.cloud.autoscaling.AutoScaling.TriggerListener} config.
*/
public static class ListenerConfig {
- public String trigger;
- public List<String> stages;
- public String listenerClass;
- public List<Map<String, String>> beforeActions;
- public List<Map<String, String>> afterActions;
+ public final String trigger;
+ public final List<AutoScaling.TriggerStage> stages;
+ public final String listenerClass;
+ public final List<Map<String, String>> beforeActions;
+ public final List<Map<String, String>> afterActions;
+ public final Map<String, Object> properties = new HashMap<>();
public ListenerConfig(Map<String, Object> properties) {
+ this.properties.putAll(properties);
trigger = (String)properties.get(AutoScalingParams.TRIGGER);
- stages = (List<String>)properties.getOrDefault(AutoScalingParams.STAGE, Collections.emptyList());
+ List<String> stageNames = (List<String>)properties.getOrDefault(AutoScalingParams.STAGE, Collections.emptyList());
+ stages = new ArrayList<>(stageNames.size());
+ for (String name : stageNames) {
+ try {
+ AutoScaling.TriggerStage stage = AutoScaling.TriggerStage.valueOf(name.toUpperCase(Locale.ROOT));
+ stages.add(stage);
+ } catch (Exception e) {
+ LOG.warn("Invalid stage name '" + name + "' in listener config, skipping: " + properties);
+ }
+ }
listenerClass = (String)properties.get(AutoScalingParams.CLASS);
beforeActions = (List<Map<String, String>>)properties.getOrDefault(AutoScalingParams.BEFORE_ACTION, Collections.emptyList());
afterActions = (List<Map<String, String>>)properties.getOrDefault(AutoScalingParams.AFTER_ACTION, Collections.emptyList());
@@ -80,6 +96,13 @@ public class AutoScalingConfig {
}
/**
+ * Return the original JSON map representation that was used for building this config.
+ */
+ public Map<String, Object> getJsonMap() {
+ return jsonMap;
+ }
+
+ /**
* Get {@link Policy} configuration.
*/
public Policy getPolicy() {
@@ -108,6 +131,26 @@ public class AutoScalingConfig {
}
/**
+ * Check whether triggers for specific event type exist.
+ * @param types list of event types
+ * @return true if there's at least one trigger matching at least one event type,
+ * false otherwise,
+ */
+ public boolean hasTriggerForEvents(AutoScaling.EventType... types) {
+ if (types == null || types.length == 0) {
+ return false;
+ }
+ for (TriggerConfig config : getTriggerConfigs().values()) {
+ for (AutoScaling.EventType type : types) {
+ if (config.eventType.equals(type)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ /**
* Get listener configurations.
*/
public Map<String, ListenerConfig> getListenerConfigs() {
@@ -126,22 +169,30 @@ public class AutoScalingConfig {
}
/**
- * Check whether triggers for specific event type exist.
- * @param types list of event types
- * @return true if there's at least one trigger matching at least one event type,
- * false otherwise,
+ * Get listeners for a specific trigger name.
+ * @param triggerName name of the trigger
+ * @return a map where keys are the stages and values are the list of listener configs that apply to a
+ * particular triggerName and stage.
*/
- public boolean hasTriggerForEvents(AutoScaling.EventType... types) {
- if (types == null || types.length == 0) {
- return false;
- }
- for (TriggerConfig config : getTriggerConfigs().values()) {
- for (AutoScaling.EventType type : types) {
- if (config.eventType.equals(type)) {
- return true;
+ public Map<AutoScaling.TriggerStage, List<ListenerConfig>> getListenerConfigsForTrigger(String triggerName) {
+ Map<String, ListenerConfig> configs = getListenerConfigs();
+ if (configs.isEmpty()) {
+ return Collections.emptyMap();
+ } else {
+ Map<AutoScaling.TriggerStage, List<ListenerConfig>> res = new HashMap<>();
+ for (ListenerConfig cfg : configs.values()) {
+ if (triggerName.equals(cfg.trigger)) {
+ for (AutoScaling.TriggerStage stage : cfg.stages) {
+ List<ListenerConfig> cfgsPerStage = res.get(stage);
+ if (cfgsPerStage == null) {
+ cfgsPerStage = new ArrayList<>();
+ res.put(stage, cfgsPerStage);
+ }
+ cfgsPerStage.add(cfg);
+ }
}
}
+ return res;
}
- return false;
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a35a889/solr/core/src/java/org/apache/solr/cloud/autoscaling/HttpTriggerListener.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/HttpTriggerListener.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/HttpTriggerListener.java
new file mode 100644
index 0000000..57a9c31
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/HttpTriggerListener.java
@@ -0,0 +1,38 @@
+/*
+ * 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 org.apache.http.client.HttpClient;
+import org.apache.solr.core.CoreContainer;
+
+/**
+ * Simple HTTP callback that sends trigger events as JSON.
+ */
+public class HttpTriggerListener implements AutoScaling.TriggerListener {
+
+ private HttpClient httpClient;
+
+ @Override
+ public void init(CoreContainer coreContainer) {
+ httpClient = coreContainer.getUpdateShardHandler().getHttpClient();
+ }
+
+ @Override
+ public void onEvent(AutoScaling.TriggerStage stage, String actionName, TriggerEvent event) {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a35a889/solr/core/src/java/org/apache/solr/cloud/autoscaling/LogTriggerListener.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/LogTriggerListener.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/LogTriggerListener.java
new file mode 100644
index 0000000..a4d4fdd
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/LogTriggerListener.java
@@ -0,0 +1,42 @@
+/*
+ * 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 org.apache.solr.core.CoreContainer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Implementation of {@link org.apache.solr.cloud.autoscaling.AutoScaling.TriggerListener} that reports
+ * events to a log.
+ */
+public class LogTriggerListener implements AutoScaling.TriggerListener {
+ private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Override
+ public void init(CoreContainer coreContainer) {
+
+ }
+
+ @Override
+ public void onEvent(AutoScaling.TriggerStage stage, String actionName, TriggerEvent event) {
+ LOG.info("stage={}, actionName={}, event={}", stage, actionName, event);
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a35a889/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoScalingHandlerTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoScalingHandlerTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoScalingHandlerTest.java
index 06c2920..be7fb0c 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoScalingHandlerTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/AutoScalingHandlerTest.java
@@ -356,7 +356,7 @@ public class AutoScalingHandlerTest extends SolrCloudTestCase {
"'trigger' : 'node_lost_trigger'," +
"'stage' : ['STARTED','ABORTED','SUCCEEDED']," +
"'beforeAction' : 'execute_plan'," +
- "'class' : 'org.apache.solr.cloud.autoscaling.AutoScaling$HttpCallbackListener'," +
+ "'class' : 'org.apache.solr.cloud.autoscaling.AutoScaling$HttpTriggerListener'," +
"'url' : 'http://xyz.com/on_node_lost?node={$LOST_NODE_NAME}'" +
"}" +
"}";
@@ -371,7 +371,7 @@ public class AutoScalingHandlerTest extends SolrCloudTestCase {
assertTrue(listeners.containsKey("xyz"));
Map<String, Object> xyzListener = (Map<String, Object>) listeners.get("xyz");
assertEquals(5, xyzListener.size());
- assertEquals("org.apache.solr.cloud.autoscaling.AutoScaling$HttpCallbackListener", xyzListener.get("class").toString());
+ assertEquals("org.apache.solr.cloud.autoscaling.AutoScaling$HttpTriggerListener", xyzListener.get("class").toString());
String removeTriggerCommand = "{" +
"'remove-trigger' : {" +
@@ -422,7 +422,7 @@ public class AutoScalingHandlerTest extends SolrCloudTestCase {
"'trigger' : 'node_lost_trigger'," +
"'stage' : ['STARTED','ABORTED','SUCCEEDED']," +
"'beforeAction' : 'execute_plan'," +
- "'class' : 'org.apache.solr.cloud.autoscaling.AutoScaling$HttpCallbackListener'," +
+ "'class' : 'org.apache.solr.cloud.autoscaling.AutoScaling$HttpTriggerListener'," +
"'url' : 'http://xyz.com/on_node_lost?node={$LOST_NODE_NAME}'" +
"}" +
"}";
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a35a889/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 50f7e3d..a2beed4 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
@@ -43,7 +43,7 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
private static AtomicBoolean actionInitCalled = new AtomicBoolean(false);
private static AtomicBoolean actionCloseCalled = new AtomicBoolean(false);
- private AutoScaling.TriggerListener noFirstRunListener = event -> {
+ private AutoScaling.EventProcessor noFirstRunProcessor = event -> {
fail("Did not expect the listener to fire on first run!");
return true;
};
@@ -73,7 +73,7 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
Map<String, Object> props = createTriggerProps(waitForSeconds);
try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container)) {
- trigger.setProcessor(noFirstRunListener);
+ trigger.setProcessor(noFirstRunProcessor);
trigger.run();
JettySolrRunner newNode = cluster.startJettySolrRunner();
@@ -112,7 +112,7 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container)) {
final long waitTime = 2;
props.put("waitFor", waitTime);
- trigger.setProcessor(noFirstRunListener);
+ trigger.setProcessor(noFirstRunProcessor);
trigger.run();
JettySolrRunner newNode = cluster.startJettySolrRunner();
@@ -196,7 +196,7 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
CoreContainer container = cluster.getJettySolrRunners().get(0).getCoreContainer();
Map<String, Object> props = createTriggerProps(0);
try (NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container)) {
- trigger.setProcessor(noFirstRunListener);
+ trigger.setProcessor(noFirstRunProcessor);
trigger.run(); // starts tracking live nodes
JettySolrRunner newNode = cluster.startJettySolrRunner();
@@ -232,7 +232,7 @@ public class NodeAddedTriggerTest extends SolrCloudTestCase {
// add a new node but update the trigger before the waitFor period expires
// and assert that the new trigger still fires
NodeAddedTrigger trigger = new NodeAddedTrigger("node_added_trigger", props, container);
- trigger.setProcessor(noFirstRunListener);
+ trigger.setProcessor(noFirstRunProcessor);
trigger.run();
JettySolrRunner newNode = cluster.startJettySolrRunner();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/6a35a889/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 bb3339c..2492ce2 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
@@ -43,7 +43,7 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
private static AtomicBoolean actionInitCalled = new AtomicBoolean(false);
private static AtomicBoolean actionCloseCalled = new AtomicBoolean(false);
- private AutoScaling.TriggerListener noFirstRunListener = event -> {
+ private AutoScaling.EventProcessor noFirstRunProcessor = event -> {
fail("Did not expect the listener to fire on first run!");
return true;
};
@@ -74,7 +74,7 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
Map<String, Object> props = createTriggerProps(waitForSeconds);
try (NodeLostTrigger trigger = new NodeLostTrigger("node_lost_trigger", props, container)) {
- trigger.setProcessor(noFirstRunListener);
+ trigger.setProcessor(noFirstRunProcessor);
trigger.run();
String lostNodeName = cluster.getJettySolrRunner(1).getNodeName();
cluster.stopJettySolrRunner(1);
@@ -115,7 +115,7 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
try (NodeLostTrigger trigger = new NodeLostTrigger("node_lost_trigger", props, container)) {
final long waitTime = 2;
props.put("waitFor", waitTime);
- trigger.setProcessor(noFirstRunListener);
+ trigger.setProcessor(noFirstRunProcessor);
trigger.run();
JettySolrRunner lostNode = cluster.getJettySolrRunner(1);
@@ -210,7 +210,7 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
CoreContainer container = cluster.getJettySolrRunners().get(0).getCoreContainer();
Map<String, Object> props = createTriggerProps(0);
try (NodeLostTrigger trigger = new NodeLostTrigger("node_added_trigger", props, container)) {
- trigger.setProcessor(noFirstRunListener);
+ trigger.setProcessor(noFirstRunProcessor);
JettySolrRunner newNode = cluster.startJettySolrRunner();
cluster.waitForAllNodes(5);
@@ -263,7 +263,7 @@ public class NodeLostTriggerTest extends SolrCloudTestCase {
// and assert that the new trigger still fires
NodeLostTrigger trigger = new NodeLostTrigger("node_lost_trigger", props, container);
- trigger.setProcessor(noFirstRunListener);
+ trigger.setProcessor(noFirstRunProcessor);
trigger.run();
// stop the newly created node