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 6c40ca21ba..66a4ee7c4a 100644 --- a/api/src/main/java/org/apache/unomi/api/ClusterNode.java +++ b/api/src/main/java/org/apache/unomi/api/ClusterNode.java @@ -17,15 +17,15 @@ package org.apache.unomi.api; -import java.io.Serializable; - /** * Information about a cluster node. */ -public class ClusterNode implements Serializable { +public class ClusterNode extends Item { private static final long serialVersionUID = 1281422346318230514L; + public static final String ITEM_TYPE = "clusterNode"; + private double cpuLoad; private double[] loadAverage; private String publicHostAddress; @@ -33,11 +33,18 @@ public class ClusterNode implements Serializable { 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); } /** @@ -165,4 +172,58 @@ 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 299ac90988..f100be2864 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 @@ -19,7 +19,6 @@ import org.apache.unomi.api.ClusterNode; -import java.io.Serializable; import java.util.Date; import java.util.List; @@ -51,11 +50,4 @@ 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/pom.xml b/extensions/router/router-core/pom.xml index a8ce97fd49..8b7d46aadf 100644 --- a/extensions/router/router-core/pom.xml +++ b/extensions/router/router-core/pom.xml @@ -133,16 +133,6 @@ ${kafka.client.version} provided - - org.apache.karaf.cellar - org.apache.karaf.cellar.core - provided - - - org.apache.karaf.cellar - org.apache.karaf.cellar.config - provided - 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 67219f9c56..87d275456e 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 @@ -21,7 +21,6 @@ import org.apache.camel.component.jackson.JacksonDataFormat; import org.apache.camel.core.osgi.OsgiDefaultCamelContext; import org.apache.camel.model.RouteDefinition; -import org.apache.unomi.api.services.ClusterService; import org.apache.unomi.api.services.ConfigSharingService; import org.apache.unomi.api.services.ProfileService; import org.apache.unomi.persistence.spi.PersistenceService; @@ -31,7 +30,6 @@ 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; @@ -75,7 +73,6 @@ public class RouterCamelContext implements IRouterCamelContext { private String allowedEndpoints; private BundleContext bundleContext; private ConfigSharingService configSharingService; - private ClusterService clusterService; // TODO UNOMI-572: when fixing UNOMI-572 please remove the usage of the custom ScheduledExecutorService and re-introduce the Unomi Scheduler Service private ScheduledExecutorService scheduler; @@ -102,10 +99,6 @@ public void setConfigSharingService(ConfigSharingService configSharingService) { this.configSharingService = configSharingService; } - public void setClusterService(ClusterService clusterService) { - this.clusterService = clusterService; - } - public void setTracing(boolean tracing) { camelContext.setTracing(true); } @@ -240,12 +233,6 @@ 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 { @@ -266,11 +253,6 @@ 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); - } } } @@ -291,11 +273,6 @@ 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/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/event/UpdateCamelRouteEvent.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/event/UpdateCamelRouteEvent.java deleted file mode 100644 index 2f3d2cb3f1..0000000000 --- a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/event/UpdateCamelRouteEvent.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.unomi.router.core.event; - -import org.apache.karaf.cellar.core.event.Event; - -/** - * @author dgaillard - */ -public class UpdateCamelRouteEvent extends Event { - private String routeId; - - public UpdateCamelRouteEvent(String id) { - super(id); - } - - public String getRouteId() { - return routeId; - } - - public void setRouteId(String routeId) { - this.routeId = routeId; - } -} diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/event/UpdateCamelRouteEventHandler.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/event/UpdateCamelRouteEventHandler.java deleted file mode 100644 index f43f2b629d..0000000000 --- a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/event/UpdateCamelRouteEventHandler.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.unomi.router.core.event; - -import org.apache.commons.lang3.StringUtils; -import org.apache.karaf.cellar.config.Constants; -import org.apache.karaf.cellar.core.CellarSupport; -import org.apache.karaf.cellar.core.control.Switch; -import org.apache.karaf.cellar.core.event.EventHandler; -import org.apache.karaf.cellar.core.event.EventType; -import org.apache.unomi.router.core.context.RouterCamelContext; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * @author dgaillard - */ -public class UpdateCamelRouteEventHandler extends CellarSupport implements EventHandler { - private static final Logger LLOGGER = LoggerFactory.getLogger(UpdateCamelRouteEventHandler.class.getName()); - - private RouterCamelContext routerCamelContext; - - @Override - public void handle(UpdateCamelRouteEvent event) { - LLOGGER.debug("Handle event"); - if (isAllowed(event.getSourceGroup(), Constants.CATEGORY, event.getId(), EventType.INBOUND)) { - LLOGGER.debug("Event is allowed"); - // check if it's not a "local" event - if (event.getSourceNode() != null && event.getSourceNode().getId().equalsIgnoreCase(clusterManager.getNode().getId())) { - LLOGGER.debug("Cluster event is local (coming from local synchronizer or listener)"); - return; - } - - try { - LLOGGER.debug("Event id is {}", event.getId()); - if (event.getId().equals(RouterCamelContext.EVENT_ID_REMOVE) && StringUtils.isNotBlank(event.getRouteId())) { - routerCamelContext.killExistingRoute(event.getRouteId(), false); - } else if ((event.getId().equals(RouterCamelContext.EVENT_ID_IMPORT))) { - routerCamelContext.updateProfileImportReaderRoute(event.getRouteId(), false); - } else if (event.getId().equals(RouterCamelContext.EVENT_ID_EXPORT)) { - routerCamelContext.updateProfileExportReaderRoute(event.getRouteId(), false); - } - } catch (Exception e) { - LLOGGER.error("Error when executing event", e); - } - } - } - - @Override - public Class getType() { - return UpdateCamelRouteEvent.class; - } - - @Override - public Switch getSwitch() { - return null; - } - - public void setRouterCamelContext(RouterCamelContext routerCamelContext) { - this.routerCamelContext = routerCamelContext; - } -} diff --git a/extensions/router/router-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/extensions/router/router-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml index d7b7a36c00..281c93f912 100644 --- a/extensions/router/router-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml +++ b/extensions/router/router-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml @@ -112,7 +112,6 @@ - @@ -120,18 +119,6 @@ - - - - - - - - - - - - @@ -141,8 +128,6 @@ - - - \ No newline at end of file + diff --git a/itests/src/test/java/org/apache/unomi/itests/BaseIT.java b/itests/src/test/java/org/apache/unomi/itests/BaseIT.java index 38b7f0c66d..a98d90d7db 100644 --- a/itests/src/test/java/org/apache/unomi/itests/BaseIT.java +++ b/itests/src/test/java/org/apache/unomi/itests/BaseIT.java @@ -256,11 +256,6 @@ public Option[] config() { editConfigurationFilePut("etc/custom.system.properties", "org.apache.unomi.elasticsearch.rollover.maxDocs", "300"), systemProperty("org.ops4j.pax.exam.rbc.rmi.port").value("1199"), - systemProperty("org.apache.unomi.hazelcast.group.name").value("cellar"), - systemProperty("org.apache.unomi.hazelcast.group.password").value("pass"), - systemProperty("org.apache.unomi.hazelcast.network.port").value("5701"), - systemProperty("org.apache.unomi.hazelcast.tcp-ip.members").value("127.0.0.1"), - systemProperty("org.apache.unomi.hazelcast.tcp-ip.interface").value("127.0.0.1"), systemProperty("org.apache.unomi.healthcheck.enabled").value("true"), logLevel(LogLevel.INFO), @@ -383,6 +378,17 @@ public void updateServices() throws InterruptedException { segmentService = getService(SegmentService.class); } + /** + * Updates an OSGi configuration with a single property value and optionally waits for the service to be reregistered. + * If serviceName is null, the method will not wait for service re-registration. + * + * @param serviceName The fully qualified name of the service to wait for, or null to skip waiting + * @param configPid The persistent identifier of the configuration to update + * @param propName The name of the property to update + * @param propValue The new value for the property + * @throws InterruptedException If the thread is interrupted while waiting for service reregistration + * @throws IOException If an error occurs while updating the configuration + */ public void updateConfiguration(String serviceName, String configPid, String propName, Object propValue) throws InterruptedException, IOException { Map props = new HashMap<>(); @@ -390,20 +396,43 @@ public void updateConfiguration(String serviceName, String configPid, String pro updateConfiguration(serviceName, configPid, props); } + /** + * Updates an OSGi configuration with multiple property values and optionally waits for the service to be reregistered. + * If serviceName is null, the method will not wait for service re-registration. + * + * @param serviceName The fully qualified name of the service to wait for, or null to skip waiting + * @param configPid The persistent identifier of the configuration to update + * @param propsToSet A map of property names to their new values + * @throws InterruptedException If the thread is interrupted while waiting for service reregistration + * @throws IOException If an error occurs while updating the configuration + */ public void updateConfiguration(String serviceName, String configPid, Map propsToSet) throws InterruptedException, IOException { org.osgi.service.cm.Configuration cfg = configurationAdmin.getConfiguration(configPid); Dictionary props = cfg.getProperties(); + + // Handle case where properties haven't been initialized yet + final Dictionary finalProps = (props != null) ? props : new Hashtable<>(); + + // Add new properties to the dictionary for (Map.Entry propToSet : propsToSet.entrySet()) { - props.put(propToSet.getKey(), propToSet.getValue()); + finalProps.put(propToSet.getKey(), propToSet.getValue()); } - waitForReRegistration(serviceName, () -> { - try { - cfg.update(props); - } catch (IOException ignored) { - } - }); + // If serviceName is null, don't wait for service re-registration + if (serviceName == null) { + LOGGER.info("Updating configuration {} without waiting for service restart", configPid); + cfg.update(finalProps); + // Give the configuration change handler time to process + Thread.sleep(1000); + } else { + waitForReRegistration(serviceName, () -> { + try { + cfg.update(finalProps); + } catch (IOException ignored) { + } + }); + } waitForStartup(); diff --git a/itests/src/test/java/org/apache/unomi/itests/ProfileServiceIT.java b/itests/src/test/java/org/apache/unomi/itests/ProfileServiceIT.java index 623904938c..479ba56577 100644 --- a/itests/src/test/java/org/apache/unomi/itests/ProfileServiceIT.java +++ b/itests/src/test/java/org/apache/unomi/itests/ProfileServiceIT.java @@ -168,7 +168,7 @@ public void testGetProfileWithWrongScrollerIdThrowException() } } - updateConfiguration(PersistenceService.class.getName(), "org.apache.unomi.persistence.elasticsearch", "throwExceptions", true); + updateConfiguration(null, "org.apache.unomi.persistence.elasticsearch", "throwExceptions", true); Query query = new Query(); query.setLimit(2); @@ -181,7 +181,7 @@ public void testGetProfileWithWrongScrollerIdThrowException() } catch (RuntimeException ex) { // Should get here since this scenario should throw exception } finally { - updateConfiguration(PersistenceService.class.getName(), "org.apache.unomi.persistence.elasticsearch", "throwExceptions", + updateConfiguration(null, "org.apache.unomi.persistence.elasticsearch", "throwExceptions", throwExceptionCurrent); } } diff --git a/itests/src/test/java/org/apache/unomi/itests/RuleServiceIT.java b/itests/src/test/java/org/apache/unomi/itests/RuleServiceIT.java index 1d5a23d9de..04ef4904a7 100644 --- a/itests/src/test/java/org/apache/unomi/itests/RuleServiceIT.java +++ b/itests/src/test/java/org/apache/unomi/itests/RuleServiceIT.java @@ -151,14 +151,14 @@ public void testRuleOptimizationPerf() throws NoSuchFieldException, IllegalAcces Profile profile = new Profile(UUID.randomUUID().toString()); Session session = new Session(UUID.randomUUID().toString(), profile, new Date(), TEST_SCOPE); - updateConfiguration(RulesService.class.getName(), "org.apache.unomi.services", "rules.optimizationActivated", "false"); + updateConfiguration(null, "org.apache.unomi.services", "rules.optimizationActivated", "false"); rulesService = getService(RulesService.class); eventService = getService(EventService.class); LOGGER.info("Running unoptimized rules performance test..."); long unoptimizedRunTime = runEventTest(profile, session); - updateConfiguration(RulesService.class.getName(), "org.apache.unomi.services", "rules.optimizationActivated", "true"); + updateConfiguration(null, "org.apache.unomi.services", "rules.optimizationActivated", "true"); rulesService = getService(RulesService.class); eventService = getService(EventService.class); diff --git a/kar/pom.xml b/kar/pom.xml index 226e3c85cf..9a38f9b78a 100644 --- a/kar/pom.xml +++ b/kar/pom.xml @@ -143,13 +143,6 @@ joda-time joda-time - - org.apache.karaf.cellar - apache-karaf-cellar - features - xml - runtime - org.apache.unomi unomi-web-tracker-wab diff --git a/kar/src/main/feature/feature.xml b/kar/src/main/feature/feature.xml index 9de01623ac..6355a69e43 100644 --- a/kar/src/main/feature/feature.xml +++ b/kar/src/main/feature/feature.xml @@ -19,7 +19,6 @@ mvn:org.apache.cxf.karaf/apache-cxf/${cxf.version}/xml/features - mvn:org.apache.karaf.cellar/apache-karaf-cellar/${version.karaf.cellar}/xml/features wrap @@ -29,7 +28,6 @@ cxf-features-metrics cxf-rs-security-cors cxf-rs-description-openapi-v3 - cellar eventadmin feature shell-compat @@ -40,7 +38,6 @@ mvn:org.apache.unomi/unomi-services/${project.version}/cfg/servicescfg mvn:org.apache.unomi/unomi-services/${project.version}/cfg/thirdpartycfg mvn:org.apache.unomi/unomi-services/${project.version}/cfg/clustercfg - mvn:org.apache.unomi/unomi-services/${project.version}/xml/hazelcastconfig mvn:org.apache.unomi/cxs-geonames-services/${project.version}/cfg/geonamescfg mvn:org.apache.unomi/unomi-groovy-actions-services/${project.version}/cfg/groovyactionscfg mvn:org.apache.unomi/unomi-json-schema-services/${project.version}/cfg/schemacfg diff --git a/manual/src/main/asciidoc/building-and-deploying.adoc b/manual/src/main/asciidoc/building-and-deploying.adoc index 840639e6ee..3de47041f9 100644 --- a/manual/src/main/asciidoc/building-and-deploying.adoc +++ b/manual/src/main/asciidoc/building-and-deploying.adoc @@ -144,12 +144,11 @@ files (at the end of the file): export KARAF_OPTS="$KARAF_OPTS -Xmx3G" ---- -Install the WAR support, CXF and Karaf Cellar into Karaf by doing the following in the Karaf command line: +Install the WAR support and CXF into Karaf by doing the following in the Karaf command line: [source] ---- feature:repo-add cxf-jaxrs 3.3.4 - feature:repo-add cellar 4.1.3 feature:repo-add mvn:org.apache.unomi/unomi-kar/VERSION/xml/features feature:install unomi-kar ---- diff --git a/manual/src/main/asciidoc/clustering.adoc b/manual/src/main/asciidoc/clustering.adoc index be27ccfd95..9d31aa6a2b 100644 --- a/manual/src/main/asciidoc/clustering.adoc +++ b/manual/src/main/asciidoc/clustering.adoc @@ -13,7 +13,7 @@ // === Cluster setup -Apache Karaf relies on Apache Karaf Cellar, which in turn uses Hazelcast to discover and configure its cluster. +Apache Karaf relies on Persistence to register nodes and manage cluster. You can control most of the important clustering settings through the centralized configuration file at @@ -21,19 +21,9 @@ You can control most of the important clustering settings through the centralize And notably using the following properties: - org.apache.unomi.hazelcast.group.name=${env:UNOMI_HAZELCAST_GROUP_NAME:-cellar} - org.apache.unomi.hazelcast.group.password=${env:UNOMI_HAZELCAST_GROUP_PASSWORD:-pass} - # This list can be comma separated and use ranges such as 192.168.1.0-7,192.168.1.21 - org.apache.unomi.hazelcast.tcp-ip.members=${env:UNOMI_HAZELCAST_TCPIP_MEMBERS:-127.0.0.1} - org.apache.unomi.hazelcast.tcp-ip.interface=${env:UNOMI_HAZELCAST_TCPIP_INTERFACE:-127.0.0.1} - org.apache.unomi.hazelcast.network.port=${env:UNOMI_HAZELCAST_NETWORK_PORT:-5701} + 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} + org.apache.unomi.cluster.nodeId=${env:UNOMI_CLUSTER_NODEID:-unomi-node-1} + org.apache.unomi.cluster.nodeStatisticsUpdateFrequency=${env:UNOMI_CLUSTER_NODESTATISTICS_UPDATEFREQUENCY:-10000} -If you need more fine-grained control over the Hazelcast configuration you could also edit the following file: - -[source] ----- -etc/hazelcast.xml ----- - -Note that it would be best to keep all configuration in the centralized custom configuration, for example by adding -placeholders in the hazelcast.xml file if need be and adding the properties to the centralized configuration file. +Note that it is mandatory to set a different `org.apache.unomi.cluster.nodeId` for each node in the cluster. diff --git a/package/pom.xml b/package/pom.xml index 13af2a5be2..265e919738 100644 --- a/package/pom.xml +++ b/package/pom.xml @@ -80,13 +80,6 @@ xml runtime - - org.apache.karaf.cellar - apache-karaf-cellar - features - xml - runtime - org.apache.unomi @@ -192,17 +185,6 @@ org.apache.unomi.persistence.elasticsearch.cfg - - org.apache.unomi - unomi-services - ${project.version} - hazelcastconfig - xml - - ${project.build.directory}/assembly/etc - - hazelcast.xml - org.apache.unomi unomi-services @@ -362,7 +344,6 @@ system war cxf-jaxrs - cellar aries-blueprint shell-compat unomi-kar diff --git a/package/src/main/resources/etc/custom.system.properties b/package/src/main/resources/etc/custom.system.properties index 5e97437c3e..7b1b576eb0 100644 --- a/package/src/main/resources/etc/custom.system.properties +++ b/package/src/main/resources/etc/custom.system.properties @@ -18,14 +18,6 @@ ################################################################################ # This following file is used to customize system properties for the Apache Unomi application running in Apache Karaf. ${optionals}=unomi.custom.system.properties -####################################################################################################################### -## Hazelcast clustering settings ## -####################################################################################################################### -org.apache.unomi.hazelcast.group.name=${env:UNOMI_HAZELCAST_GROUP_NAME:-cellar} -org.apache.unomi.hazelcast.group.password=${env:UNOMI_HAZELCAST_GROUP_PASSWORD:-pass} -# This list can be comma separated and use ranges such as 192.168.1.0-7,192.168.1.21 -org.apache.unomi.hazelcast.tcp-ip.members=${env:UNOMI_HAZELCAST_TCPIP_MEMBERS:-127.0.0.1} -org.apache.unomi.hazelcast.network.port=${env:UNOMI_HAZELCAST_NETWORK_PORT:-5701} ####################################################################################################################### ## Security settings ## @@ -81,11 +73,14 @@ 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/package/src/main/resources/etc/org.apache.karaf.cellar.groups.cfg b/package/src/main/resources/etc/org.apache.karaf.cellar.groups.cfg deleted file mode 100644 index 9542b735ee..0000000000 --- a/package/src/main/resources/etc/org.apache.karaf.cellar.groups.cfg +++ /dev/null @@ -1,81 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You under the Apache License, Version 2.0 -# (the "License"); you may not use this file except in compliance with -# the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# -# -# This property stores the cluster groups for which the local node is member -# -groups = default - -# -# Filtering of the bundles in the default cluster group -# -default.bundle.whitelist.inbound = * -default.bundle.whitelist.outbound = * -default.bundle.blacklist.inbound = *.xml -default.bundle.blacklist.outbound = *.xml - -# -# Filtering of the configurations in the default cluster group -# -default.config.whitelist.inbound = * -default.config.whitelist.outbound = * -default.config.blacklist.inbound = org.apache.felix.fileinstall*, \ - org.apache.karaf.cellar*, \ - org.apache.karaf.management, \ - org.apache.karaf.shell, \ - org.ops4j.pax.logging, \ - org.ops4j.pax.web, \ - org.apache.aries.transaction, \ - org.apache.unomi.cluster, \ - org.apache.unomi.geonames, \ - org.apache.unomi.persistence.elasticsearch, \ - org.apache.unomi.router, \ - org.apache.unomi.plugins.request -default.config.blacklist.outbound = org.apache.felix.fileinstall*, \ - org.apache.karaf.cellar*, \ - org.apache.karaf.management, \ - org.apache.karaf.shell, \ - org.ops4j.pax.logging, \ - org.ops4j.pax.web, \ - org.apache.aries.transaction, \ - org.apache.unomi.cluster, \ - org.apache.unomi.geonames, \ - org.apache.unomi.persistence.elasticsearch, \ - org.apache.unomi.router, \ - org.apache.unomi.plugins.request - -# -# Filtering of the features in the default cluster group -# -default.feature.whitelist.inbound = * -default.feature.whitelist.outbound = * -default.feature.blacklist.inbound = none -default.feature.blacklist.outbound = none - -# -# The following properties define the behavior to use when the node joins the cluster (the usage of the bootstrap -# synchronizer), per cluster group and per resource. -# The following values are accepted: -# disabled: means that the synchronizer is not used, meaning the node or the cluster are not updated at all -# cluster: if the node is the first one in the cluster, it pushes its local state to the cluster, else it's not the -# first node of the cluster, the node will update its local state with the cluster one (meaning that the cluster -# is the master) -# node: in this case, the node is the master, it means that the cluster state will be overwritten by the node state. -# -default.bundle.sync = disabled -default.config.sync = disabled -default.feature.sync = disabled -default.obr.urls.sync = disabled diff --git a/persistence-elasticsearch/core/pom.xml b/persistence-elasticsearch/core/pom.xml index ee5afc77f7..810e38807c 100644 --- a/persistence-elasticsearch/core/pom.xml +++ b/persistence-elasticsearch/core/pom.xml @@ -192,12 +192,6 @@ junit test - - com.hazelcast - hazelcast-all - 3.12.8 - provided - org.apache.unomi @@ -226,7 +220,6 @@ com.google.apphosting.api;resolution:=optional, com.google.common.geometry;resolution:=optional, com.google.errorprone.annotations.concurrent;resolution:=optional, - com.hazelcast.core;version="[3.12,4)";resolution:=optional, com.lmax.disruptor;resolution:=optional, com.lmax.disruptor.dsl;resolution:=optional, com.sun.management;resolution:=optional, diff --git a/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java b/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java index 27e874e523..372fb6950e 100644 --- a/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java +++ b/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java @@ -40,6 +40,7 @@ import org.apache.unomi.persistence.elasticsearch.conditions.*; import org.apache.unomi.persistence.spi.PersistenceService; import org.apache.unomi.persistence.spi.aggregate.*; +import org.apache.unomi.persistence.spi.config.ConfigurationUpdateHelper; import org.elasticsearch.ElasticsearchStatusException; import org.elasticsearch.Version; import org.elasticsearch.action.DocWriteRequest; @@ -112,6 +113,7 @@ import org.elasticsearch.search.sort.SortOrder; import org.elasticsearch.tasks.TaskId; import org.osgi.framework.*; +import org.osgi.service.cm.ManagedService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -134,7 +136,7 @@ import static org.elasticsearch.index.query.QueryBuilders.termQuery; @SuppressWarnings("rawtypes") -public class ElasticSearchPersistenceServiceImpl implements PersistenceService, SynchronousBundleListener { +public class ElasticSearchPersistenceServiceImpl implements PersistenceService, SynchronousBundleListener, ManagedService { public static final String BULK_PROCESSOR_BULK_SIZE = "bulkProcessor.bulkSize"; public static final String BULK_PROCESSOR_FLUSH_INTERVAL = "bulkProcessor.flushInterval"; @@ -717,6 +719,39 @@ public void bundleChanged(BundleEvent event) { } } + @Override + public void updated(Dictionary properties) { + Map propertyMappings = new HashMap<>(); + + // Boolean properties + propertyMappings.put("throwExceptions", ConfigurationUpdateHelper.booleanProperty(this::setThrowExceptions)); + propertyMappings.put("alwaysOverwrite", ConfigurationUpdateHelper.booleanProperty(this::setAlwaysOverwrite)); + propertyMappings.put("useBatchingForSave", ConfigurationUpdateHelper.booleanProperty(this::setUseBatchingForSave)); + propertyMappings.put("useBatchingForUpdate", ConfigurationUpdateHelper.booleanProperty(this::setUseBatchingForUpdate)); + propertyMappings.put("aggQueryThrowOnMissingDocs", ConfigurationUpdateHelper.booleanProperty(this::setAggQueryThrowOnMissingDocs)); + + // String properties + propertyMappings.put("logLevelRestClient", ConfigurationUpdateHelper.stringProperty(this::setLogLevelRestClient)); + propertyMappings.put("clientSocketTimeout", ConfigurationUpdateHelper.stringProperty(this::setClientSocketTimeout)); + propertyMappings.put("taskWaitingTimeout", ConfigurationUpdateHelper.stringProperty(this::setTaskWaitingTimeout)); + propertyMappings.put("taskWaitingPollingInterval", ConfigurationUpdateHelper.stringProperty(this::setTaskWaitingPollingInterval)); + propertyMappings.put("aggQueryMaxResponseSizeHttp", ConfigurationUpdateHelper.stringProperty(this::setAggQueryMaxResponseSizeHttp)); + + // Integer properties + propertyMappings.put("aggregateQueryBucketSize", ConfigurationUpdateHelper.integerProperty(this::setAggregateQueryBucketSize)); + + // Custom property for itemTypeToRefreshPolicy with IOException handling + propertyMappings.put("itemTypeToRefreshPolicy", ConfigurationUpdateHelper.customProperty((value, logger) -> { + try { + setItemTypeToRefreshPolicy(value.toString()); + } catch (IOException e) { + logger.warn("Error setting itemTypeToRefreshPolicy: {}", e.getMessage()); + } + })); + + ConfigurationUpdateHelper.processConfigurationUpdates(properties, LOGGER, "ElasticSearch persistence", propertyMappings); + } + private void loadPredefinedMappings(BundleContext bundleContext, boolean forceUpdateMapping) { Enumeration predefinedMappings = bundleContext.getBundle().findEntries("META-INF/cxs/mappings", "*.json", true); if (predefinedMappings == null) { diff --git a/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/clusterNode.json b/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/clusterNode.json new file mode 100644 index 0000000000..e3cd2f76d9 --- /dev/null +++ b/persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/clusterNode.json @@ -0,0 +1,67 @@ +{ + "dynamic_templates": [ + { + "all": { + "match": "*", + "match_mapping_type": "string", + "mapping": { + "type": "text", + "analyzer": "folding", + "fields": { + "keyword": { + "type": "keyword", + "ignore_above": 256 + } + } + } + } + } + ], + "properties": { + "creationDate": { + "type": "date" + }, + "lastModificationDate": { + "type": "date" + }, + "lastSyncDate": { + "type": "date" + }, + "cpuLoad": { + "type": "double" + }, + "loadAverage": { + "type": "double" + }, + "uptime": { + "type": "long" + }, + "master": { + "type": "boolean" + }, + "data": { + "type": "boolean" + }, + "startTime": { + "type": "long" + }, + "lastHeartbeat": { + "type": "long" + }, + "serverInfo": { + "properties": { + "serverBuildDate": { + "type": "date" + }, + "eventTypes": { + "type": "object", + "enabled": false + }, + "capabilities": { + "type": "object", + "enabled": false + } + } + } + } +} diff --git a/persistence-elasticsearch/core/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/persistence-elasticsearch/core/src/main/resources/OSGI-INF/blueprint/blueprint.xml index d142351719..e376be29fa 100644 --- a/persistence-elasticsearch/core/src/main/resources/OSGI-INF/blueprint/blueprint.xml +++ b/persistence-elasticsearch/core/src/main/resources/OSGI-INF/blueprint/blueprint.xml @@ -23,7 +23,7 @@ http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.1.0 http://aries.apache.org/schemas/blueprint-cm/blueprint-cm-1.1.0.xsd"> + update-strategy="none" placeholder-prefix="${es."> @@ -85,7 +85,11 @@ org.apache.unomi.persistence.spi.PersistenceService org.osgi.framework.SynchronousBundleListener + org.osgi.service.cm.ManagedService + + + properties, Logger logger, + String serviceName, + Map propertyMappings) { + if (properties == null) { + return; + } + + logger.info("{} configuration updated, applying changes without restart", serviceName); + + try { + for (Map.Entry entry : propertyMappings.entrySet()) { + String propertyName = entry.getKey(); + PropertyMapping mapping = entry.getValue(); + + Object value = properties.get(propertyName); + if (value != null) { + try { + mapping.apply(value, logger); + logger.debug("Updated {} to: {}", propertyName, value); + } catch (Exception e) { + logger.warn("Error setting property {}: {}", propertyName, e.getMessage()); + } + } + } + } catch (Exception e) { + logger.error("Error applying configuration updates", e); + } + } + + /** + * Creates a boolean property mapping. + * + * @param setter The setter function to call with the boolean value + * @return PropertyMapping for boolean properties + */ + public static PropertyMapping booleanProperty(Consumer setter) { + return (value, logger) -> { + boolean boolValue = Boolean.parseBoolean(value.toString()); + setter.accept(boolValue); + }; + } + + /** + * Creates a string property mapping. + * + * @param setter The setter function to call with the string value + * @return PropertyMapping for string properties + */ + public static PropertyMapping stringProperty(Consumer setter) { + return (value, logger) -> { + String stringValue = value.toString(); + setter.accept(stringValue); + }; + } + + /** + * Creates an integer property mapping. + * + * @param setter The setter function to call with the integer value + * @return PropertyMapping for integer properties + */ + public static PropertyMapping integerProperty(Consumer setter) { + return (value, logger) -> { + int intValue = Integer.parseInt(value.toString()); + setter.accept(intValue); + }; + } + + /** + * Creates a long property mapping. + * + * @param setter The setter function to call with the long value + * @return PropertyMapping for long properties + */ + public static PropertyMapping longProperty(Consumer setter) { + return (value, logger) -> { + long longValue = Long.parseLong(value.toString()); + setter.accept(longValue); + }; + } + + /** + * Creates a custom property mapping for special cases. + * + * @param processor The custom processor function + * @return PropertyMapping for custom properties + */ + public static PropertyMapping customProperty(PropertyProcessor processor) { + return processor::process; + } + + /** + * Functional interface for property processing. + */ + @FunctionalInterface + public interface PropertyMapping { + /** + * Applies the property value using the appropriate setter. + * + * @param value The property value + * @param logger The logger to use + * @throws Exception if there's an error processing the property + */ + void apply(Object value, Logger logger) throws Exception; + } + + /** + * Functional interface for custom property processing. + */ + @FunctionalInterface + public interface PropertyProcessor { + /** + * Processes the property value. + * + * @param value The property value + * @param logger The logger to use + * @throws Exception if there's an error processing the property + */ + void process(Object value, Logger logger) throws Exception; + } +} diff --git a/pom.xml b/pom.xml index c537486e12..bcef0de13d 100644 --- a/pom.xml +++ b/pom.xml @@ -65,7 +65,6 @@ 2.11.4 2.11.4 4.2.15 - 4.2.1 4.13.5 7.4.2 7.11.0 @@ -830,24 +829,6 @@ xml runtime - - org.apache.karaf.cellar - apache-karaf-cellar - features - ${version.karaf.cellar} - xml - runtime - - - org.apache.karaf.cellar - org.apache.karaf.cellar.core - ${version.karaf.cellar} - - - org.apache.karaf.cellar - org.apache.karaf.cellar.config - ${version.karaf.cellar} - diff --git a/services/pom.xml b/services/pom.xml index c4cf7e4da3..c9fb46411d 100644 --- a/services/pom.xml +++ b/services/pom.xml @@ -56,6 +56,12 @@ ${project.version} provided + + org.apache.unomi + unomi-lifecycle-watcher + ${project.version} + provided + javax.servlet @@ -111,16 +117,6 @@ commons-beanutils commons-beanutils - - org.apache.karaf.cellar - org.apache.karaf.cellar.core - provided - - - org.apache.karaf.cellar - org.apache.karaf.cellar.config - provided - org.apache.unomi @@ -175,6 +171,7 @@ *;scope=compile|runtime sun.misc;resolution:=optional, + com.sun.management;resolution:=optional, * @@ -213,11 +210,6 @@ cfg clustercfg - - src/main/resources/hazelcast.xml - xml - hazelcastconfig - 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 ec4cfe5235..e71973ce35 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,75 +18,114 @@ package org.apache.unomi.services.impl.cluster; import org.apache.commons.lang3.ArrayUtils; -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.commons.lang3.StringUtils; 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.api.services.SchedulerService; +import org.apache.unomi.lifecycle.BundleWatcher; 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 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.TimeUnit; +import java.util.concurrent.*; /** * 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 PersistenceService persistenceService; + private String publicAddress; private String internalAddress; - private Map> nodeSystemStatistics = new ConcurrentHashMap<>(); - private Group group = null; - private SchedulerService schedulerService; - + //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 volatile boolean shutdownNow = false; + private volatile List cachedClusterNodes = Collections.emptyList(); - public void setPersistenceService(PersistenceService persistenceService) { - this.persistenceService = persistenceService; - } + private BundleWatcher bundleWatcher; + private ScheduledFuture updateSystemStatsFuture; + private ScheduledFuture cleanupStaleNodesFuture; - public void setKarafCellarClusterManager(ClusterManager karafCellarClusterManager) { - this.karafCellarClusterManager = karafCellarClusterManager; - } + /** + * Max time to wait for persistence service (in milliseconds) + */ + private static final long MAX_WAIT_TIME = 60000; // 60 seconds - public void setKarafCellarEventProducer(EventProducer karafCellarEventProducer) { - this.karafCellarEventProducer = karafCellarEventProducer; + /** + * 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 setKarafCellarGroupManager(GroupManager karafCellarGroupManager) { - this.karafCellarGroupManager = karafCellarGroupManager; - } + /** + * 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; + } - public void setKarafCellarGroupName(String karafCellarGroupName) { - this.karafCellarGroupName = karafCellarGroupName; + // 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) { + if (persistenceService != 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 setOsgiConfigurationAdmin(ConfigurationAdmin osgiConfigurationAdmin) { - this.osgiConfigurationAdmin = osgiConfigurationAdmin; + /** + * 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 setPublicAddress(String publicAddress) { @@ -101,8 +140,35 @@ 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() { @@ -110,211 +176,345 @@ public Map> getNodeSystemStatistics() { } public void init() { - 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; - } - } + // 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); + } - // check if the producer is ON - if (setupConfigOk && karafCellarEventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) { - LOGGER.error("Cluster event producer is OFF"); - 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 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; - } + nodeStartTime = System.currentTimeMillis(); - 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); + /* 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(); - configurations.put(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION, karafCellarClusterNodeConfiguration); - ClusterConfigurationEvent clusterConfigurationEvent = new ClusterConfigurationEvent(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION); - sendEvent(clusterConfigurationEvent); - } + LOGGER.info("Cluster service initialized with node ID: {}", nodeId); + } - TimerTask statisticsTask = new TimerTask() { - @Override - public void run() { - try { - updateSystemStats(); - } catch (Throwable t) { - LOGGER.error("Error updating system statistics", t); - } + /** + * 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; + } + */ + + // 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); } - }; - schedulerService.getScheduleExecutorService().scheduleWithFixedDelay(statisticsTask, 0, nodeStatisticsUpdateFrequency, TimeUnit.MILLISECONDS); + } + }; + /* Wait for PR UNOMI-878 to reactivate that code + schedulerService.createRecurringTask("clusterNodeStatisticsUpdate", nodeStatisticsUpdateFrequency, TimeUnit.MILLISECONDS, statisticsTask, false); + */ + updateSystemStatsFuture = 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); + } + } + }; + /* Wait for PR UNOMI-878 to reactivate that code + schedulerService.createRecurringTask("clusterStaleNodesCleanup", 60000, TimeUnit.MILLISECONDS, cleanupTask, false); + */ + cleanupStaleNodesFuture = scheduledExecutorService.scheduleAtFixedRate(cleanupTask, 100, 60000, TimeUnit.MILLISECONDS); - } - LOGGER.info("Cluster service initialized."); + LOGGER.info("Cluster service scheduled tasks initialized"); } public void destroy() { - LOGGER.info("Cluster service shutdown."); - } - - @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); - } - 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); + LOGGER.info("Cluster service shutting down..."); + shutdownNow = true; + + // Cancel scheduled tasks + if (updateSystemStatsFuture != null) { + boolean successfullyCancelled = updateSystemStatsFuture.cancel(false); + if (!successfullyCancelled) { + LOGGER.warn("Failed to cancel scheduled task: clusterNodeStatisticsUpdate"); + } else { + LOGGER.info("Scheduled task: clusterNodeStatisticsUpdate cancelled"); } - String internalEndpoint = internalNodeEndpoints.get(karafCellarNode.getId()); - if (internalEndpoint != null) { - clusterNode.setInternalHostAddress(internalEndpoint); + } + if (cleanupStaleNodesFuture != null) { + boolean successfullyCancelled = cleanupStaleNodesFuture.cancel(false); + if (!successfullyCancelled) { + LOGGER.warn("Failed to cancel scheduled task: cleanupStaleNodesFuture"); + } else { + LOGGER.info("Scheduled task: cleanupStaleNodesFuture cancelled"); } - 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)); + } + if (scheduledExecutorService != null) { + scheduledExecutorService.shutdownNow(); + try { + boolean successfullyTerminated = scheduledExecutorService.awaitTermination(10, TimeUnit.SECONDS); + if (!successfullyTerminated) { + LOGGER.warn("Failed to terminate scheduled tasks after 10 seconds..."); + } else { + LOGGER.info("Scheduled tasks terminated"); } + } catch (InterruptedException e) { + LOGGER.error("Error waiting for scheduled tasks to terminate", e); } - clusterNodes.put(karafCellarNode.getId(), clusterNode); } - return new ArrayList(clusterNodes.values()); - } - - @Override - public void purge(Date date) { - persistenceService.purge(date); - } + // Remove node from persistence service + if (persistenceService != null) { + try { + persistenceService.remove(nodeId, ClusterNode.class); + LOGGER.info("Node {} removed from cluster", nodeId); + } catch (Exception e) { + LOGGER.error("Error removing node from cluster", e); + } + } - @Override - public void purge(String scope) { - persistenceService.purge(scope); - } + // Clear references + persistenceService = null; + bundleWatcher = null; - @Override - public void sendEvent(Serializable eventObject) { - Event event = (Event) eventObject; - event.setSourceGroup(group); - event.setSourceNode(karafCellarClusterManager.getNode()); - karafCellarEventProducer.produce(event); + LOGGER.info("Cluster service shutdown."); } /** - * 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. + * Register this node in the persistence service */ - 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 Map getMapProperty(Properties properties, String propertyName, String defaultValue) { - String propertyValue = properties.getProperty(propertyName, defaultValue); - return getMapProperty(propertyValue); - } - - 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]); + private void registerNodeInPersistence() { + if (persistenceService == null) { + LOGGER.error("Cannot register node: PersistenceService not available"); + return; } - return propertyMapValue; - } - 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(","); - } + 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"); } - String oldPropertyValue = (String) properties.setProperty(propertyName, propertyValueBuilder.toString()); - if (oldPropertyValue == null) { - return null; + + updateSystemStatsForNode(clusterNode); + + boolean success = persistenceService.save(clusterNode); + if (success) { + LOGGER.info("Node {} registered in cluster", nodeId); + } else { + LOGGER.error("Failed to register node {} in cluster", nodeId); } - return getMapProperty(oldPropertyValue); } - private void updateSystemStats() { + /** + * Updates system stats for the given cluster node + */ + private void updateSystemStatsForNode(ClusterNode node) { final RuntimeMXBean remoteRuntime = ManagementFactory.getRuntimeMXBean(); long uptime = remoteRuntime.getUptime(); - ObjectName operatingSystemMXBeanName = ManagementFactory.getOperatingSystemMXBean().getObjectName(); - Double systemCpuLoad = null; + + double systemCpuLoad = 0.0; try { - systemCpuLoad = (Double) ManagementFactory.getPlatformMBeanServer().getAttribute(operatingSystemMXBeanName, "SystemCpuLoad"); - } catch (MBeanException | AttributeNotFoundException | InstanceNotFoundException | ReflectionException e) { - LOGGER.error("Error retrieving system CPU load", e); + 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); } + 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); - clusterSystemStatisticsEvent.setStatistics(systemStatistics); - nodeSystemStatistics.put(karafCellarClusterManager.getNode().getId(), systemStatistics); - sendEvent(clusterSystemStatisticsEvent); + nodeSystemStatistics.put(nodeId, systemStatistics); + } + + /** + * Updates the system statistics for this node and stores them in the persistence service + */ + private void updateSystemStats() { + if (shutdownNow) { + return; + } + + if (persistenceService == null) { + LOGGER.warn("Cannot update system stats: PersistenceService not available"); + return; + } + + // Load node from persistence + ClusterNode node = persistenceService.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 = persistenceService.save(node); + if (!success) { + LOGGER.error("Failed to update node {} statistics", nodeId); + } + + // Always refresh cluster nodes cache after attempting stats update + try { + List nodes = persistenceService.getAllItems(ClusterNode.class, 0, -1, null).getList(); + cachedClusterNodes = nodes; + } catch (Exception e) { + LOGGER.warn("Failed to refresh cluster nodes cache during stats update", e); + } + } 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; + } + + if (persistenceService == 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 = persistenceService.query(staleNodesCondition, null, ClusterNode.class, 0, -1); + + for (ClusterNode staleNode : staleNodes.getList()) { + LOGGER.info("Removing stale node: {}", staleNode.getItemId()); + persistenceService.remove(staleNode.getItemId(), ClusterNode.class); + nodeSystemStatistics.remove(staleNode.getItemId()); + } + } + + @Override + public List getClusterNodes() { + // Return cached cluster nodes, creating a defensive copy + return cachedClusterNodes.isEmpty() ? Collections.emptyList() : new ArrayList<>(cachedClusterNodes); } + @Override + public void purge(Date date) { + if (persistenceService == null) { + LOGGER.warn("Cannot purge by date: PersistenceService not available"); + return; + } + + persistenceService.purge(date); + } + + @Override + public void purge(String scope) { + if (persistenceService == null) { + LOGGER.warn("Cannot purge by scope: PersistenceService not available"); + return; + } + + persistenceService.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 persistenceService != null; + } } + diff --git a/services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterSystemStatisticsEvent.java b/services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterSystemStatisticsEvent.java deleted file mode 100644 index 3c4ec5ad77..0000000000 --- a/services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterSystemStatisticsEvent.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.unomi.services.impl.cluster; - -import org.apache.karaf.cellar.core.event.Event; - -import java.io.Serializable; -import java.util.Map; -import java.util.TreeMap; - -/** - * The cluster event used to transmit update to node system statistics. - */ -public class ClusterSystemStatisticsEvent extends Event { - - Map statistics = new TreeMap<>(); - - public ClusterSystemStatisticsEvent(String id) { - super(id); - } - - public Map getStatistics() { - return statistics; - } - - public void setStatistics(Map statistics) { - this.statistics = statistics; - } -} diff --git a/services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterSystemStatisticsEventHandler.java b/services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterSystemStatisticsEventHandler.java deleted file mode 100644 index 9eecaf4ba9..0000000000 --- a/services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterSystemStatisticsEventHandler.java +++ /dev/null @@ -1,138 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.unomi.services.impl.cluster; - -import org.apache.karaf.cellar.config.Constants; -import org.apache.karaf.cellar.core.CellarSupport; -import org.apache.karaf.cellar.core.Configurations; -import org.apache.karaf.cellar.core.Group; -import org.apache.karaf.cellar.core.control.BasicSwitch; -import org.apache.karaf.cellar.core.control.Switch; -import org.apache.karaf.cellar.core.control.SwitchStatus; -import org.apache.karaf.cellar.core.event.EventHandler; -import org.apache.karaf.cellar.core.event.EventType; -import org.osgi.service.cm.Configuration; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.Serializable; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -/** - * A Karaf Cellar event handler to process incoming events that contain system statistics updates from nodes. - */ -public class ClusterSystemStatisticsEventHandler extends CellarSupport implements EventHandler { - - public static final String SWITCH_ID = "org.apache.unomi.cluster.system.statistics.handler"; - private static final Logger LLOGGER = LoggerFactory.getLogger(ClusterSystemStatisticsEventHandler.class.getName()); - private final Switch eventSwitch = new BasicSwitch(SWITCH_ID); - private ClusterServiceImpl clusterServiceImpl; - - public void setClusterServiceImpl(ClusterServiceImpl clusterServiceImpl) { - this.clusterServiceImpl = clusterServiceImpl; - } - - public void init() { - // nothing to do - } - - public void destroy() { - // nothing to do - } - - @Override - public void handle(ClusterSystemStatisticsEvent event) { - // check if the handler is ON - if (this.getSwitch().getStatus().equals(SwitchStatus.OFF)) { - LLOGGER.debug("CELLAR SYSTEM STATS: {} switch is OFF, cluster event not handled", SWITCH_ID); - return; - } - - if (groupManager == null) { - //in rare cases for example right after installation this happens! - LLOGGER.error("CELLAR SYSTEM STATS: retrieved event {} while groupManager is not available yet!", event); - return; - } - - // check if the group is local - if (!groupManager.isLocalGroup(event.getSourceGroup().getName())) { - LLOGGER.info("CELLAR SYSTEM STATS: node is not part of the event cluster group {}",event.getSourceGroup().getName()); - return; - } - - Group group = event.getSourceGroup(); - String groupName = group.getName(); - - String pid = event.getId(); - - if (isAllowed(event.getSourceGroup(), Constants.CATEGORY, pid, EventType.INBOUND)) { - - // check if it's not a "local" event - if (event.getSourceNode() != null && event.getSourceNode().getId().equalsIgnoreCase(clusterManager.getNode().getId())) { - LLOGGER.trace("CELLAR SYSTEM STATS: cluster event is local (coming from local synchronizer or listener)"); - return; - } - - - Map nodeSystemStatistics = clusterServiceImpl.getNodeSystemStatistics().get(event.getSourceNode().getId()); - if (nodeSystemStatistics == null) { - nodeSystemStatistics = new ConcurrentHashMap<>(); - } - nodeSystemStatistics.putAll(event.getStatistics()); - clusterServiceImpl.getNodeSystemStatistics().put(event.getSourceNode().getId(), nodeSystemStatistics); - } - - } - - @Override - public Class getType() { - return ClusterSystemStatisticsEvent.class; - } - - /** - * Get the cluster configuration event handler switch. - * - * @return the cluster configuration event handler switch. - */ - @Override - public Switch getSwitch() { - // load the switch status from the config - try { - Configuration configuration = configurationAdmin.getConfiguration(Configurations.NODE, null); - if (configuration != null) { - String handlerStatus = (String) configuration.getProperties().get(Configurations.HANDLER + "." + this.getClass().getName()); - if (handlerStatus == null) { - // default value is on. - eventSwitch.turnOn(); - } else { - Boolean status = new Boolean(handlerStatus); - if (status) { - eventSwitch.turnOn(); - } else { - eventSwitch.turnOff(); - } - } - } - } catch (Exception e) { - // nothing to do - } - return eventSwitch; - } - - -} diff --git a/services/src/main/java/org/apache/unomi/services/impl/rules/RulesServiceImpl.java b/services/src/main/java/org/apache/unomi/services/impl/rules/RulesServiceImpl.java index 6702f5689d..fe3fe8fb02 100644 --- a/services/src/main/java/org/apache/unomi/services/impl/rules/RulesServiceImpl.java +++ b/services/src/main/java/org/apache/unomi/services/impl/rules/RulesServiceImpl.java @@ -30,9 +30,11 @@ import org.apache.unomi.api.services.*; import org.apache.unomi.persistence.spi.CustomObjectMapper; import org.apache.unomi.persistence.spi.PersistenceService; +import org.apache.unomi.persistence.spi.config.ConfigurationUpdateHelper; import org.apache.unomi.services.actions.ActionExecutorDispatcher; import org.apache.unomi.api.utils.ParserHelper; import org.osgi.framework.*; +import org.osgi.service.cm.ManagedService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -43,7 +45,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; -public class RulesServiceImpl implements RulesService, EventListenerService, SynchronousBundleListener { +public class RulesServiceImpl implements RulesService, EventListenerService, SynchronousBundleListener, ManagedService { public static final String TRACKED_PARAMETER = "trackedConditionParameters"; private static final Logger LOGGER = LoggerFactory.getLogger(RulesServiceImpl.class.getName()); @@ -105,6 +107,20 @@ public void setOptimizedRulesActivated(Boolean optimizedRulesActivated) { this.optimizedRulesActivated = optimizedRulesActivated; } + @Override + public void updated(Dictionary properties) { + Map propertyMappings = new HashMap<>(); + + // Boolean properties + propertyMappings.put("rules.optimizationActivated", ConfigurationUpdateHelper.booleanProperty(this::setOptimizedRulesActivated)); + + // Integer properties + propertyMappings.put("rules.refresh.interval", ConfigurationUpdateHelper.integerProperty(this::setRulesRefreshInterval)); + propertyMappings.put("rules.statistics.refresh.interval", ConfigurationUpdateHelper.integerProperty(this::setRulesStatisticsRefreshInterval)); + + ConfigurationUpdateHelper.processConfigurationUpdates(properties, LOGGER, "Rules service", propertyMappings); + } + public void postConstruct() { LOGGER.debug("postConstruct {{}}", bundleContext.getBundle()); diff --git a/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml index f49f3c8497..bf64a77d72 100644 --- a/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml +++ b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml @@ -23,7 +23,7 @@ http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.1.0 http://aries.apache.org/schemas/blueprint-cm/blueprint-cm-1.1.0.xsd"> + update-strategy="none" placeholder-prefix="${services."> @@ -53,7 +53,7 @@ - + @@ -61,7 +61,7 @@ + update-strategy="none" placeholder-prefix="${es."> @@ -70,12 +70,10 @@ - - - + @@ -183,7 +181,11 @@ org.apache.unomi.api.services.RulesService org.apache.unomi.api.services.EventListenerService org.osgi.framework.SynchronousBundleListener + org.osgi.service.cm.ManagedService + + + + - - - - - + + + @@ -411,7 +413,6 @@ - @@ -429,19 +430,4 @@ - - - - - - - - - - - - - diff --git a/services/src/main/resources/hazelcast.xml b/services/src/main/resources/hazelcast.xml deleted file mode 100644 index 2179c92c6c..0000000000 --- a/services/src/main/resources/hazelcast.xml +++ /dev/null @@ -1,222 +0,0 @@ - - - - - ${org.apache.unomi.hazelcast.group.name} - ${org.apache.unomi.hazelcast.group.password} - - http://localhost:8080/mancenter - - true - - - ${org.apache.unomi.hazelcast.network.port} - - - 0 - - - - 224.2.2.3 - 54327 - - - ${org.apache.unomi.hazelcast.tcp-ip.members} - - - my-access-key - my-secret-key - - us-west-1 - - ec2.amazonaws.com - - hazelcast-sg - type - hz-nodes - - - - 10.10.1.* - - - - - - PBEWithMD5AndDES - - thesalt - - thepass - - 19 - - - - - 16 - - 0 - - - - 0 - - 1 - - 0 - -1 - - - - - BINARY - - 1 - - 0 - - 0 - - 0 - - NONE - - 0 - - 25 - - com.hazelcast.map.merge.PassThroughMergePolicy - - - - 1 - SET - - - - 1 - SET - - - - 1 - - - - 1 - - - - 0 - - 0 - 0 - 1000 - true - CANCEL_RUNNING_OPERATION - - - - 0 - 1 - 0 - - - - 0 - - - - \ No newline at end of file diff --git a/services/src/main/resources/org.apache.unomi.cluster.cfg b/services/src/main/resources/org.apache.unomi.cluster.cfg index bfbb189d9e..eecb7e1dec 100644 --- a/services/src/main/resources/org.apache.unomi.cluster.cfg +++ b/services/src/main/resources/org.apache.unomi.cluster.cfg @@ -14,14 +14,18 @@ # 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 nodeStatisticsUpdateFrequency controls the frequency of the update of system statistics such as CPU load, +# 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, # 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} \ No newline at end of file +nodeStatisticsUpdateFrequency=${org.apache.unomi.cluster.nodeStatisticsUpdateFrequency:-10000}