From efb8266aee22849ff30e3a08f271d9e07298cbd5 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Fri, 22 Aug 2025 15:51:56 +0200 Subject: [PATCH 1/9] UNOMI-877: Remove all reference to cellar and hazelcast. UNOMI-877: Not sending event to cluster anymore. UNOMI-877: Replace Karaf Cellar and Hazelcast with PersistenceService for cluster synchronization (code isolated from branch unomi-3-dev made by Serge Huber) --- .../org/apache/unomi/api/ClusterNode.java | 65 +- .../unomi/api/services/ClusterService.java | 7 - extensions/router/router-core/pom.xml | 10 - .../core/context/RouterCamelContext.java | 22 - .../core/event/UpdateCamelRouteEvent.java | 38 - .../event/UpdateCamelRouteEventHandler.java | 76 -- .../OSGI-INF/blueprint/blueprint.xml | 16 +- .../java/org/apache/unomi/itests/BaseIT.java | 5 - kar/pom.xml | 7 - kar/src/main/feature/feature.xml | 3 - .../main/asciidoc/building-and-deploying.adoc | 3 +- manual/src/main/asciidoc/clustering.adoc | 22 +- package/pom.xml | 19 - .../resources/etc/custom.system.properties | 13 +- .../etc/org.apache.karaf.cellar.groups.cfg | 81 -- persistence-elasticsearch/core/pom.xml | 7 - pom.xml | 19 - services/pom.xml | 22 +- .../impl/cluster/ClusterServiceImpl.java | 690 +++++++++++++----- .../cluster/ClusterSystemStatisticsEvent.java | 43 -- .../ClusterSystemStatisticsEventHandler.java | 138 ---- .../OSGI-INF/blueprint/blueprint.xml | 29 +- services/src/main/resources/hazelcast.xml | 222 ------ .../resources/org.apache.unomi.cluster.cfg | 10 +- 24 files changed, 583 insertions(+), 984 deletions(-) delete mode 100644 extensions/router/router-core/src/main/java/org/apache/unomi/router/core/event/UpdateCamelRouteEvent.java delete mode 100644 extensions/router/router-core/src/main/java/org/apache/unomi/router/core/event/UpdateCamelRouteEventHandler.java delete mode 100644 package/src/main/resources/etc/org.apache.karaf.cellar.groups.cfg delete mode 100644 services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterSystemStatisticsEvent.java delete mode 100644 services/src/main/java/org/apache/unomi/services/impl/cluster/ClusterSystemStatisticsEventHandler.java delete mode 100644 services/src/main/resources/hazelcast.xml 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..e096490b9b 100644 --- a/api/src/main/java/org/apache/unomi/api/ClusterNode.java +++ b/api/src/main/java/org/apache/unomi/api/ClusterNode.java @@ -22,10 +22,12 @@ /** * 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 +35,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 +174,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..ad87775037 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,11 +51,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..c3f03876ea 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,7 +31,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 +74,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 +100,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 +234,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 +254,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 +274,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..c2e73abe7c 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 @@ -120,18 +120,6 @@ - - - - - - - - - - - - @@ -141,8 +129,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..483bacd8a3 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), 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/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..b4bc2c421e 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,28 +18,30 @@ 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.osgi.framework.BundleContext; +import org.osgi.framework.ServiceReference; +import org.osgi.util.tracker.ServiceTracker; +import org.osgi.util.tracker.ServiceTrackerCustomizer; 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; /** @@ -47,46 +49,185 @@ */ 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; + + /** + * 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; + 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 BundleContext bundleContext; + private volatile boolean shutdownNow = false; - public void setPersistenceService(PersistenceService persistenceService) { - this.persistenceService = persistenceService; + private BundleWatcher bundleWatcher; + + /** + * Max time to wait for persistence service (in milliseconds) + */ + private static final long MAX_WAIT_TIME = 60000; // 60 seconds + + /** + * 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 setKarafCellarClusterManager(ClusterManager karafCellarClusterManager) { - this.karafCellarClusterManager = karafCellarClusterManager; + /** + * 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 setKarafCellarEventProducer(EventProducer karafCellarEventProducer) { - this.karafCellarEventProducer = karafCellarEventProducer; + /** + * 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 setKarafCellarGroupManager(GroupManager karafCellarGroupManager) { - this.karafCellarGroupManager = karafCellarGroupManager; + /** + * 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 setKarafCellarGroupName(String karafCellarGroupName) { - this.karafCellarGroupName = karafCellarGroupName; + /** + * 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 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 +242,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 +278,331 @@ 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; - } - } - - // check if the producer is ON - if (setupConfigOk && karafCellarEventProducer.getSwitch().getStatus().equals(SwitchStatus.OFF)) { - LOGGER.error("Cluster event producer is OFF"); - setupConfigOk = false; - } - - // check if the config pid is allowed - if (setupConfigOk && !isClusterConfigPIDAllowed(group, Constants.CATEGORY, KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION, EventType.OUTBOUND)) { - LOGGER.error("Configuration PID " + KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION + " is blocked outbound for cluster group {}", - karafCellarGroupName); - setupConfigOk = false; - } - - if (setupConfigOk) { - Map 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); + // 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); + } - 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 persistence service to be available + try { + waitForPersistenceService(); + } catch (IllegalStateException e) { + LOGGER.error("Failed to initialize cluster service: {}", e.getMessage()); + return; + } - configurations.put(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION, karafCellarClusterNodeConfiguration); - ClusterConfigurationEvent clusterConfigurationEvent = new ClusterConfigurationEvent(KARAF_CELLAR_CLUSTER_NODE_CONFIGURATION); - sendEvent(clusterConfigurationEvent); - } + nodeStartTime = System.currentTimeMillis(); - 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); + // Register this node in the persistence service + registerNodeInPersistence(); + /* 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."); } - LOGGER.info("Cluster service initialized."); - } + */ + initializeScheduledTasks(); - public void destroy() { - LOGGER.info("Cluster service shutdown."); + LOGGER.info("Cluster service initialized with node ID: {}", nodeId); } - @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); + /** + * 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; } - for (org.apache.karaf.cellar.core.Node karafCellarNode : karafCellarNodes) { - ClusterNode clusterNode = new ClusterNode(); - String publicEndpoint = publicNodeEndpoints.get(karafCellarNode.getId()); - if (publicEndpoint != null) { - clusterNode.setPublicHostAddress(publicEndpoint); - } - String internalEndpoint = internalNodeEndpoints.get(karafCellarNode.getId()); - if (internalEndpoint != null) { - clusterNode.setInternalHostAddress(internalEndpoint); - } - 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); + */ + + // 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); } - 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("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); } } - clusterNodes.put(karafCellarNode.getId(), clusterNode); - } + }; + /* Wait for PR UNOMI-878 to reactivate that code + schedulerService.createRecurringTask("clusterStaleNodesCleanup", 60000, TimeUnit.MILLISECONDS, cleanupTask, false); + */ + scheduledExecutorService.scheduleAtFixedRate(cleanupTask, 100, 60000, TimeUnit.MILLISECONDS); - return new ArrayList(clusterNodes.values()); + LOGGER.info("Cluster service scheduled tasks initialized"); } - @Override - public void purge(Date date) { - persistenceService.purge(date); - } + 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); + } + } - @Override - public void purge(String scope) { - persistenceService.purge(scope); - } + // 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 sendEvent(Serializable eventObject) { - Event event = (Event) eventObject; - event.setSourceGroup(group); - event.setSourceNode(karafCellarClusterManager.getNode()); - karafCellarEventProducer.produce(event); + // Clear references + persistenceService = null; + bundleWatcher = null; + + 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() { + PersistenceService service = getPersistenceService(); + if (service == 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 = service.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; + } + + 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(); } + @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/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/resources/OSGI-INF/blueprint/blueprint.xml b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml index f49f3c8497..e2e1884648 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 @@ - + @@ -70,9 +70,6 @@ - - - @@ -262,13 +259,11 @@ - - - - - + + @@ -411,7 +406,6 @@ - @@ -429,19 +423,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} From 164b8644b9e5d30c2576ebb0c54035168929ce15 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Fri, 22 Aug 2025 15:57:30 +0200 Subject: [PATCH 2/9] Remove unused imports from ClusterNode, RouterCamelContext, and ClusterService. --- api/src/main/java/org/apache/unomi/api/ClusterNode.java | 2 -- .../main/java/org/apache/unomi/api/services/ClusterService.java | 1 - .../apache/unomi/router/core/context/RouterCamelContext.java | 1 - 3 files changed, 4 deletions(-) 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..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,8 +17,6 @@ package org.apache.unomi.api; -import java.io.Serializable; - /** * Information about a cluster node. */ 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..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; 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 c3f03876ea..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; From ebbf4f66af9b7200216fd2dfedc1d41e3e038157 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Sat, 23 Aug 2025 08:59:10 +0200 Subject: [PATCH 3/9] Remove unused `clusterService` property from blueprint.xml. --- .../src/main/resources/OSGI-INF/blueprint/blueprint.xml | 1 - 1 file changed, 1 deletion(-) 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 c2e73abe7c..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 @@ - From ef38a6d96eff4dd184296b0c653a857833c21c09 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Sun, 24 Aug 2025 09:01:43 +0200 Subject: [PATCH 4/9] Add `bundleWatcher` reference and properties to blueprint.xml. --- services/src/main/resources/OSGI-INF/blueprint/blueprint.xml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml index e2e1884648..69c7898865 100644 --- a/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml +++ b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml @@ -73,6 +73,7 @@ + @@ -261,6 +262,8 @@ + + From e8f2a8e0b3872e56fee959e7d25585834c8b1fb7 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 25 Aug 2025 13:12:20 +0200 Subject: [PATCH 5/9] Add clusterNode mapping definition for Elasticsearch. --- .../META-INF/cxs/mappings/clusterNode.json | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 persistence-elasticsearch/core/src/main/resources/META-INF/cxs/mappings/clusterNode.json 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 + } + } + } + } +} From 56515e821e7a8db623345a48fae016438ebac46b Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 25 Aug 2025 15:45:55 +0200 Subject: [PATCH 6/9] Reorder `shutdownNow` to ensure proper node removal from PersistenceService during shutdown. --- .../unomi/services/impl/cluster/ClusterServiceImpl.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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..a83012ce87 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 @@ -363,9 +363,7 @@ public void run() { } public void destroy() { - shutdownNow = true; - - // Remove this node from the persistence service + // Remove this node from the persistence service BEFORE setting shutdownNow PersistenceService service = getPersistenceService(); if (service != null) { try { @@ -376,6 +374,8 @@ public void destroy() { } } + shutdownNow = true; + // Close service trackers if (persistenceServiceTracker != null) { try { From ca6d3e39bf5aee0d57a76d7ece847c10f6d7219c Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Mon, 25 Aug 2025 16:13:13 +0200 Subject: [PATCH 7/9] Clarify comment on node removal timing in `destroy` method. --- .../apache/unomi/services/impl/cluster/ClusterServiceImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 a83012ce87..80fcc6397c 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 @@ -363,7 +363,7 @@ public void run() { } public void destroy() { - // Remove this node from the persistence service BEFORE setting shutdownNow + // Remove this node from the persistence service BEFORE setting shutdownNow otherwise it won't work PersistenceService service = getPersistenceService(); if (service != null) { try { From 324349538ac345601f62ad8cfedf7811ca6fa1b0 Mon Sep 17 00:00:00 2001 From: Serge Huber Date: Tue, 26 Aug 2025 17:22:46 +0200 Subject: [PATCH 8/9] Refactor configuration handling and remove unused OSGi ServiceTracker references. --- .../java/org/apache/unomi/itests/BaseIT.java | 48 ++++- .../apache/unomi/itests/ProfileServiceIT.java | 4 +- .../apache/unomi/itests/RuleServiceIT.java | 4 +- .../ElasticSearchPersistenceServiceImpl.java | 37 +++- .../OSGI-INF/blueprint/blueprint.xml | 6 +- .../spi/config/ConfigurationUpdateHelper.java | 160 ++++++++++++++ .../impl/cluster/ClusterServiceImpl.java | 204 +++++------------- .../services/impl/rules/RulesServiceImpl.java | 18 +- .../OSGI-INF/blueprint/blueprint.xml | 10 +- 9 files changed, 326 insertions(+), 165 deletions(-) create mode 100644 persistence-spi/src/main/java/org/apache/unomi/persistence/spi/config/ConfigurationUpdateHelper.java 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 483bacd8a3..a98d90d7db 100644 --- a/itests/src/test/java/org/apache/unomi/itests/BaseIT.java +++ b/itests/src/test/java/org/apache/unomi/itests/BaseIT.java @@ -378,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<>(); @@ -385,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/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/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/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 80fcc6397c..c61096f4e7 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 @@ -27,10 +27,6 @@ import org.apache.unomi.api.services.ClusterService; import org.apache.unomi.lifecycle.BundleWatcher; 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.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -39,10 +35,7 @@ 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; +import java.util.concurrent.*; /** * Implementation of the persistence service interface @@ -51,23 +44,6 @@ public class ClusterServiceImpl implements ClusterService { 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; private String publicAddress; @@ -78,24 +54,17 @@ public class ClusterServiceImpl implements ClusterService { private long nodeStartTime; private long nodeStatisticsUpdateFrequency = 10000; private Map> nodeSystemStatistics = new ConcurrentHashMap<>(); - private BundleContext bundleContext; private volatile boolean shutdownNow = false; private BundleWatcher bundleWatcher; + private ScheduledFuture updateSystemStatsFuture; + private ScheduledFuture cleanupStaleNodesFuture; /** * Max time to wait for persistence service (in milliseconds) */ private static final long MAX_WAIT_TIME = 60000; // 60 seconds - /** - * 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; - } - /** * Sets the bundle watcher used to retrieve server information * @@ -124,24 +93,12 @@ private void waitForPersistenceService() { 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) { + if (persistenceService != null) { LOGGER.info("Persistence service is now available"); return; } @@ -161,66 +118,6 @@ private void waitForPersistenceService() { throw new IllegalStateException("PersistenceService not available after waiting " + MAX_WAIT_TIME + "ms"); } - /** - * 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; - } - - /** - * 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(); - } - } - /** * For unit tests and backward compatibility - directly sets the persistence service * @param persistenceService the persistence service to set @@ -278,9 +175,6 @@ 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."; @@ -341,7 +235,7 @@ public void run() { /* Wait for PR UNOMI-878 to reactivate that code schedulerService.createRecurringTask("clusterNodeStatisticsUpdate", nodeStatisticsUpdateFrequency, TimeUnit.MILLISECONDS, statisticsTask, false); */ - scheduledExecutorService.scheduleAtFixedRate(statisticsTask, 100, nodeStatisticsUpdateFrequency, TimeUnit.MILLISECONDS); + updateSystemStatsFuture = scheduledExecutorService.scheduleAtFixedRate(statisticsTask, 100, nodeStatisticsUpdateFrequency, TimeUnit.MILLISECONDS); // Schedule cleanup of stale nodes TimerTask cleanupTask = new TimerTask() { @@ -357,34 +251,54 @@ public void run() { /* Wait for PR UNOMI-878 to reactivate that code schedulerService.createRecurringTask("clusterStaleNodesCleanup", 60000, TimeUnit.MILLISECONDS, cleanupTask, false); */ - scheduledExecutorService.scheduleAtFixedRate(cleanupTask, 100, 60000, TimeUnit.MILLISECONDS); + cleanupStaleNodesFuture = scheduledExecutorService.scheduleAtFixedRate(cleanupTask, 100, 60000, TimeUnit.MILLISECONDS); LOGGER.info("Cluster service scheduled tasks initialized"); } public void destroy() { - // Remove this node from the persistence service BEFORE setting shutdownNow otherwise it won't work - PersistenceService service = getPersistenceService(); - if (service != null) { + 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"); + } + } + 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"); + } + } + if (scheduledExecutorService != null) { + scheduledExecutorService.shutdownNow(); try { - service.remove(nodeId, ClusterNode.class); - LOGGER.info("Node {} removed from cluster", nodeId); - } catch (Exception e) { - LOGGER.error("Error removing node from cluster", e); + 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); } } - shutdownNow = true; - - // Close service trackers - if (persistenceServiceTracker != null) { + // Remove node from persistence service + if (persistenceService != null) { try { - persistenceServiceTracker.close(); - LOGGER.debug("Persistence service tracker closed"); + persistenceService.remove(nodeId, ClusterNode.class); + LOGGER.info("Node {} removed from cluster", nodeId); } catch (Exception e) { - LOGGER.debug("Error closing persistence service tracker: {}", e.getMessage()); + LOGGER.error("Error removing node from cluster", e); } - persistenceServiceTracker = null; } // Clear references @@ -398,8 +312,7 @@ public void destroy() { * Register this node in the persistence service */ private void registerNodeInPersistence() { - PersistenceService service = getPersistenceService(); - if (service == null) { + if (persistenceService == null) { LOGGER.error("Cannot register node: PersistenceService not available"); return; } @@ -423,7 +336,7 @@ private void registerNodeInPersistence() { updateSystemStatsForNode(clusterNode); - boolean success = service.save(clusterNode); + boolean success = persistenceService.save(clusterNode); if (success) { LOGGER.info("Node {} registered in cluster", nodeId); } else { @@ -481,14 +394,13 @@ private void updateSystemStats() { return; } - PersistenceService service = getPersistenceService(); - if (service == null) { + if (persistenceService == null) { LOGGER.warn("Cannot update system stats: PersistenceService not available"); return; } // Load node from persistence - ClusterNode node = service.load(nodeId, ClusterNode.class); + ClusterNode node = persistenceService.load(nodeId, ClusterNode.class); if (node == null) { LOGGER.warn("Node {} not found in persistence, re-registering", nodeId); registerNodeInPersistence(); @@ -515,7 +427,7 @@ private void updateSystemStats() { node.setLastHeartbeat(System.currentTimeMillis()); // Save back to persistence - boolean success = service.save(node); + boolean success = persistenceService.save(node); if (!success) { LOGGER.error("Failed to update node {} statistics", nodeId); } @@ -532,8 +444,7 @@ private void cleanupStaleNodes() { return; } - PersistenceService service = getPersistenceService(); - if (service == null) { + if (persistenceService == null) { LOGGER.warn("Cannot cleanup stale nodes: PersistenceService not available"); return; } @@ -552,47 +463,44 @@ private void cleanupStaleNodes() { staleNodesCondition.setParameter("comparisonOperator", "lessThan"); staleNodesCondition.setParameter("propertyValueInteger", cutoffTime); - PartialList staleNodes = service.query(staleNodesCondition, null, ClusterNode.class, 0, -1); + PartialList staleNodes = persistenceService.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); + persistenceService.remove(staleNode.getItemId(), ClusterNode.class); nodeSystemStatistics.remove(staleNode.getItemId()); } } @Override public List getClusterNodes() { - PersistenceService service = getPersistenceService(); - if (service == null) { + if (persistenceService == 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(); + return persistenceService.getAllItems(ClusterNode.class, 0, -1, null).getList(); } @Override public void purge(Date date) { - PersistenceService service = getPersistenceService(); - if (service == null) { + if (persistenceService == null) { LOGGER.warn("Cannot purge by date: PersistenceService not available"); return; } - service.purge(date); + persistenceService.purge(date); } @Override public void purge(String scope) { - PersistenceService service = getPersistenceService(); - if (service == null) { + if (persistenceService == null) { LOGGER.warn("Cannot purge by scope: PersistenceService not available"); return; } - service.purge(scope); + persistenceService.purge(scope); } /** @@ -602,7 +510,7 @@ public void purge(String scope) { * @return true if a persistence service is available (either directly set or via tracker) */ public boolean isPersistenceServiceAvailable() { - return getPersistenceService() != null; + return persistenceService != null; } } 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 69c7898865..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."> @@ -61,7 +61,7 @@ + update-strategy="none" placeholder-prefix="${es."> @@ -181,7 +181,11 @@ org.apache.unomi.api.services.RulesService org.apache.unomi.api.services.EventListenerService org.osgi.framework.SynchronousBundleListener + org.osgi.service.cm.ManagedService + + + + -