[KARAF-3981] Improve config synchronizer
Project: http://git-wip-us.apache.org/repos/asf/karaf-cellar/repo Commit: http://git-wip-us.apache.org/repos/asf/karaf-cellar/commit/f71624e5 Tree: http://git-wip-us.apache.org/repos/asf/karaf-cellar/tree/f71624e5 Diff: http://git-wip-us.apache.org/repos/asf/karaf-cellar/diff/f71624e5 Branch: refs/heads/cellar-3.0.x Commit: f71624e5c3e1af7578741edade2a9b1c81e8f3dc Parents: 96d50bb Author: Jean-Baptiste Onofré <[email protected]> Authored: Sun Sep 13 06:58:57 2015 +0200 Committer: Jean-Baptiste Onofré <[email protected]> Committed: Mon Sep 14 08:33:59 2015 +0200 ---------------------------------------------------------------------- .../karaf/cellar/bundle/BundleEventHandler.java | 2 +- .../karaf/cellar/bundle/BundleSynchronizer.java | 20 ++++-- .../config/ConfigurationEventHandler.java | 6 ++ .../config/ConfigurationSynchronizer.java | 71 +++++++++++++++----- .../config/LocalConfigurationListener.java | 6 +- .../resources/OSGI-INF/blueprint/blueprint.xml | 1 + 6 files changed, 80 insertions(+), 26 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/f71624e5/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleEventHandler.java ---------------------------------------------------------------------- diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleEventHandler.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleEventHandler.java index 468ba29..edfe2d5 100644 --- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleEventHandler.java +++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleEventHandler.java @@ -68,7 +68,7 @@ public class BundleEventHandler extends BundleSupport implements EventHandler<Cl try { // check if it's not a "local" event if (event.getSourceNode() != null && event.getSourceNode().getId().equalsIgnoreCase(clusterManager.getNode().getId())) { - LOGGER.debug("CELLAR BUNDLE: cluster event is local (coming from local synchronizer or listener)"); + LOGGER.trace("CELLAR BUNDLE: cluster event is local (coming from local synchronizer or listener)"); return; } if (isAllowed(event.getSourceGroup(), Constants.CATEGORY, event.getLocation(), EventType.INBOUND)) { http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/f71624e5/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSynchronizer.java ---------------------------------------------------------------------- diff --git a/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSynchronizer.java b/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSynchronizer.java index 9722a20..aee5a26 100644 --- a/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSynchronizer.java +++ b/bundle/src/main/java/org/apache/karaf/cellar/bundle/BundleSynchronizer.java @@ -16,6 +16,7 @@ package org.apache.karaf.cellar.bundle; import org.apache.karaf.cellar.core.Configurations; import org.apache.karaf.cellar.core.Group; import org.apache.karaf.cellar.core.Synchronizer; +import org.apache.karaf.cellar.core.control.SwitchStatus; import org.apache.karaf.cellar.core.event.EventProducer; import org.apache.karaf.cellar.core.event.EventType; import org.osgi.framework.Bundle; @@ -69,16 +70,16 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer { public void sync(Group group) { String policy = getSyncPolicy(group); if (policy == null) { - LOGGER.warn("CELLAR BUNDLE: sync policy is not defined for cluster group " + group.getName()); + LOGGER.warn("CELLAR BUNDLE: sync policy is not defined for cluster group {}", group.getName()); } if (policy.equalsIgnoreCase("cluster")) { - LOGGER.debug("CELLAR BUNDLE: sync policy set as 'cluster' for cluster group " + group.getName()); + LOGGER.debug("CELLAR BUNDLE: sync policy set as 'cluster' for cluster group {}", group.getName()); LOGGER.debug("CELLAR BUNDLE: updating node from the cluster (pull first)"); pull(group); LOGGER.debug("CELLAR BUNDLE: updating cluster from the local node (push after)"); push(group); } else if (policy.equalsIgnoreCase("node")) { - LOGGER.debug("CELLAR BUNDLE: sync policy set as 'node' for cluster group " + group.getName()); + LOGGER.debug("CELLAR BUNDLE: sync policy set as 'node' for cluster group {}", group.getName()); LOGGER.debug("CELLAR BUNDLE: updating cluster from the local node (push first)"); push(group); LOGGER.debug("CELLAR BUNDLE: updating node from the cluster (pull after)"); @@ -127,18 +128,18 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer { try { if (state.getStatus() == Bundle.INSTALLED) { if (!isInstalled(state.getLocation())) { - LOGGER.debug("CELLAR BUNDLE: installing bundle located {}", state.getLocation()); + LOGGER.debug("CELLAR BUNDLE: installing bundle located {} on node", state.getLocation()); installBundleFromLocation(state.getLocation()); } else { LOGGER.debug("CELLAR BUNDLE: bundle located {} already installed on node", state.getLocation()); } } else if (state.getStatus() == Bundle.ACTIVE) { if (!isInstalled(state.getLocation())) { - LOGGER.debug("CELLAR BUNDLE: installing bundle located {}", state.getLocation()); + LOGGER.debug("CELLAR BUNDLE: installing bundle located {} on node", state.getLocation()); installBundleFromLocation(state.getLocation()); } if (!isStarted(state.getLocation())) { - LOGGER.debug("CELLAR BUNDLE: starting bundle {}/{}", symbolicName, version); + LOGGER.debug("CELLAR BUNDLE: starting bundle {}/{} on node", symbolicName, version); startBundle(symbolicName, version); } else { LOGGER.debug("CELLAR BUNDLE: bundle located {} already started on node", state.getLocation()); @@ -165,6 +166,11 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer { @Override public void push(Group group) { + if (eventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) { + LOGGER.warn("CELLAR BUNDLE: cluster event producer is OFF"); + return; + } + if (group != null) { String groupName = group.getName(); LOGGER.debug("CELLAR BUNDLE: pushing bundles to cluster group {}", groupName); @@ -255,4 +261,4 @@ public class BundleSynchronizer extends BundleSupport implements Synchronizer { return "disabled"; } -} \ No newline at end of file +} http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/f71624e5/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationEventHandler.java ---------------------------------------------------------------------- diff --git a/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationEventHandler.java b/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationEventHandler.java index 9382414..682bc2c 100644 --- a/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationEventHandler.java +++ b/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationEventHandler.java @@ -62,6 +62,12 @@ public class ConfigurationEventHandler extends ConfigurationSupport implements E return; } + // check if it's not a "local" event + if (event.getSourceNode() != null && event.getSourceNode().getId().equalsIgnoreCase(clusterManager.getNode().getId())) { + LOGGER.trace("CELLAR CONFIG: cluster event is local (coming from local synchronizer or listener)"); + return; + } + Group group = event.getSourceGroup(); String groupName = group.getName(); http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/f71624e5/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationSynchronizer.java ---------------------------------------------------------------------- diff --git a/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationSynchronizer.java b/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationSynchronizer.java index 307ae19..84df57a 100644 --- a/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationSynchronizer.java +++ b/config/src/main/java/org/apache/karaf/cellar/config/ConfigurationSynchronizer.java @@ -38,8 +38,10 @@ public class ConfigurationSynchronizer extends ConfigurationSupport implements S private static final transient Logger LOGGER = LoggerFactory.getLogger(ConfigurationSynchronizer.class); - public ConfigurationSynchronizer() { - // nothing to do + private EventProducer eventProducer; + + public void setEventProducer(EventProducer eventProducer) { + this.eventProducer = eventProducer; } public void init() { @@ -63,19 +65,32 @@ public class ConfigurationSynchronizer extends ConfigurationSupport implements S @Override public void sync(Group group) { String policy = getSyncPolicy(group); - if (policy != null && policy.equalsIgnoreCase("cluster")) { - LOGGER.debug("CELLAR CONFIG: sync policy is set as 'cluster' for cluster group " + group.getName()); - if (clusterManager.listNodesByGroup(group).size() == 1 && clusterManager.listNodesByGroup(group).contains(clusterManager.getNode())) { - LOGGER.debug("CELLAR CONFIG: node is the first and only member of the group, pushing state"); - push(group); - } else { - LOGGER.debug("CELLAR CONFIG: pulling state"); - pull(group); - } + if (policy == null) { + LOGGER.warn("CELLAR CONFIG: sync policy is not defined for cluster group {}", group.getName()); } - if (policy != null && policy.equalsIgnoreCase("node")) { - LOGGER.debug("CELLAR CONFIG: sync policy is set as 'node' for cluster group " + group.getName()); + if (policy.equalsIgnoreCase("cluster")) { + LOGGER.debug("CELLAR CONFIG: sync policy set as 'cluster' for cluster group {}", group.getName()); + LOGGER.debug("CELLAR CONFIG: updating node from the cluster (pull first)"); + pull(group); + LOGGER.debug("CELLAR CONFIG: updating cluster from the local node (push after)"); + push(group); + } else if (policy.equalsIgnoreCase("node")) { + LOGGER.debug("CELLAR CONFIG: sync policy set as 'node' for cluster group {}", group.getName()); + LOGGER.debug("CELLAR CONFIG: updating cluster from the local node (push first)"); push(group); + LOGGER.debug("CELLAR CONFIG: updating node from the cluster (pull after)"); + pull(group); + } else if (policy.equalsIgnoreCase("clusterOnly")) { + LOGGER.debug("CELLAR CONFIG: sync policy set as 'clusterOnly' for cluster group " + group.getName()); + LOGGER.debug("CELLAR CONFIG: updating node from the cluster (pull only)"); + pull(group); + } else if (policy.equalsIgnoreCase("nodeOnly")) { + LOGGER.debug("CELLAR CONFIG: sync policy set as 'nodeOnly' for cluster group " + group.getName()); + LOGGER.debug("CELLAR CONFIG: updating cluster from the local node (push only)"); + push(group); + } else { + LOGGER.debug("CELLAR CONFIG: sync policy set as 'disabled' for cluster group " + group.getName()); + LOGGER.debug("CELLAR CONFIG: no sync"); } } @@ -107,6 +122,7 @@ public class ConfigurationSynchronizer extends ConfigurationSupport implements S localDictionary = filter(localDictionary); if (!equals(clusterDictionary, localDictionary)) { + LOGGER.debug("CELLAR CONFIG: updating configration {} on node", pid); localConfiguration.update((Dictionary) clusterDictionary); persistConfiguration(configurationAdmin, pid, clusterDictionary); } @@ -128,6 +144,11 @@ public class ConfigurationSynchronizer extends ConfigurationSupport implements S */ public void push(Group group) { + if (eventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) { + LOGGER.warn("CELLAR CONFIG: cluster event producer is OFF"); + return; + } + if (group != null) { String groupName = group.getName(); LOGGER.debug("CELLAR CONFIG: pushing configurations to cluster group {}", groupName); @@ -145,8 +166,28 @@ public class ConfigurationSynchronizer extends ConfigurationSupport implements S if (isAllowed(group, Constants.CATEGORY, pid, EventType.OUTBOUND)) { Dictionary localDictionary = localConfiguration.getProperties(); localDictionary = filter(localDictionary); - // update the configurations in the cluster group - clusterConfigurations.put(pid, dictionaryToProperties(localDictionary)); + if (!clusterConfigurations.containsKey(pid)) { + LOGGER.debug("CELLAR CONFIG: creating configuration pid {} on the cluster", pid); + // update cluster configurations + clusterConfigurations.put(pid, dictionaryToProperties(localDictionary)); + // send cluster event + ClusterConfigurationEvent event = new ClusterConfigurationEvent(pid); + event.setSourceGroup(group); + event.setSourceNode(clusterManager.getNode()); + eventProducer.produce(event); + } else { + Dictionary clusterDictionary = clusterConfigurations.get(pid); + if (!equals(clusterDictionary, localDictionary)) { + LOGGER.debug("CELLAR CONFIG: updating configuration pid {} on the cluster", pid); + // update cluster configurations + clusterConfigurations.put(pid, dictionaryToProperties(localDictionary)); + // send cluster event + ClusterConfigurationEvent event = new ClusterConfigurationEvent(pid); + event.setSourceGroup(group); + event.setSourceNode(clusterManager.getNode()); + eventProducer.produce(event); + } + } } else LOGGER.trace("CELLAR CONFIG: configuration with PID {} is marked BLOCKED OUTBOUND for cluster group {}", pid, groupName); } http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/f71624e5/config/src/main/java/org/apache/karaf/cellar/config/LocalConfigurationListener.java ---------------------------------------------------------------------- diff --git a/config/src/main/java/org/apache/karaf/cellar/config/LocalConfigurationListener.java b/config/src/main/java/org/apache/karaf/cellar/config/LocalConfigurationListener.java index 8846e70..67fb201 100644 --- a/config/src/main/java/org/apache/karaf/cellar/config/LocalConfigurationListener.java +++ b/config/src/main/java/org/apache/karaf/cellar/config/LocalConfigurationListener.java @@ -45,7 +45,7 @@ public class LocalConfigurationListener extends ConfigurationSupport implements public void configurationEvent(ConfigurationEvent event) { if (!isEnabled()) { - LOGGER.debug("CELLAR CONFIG: local listener is disabled"); + LOGGER.trace("CELLAR CONFIG: local listener is disabled"); return; } @@ -72,7 +72,7 @@ public class LocalConfigurationListener extends ConfigurationSupport implements if (clusterConfigurations.containsKey(pid)) { // update the configurations in the cluster group clusterConfigurations.remove(pid); - // broadcast the cluster event + // send the cluster event ClusterConfigurationEvent clusterConfigurationEvent = new ClusterConfigurationEvent(pid); clusterConfigurationEvent.setType(event.getType()); clusterConfigurationEvent.setSourceNode(clusterManager.getNode()); @@ -91,7 +91,7 @@ public class LocalConfigurationListener extends ConfigurationSupport implements if (!equals(localDictionary, distributedDictionary)) { // update the configurations in the cluster group clusterConfigurations.put(pid, dictionaryToProperties(localDictionary)); - // broadcast the cluster event + // send the cluster event ClusterConfigurationEvent clusterConfigurationEvent = new ClusterConfigurationEvent(pid); clusterConfigurationEvent.setSourceGroup(group); clusterConfigurationEvent.setSourceNode(clusterManager.getNode()); http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/f71624e5/config/src/main/resources/OSGI-INF/blueprint/blueprint.xml ---------------------------------------------------------------------- diff --git a/config/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/config/src/main/resources/OSGI-INF/blueprint/blueprint.xml index 1baf760..de653da 100644 --- a/config/src/main/resources/OSGI-INF/blueprint/blueprint.xml +++ b/config/src/main/resources/OSGI-INF/blueprint/blueprint.xml @@ -43,6 +43,7 @@ <property name="configurationAdmin" ref="configurationAdmin"/> <property name="groupManager" ref="groupManager"/> <property name="clusterManager" ref="clusterManager"/> + <property name="eventProducer" ref="eventProducer"/> <property name="storage" value="${storage}"/> </bean> <service ref="synchronizer" interface="org.apache.karaf.cellar.core.Synchronizer">
