services/src/main/java/org/apache/unomi/services/services/ClusterServiceImpl.java - unomi - Git at Google

 /*
  * 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.unomi.services.services;

 import org.apache.karaf.cellar.config.ClusterConfigurationEvent;
 import org.apache.karaf.cellar.config.Constants;
 import org.apache.karaf.cellar.core.*;
 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.apache.unomi.api.ClusterNode;
 import org.apache.unomi.api.services.ClusterService;
 import org.apache.unomi.persistence.spi.PersistenceService;
 import org.osgi.service.cm.ConfigurationAdmin;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import javax.management.*;
 import javax.management.remote.JMXConnector;
 import javax.management.remote.JMXConnectorFactory;
 import javax.management.remote.JMXServiceURL;
 import java.io.IOException;
 import java.lang.management.ManagementFactory;
 import java.lang.management.OperatingSystemMXBean;
 import java.lang.management.RuntimeMXBean;
 import java.net.ConnectException;
 import java.net.MalformedURLException;
 import java.util.*;

 /**
  * Implementation of the persistence service interface
  */
 public class ClusterServiceImpl implements ClusterService {

     public static final String KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION = "org.apache.unomi.nodes";
     public static final String KARAF_CLUSTER_CONFIGURATION_PUBLIC_ENDPOINTS = "publicEndpoints";
     public static final String KARAF_CLUSTER_CONFIGURATION_INTERNAL_ENDPOINTS = "internalEndpoints";
     private static final Logger logger = LoggerFactory.getLogger(ClusterServiceImpl.class.getName());
     PersistenceService persistenceService;
     private ClusterManager karafCellarClusterManager;
     private EventProducer karafCellarEventProducer;
     private GroupManager karafCellarGroupManager;
     private String karafCellarGroupName = Configurations.DEFAULT_GROUP_NAME;
     private ConfigurationAdmin osgiConfigurationAdmin;
     private String karafJMXUsername = "karaf";
     private String karafJMXPassword = "karaf";
     private int karafJMXPort = 1099;
     private String publicAddress;
     private String internalAddress;
     private Map<String, JMXConnector> jmxConnectors = new LinkedHashMap<>();

     public void setPersistenceService(PersistenceService persistenceService) {
         this.persistenceService = persistenceService;
     }

     public void setKarafCellarClusterManager(ClusterManager karafCellarClusterManager) {
         this.karafCellarClusterManager = karafCellarClusterManager;
     }

     public void setKarafCellarEventProducer(EventProducer karafCellarEventProducer) {
         this.karafCellarEventProducer = karafCellarEventProducer;
     }

     public void setKarafCellarGroupManager(GroupManager karafCellarGroupManager) {
         this.karafCellarGroupManager = karafCellarGroupManager;
     }

     public void setKarafCellarGroupName(String karafCellarGroupName) {
         this.karafCellarGroupName = karafCellarGroupName;
     }

     public void setOsgiConfigurationAdmin(ConfigurationAdmin osgiConfigurationAdmin) {
         this.osgiConfigurationAdmin = osgiConfigurationAdmin;
     }

     public void setKarafJMXUsername(String karafJMXUsername) {
         this.karafJMXUsername = karafJMXUsername;
     }

     public void setKarafJMXPassword(String karafJMXPassword) {
         this.karafJMXPassword = karafJMXPassword;
     }

     public void setKarafJMXPort(int karafJMXPort) {
         this.karafJMXPort = karafJMXPort;
     }

     public void setPublicAddress(String publicAddress) {
         this.publicAddress = publicAddress;
     }

     public void setInternalAddress(String internalAddress) {
         this.internalAddress = internalAddress;
     }

     public void init() {
         if (karafCellarEventProducer != null && karafCellarClusterManager != null) {

             boolean setupConfigOk = true;
             Group group = karafCellarGroupManager.findGroupByName(karafCellarGroupName);
             if (setupConfigOk && group == null) {
                 logger.error("Cluster group " + karafCellarGroupName + " doesn't exist, creating it...");
                 group = karafCellarGroupManager.createGroup(karafCellarGroupName);
                 if (group != null) {
                     setupConfigOk = true;
                 } else {
                     setupConfigOk = false;
                 }
             }

             // check if the producer is ON
             if (setupConfigOk && karafCellarEventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) {
                 logger.error("Cluster event producer is OFF");
                 setupConfigOk = false;
             }

             // check if the config pid is allowed
             if (setupConfigOk && !isClusterConfigPIDAllowed(group, Constants.CATEGORY, KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION, EventType.OUTBOUND)) {
                 logger.error("Configuration PID " + KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION + " is blocked outbound for cluster group " + karafCellarGroupName);
                 setupConfigOk = false;
             }

             if (setupConfigOk) {
                 Map<String, Properties> configurations = karafCellarClusterManager.getMap(Constants.CONFIGURATION_MAP + Configurations.SEPARATOR + karafCellarGroupName);
                 org.apache.karaf.cellar.core.Node thisKarafNode = karafCellarClusterManager.getNode();
                 Properties karafCellarClusterNodeConfiguration = configurations.get(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION);
                 if (karafCellarClusterNodeConfiguration == null) {
                     karafCellarClusterNodeConfiguration = new Properties();
                 }
                 Map<String, String> publicEndpoints = getMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_PUBLIC_ENDPOINTS, thisKarafNode.getId() + "=" + publicAddress);
                 publicEndpoints.put(thisKarafNode.getId(), publicAddress);
                 setMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_PUBLIC_ENDPOINTS, publicEndpoints);

                 Map<String, String> internalEndpoints = getMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_INTERNAL_ENDPOINTS, thisKarafNode.getId() + "=" + internalAddress);
                 internalEndpoints.put(thisKarafNode.getId(), internalAddress);
                 setMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_INTERNAL_ENDPOINTS, internalEndpoints);

                 configurations.put(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION, karafCellarClusterNodeConfiguration);
                 ClusterConfigurationEvent clusterConfigurationEvent = new ClusterConfigurationEvent(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION);
                 clusterConfigurationEvent.setSourceGroup(group);
                 karafCellarEventProducer.produce(clusterConfigurationEvent);
             }
         }
         logger.info("Cluster service initialized.");
     }

     public void destroy() {
         for (Map.Entry<String, JMXConnector> jmxConnectorEntry : jmxConnectors.entrySet()) {
             String url = jmxConnectorEntry.getKey();
             JMXConnector jmxConnector = jmxConnectorEntry.getValue();
             try {
                 jmxConnector.close();
             } catch (IOException e) {
                 logger.error("Error closing JMX connector for url {}", url, e);
             }
         }
         logger.info("Cluster service shutdown.");
     }

     @Override
     public List<ClusterNode> getClusterNodes() {
         Map<String, ClusterNode> clusterNodes = new LinkedHashMap<String, ClusterNode>();

         Set<org.apache.karaf.cellar.core.Node> karafCellarNodes = karafCellarClusterManager.listNodes();
         org.apache.karaf.cellar.core.Node thisKarafNode = karafCellarClusterManager.getNode();
         Map<String, Properties> clusterConfigurations = karafCellarClusterManager.getMap(Constants.CONFIGURATION_MAP + Configurations.SEPARATOR + karafCellarGroupName);
         Properties karafCellarClusterNodeConfiguration = clusterConfigurations.get(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION);
         Map<String, String> publicNodeEndpoints = new TreeMap<>();
         Map<String, String> internalNodeEndpoints = new TreeMap<>();
         if (karafCellarClusterNodeConfiguration != null) {
             publicNodeEndpoints = getMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_PUBLIC_ENDPOINTS, thisKarafNode.getId() + "=" + publicAddress);
             internalNodeEndpoints = getMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_INTERNAL_ENDPOINTS, thisKarafNode.getId() + "=" + internalAddress);
         }
         for (org.apache.karaf.cellar.core.Node karafCellarNode : karafCellarNodes) {
             ClusterNode clusterNode = new ClusterNode();
             String publicEndpoint = publicNodeEndpoints.get(karafCellarNode.getId());
             if (publicEndpoint != null) {
                 clusterNode.setPublicHostAddress(publicEndpoint);
             }
             String internalEndpoint = internalNodeEndpoints.get(karafCellarNode.getId());
             if (internalEndpoint != null) {
                 clusterNode.setInternalHostAddress(internalEndpoint);
             }
             String serviceUrl = "service:jmx:rmi:///jndi/rmi://" + karafCellarNode.getHost() + ":" + karafJMXPort + "/karaf-root";
             try {
                 JMXConnector jmxConnector = getJMXConnector(serviceUrl);
                 MBeanServerConnection mbsc = jmxConnector.getMBeanServerConnection();
                 final RuntimeMXBean remoteRuntime = ManagementFactory.newPlatformMXBeanProxy(mbsc, ManagementFactory.RUNTIME_MXBEAN_NAME, RuntimeMXBean.class);
                 clusterNode.setUptime(remoteRuntime.getUptime());
                 ObjectName operatingSystemMXBeanName = new ObjectName(ManagementFactory.OPERATING_SYSTEM_MXBEAN_NAME);
                 Double systemCpuLoad = null;
                 try {
                     systemCpuLoad = (Double) mbsc.getAttribute(operatingSystemMXBeanName, "SystemCpuLoad");
                 } catch (MBeanException e) {
                     logger.error("Error retrieving system CPU load", e);
                 } catch (AttributeNotFoundException e) {
                     logger.error("Error retrieving system CPU load", e);
                 }
                 final OperatingSystemMXBean remoteOperatingSystemMXBean = ManagementFactory.newPlatformMXBeanProxy(mbsc, ManagementFactory.OPERATING_SYSTEM_MXBEAN_NAME, OperatingSystemMXBean.class);
                 clusterNode.setLoadAverage(new double[]{remoteOperatingSystemMXBean.getSystemLoadAverage()});
                 if (systemCpuLoad != null) {
                     clusterNode.setCpuLoad(systemCpuLoad);
                 }

             } catch (MalformedURLException e) {
                 logger.error("Error connecting to remote JMX server", e);
             } catch (ConnectException ce) {
                 handleTimeouts(serviceUrl, ce);
             } catch (java.rmi.ConnectException ce) {
                 handleTimeouts(serviceUrl, ce);
             } catch (java.rmi.ConnectIOException cioe) {
                 handleTimeouts(serviceUrl, cioe);
             } catch (IOException e) {
                 logger.error("Error retrieving remote JMX data", e);
             } catch (MalformedObjectNameException e) {
                 logger.error("Error retrieving remote JMX data", e);
             } catch (InstanceNotFoundException e) {
                 logger.error("Error retrieving remote JMX data", e);
             } catch (ReflectionException e) {
                 logger.error("Error retrieving remote JMX data", e);
             }
             clusterNodes.put(karafCellarNode.getId(), clusterNode);
         }

         return new ArrayList<ClusterNode>(clusterNodes.values());
     }

     private void handleTimeouts(String serviceUrl, Throwable throwable) {
         Throwable rootCause = throwable;
         while (rootCause.getCause() != null) {
             rootCause = rootCause.getCause();
         }
         logger.warn("JMX RMI Connection error, will reconnect on next request. Active debug logging for access to detailed stack trace. Root cause=" + rootCause.getMessage());
         logger.debug("Detailed stacktrace", throwable);
         JMXConnector jmxConnector = jmxConnectors.remove(serviceUrl);
         try {
             if (jmxConnector != null) {
                 jmxConnector.close();
             }
         } catch (Throwable t) {
             // ignore any exception when closing a timed out connection.
         }
     }

     @Override
     public void purge(Date date) {
         persistenceService.purge(date);
     }

     @Override
     public void purge(String scope) {
         persistenceService.purge(scope);
     }

     /**
      * Check if a configuration is allowed.
      *
      * @param group    the cluster group.
      * @param category the configuration category constant.
      * @param pid      the configuration PID.
      * @param type     the cluster event type.
      * @return true if the cluster event type is allowed, false else.
      */
     public boolean isClusterConfigPIDAllowed(Group group, String category, String pid, EventType type) {
         CellarSupport support = new CellarSupport();
         support.setClusterManager(this.karafCellarClusterManager);
         support.setGroupManager(this.karafCellarGroupManager);
         support.setConfigurationAdmin(this.osgiConfigurationAdmin);
         return support.isAllowed(group, category, pid, type);
     }

     private JMXConnector getJMXConnector(String url) throws IOException {
         if (jmxConnectors.containsKey(url)) {
             JMXConnector jmxConnector = jmxConnectors.get(url);
             try {
                 jmxConnector.getMBeanServerConnection();
                 return jmxConnector;
             } catch (IOException e) {
                 jmxConnectors.remove(url);
                 try {
                     jmxConnector.close();
                 } catch (IOException e1) {
                     logger.warn("Closing invalid JMX connection resulted in :" + e1.getMessage() + ", this is probably ok.");
                     logger.debug("Error closing invalid JMX connection", e1);
                 }
                 if (e.getMessage() != null && e.getMessage().contains("Connection closed")) {
                     logger.warn("JMX connection to url {} was closed (Cause:{}). Reconnecting...", url, e.getMessage());
                 } else {
                     logger.error("Error using the JMX connection to url {}, closed and will reconnect", url, e);
                 }
             }
         }
         // if we reach this point either we didn't have a connector or it didn't validate
         // now let's connect to remote JMX service to retrieve information from the runtime and operating system MX beans
         JMXServiceURL jmxServiceURL = new JMXServiceURL(url);
         Map<String, Object> environment = new HashMap<String, Object>();
         if (karafJMXUsername != null && karafJMXPassword != null) {
             environment.put(JMXConnector.CREDENTIALS, new String[]{karafJMXUsername, karafJMXPassword});
         }
         JMXConnector jmxConnector = JMXConnectorFactory.connect(jmxServiceURL, environment);
         jmxConnectors.put(url, jmxConnector);
         return jmxConnector;
     }

     private Map<String, String> getMapProperty(Properties properties, String propertyName, String defaultValue) {
         String propertyValue = properties.getProperty(propertyName, defaultValue);
         return getMapProperty(propertyValue);
     }

     private Map<String, String> getMapProperty(String propertyValue) {
         String[] propertyValueArray = propertyValue.split(",");
         Map<String, String> propertyMapValue = new LinkedHashMap<>();
         for (String propertyValueElement : propertyValueArray) {
             String[] propertyValueElementPrats = propertyValueElement.split("=");
             propertyMapValue.put(propertyValueElementPrats[0], propertyValueElementPrats[1]);
         }
         return propertyMapValue;
     }

     private Map<String, String> setMapProperty(Properties properties, String propertyName, Map<String, String> propertyMapValue) {
         StringBuilder propertyValueBuilder = new StringBuilder();
         int entryCount = 0;
         for (Map.Entry<String, String> propertyMapValueEntry : propertyMapValue.entrySet()) {
             propertyValueBuilder.append(propertyMapValueEntry.getKey());
             propertyValueBuilder.append("=");
             propertyValueBuilder.append(propertyMapValueEntry.getValue());
             if (entryCount < propertyMapValue.size() - 1) {
                 propertyValueBuilder.append(",");
             }
         }
         String oldPropertyValue = (String) properties.setProperty(propertyName, propertyValueBuilder.toString());
         if (oldPropertyValue == null) {
             return null;
         }
         return getMapProperty(oldPropertyValue);
     }

 }
	/*
	* 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.unomi.services.services;

	import org.apache.karaf.cellar.config.ClusterConfigurationEvent;
	import org.apache.karaf.cellar.config.Constants;
	import org.apache.karaf.cellar.core.*;
	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.apache.unomi.api.ClusterNode;
	import org.apache.unomi.api.services.ClusterService;
	import org.apache.unomi.persistence.spi.PersistenceService;
	import org.osgi.service.cm.ConfigurationAdmin;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import javax.management.*;
	import javax.management.remote.JMXConnector;
	import javax.management.remote.JMXConnectorFactory;
	import javax.management.remote.JMXServiceURL;
	import java.io.IOException;
	import java.lang.management.ManagementFactory;
	import java.lang.management.OperatingSystemMXBean;
	import java.lang.management.RuntimeMXBean;
	import java.net.ConnectException;
	import java.net.MalformedURLException;
	import java.util.*;

	/**
	* Implementation of the persistence service interface
	*/
	public class ClusterServiceImpl implements ClusterService {

	public static final String KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION = "org.apache.unomi.nodes";
	public static final String KARAF_CLUSTER_CONFIGURATION_PUBLIC_ENDPOINTS = "publicEndpoints";
	public static final String KARAF_CLUSTER_CONFIGURATION_INTERNAL_ENDPOINTS = "internalEndpoints";
	private static final Logger logger = LoggerFactory.getLogger(ClusterServiceImpl.class.getName());
	PersistenceService persistenceService;
	private ClusterManager karafCellarClusterManager;
	private EventProducer karafCellarEventProducer;
	private GroupManager karafCellarGroupManager;
	private String karafCellarGroupName = Configurations.DEFAULT_GROUP_NAME;
	private ConfigurationAdmin osgiConfigurationAdmin;
	private String karafJMXUsername = "karaf";
	private String karafJMXPassword = "karaf";
	private int karafJMXPort = 1099;
	private String publicAddress;
	private String internalAddress;
	private Map<String, JMXConnector> jmxConnectors = new LinkedHashMap<>();

	public void setPersistenceService(PersistenceService persistenceService) {
	this.persistenceService = persistenceService;
	}

	public void setKarafCellarClusterManager(ClusterManager karafCellarClusterManager) {
	this.karafCellarClusterManager = karafCellarClusterManager;
	}

	public void setKarafCellarEventProducer(EventProducer karafCellarEventProducer) {
	this.karafCellarEventProducer = karafCellarEventProducer;
	}

	public void setKarafCellarGroupManager(GroupManager karafCellarGroupManager) {
	this.karafCellarGroupManager = karafCellarGroupManager;
	}

	public void setKarafCellarGroupName(String karafCellarGroupName) {
	this.karafCellarGroupName = karafCellarGroupName;
	}

	public void setOsgiConfigurationAdmin(ConfigurationAdmin osgiConfigurationAdmin) {
	this.osgiConfigurationAdmin = osgiConfigurationAdmin;
	}

	public void setKarafJMXUsername(String karafJMXUsername) {
	this.karafJMXUsername = karafJMXUsername;
	}

	public void setKarafJMXPassword(String karafJMXPassword) {
	this.karafJMXPassword = karafJMXPassword;
	}

	public void setKarafJMXPort(int karafJMXPort) {
	this.karafJMXPort = karafJMXPort;
	}

	public void setPublicAddress(String publicAddress) {
	this.publicAddress = publicAddress;
	}

	public void setInternalAddress(String internalAddress) {
	this.internalAddress = internalAddress;
	}

	public void init() {
	if (karafCellarEventProducer != null && karafCellarClusterManager != null) {

	boolean setupConfigOk = true;
	Group group = karafCellarGroupManager.findGroupByName(karafCellarGroupName);
	if (setupConfigOk && group == null) {
	logger.error("Cluster group " + karafCellarGroupName + " doesn't exist, creating it...");
	group = karafCellarGroupManager.createGroup(karafCellarGroupName);
	if (group != null) {
	setupConfigOk = true;
	} else {
	setupConfigOk = false;
	}
	}

	// check if the producer is ON
	if (setupConfigOk && karafCellarEventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) {
	logger.error("Cluster event producer is OFF");
	setupConfigOk = false;
	}

	// check if the config pid is allowed
	if (setupConfigOk && !isClusterConfigPIDAllowed(group, Constants.CATEGORY, KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION, EventType.OUTBOUND)) {
	logger.error("Configuration PID " + KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION + " is blocked outbound for cluster group " + karafCellarGroupName);
	setupConfigOk = false;
	}

	if (setupConfigOk) {
	Map<String, Properties> configurations = karafCellarClusterManager.getMap(Constants.CONFIGURATION_MAP + Configurations.SEPARATOR + karafCellarGroupName);
	org.apache.karaf.cellar.core.Node thisKarafNode = karafCellarClusterManager.getNode();
	Properties karafCellarClusterNodeConfiguration = configurations.get(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION);
	if (karafCellarClusterNodeConfiguration == null) {
	karafCellarClusterNodeConfiguration = new Properties();
	}
	Map<String, String> publicEndpoints = getMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_PUBLIC_ENDPOINTS, thisKarafNode.getId() + "=" + publicAddress);
	publicEndpoints.put(thisKarafNode.getId(), publicAddress);
	setMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_PUBLIC_ENDPOINTS, publicEndpoints);

	Map<String, String> internalEndpoints = getMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_INTERNAL_ENDPOINTS, thisKarafNode.getId() + "=" + internalAddress);
	internalEndpoints.put(thisKarafNode.getId(), internalAddress);
	setMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_INTERNAL_ENDPOINTS, internalEndpoints);

	configurations.put(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION, karafCellarClusterNodeConfiguration);
	ClusterConfigurationEvent clusterConfigurationEvent = new ClusterConfigurationEvent(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION);
	clusterConfigurationEvent.setSourceGroup(group);
	karafCellarEventProducer.produce(clusterConfigurationEvent);
	}
	}
	logger.info("Cluster service initialized.");
	}

	public void destroy() {
	for (Map.Entry<String, JMXConnector> jmxConnectorEntry : jmxConnectors.entrySet()) {
	String url = jmxConnectorEntry.getKey();
	JMXConnector jmxConnector = jmxConnectorEntry.getValue();
	try {
	jmxConnector.close();
	} catch (IOException e) {
	logger.error("Error closing JMX connector for url {}", url, e);
	}
	}
	logger.info("Cluster service shutdown.");
	}

	@Override
	public List<ClusterNode> getClusterNodes() {
	Map<String, ClusterNode> clusterNodes = new LinkedHashMap<String, ClusterNode>();

	Set<org.apache.karaf.cellar.core.Node> karafCellarNodes = karafCellarClusterManager.listNodes();
	org.apache.karaf.cellar.core.Node thisKarafNode = karafCellarClusterManager.getNode();
	Map<String, Properties> clusterConfigurations = karafCellarClusterManager.getMap(Constants.CONFIGURATION_MAP + Configurations.SEPARATOR + karafCellarGroupName);
	Properties karafCellarClusterNodeConfiguration = clusterConfigurations.get(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION);
	Map<String, String> publicNodeEndpoints = new TreeMap<>();
	Map<String, String> internalNodeEndpoints = new TreeMap<>();
	if (karafCellarClusterNodeConfiguration != null) {
	publicNodeEndpoints = getMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_PUBLIC_ENDPOINTS, thisKarafNode.getId() + "=" + publicAddress);
	internalNodeEndpoints = getMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_INTERNAL_ENDPOINTS, thisKarafNode.getId() + "=" + internalAddress);
	}
	for (org.apache.karaf.cellar.core.Node karafCellarNode : karafCellarNodes) {
	ClusterNode clusterNode = new ClusterNode();
	String publicEndpoint = publicNodeEndpoints.get(karafCellarNode.getId());
	if (publicEndpoint != null) {
	clusterNode.setPublicHostAddress(publicEndpoint);
	}
	String internalEndpoint = internalNodeEndpoints.get(karafCellarNode.getId());
	if (internalEndpoint != null) {
	clusterNode.setInternalHostAddress(internalEndpoint);
	}
	String serviceUrl = "service:jmx:rmi:///jndi/rmi://" + karafCellarNode.getHost() + ":" + karafJMXPort + "/karaf-root";
	try {
	JMXConnector jmxConnector = getJMXConnector(serviceUrl);
	MBeanServerConnection mbsc = jmxConnector.getMBeanServerConnection();
	final RuntimeMXBean remoteRuntime = ManagementFactory.newPlatformMXBeanProxy(mbsc, ManagementFactory.RUNTIME_MXBEAN_NAME, RuntimeMXBean.class);
	clusterNode.setUptime(remoteRuntime.getUptime());
	ObjectName operatingSystemMXBeanName = new ObjectName(ManagementFactory.OPERATING_SYSTEM_MXBEAN_NAME);
	Double systemCpuLoad = null;
	try {
	systemCpuLoad = (Double) mbsc.getAttribute(operatingSystemMXBeanName, "SystemCpuLoad");
	} catch (MBeanException e) {
	logger.error("Error retrieving system CPU load", e);
	} catch (AttributeNotFoundException e) {
	logger.error("Error retrieving system CPU load", e);
	}
	final OperatingSystemMXBean remoteOperatingSystemMXBean = ManagementFactory.newPlatformMXBeanProxy(mbsc, ManagementFactory.OPERATING_SYSTEM_MXBEAN_NAME, OperatingSystemMXBean.class);
	clusterNode.setLoadAverage(new double[]{remoteOperatingSystemMXBean.getSystemLoadAverage()});
	if (systemCpuLoad != null) {
	clusterNode.setCpuLoad(systemCpuLoad);
	}

	} catch (MalformedURLException e) {
	logger.error("Error connecting to remote JMX server", e);
	} catch (ConnectException ce) {
	handleTimeouts(serviceUrl, ce);
	} catch (java.rmi.ConnectException ce) {
	handleTimeouts(serviceUrl, ce);
	} catch (java.rmi.ConnectIOException cioe) {
	handleTimeouts(serviceUrl, cioe);
	} catch (IOException e) {
	logger.error("Error retrieving remote JMX data", e);
	} catch (MalformedObjectNameException e) {
	logger.error("Error retrieving remote JMX data", e);
	} catch (InstanceNotFoundException e) {
	logger.error("Error retrieving remote JMX data", e);
	} catch (ReflectionException e) {
	logger.error("Error retrieving remote JMX data", e);
	}
	clusterNodes.put(karafCellarNode.getId(), clusterNode);
	}

	return new ArrayList<ClusterNode>(clusterNodes.values());
	}

	private void handleTimeouts(String serviceUrl, Throwable throwable) {
	Throwable rootCause = throwable;
	while (rootCause.getCause() != null) {
	rootCause = rootCause.getCause();
	}
	logger.warn("JMX RMI Connection error, will reconnect on next request. Active debug logging for access to detailed stack trace. Root cause=" + rootCause.getMessage());
	logger.debug("Detailed stacktrace", throwable);
	JMXConnector jmxConnector = jmxConnectors.remove(serviceUrl);
	try {
	if (jmxConnector != null) {
	jmxConnector.close();
	}
	} catch (Throwable t) {
	// ignore any exception when closing a timed out connection.
	}
	}

	@Override
	public void purge(Date date) {
	persistenceService.purge(date);
	}

	@Override
	public void purge(String scope) {
	persistenceService.purge(scope);
	}

	/**
	* Check if a configuration is allowed.
	*
	* @param group the cluster group.
	* @param category the configuration category constant.
	* @param pid the configuration PID.
	* @param type the cluster event type.
	* @return true if the cluster event type is allowed, false else.
	*/
	public boolean isClusterConfigPIDAllowed(Group group, String category, String pid, EventType type) {
	CellarSupport support = new CellarSupport();
	support.setClusterManager(this.karafCellarClusterManager);
	support.setGroupManager(this.karafCellarGroupManager);
	support.setConfigurationAdmin(this.osgiConfigurationAdmin);
	return support.isAllowed(group, category, pid, type);
	}

	private JMXConnector getJMXConnector(String url) throws IOException {
	if (jmxConnectors.containsKey(url)) {
	JMXConnector jmxConnector = jmxConnectors.get(url);
	try {
	jmxConnector.getMBeanServerConnection();
	return jmxConnector;
	} catch (IOException e) {
	jmxConnectors.remove(url);
	try {
	jmxConnector.close();
	} catch (IOException e1) {
	logger.warn("Closing invalid JMX connection resulted in :" + e1.getMessage() + ", this is probably ok.");
	logger.debug("Error closing invalid JMX connection", e1);
	}
	if (e.getMessage() != null && e.getMessage().contains("Connection closed")) {
	logger.warn("JMX connection to url {} was closed (Cause:{}). Reconnecting...", url, e.getMessage());
	} else {
	logger.error("Error using the JMX connection to url {}, closed and will reconnect", url, e);
	}
	}
	}
	// if we reach this point either we didn't have a connector or it didn't validate
	// now let's connect to remote JMX service to retrieve information from the runtime and operating system MX beans
	JMXServiceURL jmxServiceURL = new JMXServiceURL(url);
	Map<String, Object> environment = new HashMap<String, Object>();
	if (karafJMXUsername != null && karafJMXPassword != null) {
	environment.put(JMXConnector.CREDENTIALS, new String[]{karafJMXUsername, karafJMXPassword});
	}
	JMXConnector jmxConnector = JMXConnectorFactory.connect(jmxServiceURL, environment);
	jmxConnectors.put(url, jmxConnector);
	return jmxConnector;
	}

	private Map<String, String> getMapProperty(Properties properties, String propertyName, String defaultValue) {
	String propertyValue = properties.getProperty(propertyName, defaultValue);
	return getMapProperty(propertyValue);
	}

	private Map<String, String> getMapProperty(String propertyValue) {
	String[] propertyValueArray = propertyValue.split(",");
	Map<String, String> propertyMapValue = new LinkedHashMap<>();
	for (String propertyValueElement : propertyValueArray) {
	String[] propertyValueElementPrats = propertyValueElement.split("=");
	propertyMapValue.put(propertyValueElementPrats[0], propertyValueElementPrats[1]);
	}
	return propertyMapValue;
	}

	private Map<String, String> setMapProperty(Properties properties, String propertyName, Map<String, String> propertyMapValue) {
	StringBuilder propertyValueBuilder = new StringBuilder();
	int entryCount = 0;
	for (Map.Entry<String, String> propertyMapValueEntry : propertyMapValue.entrySet()) {
	propertyValueBuilder.append(propertyMapValueEntry.getKey());
	propertyValueBuilder.append("=");
	propertyValueBuilder.append(propertyMapValueEntry.getValue());
	if (entryCount < propertyMapValue.size() - 1) {
	propertyValueBuilder.append(",");
	}
	}
	String oldPropertyValue = (String) properties.setProperty(propertyName, propertyValueBuilder.toString());
	if (oldPropertyValue == null) {
	return null;
	}
	return getMapProperty(oldPropertyValue);
	}

	}