Repository: karaf-cellar Updated Branches: refs/heads/master f734e334f -> 839e092ef
[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/839e092e Tree: http://git-wip-us.apache.org/repos/asf/karaf-cellar/tree/839e092e Diff: http://git-wip-us.apache.org/repos/asf/karaf-cellar/diff/839e092e Branch: refs/heads/master Commit: 839e092ef466fe2b8d3a53f8b7db2d1fec23bf07 Parents: f734e33 Author: Jean-Baptiste Onofré <[email protected]> Authored: Sun Sep 13 06:58:57 2015 +0200 Committer: Jean-Baptiste Onofré <[email protected]> Committed: Sun Sep 13 06:58:57 2015 +0200 ---------------------------------------------------------------------- .../karaf/cellar/bundle/BundleEventHandler.java | 2 +- .../karaf/cellar/bundle/BundleSynchronizer.java | 18 +++-- .../config/ConfigurationEventHandler.java | 6 ++ .../config/ConfigurationSynchronizer.java | 71 +++++++++++++++----- .../config/LocalConfigurationListener.java | 6 +- .../cellar/config/internal/osgi/Activator.java | 1 + 6 files changed, 79 insertions(+), 25 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/839e092e/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 4b2700b..589fda4 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; } // check if the pid is marked as local. http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/839e092e/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 e8124de..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); http://git-wip-us.apache.org/repos/asf/karaf-cellar/blob/839e092e/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/839e092e/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 5ff6737..79b0ade 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() { @@ -65,19 +67,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"); } } @@ -109,6 +124,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); } @@ -130,6 +146,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); @@ -147,8 +168,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/839e092e/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/839e092e/config/src/main/java/org/apache/karaf/cellar/config/internal/osgi/Activator.java ---------------------------------------------------------------------- diff --git a/config/src/main/java/org/apache/karaf/cellar/config/internal/osgi/Activator.java b/config/src/main/java/org/apache/karaf/cellar/config/internal/osgi/Activator.java index 0502943..d52746e 100644 --- a/config/src/main/java/org/apache/karaf/cellar/config/internal/osgi/Activator.java +++ b/config/src/main/java/org/apache/karaf/cellar/config/internal/osgi/Activator.java @@ -103,6 +103,7 @@ public class Activator extends BaseActivator implements ManagedService { configurationSynchronizer.setConfigurationAdmin(configurationAdmin); configurationSynchronizer.setGroupManager(groupManager); configurationSynchronizer.setClusterManager(clusterManager); + configurationSynchronizer.setEventProducer(eventProducer); configurationSynchronizer.setStorage(storage); configurationSynchronizer.init(); props = new Hashtable();
