You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@brooklyn.apache.org by al...@apache.org on 2017/11/10 10:59:20 UTC
[04/11] brooklyn-server git commit: move to `failover` package
instead of `ha` and mark beta
move to `failover` package instead of `ha` and mark beta
Project: http://git-wip-us.apache.org/repos/asf/brooklyn-server/repo
Commit: http://git-wip-us.apache.org/repos/asf/brooklyn-server/commit/0191d90e
Tree: http://git-wip-us.apache.org/repos/asf/brooklyn-server/tree/0191d90e
Diff: http://git-wip-us.apache.org/repos/asf/brooklyn-server/diff/0191d90e
Branch: refs/heads/master
Commit: 0191d90e2be59f25dcfa19bcbbd53c78d08be2d8
Parents: 62706c6
Author: Alex Heneveld <al...@cloudsoftcorp.com>
Authored: Wed Nov 8 09:32:36 2017 +0000
Committer: Alex Heneveld <al...@cloudsoftcorp.com>
Committed: Wed Nov 8 09:32:36 2017 +0000
----------------------------------------------------------------------
.../test/policy/failover/ElectPrimaryTest.java | 379 ++++++++++++++++
.../policies/ha/brooklyn/ElectPrimaryTest.java | 379 ----------------
.../policy/failover/ElectPrimaryConfig.java | 85 ++++
.../policy/failover/ElectPrimaryEffector.java | 442 +++++++++++++++++++
.../policy/failover/ElectPrimaryPolicy.java | 225 ++++++++++
.../policy/failover/PrimaryRunningEnricher.java | 75 ++++
.../failover/PropagatePrimaryEnricher.java | 204 +++++++++
.../brooklyn/policy/ha/ElectPrimaryConfig.java | 82 ----
.../policy/ha/ElectPrimaryEffector.java | 440 ------------------
.../brooklyn/policy/ha/ElectPrimaryPolicy.java | 223 ----------
.../policy/ha/PrimaryRunningEnricher.java | 74 ----
.../policy/ha/PropagatePrimaryEnricher.java | 204 ---------
.../policy/failover/ElectPrimaryTest.java | 26 ++
.../brooklyn/policy/ha/ElectPrimaryTest.java | 26 --
.../failover/elect-primary-propagate-test.yaml | 37 ++
.../elect-primary-selection-mode-test.yaml | 33 ++
.../failover/elect-primary-simple-test.yaml | 28 ++
.../policy/ha/elect-primary-propagate-test.yaml | 37 --
.../ha/elect-primary-selection-mode-test.yaml | 33 --
.../policy/ha/elect-primary-simple-test.yaml | 28 --
20 files changed, 1534 insertions(+), 1526 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0191d90e/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/test/policy/failover/ElectPrimaryTest.java
----------------------------------------------------------------------
diff --git a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/test/policy/failover/ElectPrimaryTest.java b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/test/policy/failover/ElectPrimaryTest.java
new file mode 100644
index 0000000..583da10
--- /dev/null
+++ b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/brooklyn/test/policy/failover/ElectPrimaryTest.java
@@ -0,0 +1,379 @@
+/*
+ * 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.brooklyn.camp.brooklyn.test.policy.failover;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.brooklyn.api.effector.Effector;
+import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.entity.EntityLocal;
+import org.apache.brooklyn.api.mgmt.Task;
+import org.apache.brooklyn.api.sensor.AttributeSensor;
+import org.apache.brooklyn.camp.brooklyn.AbstractYamlTest;
+import org.apache.brooklyn.config.ConfigKey;
+import org.apache.brooklyn.core.effector.AddEffector;
+import org.apache.brooklyn.core.effector.EffectorBody;
+import org.apache.brooklyn.core.effector.Effectors;
+import org.apache.brooklyn.core.entity.Attributes;
+import org.apache.brooklyn.core.entity.Dumper;
+import org.apache.brooklyn.core.entity.Entities;
+import org.apache.brooklyn.core.entity.EntityAsserts;
+import org.apache.brooklyn.core.entity.EntityPredicates;
+import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
+import org.apache.brooklyn.core.entity.trait.Startable;
+import org.apache.brooklyn.core.sensor.Sensors;
+import org.apache.brooklyn.core.test.entity.TestEntity;
+import org.apache.brooklyn.policy.failover.ElectPrimaryConfig;
+import org.apache.brooklyn.policy.failover.ElectPrimaryConfig.PrimaryDefaultSensorsAndEffectors;
+import org.apache.brooklyn.policy.failover.ElectPrimaryConfig.SelectionMode;
+import org.apache.brooklyn.policy.failover.ElectPrimaryEffector;
+import org.apache.brooklyn.test.Asserts;
+import org.apache.brooklyn.test.support.LoggingVerboseReporter;
+import org.apache.brooklyn.util.collections.MutableList;
+import org.apache.brooklyn.util.core.config.ConfigBag;
+import org.apache.brooklyn.util.exceptions.Exceptions;
+import org.apache.brooklyn.util.text.Strings;
+import org.apache.brooklyn.util.time.Duration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.ITestNGListener;
+import org.testng.TestNG;
+import org.testng.annotations.Test;
+import org.testng.reporters.FailedReporter;
+
+import com.google.common.base.Predicates;
+import com.google.common.base.Stopwatch;
+
+// TODO this test is in the CAMP package because YAML not available in policies package
+public class ElectPrimaryTest extends AbstractYamlTest {
+
+ private static final Logger log = LoggerFactory.getLogger(ElectPrimaryTest.class);
+
+ public static AttributeSensor<Entity> PRIMARY = PrimaryDefaultSensorsAndEffectors.PRIMARY;
+ public static ConfigKey<Double> WEIGHT_CONFIG = PrimaryDefaultSensorsAndEffectors.PRIMARY_WEIGHT_CONFIG;
+ public static AttributeSensor<Double> WEIGHT_SENSOR = PrimaryDefaultSensorsAndEffectors.PRIMARY_WEIGHT_SENSOR;
+
+ protected void setItemFromTestAsSimple() {
+ addCatalogItems(Strings.lines(
+ "brooklyn.catalog:",
+ " items:",
+ " - id: item-from-test",
+ " item:",
+ " type: "+TestEntity.class.getName()));
+ }
+
+ @Test
+ public void testTwoChildren() throws Exception {
+ setItemFromTestAsSimple();
+
+ Entity app = createAndStartApplication(loadYaml("classpath://org/apache/brooklyn/policy/failover/elect-primary-simple-test.yaml"));
+ EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.notNull());
+ log.info("Primary sensor is: "+app.sensors().get(PRIMARY));
+ }
+
+ @Test
+ public void testSetPreferredViaWeightConfigOnB() throws Exception {
+ runSetPreferredViaWeightConfigOnB();
+ }
+
+ public Entity runSetPreferredViaWeightConfigOnB() throws Exception {
+ setItemFromTestAsSimple();
+
+ Entity app = createAndStartApplication(loadYaml("classpath://org/apache/brooklyn/policy/failover/elect-primary-simple-test.yaml",
+ " brooklyn.config:",
+ " "+WEIGHT_CONFIG.getName()+": 1"));
+
+ EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.notNull());
+ Assert.assertEquals(app.sensors().get(PRIMARY).getDisplayName(), "b");
+ EntityAsserts.assertEntityHealthyEventually(app);
+ return app;
+ }
+
+ @Test
+ public void testSetDisallowedViaWeightConfigOnB() throws Exception {
+ setItemFromTestAsSimple();
+
+ Entity app = createAndStartApplication(loadYaml("classpath://org/apache/brooklyn/policy/failover/elect-primary-simple-test.yaml",
+ " brooklyn.config:",
+ " "+WEIGHT_CONFIG.getName()+": -1"));
+
+ EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.notNull());
+ Assert.assertEquals(app.sensors().get(PRIMARY).getDisplayName(), "a");
+ }
+
+ @Test
+ public void testFailover() throws Exception {
+ Entity app = runSetPreferredViaWeightConfigOnB();
+ Entity b = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("b"));
+
+ Entities.unmanage(b);
+ EntityAsserts.assertAttributeEventually(app, PRIMARY, EntityPredicates.displayNameEqualTo("a"));
+ }
+
+ @Test
+ public void testPropagateSensorsWithFailover() throws Exception {
+ setItemFromTestAsSimple();
+
+ Entity app = createAndStartApplication(loadYaml("classpath://org/apache/brooklyn/policy/failover/elect-primary-propagate-test.yaml"));
+
+ AttributeSensor<String> S1 = Sensors.newStringSensor("sens1");
+ AttributeSensor<String> S2 = Sensors.newStringSensor("sens2");
+ AttributeSensor<String> S3 = Sensors.newStringSensor("sens3");
+
+ Entity a = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("a"));
+ Entity b = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("b"));
+
+ EntityAsserts.assertAttributeEqualsEventually(b, S1, "hi1");
+
+ EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.notNull());
+ Assert.assertEquals(app.sensors().get(PRIMARY), b);
+ EntityAsserts.assertEntityHealthyEventually(app);
+ EntityAsserts.assertAttributeEqualsEventually(app, S1, "hi1");
+
+ b.sensors().set(S3, "hi-3-1");
+ b.sensors().set(S1, "hi2");
+ b.sensors().set(S2, "hi-2-1");
+ EntityAsserts.assertAttributeEqualsEventually(app, S1, "hi2");
+ EntityAsserts.assertAttributeEqualsEventually(app, S2, "hi-2-1");
+ // S3 not propagated
+ Asserts.assertNull(app.sensors().get(S3));
+
+ a.sensors().set(S1, "hi-a-1");
+ Entities.unmanage(b);
+ EntityAsserts.assertAttributeEqualsEventually(app, PRIMARY, a);
+ // s1 changes
+ EntityAsserts.assertAttributeEqualsEventually(app, S1, "hi-a-1");
+ // and s2 is cleared
+ EntityAsserts.assertAttributeEqualsEventually(app, S2, null);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Test
+ public void testFireCausesPromoteDemote() throws Exception {
+ Entity app = runSetPreferredViaWeightConfigOnB();
+
+ Collection<Entity> testEntities = mgmt().<Entity>lookupAll(EntityPredicates.hasInterfaceMatching(".*TestEntity"));
+ Asserts.assertSize(testEntities, 2);
+ promoteDemoteEffectorMessages.clear();
+ testEntities.forEach((entity) -> {
+ new MockPromoteDemoteEffector(ElectPrimaryConfig.PrimaryDefaultSensorsAndEffectors.PROMOTE).apply((EntityLocal) entity);
+ new MockPromoteDemoteEffector(ElectPrimaryConfig.PrimaryDefaultSensorsAndEffectors.DEMOTE).apply((EntityLocal) entity);
+ });
+
+ Entity a = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("a"));
+ Entity b = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("b"));
+
+ a.sensors().set(WEIGHT_SENSOR, 2.0d);
+ ((Startable)b).stop();
+ // or can do this; but it causes app to be on fire because it is requiring children healthy
+ // TODO what's the best way for a node to say it's allowed to have failed children?
+ // run a variant of this test using that, to ensure policy kicks in on down or on fire
+ //ServiceProblemsLogic.updateProblemsIndicator(b, "test", "force fire");
+
+ EntityAsserts.assertAttributeEventually(app, PRIMARY, EntityPredicates.displayNameEqualTo("a"));
+ // promotion and demotion should finish by the time the app is running again
+ EntityAsserts.assertEntityHealthyEventually(app);
+
+ // shouldn't be necessary as app is set starting before primary set to a, so
+ // above method checks not pass until promotion/demotion invocations have also completed
+ //Asserts.succeedsEventually(() -> Asserts.assertSize(promoteDemoteEffectorMessages, 2));
+
+ Asserts.assertTrue(promoteDemoteEffectorMessages.stream().anyMatch((s) -> s.matches("promote .*"+a.getId()+".* args=.*")),
+ "Missing/bad promotion message in: "+promoteDemoteEffectorMessages);
+ Asserts.assertTrue(promoteDemoteEffectorMessages.stream().anyMatch((s) -> s.matches("demote .*"+b.getId()+".* args=.*")),
+ "Missing/bad demotion message in: "+promoteDemoteEffectorMessages);
+
+ promoteDemoteEffectorMessages.clear();
+ }
+
+ List<String> promoteDemoteEffectorMessages = Collections.synchronizedList(MutableList.of());
+ private class MockPromoteDemoteEffector extends AddEffector {
+ public MockPromoteDemoteEffector(Effector<String> base) {
+ super(Effectors.effector(base).impl(new MockPromoteDemoteEffectorBody(base.getName())).build());
+ }
+ }
+ private class MockPromoteDemoteEffectorBody extends EffectorBody<String> {
+ private String message;
+ private MockPromoteDemoteEffectorBody(String message) {
+ this.message = message;
+ }
+ @Override
+ public String call(ConfigBag args) {
+ String result = message+" "+entity()+" args="+args.getAllConfigRaw();
+ promoteDemoteEffectorMessages.add(result);
+ return result;
+ }
+ }
+
+ // deferred:
+ // TODO support configurable parallelisation of promote/demote (in the code)
+ // TODO tests for timeout
+ // TODO tests for effector propagation in addition to sensor propagation
+
+ @Test
+ public void testSelectionModeStrictReelectWithPreference() throws Exception {
+ runSelectionModeTest(SelectionMode.STRICT, false);
+ }
+
+ @Test
+ public void testSelectionModeBestReelectWithPreference() throws Exception {
+ runSelectionModeTest(SelectionMode.BEST, false);
+ }
+
+ @Test
+ public void testSelectionModeFailoverReelectWithPreference() throws Exception {
+ runSelectionModeTest(SelectionMode.FAILOVER, false);
+ }
+
+ @Test(groups="Integration")
+ public void testSelectionModeStrictReelectWithPreferenceIntegration() throws Exception {
+ runSelectionModeTest(SelectionMode.STRICT, true);
+ }
+
+ @Test(groups="Integration")
+ public void testSelectionModeBestReelectWithPreferenceIntegration() throws Exception {
+ runSelectionModeTest(SelectionMode.BEST, true);
+ }
+
+ @Test(groups="Integration")
+ public void testSelectionModeFailoverReelectWithPreferenceIntegration() throws Exception {
+ runSelectionModeTest(SelectionMode.FAILOVER, true);
+ }
+
+ private void runSelectionModeTest(SelectionMode mode, boolean integration) throws Exception {
+ Entity app = null;
+ try {
+ setItemFromTestAsSimple();
+
+ app = createAndStartApplication(Strings.replaceAllNonRegex(
+ loadYaml("classpath://org/apache/brooklyn/policy/failover/elect-primary-selection-mode-test.yaml"),
+ "$$REPLACE$$", mode.name().toLowerCase()) );
+ EntityAsserts.assertEntityHealthyEventually(app);
+
+ Entity a = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("a"));
+ Entity b = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("b"));
+ // guarantee not running until has primary - via the enricher
+ EntityAsserts.assertAttributeEquals(app, PRIMARY, a);
+
+ b.sensors().set(WEIGHT_SENSOR, 2.0d);
+ if (mode==SelectionMode.FAILOVER) {
+ // primary won't change in failover mode
+ assertPrimaryUnchanged(app, a, integration);
+
+ // force election of best, overriding config mode
+ app.invoke(ElectPrimaryEffector.EFFECTOR, ConfigBag.newInstance().configure(
+ ElectPrimaryConfig.SELECTION_MODE, SelectionMode.BEST).getAllConfigRaw()).get();
+ }
+ EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.equalTo(b));
+
+ if (mode==SelectionMode.FAILOVER) {
+ // other tests don't apply to faliover
+ return;
+ }
+
+ b.sensors().set(WEIGHT_SENSOR, -1.0d);
+ // disabling b causes revert to a
+ EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.equalTo(a));
+
+ b.sensors().set(WEIGHT_SENSOR, 1.0d);
+ // now have a tie, should fail in strict, preserve a in best
+ if (mode==SelectionMode.STRICT) {
+ EntityAsserts.assertAttributeEqualsEventually(app, Attributes.SERVICE_STATE_ACTUAL, Lifecycle.ON_FIRE);
+ } else if (mode==SelectionMode.BEST) {
+ assertPrimaryUnchanged(app, a, integration);
+ EntityAsserts.assertEntityHealthyEventually(app);
+ }
+ try {
+ Object result = getLastElectionTask(app).get();
+ // one error was observed in above call outside of strict mode but couldn't tell what;
+ // refactored to say what the error was -- possibly task TC'd or maybe a CME;
+ // was running in a tight loop so likely we'll never see it again
+ if (mode==SelectionMode.STRICT) {
+ Asserts.shouldHaveFailedPreviously("Instead got: "+result);
+ }
+ } catch (Exception e) {
+ if (mode!=SelectionMode.STRICT) {
+ // tie should only fail in strict mode
+ Exceptions.propagate(e);
+ }
+ Asserts.expectedFailureContainsIgnoreCase(e,
+ "Cannot select primary", b.toString(), a.toString());
+ }
+
+ a.sensors().set(WEIGHT_SENSOR, -1d);
+ // disabling a reverts to b and makes it healthy
+ EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.equalTo(b));
+ EntityAsserts.assertAttributeEqualsEventually(app, Attributes.SERVICE_STATE_ACTUAL, Lifecycle.RUNNING);
+
+ // now disabling b causes failure
+ b.sensors().set(WEIGHT_SENSOR, -1d);
+ log.info("Waiting for no primary");
+ EntityAsserts.assertAttributeEqualsEventually(app, PRIMARY, null);
+ Object result = getLastElectionTask(app).get();
+ Asserts.assertStringContainsIgnoreCase(result.toString(), ElectPrimaryEffector.ResultCode.NO_PRIMARY_AVAILABLE.toString());
+
+ } catch (Throwable t) {
+ log.error("Failed: "+t, t);
+ Dumper.dumpInfo(app);
+ t.printStackTrace();
+ Exceptions.propagateIfFatal(t);
+ }
+ }
+
+ protected Task<?> getLastElectionTask(Entity app) {
+ // look up last task as noted in highlight and confirm it gave the right output
+ return mgmt().getExecutionManager().getTask(app.policies().iterator().next().getHighlights().get("lastScan").getTaskId());
+ }
+
+ private void assertPrimaryUnchanged(Entity app, Entity a, boolean integration) {
+ if (integration) {
+ EntityAsserts.assertAttributeEqualsContinually(app, PRIMARY, a);
+ } else {
+ EntityAsserts.assertAttributeEquals(app, PRIMARY, a);
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ int count = -1;
+ Stopwatch sw = Stopwatch.createStarted();
+ while (++count<100) {
+ log.info("new test run\n\n\nTEST RUN "+count+"\n");
+
+// ElectPrimaryTest t = new ElectPrimaryTest();
+// t.setUp();
+// t.testFireCausesPromoteDemote();
+// t.tearDown();
+
+ TestNG testNG = new TestNG();
+ testNG.setTestClasses(new Class[] { ElectPrimaryTest.class });
+ testNG.addListener((ITestNGListener)new LoggingVerboseReporter());
+ FailedReporter failedReporter = new FailedReporter();
+ testNG.addListener((ITestNGListener)failedReporter);
+ testNG.run();
+ if (!failedReporter.getFailedTests().isEmpty()) {
+ log.error("Failures: "+failedReporter.getFailedTests());
+ System.exit(1);
+ }
+ }
+ log.info("\n\nCompleted "+count+" runs in "+Duration.of(sw));
+ }
+}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0191d90e/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/policies/ha/brooklyn/ElectPrimaryTest.java
----------------------------------------------------------------------
diff --git a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/policies/ha/brooklyn/ElectPrimaryTest.java b/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/policies/ha/brooklyn/ElectPrimaryTest.java
deleted file mode 100644
index 638fba8..0000000
--- a/camp/camp-brooklyn/src/test/java/org/apache/brooklyn/camp/policies/ha/brooklyn/ElectPrimaryTest.java
+++ /dev/null
@@ -1,379 +0,0 @@
-/*
- * 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.brooklyn.camp.policies.ha.brooklyn;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-
-import org.apache.brooklyn.api.effector.Effector;
-import org.apache.brooklyn.api.entity.Entity;
-import org.apache.brooklyn.api.entity.EntityLocal;
-import org.apache.brooklyn.api.mgmt.Task;
-import org.apache.brooklyn.api.sensor.AttributeSensor;
-import org.apache.brooklyn.camp.brooklyn.AbstractYamlTest;
-import org.apache.brooklyn.config.ConfigKey;
-import org.apache.brooklyn.core.effector.AddEffector;
-import org.apache.brooklyn.core.effector.EffectorBody;
-import org.apache.brooklyn.core.effector.Effectors;
-import org.apache.brooklyn.core.entity.Attributes;
-import org.apache.brooklyn.core.entity.Dumper;
-import org.apache.brooklyn.core.entity.Entities;
-import org.apache.brooklyn.core.entity.EntityAsserts;
-import org.apache.brooklyn.core.entity.EntityPredicates;
-import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
-import org.apache.brooklyn.core.entity.trait.Startable;
-import org.apache.brooklyn.core.sensor.Sensors;
-import org.apache.brooklyn.core.test.entity.TestEntity;
-import org.apache.brooklyn.policy.ha.ElectPrimaryConfig;
-import org.apache.brooklyn.policy.ha.ElectPrimaryConfig.PrimaryDefaultSensorsAndEffectors;
-import org.apache.brooklyn.policy.ha.ElectPrimaryConfig.SelectionMode;
-import org.apache.brooklyn.policy.ha.ElectPrimaryEffector;
-import org.apache.brooklyn.test.Asserts;
-import org.apache.brooklyn.test.support.LoggingVerboseReporter;
-import org.apache.brooklyn.util.collections.MutableList;
-import org.apache.brooklyn.util.core.config.ConfigBag;
-import org.apache.brooklyn.util.exceptions.Exceptions;
-import org.apache.brooklyn.util.text.Strings;
-import org.apache.brooklyn.util.time.Duration;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testng.Assert;
-import org.testng.ITestNGListener;
-import org.testng.TestNG;
-import org.testng.annotations.Test;
-import org.testng.reporters.FailedReporter;
-
-import com.google.common.base.Predicates;
-import com.google.common.base.Stopwatch;
-
-// TODO this test is in the CAMP package because YAML not available in policies package
-public class ElectPrimaryTest extends AbstractYamlTest {
-
- private static final Logger log = LoggerFactory.getLogger(ElectPrimaryTest.class);
-
- public static AttributeSensor<Entity> PRIMARY = PrimaryDefaultSensorsAndEffectors.PRIMARY;
- public static ConfigKey<Double> WEIGHT_CONFIG = PrimaryDefaultSensorsAndEffectors.PRIMARY_WEIGHT_CONFIG;
- public static AttributeSensor<Double> WEIGHT_SENSOR = PrimaryDefaultSensorsAndEffectors.PRIMARY_WEIGHT_SENSOR;
-
- protected void setItemFromTestAsSimple() {
- addCatalogItems(Strings.lines(
- "brooklyn.catalog:",
- " items:",
- " - id: item-from-test",
- " item:",
- " type: "+TestEntity.class.getName()));
- }
-
- @Test
- public void testTwoChildren() throws Exception {
- setItemFromTestAsSimple();
-
- Entity app = createAndStartApplication(loadYaml("classpath://org/apache/brooklyn/policy/ha/elect-primary-simple-test.yaml"));
- EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.notNull());
- log.info("Primary sensor is: "+app.sensors().get(PRIMARY));
- }
-
- @Test
- public void testSetPreferredViaWeightConfigOnB() throws Exception {
- runSetPreferredViaWeightConfigOnB();
- }
-
- public Entity runSetPreferredViaWeightConfigOnB() throws Exception {
- setItemFromTestAsSimple();
-
- Entity app = createAndStartApplication(loadYaml("classpath://org/apache/brooklyn/policy/ha/elect-primary-simple-test.yaml",
- " brooklyn.config:",
- " "+WEIGHT_CONFIG.getName()+": 1"));
-
- EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.notNull());
- Assert.assertEquals(app.sensors().get(PRIMARY).getDisplayName(), "b");
- EntityAsserts.assertEntityHealthyEventually(app);
- return app;
- }
-
- @Test
- public void testSetDisallowedViaWeightConfigOnB() throws Exception {
- setItemFromTestAsSimple();
-
- Entity app = createAndStartApplication(loadYaml("classpath://org/apache/brooklyn/policy/ha/elect-primary-simple-test.yaml",
- " brooklyn.config:",
- " "+WEIGHT_CONFIG.getName()+": -1"));
-
- EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.notNull());
- Assert.assertEquals(app.sensors().get(PRIMARY).getDisplayName(), "a");
- }
-
- @Test
- public void testFailover() throws Exception {
- Entity app = runSetPreferredViaWeightConfigOnB();
- Entity b = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("b"));
-
- Entities.unmanage(b);
- EntityAsserts.assertAttributeEventually(app, PRIMARY, EntityPredicates.displayNameEqualTo("a"));
- }
-
- @Test
- public void testPropagateSensorsWithFailover() throws Exception {
- setItemFromTestAsSimple();
-
- Entity app = createAndStartApplication(loadYaml("classpath://org/apache/brooklyn/policy/ha/elect-primary-propagate-test.yaml"));
-
- AttributeSensor<String> S1 = Sensors.newStringSensor("sens1");
- AttributeSensor<String> S2 = Sensors.newStringSensor("sens2");
- AttributeSensor<String> S3 = Sensors.newStringSensor("sens3");
-
- Entity a = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("a"));
- Entity b = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("b"));
-
- EntityAsserts.assertAttributeEqualsEventually(b, S1, "hi1");
-
- EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.notNull());
- Assert.assertEquals(app.sensors().get(PRIMARY), b);
- EntityAsserts.assertEntityHealthyEventually(app);
- EntityAsserts.assertAttributeEqualsEventually(app, S1, "hi1");
-
- b.sensors().set(S3, "hi-3-1");
- b.sensors().set(S1, "hi2");
- b.sensors().set(S2, "hi-2-1");
- EntityAsserts.assertAttributeEqualsEventually(app, S1, "hi2");
- EntityAsserts.assertAttributeEqualsEventually(app, S2, "hi-2-1");
- // S3 not propagated
- Asserts.assertNull(app.sensors().get(S3));
-
- a.sensors().set(S1, "hi-a-1");
- Entities.unmanage(b);
- EntityAsserts.assertAttributeEqualsEventually(app, PRIMARY, a);
- // s1 changes
- EntityAsserts.assertAttributeEqualsEventually(app, S1, "hi-a-1");
- // and s2 is cleared
- EntityAsserts.assertAttributeEqualsEventually(app, S2, null);
- }
-
- @SuppressWarnings("deprecation")
- @Test
- public void testFireCausesPromoteDemote() throws Exception {
- Entity app = runSetPreferredViaWeightConfigOnB();
-
- Collection<Entity> testEntities = mgmt().<Entity>lookupAll(EntityPredicates.hasInterfaceMatching(".*TestEntity"));
- Asserts.assertSize(testEntities, 2);
- promoteDemoteEffectorMessages.clear();
- testEntities.forEach((entity) -> {
- new MockPromoteDemoteEffector(ElectPrimaryConfig.PrimaryDefaultSensorsAndEffectors.PROMOTE).apply((EntityLocal) entity);
- new MockPromoteDemoteEffector(ElectPrimaryConfig.PrimaryDefaultSensorsAndEffectors.DEMOTE).apply((EntityLocal) entity);
- });
-
- Entity a = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("a"));
- Entity b = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("b"));
-
- a.sensors().set(WEIGHT_SENSOR, 2.0d);
- ((Startable)b).stop();
- // or can do this; but it causes app to be on fire because it is requiring children healthy
- // TODO what's the best way for a node to say it's allowed to have failed children?
- // run a variant of this test using that, to ensure policy kicks in on down or on fire
- //ServiceProblemsLogic.updateProblemsIndicator(b, "test", "force fire");
-
- EntityAsserts.assertAttributeEventually(app, PRIMARY, EntityPredicates.displayNameEqualTo("a"));
- // promotion and demotion should finish by the time the app is running again
- EntityAsserts.assertEntityHealthyEventually(app);
-
- // shouldn't be necessary as app is set starting before primary set to a, so
- // above method checks not pass until promotion/demotion invocations have also completed
- //Asserts.succeedsEventually(() -> Asserts.assertSize(promoteDemoteEffectorMessages, 2));
-
- Asserts.assertTrue(promoteDemoteEffectorMessages.stream().anyMatch((s) -> s.matches("promote .*"+a.getId()+".* args=.*")),
- "Missing/bad promotion message in: "+promoteDemoteEffectorMessages);
- Asserts.assertTrue(promoteDemoteEffectorMessages.stream().anyMatch((s) -> s.matches("demote .*"+b.getId()+".* args=.*")),
- "Missing/bad demotion message in: "+promoteDemoteEffectorMessages);
-
- promoteDemoteEffectorMessages.clear();
- }
-
- List<String> promoteDemoteEffectorMessages = Collections.synchronizedList(MutableList.of());
- private class MockPromoteDemoteEffector extends AddEffector {
- public MockPromoteDemoteEffector(Effector<String> base) {
- super(Effectors.effector(base).impl(new MockPromoteDemoteEffectorBody(base.getName())).build());
- }
- }
- private class MockPromoteDemoteEffectorBody extends EffectorBody<String> {
- private String message;
- private MockPromoteDemoteEffectorBody(String message) {
- this.message = message;
- }
- @Override
- public String call(ConfigBag args) {
- String result = message+" "+entity()+" args="+args.getAllConfigRaw();
- promoteDemoteEffectorMessages.add(result);
- return result;
- }
- }
-
- // deferred:
- // TODO support configurable parallelisation of promote/demote (in the code)
- // TODO tests for timeout
- // TODO tests for effector propagation in addition to sensor propagation
-
- @Test
- public void testSelectionModeStrictReelectWithPreference() throws Exception {
- runSelectionModeTest(SelectionMode.STRICT, false);
- }
-
- @Test
- public void testSelectionModeBestReelectWithPreference() throws Exception {
- runSelectionModeTest(SelectionMode.BEST, false);
- }
-
- @Test
- public void testSelectionModeFailoverReelectWithPreference() throws Exception {
- runSelectionModeTest(SelectionMode.FAILOVER, false);
- }
-
- @Test(groups="Integration")
- public void testSelectionModeStrictReelectWithPreferenceIntegration() throws Exception {
- runSelectionModeTest(SelectionMode.STRICT, true);
- }
-
- @Test(groups="Integration")
- public void testSelectionModeBestReelectWithPreferenceIntegration() throws Exception {
- runSelectionModeTest(SelectionMode.BEST, true);
- }
-
- @Test(groups="Integration")
- public void testSelectionModeFailoverReelectWithPreferenceIntegration() throws Exception {
- runSelectionModeTest(SelectionMode.FAILOVER, true);
- }
-
- private void runSelectionModeTest(SelectionMode mode, boolean integration) throws Exception {
- Entity app = null;
- try {
- setItemFromTestAsSimple();
-
- app = createAndStartApplication(Strings.replaceAllNonRegex(
- loadYaml("classpath://org/apache/brooklyn/policy/ha/elect-primary-selection-mode-test.yaml"),
- "$$REPLACE$$", mode.name().toLowerCase()) );
- EntityAsserts.assertEntityHealthyEventually(app);
-
- Entity a = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("a"));
- Entity b = (Entity)mgmt().<Entity>lookup(EntityPredicates.displayNameEqualTo("b"));
- // guarantee not running until has primary - via the enricher
- EntityAsserts.assertAttributeEquals(app, PRIMARY, a);
-
- b.sensors().set(WEIGHT_SENSOR, 2.0d);
- if (mode==SelectionMode.FAILOVER) {
- // primary won't change in failover mode
- assertPrimaryUnchanged(app, a, integration);
-
- // force election of best, overriding config mode
- app.invoke(ElectPrimaryEffector.EFFECTOR, ConfigBag.newInstance().configure(
- ElectPrimaryConfig.SELECTION_MODE, SelectionMode.BEST).getAllConfigRaw()).get();
- }
- EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.equalTo(b));
-
- if (mode==SelectionMode.FAILOVER) {
- // other tests don't apply to faliover
- return;
- }
-
- b.sensors().set(WEIGHT_SENSOR, -1.0d);
- // disabling b causes revert to a
- EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.equalTo(a));
-
- b.sensors().set(WEIGHT_SENSOR, 1.0d);
- // now have a tie, should fail in strict, preserve a in best
- if (mode==SelectionMode.STRICT) {
- EntityAsserts.assertAttributeEqualsEventually(app, Attributes.SERVICE_STATE_ACTUAL, Lifecycle.ON_FIRE);
- } else if (mode==SelectionMode.BEST) {
- assertPrimaryUnchanged(app, a, integration);
- EntityAsserts.assertEntityHealthyEventually(app);
- }
- try {
- Object result = getLastElectionTask(app).get();
- // one error was observed in above call outside of strict mode but couldn't tell what;
- // refactored to say what the error was -- possibly task TC'd or maybe a CME;
- // was running in a tight loop so likely we'll never see it again
- if (mode==SelectionMode.STRICT) {
- Asserts.shouldHaveFailedPreviously("Instead got: "+result);
- }
- } catch (Exception e) {
- if (mode!=SelectionMode.STRICT) {
- // tie should only fail in strict mode
- Exceptions.propagate(e);
- }
- Asserts.expectedFailureContainsIgnoreCase(e,
- "Cannot select primary", b.toString(), a.toString());
- }
-
- a.sensors().set(WEIGHT_SENSOR, -1d);
- // disabling a reverts to b and makes it healthy
- EntityAsserts.assertAttributeEventually(app, PRIMARY, Predicates.equalTo(b));
- EntityAsserts.assertAttributeEqualsEventually(app, Attributes.SERVICE_STATE_ACTUAL, Lifecycle.RUNNING);
-
- // now disabling b causes failure
- b.sensors().set(WEIGHT_SENSOR, -1d);
- log.info("Waiting for no primary");
- EntityAsserts.assertAttributeEqualsEventually(app, PRIMARY, null);
- Object result = getLastElectionTask(app).get();
- Asserts.assertStringContainsIgnoreCase(result.toString(), ElectPrimaryEffector.ResultCode.NO_PRIMARY_AVAILABLE.toString());
-
- } catch (Throwable t) {
- log.error("Failed: "+t, t);
- Dumper.dumpInfo(app);
- t.printStackTrace();
- Exceptions.propagateIfFatal(t);
- }
- }
-
- protected Task<?> getLastElectionTask(Entity app) {
- // look up last task as noted in highlight and confirm it gave the right output
- return mgmt().getExecutionManager().getTask(app.policies().iterator().next().getHighlights().get("lastScan").getTaskId());
- }
-
- private void assertPrimaryUnchanged(Entity app, Entity a, boolean integration) {
- if (integration) {
- EntityAsserts.assertAttributeEqualsContinually(app, PRIMARY, a);
- } else {
- EntityAsserts.assertAttributeEquals(app, PRIMARY, a);
- }
- }
-
- public static void main(String[] args) throws Exception {
- int count = -1;
- Stopwatch sw = Stopwatch.createStarted();
- while (++count<100) {
- log.info("new test run\n\n\nTEST RUN "+count+"\n");
-
-// ElectPrimaryTest t = new ElectPrimaryTest();
-// t.setUp();
-// t.testFireCausesPromoteDemote();
-// t.tearDown();
-
- TestNG testNG = new TestNG();
- testNG.setTestClasses(new Class[] { ElectPrimaryTest.class });
- testNG.addListener((ITestNGListener)new LoggingVerboseReporter());
- FailedReporter failedReporter = new FailedReporter();
- testNG.addListener((ITestNGListener)failedReporter);
- testNG.run();
- if (!failedReporter.getFailedTests().isEmpty()) {
- log.error("Failures: "+failedReporter.getFailedTests());
- System.exit(1);
- }
- }
- log.info("\n\nCompleted "+count+" runs in "+Duration.of(sw));
- }
-}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0191d90e/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryConfig.java
----------------------------------------------------------------------
diff --git a/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryConfig.java b/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryConfig.java
new file mode 100644
index 0000000..e9f2803
--- /dev/null
+++ b/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryConfig.java
@@ -0,0 +1,85 @@
+/*
+ * 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.brooklyn.policy.failover;
+
+import org.apache.brooklyn.api.effector.Effector;
+import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.sensor.AttributeSensor;
+import org.apache.brooklyn.config.ConfigKey;
+import org.apache.brooklyn.core.config.ConfigKeys;
+import org.apache.brooklyn.core.effector.Effectors;
+import org.apache.brooklyn.core.sensor.Sensors;
+import org.apache.brooklyn.util.time.Duration;
+
+import com.google.common.annotations.Beta;
+
+@Beta
+public interface ElectPrimaryConfig {
+
+ public enum TargetMode { CHILDREN, MEMBERS, AUTO }
+ public static final ConfigKey<TargetMode> TARGET_MODE = ConfigKeys.builder(TargetMode.class, "primary.target.mode")
+ .description("where should the policy look for primary candidates; one of 'children', 'members', or 'auto' (members if it's a group)")
+ .defaultValue(TargetMode.AUTO)
+ .build();
+
+ public enum SelectionMode { FAILOVER, BEST, STRICT }
+ public static final ConfigKey<SelectionMode> SELECTION_MODE = ConfigKeys.builder(SelectionMode.class, "primary.selection.mode")
+ .description("under what circumstances should the primary change: `failover` to change only if an existing primary is unhealthy, `best` to change so one with the highest weight is always selected, or `strict` to act as `best` but fail if several advertise the highest weight (for use when the weight sensor is updated by the nodes and should tell us unambiguously who was elected)")
+ .defaultValue(SelectionMode.FAILOVER)
+ .build();
+
+ public static final ConfigKey<Duration> BEST_WAIT_TIMEOUT = ConfigKeys.newDurationConfigKey("primary.stopped.wait.timeout",
+ "if the highest-ranking primary is not starting, the effector will wait this long for it to be starting before picking a less highly-weighted primary; "
+ + "default 5s, typically long enough to avoid races where multiple children are started concurrently but they complete extremely quickly and one completes before a better one starts, "
+ + "the short duration is sufficient for the theoretical best to become waiting where the `primary.starting.wait.timeout` applies",
+ Duration.seconds(5));
+
+ public static final ConfigKey<Duration> BEST_STARTING_WAIT_TIMEOUT = ConfigKeys.newDurationConfigKey("primary.starting.wait.timeout",
+ "if the highest-ranking primary is starting, the effector will wait this long for it to be running before picking a less highly-weighted primary "
+ + "(or in the case of `strict` before failing if there are ties); "
+ + "default 5m, typically long enough to avoid races where multiple children are started and a sub-optimal one comes online before the best one",
+ Duration.minutes(5));
+
+ public static final ConfigKey<String> PRIMARY_SENSOR_NAME = ConfigKeys.newStringConfigKey("primary.sensor.name",
+ "name of sensor to publish, defaulting to 'primary'",
+ PrimaryDefaultSensorsAndEffectors.PRIMARY.getName());
+
+ public static final ConfigKey<String> PRIMARY_WEIGHT_NAME = ConfigKeys.newStringConfigKey("primary.weight.name",
+ "config key or sensor to scan from candidate nodes to determine who should be primary",
+ PrimaryDefaultSensorsAndEffectors.PRIMARY_WEIGHT_SENSOR.getName());
+
+ public static final ConfigKey<String> PROMOTE_EFFECTOR_NAME = ConfigKeys.newStringConfigKey("primary.promote.effector.name",
+ "effector to invoke on promotion, default `promote` and with no error if not present (but if set explicitly it will cause an error if not present)",
+ PrimaryDefaultSensorsAndEffectors.PROMOTE.getName());
+
+ public static final ConfigKey<String> DEMOTE_EFFECTOR_NAME = ConfigKeys.newStringConfigKey("primary.demote.effector.name",
+ "effector to invoke on demotion, default `demote` and with no error if not present (but if set explicitly it will cause an error if not present)",
+ PrimaryDefaultSensorsAndEffectors.DEMOTE.getName());
+
+
+ public interface PrimaryDefaultSensorsAndEffectors {
+ public static final AttributeSensor<Entity> PRIMARY = Sensors.newSensor(Entity.class, "primary");
+ public static final ConfigKey<Double> PRIMARY_WEIGHT_CONFIG = ConfigKeys.newDoubleConfigKey("ha.primary.weight");
+ public static final AttributeSensor<Double> PRIMARY_WEIGHT_SENSOR = Sensors.newDoubleSensor("ha.primary.weight");
+
+ public static final Effector<String> PROMOTE = Effectors.effector(String.class, "promote").buildAbstract();
+ public static final Effector<String> DEMOTE = Effectors.effector(String.class, "demote").buildAbstract();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0191d90e/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryEffector.java
----------------------------------------------------------------------
diff --git a/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryEffector.java b/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryEffector.java
new file mode 100644
index 0000000..aa28f1a
--- /dev/null
+++ b/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryEffector.java
@@ -0,0 +1,442 @@
+/*
+ * 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.brooklyn.policy.failover;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+import org.apache.brooklyn.api.effector.Effector;
+import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.entity.EntityInitializer;
+import org.apache.brooklyn.api.entity.Group;
+import org.apache.brooklyn.api.mgmt.Task;
+import org.apache.brooklyn.core.config.ConfigKeys;
+import org.apache.brooklyn.core.effector.EffectorBody;
+import org.apache.brooklyn.core.effector.EffectorTasks.EffectorBodyTaskFactory;
+import org.apache.brooklyn.core.effector.Effectors;
+import org.apache.brooklyn.core.entity.Attributes;
+import org.apache.brooklyn.core.entity.Entities;
+import org.apache.brooklyn.core.entity.EntityInternal;
+import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
+import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic;
+import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.ServiceProblemsLogic;
+import org.apache.brooklyn.core.sensor.Sensors;
+import org.apache.brooklyn.util.collections.MutableList;
+import org.apache.brooklyn.util.collections.MutableMap;
+import org.apache.brooklyn.util.core.config.ConfigBag;
+import org.apache.brooklyn.util.core.task.DynamicTasks;
+import org.apache.brooklyn.util.core.task.Tasks;
+import org.apache.brooklyn.util.exceptions.Exceptions;
+import org.apache.brooklyn.util.exceptions.UserFacingException;
+import org.apache.brooklyn.util.time.Duration;
+import org.apache.brooklyn.util.time.Time;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.Beta;
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.Iterables;
+
+/**
+This effector will scan candidates among children or members to determine which should be noted as "primary".
+The primary is selected from service-up candidates based on a numeric weight as a sensor or config on the candidates
+(`ha.primary.weight`, unless overridden), with higher weights being preferred.
+In the case of ties, or a new candidate emerging with a weight higher than a current healthy primary,
+behaviour can be configured with `primary.selection.mode`.
+
+Returns a map containing a message, newPrimary, oldPrimary, and a {@link ResultCode} code.
+*/
+@Beta
+public class ElectPrimaryEffector implements EntityInitializer, ElectPrimaryConfig {
+
+ private static final Logger log = LoggerFactory.getLogger(ElectPrimaryEffector.class);
+
+ public static enum ResultCode { PRIMARY_UNCHANGED, NEW_PRIMARY_ELECTED, NO_PRIMARY_AVAILABLE }
+
+ public static final Effector<Object> EFFECTOR = Effectors.effector(Object.class, "electPrimary").
+ description("Scan to detect whether there is or should be a new primary").buildAbstract();
+
+ private final ConfigBag paramsCreationTime;
+
+ public ElectPrimaryEffector(ConfigBag params) {
+ this.paramsCreationTime = params;
+ }
+
+ public ElectPrimaryEffector(Map<String,String> params) {
+ this(ConfigBag.newInstance(params));
+ }
+
+
+ // wire up the entity to call the task factory to create the task on invocation
+
+ @Override
+ public void apply(@SuppressWarnings("deprecation") org.apache.brooklyn.api.entity.EntityLocal entity) {
+ ((EntityInternal)entity).getMutableEntityType().addEffector(makeEffector(paramsCreationTime));
+ }
+
+ public static Effector<Object> makeEffector(ConfigBag params) {
+ return Effectors.effector(EFFECTOR).impl(new EffectorBodyTaskFactory<Object>(new ElectPrimaryEffectorBody(params))).build();
+ }
+
+ protected static class ElectPrimaryEffectorBody extends EffectorBody<Object> {
+ private final ConfigBag paramsCreationTime;
+
+ public ElectPrimaryEffectorBody(ConfigBag paramsCreationTime) {
+ this.paramsCreationTime = paramsCreationTime;
+ }
+
+ // these are the actual tasks we do
+
+ @Override
+ public Object call(ConfigBag paramsInvocationTime) {
+ ConfigBag params = ConfigBag.newInstanceCopying(paramsCreationTime).copy(paramsInvocationTime);
+
+ try {
+ Entity newPrimary = DynamicTasks.queue("check primaries", new CheckPrimaries(params)).getUnchecked();
+
+ Entity currentActive = getCurrentActive(params);
+ if (newPrimary==null) {
+// If no primary can be found, the effector will:
+// * add a "primary-election" problem so that service state logic, if applicable, will know that the entity is unhealthy
+// * set service up false
+// * if the local entity is expected to be RUNNING, it will set actual state to ON_FIRE
+// * if the local entity has no expectation, it will set actual state to STOPPED
+// * demote any old primary
+ ServiceProblemsLogic.updateProblemsIndicator(entity(), "primary", "No primary could be found");
+ entity().sensors().set(Sensors.newSensor(Entity.class, params.get(PRIMARY_SENSOR_NAME)), null);
+ entity().sensors().set(Attributes.SERVICE_UP, false);
+ if (Lifecycle.RUNNING.equals( entity().getAttribute(Attributes.SERVICE_STATE_EXPECTED) )) {
+ entity().sensors().set(Attributes.SERVICE_STATE_ACTUAL, Lifecycle.ON_FIRE);
+ } else {
+ entity().sensors().set(Attributes.SERVICE_STATE_ACTUAL, Lifecycle.STOPPED);
+ }
+ DynamicTasks.queue(Tasks.create("demote "+currentActive, new Demote(params)) ).getUnchecked();
+ return MutableMap.of("code", ResultCode.NO_PRIMARY_AVAILABLE, "message", "No primary available", "primary", null);
+ }
+
+ if (newPrimary.equals(currentActive)) {
+ // If there is a primary and it is unchanged, the effector will end.
+ return MutableMap.of("code", ResultCode.PRIMARY_UNCHANGED, "message", "No change required", "primary", newPrimary);
+ }
+
+ log.info("Detected new primary "+newPrimary+" at "+entity()+" (previously had "+currentActive+")");
+// If a new primary is detected, the effector will:
+// * set the local entity to the STARTING state
+// * clear any "primary-election" problem
+// * publish the new primary in a sensor called `primary` (or the sensor set in `primary.sensor.name`)
+// * cancel any other ongoing promote calls, and if there is an ongoing demote call on the entity being promoted, cancel that also
+// * in parallel
+// * invoke `promote` (or the effector called `primary.promote.effector.name`) on the local entity or the entity being promoted
+// * invoke `demote` (or the effector called `primary.promote.effector.name`) on the local entity or the entity being demoted, if an entity is being demoted
+// * set service up true
+// * set the local entity to the RUNNING state
+
+ boolean wasRunning = entity().sensors().get(Attributes.SERVICE_STATE_ACTUAL) == Lifecycle.RUNNING;
+ if (wasRunning) {
+ log.debug("Transititioning "+entity()+" to starting while promoting/demoting");
+ ServiceStateLogic.setExpectedState(entity(), Lifecycle.STARTING);
+ }
+ ServiceProblemsLogic.clearProblemsIndicator(entity(), "primary");
+ entity().sensors().set(Sensors.newSensor(Entity.class, params.get(PRIMARY_SENSOR_NAME)), newPrimary);
+ try {
+ // TODO cancel other promote/demote calls
+
+ promoteAndDemote(params, currentActive, newPrimary);
+
+ log.debug("Promoted/demoted primary for "+entity()+", now setting service up "+(wasRunning ? "and running" : "(but not setting as 'running' because it wasn't 'running' before)"));
+ entity().sensors().set(Attributes.SERVICE_UP, true);
+ if (wasRunning) ServiceStateLogic.setExpectedState(entity(), Lifecycle.RUNNING);
+
+ } catch (Exception e) {
+ Exceptions.propagateIfFatal(e);
+ log.debug("Error promoting/demoting primary for "+entity()+" (rethrowing): "+e);
+ ServiceProblemsLogic.updateProblemsIndicator(entity(), "primary", e);
+ ServiceStateLogic.setExpectedStateRunningWithErrors(entity());
+ Exceptions.propagate(e);
+ }
+
+ return MutableMap.of("code", ResultCode.NEW_PRIMARY_ELECTED, "message", "New primary found", "primary", newPrimary);
+
+ } catch (Exception e) {
+ Exceptions.propagateIfFatal(e);
+
+ if (Entities.isNoLongerManaged(entity())) {
+ // ignore errors if shutting down
+ return "<no-longer-managed>";
+ }
+
+ Lifecycle expected = ServiceStateLogic.getExpectedState(entity());
+ if (expected==Lifecycle.RUNNING || expected==Lifecycle.STARTING) {
+ // including SelectionModeStrictFailed
+ log.warn("Error electing new primary at "+entity()+": "+Exceptions.collapseText(e));
+ ServiceProblemsLogic.updateProblemsIndicator(entity(), "primary", "Error electing primary: "+
+ Exceptions.collapseText(e));
+ entity().sensors().set(Attributes.SERVICE_STATE_ACTUAL, Lifecycle.ON_FIRE);
+
+ throw Exceptions.propagateAnnotated("Error electing primary (when "+expected.toString().toLowerCase()+")", e);
+ }
+
+ throw Exceptions.propagateAnnotated("Error electing primary (when not starting/running)", e);
+ }
+ }
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ protected void promoteAndDemote(ConfigBag params, Entity oldPrimary, Entity newPrimary) {
+ params.configureStringKey("oldPrimary", oldPrimary);
+ params.configureStringKey("newPrimary", newPrimary);
+ MutableList<Task> tasks = MutableList.<Task>of();
+
+ if (newPrimary!=null) tasks.append(Tasks.create("promote "+newPrimary, new Promote(params)));
+ else tasks.append(Tasks.warning("No new primary; nothing to promote", null, false));
+
+ if (oldPrimary!=null) tasks.append(Tasks.create("demote "+oldPrimary, new Demote(params)));
+ else tasks.append(Tasks.warning("No old primary; nothing to demote", null, false));
+
+ log.debug("Running "+tasks);
+ List<?> result = DynamicTasks.queue(Tasks.parallel("promote/demote", (List<Task<?>>)(List) tasks)).getUnchecked();
+ log.debug("Ran "+tasks+", results: "+result);
+ }
+
+ private Entity getCurrentActive(ConfigBag params) {
+ return entity().getAttribute(Sensors.newSensor(Entity.class, params.get(PRIMARY_SENSOR_NAME)));
+ }
+
+ protected class CheckPrimaries implements Callable<Entity> {
+ final ConfigBag params;
+ public CheckPrimaries(ConfigBag params) { this.params = params; }
+
+ @Override
+ public Entity call() throws Exception {
+ Stopwatch elapsedTime = Stopwatch.createStarted();
+ boolean extendedForNonViableRunning = false;
+ outer: while (true) {
+ TargetMode target = params.get(TARGET_MODE);
+ Iterable<Entity> candidates =
+ target==TargetMode.CHILDREN ? entity().getChildren() :
+ target==TargetMode.MEMBERS ? ((Group)entity()).getMembers() :
+ // auto - prefer members
+ entity() instanceof Group ? ((Group)entity()).getMembers() : entity().getChildren();
+
+ SelectionMode mode = params.get(SELECTION_MODE);
+ Entity currentActive = getCurrentActive(params);
+ if (mode==SelectionMode.FAILOVER && currentActive!=null && Iterables.contains(candidates, currentActive) && isViable(currentActive)) {
+ return currentActive;
+ }
+ // find best live and/or time to wait
+ Duration delayForBest = Duration.ZERO;
+ Duration period = Duration.millis(10);
+
+ if (candidates.iterator().hasNext()) {
+ List<WeightedEntity> weightedEntities = MutableList.of();
+ for (Entity candidate: candidates) {
+ weightedEntities.add(new WeightedEntity(candidate));
+ }
+ Collections.sort(weightedEntities);
+ WeightedEntity anyBestTheoreticalW = weightedEntities.iterator().next();
+ WeightedEntity bestLive = null;
+ for (WeightedEntity w: weightedEntities) {
+ if (w.score < -0.00001) {
+ // this is disallowed, as are all following
+ break;
+ }
+ if (Math.abs(anyBestTheoreticalW.score - w.score) >= 0.0000000001) {
+ if (bestLive==null && !delayForBest.isLongerThan(elapsedTime)) {
+ // no viable highest-score and can't wait any longer;
+ // take the next highest that we find
+ if (isViable(w.entity)) {
+ log.debug("Theoretical best primary at "+entity()+" ("+anyBestTheoreticalW+", maybe others) not available, using next best: "+w);
+ return w.entity;
+ } else {
+ continue;
+ }
+ } else {
+ // don't bother looking at non-highest scoring nodes, we know what to do
+ break;
+ }
+ }
+
+ // examining the highest scoring ones
+
+ if (isViable(w.entity)) {
+ log.debug("Viable best primary at "+entity()+" detected: "+w);
+ if (bestLive==null) {
+ bestLive = w;
+ } else if (mode==SelectionMode.STRICT) {
+ // in strict mode, should only have one node with the best score,
+ // so this is a failure. we might want to allow some grace if a
+ // method is promoting a new one by changing (non-atomically) which
+ // single entity has a weight 1 where others are 0.
+ // but preferred way to avoid that is to increment weight at new primary instead.
+ throw new SelectionModeStrictFailed(w.entity, bestLive.entity, bestLive.score);
+ } else if (w.entity.equals(currentActive)) {
+ // always prefer current active if it is best
+ // (safe to bail out here but we won't yet)
+ bestLive = w;
+ } else {
+ // two equally good bests, neither current active
+ // could prefer either but go with the first (no-op here)
+ }
+
+ } else {
+ // this best not viable - determine why and how long to wait
+ Lifecycle state = w.entity.getAttribute(Attributes.SERVICE_STATE_ACTUAL);
+ log.debug("Theoretical best primary at "+entity()+": "+w.entity+" "+state+" (not viable); may re-check");
+ if (state==Lifecycle.STARTING) {
+ delayForBest = Duration.max(delayForBest, entity().config().get(BEST_STARTING_WAIT_TIMEOUT));
+
+ } else if (state==Lifecycle.RUNNING) {
+
+ // running, not on fire, but not viable - either a race (has just become viable)
+ // or a coding error (caller did not set "up" correctly); give extra 5s if we haven't already
+ if (!extendedForNonViableRunning) {
+ delayForBest = Duration.max(delayForBest, Duration.of(elapsedTime).add(entity().config().get(BEST_WAIT_TIMEOUT)));
+ extendedForNonViableRunning = true;
+ }
+ } else {
+ delayForBest = Duration.max(delayForBest, entity().config().get(BEST_WAIT_TIMEOUT));
+ }
+ }
+ }
+
+ // finished looking at all nodes, or all theoretical bests and found one
+ if (bestLive!=null) {
+ // found a best live (preferring current if viable,
+ // but doesn't wait for current)
+ return bestLive.entity;
+ }
+ } else {
+ delayForBest = entity().config().get(BEST_WAIT_TIMEOUT);
+ }
+ Duration delay = delayForBest.subtract(Duration.of(elapsedTime));
+ if (delay.isPositive()) {
+ delay = Duration.min(delay, period);
+ period = Duration.min(Duration.ONE_SECOND, period.multiply(1.5));
+ log.debug("Delaying "+delay+" ("+delayForBest+" allowed, "+Duration.of(elapsedTime)+" elapsed) then rechecking for best primary at "+entity());
+ // there was a theoretical best that wasn't started
+ Time.sleep(delay);
+ continue outer;
+ }
+
+ // none viable or worth waiting for
+ return null;
+ }
+
+ }
+
+ private class WeightedEntity implements Comparable<WeightedEntity> {
+ public final Entity entity;
+ public final double score;
+
+ public WeightedEntity(Entity candidate) {
+ this.entity = candidate;
+ this.score = score(candidate);
+ }
+
+ @Override
+ public int compareTo(WeightedEntity o) {
+ double v = o.score - this.score;
+ return (v > 0.00000001 ? 1 : v<-0.00000001 ? -1 : 0);
+ }
+ @Override
+ public String toString() {
+ return entity+":"+score;
+ }
+ }
+
+ protected boolean isViable(Entity candidate) {
+ if (!Lifecycle.RUNNING.equals( candidate.getAttribute(Attributes.SERVICE_STATE_ACTUAL) )) return false;
+ if (!Boolean.TRUE.equals( candidate.getAttribute(Attributes.SERVICE_UP) )) return false;
+ if (score(candidate) <= -0.000000001) return false;
+ return true;
+ }
+
+ private double score(Entity candidate) {
+ Double s = candidate.getAttribute(Sensors.newDoubleSensor(params.get(PRIMARY_WEIGHT_NAME)));
+ if (s!=null) return s;
+ s = candidate.getConfig(ConfigKeys.newDoubleConfigKey(params.get(PRIMARY_WEIGHT_NAME)));
+ if (s!=null) return s;
+ return 0;
+ }
+ }
+
+ protected class Promote implements Callable<Object> {
+ final ConfigBag params;
+ public Promote(ConfigBag params) { this.params = params; }
+
+ @Override
+ public Object call() throws Exception {
+ String promoteEffectorName = params.get(PROMOTE_EFFECTOR_NAME);
+ Effector<?> eff = entity().getEffector(promoteEffectorName);
+ if (eff!=null) {
+ return DynamicTasks.queue( Effectors.invocation(entity(), eff, params) ).asTask().getUnchecked();
+ }
+ EntityInternal newPrimary = (EntityInternal)params.getStringKey("newPrimary");
+ if (newPrimary==null) {
+ return "Nothing to demote; no new primary";
+ }
+ eff = newPrimary.getEffector(promoteEffectorName);
+ if (eff!=null) {
+ return DynamicTasks.queue( Effectors.invocation(newPrimary, eff, params) ).asTask().getUnchecked();
+ }
+ return "No promotion effector '"+promoteEffectorName+"'; nothing to do";
+ }
+ }
+
+ protected class Demote implements Callable<Object> {
+ final ConfigBag params;
+ public Demote(ConfigBag params) { this.params = params; }
+
+ @Override
+ public Object call() throws Exception {
+ String demoteEffectorName = params.get(DEMOTE_EFFECTOR_NAME);
+ Effector<?> eff = entity().getEffector(demoteEffectorName);
+ if (eff!=null) {
+ return DynamicTasks.queue( Effectors.invocation(entity(), eff, params) ).asTask().getUnchecked();
+ }
+ EntityInternal oldPrimary = (EntityInternal)params.getStringKey("oldPrimary");
+ if (oldPrimary==null) {
+ return "Nothing to demote; no old primary";
+ }
+ eff = oldPrimary.getEffector(demoteEffectorName);
+ if (eff!=null) {
+ return DynamicTasks.queue( Effectors.invocation(oldPrimary, eff, params) ).asTask().getUnchecked();
+ }
+ return "No demotion effector '"+demoteEffectorName+"'; nothing to do";
+ }
+ }
+ }
+
+ private static class SelectionModeStrictFailed extends UserFacingException {
+ private static final long serialVersionUID = -6253854814553229953L;
+
+ // fields not needed at present
+// final Entity entity1;
+// final Entity entity2;
+// final double score;
+
+ public SelectionModeStrictFailed(Entity entity1, Entity entity2, double score) {
+ super("Cannot select primary in strict mode: entities "+entity1+" and "+entity2+" have same score "+score);
+// this.entity1 = entity1;
+// this.entity2 = entity2;
+// this.score = score;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0191d90e/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryPolicy.java
----------------------------------------------------------------------
diff --git a/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryPolicy.java b/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryPolicy.java
new file mode 100644
index 0000000..edcef22
--- /dev/null
+++ b/policy/src/main/java/org/apache/brooklyn/policy/failover/ElectPrimaryPolicy.java
@@ -0,0 +1,225 @@
+/*
+ * 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.brooklyn.policy.failover;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.entity.Group;
+import org.apache.brooklyn.api.mgmt.Task;
+import org.apache.brooklyn.api.sensor.AttributeSensor;
+import org.apache.brooklyn.api.sensor.EnricherSpec;
+import org.apache.brooklyn.api.sensor.Sensor;
+import org.apache.brooklyn.api.sensor.SensorEvent;
+import org.apache.brooklyn.api.sensor.SensorEventListener;
+import org.apache.brooklyn.config.ConfigKey;
+import org.apache.brooklyn.core.config.ConfigKeys;
+import org.apache.brooklyn.core.effector.Effectors;
+import org.apache.brooklyn.core.entity.AbstractEntity;
+import org.apache.brooklyn.core.entity.Attributes;
+import org.apache.brooklyn.core.entity.Entities;
+import org.apache.brooklyn.core.entity.EntityInternal;
+import org.apache.brooklyn.core.entity.StartableApplication;
+import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
+import org.apache.brooklyn.core.mgmt.BrooklynTaskTags;
+import org.apache.brooklyn.core.policy.AbstractPolicy;
+import org.apache.brooklyn.core.sensor.Sensors;
+import org.apache.brooklyn.entity.group.DynamicGroup;
+import org.apache.brooklyn.util.collections.MutableList;
+import org.apache.brooklyn.util.collections.QuorumCheck.QuorumChecks;
+import org.apache.brooklyn.util.core.task.DynamicTasks;
+import org.apache.brooklyn.util.exceptions.Exceptions;
+import org.apache.brooklyn.util.exceptions.UserFacingException;
+import org.apache.brooklyn.util.text.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.Beta;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
+import com.google.common.reflect.TypeToken;
+
+/**
+
+The ElectPrimaryPolicy acts to keep exactly one of its children or members as primary, promoting and demoting them when required.
+
+A simple use case is where we have two children, call them North and South, and we wish for North to be primary. If North fails, however, we want to promote and fail over to South. This can be done by:
+
+* adding this policy at the parent
+* setting ` ha.primary.weight` on North
+* optionally defining `promote` on North and South (if action is required there to promote it)
+* observing the `primary` sensor to see which is primary
+* optionally setting `propagate.primary.sensors: [ main.uri ]` to publish `main.uri` from whichever of North or South is active
+* optionally setting `primary.selection.mode: best` to switch back to North if it comes back online
+
+The policy works by listening for service-up changes in the target pool (children or members) and listening for `ha.primary.weight` sensor values from those elements.
+On any change, it invokes an effector to perform the primary election.
+By default, the effector invoked is `electPrimary`, but this can be changed with the `primary.election.effector` config key. If this effector does not exist, the policy will add a default behaviour using `ElectPrimaryEffector`. Details of the election are described in that effector, but to summarize, it will find an appropriate primary from the target pool and publish a sensor indicating who the new primary is.
+If the effector is not defined this policy will add one with the standard election behaviour
+({@link ElectPrimaryEffector}).
+That effector will also invoke `promote` and `demote` on the relevant entities.
+
+All the `primary.*` parameters accepted by that effector can be defined on this policy
+and will be passed to the effector, along with an `event` parameter indicating the sensor which triggered the election.
+
+If no quorum.up or quorum.running is set on the entity, both will be set to a constant 1.
+
+ */
+@Beta
+public class ElectPrimaryPolicy extends AbstractPolicy implements ElectPrimaryConfig {
+
+ private static final Logger log = LoggerFactory.getLogger(ElectPrimaryPolicy.class);
+
+ public static ConfigKey<String> EFFECTOR_NAME = ConfigKeys.newStringConfigKey("primary.election.effector",
+ "The effector to invoke to perform the scan; if not set, it will use electPrimary and create if necessary",
+ "electPrimary");
+
+ @SuppressWarnings("serial")
+ public static ConfigKey<Collection<? extends Object>> PROPAGATE_PRIMARY_SENSORS = ConfigKeys.newConfigKey(
+ new TypeToken<Collection<? extends Object>>() {},
+ "propagate.primary.sensors");
+
+ @Override
+ public void setEntity(@SuppressWarnings("deprecation") org.apache.brooklyn.api.entity.EntityLocal entity) {
+ super.setEntity(entity);
+
+ checkAndMaybeAddEffector(entity);
+ checkQuorums(entity);
+ addSubscriptions(entity);
+
+ Collection<? extends Object> sensorsToPropagate = config().get(PROPAGATE_PRIMARY_SENSORS);
+ if (sensorsToPropagate!=null) {
+ List<Sensor<?>> realSensors = MutableList.of();
+ for (Object s: sensorsToPropagate) {
+ if (s instanceof String) s = Sensors.newSensor(Object.class, (String) s);
+ if (s instanceof Sensor) {
+ realSensors.add((Sensor<?>)s);
+ } else {
+ throw new IllegalArgumentException("Config "+PROPAGATE_PRIMARY_SENSORS.getName()+" had invalid entry "
+ + "'"+s+"'; expected string or sensor");
+ }
+ }
+ entity.enrichers().add(EnricherSpec.create(PropagatePrimaryEnricher.class)
+ .configure(PropagatePrimaryEnricher.PROPAGATE_EFFECTORS, false)
+ .configure(PropagatePrimaryEnricher.PROPAGATING, realSensors));
+ }
+ }
+
+ protected void checkAndMaybeAddEffector(@SuppressWarnings("deprecation") org.apache.brooklyn.api.entity.EntityLocal entity) {
+ String effName = config().get(EFFECTOR_NAME);
+ if (((EntityInternal)entity).getEffector(effName)==null) {
+ // effector not defined
+ if (config().getRaw(EFFECTOR_NAME).isAbsent()) {
+ log.debug("No effector '"+effName+"' present at "+entity+"; creating default");
+ // if not set, we can create the default
+ new ElectPrimaryEffector(config().getBag()).apply(entity);
+
+ } else {
+ // otherwise it's an error, fail
+ throw new IllegalStateException("No such effector '"+effName+"' on "+entity);
+ }
+ }
+ }
+
+ private void checkQuorums(Entity entity) {
+ // set all quorums to 1 if not explicitly set
+ if ( ((ConfigurationSupportInternal)entity.config()).getRaw(StartableApplication.UP_QUORUM_CHECK).isAbsent() ) {
+ entity.config().set(StartableApplication.UP_QUORUM_CHECK, QuorumChecks.newInstance(1, 0.0, false));
+ }
+ if ( ((ConfigurationSupportInternal)entity.config()).getRaw(StartableApplication.RUNNING_QUORUM_CHECK).isAbsent() ) {
+ entity.config().set(StartableApplication.RUNNING_QUORUM_CHECK, QuorumChecks.newInstance(1, 0.0, false));
+ }
+ }
+
+ protected void addSubscriptions(Entity entity) {
+ String weightSensorName = config().get(PRIMARY_WEIGHT_NAME);
+ highlightTriggers("Listening for "+weightSensorName+" and service up, state on all children " + (entity instanceof Group ? "and members " : ""));
+
+ Change<Entity> candidateSetChange = new Change<Entity>();
+ subscriptions().subscribe(entity, AbstractEntity.CHILD_ADDED, candidateSetChange);
+ subscriptions().subscribe(entity, AbstractEntity.CHILD_REMOVED, candidateSetChange);
+ subscriptions().subscribe(entity, DynamicGroup.MEMBER_ADDED, candidateSetChange);
+ subscriptions().subscribe(entity, DynamicGroup.MEMBER_REMOVED, candidateSetChange);
+
+ Change<Boolean> candidateUpChange = new Change<Boolean>();
+ subscriptions().subscribeToChildren(entity, Attributes.SERVICE_UP, candidateUpChange);
+ if (entity instanceof Group) {
+ subscriptions().subscribeToMembers(((Group)entity), Attributes.SERVICE_UP, candidateUpChange);
+ }
+
+ Change<Lifecycle> candidateLifecycleChange = new Change<Lifecycle>();
+ subscriptions().subscribeToChildren(entity, Attributes.SERVICE_STATE_ACTUAL, candidateLifecycleChange);
+ if (entity instanceof Group) {
+ subscriptions().subscribeToMembers(((Group)entity), Attributes.SERVICE_STATE_ACTUAL, candidateLifecycleChange);
+ }
+
+ Change<Number> candidateWeightChange = new Change<Number>();
+ AttributeSensor<Number> weightSensor = Sensors.newSensor(Number.class, weightSensorName);
+ subscriptions().subscribeToChildren(entity, weightSensor, candidateWeightChange);
+ if (entity instanceof Group) {
+ subscriptions().subscribeToMembers(((Group)entity), weightSensor, candidateWeightChange);
+ }
+ }
+
+ public class Change<T> implements SensorEventListener<T> {
+ public void onEvent(SensorEvent<T> event) { rescan(event); }
+ }
+
+ public void rescan(SensorEvent<?> event) {
+ String code = null;
+ try {
+ String effName = config().get(EFFECTOR_NAME);
+ if (log.isTraceEnabled()) {
+ log.trace("Policy "+this+" got event "+event+"; triggering rescan with "+effName);
+ }
+ Task<?> task = Effectors.invocation(entity, Preconditions.checkNotNull( ((EntityInternal)entity).getEffector(effName) ), config().getBag()).asTask();
+ BrooklynTaskTags.addTagDynamically(task, BrooklynTaskTags.NON_TRANSIENT_TASK_TAG);
+
+ highlight("lastScan", "Running "+effName+" on "+event.getSensor().getName()+" from "+event.getSource().getId(), task);
+
+ Object result = DynamicTasks.get(task);
+ if (result instanceof Map) code = Strings.toString( ((Map<?,?>)result).get("code") );
+
+ if (ElectPrimaryEffector.ResultCode.NEW_PRIMARY_ELECTED.name().equalsIgnoreCase(code)) {
+ highlightAction("New primary elected: "+((Map<?,?>)result).get("primary"), null);
+ }
+ if (ElectPrimaryEffector.ResultCode.NO_PRIMARY_AVAILABLE.name().equalsIgnoreCase(code)) {
+ highlightViolation("No primary available");
+ }
+ } catch (Throwable e) {
+ Exceptions.propagateIfFatal(e);
+ if (Entities.isNoLongerManaged(entity)) throw Exceptions.propagate(e);
+
+ Throwable root = Throwables.getRootCause(e);
+ if (root instanceof UserFacingException) {
+ // prefer user-facing root cause to history; mainly for SelectionModeStrictFailed.
+ // possibly not right if it's another (eg exposition on why promote failed)
+ e = root;
+ }
+ if (e instanceof UserFacingException) {
+ log.warn("Error running policy "+this+" on "+entity+": "+Exceptions.collapseText(e));
+ } else {
+ log.warn("Error running policy "+this+" on "+entity+": "+Exceptions.collapseText(e), e);
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/brooklyn-server/blob/0191d90e/policy/src/main/java/org/apache/brooklyn/policy/failover/PrimaryRunningEnricher.java
----------------------------------------------------------------------
diff --git a/policy/src/main/java/org/apache/brooklyn/policy/failover/PrimaryRunningEnricher.java b/policy/src/main/java/org/apache/brooklyn/policy/failover/PrimaryRunningEnricher.java
new file mode 100644
index 0000000..0353aa8
--- /dev/null
+++ b/policy/src/main/java/org/apache/brooklyn/policy/failover/PrimaryRunningEnricher.java
@@ -0,0 +1,75 @@
+/*
+ * 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.brooklyn.policy.failover;
+
+import org.apache.brooklyn.api.entity.Entity;
+import org.apache.brooklyn.api.sensor.SensorEvent;
+import org.apache.brooklyn.api.sensor.SensorEventListener;
+import org.apache.brooklyn.config.ConfigKey;
+import org.apache.brooklyn.core.enricher.AbstractEnricher;
+import org.apache.brooklyn.core.entity.Attributes;
+import org.apache.brooklyn.core.entity.lifecycle.Lifecycle;
+import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic;
+import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.ServiceNotUpLogic;
+import org.apache.brooklyn.core.entity.lifecycle.ServiceStateLogic.ServiceProblemsLogic;
+import org.apache.brooklyn.core.sensor.Sensors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.Beta;
+
+/** Records if the elected primary child/member is running, updating service state of this entity
+ * if it isn't running when it should be via the service problems map which in the event of
+ * an issue will contain more information about primary status. */
+@Beta
+public class PrimaryRunningEnricher extends AbstractEnricher implements SensorEventListener<Object> {
+
+ private static final Logger log = LoggerFactory.getLogger(PrimaryRunningEnricher.class);
+
+ public static final ConfigKey<String> PRIMARY_SENSOR_NAME = ElectPrimaryConfig.PRIMARY_SENSOR_NAME;
+
+ public void setEntity(@SuppressWarnings("deprecation") org.apache.brooklyn.api.entity.EntityLocal entity) {
+ super.setEntity(entity);
+ subscriptions().subscribe(entity, Sensors.newSensor(Entity.class, config().get(PRIMARY_SENSOR_NAME)), this);
+ subscriptions().subscribeToChildren(entity, Attributes.SERVICE_UP, this);
+ subscriptions().subscribeToChildren(entity, Attributes.SERVICE_STATE_ACTUAL, this);
+ highlightTriggers("Listening for "+config().get(PRIMARY_SENSOR_NAME)+" locally and service up and state at children");
+ onEvent(null);
+ }
+
+ @Override
+ public void onEvent(SensorEvent<Object> event) {
+ Entity primary = entity.getAttribute( Sensors.newSensor(Entity.class, config().get(PRIMARY_SENSOR_NAME)) );
+ if (primary==null) {
+ ServiceNotUpLogic.updateNotUpIndicator(entity, "primary.enricher", "no primary found");
+ ServiceProblemsLogic.updateProblemsIndicator(entity, "primary.enricher", "no primary found");
+ } else if (Lifecycle.RUNNING.equals(primary.getAttribute(Attributes.SERVICE_STATE_ACTUAL)) &&
+ Boolean.TRUE.equals(primary.getAttribute(Attributes.SERVICE_UP))) {
+ if (ServiceStateLogic.getMapSensorEntry(entity, Attributes.SERVICE_PROBLEMS, "primary.enricher")!=null) {
+ log.info("Primary "+primary+" at "+entity+" detected as healthy");
+ ServiceProblemsLogic.clearProblemsIndicator(entity, "primary.enricher");
+ ServiceNotUpLogic.clearNotUpIndicator(entity, "primary.enricher");
+ }
+ } else {
+ log.warn("Primary "+primary+" at "+entity+" detected as down or otherwise unhealthy");
+ ServiceProblemsLogic.updateProblemsIndicator(entity, "primary.enricher", "Primary "+primary+" not in healthy state");
+ }
+ }
+
+}