This is an automated email from the ASF dual-hosted git repository.
sergehuber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new 854d0d615 UNOMI-1006: Generate a node id per node and run the
heartbeat on every node (#916)
854d0d615 is described below
commit 854d0d61545623f039a406c6edacdf20b4aa8068
Author: Serge Huber <[email protected]>
AuthorDate: Thu Oct 15 06:49:46 2026 +0200
UNOMI-1006: Generate a node id per node and run the heartbeat on every node
(#916)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/unomi/api/ClusterNode.java | 39 ++++
docker/README.md | 2 +
.../test/java/org/apache/unomi/itests/AllITs.java | 1 +
.../org/apache/unomi/itests/NodeIdentityIT.java | 85 ++++++++
kar/src/main/feature/feature.xml | 1 +
lifecycle-watcher/pom.xml | 41 ++++
.../org/apache/unomi/lifecycle/BundleWatcher.java | 15 ++
.../apache/unomi/lifecycle/BundleWatcherImpl.java | 92 +++++++++
.../org/apache/unomi/lifecycle/NodeIdentity.java | 103 ++++++++++
.../resources/OSGI-INF/blueprint/blueprint.xml | 3 +
.../src/main/resources/org.apache.unomi.node.cfg | 26 +++
.../lifecycle/BundleWatcherNodeIdentityTest.java | 221 +++++++++++++++++++++
.../apache/unomi/lifecycle/NodeIdentityTest.java | 95 +++++++++
manual/src/main/asciidoc/clustering.adoc | 37 +++-
manual/src/main/asciidoc/scheduler.adoc | 11 +-
package/pom.xml | 9 +
.../main/resources/etc/custom.system.properties | 12 +-
.../services/impl/cluster/ClusterServiceImpl.java | 14 +-
.../scheduler/PersistenceSchedulerProvider.java | 4 +-
.../impl/scheduler/SchedulerServiceImpl.java | 12 +-
.../resources/OSGI-INF/blueprint/blueprint.xml | 4 +-
.../main/resources/org.apache.unomi.cluster.cfg | 8 +-
.../main/resources/org.apache.unomi.services.cfg | 6 +-
.../impl/cluster/ClusterServiceImplTest.java | 60 +++++-
.../PersistenceSchedulerProviderTest.java | 23 +++
.../impl/scheduler/SchedulerServiceNodeIdTest.java | 39 ++++
26 files changed, 926 insertions(+), 37 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 302d0b533..3a088b86a 100644
--- a/api/src/main/java/org/apache/unomi/api/ClusterNode.java
+++ b/api/src/main/java/org/apache/unomi/api/ClusterNode.java
@@ -39,6 +39,8 @@ public class ClusterNode extends Item {
private boolean master;
private boolean data;
private long startTime;
+ private String nodeName;
+ private Boolean executorNode;
private long lastHeartbeat;
// Server information
@@ -196,6 +198,43 @@ public class ClusterNode extends Item {
this.startTime = startTime;
}
+ /**
+ * User friendly name of this node. Unlike the item id it is only
informative and may change
+ * between restarts.
+ *
+ * @return node name, or {@code null} when unknown
+ */
+ public String getNodeName() {
+ return nodeName;
+ }
+
+ /**
+ * Sets the user friendly name of this node.
+ *
+ * @param nodeName node name
+ */
+ public void setNodeName(String nodeName) {
+ this.nodeName = nodeName;
+ }
+
+ /**
+ * Whether this node runs scheduled tasks ({@code scheduler.executorNode}).
+ *
+ * @return {@code true} or {@code false}, or {@code null} when the node
did not report it
+ */
+ public Boolean getExecutorNode() {
+ return executorNode;
+ }
+
+ /**
+ * Sets whether this node runs scheduled tasks.
+ *
+ * @param executorNode whether the node runs scheduled tasks
+ */
+ public void setExecutorNode(Boolean executorNode) {
+ this.executorNode = executorNode;
+ }
+
/**
* When this node last sent a heartbeat (milliseconds since epoch).
*
diff --git a/docker/README.md b/docker/README.md
index 426c72f40..06e9fbe30 100644
--- a/docker/README.md
+++ b/docker/README.md
@@ -126,6 +126,8 @@ Note: Linux doesn't support the host.docker.internal DNS
lookup method yet, it s
### Common Variables
- `UNOMI_ROOT_PASSWORD`: Required admin (`karaf`) password — no known default
- `UNOMI_HEALTHCHECK_PASSWORD`: Required health-check (`health`) password — no
known default
+- `UNOMI_CLUSTER_NODEID`: Optional node id, unique per container. When unset,
Unomi generates one on the first start and stores it inside the container; set
this variable to keep the same id when the container is recreated
+- `UNOMI_CLUSTER_NODENAME`: Optional user friendly node name (defaults to the
container host name); it is the prefix of a generated id
- `UNOMI_AUTO_START`: Boolean to specify if unomi auto start with karaf
(defaults to `true`)
- `UNOMI_DISTRIBUTION`: Specifies the Unomi Distribution Feature to use
(`unomi-distribution-elasticsearch` or `unomi-distribution-opensearch`,
defaults to `unomi-distribution-elasticsearch`)
diff --git a/itests/src/test/java/org/apache/unomi/itests/AllITs.java
b/itests/src/test/java/org/apache/unomi/itests/AllITs.java
index 5e07fde44..50f160411 100644
--- a/itests/src/test/java/org/apache/unomi/itests/AllITs.java
+++ b/itests/src/test/java/org/apache/unomi/itests/AllITs.java
@@ -96,6 +96,7 @@ import org.junit.runners.Suite.SuiteClasses;
TenantIT.class,
BackgroundTenantTasksIT.class,
SchedulerIT.class,
+ NodeIdentityIT.class,
EventsCollectorIT.class,
RolloverIT.class,
HealthCheckIT.class
diff --git a/itests/src/test/java/org/apache/unomi/itests/NodeIdentityIT.java
b/itests/src/test/java/org/apache/unomi/itests/NodeIdentityIT.java
new file mode 100644
index 000000000..da8efe972
--- /dev/null
+++ b/itests/src/test/java/org/apache/unomi/itests/NodeIdentityIT.java
@@ -0,0 +1,85 @@
+/*
+ * 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.itests;
+
+import org.apache.unomi.api.ClusterNode;
+import org.apache.unomi.api.services.ClusterService;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.junit.PaxExam;
+import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
+import org.ops4j.pax.exam.spi.reactors.PerSuite;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Objects;
+import java.util.Properties;
+
+import static org.junit.Assert.*;
+
+/**
+ * The ITs configure no node id, so the container goes through what a fresh
install does: the lifecycle
+ * bundle generates an id, Karaf hands it to the cluster and scheduler
services through their
+ * configurations, and it is written back to etc/ for the next start.
+ */
+@RunWith(PaxExam.class)
+@ExamReactorStrategy(PerSuite.class)
+public class NodeIdentityIT extends BaseIT {
+
+ private static final String NODE_CONFIG_FILE = "org.apache.unomi.node.cfg";
+
+ @Test
+ public void testGeneratedNodeIdIsUsedByClusterAndSchedulerAndStored()
throws Exception {
+ String nodeId = bundleWatcher.getNodeId();
+ String nodeName = bundleWatcher.getNodeName();
+ assertNotNull("The lifecycle bundle should have resolved a node id",
nodeId);
+ assertTrue("Node id " + nodeId + " should be generated as <name>-<8
hex characters>", nodeId.matches(".+-[0-9a-f]{8}"));
+ assertFalse("The node should have a name", nodeName == null ||
nodeName.trim().isEmpty());
+
+ assertEquals("The scheduler should use the node id", nodeId,
schedulerService.getNodeId());
+
+ ClusterService clusterService = getOsgiService(ClusterService.class,
600000);
+ ClusterNode clusterNode = keepTrying("Node " + nodeId + " should be
registered in the cluster",
+ () -> clusterService.getClusterNodes().stream().filter(node ->
nodeId.equals(node.getItemId())).findFirst().orElse(null),
+ Objects::nonNull, DEFAULT_TRYING_TIMEOUT,
DEFAULT_TRYING_TRIES);
+ assertEquals("The cluster record should carry the node name",
nodeName, clusterNode.getNodeName());
+ assertEquals("The cluster record should say the node runs tasks",
Boolean.TRUE, clusterNode.getExecutorNode());
+
+ // Karaf writes the configuration back to its file asynchronously
+ Path nodeConfigFile = Paths.get(System.getProperty("karaf.etc"),
NODE_CONFIG_FILE);
+ Properties stored = keepTrying("Node id " + nodeId + " should be
stored in " + nodeConfigFile,
+ () -> load(nodeConfigFile), properties ->
nodeId.equals(properties.getProperty("nodeId")),
+ DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
+ assertEquals(nodeName, stored.getProperty("nodeName"));
+ assertTrue("The explanations in " + NODE_CONFIG_FILE + " should
survive the update",
+ new String(Files.readAllBytes(nodeConfigFile),
StandardCharsets.UTF_8).contains("Identity of this Apache Unomi node"));
+ }
+
+ private static Properties load(Path file) {
+ Properties properties = new Properties();
+ try (Reader reader = Files.newBufferedReader(file,
StandardCharsets.UTF_8)) {
+ properties.load(reader);
+ } catch (IOException e) {
+ // Not there yet, or being written: the caller retries
+ }
+ return properties;
+ }
+}
diff --git a/kar/src/main/feature/feature.xml b/kar/src/main/feature/feature.xml
index dfecb485e..832ce089d 100644
--- a/kar/src/main/feature/feature.xml
+++ b/kar/src/main/feature/feature.xml
@@ -65,6 +65,7 @@
<bundle>mvn:org.apache.groovy/groovy-xml/${groovy.version}</bundle>
<bundle>mvn:org.apache.groovy/groovy-json/${groovy.version}</bundle>
+ <configfile
finalname="/etc/org.apache.unomi.node.cfg">mvn:org.apache.unomi/unomi-lifecycle-watcher/${project.version}/cfg/nodecfg</configfile>
<bundle>mvn:org.apache.unomi/unomi-lifecycle-watcher/${project.version}</bundle>
<bundle>mvn:org.apache.unomi/unomi-api/${project.version}</bundle>
<bundle>mvn:org.apache.unomi/unomi-common/${project.version}</bundle>
diff --git a/lifecycle-watcher/pom.xml b/lifecycle-watcher/pom.xml
index b1556c5c7..25f4f7e78 100644
--- a/lifecycle-watcher/pom.xml
+++ b/lifecycle-watcher/pom.xml
@@ -46,6 +46,11 @@
<artifactId>osgi.core</artifactId>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.osgi</groupId>
+ <artifactId>org.osgi.service.cm</artifactId>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.unomi</groupId>
<artifactId>unomi-api</artifactId>
@@ -66,6 +71,18 @@
<artifactId>slf4j-api</artifactId>
<scope>provided</scope>
</dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
@@ -107,6 +124,30 @@
</instructions>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>attach-artifacts</id>
+ <phase>package</phase>
+ <goals>
+ <goal>attach-artifact</goal>
+ </goals>
+ <configuration>
+ <artifacts>
+ <artifact>
+ <file>
+
src/main/resources/org.apache.unomi.node.cfg
+ </file>
+ <type>cfg</type>
+ <classifier>nodecfg</classifier>
+ </artifact>
+ </artifacts>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
</project>
diff --git
a/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/BundleWatcher.java
b/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/BundleWatcher.java
index 6e8ef96da..e76c92a55 100644
---
a/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/BundleWatcher.java
+++
b/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/BundleWatcher.java
@@ -35,6 +35,21 @@ public interface BundleWatcher {
*/
List<ServerInfo> getServerInfos();
+ /**
+ * Retrieves the id of this node, which stays the same across restarts. It
is the id the cluster
+ * and the scheduler use, unless they are configured with their own.
+ *
+ * @return the node id
+ */
+ String getNodeId();
+
+ /**
+ * Retrieves the user friendly name of this node.
+ *
+ * @return the node name
+ */
+ String getNodeName();
+
/**
* Indicates whether Unomi startup has completed. Implementations
typically track
* required bundles and initialization tasks to decide when the system is
fully ready.
diff --git
a/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/BundleWatcherImpl.java
b/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/BundleWatcherImpl.java
index 63850456c..72423cc1c 100644
---
a/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/BundleWatcherImpl.java
+++
b/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/BundleWatcherImpl.java
@@ -19,6 +19,8 @@ package org.apache.unomi.lifecycle;
import org.apache.commons.lang3.StringUtils;
import org.apache.unomi.api.ServerInfo;
import org.osgi.framework.*;
+import org.osgi.service.cm.Configuration;
+import org.osgi.service.cm.ConfigurationAdmin;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,6 +31,7 @@ import java.net.URL;
import java.text.MessageFormat;
import java.util.*;
import java.util.concurrent.*;
+import java.util.function.Function;
import java.util.stream.Collectors;
/**
@@ -39,6 +42,17 @@ public class BundleWatcherImpl implements
SynchronousBundleListener, ServiceList
private static final Logger LOGGER =
LoggerFactory.getLogger(BundleWatcherImpl.class.getName());
+ /** What the operator set (UNOMI_CLUSTER_NODEID / UNOMI_CLUSTER_NODENAME),
blank when the id is left to Unomi. */
+ static final String CLUSTER_NODE_ID_PROPERTY =
"org.apache.unomi.cluster.nodeId";
+ static final String CLUSTER_NODE_NAME_PROPERTY =
"org.apache.unomi.cluster.nodeName";
+ static final String CLUSTER_NODE_ID_VARIABLE = "UNOMI_CLUSTER_NODEID";
+ /** Set by an operator who wants the scheduler to have an id of its own. */
+ static final String SCHEDULER_NODE_ID_OVERRIDE_PROPERTY =
"org.apache.unomi.scheduler.nodeId";
+ /** {@code <pid>.<key>}: what Karaf substitutes for {@code
scheduler.nodeId} in org.apache.unomi.services. */
+ static final String SCHEDULER_NODE_ID_PROPERTY =
"org.apache.unomi.services.scheduler.nodeId";
+ /** Default of the node id in the etc/custom.system.properties of earlier
versions, the same on every node. */
+ static final String FORMER_DEFAULT_NODE_ID = "unomi-node-1";
+
private long startupTime;
private Map<String, Boolean> requiredBundles = new ConcurrentHashMap<>();
private Map<String, Boolean> requiredBundlesFromFeatures = new
ConcurrentHashMap<>();
@@ -51,6 +65,10 @@ public class BundleWatcherImpl implements
SynchronousBundleListener, ServiceList
private long matchedRequiredServicesCount = 0;
private BundleContext bundleContext;
+ private ConfigurationAdmin configurationAdmin;
+ private NodeIdentity nodeIdentity;
+ /** Environment lookup, replaceable in tests. */
+ Function<String, String> environment = System::getenv;
private volatile boolean startupMessageAlreadyDisplayed = false;
private boolean shutdownMessageAlreadyDisplayed = false;
@@ -86,7 +104,12 @@ public class BundleWatcherImpl implements
SynchronousBundleListener, ServiceList
this.bundleContext = bundleContext;
}
+ public void setConfigurationAdmin(ConfigurationAdmin configurationAdmin) {
+ this.configurationAdmin = configurationAdmin;
+ }
+
public void init() {
+ resolveNodeIdentity();
scheduler = Executors.newSingleThreadScheduledExecutor();
checkExistingBundles();
bundleContext.addBundleListener(this);
@@ -96,6 +119,75 @@ public class BundleWatcherImpl implements
SynchronousBundleListener, ServiceList
LOGGER.info("Bundle watcher initialized.");
}
+ /**
+ * Decides the identity of this node and makes it the id of the cluster
and scheduler services.
+ * <p>
+ * This runs before the services bundle is created (it needs the {@link
BundleWatcher} service). The
+ * two system properties set here are named after the configuration and
key they stand for, which is
+ * what Karaf substitutes for {@code nodeId} in org.apache.unomi.cluster
and {@code scheduler.nodeId}
+ * in org.apache.unomi.services when a bundle reads them. The identity is
remembered in the
+ * {@link NodeIdentity#PID} configuration, which Karaf writes back to
etc/, so the node keeps its id
+ * across restarts.
+ */
+ void resolveNodeIdentity() {
+ String configuredId = System.getProperty(CLUSTER_NODE_ID_PROPERTY);
+ if (FORMER_DEFAULT_NODE_ID.equals(configuredId) &&
!FORMER_DEFAULT_NODE_ID.equals(environment.apply(CLUSTER_NODE_ID_VARIABLE))) {
+ LOGGER.warn("Ignoring node id {}: it is the default of an
etc/custom.system.properties kept from an earlier "
+ + "version, which every node would share. Set {}={} to
keep it.",
+ FORMER_DEFAULT_NODE_ID, CLUSTER_NODE_ID_VARIABLE,
FORMER_DEFAULT_NODE_ID);
+ configuredId = null;
+ }
+
+ // Only set once the stored identity could be read: one that could not
be read must not be overwritten
+ Configuration configuration = null;
+ Dictionary<String, Object> stored = new Hashtable<>();
+ try {
+ Configuration readable =
configurationAdmin.getConfiguration(NodeIdentity.PID, "?");
+ Dictionary<String, Object> properties = readable.getProperties();
+ if (properties != null) {
+ stored = properties;
+ }
+ configuration = readable;
+ } catch (Exception e) {
+ LOGGER.warn("Could not read the stored node identity ({}): the
node id will only be valid until this node stops",
+ NodeIdentity.PID, e);
+ }
+ String storedId = Objects.toString(stored.get(NodeIdentity.NODE_ID),
null);
+ String storedName =
Objects.toString(stored.get(NodeIdentity.NODE_NAME), null);
+
+ NodeIdentity identity = NodeIdentity.resolve(configuredId, storedId,
System.getProperty(CLUSTER_NODE_NAME_PROPERTY),
+ NodeIdentity.localHostName(), NodeIdentity::randomSuffix);
+
+ if (configuration != null && !(identity.nodeId.equals(storedId) &&
identity.nodeName.equals(storedName))) {
+ try {
+ // The other properties stay: felix.fileinstall.filename is
what ties the configuration to its file
+ stored.put(NodeIdentity.NODE_ID, identity.nodeId);
+ stored.put(NodeIdentity.NODE_NAME, identity.nodeName);
+ configuration.update(stored);
+ } catch (Exception e) {
+ LOGGER.warn("Could not store the node identity: node id {} is
only valid until this node stops. "
+ + "Set {} to keep it.", identity.nodeId,
CLUSTER_NODE_ID_VARIABLE, e);
+ }
+ }
+
+ System.setProperty(CLUSTER_NODE_ID_PROPERTY, identity.nodeId);
+ if
(StringUtils.isBlank(System.getProperty(SCHEDULER_NODE_ID_OVERRIDE_PROPERTY))) {
+ System.setProperty(SCHEDULER_NODE_ID_PROPERTY, identity.nodeId);
+ }
+ nodeIdentity = identity;
+ LOGGER.info("Node identity: id={} ({}), name={}", identity.nodeId,
identity.source, identity.nodeName);
+ }
+
+ @Override
+ public String getNodeId() {
+ return nodeIdentity.nodeId;
+ }
+
+ @Override
+ public String getNodeName() {
+ return nodeIdentity.nodeName;
+ }
+
@Override
public List<ServerInfo> getServerInfos() {
return serverInfos;
diff --git
a/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/NodeIdentity.java
b/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/NodeIdentity.java
new file mode 100644
index 000000000..0ae41cd5f
--- /dev/null
+++
b/lifecycle-watcher/src/main/java/org/apache/unomi/lifecycle/NodeIdentity.java
@@ -0,0 +1,103 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.net.InetAddress;
+import java.security.SecureRandom;
+import java.util.Locale;
+import java.util.function.Supplier;
+
+/**
+ * Identity of this Unomi node: an id that stays the same across restarts and
a user friendly name.
+ * <p>
+ * The id is the configured one when there is one, otherwise the id this node
used before, otherwise
+ * a new one built from the node name and a random suffix (for example {@code
web-1-3f9a2c1b}).
+ * The name is the configured one, otherwise the host name. Unlike the id, the
name is resolved
+ * again at every start, so it follows the configuration or the host name.
+ */
+final class NodeIdentity {
+
+ /** Configuration (stored as {@code etc/org.apache.unomi.node.cfg}) that
remembers the identity. */
+ static final String PID = "org.apache.unomi.node";
+ static final String NODE_ID = "nodeId";
+ static final String NODE_NAME = "nodeName";
+
+ static final String DEFAULT_NODE_NAME = "unomi";
+ private static final int MAX_NAME_LENGTH_IN_ID = 40;
+ private static final SecureRandom RANDOM = new SecureRandom();
+
+ final String nodeId;
+ final String nodeName;
+ /** Where the id comes from: {@code configured}, {@code stored} or {@code
generated}. */
+ final String source;
+
+ private NodeIdentity(String nodeId, String nodeName, String source) {
+ this.nodeId = nodeId;
+ this.nodeName = nodeName;
+ this.source = source;
+ }
+
+ /**
+ * @param configuredId id set by the operator, blank when none
+ * @param storedId id this node used at its previous start, blank
when none
+ * @param configuredName name set by the operator, blank when none
+ * @param hostName host name of this machine, blank when unknown
+ * @param suffix source of the random suffix of a generated id
+ * @return the identity to use for this start
+ */
+ static NodeIdentity resolve(String configuredId, String storedId, String
configuredName, String hostName,
+ Supplier<String> suffix) {
+ String nodeName = StringUtils.firstNonBlank(configuredName, hostName,
DEFAULT_NODE_NAME).trim();
+ if (StringUtils.isNotBlank(configuredId)) {
+ return new NodeIdentity(configuredId.trim(), nodeName,
"configured");
+ }
+ if (StringUtils.isNotBlank(storedId)) {
+ return new NodeIdentity(storedId.trim(), nodeName, "stored");
+ }
+ return new NodeIdentity(toIdPrefix(nodeName) + "-" + suffix.get(),
nodeName, "generated");
+ }
+
+ /**
+ * Reduces a name to what is safe everywhere a node id ends up (document
ids, logs, URLs).
+ */
+ static String toIdPrefix(String nodeName) {
+ String prefix =
nodeName.toLowerCase(Locale.ROOT).replaceAll("[^a-z0-9]+", "-");
+ prefix = StringUtils.strip(StringUtils.left(prefix,
MAX_NAME_LENGTH_IN_ID), "-");
+ return prefix.isEmpty() ? DEFAULT_NODE_NAME : prefix;
+ }
+
+ static String randomSuffix() {
+ return String.format("%08x", RANDOM.nextInt());
+ }
+
+ /**
+ * The environment is asked first, since it is what a container sets.
+ */
+ static String localHostName() {
+ String hostName = StringUtils.firstNonBlank(System.getenv("HOSTNAME"),
System.getenv("COMPUTERNAME"));
+ if (hostName != null) {
+ return hostName;
+ }
+ try {
+ return InetAddress.getLocalHost().getHostName();
+ } catch (Exception e) {
+ return null;
+ }
+ }
+}
diff --git
a/lifecycle-watcher/src/main/resources/OSGI-INF/blueprint/blueprint.xml
b/lifecycle-watcher/src/main/resources/OSGI-INF/blueprint/blueprint.xml
index ede5554de..343758c1b 100644
--- a/lifecycle-watcher/src/main/resources/OSGI-INF/blueprint/blueprint.xml
+++ b/lifecycle-watcher/src/main/resources/OSGI-INF/blueprint/blueprint.xml
@@ -30,8 +30,11 @@
</cm:default-properties>
</cm:property-placeholder>
+ <reference id="configurationAdmin"
interface="org.osgi.service.cm.ConfigurationAdmin"/>
+
<bean id="bundleWatcherImpl" init-method="init" destroy-method="destroy"
class="org.apache.unomi.lifecycle.BundleWatcherImpl">
<property name="bundleContext" ref="blueprintBundleContext"/>
+ <property name="configurationAdmin" ref="configurationAdmin"/>
<property name="requiredServices"
value="${lifecycle.requiredServices}" />
<property name="checkStartupStateRefreshInterval"
value="${lifecycle.checkStartupState.refresh.interval}"/>
<property name="requiredBundles">
diff --git a/lifecycle-watcher/src/main/resources/org.apache.unomi.node.cfg
b/lifecycle-watcher/src/main/resources/org.apache.unomi.node.cfg
new file mode 100644
index 000000000..0c5307fdb
--- /dev/null
+++ b/lifecycle-watcher/src/main/resources/org.apache.unomi.node.cfg
@@ -0,0 +1,26 @@
+#
+# 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.
+#
+
+# Identity of this Apache Unomi node, written by Unomi itself at startup.
+#
+# nodeId is generated on the first start and reused afterwards, so the node
keeps its place in the
+# cluster across restarts. Setting UNOMI_CLUSTER_NODEID
(org.apache.unomi.cluster.nodeId) replaces it.
+# To get a new id, unset that variable, blank the value below and restart; do
not delete this file.
+nodeId=
+
+# User friendly name, from UNOMI_CLUSTER_NODENAME
(org.apache.unomi.cluster.nodeName) or the host name.
+nodeName=
diff --git
a/lifecycle-watcher/src/test/java/org/apache/unomi/lifecycle/BundleWatcherNodeIdentityTest.java
b/lifecycle-watcher/src/test/java/org/apache/unomi/lifecycle/BundleWatcherNodeIdentityTest.java
new file mode 100644
index 000000000..7bd626c5f
--- /dev/null
+++
b/lifecycle-watcher/src/test/java/org/apache/unomi/lifecycle/BundleWatcherNodeIdentityTest.java
@@ -0,0 +1,221 @@
+/*
+ * 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.lifecycle;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.osgi.service.cm.Configuration;
+import org.osgi.service.cm.ConfigurationAdmin;
+
+import java.io.IOException;
+import java.util.Dictionary;
+import java.util.HashMap;
+import java.util.Hashtable;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * How the lifecycle bundle decides, remembers and publishes the identity of
the node at startup.
+ */
+class BundleWatcherNodeIdentityTest {
+
+ private static final String FILE_NAME_KEY = "felix.fileinstall.filename";
+ private static final String FILE_NAME =
"file:/opt/unomi/etc/org.apache.unomi.node.cfg";
+ private static final String[] PROPERTIES = {
+ BundleWatcherImpl.CLUSTER_NODE_ID_PROPERTY,
BundleWatcherImpl.CLUSTER_NODE_NAME_PROPERTY,
+ BundleWatcherImpl.SCHEDULER_NODE_ID_OVERRIDE_PROPERTY,
BundleWatcherImpl.SCHEDULER_NODE_ID_PROPERTY};
+
+ private ConfigurationAdmin configurationAdmin;
+ private Configuration configuration;
+ private final Map<String, String> environment = new HashMap<>();
+
+ @BeforeEach
+ void setUp() throws IOException {
+ clearProperties();
+ // Fixed name, so the tests do not depend on the host they run on
+ System.setProperty(BundleWatcherImpl.CLUSTER_NODE_NAME_PROPERTY,
"web-1");
+ configuration = mock(Configuration.class);
+ configurationAdmin = mock(ConfigurationAdmin.class);
+ when(configurationAdmin.getConfiguration(NodeIdentity.PID,
"?")).thenReturn(configuration);
+ }
+
+ @AfterEach
+ void clearProperties() {
+ for (String property : PROPERTIES) {
+ System.clearProperty(property);
+ }
+ }
+
+ /** What Karaf holds for the configuration once it has loaded
etc/org.apache.unomi.node.cfg. */
+ private void stored(String nodeId, String nodeName) {
+ Dictionary<String, Object> properties = new Hashtable<>();
+ properties.put(NodeIdentity.NODE_ID, nodeId);
+ properties.put(NodeIdentity.NODE_NAME, nodeName);
+ properties.put(FILE_NAME_KEY, FILE_NAME);
+ when(configuration.getProperties()).thenReturn(properties);
+ }
+
+ /** One JVM start: Karaf sets the properties from the environment, then
the lifecycle bundle starts. */
+ private BundleWatcherImpl start() {
+ BundleWatcherImpl bundleWatcher = new BundleWatcherImpl();
+ bundleWatcher.setConfigurationAdmin(configurationAdmin);
+ bundleWatcher.environment = environment::get;
+ bundleWatcher.resolveNodeIdentity();
+ return bundleWatcher;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Dictionary<String, Object> storedByUpdate() throws IOException {
+ ArgumentCaptor<Dictionary<String, Object>> captor =
ArgumentCaptor.forClass(Dictionary.class);
+ verify(configuration).update(captor.capture());
+ return captor.getValue();
+ }
+
+ private static void assertGenerated(String nodeId) {
+ assertTrue(nodeId.matches("web-1-[0-9a-f]{8}"), nodeId);
+ }
+
+ @Test
+ void firstStartGeneratesStoresAndPublishesTheId() throws IOException {
+ stored("", "");
+
+ BundleWatcherImpl bundleWatcher = start();
+
+ String nodeId = bundleWatcher.getNodeId();
+ assertGenerated(nodeId);
+ assertEquals("web-1", bundleWatcher.getNodeName());
+
+ Dictionary<String, Object> update = storedByUpdate();
+ assertEquals(nodeId, update.get(NodeIdentity.NODE_ID));
+ assertEquals("web-1", update.get(NodeIdentity.NODE_NAME));
+ // Without it Karaf would no longer write the configuration back to
its file in etc/
+ assertEquals(FILE_NAME, update.get(FILE_NAME_KEY));
+
+ // What the cluster and scheduler configurations are resolved with
+ assertEquals(nodeId,
System.getProperty(BundleWatcherImpl.CLUSTER_NODE_ID_PROPERTY));
+ assertEquals(nodeId,
System.getProperty(BundleWatcherImpl.SCHEDULER_NODE_ID_PROPERTY));
+ }
+
+ @Test
+ void restartReusesTheStoredIdWithoutRewritingIt() throws IOException {
+ stored("web-1-99999999", "web-1");
+
+ BundleWatcherImpl bundleWatcher = start();
+
+ assertEquals("web-1-99999999", bundleWatcher.getNodeId());
+ assertEquals("web-1-99999999",
System.getProperty(BundleWatcherImpl.CLUSTER_NODE_ID_PROPERTY));
+ assertEquals("web-1-99999999",
System.getProperty(BundleWatcherImpl.SCHEDULER_NODE_ID_PROPERTY));
+ verify(configuration, never()).update(any());
+ }
+
+ @Test
+ void renamingTheNodeKeepsItsId() throws IOException {
+ stored("web-1-99999999", "web-1");
+ System.setProperty(BundleWatcherImpl.CLUSTER_NODE_NAME_PROPERTY,
"frontend");
+
+ BundleWatcherImpl bundleWatcher = start();
+
+ assertEquals("web-1-99999999", bundleWatcher.getNodeId());
+ assertEquals("frontend", bundleWatcher.getNodeName());
+ assertEquals("frontend", storedByUpdate().get(NodeIdentity.NODE_NAME));
+ }
+
+ @Test
+ void configuredIdWinsAndIsRememberedOnceTheVariableIsRemoved() throws
IOException {
+ stored("web-1-99999999", "web-1");
+ System.setProperty(BundleWatcherImpl.CLUSTER_NODE_ID_PROPERTY,
"unomi-node-7");
+
+ assertEquals("unomi-node-7", start().getNodeId());
+ Dictionary<String, Object> update = storedByUpdate();
+ assertEquals("unomi-node-7", update.get(NodeIdentity.NODE_ID));
+
+ // Next start without UNOMI_CLUSTER_NODEID: the id given back to the
node stays
+ System.clearProperty(BundleWatcherImpl.CLUSTER_NODE_ID_PROPERTY);
+ when(configuration.getProperties()).thenReturn(update);
+ assertEquals("unomi-node-7", start().getNodeId());
+ }
+
+ /**
+ * An etc/custom.system.properties kept from an earlier version sets the
id of every node to the same
+ * default. Only an operator who really chose that value, through the
environment, gets it.
+ */
+ @Test
+ void formerSharedDefaultIsIgnoredUnlessChosenInTheEnvironment() throws
IOException {
+ stored("", "");
+ System.setProperty(BundleWatcherImpl.CLUSTER_NODE_ID_PROPERTY,
BundleWatcherImpl.FORMER_DEFAULT_NODE_ID);
+
+ assertGenerated(start().getNodeId());
+
+ System.setProperty(BundleWatcherImpl.CLUSTER_NODE_ID_PROPERTY,
BundleWatcherImpl.FORMER_DEFAULT_NODE_ID);
+ environment.put(BundleWatcherImpl.CLUSTER_NODE_ID_VARIABLE,
BundleWatcherImpl.FORMER_DEFAULT_NODE_ID);
+ assertEquals(BundleWatcherImpl.FORMER_DEFAULT_NODE_ID,
start().getNodeId());
+ }
+
+ @Test
+ void schedulerOverrideIsLeftAlone() {
+ stored("web-1-99999999", "web-1");
+
System.setProperty(BundleWatcherImpl.SCHEDULER_NODE_ID_OVERRIDE_PROPERTY,
"scheduler-a");
+
+ start();
+
+ assertEquals("web-1-99999999",
System.getProperty(BundleWatcherImpl.CLUSTER_NODE_ID_PROPERTY));
+
assertNull(System.getProperty(BundleWatcherImpl.SCHEDULER_NODE_ID_PROPERTY));
+ }
+
+ @Test
+ void worksWhenTheConfigurationDoesNotExistYet() throws IOException {
+ when(configuration.getProperties()).thenReturn(null);
+
+ String nodeId = start().getNodeId();
+
+ assertGenerated(nodeId);
+ assertEquals(nodeId, storedByUpdate().get(NodeIdentity.NODE_ID));
+ }
+
+ @Test
+ void nodeStillGetsAnIdWhenItCannotBeStored() throws IOException {
+ stored("", "");
+ doThrow(new IOException("etc is
read-only")).when(configuration).update(any());
+
+ String nodeId = start().getNodeId();
+
+ assertGenerated(nodeId);
+ assertEquals(nodeId,
System.getProperty(BundleWatcherImpl.CLUSTER_NODE_ID_PROPERTY));
+ }
+
+ @Test
+ void storedIdentityThatCannotBeReadIsNotOverwritten() throws IOException {
+ when(configuration.getProperties()).thenThrow(new
IllegalStateException("unavailable"));
+
+ String nodeId = start().getNodeId();
+
+ assertGenerated(nodeId);
+ assertEquals(nodeId,
System.getProperty(BundleWatcherImpl.SCHEDULER_NODE_ID_PROPERTY));
+ verify(configuration, never()).update(any());
+ }
+}
diff --git
a/lifecycle-watcher/src/test/java/org/apache/unomi/lifecycle/NodeIdentityTest.java
b/lifecycle-watcher/src/test/java/org/apache/unomi/lifecycle/NodeIdentityTest.java
new file mode 100644
index 000000000..0a5bc6559
--- /dev/null
+++
b/lifecycle-watcher/src/test/java/org/apache/unomi/lifecycle/NodeIdentityTest.java
@@ -0,0 +1,95 @@
+/*
+ * 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.lifecycle;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class NodeIdentityTest {
+
+ private static NodeIdentity resolve(String configuredId, String storedId,
String configuredName, String hostName) {
+ return NodeIdentity.resolve(configuredId, storedId, configuredName,
hostName, () -> "0a1b2c3d");
+ }
+
+ @Test
+ void generatesIdFromNameWhenNothingIsConfiguredOrStored() {
+ NodeIdentity identity = resolve(null, " ", null, "web-1");
+ assertEquals("web-1-0a1b2c3d", identity.nodeId);
+ assertEquals("web-1", identity.nodeName);
+ assertEquals("generated", identity.source);
+ }
+
+ @Test
+ void storedIdIsReusedAndDoesNotFollowNameChanges() {
+ NodeIdentity identity = resolve("", "web-1-99999999", "frontend",
"web-2");
+ assertEquals("web-1-99999999", identity.nodeId);
+ assertEquals("frontend", identity.nodeName);
+ assertEquals("stored", identity.source);
+ }
+
+ @Test
+ void configuredIdWinsOverStoredId() {
+ NodeIdentity identity = resolve(" unomi-node-7 ", "web-1-99999999",
null, "web-1");
+ assertEquals("unomi-node-7", identity.nodeId);
+ assertEquals("configured", identity.source);
+ }
+
+ @Test
+ void configuredNameWinsOverHostName() {
+ NodeIdentity identity = resolve(null, null, "Frontend A", "web-1");
+ assertEquals("Frontend A", identity.nodeName);
+ assertEquals("frontend-a-0a1b2c3d", identity.nodeId);
+ }
+
+ @Test
+ void fallsBackToDefaultNameWithoutHostName() {
+ NodeIdentity identity = resolve(null, null, " ", null);
+ assertEquals("unomi", identity.nodeName);
+ assertEquals("unomi-0a1b2c3d", identity.nodeId);
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "web-1, web-1",
+ "Web_1.Example.COM, web-1-example-com",
+ "' node ', node",
+ "--a--, a",
+ "éà, unomi",
+ "0123456789012345678901234567890123456789-tail,
0123456789012345678901234567890123456789"
+ })
+ void idPrefixIsSafeForIdsLogsAndUrls(String nodeName, String
expectedPrefix) {
+ assertEquals(expectedPrefix, NodeIdentity.toIdPrefix(nodeName));
+ }
+
+ @Test
+ void randomSuffixIsEightHexCharactersAndVaries() {
+ Set<String> suffixes = new HashSet<>();
+ for (int i = 0; i < 200; i++) {
+ String suffix = NodeIdentity.randomSuffix();
+ assertTrue(suffix.matches("[0-9a-f]{8}"), suffix);
+ suffixes.add(suffix);
+ }
+ assertTrue(suffixes.size() > 190, "suffixes should differ from one
another");
+ }
+}
diff --git a/manual/src/main/asciidoc/clustering.adoc
b/manual/src/main/asciidoc/clustering.adoc
index 6ffd776e4..356bc75ab 100644
--- a/manual/src/main/asciidoc/clustering.adoc
+++ b/manual/src/main/asciidoc/clustering.adoc
@@ -30,23 +30,52 @@ Configure each node in `etc/unomi.custom.system.properties`
(or via environment
----
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.nodeId=${env:UNOMI_CLUSTER_NODEID:-}
+org.apache.unomi.cluster.nodeName=${env:UNOMI_CLUSTER_NODENAME:-}
org.apache.unomi.cluster.nodeStatisticsUpdateFrequency=${env:UNOMI_CLUSTER_NODESTATISTICS_UPDATEFREQUENCY:-10000}
----
* **public.address** — URL other nodes and operators use to reach this node
(HTTP, port 8181 by default).
* **internal.address** — HTTPS URL for the private REST API (port 9443 by
default).
-* **nodeId** — **Must be unique** on every node in the cluster (for example
`unomi-node-1`, `unomi-node-2`).
+* **nodeId** — Optional. Identifies the node in the cluster and in the
scheduler; when it is not set, Unomi
+ generates one (see <<_node_identity,Node identity>>). When you set it, it
**must be unique** in the cluster.
+* **nodeName** — Optional user friendly name, shown next to the id. Defaults
to the host name.
* **nodeStatisticsUpdateFrequency** — Interval in milliseconds between
heartbeat / statistics updates (default 10 seconds).
-If `nodeId` is missing or duplicate, cluster registration fails at startup.
+[#_node_identity]
+==== Node identity
+
+Every node needs an id that no other node uses and that it keeps across
restarts: the id owns the node's cluster
+record and its scheduler task locks. Nothing has to be configured for this.
+
+* On its first start a node generates an id made of its name and a random
suffix, for example `web-1-3f9a2c1b`, and
+ stores it with its name in `etc/org.apache.unomi.node.cfg`. Later starts
reuse it. A second node, container or
+ copy of the image generates its own, so a single node grows into a cluster
without any change.
+* `UNOMI_CLUSTER_NODEID` (`org.apache.unomi.cluster.nodeId`) sets the id
explicitly. It replaces the stored id and
+ is stored in turn, so the node keeps it after the variable is removed. Unomi
does not check that an id you set is
+ unique: two nodes with the same id share one cluster record and the same
task locks.
+* The name comes from `UNOMI_CLUSTER_NODENAME`
(`org.apache.unomi.cluster.nodeName`), or from the host name. It is
+ read again at every start: renaming a node does not change its id.
+* The scheduler uses the same id.
+
+The stored id lives in the node's `etc/` directory and is lost with it, for
example when a container is recreated.
+The node then joins under a new id; the old cluster record is removed once its
heartbeat is stale and its task
+locks expire. To give a node its previous id back, start it once with
`UNOMI_CLUSTER_NODEID` set to that id (it is
+in the node's log at startup and in the cluster REST endpoint). To make a node
take a new id, unset
+`UNOMI_CLUSTER_NODEID`, blank `nodeId` in `etc/org.apache.unomi.node.cfg` and
restart the node; do not delete the
+file, Unomi could then no longer save the id there.
+
+Copying an installation together with its `etc/` directory (a cloned virtual
machine, an image built from a node
+that already ran, an `etc/` volume shared by several containers) copies the id
too: blank it on the copies.
==== How it works
On startup the cluster service:
1. Registers this node in persistence (`ClusterNode` item).
-2. Schedules a recurring task to update CPU, load, and uptime statistics.
+2. Schedules a recurring task to update CPU, load, and uptime statistics. It
is the node's heartbeat and runs on
+ every node, including those with `scheduler.executorNode=false`; the record
also says whether the node runs
+ scheduled tasks.
3. Schedules stale-node cleanup (nodes with no heartbeat for roughly three
times the update interval are removed).
Other nodes see the same registry through the shared search backend. The
health-check extension reports
diff --git a/manual/src/main/asciidoc/scheduler.adoc
b/manual/src/main/asciidoc/scheduler.adoc
index 8d902cd02..d42806766 100644
--- a/manual/src/main/asciidoc/scheduler.adoc
+++ b/manual/src/main/asciidoc/scheduler.adoc
@@ -447,7 +447,9 @@ end note
==== Node affinity timeline
-For a due exclusive task with active nodes `[N0, N1, N2]` (sorted) and primary
= `hash(taskId) % 3`:
+For a due exclusive task with active nodes `[N0, N1, N2]` (sorted) and primary
= `hash(taskId) % 3`. Active nodes are
+the nodes that sent a heartbeat in the last 5 minutes, leaving out those with
`scheduler.executorNode=false`, which
+never run exclusive tasks:
[cols="1,2,3",options="header"]
|===
@@ -510,7 +512,8 @@ Every checker tick (~1s) on a node runs recovery:
==== Operational checklist for clustering
-* Set a **stable unique** `scheduler.nodeId` per JVM.
+* Each JVM needs a **stable unique** node id. Unomi generates and remembers
one per node
+ (see <<_node_identity,Node identity>>) and the scheduler uses it.
* Put at least one `scheduler.executorNode=true` in the cluster for persistent
work.
* Use `disallowParallelExecution()` for singleton side-effecting jobs.
* Keep `scheduler.lockTimeout` at ~5s or above (default 10s): lease renewal
keeps long executions safe automatically, but the heartbeat needs headroom over
the `wait_for` refresh cycle (see
<<_renewal_timing_limits_with_the_default_1_s_refresh_interval,Renewal timing
limits>>).
@@ -610,7 +613,7 @@ Scheduler settings live in
`etc/org.apache.unomi.services.cfg` (OSGi PID `org.ap
|Property |Default |Description
|`scheduler.thread.poolSize` |`5` |Thread pool size for task execution
-|`scheduler.nodeId` |`test-scheduler-node` |Unique ID for this node in
scheduler logs and task metadata. Set a stable value per node in production
clusters.
+|`scheduler.nodeId` |the node id |Unique ID for this node in scheduler locks
and task metadata. Unomi provides the id of the node (see <<_node_identity,Node
identity>>); a value set here is ignored. To give the scheduler an id of its
own, set the `org.apache.unomi.scheduler.nodeId` system property, to a value
unique per node.
|`scheduler.executorNode` |`true` |Whether this node runs tasks. Set `false`
on dedicated REST-only nodes if you split roles.
|`scheduler.lockTimeout` |`10000` |Lock timeout in milliseconds
|`scheduler.purgeTaskEnabled` |`true` |Enables the built-in periodic purge of
old completed tasks
@@ -623,11 +626,9 @@ Scheduler settings live in
`etc/org.apache.unomi.services.cfg` (OSGi PID `org.ap
----
# Node 1 and 2: REST + task execution
scheduler.executorNode=true
-scheduler.nodeId=unomi-executor-1
# Node 3: API only, no task execution
scheduler.executorNode=false
-scheduler.nodeId=unomi-api-3
----
After changing configuration, restart the Unomi bundle or the Karaf container.
diff --git a/package/pom.xml b/package/pom.xml
index e633c44aa..eb3debe16 100644
--- a/package/pom.xml
+++ b/package/pom.xml
@@ -229,6 +229,15 @@
</outputDirectory>
<destFileName>org.apache.unomi.persistence.opensearch.cfg</destFileName>
</artifactItem>
+ <artifactItem>
+ <groupId>org.apache.unomi</groupId>
+
<artifactId>unomi-lifecycle-watcher</artifactId>
+ <version>${project.version}</version>
+ <classifier>nodecfg</classifier>
+ <type>cfg</type>
+
<outputDirectory>${project.build.directory}/assembly/etc</outputDirectory>
+
<destFileName>org.apache.unomi.node.cfg</destFileName>
+ </artifactItem>
<artifactItem>
<groupId>org.apache.unomi</groupId>
<artifactId>unomi-services</artifactId>
diff --git a/package/src/main/resources/etc/custom.system.properties
b/package/src/main/resources/etc/custom.system.properties
index 126afb1a4..eb1aeed2d 100644
--- a/package/src/main/resources/etc/custom.system.properties
+++ b/package/src/main/resources/etc/custom.system.properties
@@ -84,10 +84,14 @@
org.apache.unomi.admin.servlet.context=${env:UNOMI_ADMIN_CONTEXT:-/cxs}
# 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 nodeId uniquely identifies this node in the cluster and in the
scheduler. Leave it unset to let
+# Unomi generate one on the first start and reuse it afterwards (it is
remembered in
+# etc/org.apache.unomi.node.cfg). Set it to choose the id yourself, or to give
a node its previous id
+# back; it must be unique in the cluster.
+org.apache.unomi.cluster.nodeId=${env:UNOMI_CLUSTER_NODEID:-}
+# User friendly name of this node, shown next to its id and used as the prefix
of a generated id.
+# Defaults to the host name.
+org.apache.unomi.cluster.nodeName=${env:UNOMI_CLUSTER_NODENAME:-}
# The nodeStatisticsUpdateFrequency controls the frequency of the update of
system statistics such as CPU load,
# system load average and uptime. This value is set in milliseconds and is set
to 10 seconds by default. Each node
# will retrieve the local values and broadcast them through a cluster event to
all the other nodes to update
diff --git
a/services/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 65fb93abe..e56e8dd27 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
@@ -174,7 +174,14 @@ public class ClusterServiceImpl implements ClusterService {
}
}
};
- schedulerService.createRecurringTask("clusterNodeStatisticsUpdate",
nodeStatisticsUpdateFrequency, TimeUnit.MILLISECONDS, statisticsTask, false);
+ // Every node heartbeats, executor or not: a node that stops is
removed from the cluster as stale
+ schedulerService.newTask("clusterNodeStatisticsUpdate")
+ .withPeriod(nodeStatisticsUpdateFrequency,
TimeUnit.MILLISECONDS)
+ .withFixedRate()
+ .withSimpleExecutor(statisticsTask)
+ .nonPersistent()
+ .runOnAllNodes()
+ .schedule();
// Schedule cleanup of stale nodes
TimerTask cleanupTask = new TimerTask() {
@@ -285,6 +292,8 @@ public class ClusterServiceImpl implements ClusterService {
ClusterNode clusterNode = new ClusterNode();
clusterNode.setItemId(nodeId);
+ clusterNode.setNodeName(bundleWatcher != null ?
bundleWatcher.getNodeName() : null);
+ clusterNode.setExecutorNode(schedulerService != null ?
schedulerService.isExecutorNode() : null);
clusterNode.setPublicHostAddress(publicAddress);
clusterNode.setInternalHostAddress(internalAddress);
clusterNode.setStartTime(nodeStartTime);
@@ -376,6 +385,9 @@ public class ClusterServiceImpl implements ClusterService {
try {
// Update its stats
updateSystemStatsForNode(node);
+ if (schedulerService != null) {
+ node.setExecutorNode(schedulerService.isExecutorNode());
+ }
// Update server info if needed
if (bundleWatcher != null &&
!bundleWatcher.getServerInfos().isEmpty()) {
diff --git
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/PersistenceSchedulerProvider.java
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/PersistenceSchedulerProvider.java
index 77a383807..cb538fd47 100644
---
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/PersistenceSchedulerProvider.java
+++
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/PersistenceSchedulerProvider.java
@@ -394,7 +394,9 @@ public class PersistenceSchedulerProvider implements
SchedulerProvider {
long cutoffTime = System.currentTimeMillis() - (5 * 60 *
1000); // 5 minutes threshold
for (ClusterNode node : clusterNodes) {
- if (node.getLastHeartbeat() > cutoffTime) {
+ // A node that does not run tasks must not be picked
as the primary owner of one; a node
+ // that does not report its role (earlier version)
still runs tasks.
+ if (node.getLastHeartbeat() > cutoffTime &&
!Boolean.FALSE.equals(node.getExecutorNode())) {
activeNodes.add(node.getItemId());
}
}
diff --git
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
index 5e84267da..ee10a362c 100644
---
a/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
+++
b/services/src/main/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceImpl.java
@@ -771,6 +771,14 @@ public class SchedulerServiceImpl implements
SchedulerService {
* Blueprint post-construct hook; starts task checking on executor nodes.
*/
public void postConstruct() {
+ // The id comes from the node identity (see BundleWatcher). A blank
one would be shared by every node.
+ if (nodeId == null || nodeId.isBlank()) {
+ String errorMessage = "CRITICAL: scheduler nodeId is not set.
Check that scheduler.nodeId is present in "
+ + "etc/org.apache.unomi.services.cfg.";
+ LOGGER.error(errorMessage);
+ throw new IllegalStateException(errorMessage);
+ }
+
if (bundleContext == null) {
LOGGER.error("BundleContext is null, cannot initialize service
trackers");
return;
@@ -799,10 +807,6 @@ public class SchedulerServiceImpl implements
SchedulerService {
queuePendingOperation(OperationType.INITIALIZE_TASK_PURGE,
"Initialize task purge");
}
- if (nodeId == null) {
- nodeId = UUID.randomUUID().toString();
- }
-
LOGGER.info("Scheduler service initialized. Node ID: {}, Executor
node: {}, Thread pool size: {}",
nodeId, executorNode, Math.max(MIN_THREAD_POOL_SIZE,
threadPoolSize));
diff --git a/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml
b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml
index 6002b721a..96d69dfff 100644
--- a/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml
+++ b/services/src/main/resources/OSGI-INF/blueprint/blueprint.xml
@@ -117,7 +117,7 @@
<cm:property name="scheduler.thread.poolSize" value="5"/>
<cm:property name="scheduler.executorNode" value="true" />
<cm:property name="scheduler.completedTaskTtlDays" value="30" />
- <cm:property name="scheduler.nodeId" value="test-scheduler-node" />
+ <cm:property name="scheduler.nodeId" value="" />
<cm:property name="scheduler.lockTimeout" value="10000" />
<cm:property name="scheduler.purgeTaskEnabled" value="true" />
<cm:property name="scopes.refresh.interval" value="1000"/>
@@ -129,7 +129,7 @@
<cm:property-placeholder persistent-id="org.apache.unomi.cluster"
update-strategy="reload"
placeholder-prefix="${cluster.">
<cm:default-properties>
- <cm:property name="nodeId" value="unomi-node-1"/>
+ <cm:property name="nodeId" value=""/>
<cm:property name="contextserver.publicAddress"
value="https://localhost:9443"/>
<cm:property name="contextserver.internalAddress"
value="http://127.0.0.1:8181"/>
<cm:property name="nodeStatisticsUpdateFrequency" value="10000"/>
diff --git a/services/src/main/resources/org.apache.unomi.cluster.cfg
b/services/src/main/resources/org.apache.unomi.cluster.cfg
index 44fd720db..2d1f5ee89 100644
--- a/services/src/main/resources/org.apache.unomi.cluster.cfg
+++ b/services/src/main/resources/org.apache.unomi.cluster.cfg
@@ -19,10 +19,10 @@
contextserver.publicAddress=${org.apache.unomi.cluster.public.address:-http://localhost:8181}
contextserver.internalAddress=${org.apache.unomi.cluster.internal.address:-https://localhost:9443}
#
-# The nodeId is a required setting that uniquely identifies this node in the
cluster.
-# It must be set to a unique value for each node in the cluster.
-# Example: nodeId=node1
-nodeId=${org.apache.unomi.cluster.nodeId:-unomi-node-1}
+# The nodeId uniquely identifies this node in the cluster. Do not set it here:
when
+# org.apache.unomi.cluster.nodeId (UNOMI_CLUSTER_NODEID) is not set, Unomi
generates an id on the
+# first start, remembers it in etc/org.apache.unomi.node.cfg and provides it
through that property.
+nodeId=${org.apache.unomi.cluster.nodeId:-}
#
# 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
diff --git a/services/src/main/resources/org.apache.unomi.services.cfg
b/services/src/main/resources/org.apache.unomi.services.cfg
index 86c1b8a1a..dccffafda 100644
--- a/services/src/main/resources/org.apache.unomi.services.cfg
+++ b/services/src/main/resources/org.apache.unomi.services.cfg
@@ -89,8 +89,10 @@
rules.optimizationActivated=${org.apache.unomi.rules.optimizationActivated:-true
# The number of threads to compose the pool size of the scheduler.
scheduler.thread.poolSize=${org.apache.unomi.scheduler.thread.poolSize:-5}
-# The node id to use for the scheduler.
-scheduler.nodeId=${org.apache.unomi.scheduler.nodeId:-test-scheduler-node}
+# The node id to use for the scheduler. Do not set it here: Unomi provides the
id of the node (see
+# etc/org.apache.unomi.node.cfg). Set org.apache.unomi.scheduler.nodeId only
to give the scheduler an id
+# of its own, which must then be unique per node.
+scheduler.nodeId=${org.apache.unomi.scheduler.nodeId:-}
# The lock timeout to use for the scheduler.
scheduler.lockTimeout=${org.apache.unomi.scheduler.lockTimeout:-10000}
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/cluster/ClusterServiceImplTest.java
b/services/src/test/java/org/apache/unomi/services/impl/cluster/ClusterServiceImplTest.java
index bd7bc231f..8819220b6 100644
---
a/services/src/test/java/org/apache/unomi/services/impl/cluster/ClusterServiceImplTest.java
+++
b/services/src/test/java/org/apache/unomi/services/impl/cluster/ClusterServiceImplTest.java
@@ -18,6 +18,7 @@ package org.apache.unomi.services.impl.cluster;
import org.apache.unomi.api.ClusterNode;
import org.apache.unomi.api.ServerInfo;
+import org.apache.unomi.api.services.SchedulerService;
import org.apache.unomi.lifecycle.BundleWatcher;
import org.apache.unomi.persistence.spi.PersistenceService;
import
org.apache.unomi.persistence.spi.conditions.evaluator.ConditionEvaluatorDispatcher;
@@ -157,6 +158,45 @@ public class ClusterServiceImplTest {
}
// Add a new test to verify behavior when BundleWatcher is not available
+ @Test
+ public void testHeartbeatRunsOnNonExecutorNodeAndReportsItsRole() {
+ ClusterServiceImpl apiNode =
TestHelper.createClusterService(persistenceService, "api-node", PUBLIC_ADDRESS,
INTERNAL_ADDRESS, bundleContext);
+ apiNode.setNodeStatisticsUpdateFrequency(100);
+ SchedulerServiceImpl apiScheduler = TestHelper.createSchedulerService(
+ "api-node", persistenceService, executionContextManager,
bundleContext, apiNode, -1, false, true);
+ apiNode.setSchedulerService(apiScheduler);
+ try {
+ apiNode.init();
+ ClusterNode registered = persistenceService.load("api-node",
ClusterNode.class);
+ assertEquals(Boolean.FALSE, registered.getExecutorNode());
+ long initialHeartbeat = registered.getLastHeartbeat();
+
+ // Without a heartbeat the node would be removed from the cluster
as stale
+ ClusterNode updated = null;
+ for (int i = 0; i < 100 && (updated == null ||
updated.getLastHeartbeat() <= initialHeartbeat); i++) {
+ Thread.sleep(100);
+ updated = persistenceService.load("api-node",
ClusterNode.class);
+ }
+ assertTrue(updated.getLastHeartbeat() > initialHeartbeat, "the
heartbeat of a non-executor node should run");
+ assertEquals(Boolean.FALSE, updated.getExecutorNode());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ fail("interrupted");
+ } finally {
+ apiScheduler.preDestroy();
+ }
+ }
+
+ @Test
+ public void testInitRegistersNodeNameFromBundleWatcher() {
+ when(bundleWatcher.getNodeName()).thenReturn("web-1");
+ clusterService.setBundleWatcher(bundleWatcher);
+
+ clusterService.init();
+
+ assertEquals("web-1", persistenceService.load(TEST_NODE_ID,
ClusterNode.class).getNodeName());
+ }
+
@Test
public void testInitRegistersNodeInPersistenceWithoutBundleWatcher() {
// Execute without setting a BundleWatcher
@@ -174,26 +214,26 @@ public class ClusterServiceImplTest {
@Test
public void testInitSchedulesStatisticsUpdateTask() {
// We need to use a mock scheduler service for this test to verify the
task creation
- SchedulerServiceImpl mockSchedulerService =
mock(SchedulerServiceImpl.class);
+ SchedulerService mockSchedulerService = mock(SchedulerService.class);
+ SchedulerService.TaskBuilder heartbeatBuilder =
mock(SchedulerService.TaskBuilder.class, RETURNS_SELF);
+
when(mockSchedulerService.newTask("clusterNodeStatisticsUpdate")).thenReturn(heartbeatBuilder);
clusterService.setSchedulerService(mockSchedulerService);
// Execute
clusterService.init();
- // Verify
- verify(mockSchedulerService).createRecurringTask(
- eq("clusterNodeStatisticsUpdate"),
- eq(NODE_STATISTICS_UPDATE_FREQUENCY),
- eq(TimeUnit.MILLISECONDS),
- any(TimerTask.class),
- eq(false)
- );
+ // Verify: a local, recurring heartbeat that runs on every node,
executor or not
+ verify(heartbeatBuilder).withPeriod(NODE_STATISTICS_UPDATE_FREQUENCY,
TimeUnit.MILLISECONDS);
+ verify(heartbeatBuilder).nonPersistent();
+ verify(heartbeatBuilder).runOnAllNodes();
+ verify(heartbeatBuilder).schedule();
}
@Test
public void testInitSchedulesStaleNodesCleanupTask() {
// We need to use a mock scheduler service for this test to verify the
task creation
- SchedulerServiceImpl mockSchedulerService =
mock(SchedulerServiceImpl.class);
+ SchedulerService mockSchedulerService = mock(SchedulerService.class);
+
when(mockSchedulerService.newTask(any())).thenReturn(mock(SchedulerService.TaskBuilder.class,
RETURNS_SELF));
clusterService.setSchedulerService(mockSchedulerService);
// Execute
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/PersistenceSchedulerProviderTest.java
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/PersistenceSchedulerProviderTest.java
index f191c210b..b05619bbd 100644
---
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/PersistenceSchedulerProviderTest.java
+++
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/PersistenceSchedulerProviderTest.java
@@ -164,6 +164,29 @@ public class PersistenceSchedulerProviderTest {
assertFalse(nodes.contains("stale"));
}
+ @Test
+ public void testGetActiveNodesLeavesOutNodesThatDoNotRunTasks() {
+ ClusterNode executor = new ClusterNode();
+ executor.setItemId("executor");
+ executor.setExecutorNode(true);
+ executor.setLastHeartbeat(System.currentTimeMillis());
+ ClusterNode apiOnly = new ClusterNode();
+ apiOnly.setItemId("api-only");
+ apiOnly.setExecutorNode(false);
+ apiOnly.setLastHeartbeat(System.currentTimeMillis());
+ // Written by an earlier version, which does not report the role
+ ClusterNode unknown = new ClusterNode();
+ unknown.setItemId("unknown-role");
+ unknown.setLastHeartbeat(System.currentTimeMillis());
+
when(clusterService.getClusterNodes()).thenReturn(Arrays.asList(executor,
apiOnly, unknown));
+
+ List<String> nodes = provider.getActiveNodes();
+ assertTrue(nodes.contains("provider-node"));
+ assertTrue(nodes.contains("executor"));
+ assertTrue(nodes.contains("unknown-role"));
+ assertFalse(nodes.contains("api-only"));
+ }
+
@Test
public void testGetActiveNodesFallbackWhenClusterUnavailable() {
when(clusterService.getClusterNodes()).thenThrow(new
RuntimeException("down"));
diff --git
a/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceNodeIdTest.java
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceNodeIdTest.java
new file mode 100644
index 000000000..e0624ad64
--- /dev/null
+++
b/services/src/test/java/org/apache/unomi/services/impl/scheduler/SchedulerServiceNodeIdTest.java
@@ -0,0 +1,39 @@
+/*
+ * 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.scheduler;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class SchedulerServiceNodeIdTest {
+
+ /**
+ * A blank id would be the lock owner on every node that has it, so the
scheduler must not start
+ * with one, like the cluster service.
+ */
+ @ParameterizedTest
+ @NullSource
+ @ValueSource(strings = {"", " "})
+ void postConstructFailsOnBlankNodeId(String blankNodeId) {
+ SchedulerServiceImpl scheduler = new SchedulerServiceImpl();
+ scheduler.setNodeId(blankNodeId);
+ assertThrows(IllegalStateException.class, scheduler::postConstruct);
+ }
+}
