diff --git a/api/src/main/java/org/apache/unomi/api/ClusterNode.java b/api/src/main/java/org/apache/unomi/api/ClusterNode.java index e096490b9b..6c40ca21ba 100644 --- a/api/src/main/java/org/apache/unomi/api/ClusterNode.java +++ b/api/src/main/java/org/apache/unomi/api/ClusterNode.java @@ -22,12 +22,10 @@ /** * Information about a cluster node. */ -public class ClusterNode extends Item { +public class ClusterNode implements Serializable { private static final long serialVersionUID = 1281422346318230514L; - public static final String ITEM_TYPE = "clusterNode"; - private double cpuLoad; private double[] loadAverage; private String publicHostAddress; @@ -35,18 +33,11 @@ public class ClusterNode extends Item { private long uptime; private boolean master; private boolean data; - private long startTime; - private long lastHeartbeat; - - // Server information - private ServerInfo serverInfo; /** * Instantiates a new Cluster node. */ public ClusterNode() { - super(); - setItemType(ITEM_TYPE); } /** @@ -174,58 +165,4 @@ public boolean isData() { public void setData(boolean data) { this.data = data; } - - /** - * Retrieves the node start time in milliseconds. - * - * @return the start time - */ - public long getStartTime() { - return startTime; - } - - /** - * Sets the node start time in milliseconds. - * - * @param startTime the start time - */ - public void setStartTime(long startTime) { - this.startTime = startTime; - } - - /** - * Retrieves the last heartbeat time in milliseconds. - * - * @return the last heartbeat time - */ - public long getLastHeartbeat() { - return lastHeartbeat; - } - - /** - * Sets the last heartbeat time in milliseconds. - * - * @param lastHeartbeat the last heartbeat time - */ - public void setLastHeartbeat(long lastHeartbeat) { - this.lastHeartbeat = lastHeartbeat; - } - - /** - * Gets the server information. - * - * @return the server information - */ - public ServerInfo getServerInfo() { - return serverInfo; - } - - /** - * Sets the server information. - * - * @param serverInfo the server information - */ - public void setServerInfo(ServerInfo serverInfo) { - this.serverInfo = serverInfo; - } } diff --git a/api/src/main/java/org/apache/unomi/api/services/ClusterService.java b/api/src/main/java/org/apache/unomi/api/services/ClusterService.java index ad87775037..299ac90988 100644 --- a/api/src/main/java/org/apache/unomi/api/services/ClusterService.java +++ b/api/src/main/java/org/apache/unomi/api/services/ClusterService.java @@ -51,4 +51,11 @@ public interface ClusterService { */ void purge(final String scope); + /** + * This function will send an event to the nodes of the cluster + * The function takes a Serializable to avoid dependency on any clustering framework + * + * @param event this object will be cast to a org.apache.karaf.cellar.core.event.Event object + */ + void sendEvent(Serializable event); } diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/RouterCamelContext.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/RouterCamelContext.java index c7e9531327..67219f9c56 100644 --- a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/RouterCamelContext.java +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/RouterCamelContext.java @@ -31,6 +31,7 @@ import org.apache.unomi.router.api.RouterConstants; import org.apache.unomi.router.api.services.ImportExportConfigurationService; import org.apache.unomi.router.api.services.ProfileExportService; +import org.apache.unomi.router.core.event.UpdateCamelRouteEvent; import org.apache.unomi.router.core.processor.ExportRouteCompletionProcessor; import org.apache.unomi.router.core.processor.ImportConfigByFileNameProcessor; import org.apache.unomi.router.core.processor.ImportRouteCompletionProcessor; @@ -239,6 +240,12 @@ public void killExistingRoute(String routeId, boolean fireEvent) throws Exceptio camelContext.removeRouteDefinition(routeDefinition); } } + + if (fireEvent) { + UpdateCamelRouteEvent event = new UpdateCamelRouteEvent(EVENT_ID_REMOVE); + event.setRouteId(routeId); + clusterService.sendEvent(event); + } } public void updateProfileImportReaderRoute(String configId, boolean fireEvent) throws Exception { @@ -259,6 +266,11 @@ public void updateProfileImportReaderRoute(String configId, boolean fireEvent) t builder.setJacksonDataFormat(jacksonDataFormat); builder.setContext(camelContext); camelContext.addRoutes(builder); + + if (fireEvent) { + UpdateCamelRouteEvent event = new UpdateCamelRouteEvent(EVENT_ID_IMPORT); + clusterService.sendEvent(event); + } } } @@ -279,6 +291,11 @@ public void updateProfileExportReaderRoute(String configId, boolean fireEvent) t profileExportCollectRouteBuilder.setJacksonDataFormat(jacksonDataFormat); profileExportCollectRouteBuilder.setContext(camelContext); camelContext.addRoutes(profileExportCollectRouteBuilder); + + if (fireEvent) { + UpdateCamelRouteEvent event = new UpdateCamelRouteEvent(EVENT_ID_EXPORT); + clusterService.sendEvent(event); + } } } diff --git a/package/src/main/resources/etc/custom.system.properties b/package/src/main/resources/etc/custom.system.properties index 46a262816f..5e97437c3e 100644 --- a/package/src/main/resources/etc/custom.system.properties +++ b/package/src/main/resources/etc/custom.system.properties @@ -81,14 +81,11 @@ org.apache.unomi.admin.servlet.context=${env:UNOMI_ADMIN_CONTEXT:-/cxs} ####################################################################################################################### ## Cluster Settings ## ####################################################################################################################### +org.apache.unomi.cluster.group=${env:UNOMI_CLUSTER_GROUP:-default} # To simplify testing we set the public address to use HTTP, but for production environments it is highly recommended # to switch to using HTTPS with a proper SSL certificate installed. org.apache.unomi.cluster.public.address=${env:UNOMI_CLUSTER_PUBLIC_ADDRESS:-http://localhost:8181} org.apache.unomi.cluster.internal.address=${env:UNOMI_CLUSTER_INTERNAL_ADDRESS:-https://localhost:9443} -# The nodeId is a required setting that uniquely identifies this node in the cluster. -# It must be set to a unique value for each node in the cluster. -# Example: nodeId=node1 -org.apache.unomi.cluster.nodeId=${env:UNOMI_CLUSTER_NODEID:-unomi-node-1} # The nodeStatisticsUpdateFrequency controls the frequency of the update of system statistics such as CPU load, # system load average and uptime. This value is set in milliseconds and is set to 10 seconds by default. Each node # will retrieve the local values and broadcast them through a cluster event to all the other nodes to update diff --git a/services/pom.xml b/services/pom.xml index 9a0862db8b..c4cf7e4da3 100644 --- a/services/pom.xml +++ b/services/pom.xml @@ -56,12 +56,6 @@ ${project.version} provided - - org.apache.unomi - unomi-lifecycle-watcher - ${project.version} - provided - javax.servlet @@ -181,7 +175,6 @@ *;scope=compile|runtime sun.misc;resolution:=optional, - com.sun.management;resolution:=optional, * diff --git a/services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterServiceImpl.java b/services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterServiceImpl.java index b4bc2c421e..ec4cfe5235 100644 --- a/services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterServiceImpl.java +++ b/services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterServiceImpl.java @@ -18,30 +18,28 @@ package org.apache.unomi.services.impl.cluster; import org.apache.commons.lang3.ArrayUtils; -import org.apache.commons.lang3.StringUtils; +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.Event; +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.PartialList; -import org.apache.unomi.api.ServerInfo; -import org.apache.unomi.api.conditions.Condition; -import org.apache.unomi.api.conditions.ConditionType; import org.apache.unomi.api.services.ClusterService; -import org.apache.unomi.lifecycle.BundleWatcher; +import org.apache.unomi.api.services.SchedulerService; import org.apache.unomi.persistence.spi.PersistenceService; -import org.osgi.framework.BundleContext; -import org.osgi.framework.ServiceReference; -import org.osgi.util.tracker.ServiceTracker; -import org.osgi.util.tracker.ServiceTrackerCustomizer; +import org.osgi.service.cm.ConfigurationAdmin; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.management.*; import java.io.Serializable; import java.lang.management.ManagementFactory; import java.lang.management.OperatingSystemMXBean; import java.lang.management.RuntimeMXBean; import java.util.*; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; /** @@ -49,185 +47,46 @@ */ 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()); - - /** - * We use ServiceTracker instead of Blueprint dependency injection due to a known bug in Apache Aries Blueprint - * where service dependencies are not properly shut down in reverse order of their initialization. - * - * The bug manifests in two ways: - * 1. Services are not shut down in reverse order of their initialization, causing potential deadlocks - * 2. The PersistenceService is often shut down before other services that depend on it, leading to timeout waits - * - * By using ServiceTracker, we have explicit control over: - * - Service lifecycle management - * - Shutdown order - * - Service availability checks - * - Graceful degradation when services become unavailable - */ - private ServiceTracker persistenceServiceTracker; - - // Keep direct reference for backward compatibility and unit tests - private PersistenceService persistenceService; - + PersistenceService persistenceService; + private ClusterManager karafCellarClusterManager; + private EventProducer karafCellarEventProducer; + private GroupManager karafCellarGroupManager; + private String karafCellarGroupName = Configurations.DEFAULT_GROUP_NAME; + private ConfigurationAdmin osgiConfigurationAdmin; private String publicAddress; private String internalAddress; - //private SchedulerService schedulerService; /* Wait for PR UNOMI-878 to reactivate that code - private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3); - private String nodeId; - private long nodeStartTime; - private long nodeStatisticsUpdateFrequency = 10000; - private Map> nodeSystemStatistics = new ConcurrentHashMap<>(); - private BundleContext bundleContext; - private volatile boolean shutdownNow = false; - - private BundleWatcher bundleWatcher; + private Map> nodeSystemStatistics = new ConcurrentHashMap<>(); + private Group group = null; + private SchedulerService schedulerService; - /** - * Max time to wait for persistence service (in milliseconds) - */ - private static final long MAX_WAIT_TIME = 60000; // 60 seconds + private long nodeStatisticsUpdateFrequency = 10000; - /** - * Sets the bundle context, which is needed to create service trackers - * @param bundleContext the OSGi bundle context - */ - public void setBundleContext(BundleContext bundleContext) { - this.bundleContext = bundleContext; + public void setPersistenceService(PersistenceService persistenceService) { + this.persistenceService = persistenceService; } - /** - * Sets the bundle watcher used to retrieve server information - * - * @param bundleWatcher the bundle watcher - */ - public void setBundleWatcher(BundleWatcher bundleWatcher) { - this.bundleWatcher = bundleWatcher; - LOGGER.info("BundleWatcher service set"); + public void setKarafCellarClusterManager(ClusterManager karafCellarClusterManager) { + this.karafCellarClusterManager = karafCellarClusterManager; } - /** - * Waits for the persistence service to become available. - * This method will retry getting the persistence service with exponential backoff - * until it's available or until the maximum wait time is reached. - * - * @throws IllegalStateException if the persistence service is not available after the maximum wait time - */ - private void waitForPersistenceService() { - if (shutdownNow) { - return; - } - - // If persistence service is directly set (e.g., in unit tests), no need to wait - if (persistenceService != null) { - LOGGER.debug("Persistence service is already available, no need to wait"); - return; - } - - // If no bundle context, we can't get the service via tracker - if (bundleContext == null) { - LOGGER.error("No BundleContext available, cannot wait for persistence service"); - throw new IllegalStateException("No BundleContext available to get persistence service"); - } - - // Initialize service tracker if needed - if (persistenceServiceTracker == null) { - initializeServiceTrackers(); - } - - // Try to get the service with retries - long startTime = System.currentTimeMillis(); - long waitTime = 50; // Start with 50ms wait time - - while (System.currentTimeMillis() - startTime < MAX_WAIT_TIME) { - PersistenceService service = getPersistenceService(); - if (service != null) { - LOGGER.info("Persistence service is now available"); - return; - } - - try { - LOGGER.debug("Waiting for persistence service... ({}ms elapsed)", System.currentTimeMillis() - startTime); - Thread.sleep(waitTime); - // Exponential backoff with a maximum of 5 seconds - waitTime = Math.min(waitTime * 2, 5000); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - LOGGER.error("Interrupted while waiting for persistence service", e); - break; - } - } - - throw new IllegalStateException("PersistenceService not available after waiting " + MAX_WAIT_TIME + "ms"); + public void setKarafCellarEventProducer(EventProducer karafCellarEventProducer) { + this.karafCellarEventProducer = karafCellarEventProducer; } - /** - * Safely gets the persistence service, either from the direct reference (for tests) - * or from the service tracker (for OSGi runtime) - * @return the persistence service or null if not available - */ - private PersistenceService getPersistenceService() { - if (shutdownNow) return null; - - // For unit tests or if already directly set - if (persistenceService != null) { - return persistenceService; - } - - // Otherwise try to get from service tracker - return persistenceServiceTracker != null ? persistenceServiceTracker.getService() : null; + public void setKarafCellarGroupManager(GroupManager karafCellarGroupManager) { + this.karafCellarGroupManager = karafCellarGroupManager; } - /** - * Initialize service tracker for PersistenceService - */ - private void initializeServiceTrackers() { - if (bundleContext == null) { - LOGGER.warn("BundleContext is null, cannot initialize service trackers"); - return; - } - - // Only create service tracker if direct reference isn't set - if (persistenceService == null) { - LOGGER.info("Initializing PersistenceService tracker"); - persistenceServiceTracker = new ServiceTracker<>( - bundleContext, - PersistenceService.class, - new ServiceTrackerCustomizer() { - @Override - public PersistenceService addingService(ServiceReference reference) { - PersistenceService service = bundleContext.getService(reference); - if (service != null) { - persistenceService = service; - LOGGER.info("PersistenceService acquired through tracker"); - } - return service; - } - - @Override - public void modifiedService(ServiceReference reference, PersistenceService service) { - // No action needed - } - - @Override - public void removedService(ServiceReference reference, PersistenceService service) { - LOGGER.info("PersistenceService removed"); - persistenceService = null; - bundleContext.ungetService(reference); - } - } - ); - persistenceServiceTracker.open(); - } + public void setKarafCellarGroupName(String karafCellarGroupName) { + this.karafCellarGroupName = karafCellarGroupName; } - /** - * For unit tests and backward compatibility - directly sets the persistence service - * @param persistenceService the persistence service to set - */ - public void setPersistenceService(PersistenceService persistenceService) { - this.persistenceService = persistenceService; - LOGGER.info("PersistenceService set directly"); + public void setOsgiConfigurationAdmin(ConfigurationAdmin osgiConfigurationAdmin) { + this.osgiConfigurationAdmin = osgiConfigurationAdmin; } public void setPublicAddress(String publicAddress) { @@ -242,35 +101,8 @@ public void setNodeStatisticsUpdateFrequency(long nodeStatisticsUpdateFrequency) this.nodeStatisticsUpdateFrequency = nodeStatisticsUpdateFrequency; } - /* Wait for PR UNOMI-878 to reactivate that code public void setSchedulerService(SchedulerService schedulerService) { this.schedulerService = schedulerService; - - // If we're already initialized, initialize scheduled tasks now - // This handles the case when ClusterService was initialized before SchedulerService was set - if (schedulerService != null && System.currentTimeMillis() > nodeStartTime && nodeStartTime > 0) { - LOGGER.info("SchedulerService was set after ClusterService initialization, initializing scheduled tasks now"); - initializeScheduledTasks(); - } - } - */ - - /* Wait for PR UNOMI-878 to reactivate that code - /** - * Unbind method for the scheduler service, called by the OSGi framework when the service is unregistered - * @param schedulerService The scheduler service being unregistered - */ - /* - public void unsetSchedulerService(SchedulerService schedulerService) { - if (this.schedulerService == schedulerService) { - LOGGER.info("SchedulerService was unset"); - this.schedulerService = null; - } - } - */ - - public void setNodeId(String nodeId) { - this.nodeId = nodeId; } public Map> getNodeSystemStatistics() { @@ -278,331 +110,211 @@ public Map> getNodeSystemStatistics() { } public void init() { - // Initialize service trackers if not set directly (common in unit tests) - initializeServiceTrackers(); - - // Validate that nodeId is provided - if (StringUtils.isBlank(nodeId)) { - String errorMessage = "CRITICAL: nodeId is not set. This is a required setting for cluster operation."; - LOGGER.error(errorMessage); - throw new IllegalStateException(errorMessage); - } + if (karafCellarEventProducer != null && karafCellarClusterManager != null) { + + boolean setupConfigOk = true; + group = karafCellarGroupManager.findGroupByName(karafCellarGroupName); + if (setupConfigOk && group == null) { + LOGGER.error("Cluster group {} doesn't exist, creating it...", karafCellarGroupName); + group = karafCellarGroupManager.createGroup(karafCellarGroupName); + if (group != null) { + setupConfigOk = true; + } else { + setupConfigOk = false; + } + } - // Wait for persistence service to be available - try { - waitForPersistenceService(); - } catch (IllegalStateException e) { - LOGGER.error("Failed to initialize cluster service: {}", e.getMessage()); - return; - } + // check if the producer is ON + if (setupConfigOk && karafCellarEventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) { + LOGGER.error("Cluster event producer is OFF"); + setupConfigOk = false; + } - nodeStartTime = System.currentTimeMillis(); + // 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 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 publicEndpoints = getMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_PUBLIC_ENDPOINTS, thisKarafNode.getId() + "=" + publicAddress); + publicEndpoints.put(thisKarafNode.getId(), publicAddress); + setMapProperty(karafCellarClusterNodeConfiguration, KARAF_CLUSTER_CONFIGURATION_PUBLIC_ENDPOINTS, publicEndpoints); - // Register this node in the persistence service - registerNodeInPersistence(); + Map 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); + sendEvent(clusterConfigurationEvent); + } + + TimerTask statisticsTask = new TimerTask() { + @Override + public void run() { + try { + updateSystemStats(); + } catch (Throwable t) { + LOGGER.error("Error updating system statistics", t); + } + } + }; + schedulerService.getScheduleExecutorService().scheduleWithFixedDelay(statisticsTask, 0, nodeStatisticsUpdateFrequency, TimeUnit.MILLISECONDS); - /* Wait for PR UNOMI-878 to reactivate that code - /* - // Only initialize scheduled tasks if scheduler service is available - if (schedulerService != null) { - initializeScheduledTasks(); - } else { - LOGGER.warn("SchedulerService not available during ClusterService initialization. Scheduled tasks will not be registered. They will be registered when SchedulerService becomes available."); } - */ - initializeScheduledTasks(); + LOGGER.info("Cluster service initialized."); + } - LOGGER.info("Cluster service initialized with node ID: {}", nodeId); + public void destroy() { + LOGGER.info("Cluster service shutdown."); } - /** - * Initializes scheduled tasks for cluster management. - * This method can be called later if schedulerService wasn't available during init. - */ - public void initializeScheduledTasks() { - /* Wait for PR UNOMI-878 to reactivate that code - if (schedulerService == null) { - LOGGER.error("Cannot initialize scheduled tasks: SchedulerService is not set"); - return; + @Override + public List getClusterNodes() { + Map clusterNodes = new LinkedHashMap(); + + Set karafCellarNodes = karafCellarClusterManager.listNodes(); + org.apache.karaf.cellar.core.Node thisKarafNode = karafCellarClusterManager.getNode(); + Map clusterConfigurations = karafCellarClusterManager.getMap(Constants.CONFIGURATION_MAP + Configurations.SEPARATOR + karafCellarGroupName); + Properties karafCellarClusterNodeConfiguration = clusterConfigurations.get(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION); + Map publicNodeEndpoints = new TreeMap<>(); + Map 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); } - */ - - // Schedule regular updates of the node statistics - TimerTask statisticsTask = new TimerTask() { - @Override - public void run() { - try { - updateSystemStats(); - } catch (Throwable t) { - LOGGER.error("Error updating system statistics", t); - } + 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); } - }; - /* Wait for PR UNOMI-878 to reactivate that code - schedulerService.createRecurringTask("clusterNodeStatisticsUpdate", nodeStatisticsUpdateFrequency, TimeUnit.MILLISECONDS, statisticsTask, false); - */ - scheduledExecutorService.scheduleAtFixedRate(statisticsTask, 100, nodeStatisticsUpdateFrequency, TimeUnit.MILLISECONDS); - - // Schedule cleanup of stale nodes - TimerTask cleanupTask = new TimerTask() { - @Override - public void run() { - try { - cleanupStaleNodes(); - } catch (Throwable t) { - LOGGER.error("Error cleaning up stale nodes", t); + String internalEndpoint = internalNodeEndpoints.get(karafCellarNode.getId()); + if (internalEndpoint != null) { + clusterNode.setInternalHostAddress(internalEndpoint); + } + Map nodeStatistics = nodeSystemStatistics.get(karafCellarNode.getId()); + if (nodeStatistics != null) { + Long uptime = (Long) nodeStatistics.get("uptime"); + if (uptime != null) { + clusterNode.setUptime(uptime); + } + Double systemCpuLoad = (Double) nodeStatistics.get("systemCpuLoad"); + if (systemCpuLoad != null) { + clusterNode.setCpuLoad(systemCpuLoad); + } + List loadAverage = (List) nodeStatistics.get("systemLoadAverage"); + if (loadAverage != null) { + Double[] loadAverageArray = loadAverage.toArray(new Double[loadAverage.size()]); + ArrayUtils.toPrimitive(loadAverageArray); + clusterNode.setLoadAverage(ArrayUtils.toPrimitive(loadAverageArray)); } } - }; - /* Wait for PR UNOMI-878 to reactivate that code - schedulerService.createRecurringTask("clusterStaleNodesCleanup", 60000, TimeUnit.MILLISECONDS, cleanupTask, false); - */ - scheduledExecutorService.scheduleAtFixedRate(cleanupTask, 100, 60000, TimeUnit.MILLISECONDS); + clusterNodes.put(karafCellarNode.getId(), clusterNode); + } - LOGGER.info("Cluster service scheduled tasks initialized"); + return new ArrayList(clusterNodes.values()); } - public void destroy() { - shutdownNow = true; - - // Remove this node from the persistence service - PersistenceService service = getPersistenceService(); - if (service != null) { - try { - service.remove(nodeId, ClusterNode.class); - LOGGER.info("Node {} removed from cluster", nodeId); - } catch (Exception e) { - LOGGER.error("Error removing node from cluster", e); - } - } - - // Close service trackers - if (persistenceServiceTracker != null) { - try { - persistenceServiceTracker.close(); - LOGGER.debug("Persistence service tracker closed"); - } catch (Exception e) { - LOGGER.debug("Error closing persistence service tracker: {}", e.getMessage()); - } - persistenceServiceTracker = null; - } + @Override + public void purge(Date date) { + persistenceService.purge(date); + } - // Clear references - persistenceService = null; - bundleWatcher = null; + @Override + public void purge(String scope) { + persistenceService.purge(scope); + } - LOGGER.info("Cluster service shutdown."); + @Override + public void sendEvent(Serializable eventObject) { + Event event = (Event) eventObject; + event.setSourceGroup(group); + event.setSourceNode(karafCellarClusterManager.getNode()); + karafCellarEventProducer.produce(event); } /** - * Register this node in the persistence service + * 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. */ - private void registerNodeInPersistence() { - PersistenceService service = getPersistenceService(); - if (service == null) { - LOGGER.error("Cannot register node: PersistenceService not available"); - return; - } + 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); + } - ClusterNode clusterNode = new ClusterNode(); - clusterNode.setItemId(nodeId); - clusterNode.setPublicHostAddress(publicAddress); - clusterNode.setInternalHostAddress(internalAddress); - clusterNode.setStartTime(nodeStartTime); - clusterNode.setLastHeartbeat(System.currentTimeMillis()); - - // Set server information if BundleWatcher is available - if (bundleWatcher != null && !bundleWatcher.getServerInfos().isEmpty()) { - ServerInfo serverInfo = bundleWatcher.getServerInfos().get(0); - clusterNode.setServerInfo(serverInfo); - LOGGER.info("Added server info to node: version={}, build={}", - serverInfo.getServerVersion(), serverInfo.getServerBuildNumber()); - } else { - LOGGER.warn("BundleWatcher not available at registration time, server info will not be available"); - } + private Map getMapProperty(Properties properties, String propertyName, String defaultValue) { + String propertyValue = properties.getProperty(propertyName, defaultValue); + return getMapProperty(propertyValue); + } - updateSystemStatsForNode(clusterNode); + private Map getMapProperty(String propertyValue) { + String[] propertyValueArray = propertyValue.split(","); + Map propertyMapValue = new LinkedHashMap<>(); + for (String propertyValueElement : propertyValueArray) { + String[] propertyValueElementPrats = propertyValueElement.split("="); + propertyMapValue.put(propertyValueElementPrats[0], propertyValueElementPrats[1]); + } + return propertyMapValue; + } - boolean success = service.save(clusterNode); - if (success) { - LOGGER.info("Node {} registered in cluster", nodeId); - } else { - LOGGER.error("Failed to register node {} in cluster", nodeId); + private Map setMapProperty(Properties properties, String propertyName, Map propertyMapValue) { + StringBuilder propertyValueBuilder = new StringBuilder(); + int entryCount = 0; + for (Map.Entry 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); } - /** - * Updates system stats for the given cluster node - */ - private void updateSystemStatsForNode(ClusterNode node) { + private void updateSystemStats() { final RuntimeMXBean remoteRuntime = ManagementFactory.getRuntimeMXBean(); long uptime = remoteRuntime.getUptime(); - - double systemCpuLoad = 0.0; + ObjectName operatingSystemMXBeanName = ManagementFactory.getOperatingSystemMXBean().getObjectName(); + Double systemCpuLoad = null; try { - systemCpuLoad = ((com.sun.management.OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean()).getSystemCpuLoad(); - // Check for NaN value which Elasticsearch and OpenSearch don't support for float fields - if (Double.isNaN(systemCpuLoad)) { - LOGGER.debug("System CPU load is NaN, setting to 0.0"); - systemCpuLoad = 0.0; - } - } catch (Exception e) { - LOGGER.debug("Error retrieving system CPU load", e); + systemCpuLoad = (Double) ManagementFactory.getPlatformMBeanServer().getAttribute(operatingSystemMXBeanName, "SystemCpuLoad"); + } catch (MBeanException | AttributeNotFoundException | InstanceNotFoundException | ReflectionException e) { + LOGGER.error("Error retrieving system CPU load", e); } - final OperatingSystemMXBean operatingSystemMXBean = ManagementFactory.getOperatingSystemMXBean(); double systemLoadAverage = operatingSystemMXBean.getSystemLoadAverage(); - // Check for NaN value which Elasticsearch/OpenSearch doesn't support for float fields - if (Double.isNaN(systemLoadAverage)) { - LOGGER.debug("System load average is NaN, setting to 0.0"); - systemLoadAverage = 0.0; - } - - node.setCpuLoad(systemCpuLoad); - node.setUptime(uptime); + ClusterSystemStatisticsEvent clusterSystemStatisticsEvent = new ClusterSystemStatisticsEvent("org.apache.unomi.cluster.system.statistics"); + Map systemStatistics = new TreeMap<>(); ArrayList systemLoadAverageArray = new ArrayList<>(); systemLoadAverageArray.add(systemLoadAverage); - node.setLoadAverage(ArrayUtils.toPrimitive(systemLoadAverageArray.toArray(new Double[0]))); - - // Store system statistics in memory as well - Map systemStatistics = new TreeMap<>(); systemStatistics.put("systemLoadAverage", systemLoadAverageArray); systemStatistics.put("systemCpuLoad", systemCpuLoad); systemStatistics.put("uptime", uptime); - nodeSystemStatistics.put(nodeId, systemStatistics); - } - - /** - * Updates the system statistics for this node and stores them in the persistence service - */ - private void updateSystemStats() { - if (shutdownNow) { - return; - } - - PersistenceService service = getPersistenceService(); - if (service == null) { - LOGGER.warn("Cannot update system stats: PersistenceService not available"); - return; - } - - // Load node from persistence - ClusterNode node = service.load(nodeId, ClusterNode.class); - if (node == null) { - LOGGER.warn("Node {} not found in persistence, re-registering", nodeId); - registerNodeInPersistence(); - return; - } - - try { - // Update its stats - updateSystemStatsForNode(node); - - // Update server info if needed - if (bundleWatcher != null && !bundleWatcher.getServerInfos().isEmpty()) { - ServerInfo currentInfo = bundleWatcher.getServerInfos().get(0); - // Check if server info needs updating - if (node.getServerInfo() == null || - !currentInfo.getServerVersion().equals(node.getServerInfo().getServerVersion())) { - - node.setServerInfo(currentInfo); - LOGGER.info("Updated server info for node {}: version={}, build={}", - nodeId, currentInfo.getServerVersion(), currentInfo.getServerBuildNumber()); - } - } - - node.setLastHeartbeat(System.currentTimeMillis()); - - // Save back to persistence - boolean success = service.save(node); - if (!success) { - LOGGER.error("Failed to update node {} statistics", nodeId); - } - } catch (Exception e) { - LOGGER.error("Error updating system statistics for node {}: {}", nodeId, e.getMessage(), e); - } - } - - /** - * Removes stale nodes from the cluster - */ - private void cleanupStaleNodes() { - if (shutdownNow) { - return; - } - - PersistenceService service = getPersistenceService(); - if (service == null) { - LOGGER.warn("Cannot cleanup stale nodes: PersistenceService not available"); - return; - } - - long cutoffTime = System.currentTimeMillis() - (nodeStatisticsUpdateFrequency * 3); // Node is stale if no heartbeat for 3x the update frequency - - Condition staleNodesCondition = new Condition(); - ConditionType propertyConditionType = new ConditionType(); - propertyConditionType.setItemId("propertyCondition"); - propertyConditionType.setItemType(ConditionType.ITEM_TYPE); - propertyConditionType.setConditionEvaluator("propertyConditionEvaluator"); - propertyConditionType.setQueryBuilder("propertyConditionESQueryBuilder"); - staleNodesCondition.setConditionType(propertyConditionType); - staleNodesCondition.setConditionTypeId("propertyCondition"); - staleNodesCondition.setParameter("propertyName", "lastHeartbeat"); - staleNodesCondition.setParameter("comparisonOperator", "lessThan"); - staleNodesCondition.setParameter("propertyValueInteger", cutoffTime); - - PartialList staleNodes = service.query(staleNodesCondition, null, ClusterNode.class, 0, -1); - - for (ClusterNode staleNode : staleNodes.getList()) { - LOGGER.info("Removing stale node: {}", staleNode.getItemId()); - service.remove(staleNode.getItemId(), ClusterNode.class); - nodeSystemStatistics.remove(staleNode.getItemId()); - } - } - - @Override - public List getClusterNodes() { - PersistenceService service = getPersistenceService(); - if (service == null) { - LOGGER.warn("Cannot get cluster nodes: PersistenceService not available"); - return Collections.emptyList(); - } - - // Query all nodes from the persistence service - return service.getAllItems(ClusterNode.class, 0, -1, null).getList(); + clusterSystemStatisticsEvent.setStatistics(systemStatistics); + nodeSystemStatistics.put(karafCellarClusterManager.getNode().getId(), systemStatistics); + sendEvent(clusterSystemStatisticsEvent); } - @Override - public void purge(Date date) { - PersistenceService service = getPersistenceService(); - if (service == null) { - LOGGER.warn("Cannot purge by date: PersistenceService not available"); - return; - } - - service.purge(date); - } - - @Override - public void purge(String scope) { - PersistenceService service = getPersistenceService(); - if (service == null) { - LOGGER.warn("Cannot purge by scope: PersistenceService not available"); - return; - } - - service.purge(scope); - } - - /** - * Check if a persistence service is available. - * This can be used to quickly check before performing operations. - * - * @return true if a persistence service is available (either directly set or via tracker) - */ - public boolean isPersistenceServiceAvailable() { - return getPersistenceService() != null; - } } - diff --git a/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml index 994c5598dc..f49f3c8497 100644 --- a/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml +++ b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml @@ -53,7 +53,7 @@ - + @@ -262,11 +262,13 @@ - + + + + + - diff --git a/services/src/main/resources/org.apache.unomi.cluster.cfg b/services/src/main/resources/org.apache.unomi.cluster.cfg index eecb7e1dec..bfbb189d9e 100644 --- a/services/src/main/resources/org.apache.unomi.cluster.cfg +++ b/services/src/main/resources/org.apache.unomi.cluster.cfg @@ -14,18 +14,14 @@ # See the License for the specific language governing permissions and # limitations under the License. # +group=${org.apache.unomi.cluster.group:-default} # To simplify testing we set the public address to use HTTP, but for production environments it is highly recommended # to switch to using HTTPS with a proper SSL certificate installed. contextserver.publicAddress=${org.apache.unomi.cluster.public.address:-http://localhost:8181} contextserver.internalAddress=${org.apache.unomi.cluster.internal.address:-https://localhost:9443} # -# The nodeId is a required setting that uniquely identifies this node in the cluster. -# It must be set to a unique value for each node in the cluster. -# Example: nodeId=node1 -nodeId=${org.apache.unomi.cluster.nodeId:-unomi-node-1} -# -## The nodeStatisticsUpdateFrequency controls the frequency of the update of system statistics such as CPU load, +# The nodeStatisticsUpdateFrequency controls the frequency of the update of system statistics such as CPU load, # system load average and uptime. This value is set in milliseconds and is set to 10 seconds by default. Each node # will retrieve the local values and broadcast them through a cluster event to all the other nodes to update # the global cluster statistics. -nodeStatisticsUpdateFrequency=${org.apache.unomi.cluster.nodeStatisticsUpdateFrequency:-10000} +nodeStatisticsUpdateFrequency=${org.apache.unomi.cluster.nodeStatisticsUpdateFrequency:-10000} \ No newline at end of file