+if ($env:MVNW_REPOURL) {
+ $MVNW_REPO_PATTERN = if ($USE_MVND -eq $False) { "/org/apache/maven/" } else { "/maven/mvnd/" }
+ $distributionUrl = "$env:MVNW_REPOURL$MVNW_REPO_PATTERN$($distributionUrl -replace "^.*$MVNW_REPO_PATTERN",'')"
+}
+$distributionUrlName = $distributionUrl -replace '^.*/',''
+$distributionUrlNameMain = $distributionUrlName -replace '\.[^.]*$','' -replace '-bin$',''
+
+$MAVEN_M2_PATH = "$HOME/.m2"
+if ($env:MAVEN_USER_HOME) {
+ $MAVEN_M2_PATH = "$env:MAVEN_USER_HOME"
+}
+
+if (-not (Test-Path -Path $MAVEN_M2_PATH)) {
+ New-Item -Path $MAVEN_M2_PATH -ItemType Directory | Out-Null
+}
+
+$MAVEN_WRAPPER_DISTS = $null
+if ((Get-Item $MAVEN_M2_PATH).Target[0] -eq $null) {
+ $MAVEN_WRAPPER_DISTS = "$MAVEN_M2_PATH/wrapper/dists"
+} else {
+ $MAVEN_WRAPPER_DISTS = (Get-Item $MAVEN_M2_PATH).Target[0] + "/wrapper/dists"
+}
+
+$MAVEN_HOME_PARENT = "$MAVEN_WRAPPER_DISTS/$distributionUrlNameMain"
+$MAVEN_HOME_NAME = ([System.Security.Cryptography.SHA256]::Create().ComputeHash([byte[]][char[]]$distributionUrl) | ForEach-Object {$_.ToString("x2")}) -join ''
+$MAVEN_HOME = "$MAVEN_HOME_PARENT/$MAVEN_HOME_NAME"
+
+if (Test-Path -Path "$MAVEN_HOME" -PathType Container) {
+ Write-Verbose "found existing MAVEN_HOME at $MAVEN_HOME"
+ Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"
+ exit $?
+}
+
+if (! $distributionUrlNameMain -or ($distributionUrlName -eq $distributionUrlNameMain)) {
+ Write-Error "distributionUrl is not valid, must end with *-bin.zip, but found $distributionUrl"
+}
+
+# prepare tmp dir
+$TMP_DOWNLOAD_DIR_HOLDER = New-TemporaryFile
+$TMP_DOWNLOAD_DIR = New-Item -Itemtype Directory -Path "$TMP_DOWNLOAD_DIR_HOLDER.dir"
+$TMP_DOWNLOAD_DIR_HOLDER.Delete() | Out-Null
+trap {
+ if ($TMP_DOWNLOAD_DIR.Exists) {
+ try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+ catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+ }
+}
+
+New-Item -Itemtype Directory -Path "$MAVEN_HOME_PARENT" -Force | Out-Null
+
+# Download and Install Apache Maven
+Write-Verbose "Couldn't find MAVEN_HOME, downloading and installing it ..."
+Write-Verbose "Downloading from: $distributionUrl"
+Write-Verbose "Downloading to: $TMP_DOWNLOAD_DIR/$distributionUrlName"
+
+$webclient = New-Object System.Net.WebClient
+if ($env:MVNW_USERNAME -and $env:MVNW_PASSWORD) {
+ $webclient.Credentials = New-Object System.Net.NetworkCredential($env:MVNW_USERNAME, $env:MVNW_PASSWORD)
+}
+[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
+$webclient.DownloadFile($distributionUrl, "$TMP_DOWNLOAD_DIR/$distributionUrlName") | Out-Null
+
+# If specified, validate the SHA-256 sum of the Maven distribution zip file
+$distributionSha256Sum = (Get-Content -Raw "$scriptDir/.mvn/wrapper/maven-wrapper.properties" | ConvertFrom-StringData).distributionSha256Sum
+if ($distributionSha256Sum) {
+ if ($USE_MVND) {
+ Write-Error "Checksum validation is not supported for maven-mvnd. `nPlease disable validation by removing 'distributionSha256Sum' from your maven-wrapper.properties."
+ }
+ Import-Module $PSHOME\Modules\Microsoft.PowerShell.Utility -Function Get-FileHash
+ if ((Get-FileHash "$TMP_DOWNLOAD_DIR/$distributionUrlName" -Algorithm SHA256).Hash.ToLower() -ne $distributionSha256Sum) {
+ Write-Error "Error: Failed to validate Maven distribution SHA-256, your Maven distribution might be compromised. If you updated your Maven version, you need to update the specified distributionSha256Sum property."
+ }
+}
+
+# unzip and move
+Expand-Archive "$TMP_DOWNLOAD_DIR/$distributionUrlName" -DestinationPath "$TMP_DOWNLOAD_DIR" | Out-Null
+
+# Find the actual extracted directory name (handles snapshots where filename != directory name)
+$actualDistributionDir = ""
+
+# First try the expected directory name (for regular distributions)
+$expectedPath = Join-Path "$TMP_DOWNLOAD_DIR" "$distributionUrlNameMain"
+$expectedMvnPath = Join-Path "$expectedPath" "bin/$MVN_CMD"
+if ((Test-Path -Path $expectedPath -PathType Container) -and (Test-Path -Path $expectedMvnPath -PathType Leaf)) {
+ $actualDistributionDir = $distributionUrlNameMain
+}
+
+# If not found, search for any directory with the Maven executable (for snapshots)
+if (!$actualDistributionDir) {
+ Get-ChildItem -Path "$TMP_DOWNLOAD_DIR" -Directory | ForEach-Object {
+ $testPath = Join-Path $_.FullName "bin/$MVN_CMD"
+ if (Test-Path -Path $testPath -PathType Leaf) {
+ $actualDistributionDir = $_.Name
+ }
+ }
+}
+
+if (!$actualDistributionDir) {
+ Write-Error "Could not find Maven distribution directory in extracted archive"
+}
+
+Write-Verbose "Found extracted Maven distribution directory: $actualDistributionDir"
+Rename-Item -Path "$TMP_DOWNLOAD_DIR/$actualDistributionDir" -NewName $MAVEN_HOME_NAME | Out-Null
+try {
+ Move-Item -Path "$TMP_DOWNLOAD_DIR/$MAVEN_HOME_NAME" -Destination $MAVEN_HOME_PARENT | Out-Null
+} catch {
+ if (! (Test-Path -Path "$MAVEN_HOME" -PathType Container)) {
+ Write-Error "fail to move MAVEN_HOME"
+ }
+} finally {
+ try { Remove-Item $TMP_DOWNLOAD_DIR -Recurse -Force | Out-Null }
+ catch { Write-Warning "Cannot remove $TMP_DOWNLOAD_DIR" }
+}
+
+Write-Output "MVN_CMD=$MAVEN_HOME/bin/$MVN_CMD"
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..800a165
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,241 @@
+
+
+
+
+
+
+ 0.1.0-RC.1
+ 21
+ 21
+ 0.8.15
+ 4.33.0
+
+
+
+ com.kingsrook
+ qbit-build-parent
+ 2.0.0
+
+
+ com.kingsrook.qbits
+ qbit-cluster
+ ${revision}
+ jar
+
+ QBit Cluster
+ QBit for running a QQQ application on N nodes: node registry with heartbeats, lost-node detection and alerts, lease-based leadership for singletons, drain requests and an Admin view of the nodes
+ https://github.com/QRun-IO/qbit-cluster
+
+ 4.0.0
+
+
+
+ Apache License, Version 2.0
+ https://www.apache.org/licenses/LICENSE-2.0
+ repo
+
+
+
+
+
+ kingsrook
+ Kingsrook Team
+ contact@kingsrook.com
+ Kingsrook, LLC
+ https://kingsrook.com
+
+
+
+
+ scm:git:git://github.com/QRun-IO/qbit-cluster.git
+ scm:git:ssh://github.com:QRun-IO/qbit-cluster.git
+ https://github.com/QRun-IO/qbit-cluster
+ HEAD
+
+
+
+
+ com.kingsrook.qqq
+ qqq-backend-core
+
+
+
+
+
+ com.kingsrook.qqq
+ qqq-backend-module-rdbms
+ test
+
+
+ com.h2database
+ h2
+ 2.2.220
+ test
+
+
+ org.liquibase
+ liquibase-core
+ ${liquibase.version}
+ test
+
+
+ org.junit.jupiter
+ junit-jupiter
+ test
+
+
+ org.assertj
+ assertj-core
+ 3.27.7
+ test
+
+
+
+
+
+
+
+ src/main/resources
+ true
+
+ **/*.properties
+
+
+
+ src/main/resources
+ false
+
+ **/*.properties
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-checkstyle-plugin
+ 3.3.1
+
+
+ com.puppycrawl.tools
+ checkstyle
+ 10.12.6
+
+
+
+ checkstyle/config.xml
+ checkstyle/license.txt
+ true
+ true
+ warning
+ true
+
+
+
+ validate
+ validate
+ check
+
+
+
+
+ org.jacoco
+ jacoco-maven-plugin
+ ${jacoco.version}
+
+
+ prepare-agent
+
+ prepare-agent
+
+
+
+ report
+ verify
+
+ report
+
+
+
+ check
+
+ check
+
+
+
+
+ BUNDLE
+
+
+ INSTRUCTION
+ COVEREDRATIO
+ 0.70
+
+
+ CLASS
+ COVEREDRATIO
+ 0.90
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ qqq-snapshot
+
+ true
+
+
+ 4.1.0-SNAPSHOT
+
+
+
+
+ com.kingsrook.qqq
+ qqq-bom-pom
+ ${qqq.snapshot.version}
+ pom
+ import
+
+
+
+
+
+ sonatype-snapshots
+ https://central.sonatype.com/repository/maven-snapshots/
+
+ false
+
+
+ true
+
+
+
+
+
+
diff --git a/src/main/java/com/kingsrook/qbits/cluster/ClusterLeases.java b/src/main/java/com/kingsrook/qbits/cluster/ClusterLeases.java
new file mode 100644
index 0000000..e493516
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/ClusterLeases.java
@@ -0,0 +1,348 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster;
+
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+import com.kingsrook.qbits.cluster.model.ClusterTables;
+import com.kingsrook.qbits.cluster.runtime.ClusterSystem;
+import com.kingsrook.qqq.backend.core.actions.tables.ConditionalUpdateAction;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.cluster.QNodeRoles;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.ConditionalUpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.ACQUIRED_AT;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.EXPIRES_AT;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.HOLDER_NODE_ID;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.LEASE;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.LEASE_VERSION;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.MODIFY_DATE;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.NAME;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
+
+
+/*******************************************************************************
+ ** Lease-based leadership for singletons that are not Quartz jobs: at most one
+ ** node holds a named lease at a time, until it expires.
+ **
+ **
+ ** if(ClusterLeases.tryAcquire("my-singleton", Duration.ofSeconds(60)))
+ ** {
+ ** // this node is the leader for (at most) the next 60 seconds;
+ ** // call tryAcquire (or renew) again well before then to keep it
+ ** }
+ **
+ **
+ ** Every change is ONE atomic compare-and-set (core ConditionalUpdateAction;
+ ** for an RDBMS a single UPDATE cluster_lease SET ... WHERE name = ? AND
+ ** version = ? AND (expires_at < now OR holder_node_id = me OR holder_node_id
+ ** IS NULL)), so two nodes can never both win. Times come from the backend's
+ ** clock (the database clock on an RDBMS), not each pod's own. The version
+ ** counter goes up on every change.
+ **
+ ** The static methods act for this node (QNodeRoles.nodeId()); as(nodeId) acts
+ ** for a given node id (tests, tools). They run as the system user, in the
+ ** current QContext's instance.
+ *******************************************************************************/
+public final class ClusterLeases
+{
+ private static final QLogger LOG = QLogger.getLogger(ClusterLeases.class);
+
+
+
+ /*******************************************************************************
+ ** Utility class.
+ *******************************************************************************/
+ private ClusterLeases()
+ {
+ }
+
+
+
+ /*******************************************************************************
+ ** Acquire (or extend) the named lease for this node. True when this node
+ ** holds it for the next ttl.
+ *******************************************************************************/
+ public static boolean tryAcquire(String name, Duration ttl) throws QException
+ {
+ return (as(QNodeRoles.nodeId()).tryAcquire(name, ttl));
+ }
+
+
+
+ /*******************************************************************************
+ ** Extend a lease this node already holds. False when another node has it.
+ *******************************************************************************/
+ public static boolean renew(String name, Duration ttl) throws QException
+ {
+ return (as(QNodeRoles.nodeId()).renew(name, ttl));
+ }
+
+
+
+ /*******************************************************************************
+ ** Give up a lease this node holds (a no-op otherwise), so another node can
+ ** take it at once.
+ *******************************************************************************/
+ public static boolean release(String name) throws QException
+ {
+ return (as(QNodeRoles.nodeId()).release(name));
+ }
+
+
+
+ /*******************************************************************************
+ ** Whether this node holds the named lease and it has not expired.
+ *******************************************************************************/
+ public static boolean isLeader(String name) throws QException
+ {
+ return (as(QNodeRoles.nodeId()).isLeader(name));
+ }
+
+
+
+ /*******************************************************************************
+ ** Leases for a given node id.
+ *******************************************************************************/
+ public static Holder as(String nodeId)
+ {
+ return (new Holder(nodeId));
+ }
+
+
+
+ /*******************************************************************************
+ ** The lease operations for one node id.
+ *******************************************************************************/
+ public static final class Holder
+ {
+ private final String nodeId;
+
+
+
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ private Holder(String nodeId)
+ {
+ if(nodeId == null || nodeId.isBlank())
+ {
+ throw (new IllegalArgumentException("A lease holder needs a node id"));
+ }
+ this.nodeId = nodeId;
+ }
+
+
+
+ /*******************************************************************************
+ ** Acquire, or extend, the lease: wins when it is free, expired or already
+ ** ours.
+ *******************************************************************************/
+ public boolean tryAcquire(String name, Duration ttl) throws QException
+ {
+ requireTtl(ttl);
+ return (ClusterSystem.call(() ->
+ {
+ Instant now = now();
+ QRecord lease = read(name);
+ if(lease == null)
+ {
+ return (insert(name, now, ttl));
+ }
+
+ boolean ours = nodeId.equals(lease.getValueString(HOLDER_NODE_ID));
+ QQueryFilter freeOrOurs = new QQueryFilter()
+ .withBooleanOperator(QQueryFilter.BooleanOperator.OR)
+ .withCriteria(new QFilterCriteria(EXPIRES_AT, QCriteriaOperator.LESS_THAN, now))
+ .withCriteria(new QFilterCriteria(HOLDER_NODE_ID, QCriteriaOperator.EQUALS, nodeId))
+ .withCriteria(new QFilterCriteria(HOLDER_NODE_ID, QCriteriaOperator.IS_BLANK));
+ boolean won = compareAndSet(lease, freeOrOurs, nodeId, now.plus(ttl), ours && lease.getValueInstant(ACQUIRED_AT) != null ? lease.getValueInstant(ACQUIRED_AT) : now, now);
+ if(won && !ours)
+ {
+ LOG.info("Acquired cluster lease", logPair("lease", name), logPair("nodeId", nodeId), logPair("previousHolder", lease.getValueString(HOLDER_NODE_ID)));
+ }
+ return (won);
+ }));
+ }
+
+
+
+ /*******************************************************************************
+ ** Extend the lease only if this node holds it (expired or not, as long as
+ ** no other node took it meanwhile).
+ *******************************************************************************/
+ public boolean renew(String name, Duration ttl) throws QException
+ {
+ requireTtl(ttl);
+ return (ClusterSystem.call(() ->
+ {
+ Instant now = now();
+ QRecord lease = read(name);
+ if(lease == null || !nodeId.equals(lease.getValueString(HOLDER_NODE_ID)))
+ {
+ return (false);
+ }
+ Instant acquiredAt = lease.getValueInstant(ACQUIRED_AT) == null ? now : lease.getValueInstant(ACQUIRED_AT);
+ return (compareAndSet(lease, new QQueryFilter(new QFilterCriteria(HOLDER_NODE_ID, QCriteriaOperator.EQUALS, nodeId)), nodeId, now.plus(ttl), acquiredAt, now));
+ }));
+ }
+
+
+
+ /*******************************************************************************
+ ** Release the lease if this node holds it.
+ *******************************************************************************/
+ public boolean release(String name) throws QException
+ {
+ return (ClusterSystem.call(() ->
+ {
+ QRecord lease = read(name);
+ if(lease == null || !nodeId.equals(lease.getValueString(HOLDER_NODE_ID)))
+ {
+ return (false);
+ }
+ Instant now = now();
+ return (compareAndSet(lease, new QQueryFilter(new QFilterCriteria(HOLDER_NODE_ID, QCriteriaOperator.EQUALS, nodeId)), null, now, null, now));
+ }));
+ }
+
+
+
+ /*******************************************************************************
+ ** Whether this node holds the lease and it has not expired.
+ *******************************************************************************/
+ public boolean isLeader(String name) throws QException
+ {
+ return (ClusterSystem.call(() ->
+ {
+ QRecord lease = read(name);
+ if(lease == null || !nodeId.equals(lease.getValueString(HOLDER_NODE_ID)))
+ {
+ return (false);
+ }
+ Instant expiresAt = lease.getValueInstant(EXPIRES_AT);
+ return (expiresAt != null && expiresAt.isAfter(now()));
+ }));
+ }
+
+
+
+ /*******************************************************************************
+ ** The CAS: the row must still have the version we read, and pass the guard.
+ *******************************************************************************/
+ private boolean compareAndSet(QRecord lease, QQueryFilter guard, String holder, Instant expiresAt, Instant acquiredAt, Instant now) throws QException
+ {
+ Integer version = lease.getValueInteger(LEASE_VERSION) == null ? 0 : lease.getValueInteger(LEASE_VERSION);
+ QQueryFilter filter = new QQueryFilter()
+ .withCriteria(new QFilterCriteria(NAME, QCriteriaOperator.EQUALS, lease.getValueString(NAME)))
+ .withCriteria(lease.getValueInteger(LEASE_VERSION) == null
+ ? new QFilterCriteria(LEASE_VERSION, QCriteriaOperator.IS_BLANK)
+ : new QFilterCriteria(LEASE_VERSION, QCriteriaOperator.EQUALS, version))
+ .withSubFilter(guard);
+ int count = new ConditionalUpdateAction().execute(new ConditionalUpdateInput(LEASE)
+ .withFilter(filter)
+ .withValue(HOLDER_NODE_ID, holder)
+ .withValue(EXPIRES_AT, expiresAt)
+ .withValue(ACQUIRED_AT, acquiredAt)
+ .withValue(LEASE_VERSION, version + 1)
+ .withValue(MODIFY_DATE, now));
+ return (count == 1);
+ }
+
+
+
+ /*******************************************************************************
+ ** First holder: insert the row. A concurrent insert by another node fails
+ ** on the unique name; then whoever's row is there decides.
+ *******************************************************************************/
+ private boolean insert(String name, Instant now, Duration ttl) throws QException
+ {
+ try
+ {
+ QRecord inserted = new InsertAction().execute(new InsertInput(LEASE).withRecord(new QRecord()
+ .withValue(NAME, name)
+ .withValue(HOLDER_NODE_ID, nodeId)
+ .withValue(EXPIRES_AT, now.plus(ttl))
+ .withValue(ACQUIRED_AT, now)
+ .withValue(LEASE_VERSION, 1))).getRecords().get(0);
+ if(CollectionUtils.nullSafeIsEmpty(inserted.getErrors()))
+ {
+ LOG.info("Acquired cluster lease", logPair("lease", name), logPair("nodeId", nodeId));
+ return (true);
+ }
+ }
+ catch(QException e)
+ {
+ LOG.debug("Lease insert lost a race; reading the winner", logPair("lease", name));
+ }
+ QRecord lease = read(name);
+ return (lease != null && nodeId.equals(lease.getValueString(HOLDER_NODE_ID)));
+ }
+
+
+
+ /*******************************************************************************
+ ** The lease row, or null.
+ *******************************************************************************/
+ private QRecord read(String name) throws QException
+ {
+ List records = new QueryAction().execute(new QueryInput(LEASE)
+ .withFilter(new QQueryFilter(new QFilterCriteria(NAME, QCriteriaOperator.EQUALS, (Serializable) name)))).getRecords();
+ return (records.isEmpty() ? null : records.get(0));
+ }
+
+
+
+ /*******************************************************************************
+ ** The backend clock.
+ *******************************************************************************/
+ private Instant now() throws QException
+ {
+ return (new ConditionalUpdateAction().databaseNow(ClusterTables.LEASE));
+ }
+
+
+
+ /*******************************************************************************
+ ** A lease must last a positive time.
+ *******************************************************************************/
+ private static void requireTtl(Duration ttl)
+ {
+ if(ttl == null || ttl.isNegative() || ttl.isZero())
+ {
+ throw (new IllegalArgumentException("A lease ttl must be positive"));
+ }
+ }
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/ClusterQBitConfig.java b/src/main/java/com/kingsrook/qbits/cluster/ClusterQBitConfig.java
new file mode 100644
index 0000000..7b82714
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/ClusterQBitConfig.java
@@ -0,0 +1,386 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster;
+
+
+import java.util.List;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitConfig;
+import com.kingsrook.qqq.backend.core.utils.StringUtils;
+
+
+/*******************************************************************************
+ ** Host configuration of the Cluster qbit.
+ **
+ ** Table names are fixed (cluster_node, cluster_lease, cluster_setting): the
+ ** QRun qbit standard names a qbit's tables with the qbit's own prefix, and the
+ ** shipped changelog creates exactly those. The app's place in the navigation
+ ** is the host's choice: set parentAppName (for example "admin"), or attach the
+ ** app yourself.
+ *******************************************************************************/
+public class ClusterQBitConfig implements QBitConfig
+{
+ private String backendName;
+ private String appName = "cluster";
+ private String appLabel = "Cluster";
+ private String parentAppName;
+ private String permissionBaseName = "cluster";
+ private String drainPermissionName = "cluster.drain";
+ private String appVersion;
+ private Integer widgetRefreshSeconds = 15;
+ private Boolean registerRuntimeService = true;
+
+
+
+ /*******************************************************************************
+ ** Report invalid settings before adding any metadata.
+ *******************************************************************************/
+ @Override
+ public void validate(QInstance qInstance, List errors)
+ {
+ assertCondition(StringUtils.hasContent(backendName), "backendName is required for the Cluster QBit", errors);
+ assertCondition(backendName == null || qInstance.getBackend(backendName) != null, "backendName must name a backend in the instance", errors);
+ assertCondition(appName != null && appName.matches("[A-Za-z][A-Za-z0-9_]*"), "appName must be a valid name", errors);
+ assertCondition(parentAppName == null || (!parentAppName.equals(appName) && qInstance.getApp(parentAppName) != null), "parentAppName must name an existing, different app", errors);
+ assertCondition(StringUtils.hasContent(permissionBaseName), "permissionBaseName is required", errors);
+ assertCondition(StringUtils.hasContent(drainPermissionName), "drainPermissionName is required", errors);
+ assertCondition(widgetRefreshSeconds != null && widgetRefreshSeconds > 0, "widgetRefreshSeconds must be positive", errors);
+ }
+
+
+
+ /*******************************************************************************
+ ** Tables without their own backend use the configured backend.
+ *******************************************************************************/
+ @Override
+ public String getDefaultBackendNameForTables()
+ {
+ return (backendName);
+ }
+
+
+
+ /*******************************************************************************
+ ** The permission that reads the cluster tables, app and widget.
+ *******************************************************************************/
+ public String getReadPermissionName()
+ {
+ return (permissionBaseName + ".read");
+ }
+
+
+
+ /*******************************************************************************
+ ** The permission that edits the cluster tables.
+ *******************************************************************************/
+ public String getWritePermissionName()
+ {
+ return (permissionBaseName + ".write");
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for backendName
+ *******************************************************************************/
+ public String getBackendName()
+ {
+ return (this.backendName);
+ }
+
+
+
+ /*******************************************************************************
+ ** Setter for backendName
+ *******************************************************************************/
+ public void setBackendName(String backendName)
+ {
+ this.backendName = backendName;
+ }
+
+
+
+ /*******************************************************************************
+ ** Fluent setter for backendName: the backend holding the cluster_ tables
+ ** (required). Lease compare-and-set needs its CONDITIONAL_UPDATE capability
+ ** (the RDBMS and memory backends have it).
+ *******************************************************************************/
+ public ClusterQBitConfig withBackendName(String backendName)
+ {
+ this.backendName = backendName;
+ return (this);
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for appName
+ *******************************************************************************/
+ public String getAppName()
+ {
+ return (this.appName);
+ }
+
+
+
+ /*******************************************************************************
+ ** Setter for appName
+ *******************************************************************************/
+ public void setAppName(String appName)
+ {
+ this.appName = appName;
+ }
+
+
+
+ /*******************************************************************************
+ ** Fluent setter for appName (default "cluster").
+ *******************************************************************************/
+ public ClusterQBitConfig withAppName(String appName)
+ {
+ this.appName = appName;
+ return (this);
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for appLabel
+ *******************************************************************************/
+ public String getAppLabel()
+ {
+ return (this.appLabel);
+ }
+
+
+
+ /*******************************************************************************
+ ** Setter for appLabel
+ *******************************************************************************/
+ public void setAppLabel(String appLabel)
+ {
+ this.appLabel = appLabel;
+ }
+
+
+
+ /*******************************************************************************
+ ** Fluent setter for appLabel (default "Cluster").
+ *******************************************************************************/
+ public ClusterQBitConfig withAppLabel(String appLabel)
+ {
+ this.appLabel = appLabel;
+ return (this);
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for parentAppName
+ *******************************************************************************/
+ public String getParentAppName()
+ {
+ return (this.parentAppName);
+ }
+
+
+
+ /*******************************************************************************
+ ** Setter for parentAppName
+ *******************************************************************************/
+ public void setParentAppName(String parentAppName)
+ {
+ this.parentAppName = parentAppName;
+ }
+
+
+
+ /*******************************************************************************
+ ** Fluent setter for parentAppName: an existing app to attach the Cluster app
+ ** under (for example "admin"). Null (the default) leaves placement to the host.
+ *******************************************************************************/
+ public ClusterQBitConfig withParentAppName(String parentAppName)
+ {
+ this.parentAppName = parentAppName;
+ return (this);
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for permissionBaseName
+ *******************************************************************************/
+ public String getPermissionBaseName()
+ {
+ return (this.permissionBaseName);
+ }
+
+
+
+ /*******************************************************************************
+ ** Setter for permissionBaseName
+ *******************************************************************************/
+ public void setPermissionBaseName(String permissionBaseName)
+ {
+ this.permissionBaseName = permissionBaseName;
+ }
+
+
+
+ /*******************************************************************************
+ ** Fluent setter for permissionBaseName (default "cluster": the tables, app
+ ** and widget use cluster.read and cluster.write).
+ *******************************************************************************/
+ public ClusterQBitConfig withPermissionBaseName(String permissionBaseName)
+ {
+ this.permissionBaseName = permissionBaseName;
+ return (this);
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for drainPermissionName
+ *******************************************************************************/
+ public String getDrainPermissionName()
+ {
+ return (this.drainPermissionName);
+ }
+
+
+
+ /*******************************************************************************
+ ** Setter for drainPermissionName
+ *******************************************************************************/
+ public void setDrainPermissionName(String drainPermissionName)
+ {
+ this.drainPermissionName = drainPermissionName;
+ }
+
+
+
+ /*******************************************************************************
+ ** Fluent setter for drainPermissionName: the exact permission that runs Drain
+ ** node and Cancel drain (default "cluster.drain").
+ *******************************************************************************/
+ public ClusterQBitConfig withDrainPermissionName(String drainPermissionName)
+ {
+ this.drainPermissionName = drainPermissionName;
+ return (this);
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for appVersion
+ *******************************************************************************/
+ public String getAppVersion()
+ {
+ return (this.appVersion);
+ }
+
+
+
+ /*******************************************************************************
+ ** Setter for appVersion
+ *******************************************************************************/
+ public void setAppVersion(String appVersion)
+ {
+ this.appVersion = appVersion;
+ }
+
+
+
+ /*******************************************************************************
+ ** Fluent setter for appVersion: the application version each node reports.
+ ** Without it, the environment variable QQQ_APP_VERSION, else the QQQ version.
+ *******************************************************************************/
+ public ClusterQBitConfig withAppVersion(String appVersion)
+ {
+ this.appVersion = appVersion;
+ return (this);
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for widgetRefreshSeconds
+ *******************************************************************************/
+ public Integer getWidgetRefreshSeconds()
+ {
+ return (this.widgetRefreshSeconds);
+ }
+
+
+
+ /*******************************************************************************
+ ** Setter for widgetRefreshSeconds
+ *******************************************************************************/
+ public void setWidgetRefreshSeconds(Integer widgetRefreshSeconds)
+ {
+ this.widgetRefreshSeconds = widgetRefreshSeconds;
+ }
+
+
+
+ /*******************************************************************************
+ ** Fluent setter for widgetRefreshSeconds (the Nodes widget, default 15).
+ *******************************************************************************/
+ public ClusterQBitConfig withWidgetRefreshSeconds(Integer widgetRefreshSeconds)
+ {
+ this.widgetRefreshSeconds = widgetRefreshSeconds;
+ return (this);
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for registerRuntimeService
+ *******************************************************************************/
+ public Boolean getRegisterRuntimeService()
+ {
+ return (this.registerRuntimeService);
+ }
+
+
+
+ /*******************************************************************************
+ ** Setter for registerRuntimeService
+ *******************************************************************************/
+ public void setRegisterRuntimeService(Boolean registerRuntimeService)
+ {
+ this.registerRuntimeService = registerRuntimeService;
+ }
+
+
+
+ /*******************************************************************************
+ ** Fluent setter for registerRuntimeService: add ClusterRuntimeService to the
+ ** instance's runtime services (default true), so QApplicationLauncher starts
+ ** the heartbeat. A host that starts services itself starts it directly.
+ *******************************************************************************/
+ public ClusterQBitConfig withRegisterRuntimeService(Boolean registerRuntimeService)
+ {
+ this.registerRuntimeService = registerRuntimeService;
+ return (this);
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/ClusterQBitProducer.java b/src/main/java/com/kingsrook/qbits/cluster/ClusterQBitProducer.java
new file mode 100644
index 0000000..37c108c
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/ClusterQBitProducer.java
@@ -0,0 +1,249 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster;
+
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import com.kingsrook.qbits.cluster.health.ClusterHeartbeatHealthCheck;
+import com.kingsrook.qbits.cluster.model.ClusterTables;
+import com.kingsrook.qbits.cluster.processes.ClusterNodeActions;
+import com.kingsrook.qbits.cluster.runtime.ClusterAlerts;
+import com.kingsrook.qbits.cluster.runtime.ClusterRuntimeService;
+import com.kingsrook.qbits.cluster.widgets.ClusterNodesWidgetRenderer;
+import com.kingsrook.qqq.backend.core.actions.permissions.UseOtherPermissionNameCustomPermissionChecker;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.health.HealthCategory;
+import com.kingsrook.qqq.backend.core.health.HealthCheckKind;
+import com.kingsrook.qqq.backend.core.health.HealthCheckMetaData;
+import com.kingsrook.qqq.backend.core.health.HealthChecksMetaData;
+import com.kingsrook.qqq.backend.core.model.dashboard.widgets.WidgetType;
+import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerMultiOutput;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.metadata.dashboard.QWidgetMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.help.bundles.HelpBundles;
+import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.layout.QIcon;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.PermissionLevel;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.QPermissionRules;
+import com.kingsrook.qqq.backend.core.model.metadata.possiblevalues.QPossibleValueSource;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitConfigValidationException;
+import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaDataProducer;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
+
+
+/*******************************************************************************
+ ** Produces the Cluster QBit into an instance:
+ ** - tables cluster_node, cluster_lease, cluster_setting and the node status
+ ** possible values;
+ ** - record actions Drain node and Cancel drain on cluster_node;
+ ** - the Nodes widget and the Cluster app (its home), attached under
+ ** parentAppName when the config names one;
+ ** - the cluster.node.lost alert type, the cluster.node.heartbeat health check
+ ** and the help bundle (help/qbit-cluster.yaml);
+ ** - ClusterRuntimeService among the instance's runtime services.
+ **
+ **
+ ** new ClusterQBitProducer()
+ ** .withQBitConfig(new ClusterQBitConfig().withBackendName("rdbms").withParentAppName("admin"))
+ ** .produce(qInstance);
+ **
+ *******************************************************************************/
+public class ClusterQBitProducer implements QBitMetaDataProducer
+{
+ public static final String GROUP_ID = "com.kingsrook.qbits";
+ public static final String ARTIFACT_ID = "qbit-cluster";
+ public static final String NAMESPACE = "cluster";
+ public static final String HELP_BUNDLE = "help/qbit-cluster.yaml";
+
+ private ClusterQBitConfig qBitConfig;
+
+
+
+ /*******************************************************************************
+ ** Fluent setter for the config.
+ *******************************************************************************/
+ public ClusterQBitProducer withQBitConfig(ClusterQBitConfig config)
+ {
+ this.qBitConfig = config;
+ return (this);
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for the config.
+ *******************************************************************************/
+ @Override
+ public ClusterQBitConfig getQBitConfig()
+ {
+ return (qBitConfig);
+ }
+
+
+
+ /*******************************************************************************
+ ** Stable namespace, independent of the host's navigation label.
+ *******************************************************************************/
+ @Override
+ public String getNamespace()
+ {
+ return (NAMESPACE);
+ }
+
+
+
+ /*******************************************************************************
+ ** The qbit's identity, with the Maven-filtered version.
+ *******************************************************************************/
+ @Override
+ public QBitMetaData getQBitMetaData()
+ {
+ return (new QBitMetaData().withGroupId(GROUP_ID).withArtifactId(ARTIFACT_ID).withVersion(version()).withNamespace(NAMESPACE).withConfig(qBitConfig));
+ }
+
+
+
+ /*******************************************************************************
+ ** The packaged version.
+ *******************************************************************************/
+ public static String version()
+ {
+ try(InputStream stream = ClusterQBitProducer.class.getResourceAsStream("/qbit-cluster.properties"))
+ {
+ Properties properties = new Properties();
+ if(stream != null)
+ {
+ properties.load(stream);
+ }
+ return (properties.getProperty("version", "unknown"));
+ }
+ catch(IOException e)
+ {
+ throw (new IllegalStateException("Cannot read the qbit-cluster version resource", e));
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Validate, then add everything to the instance. Returns empty output, as
+ ** everything is already added.
+ *******************************************************************************/
+ @Override
+ public MetaDataProducerMultiOutput produce(QInstance qInstance) throws QException
+ {
+ if(qBitConfig == null)
+ {
+ throw (new QException("ClusterQBitConfig is required"));
+ }
+ List errors = new ArrayList<>();
+ qBitConfig.validate(qInstance, errors);
+ if(!errors.isEmpty())
+ {
+ throw (new QBitConfigValidationException(qBitConfig, errors));
+ }
+
+ QBitMetaData qBit = getQBitMetaData();
+ qInstance.addQBit(qBit);
+
+ for(QPossibleValueSource possibleValueSource : ClusterTables.possibleValueSources())
+ {
+ qInstance.addPossibleValueSource(possibleValueSource);
+ }
+ List tables = ClusterTables.tables(qBitConfig);
+ for(QTableMetaData table : tables)
+ {
+ table.setSourceQBitName(qBit.getName());
+ qInstance.addTable(table);
+ }
+ for(QProcessMetaData process : ClusterNodeActions.processes(qBitConfig))
+ {
+ process.setSourceQBitName(qBit.getName());
+ qInstance.addProcess(process);
+ }
+
+ qInstance.addWidget(new QWidgetMetaData()
+ .withName(ClusterNodesWidgetRenderer.NAME)
+ .withLabel("Nodes")
+ .withType(WidgetType.TABLE.getType())
+ .withGridColumns(12)
+ .withPermissionRules(readRules())
+ .withCodeReference(new QCodeReference(ClusterNodesWidgetRenderer.class))
+ .withDefaultValue("refreshSeconds", qBitConfig.getWidgetRefreshSeconds()));
+
+ QAppMetaData app = new QAppMetaData()
+ .withName(qBitConfig.getAppName())
+ .withLabel(qBitConfig.getAppLabel())
+ .withIcon(new QIcon("hub"))
+ .withPermissionRules(readRules())
+ .withWidgets(new ArrayList<>(List.of(ClusterNodesWidgetRenderer.NAME)));
+ tables.forEach(app::withChild);
+ qInstance.addApp(app);
+ if(qBitConfig.getParentAppName() != null)
+ {
+ QAppMetaData parent = qInstance.getApp(qBitConfig.getParentAppName());
+ if(CollectionUtils.nonNullList(parent.getChildren()).stream().noneMatch(child -> app.getName().equals(child.getName())))
+ {
+ parent.withChild(app);
+ }
+ }
+
+ ClusterAlerts.register(qInstance, qBit.getName());
+
+ HealthChecksMetaData.ofOrWithNew(qInstance).withCheck(new HealthCheckMetaData()
+ .withName(ClusterHeartbeatHealthCheck.NAME)
+ .withLabel("Cluster heartbeat")
+ .withComponent(NAMESPACE)
+ .withKind(HealthCheckKind.DEPENDENCY)
+ .withCategory(HealthCategory.CORE)
+ .withIntervalSeconds(60)
+ .withCheck(new QCodeReference(ClusterHeartbeatHealthCheck.class)));
+
+ if(Boolean.TRUE.equals(qBitConfig.getRegisterRuntimeService())
+ && CollectionUtils.nonNullList(qInstance.getRuntimeServices()).stream().noneMatch(reference -> ClusterRuntimeService.class.getName().equals(reference.getName())))
+ {
+ qInstance.withRuntimeService(new QCodeReference(ClusterRuntimeService.class));
+ }
+
+ HelpBundles.register(qInstance, HELP_BUNDLE);
+ return (new MetaDataProducerMultiOutput());
+ }
+
+
+
+ /*******************************************************************************
+ ** The app and widget need the exact read permission (cluster.read).
+ *******************************************************************************/
+ private QPermissionRules readRules()
+ {
+ return (new QPermissionRules().withLevel(PermissionLevel.HAS_ACCESS_PERMISSION)
+ .withCustomPermissionChecker(UseOtherPermissionNameCustomPermissionChecker.build(qBitConfig.getReadPermissionName())));
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/health/ClusterHeartbeatHealthCheck.java b/src/main/java/com/kingsrook/qbits/cluster/health/ClusterHeartbeatHealthCheck.java
new file mode 100644
index 0000000..79bc159
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/health/ClusterHeartbeatHealthCheck.java
@@ -0,0 +1,112 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.health;
+
+
+import java.time.Duration;
+import java.time.Instant;
+import com.kingsrook.qbits.cluster.model.ClusterSettings;
+import com.kingsrook.qbits.cluster.model.ClusterTables;
+import com.kingsrook.qbits.cluster.runtime.ClusterRuntime;
+import com.kingsrook.qbits.cluster.runtime.ClusterSystem;
+import com.kingsrook.qqq.backend.core.actions.tables.ConditionalUpdateAction;
+import com.kingsrook.qqq.backend.core.health.HealthCheck;
+import com.kingsrook.qqq.backend.core.health.HealthCheckInput;
+import com.kingsrook.qqq.backend.core.health.HealthCheckOutcome;
+import com.kingsrook.qqq.backend.core.health.HealthDetails;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+
+
+/*******************************************************************************
+ ** cluster.node.heartbeat: this node's own heartbeat is current.
+ ** - UP: the last heartbeat is within two heartbeat intervals;
+ ** - DEGRADED: older than that, but not yet lost_after_seconds (writes are slow
+ ** or failing);
+ ** - DOWN: older than lost_after_seconds (other nodes will mark this one LOST),
+ ** or the row is missing or cannot be read;
+ ** - UNKNOWN: the cluster runtime is not running in this JVM.
+ *******************************************************************************/
+public class ClusterHeartbeatHealthCheck implements HealthCheck
+{
+ public static final String NAME = "cluster.node.heartbeat";
+
+
+
+ /*******************************************************************************
+ ** Check this node's row.
+ *******************************************************************************/
+ @Override
+ public HealthCheckOutcome run(HealthCheckInput input)
+ {
+ ClusterRuntime runtime = ClusterRuntime.current();
+ if(runtime == null)
+ {
+ return (HealthCheckOutcome.unknown("The cluster runtime is not running on this node"));
+ }
+
+ QInstance qInstance = input.getQInstance() == null ? runtime.getQInstance() : input.getQInstance();
+ try
+ {
+ return (ClusterSystem.call(qInstance, () ->
+ {
+ QRecord node = runtime.getRegistry().read();
+ if(node == null)
+ {
+ return (HealthCheckOutcome.down("This node is not registered in the cluster"));
+ }
+ Instant heartbeatAt = node.getValueInstant(ClusterTables.HEARTBEAT_AT);
+ if(heartbeatAt == null)
+ {
+ return (HealthCheckOutcome.down("This node has no heartbeat"));
+ }
+
+ ClusterSettings settings = runtime.getSettings();
+ long age = Math.max(0, Duration.between(heartbeatAt, new ConditionalUpdateAction().databaseNow(ClusterTables.NODE)).toSeconds());
+ HealthDetails details = new HealthDetails()
+ .withValue("nodeId", runtime.getRegistry().getNodeId())
+ .withValue("status", node.getValueString(ClusterTables.STATUS))
+ .withValue("heartbeatAgeSeconds", age)
+ .withValue("heartbeatSeconds", settings.heartbeatSeconds());
+
+ HealthCheckOutcome outcome;
+ if(age <= 2L * settings.heartbeatSeconds())
+ {
+ outcome = HealthCheckOutcome.up("Heartbeat is current");
+ }
+ else if(age < settings.lostAfterSeconds())
+ {
+ outcome = HealthCheckOutcome.degraded("Heartbeat is late (" + age + " seconds old)");
+ }
+ else
+ {
+ outcome = HealthCheckOutcome.down("Heartbeat is " + age + " seconds old; other nodes treat this node as lost");
+ }
+ return (outcome.withDetails(details));
+ }));
+ }
+ catch(Exception e)
+ {
+ return (HealthCheckOutcome.down("This node's heartbeat could not be read"));
+ }
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/model/ClusterSettingValidator.java b/src/main/java/com/kingsrook/qbits/cluster/model/ClusterSettingValidator.java
new file mode 100644
index 0000000..17e74a3
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/model/ClusterSettingValidator.java
@@ -0,0 +1,67 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.model;
+
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
+import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
+
+
+/*******************************************************************************
+ ** Rejects a cluster_setting value that the runtime could not use (for example
+ ** heartbeat_seconds = "soon"), so a typo is caught on save rather than being
+ ** silently replaced by the default. Unknown keys are allowed.
+ *******************************************************************************/
+public class ClusterSettingValidator implements TableCustomizerInterface
+{
+
+ /*******************************************************************************
+ ** Validate inserts and updates (an update may carry only the value; the key
+ ** then comes from the stored row).
+ *******************************************************************************/
+ @Override
+ public List preInsertOrUpdate(AbstractActionInput input, List records, boolean isPreview, Optional> oldRecordList)
+ {
+ Map oldById = new HashMap<>();
+ oldRecordList.ifPresent(list -> list.forEach(old -> oldById.put(old.getValue(ClusterTables.ID), old)));
+
+ for(QRecord record : records)
+ {
+ QRecord old = oldById.get(record.getValue(ClusterTables.ID));
+ String key = record.getValues().containsKey(ClusterTables.SETTING_KEY) ? record.getValueString(ClusterTables.SETTING_KEY) : old == null ? null : old.getValueString(ClusterTables.SETTING_KEY);
+ String value = record.getValues().containsKey(ClusterTables.SETTING_VALUE) ? record.getValueString(ClusterTables.SETTING_VALUE) : old == null ? null : old.getValueString(ClusterTables.SETTING_VALUE);
+ String error = ClusterSettings.validate(key, value);
+ if(error != null)
+ {
+ record.addError(new BadInputStatusMessage(error));
+ }
+ }
+ return (records);
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/model/ClusterSettings.java b/src/main/java/com/kingsrook/qbits/cluster/model/ClusterSettings.java
new file mode 100644
index 0000000..3207865
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/model/ClusterSettings.java
@@ -0,0 +1,195 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.model;
+
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
+
+
+/*******************************************************************************
+ ** The cluster's settings (cluster_setting rows), with the defaults used when a
+ ** row is missing or its value is not valid. The defaults are the same values
+ ** the shipped defaults changelog seeds.
+ **
+ ** @param heartbeatSeconds how often each node writes its heartbeat
+ ** @param lostAfterSeconds a node whose heartbeat is older than this is LOST
+ ** @param retainStoppedDays STOPPED and LOST rows older than this are deleted
+ ** @param alertOnLost whether a lost node raises the cluster.node.lost alert
+ *******************************************************************************/
+public record ClusterSettings(int heartbeatSeconds, int lostAfterSeconds, int retainStoppedDays, boolean alertOnLost)
+{
+ public static final String HEARTBEAT_SECONDS = "heartbeat_seconds";
+ public static final String LOST_AFTER_SECONDS = "lost_after_seconds";
+ public static final String RETAIN_STOPPED_DAYS = "retain_stopped_days";
+ public static final String ALERT_ON_LOST = "alert_on_lost";
+
+ /** each known key, with its default value and its description (as seeded). */
+ public static final Map> DEFAULTS = defaults();
+
+ private static final QLogger LOG = QLogger.getLogger(ClusterSettings.class);
+
+
+
+ /*******************************************************************************
+ ** The built-in defaults.
+ *******************************************************************************/
+ public static ClusterSettings defaultSettings()
+ {
+ return (new ClusterSettings(15, 60, 7, true));
+ }
+
+
+
+ /*******************************************************************************
+ ** Read the settings (the caller supplies a session that may read the table).
+ ** A missing or invalid row keeps its default.
+ *******************************************************************************/
+ public static ClusterSettings load() throws QException
+ {
+ Map values = new LinkedHashMap<>();
+ for(QRecord record : new QueryAction().execute(new QueryInput(ClusterTables.SETTING)).getRecords())
+ {
+ values.put(record.getValueString(ClusterTables.SETTING_KEY), record.getValueString(ClusterTables.SETTING_VALUE));
+ }
+ return (from(values));
+ }
+
+
+
+ /*******************************************************************************
+ ** Settings from key/value pairs.
+ *******************************************************************************/
+ public static ClusterSettings from(Map values)
+ {
+ ClusterSettings defaults = defaultSettings();
+ int heartbeat = positive(values, HEARTBEAT_SECONDS, defaults.heartbeatSeconds());
+ int lostAfter = positive(values, LOST_AFTER_SECONDS, defaults.lostAfterSeconds());
+ if(lostAfter <= heartbeat)
+ {
+ LOG.warn("Cluster setting lost_after_seconds must be greater than heartbeat_seconds; using three heartbeats", logPair("heartbeatSeconds", heartbeat), logPair("lostAfterSeconds", lostAfter));
+ lostAfter = heartbeat * 3;
+ }
+ int retain = positive(values, RETAIN_STOPPED_DAYS, defaults.retainStoppedDays());
+ Boolean alert = parseBoolean(values.get(ALERT_ON_LOST));
+ return (new ClusterSettings(heartbeat, lostAfter, retain, alert == null ? defaults.alertOnLost() : alert));
+ }
+
+
+
+ /*******************************************************************************
+ ** Why a value is not valid for a key (null when it is, or the key is not a
+ ** known one).
+ *******************************************************************************/
+ public static String validate(String key, String value)
+ {
+ if(key == null || !DEFAULTS.containsKey(key))
+ {
+ return (null);
+ }
+ if(ALERT_ON_LOST.equals(key))
+ {
+ return (parseBoolean(value) == null ? "The value of " + key + " must be true or false." : null);
+ }
+ Integer number = parseInteger(value);
+ return (number == null || number < 1 ? "The value of " + key + " must be a whole number of at least 1." : null);
+ }
+
+
+
+ /*******************************************************************************
+ ** A positive integer setting, or the default.
+ *******************************************************************************/
+ private static int positive(Map values, String key, int defaultValue)
+ {
+ String value = values.get(key);
+ if(value == null)
+ {
+ return (defaultValue);
+ }
+ Integer number = parseInteger(value);
+ if(number == null || number < 1)
+ {
+ LOG.warn("Ignoring a cluster setting that is not a positive whole number", logPair("key", key), logPair("value", value));
+ return (defaultValue);
+ }
+ return (number);
+ }
+
+
+
+ /*******************************************************************************
+ ** Parse an integer, or null.
+ *******************************************************************************/
+ private static Integer parseInteger(String value)
+ {
+ try
+ {
+ return (value == null ? null : Integer.valueOf(value.trim()));
+ }
+ catch(NumberFormatException e)
+ {
+ return (null);
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Parse true / false (any case), or null.
+ *******************************************************************************/
+ private static Boolean parseBoolean(String value)
+ {
+ if(value == null)
+ {
+ return (null);
+ }
+ String trimmed = value.trim();
+ if("true".equalsIgnoreCase(trimmed))
+ {
+ return (true);
+ }
+ return ("false".equalsIgnoreCase(trimmed) ? false : null);
+ }
+
+
+
+ /*******************************************************************************
+ ** The seeded keys, values and descriptions (in seed order).
+ *******************************************************************************/
+ private static Map> defaults()
+ {
+ Map> map = new LinkedHashMap<>();
+ map.put(HEARTBEAT_SECONDS, List.of("15", "How often, in seconds, each node records that it is alive."));
+ map.put(LOST_AFTER_SECONDS, List.of("60", "A node with no heartbeat for this many seconds is marked Lost."));
+ map.put(RETAIN_STOPPED_DAYS, List.of("7", "Days to keep the rows of stopped and lost nodes before they are deleted."));
+ map.put(ALERT_ON_LOST, List.of("true", "Whether a lost node raises an alert (true or false)."));
+ return (java.util.Collections.unmodifiableMap(map));
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/model/ClusterTables.java b/src/main/java/com/kingsrook/qbits/cluster/model/ClusterTables.java
new file mode 100644
index 0000000..07c694d
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/model/ClusterTables.java
@@ -0,0 +1,273 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.model;
+
+
+import java.util.List;
+import com.kingsrook.qbits.cluster.ClusterQBitConfig;
+import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.DisplayFormat;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.layout.QIcon;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.PermissionLevel;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.QPermissionRules;
+import com.kingsrook.qqq.backend.core.model.metadata.possiblevalues.QPossibleValueSource;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.Capability;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QFieldSection;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.Tier;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.UniqueKey;
+
+
+/*******************************************************************************
+ ** The three cluster_ tables. Each is an ordinary QQQ table (query, view,
+ ** export, API), secured by the read / write permissions of the configured
+ ** permission base name rather than by hiding it:
+ ** - cluster_node: one row per node, written by the nodes themselves (no insert
+ ** or edit in the UI; Drain node / Cancel drain are record actions; old rows
+ ** may be deleted);
+ ** - cluster_lease: leadership leases (ClusterLeases); an administrator may
+ ** edit a lease (for example clear its holder to force a hand-over);
+ ** - cluster_setting: the cluster's settings, seeded with defaults.
+ **
+ ** Column names are the snake_case of the field names (the shipped Liquibase
+ ** changelog db/liquibase/qbit-cluster/cluster.changelog.sql).
+ *******************************************************************************/
+public final class ClusterTables
+{
+ public static final String NODE = "cluster_node";
+ public static final String LEASE = "cluster_lease";
+ public static final String SETTING = "cluster_setting";
+
+ public static final List ALL = List.of(NODE, LEASE, SETTING);
+
+ ///////////////////////
+ // cluster_node //
+ ///////////////////////
+ public static final String ID = "id";
+ public static final String NODE_ID = "nodeId";
+ public static final String POD_NAME = "podName";
+ public static final String K8S_NODE_NAME = "k8sNodeName";
+ public static final String ZONE = "zone";
+ public static final String DEPLOYMENT = "deployment";
+ public static final String ROLES = "roles";
+ public static final String VERSION = "version";
+ public static final String STARTED_AT = "startedAt";
+ public static final String HEARTBEAT_AT = "heartbeatAt";
+ public static final String STATUS = "status";
+ public static final String WORKER_THREADS = "workerThreads";
+ public static final String SCHEDULER_ACTIVE = "schedulerActive";
+ public static final String ESB_TRIGGERS_ACTIVE = "esbTriggersActive";
+ public static final String ASYNC_JOBS_RUNNING = "asyncJobsRunning";
+ public static final String HEAP_USED_MB = "heapUsedMb";
+ public static final String HEAP_MAX_MB = "heapMaxMb";
+ public static final String DRAIN_REQUESTED = "drainRequested";
+ public static final String DRAIN_REQUESTED_BY = "drainRequestedBy";
+ public static final String DRAIN_REASON = "drainReason";
+ public static final String CREATE_DATE = "createDate";
+ public static final String MODIFY_DATE = "modifyDate";
+
+ ///////////////////////
+ // cluster_lease //
+ ///////////////////////
+ public static final String NAME = "name";
+ public static final String HOLDER_NODE_ID = "holderNodeId";
+ public static final String EXPIRES_AT = "expiresAt";
+ public static final String LEASE_VERSION = "version";
+ public static final String ACQUIRED_AT = "acquiredAt";
+
+ ///////////////////////
+ // cluster_setting //
+ ///////////////////////
+ public static final String SETTING_KEY = "settingKey";
+ public static final String SETTING_VALUE = "settingValue";
+ public static final String DESCRIPTION = "description";
+
+
+
+ /*******************************************************************************
+ ** Utility class.
+ *******************************************************************************/
+ private ClusterTables()
+ {
+ }
+
+
+
+ /*******************************************************************************
+ ** The possible-value sources: node statuses, and nodes (for lease holders).
+ *******************************************************************************/
+ public static List possibleValueSources()
+ {
+ return (List.of(QPossibleValueSource.newForEnum(NodeStatus.NAME, NodeStatus.values())));
+ }
+
+
+
+ /*******************************************************************************
+ ** Every table.
+ *******************************************************************************/
+ public static List tables(ClusterQBitConfig config)
+ {
+ return (List.of(nodeTable(config), leaseTable(config), settingTable(config)));
+ }
+
+
+
+ /*******************************************************************************
+ ** Read / write permissions on the configured base name (cluster.read and
+ ** cluster.write by default).
+ *******************************************************************************/
+ public static QPermissionRules tablePermissionRules(ClusterQBitConfig config)
+ {
+ return (new QPermissionRules().withLevel(PermissionLevel.READ_WRITE_PERMISSIONS).withPermissionBaseName(config.getPermissionBaseName()));
+ }
+
+
+
+ /*******************************************************************************
+ ** cluster_node
+ *******************************************************************************/
+ public static QTableMetaData nodeTable(ClusterQBitConfig config)
+ {
+ QTableMetaData table = base(config, NODE, "Nodes", "dns")
+ .withRecordLabelFormat("%s")
+ .withRecordLabelFields(NODE_ID)
+ .withUniqueKey(new UniqueKey(NODE_ID))
+ .withField(field(NODE_ID, "node_id", QFieldType.STRING).withLabel("Node").withMaxLength(250).withIsRequired(true))
+ .withField(field(POD_NAME, "pod_name", QFieldType.STRING).withLabel("Pod").withMaxLength(250))
+ .withField(field(K8S_NODE_NAME, "k8s_node_name", QFieldType.STRING).withLabel("Kubernetes Node").withMaxLength(250))
+ .withField(field(ZONE, "zone", QFieldType.STRING).withLabel("Zone").withMaxLength(100))
+ .withField(field(DEPLOYMENT, "deployment", QFieldType.STRING).withLabel("Deployment").withMaxLength(250))
+ .withField(field(ROLES, "roles", QFieldType.STRING).withLabel("Roles").withMaxLength(100))
+ .withField(field(VERSION, "version", QFieldType.STRING).withLabel("Version").withMaxLength(100))
+ .withField(field(STARTED_AT, "started_at", QFieldType.DATE_TIME).withLabel("Started"))
+ .withField(field(HEARTBEAT_AT, "heartbeat_at", QFieldType.DATE_TIME).withLabel("Last Heartbeat"))
+ .withField(field(STATUS, "status", QFieldType.STRING).withLabel("Status").withMaxLength(20).withPossibleValueSourceName(NodeStatus.NAME))
+ .withField(field(WORKER_THREADS, "worker_threads", QFieldType.INTEGER).withLabel("Job Worker Threads"))
+ .withField(field(SCHEDULER_ACTIVE, "scheduler_active", QFieldType.BOOLEAN).withLabel("Runs Schedules"))
+ .withField(field(ESB_TRIGGERS_ACTIVE, "esb_triggers_active", QFieldType.BOOLEAN).withLabel("Runs ESB Triggers"))
+ .withField(field(ASYNC_JOBS_RUNNING, "async_jobs_running", QFieldType.INTEGER).withLabel("Steps Running"))
+ .withField(field(HEAP_USED_MB, "heap_used_mb", QFieldType.INTEGER).withLabel("Memory Used (MB)").withDisplayFormat(DisplayFormat.COMMAS))
+ .withField(field(HEAP_MAX_MB, "heap_max_mb", QFieldType.INTEGER).withLabel("Memory Limit (MB)").withDisplayFormat(DisplayFormat.COMMAS))
+ .withField(field(DRAIN_REQUESTED, "drain_requested", QFieldType.BOOLEAN).withLabel("Drain Requested"))
+ .withField(field(DRAIN_REQUESTED_BY, "drain_requested_by", QFieldType.STRING).withLabel("Drain Requested By").withMaxLength(250))
+ .withField(field(DRAIN_REASON, "drain_reason", QFieldType.STRING).withLabel("Drain Reason").withMaxLength(500))
+ .withSection(new QFieldSection("identity", "Node", new QIcon("dns"), Tier.T1, List.of(ID, NODE_ID, STATUS, ROLES, HEARTBEAT_AT)))
+ .withSection(new QFieldSection("placement", "Where It Runs", new QIcon("place"), Tier.T2, List.of(POD_NAME, K8S_NODE_NAME, ZONE, DEPLOYMENT, VERSION)))
+ .withSection(new QFieldSection("work", "Work and Memory", new QIcon("memory"), Tier.T2,
+ List.of(WORKER_THREADS, SCHEDULER_ACTIVE, ESB_TRIGGERS_ACTIVE, ASYNC_JOBS_RUNNING, HEAP_USED_MB, HEAP_MAX_MB)))
+ .withSection(new QFieldSection("drain", "Drain", new QIcon("logout"), Tier.T2, List.of(DRAIN_REQUESTED, DRAIN_REQUESTED_BY, DRAIN_REASON)))
+ .withSection(dates(STARTED_AT));
+ table.getFields().values().forEach(field -> field.setIsEditable(false));
+ table.withoutCapabilities(Capability.TABLE_INSERT, Capability.TABLE_UPDATE);
+ return (table);
+ }
+
+
+
+ /*******************************************************************************
+ ** cluster_lease
+ *******************************************************************************/
+ public static QTableMetaData leaseTable(ClusterQBitConfig config)
+ {
+ QTableMetaData table = base(config, LEASE, "Leases", "key")
+ .withRecordLabelFormat("%s")
+ .withRecordLabelFields(NAME)
+ .withUniqueKey(new UniqueKey(NAME))
+ .withField(field(NAME, "name", QFieldType.STRING).withLabel("Name").withMaxLength(100).withIsRequired(true).withIsEditable(false))
+ .withField(field(HOLDER_NODE_ID, "holder_node_id", QFieldType.STRING).withLabel("Held By").withMaxLength(250))
+ .withField(field(EXPIRES_AT, "expires_at", QFieldType.DATE_TIME).withLabel("Expires"))
+ .withField(field(LEASE_VERSION, "version", QFieldType.INTEGER).withLabel("Version").withIsEditable(false))
+ .withField(field(ACQUIRED_AT, "acquired_at", QFieldType.DATE_TIME).withLabel("Held Since").withIsEditable(false))
+ .withSection(new QFieldSection("identity", "Lease", new QIcon("key"), Tier.T1, List.of(ID, NAME, HOLDER_NODE_ID, EXPIRES_AT)))
+ .withSection(dates(ACQUIRED_AT, LEASE_VERSION));
+ table.withoutCapabilities(Capability.TABLE_INSERT);
+ return (table);
+ }
+
+
+
+ /*******************************************************************************
+ ** cluster_setting
+ *******************************************************************************/
+ public static QTableMetaData settingTable(ClusterQBitConfig config)
+ {
+ QTableMetaData table = base(config, SETTING, "Settings", "tune")
+ .withRecordLabelFormat("%s")
+ .withRecordLabelFields(SETTING_KEY)
+ .withUniqueKey(new UniqueKey(SETTING_KEY))
+ .withField(field(SETTING_KEY, "setting_key", QFieldType.STRING).withLabel("Setting").withMaxLength(100).withIsRequired(true))
+ .withField(field(SETTING_VALUE, "setting_value", QFieldType.STRING).withLabel("Value").withMaxLength(250).withIsRequired(true))
+ .withField(field(DESCRIPTION, "description", QFieldType.STRING).withLabel("Description").withMaxLength(500))
+ .withSection(new QFieldSection("identity", "Setting", new QIcon("tune"), Tier.T1, List.of(ID, SETTING_KEY, SETTING_VALUE, DESCRIPTION)))
+ .withSection(dates());
+ QCodeReference validator = new QCodeReference(ClusterSettingValidator.class);
+ table.withCustomizer(TableCustomizers.PRE_INSERT_RECORD, validator);
+ table.withCustomizer(TableCustomizers.PRE_UPDATE_RECORD, validator);
+ return (table);
+ }
+
+
+
+ /*******************************************************************************
+ ** The id, create and modify date fields and the table basics.
+ *******************************************************************************/
+ private static QTableMetaData base(ClusterQBitConfig config, String name, String label, String icon)
+ {
+ return (new QTableMetaData()
+ .withName(name)
+ .withLabel(label)
+ .withIcon(new QIcon(icon))
+ .withBackendName(config.getBackendName())
+ .withPrimaryKeyField(ID)
+ .withPermissionRules(tablePermissionRules(config))
+ .withField(field(ID, "id", QFieldType.INTEGER).withLabel("Id").withIsEditable(false))
+ .withField(field(CREATE_DATE, "create_date", QFieldType.DATE_TIME).withLabel("Create Date").withIsEditable(false))
+ .withField(field(MODIFY_DATE, "modify_date", QFieldType.DATE_TIME).withLabel("Modify Date").withIsEditable(false)));
+ }
+
+
+
+ /*******************************************************************************
+ ** A field with its column name.
+ *******************************************************************************/
+ private static QFieldMetaData field(String name, String column, QFieldType type)
+ {
+ return (new QFieldMetaData(name, type).withBackendName(column));
+ }
+
+
+
+ /*******************************************************************************
+ ** The dates section (with any extra fields first).
+ *******************************************************************************/
+ private static QFieldSection dates(String... extraFields)
+ {
+ List fields = new java.util.ArrayList<>(List.of(extraFields));
+ fields.add(CREATE_DATE);
+ fields.add(MODIFY_DATE);
+ return (new QFieldSection("dates", "Dates", new QIcon("calendar_month"), Tier.T3, fields));
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/model/NodeStatus.java b/src/main/java/com/kingsrook/qbits/cluster/model/NodeStatus.java
new file mode 100644
index 0000000..6646eb4
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/model/NodeStatus.java
@@ -0,0 +1,78 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.model;
+
+
+import com.kingsrook.qqq.backend.core.model.metadata.possiblevalues.PossibleValueEnum;
+
+
+/*******************************************************************************
+ ** Where a node is in its life (cluster_node.status):
+ ** - UP: running and sending heartbeats;
+ ** - DRAINING: finishing in-flight work, not taking new requests or work;
+ ** - LOST: stopped sending heartbeats without saying goodbye (crash, kill,
+ ** network partition) - set by the sweeper;
+ ** - STOPPED: shut down cleanly.
+ *******************************************************************************/
+public enum NodeStatus implements PossibleValueEnum
+{
+ UP("Up"),
+ DRAINING("Draining"),
+ LOST("Lost"),
+ STOPPED("Stopped");
+
+ public static final String NAME = "cluster_node_status";
+
+ private final String label;
+
+
+
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ NodeStatus(String label)
+ {
+ this.label = label;
+ }
+
+
+
+ /*******************************************************************************
+ ** Possible value id.
+ *******************************************************************************/
+ @Override
+ public String getPossibleValueId()
+ {
+ return (name());
+ }
+
+
+
+ /*******************************************************************************
+ ** Possible value label.
+ *******************************************************************************/
+ @Override
+ public String getPossibleValueLabel()
+ {
+ return (label);
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/processes/ClusterNodeActionSteps.java b/src/main/java/com/kingsrook/qbits/cluster/processes/ClusterNodeActionSteps.java
new file mode 100644
index 0000000..eb56998
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/processes/ClusterNodeActionSteps.java
@@ -0,0 +1,522 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.processes;
+
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import com.kingsrook.qbits.cluster.model.ClusterTables;
+import com.kingsrook.qbits.cluster.model.NodeStatus;
+import com.kingsrook.qbits.cluster.runtime.ClusterSystem;
+import com.kingsrook.qqq.backend.core.actions.tables.ConditionalUpdateAction;
+import com.kingsrook.qqq.backend.core.actions.tables.CountAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.Status;
+import com.kingsrook.qqq.backend.core.model.actions.tables.count.CountInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.ConditionalUpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractLoadStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractTransformStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ProcessSummaryProviderInterface;
+import com.kingsrook.qqq.backend.core.utils.StringUtils;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.DRAIN_REASON;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.DRAIN_REQUESTED;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.DRAIN_REQUESTED_BY;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.ID;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.MODIFY_DATE;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.NODE;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.NODE_ID;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.STATUS;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
+
+
+/*******************************************************************************
+ ** The steps of the two cluster_node record actions, each a standard streamed
+ ** ETL (Review: what will happen and what is skipped and why; Result: what
+ ** happened):
+ ** - Drain node: sets the drain request (who, why) on nodes that are UP. The
+ ** node picks it up on its next heartbeat and starts draining.
+ ** - Cancel drain: clears a request the node has not acted on yet. A node that
+ ** is already draining cannot be undrained.
+ ** Both writes are compare-and-set on the node's status, so a node that starts
+ ** draining between Review and Result is reported, never overwritten.
+ *******************************************************************************/
+public final class ClusterNodeActionSteps
+{
+ public static final String FIELD_REASON = "clusterDrainReason";
+
+ private static final QLogger LOG = QLogger.getLogger(ClusterNodeActionSteps.class);
+
+
+
+ /*******************************************************************************
+ ** Utility class.
+ *******************************************************************************/
+ private ClusterNodeActionSteps()
+ {
+ }
+
+
+
+ /*******************************************************************************
+ ** Review for Drain node.
+ *******************************************************************************/
+ public static class DrainTransform extends BaseTransform
+ {
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ public DrainTransform()
+ {
+ super("node will be asked to drain", "nodes will be asked to drain");
+ }
+
+
+
+ /*******************************************************************************
+ ** Only an UP node without a pending request can be drained.
+ *******************************************************************************/
+ @Override
+ String skipReason(QRecord node)
+ {
+ if(!NodeStatus.UP.name().equals(node.getValueString(STATUS)))
+ {
+ return ("Only a node that is up can be drained; these are draining, lost or stopped.");
+ }
+ if(Boolean.TRUE.equals(node.getValueBoolean(DRAIN_REQUESTED)))
+ {
+ return ("A drain was already requested for these nodes.");
+ }
+ return (null);
+ }
+
+
+
+ /*******************************************************************************
+ ** Warn when the drain would leave no node up.
+ *******************************************************************************/
+ @Override
+ void addWarnings(List lines, int willCount)
+ {
+ try
+ {
+ Integer up = ClusterSystem.call(() -> new CountAction().execute(new CountInput(NODE).withFilter(new QQueryFilter()
+ .withCriteria(new QFilterCriteria(STATUS, QCriteriaOperator.EQUALS, NodeStatus.UP.name()))
+ .withSubFilter(notRequested()))).getCount());
+ if(up != null && willCount > 0 && up - willCount <= 0)
+ {
+ lines.add(new ProcessSummaryLine(Status.WARNING, null, "No other node is up: after this drain, no node will take new requests or work until one starts."));
+ }
+ }
+ catch(QException e)
+ {
+ LOG.debug("Could not count the nodes that are up", e);
+ }
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Result for Drain node.
+ *******************************************************************************/
+ public static class DrainLoad extends BaseLoad
+ {
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ public DrainLoad()
+ {
+ super("node will be asked to drain", "nodes will be asked to drain", "node was asked to drain", "nodes were asked to drain");
+ }
+
+
+
+ /*******************************************************************************
+ ** Set the request where the node is still UP with no request.
+ *******************************************************************************/
+ @Override
+ boolean apply(RunBackendStepInput input, QRecord node, Instant now) throws QException
+ {
+ String reason = input.getValueString(FIELD_REASON);
+ String by = userLabel(QContext.getQSession());
+ int count = ClusterSystem.call(() -> new ConditionalUpdateAction().execute(new ConditionalUpdateInput(NODE)
+ .withFilter(new QQueryFilter()
+ .withCriteria(new QFilterCriteria(ID, QCriteriaOperator.EQUALS, node.getValue(ID)))
+ .withCriteria(new QFilterCriteria(STATUS, QCriteriaOperator.EQUALS, NodeStatus.UP.name()))
+ .withSubFilter(notRequested()))
+ .withValue(DRAIN_REQUESTED, true)
+ .withValue(DRAIN_REQUESTED_BY, by)
+ .withValue(DRAIN_REASON, StringUtils.hasContent(reason) ? truncate(reason.trim(), 500) : null)
+ .withValue(MODIFY_DATE, now)));
+ if(count == 1)
+ {
+ LOG.info("Cluster node drain requested", logPair("nodeId", node.getValueString(NODE_ID)), logPair("requestedBy", by), logPair("reason", reason));
+ }
+ return (count == 1);
+ }
+
+
+
+ /*******************************************************************************
+ ** Why a node could not be asked.
+ *******************************************************************************/
+ @Override
+ String failedMessage()
+ {
+ return ("These nodes stopped being up, or were already asked, before the drain was saved.");
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Review for Cancel drain.
+ *******************************************************************************/
+ public static class CancelTransform extends BaseTransform
+ {
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ public CancelTransform()
+ {
+ super("node's drain request will be cancelled", "nodes' drain requests will be cancelled");
+ }
+
+
+
+ /*******************************************************************************
+ ** Only a pending request on an UP node can be cancelled.
+ *******************************************************************************/
+ @Override
+ String skipReason(QRecord node)
+ {
+ if(!Boolean.TRUE.equals(node.getValueBoolean(DRAIN_REQUESTED)))
+ {
+ return ("No drain was requested for these nodes.");
+ }
+ if(!NodeStatus.UP.name().equals(node.getValueString(STATUS)))
+ {
+ return ("These nodes have already started draining (or stopped). Draining cannot be undone; restart the node instead.");
+ }
+ return (null);
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Result for Cancel drain.
+ *******************************************************************************/
+ public static class CancelLoad extends BaseLoad
+ {
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ public CancelLoad()
+ {
+ super("node's drain request will be cancelled", "nodes' drain requests will be cancelled", "node's drain request was cancelled", "nodes' drain requests were cancelled");
+ }
+
+
+
+ /*******************************************************************************
+ ** Clear the request while the node is still UP.
+ *******************************************************************************/
+ @Override
+ boolean apply(RunBackendStepInput input, QRecord node, Instant now) throws QException
+ {
+ int count = ClusterSystem.call(() -> new ConditionalUpdateAction().execute(new ConditionalUpdateInput(NODE)
+ .withFilter(new QQueryFilter()
+ .withCriteria(new QFilterCriteria(ID, QCriteriaOperator.EQUALS, node.getValue(ID)))
+ .withCriteria(new QFilterCriteria(STATUS, QCriteriaOperator.EQUALS, NodeStatus.UP.name()))
+ .withCriteria(new QFilterCriteria(DRAIN_REQUESTED, QCriteriaOperator.EQUALS, true)))
+ .withValue(DRAIN_REQUESTED, false)
+ .withValue(DRAIN_REQUESTED_BY, null)
+ .withValue(DRAIN_REASON, null)
+ .withValue(MODIFY_DATE, now)));
+ if(count == 1)
+ {
+ LOG.info("Cluster node drain request cancelled", logPair("nodeId", node.getValueString(NODE_ID)), logPair("by", userLabel(QContext.getQSession())));
+ }
+ return (count == 1);
+ }
+
+
+
+ /*******************************************************************************
+ ** Why a request could not be cancelled.
+ *******************************************************************************/
+ @Override
+ String failedMessage()
+ {
+ return ("These nodes started draining before the cancel was saved; draining cannot be undone.");
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Review: sort the selected nodes into "will" and "skipped (why)".
+ *******************************************************************************/
+ abstract static class BaseTransform extends AbstractTransformStep implements ProcessSummaryProviderInterface
+ {
+ private final ProcessSummaryLine will;
+ private final List skipped = new ArrayList<>();
+
+
+
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ BaseTransform(String singular, String plural)
+ {
+ will = line(Status.OK, singular, plural, singular, plural);
+ }
+
+
+
+ /*******************************************************************************
+ ** Why a node is skipped, or null.
+ *******************************************************************************/
+ abstract String skipReason(QRecord node);
+
+
+
+ /*******************************************************************************
+ ** Extra Review lines.
+ *******************************************************************************/
+ void addWarnings(List lines, int willCount)
+ {
+ }
+
+
+
+ /*******************************************************************************
+ ** Sort one page.
+ *******************************************************************************/
+ @Override
+ public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output)
+ {
+ for(QRecord node : input.getRecords())
+ {
+ String reason = skipReason(node);
+ if(reason == null)
+ {
+ will.incrementCountAndAddPrimaryKey(node.getValue(ID));
+ output.addRecord(node);
+ }
+ else
+ {
+ skippedLine(reason).incrementCountAndAddPrimaryKey(node.getValue(ID));
+ }
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** The Review lines (the skipped lines repeat on the Result screen).
+ *******************************************************************************/
+ @Override
+ public ArrayList getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+ {
+ ArrayList lines = new ArrayList<>();
+ if(!isForResultScreen)
+ {
+ will.addSelfToListIfAnyCount(lines);
+ addWarnings(lines, will.getCount() == null ? 0 : will.getCount());
+ }
+ for(ProcessSummaryLine line : skipped)
+ {
+ line.addSelfToListIfAnyCount(lines);
+ }
+ return (lines);
+ }
+
+
+
+ /*******************************************************************************
+ ** The skipped line for a reason.
+ *******************************************************************************/
+ private ProcessSummaryLine skippedLine(String reason)
+ {
+ for(ProcessSummaryLine line : skipped)
+ {
+ if(reason.equals(line.getMessage()))
+ {
+ return (line);
+ }
+ }
+ ProcessSummaryLine line = new ProcessSummaryLine(Status.INFO, 0, reason);
+ line.setSingularFutureMessage("node will be skipped. " + reason);
+ line.setPluralFutureMessage("nodes will be skipped. " + reason);
+ line.setSingularPastMessage("node was skipped. " + reason);
+ line.setPluralPastMessage("nodes were skipped. " + reason);
+ line.setMessage(reason);
+ skipped.add(line);
+ return (line);
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Result: apply the change to each node.
+ *******************************************************************************/
+ abstract static class BaseLoad extends AbstractLoadStep implements ProcessSummaryProviderInterface
+ {
+ private final ProcessSummaryLine done;
+ private final ProcessSummaryLine failed;
+
+
+
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ BaseLoad(String singularFuture, String pluralFuture, String singularPast, String pluralPast)
+ {
+ done = line(Status.OK, singularFuture, pluralFuture, singularPast, pluralPast);
+ failed = new ProcessSummaryLine(Status.ERROR, 0, null);
+ }
+
+
+
+ /*******************************************************************************
+ ** Change one node; false when its state changed first.
+ *******************************************************************************/
+ abstract boolean apply(RunBackendStepInput input, QRecord node, Instant now) throws QException;
+
+
+
+ /*******************************************************************************
+ ** The message for nodes that could not be changed.
+ *******************************************************************************/
+ abstract String failedMessage();
+
+
+
+ /*******************************************************************************
+ ** Apply one page.
+ *******************************************************************************/
+ @Override
+ public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+ {
+ Instant now = new ConditionalUpdateAction().databaseNow(NODE);
+ for(QRecord node : input.getRecords())
+ {
+ if(apply(input, node, now))
+ {
+ done.incrementCountAndAddPrimaryKey(node.getValue(ID));
+ }
+ else
+ {
+ failed.incrementCountAndAddPrimaryKey(node.getValue(ID));
+ }
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** The Result lines.
+ *******************************************************************************/
+ @Override
+ public ArrayList getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+ {
+ ArrayList lines = new ArrayList<>();
+ done.addSelfToListIfAnyCount(lines);
+ failed.setSingularPastMessage("node was not changed. " + failedMessage());
+ failed.setPluralPastMessage("nodes were not changed. " + failedMessage());
+ failed.addSelfToListIfAnyCount(lines);
+ if(getTransformStep() instanceof ProcessSummaryProviderInterface transform)
+ {
+ lines.addAll(transform.getProcessSummary(output, true));
+ }
+ return (lines);
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** drain_requested is false or empty.
+ *******************************************************************************/
+ static QQueryFilter notRequested()
+ {
+ return (new QQueryFilter()
+ .withBooleanOperator(QQueryFilter.BooleanOperator.OR)
+ .withCriteria(new QFilterCriteria(DRAIN_REQUESTED, QCriteriaOperator.EQUALS, false))
+ .withCriteria(new QFilterCriteria(DRAIN_REQUESTED, QCriteriaOperator.IS_BLANK)));
+ }
+
+
+
+ /*******************************************************************************
+ ** Who the session is, for drain_requested_by.
+ *******************************************************************************/
+ static String userLabel(QSession session)
+ {
+ if(session == null || session.getUser() == null)
+ {
+ return ("an administrator");
+ }
+ String label = StringUtils.hasContent(session.getUser().getFullName()) ? session.getUser().getFullName() : session.getUser().getIdReference();
+ return (StringUtils.hasContent(label) ? truncate(label, 250) : "an administrator");
+ }
+
+
+
+ /*******************************************************************************
+ ** A counted summary line.
+ *******************************************************************************/
+ static ProcessSummaryLine line(Status status, String singularFuture, String pluralFuture, String singularPast, String pluralPast)
+ {
+ ProcessSummaryLine line = new ProcessSummaryLine(status, 0, null);
+ line.setSingularFutureMessage(singularFuture);
+ line.setPluralFutureMessage(pluralFuture);
+ line.setSingularPastMessage(singularPast);
+ line.setPluralPastMessage(pluralPast);
+ return (line);
+ }
+
+
+
+ /*******************************************************************************
+ ** Cut text to a column's length.
+ *******************************************************************************/
+ static String truncate(String value, int max)
+ {
+ return (value.length() <= max ? value : value.substring(0, max));
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/processes/ClusterNodeActions.java b/src/main/java/com/kingsrook/qbits/cluster/processes/ClusterNodeActions.java
new file mode 100644
index 0000000..38b7412
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/processes/ClusterNodeActions.java
@@ -0,0 +1,111 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.processes;
+
+
+import java.util.List;
+import com.kingsrook.qbits.cluster.ClusterQBitConfig;
+import com.kingsrook.qbits.cluster.model.ClusterTables;
+import com.kingsrook.qqq.backend.core.actions.permissions.UseOtherPermissionNameCustomPermissionChecker;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.layout.QIcon;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.PermissionLevel;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.QPermissionRules;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QComponentType;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QFrontendComponentMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QFrontendStepMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractLoadStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractTransformStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ExtractViaQueryStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.StreamedETLWithFrontendProcess;
+
+
+/*******************************************************************************
+ ** The cluster_node record actions (never in the navigation): Drain node and
+ ** Cancel drain. Each is a standard streamed ETL process on cluster_node with
+ ** selected records in, a Review screen (the confirmation) and a Result
+ ** screen. Both need the configured drain permission (cluster.drain).
+ *******************************************************************************/
+public final class ClusterNodeActions
+{
+ public static final String DRAIN_NODE = "clusterDrainNode";
+ public static final String CANCEL_DRAIN = "clusterCancelDrain";
+
+
+
+ /*******************************************************************************
+ ** Utility class.
+ *******************************************************************************/
+ private ClusterNodeActions()
+ {
+ }
+
+
+
+ /*******************************************************************************
+ ** Both processes.
+ *******************************************************************************/
+ public static List processes(ClusterQBitConfig config)
+ {
+ QProcessMetaData drain = action(config, DRAIN_NODE, "Drain node", "logout", ClusterNodeActionSteps.DrainTransform.class, ClusterNodeActionSteps.DrainLoad.class,
+ "This node will stop taking new requests and work, finish what it is doing, and keep running until it is stopped or restarted.");
+ drain.withStep(0, new QFrontendStepMetaData()
+ .withName("input")
+ .withLabel("Drain node")
+ .withComponent(new QFrontendComponentMetaData().withType(QComponentType.EDIT_FORM))
+ .withFormFields(List.of(new QFieldMetaData(ClusterNodeActionSteps.FIELD_REASON, QFieldType.STRING).withLabel("Reason").withMaxLength(500))));
+
+ QProcessMetaData cancel = action(config, CANCEL_DRAIN, "Cancel drain", "login", ClusterNodeActionSteps.CancelTransform.class, ClusterNodeActionSteps.CancelLoad.class,
+ "The drain request will be withdrawn before the node acts on it.");
+ return (List.of(drain, cancel));
+ }
+
+
+
+ /*******************************************************************************
+ ** One record action.
+ *******************************************************************************/
+ private static QProcessMetaData action(ClusterQBitConfig config, String name, String label, String icon, Class extends AbstractTransformStep> transform,
+ Class extends AbstractLoadStep> load, String previewMessage)
+ {
+ QProcessMetaData process = StreamedETLWithFrontendProcess.processMetaDataBuilder()
+ .withName(name)
+ .withLabel(label)
+ .withIcon(new QIcon(icon))
+ .withTableName(ClusterTables.NODE)
+ .withSourceTable(ClusterTables.NODE)
+ .withDestinationTable(ClusterTables.NODE)
+ .withExtractStepClass(ExtractViaQueryStep.class)
+ .withTransformStepClass(transform)
+ .withLoadStepClass(load)
+ .withSupportsFullValidation(true)
+ .withDoFullValidation(true)
+ .withMinInputRecords(1)
+ .withPreviewMessage(previewMessage)
+ .getProcessMetaData();
+ process.setPermissionRules(new QPermissionRules().withLevel(PermissionLevel.HAS_ACCESS_PERMISSION)
+ .withCustomPermissionChecker(UseOtherPermissionNameCustomPermissionChecker.build(config.getDrainPermissionName())));
+ return (process);
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterAlerts.java b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterAlerts.java
new file mode 100644
index 0000000..b519f30
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterAlerts.java
@@ -0,0 +1,101 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.runtime;
+
+
+import com.kingsrook.qbits.cluster.model.ClusterTables;
+import com.kingsrook.qqq.backend.core.alerts.AlertRecordRef;
+import com.kingsrook.qqq.backend.core.alerts.AlertSeverity;
+import com.kingsrook.qqq.backend.core.alerts.AlertSignal;
+import com.kingsrook.qqq.backend.core.alerts.AlertsInstanceMetaData;
+import com.kingsrook.qqq.backend.core.alerts.QAlertTypeMetaData;
+import com.kingsrook.qqq.backend.core.alerts.QAlerts;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+
+
+/*******************************************************************************
+ ** The cluster.node.lost alert, through core's alerts SPI (QAlerts). With
+ ** qbit-alerts installed it becomes an Alerts record and notifications;
+ ** without it, core logs it. Raising never fails the sweep (QAlerts catches).
+ *******************************************************************************/
+public final class ClusterAlerts
+{
+ public static final String NODE_LOST = "cluster.node.lost";
+
+
+
+ /*******************************************************************************
+ ** Utility class.
+ *******************************************************************************/
+ private ClusterAlerts()
+ {
+ }
+
+
+
+ /*******************************************************************************
+ ** Declare the alert type (once per instance).
+ *******************************************************************************/
+ public static void register(QInstance qInstance, String qBitName)
+ {
+ if(QAlertTypeMetaData.of(qInstance, NODE_LOST) != null)
+ {
+ return;
+ }
+ AlertsInstanceMetaData.ofOrWithNew(qInstance).withAlertType(new QAlertTypeMetaData()
+ .withName(NODE_LOST)
+ .withLabel("Cluster node lost")
+ .withDefaultSeverity(AlertSeverity.WARNING)
+ .withSubjectTableName(ClusterTables.NODE)
+ .withSourceQBitName(qBitName)
+ .withDescription("A node of the application stopped sending heartbeats without shutting down cleanly.")
+ .withHowToFix("Check why the pod or server stopped (crash, out of memory, eviction, network). The other nodes keep serving; "
+ + "a replacement usually starts on its own. The alert resolves when the node's heartbeat returns or its row is purged."));
+ }
+
+
+
+ /*******************************************************************************
+ ** Raise the alert for a node marked LOST (deduplicated by node id).
+ *******************************************************************************/
+ public static void raiseLost(QRecord node, long secondsSinceHeartbeat)
+ {
+ String nodeId = node.getValueString(ClusterTables.NODE_ID);
+ QAlerts.raise(new AlertSignal(NODE_LOST)
+ .withDedupKey(nodeId)
+ .withTitle("Node " + nodeId + " stopped sending heartbeats")
+ .withSummary("No heartbeat for " + secondsSinceHeartbeat + " seconds. Roles: " + node.getValueString(ClusterTables.ROLES) + ".")
+ .withRecordRef(new AlertRecordRef(ClusterTables.NODE, node.getValue(ClusterTables.ID)))
+ .withAttribute("nodeId", nodeId));
+ }
+
+
+
+ /*******************************************************************************
+ ** Resolve a node's alert (a no-op when none is open).
+ *******************************************************************************/
+ public static void resolveLost(String nodeId, String note)
+ {
+ QAlerts.resolve(NODE_LOST, nodeId, note);
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterNodeInfo.java b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterNodeInfo.java
new file mode 100644
index 0000000..8eb91f8
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterNodeInfo.java
@@ -0,0 +1,201 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.runtime;
+
+
+import java.io.InputStream;
+import java.util.Map;
+import java.util.Properties;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import com.kingsrook.qbits.cluster.ClusterQBitConfig;
+import com.kingsrook.qqq.backend.core.actions.async.AsyncJobRegistry;
+import com.kingsrook.qqq.backend.core.cluster.QNodeRoles;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.utils.StringUtils;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.ASYNC_JOBS_RUNNING;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.DEPLOYMENT;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.ESB_TRIGGERS_ACTIVE;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.HEAP_MAX_MB;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.HEAP_USED_MB;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.K8S_NODE_NAME;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.NODE_ID;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.POD_NAME;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.ROLES;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.SCHEDULER_ACTIVE;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.VERSION;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.WORKER_THREADS;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.ZONE;
+
+
+/*******************************************************************************
+ ** What this node reports about itself. Placement comes from the Kubernetes
+ ** Downward API environment (each optional): QQQ_POD_NAME, QQQ_K8S_NODE_NAME,
+ ** QQQ_ZONE, QQQ_DEPLOYMENT. Roles and role flags come from core QNodeRoles,
+ ** running steps from AsyncJobRegistry, memory from the JVM.
+ *******************************************************************************/
+public class ClusterNodeInfo
+{
+ public static final String ENV_POD_NAME = "QQQ_POD_NAME";
+ public static final String ENV_K8S_NODE = "QQQ_K8S_NODE_NAME";
+ public static final String ENV_ZONE = "QQQ_ZONE";
+ public static final String ENV_DEPLOYMENT = "QQQ_DEPLOYMENT";
+ public static final String ENV_APP_VERSION = "QQQ_APP_VERSION";
+
+ private static final long MB = 1024L * 1024L;
+
+ private final ClusterQBitConfig config;
+ private final Function environment;
+
+
+
+ /*******************************************************************************
+ ** Constructor (the process environment).
+ *******************************************************************************/
+ public ClusterNodeInfo(ClusterQBitConfig config)
+ {
+ this(config, System::getenv);
+ }
+
+
+
+ /*******************************************************************************
+ ** Constructor with an environment lookup (tests).
+ *******************************************************************************/
+ public ClusterNodeInfo(ClusterQBitConfig config, Function environment)
+ {
+ this.config = config;
+ this.environment = environment;
+ }
+
+
+
+ /*******************************************************************************
+ ** This node's id.
+ *******************************************************************************/
+ public String nodeId()
+ {
+ return (QNodeRoles.nodeId());
+ }
+
+
+
+ /*******************************************************************************
+ ** The fixed fields, written when the node starts.
+ *******************************************************************************/
+ public QRecord identity()
+ {
+ return (new QRecord()
+ .withValue(NODE_ID, nodeId())
+ .withValue(POD_NAME, env(ENV_POD_NAME))
+ .withValue(K8S_NODE_NAME, env(ENV_K8S_NODE))
+ .withValue(ZONE, env(ENV_ZONE))
+ .withValue(DEPLOYMENT, env(ENV_DEPLOYMENT))
+ .withValue(ROLES, roles())
+ .withValue(VERSION, version()));
+ }
+
+
+
+ /*******************************************************************************
+ ** The live fields, written on every heartbeat.
+ *******************************************************************************/
+ public Map live()
+ {
+ QNodeRoles.Settings settings = QNodeRoles.get();
+ Runtime runtime = Runtime.getRuntime();
+ Map values = new java.util.LinkedHashMap<>();
+ values.put(WORKER_THREADS, settings.workerThreads());
+ values.put(SCHEDULER_ACTIVE, settings.scheduler());
+ values.put(ESB_TRIGGERS_ACTIVE, settings.esbTriggers());
+ values.put(ASYNC_JOBS_RUNNING, AsyncJobRegistry.runningCount());
+ values.put(HEAP_USED_MB, (int) ((runtime.totalMemory() - runtime.freeMemory()) / MB));
+ values.put(HEAP_MAX_MB, runtime.maxMemory() == Long.MAX_VALUE ? null : (int) (runtime.maxMemory() / MB));
+ return (values);
+ }
+
+
+
+ /*******************************************************************************
+ ** This node's roles as a comma list, in role order (WEB,WORKER,...).
+ *******************************************************************************/
+ public static String roles()
+ {
+ return (QNodeRoles.get().roles().stream().sorted().map(Enum::name).collect(Collectors.joining(",")));
+ }
+
+
+
+ /*******************************************************************************
+ ** The application version: the config, else QQQ_APP_VERSION, else "qqq " and
+ ** the QQQ version.
+ *******************************************************************************/
+ public String version()
+ {
+ if(StringUtils.hasContent(config.getAppVersion()))
+ {
+ return (config.getAppVersion());
+ }
+ String fromEnvironment = env(ENV_APP_VERSION);
+ if(fromEnvironment != null)
+ {
+ return (fromEnvironment);
+ }
+ String qqqVersion = qqqVersion();
+ return (qqqVersion == null ? null : "qqq " + qqqVersion);
+ }
+
+
+
+ /*******************************************************************************
+ ** The QQQ (core) version, from its Maven metadata in the jar.
+ *******************************************************************************/
+ static String qqqVersion()
+ {
+ try(InputStream stream = QInstance.class.getResourceAsStream("/META-INF/maven/com.kingsrook.qqq/qqq-backend-core/pom.properties"))
+ {
+ if(stream == null)
+ {
+ return (QInstance.class.getPackage().getImplementationVersion());
+ }
+ Properties properties = new Properties();
+ properties.load(stream);
+ return (properties.getProperty("version"));
+ }
+ catch(Exception e)
+ {
+ return (null);
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** A trimmed environment value, or null.
+ *******************************************************************************/
+ private String env(String name)
+ {
+ String value = environment.apply(name);
+ return (StringUtils.hasContent(value) ? value.trim() : null);
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterNodeRegistry.java b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterNodeRegistry.java
new file mode 100644
index 0000000..a016a6a
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterNodeRegistry.java
@@ -0,0 +1,285 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.runtime;
+
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.cluster.model.ClusterTables;
+import com.kingsrook.qbits.cluster.model.NodeStatus;
+import com.kingsrook.qqq.backend.core.actions.tables.ConditionalUpdateAction;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.cluster.QNodeLifecycle;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.ConditionalUpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
+import com.kingsrook.qqq.backend.core.utils.StringUtils;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.DRAIN_REASON;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.DRAIN_REQUESTED;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.DRAIN_REQUESTED_BY;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.HEARTBEAT_AT;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.MODIFY_DATE;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.NODE;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.NODE_ID;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.STARTED_AT;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.STATUS;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
+
+
+/*******************************************************************************
+ ** This node's row in cluster_node: registered (upserted) at start, refreshed by
+ ** each heartbeat, and the place a drain request arrives. Writes are
+ ** compare-and-set updates keyed by node id (no audits or record events: this
+ ** is bookkeeping, every 15 seconds per node); the first registration inserts.
+ ** Every method runs as the system user in the given instance.
+ *******************************************************************************/
+public class ClusterNodeRegistry
+{
+ private static final QLogger LOG = QLogger.getLogger(ClusterNodeRegistry.class);
+
+ private final QInstance qInstance;
+ private final ClusterNodeInfo info;
+ private final String nodeId;
+
+ private volatile boolean stopped = false;
+
+
+
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ public ClusterNodeRegistry(QInstance qInstance, ClusterNodeInfo info)
+ {
+ this.qInstance = qInstance;
+ this.info = info;
+ this.nodeId = info.nodeId();
+ }
+
+
+
+ /*******************************************************************************
+ ** Register this node: insert or refresh its row with status UP (or DRAINING
+ ** if it already is), a new start time, and any old drain request cleared (a
+ ** new process does not inherit its predecessor's request).
+ *******************************************************************************/
+ public void register() throws QException
+ {
+ ClusterSystem.call(qInstance, () ->
+ {
+ Instant now = now();
+ QRecord existing = read();
+
+ Map values = new LinkedHashMap<>(info.identity().getValues());
+ values.putAll(info.live());
+ values.put(STATUS, localStatus().name());
+ values.put(STARTED_AT, now);
+ values.put(HEARTBEAT_AT, now);
+ values.put(DRAIN_REQUESTED, false);
+ values.put(DRAIN_REQUESTED_BY, null);
+ values.put(DRAIN_REASON, null);
+
+ if(existing == null)
+ {
+ QRecord record = new QRecord();
+ record.setValues(values);
+ QRecord inserted = new InsertAction().execute(new InsertInput(NODE).withRecord(record)).getRecords().get(0);
+ if(CollectionUtils.nullSafeHasContents(inserted.getErrors()))
+ {
+ throw (new QException("Could not register cluster node " + nodeId + ": " + inserted.getErrors().get(0).getMessage()));
+ }
+ }
+ else
+ {
+ values.put(MODIFY_DATE, now);
+ update(values);
+ resolveIfWasLost(existing);
+ }
+ LOG.info("Registered cluster node", logPair("nodeId", nodeId), logPair("roles", values.get(ClusterTables.ROLES)), logPair("version", values.get(ClusterTables.VERSION)));
+ return (null);
+ });
+ }
+
+
+
+ /*******************************************************************************
+ ** One heartbeat: pick up a drain request (start draining), then write the
+ ** live fields and the heartbeat time. A missing row (deleted, or purged
+ ** while the node was paused) is registered again.
+ *******************************************************************************/
+ public void heartbeat() throws QException
+ {
+ if(stopped)
+ {
+ return;
+ }
+
+ QRecord existing = ClusterSystem.call(qInstance, this::read);
+ if(existing == null)
+ {
+ LOG.info("This node's cluster row is missing; registering again", logPair("nodeId", nodeId));
+ register();
+ return;
+ }
+
+ if(Boolean.TRUE.equals(existing.getValueBoolean(DRAIN_REQUESTED)) && !QNodeLifecycle.isDraining())
+ {
+ String by = StringUtils.hasContent(existing.getValueString(DRAIN_REQUESTED_BY)) ? existing.getValueString(DRAIN_REQUESTED_BY) : "an administrator";
+ String reason = existing.getValueString(DRAIN_REASON);
+ LOG.info("Drain requested for this node", logPair("nodeId", nodeId), logPair("requestedBy", by), logPair("reason", reason));
+ QNodeLifecycle.startDraining("requested by " + by + (StringUtils.hasContent(reason) ? ": " + reason : ""));
+ }
+
+ ClusterSystem.call(qInstance, () ->
+ {
+ Map values = new LinkedHashMap<>(info.live());
+ values.put(HEARTBEAT_AT, now());
+ values.put(STATUS, localStatus().name());
+ update(values);
+ resolveIfWasLost(existing);
+ return (null);
+ });
+ }
+
+
+
+ /*******************************************************************************
+ ** Record a status change of this node (DRAINING, STOPPED). STOPPED also
+ ** records the time, which retention counts from, and stops heartbeats.
+ *******************************************************************************/
+ public void markStatus(NodeStatus status) throws QException
+ {
+ if(stopped)
+ {
+ return;
+ }
+ if(status == NodeStatus.STOPPED)
+ {
+ stopped = true;
+ }
+
+ ClusterSystem.call(qInstance, () ->
+ {
+ Instant now = now();
+ Map values = new LinkedHashMap<>();
+ values.put(STATUS, status.name());
+ values.put(MODIFY_DATE, now);
+ if(status == NodeStatus.STOPPED)
+ {
+ values.put(HEARTBEAT_AT, now);
+ }
+ update(values);
+ LOG.info("Cluster node status", logPair("nodeId", nodeId), logPair("status", status));
+ return (null);
+ });
+ }
+
+
+
+ /*******************************************************************************
+ ** This node's row, or null (the caller runs as the system user).
+ *******************************************************************************/
+ public QRecord read() throws QException
+ {
+ List records = new QueryAction().execute(new QueryInput(NODE)
+ .withFilter(new QQueryFilter(new QFilterCriteria(NODE_ID, QCriteriaOperator.EQUALS, nodeId)))).getRecords();
+ return (records.isEmpty() ? null : records.get(0));
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for nodeId
+ *******************************************************************************/
+ public String getNodeId()
+ {
+ return (nodeId);
+ }
+
+
+
+ /*******************************************************************************
+ ** Whether this node was marked STOPPED.
+ *******************************************************************************/
+ public boolean isStopped()
+ {
+ return (stopped);
+ }
+
+
+
+ /*******************************************************************************
+ ** UP, or DRAINING once core's lifecycle says so.
+ *******************************************************************************/
+ static NodeStatus localStatus()
+ {
+ return (QNodeLifecycle.isDraining() ? NodeStatus.DRAINING : NodeStatus.UP);
+ }
+
+
+
+ /*******************************************************************************
+ ** The update of this node's row.
+ *******************************************************************************/
+ private void update(Map values) throws QException
+ {
+ new ConditionalUpdateAction().execute(new ConditionalUpdateInput(NODE)
+ .withFilter(new QQueryFilter(new QFilterCriteria(NODE_ID, QCriteriaOperator.EQUALS, nodeId)))
+ .withValues(values));
+ }
+
+
+
+ /*******************************************************************************
+ ** A node marked LOST that beats again (a long pause, a partition) resolves
+ ** its alert.
+ *******************************************************************************/
+ private void resolveIfWasLost(QRecord existing)
+ {
+ if(NodeStatus.LOST.name().equals(existing.getValueString(STATUS)))
+ {
+ LOG.info("A node marked lost is sending heartbeats again", logPair("nodeId", nodeId));
+ ClusterAlerts.resolveLost(nodeId, "The node is sending heartbeats again.");
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** The backend clock.
+ *******************************************************************************/
+ private static Instant now() throws QException
+ {
+ return (new ConditionalUpdateAction().databaseNow(NODE));
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterRuntime.java b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterRuntime.java
new file mode 100644
index 0000000..b99be75
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterRuntime.java
@@ -0,0 +1,316 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.runtime;
+
+
+import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import com.kingsrook.qbits.cluster.ClusterLeases;
+import com.kingsrook.qbits.cluster.ClusterQBitConfig;
+import com.kingsrook.qbits.cluster.model.ClusterSettings;
+import com.kingsrook.qbits.cluster.model.NodeStatus;
+import com.kingsrook.qqq.backend.core.cluster.QNodeLifecycle;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
+
+
+/*******************************************************************************
+ ** This node's cluster runtime: one daemon thread that
+ ** 1. registers the node (status UP) at start;
+ ** 2. every heartbeat_seconds writes the heartbeat and live fields, and picks
+ ** up a drain request (QNodeLifecycle.startDraining);
+ ** 3. on the node holding the "cluster-sweeper" lease, runs the lost-node
+ ** sweep and purge (ClusterSweeper).
+ ** A QNodeLifecycle drain listener marks the node DRAINING (whatever started
+ ** the drain: a request, or SIGTERM); stop (the runtime service, or the JVM
+ ** shutdown hook) marks it STOPPED and releases the lease. A failed tick is
+ ** logged and retried on the next one; it never stops the thread.
+ *******************************************************************************/
+public class ClusterRuntime
+{
+ private static final QLogger LOG = QLogger.getLogger(ClusterRuntime.class);
+
+ private static volatile ClusterRuntime current;
+
+ private final QInstance qInstance;
+ private final ClusterNodeRegistry registry;
+ private final ClusterSweeper sweeper;
+
+ private volatile ClusterSettings settings = ClusterSettings.defaultSettings();
+ private volatile ScheduledExecutorService executor;
+ private volatile Thread shutdownHook;
+ private volatile boolean stopped = false;
+
+
+
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ public ClusterRuntime(QInstance qInstance, ClusterQBitConfig config)
+ {
+ this(qInstance, new ClusterNodeInfo(config));
+ }
+
+
+
+ /*******************************************************************************
+ ** Constructor with a node-info source (tests).
+ *******************************************************************************/
+ public ClusterRuntime(QInstance qInstance, ClusterNodeInfo info)
+ {
+ this.qInstance = qInstance;
+ this.registry = new ClusterNodeRegistry(qInstance, info);
+ this.sweeper = new ClusterSweeper(qInstance);
+ }
+
+
+
+ /*******************************************************************************
+ ** The runtime running in this JVM, or null.
+ *******************************************************************************/
+ public static ClusterRuntime current()
+ {
+ return (current);
+ }
+
+
+
+ /*******************************************************************************
+ ** Register the node and the drain listener, without the background thread
+ ** (start() does both; tests drive tick() themselves).
+ *******************************************************************************/
+ public void initialize() throws QException
+ {
+ settings = loadSettings();
+ registry.register();
+ QNodeLifecycle.onDrain(reason ->
+ {
+ try
+ {
+ registry.markStatus(NodeStatus.DRAINING);
+ releaseLease();
+ }
+ catch(Exception e)
+ {
+ LOG.warn("Could not record that this node is draining", e, logPair("nodeId", registry.getNodeId()));
+ }
+ });
+ current = this;
+ }
+
+
+
+ /*******************************************************************************
+ ** Initialize, then heartbeat on a daemon thread, and mark the node STOPPED
+ ** if the JVM shuts down first.
+ *******************************************************************************/
+ public void start() throws QException
+ {
+ initialize();
+ executor = Executors.newSingleThreadScheduledExecutor(runnable ->
+ {
+ Thread thread = new Thread(runnable, "qqq-cluster-heartbeat");
+ thread.setDaemon(true);
+ return (thread);
+ });
+ scheduleNext();
+
+ shutdownHook = new Thread(this::stop, "qqq-cluster-shutdown");
+ Runtime.getRuntime().addShutdownHook(shutdownHook);
+ LOG.info("Cluster runtime started", logPair("nodeId", registry.getNodeId()), logPair("heartbeatSeconds", settings.heartbeatSeconds()));
+ }
+
+
+
+ /*******************************************************************************
+ ** One heartbeat, then (on the leader) one sweep.
+ *******************************************************************************/
+ public void tick()
+ {
+ if(stopped)
+ {
+ return;
+ }
+ try
+ {
+ settings = loadSettings();
+ registry.heartbeat();
+
+ if(QNodeLifecycle.isDraining())
+ {
+ releaseLease();
+ return;
+ }
+
+ ClusterLeases.Holder lease = ClusterLeases.as(registry.getNodeId());
+ boolean leader = ClusterSystem.call(qInstance, () -> lease.tryAcquire(ClusterSweeper.LEASE_NAME, Duration.ofSeconds(settings.lostAfterSeconds())));
+ if(leader)
+ {
+ sweeper.sweep(settings);
+ }
+ }
+ catch(Exception e)
+ {
+ LOG.warn("Cluster heartbeat failed; retrying on the next one", e, logPair("nodeId", registry.getNodeId()));
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Stop (idempotent): stop heartbeats, release the sweeper lease, mark the
+ ** node STOPPED. Never throws.
+ *******************************************************************************/
+ public void stop()
+ {
+ synchronized(this)
+ {
+ if(stopped)
+ {
+ return;
+ }
+ stopped = true;
+ }
+
+ if(executor != null)
+ {
+ executor.shutdownNow();
+ }
+ try
+ {
+ releaseLease();
+ registry.markStatus(NodeStatus.STOPPED);
+ }
+ catch(Exception e)
+ {
+ LOG.warn("Could not mark this cluster node stopped", e, logPair("nodeId", registry.getNodeId()));
+ }
+ removeShutdownHook();
+ if(current == this)
+ {
+ current = null;
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for registry
+ *******************************************************************************/
+ public ClusterNodeRegistry getRegistry()
+ {
+ return (registry);
+ }
+
+
+
+ /*******************************************************************************
+ ** The settings read on the last tick.
+ *******************************************************************************/
+ public ClusterSettings getSettings()
+ {
+ return (settings);
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for qInstance
+ *******************************************************************************/
+ public QInstance getQInstance()
+ {
+ return (qInstance);
+ }
+
+
+
+ /*******************************************************************************
+ ** Run a tick, then schedule the next at the current heartbeat interval.
+ *******************************************************************************/
+ private void scheduleNext()
+ {
+ ScheduledExecutorService service = executor;
+ if(stopped || service == null || service.isShutdown())
+ {
+ return;
+ }
+ service.schedule(() ->
+ {
+ tick();
+ scheduleNext();
+ }, settings.heartbeatSeconds(), TimeUnit.SECONDS);
+ }
+
+
+
+ /*******************************************************************************
+ ** Release the sweeper lease if this node holds it.
+ *******************************************************************************/
+ private void releaseLease() throws QException
+ {
+ ClusterLeases.Holder lease = ClusterLeases.as(registry.getNodeId());
+ ClusterSystem.call(qInstance, () -> lease.release(ClusterSweeper.LEASE_NAME));
+ }
+
+
+
+ /*******************************************************************************
+ ** The settings, or the last ones read when they cannot be read.
+ *******************************************************************************/
+ private ClusterSettings loadSettings()
+ {
+ try
+ {
+ return (ClusterSystem.call(qInstance, ClusterSettings::load));
+ }
+ catch(Exception e)
+ {
+ LOG.warn("Could not read the cluster settings; keeping the current ones", e);
+ return (settings);
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Remove the JVM hook, unless the JVM is already shutting down.
+ *******************************************************************************/
+ private void removeShutdownHook()
+ {
+ Thread hook = shutdownHook;
+ if(hook != null && Thread.currentThread() != hook)
+ {
+ try
+ {
+ Runtime.getRuntime().removeShutdownHook(hook);
+ }
+ catch(IllegalStateException e)
+ {
+ LOG.debug("The JVM is shutting down; the cluster hook stays");
+ }
+ }
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterRuntimeService.java b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterRuntimeService.java
new file mode 100644
index 0000000..9ee0213
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterRuntimeService.java
@@ -0,0 +1,119 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.runtime;
+
+
+import com.kingsrook.qbits.cluster.ClusterQBitConfig;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.instances.QRuntimeServiceInterface;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData;
+
+
+/*******************************************************************************
+ ** Starts and stops this node's cluster runtime with the application.
+ ** QApplicationLauncher runs it from the instance's runtime services (the
+ ** producer registers it); a host that starts services itself (for example
+ ** with QApplicationJavalinServer) calls start(qInstance) after the server
+ ** starts, and stop() at shutdown.
+ *******************************************************************************/
+public class ClusterRuntimeService implements QRuntimeServiceInterface
+{
+ private static final QLogger LOG = QLogger.getLogger(ClusterRuntimeService.class);
+
+ private ClusterRuntime runtime;
+
+
+
+ /*******************************************************************************
+ ** Name of the service.
+ *******************************************************************************/
+ @Override
+ public String getName()
+ {
+ return ("cluster");
+ }
+
+
+
+ /*******************************************************************************
+ ** Register this node and start heartbeats.
+ *******************************************************************************/
+ @Override
+ public void start(QInstance qInstance) throws QException
+ {
+ ClusterQBitConfig config = findConfig(qInstance);
+ if(config == null)
+ {
+ LOG.warn("Cluster runtime service started, but no produced Cluster QBit was found in the QInstance");
+ return;
+ }
+ runtime = new ClusterRuntime(qInstance, config);
+ runtime.start();
+ }
+
+
+
+ /*******************************************************************************
+ ** Mark this node stopped.
+ *******************************************************************************/
+ @Override
+ public void stop()
+ {
+ if(runtime != null)
+ {
+ runtime.stop();
+ }
+ }
+
+
+
+ /*******************************************************************************
+ ** Getter for the runtime (null before start).
+ *******************************************************************************/
+ public ClusterRuntime getRuntime()
+ {
+ return (runtime);
+ }
+
+
+
+ /*******************************************************************************
+ ** The config of the instance's Cluster qbit, or null.
+ *******************************************************************************/
+ static ClusterQBitConfig findConfig(QInstance qInstance)
+ {
+ if(qInstance == null || qInstance.getQBits() == null)
+ {
+ return (null);
+ }
+ for(QBitMetaData qBit : qInstance.getQBits().values())
+ {
+ if(qBit.getConfig() instanceof ClusterQBitConfig config)
+ {
+ return (config);
+ }
+ }
+ return (null);
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterSweeper.java b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterSweeper.java
new file mode 100644
index 0000000..d3c7698
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterSweeper.java
@@ -0,0 +1,180 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.runtime;
+
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import com.kingsrook.qbits.cluster.model.ClusterSettings;
+import com.kingsrook.qbits.cluster.model.ClusterTables;
+import com.kingsrook.qbits.cluster.model.NodeStatus;
+import com.kingsrook.qqq.backend.core.actions.tables.ConditionalUpdateAction;
+import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.ConditionalUpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.HEARTBEAT_AT;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.ID;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.MODIFY_DATE;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.NODE;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.NODE_ID;
+import static com.kingsrook.qbits.cluster.model.ClusterTables.STATUS;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
+
+
+/*******************************************************************************
+ ** The lost-node sweep, run by the one node holding the "cluster-sweeper" lease:
+ ** - a node that is UP or DRAINING with a heartbeat older than
+ ** lost_after_seconds becomes LOST (compare-and-set on its status and
+ ** heartbeat, so a node that beats in between stays UP), and raises the
+ ** cluster.node.lost alert when alert_on_lost is true;
+ ** - STOPPED and LOST rows whose last heartbeat is older than
+ ** retain_stopped_days are deleted (their alerts resolved).
+ ** Every step is idempotent: a sweep interrupted by a crash is simply redone
+ ** by the next leader, so the purge needs no other durable mechanism.
+ *******************************************************************************/
+public class ClusterSweeper
+{
+ public static final String LEASE_NAME = "cluster-sweeper";
+
+ private static final QLogger LOG = QLogger.getLogger(ClusterSweeper.class);
+
+ private static final List LIVE_STATUSES = List.of(NodeStatus.UP.name(), NodeStatus.DRAINING.name());
+ private static final List ENDED_STATUSES = List.of(NodeStatus.STOPPED.name(), NodeStatus.LOST.name());
+
+ private final QInstance qInstance;
+
+
+
+ /*******************************************************************************
+ ** What one sweep did.
+ **
+ ** @param lostNodeIds nodes marked LOST
+ ** @param purged rows deleted
+ *******************************************************************************/
+ public record Result(List lostNodeIds, int purged)
+ {
+ }
+
+
+
+ /*******************************************************************************
+ ** Constructor.
+ *******************************************************************************/
+ public ClusterSweeper(QInstance qInstance)
+ {
+ this.qInstance = qInstance;
+ }
+
+
+
+ /*******************************************************************************
+ ** One sweep with the given settings (the caller holds the lease).
+ *******************************************************************************/
+ public Result sweep(ClusterSettings settings) throws QException
+ {
+ return (ClusterSystem.call(qInstance, () ->
+ {
+ Instant now = new ConditionalUpdateAction().databaseNow(NODE);
+ return (new Result(markLost(settings, now), purge(settings, now)));
+ }));
+ }
+
+
+
+ /*******************************************************************************
+ ** Mark stale live nodes LOST.
+ *******************************************************************************/
+ private List markLost(ClusterSettings settings, Instant now) throws QException
+ {
+ Instant cutoff = now.minusSeconds(settings.lostAfterSeconds());
+ QQueryFilter stale = new QQueryFilter()
+ .withCriteria(new QFilterCriteria(STATUS, QCriteriaOperator.IN, LIVE_STATUSES))
+ .withCriteria(new QFilterCriteria(HEARTBEAT_AT, QCriteriaOperator.LESS_THAN, cutoff));
+
+ List lost = new ArrayList<>();
+ for(QRecord node : new QueryAction().execute(new QueryInput(NODE).withFilter(stale)).getRecords())
+ {
+ QQueryFilter guard = new QQueryFilter()
+ .withCriteria(new QFilterCriteria(ID, QCriteriaOperator.EQUALS, node.getValue(ID)))
+ .withCriteria(new QFilterCriteria(STATUS, QCriteriaOperator.IN, LIVE_STATUSES))
+ .withCriteria(new QFilterCriteria(HEARTBEAT_AT, QCriteriaOperator.LESS_THAN, cutoff));
+ int count = new ConditionalUpdateAction().execute(new ConditionalUpdateInput(NODE).withFilter(guard)
+ .withValue(STATUS, NodeStatus.LOST.name())
+ .withValue(MODIFY_DATE, now));
+ if(count == 1)
+ {
+ String nodeId = node.getValueString(NODE_ID);
+ Instant lastHeartbeat = node.getValueInstant(HEARTBEAT_AT);
+ long seconds = lastHeartbeat == null ? settings.lostAfterSeconds() : Duration.between(lastHeartbeat, now).toSeconds();
+ LOG.warn("Cluster node lost: no heartbeat", logPair("nodeId", nodeId), logPair("secondsSinceHeartbeat", seconds), logPair("previousStatus", node.getValueString(STATUS)));
+ lost.add(nodeId);
+ if(settings.alertOnLost())
+ {
+ ClusterAlerts.raiseLost(node, seconds);
+ }
+ }
+ }
+ return (lost);
+ }
+
+
+
+ /*******************************************************************************
+ ** Delete old STOPPED / LOST rows.
+ *******************************************************************************/
+ private int purge(ClusterSettings settings, Instant now) throws QException
+ {
+ Instant cutoff = now.minus(Duration.ofDays(settings.retainStoppedDays()));
+ QQueryFilter old = new QQueryFilter()
+ .withCriteria(new QFilterCriteria(STATUS, QCriteriaOperator.IN, ENDED_STATUSES))
+ .withCriteria(new QFilterCriteria(HEARTBEAT_AT, QCriteriaOperator.LESS_THAN, cutoff));
+
+ List ids = new ArrayList<>();
+ for(QRecord node : new QueryAction().execute(new QueryInput(NODE).withFilter(old)).getRecords())
+ {
+ ids.add(node.getValue(ID));
+ if(NodeStatus.LOST.name().equals(node.getValueString(STATUS)))
+ {
+ ClusterAlerts.resolveLost(node.getValueString(NODE_ID), "The lost node's row was removed after " + settings.retainStoppedDays() + " days.");
+ }
+ }
+ if(ids.isEmpty())
+ {
+ return (0);
+ }
+ new DeleteAction().execute(new DeleteInput(NODE).withPrimaryKeys(ids));
+ LOG.info("Purged old cluster node rows", logPair("count", ids.size()), logPair("retainDays", settings.retainStoppedDays()));
+ return (ids.size());
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterSystem.java b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterSystem.java
new file mode 100644
index 0000000..c657d9f
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/runtime/ClusterSystem.java
@@ -0,0 +1,100 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.runtime;
+
+
+import com.kingsrook.qqq.backend.core.context.CapturedContext;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
+
+
+/*******************************************************************************
+ ** Runs the qbit's own bookkeeping on the cluster_ tables as the system user
+ ** (never a person's work), then restores the caller's context.
+ *******************************************************************************/
+public final class ClusterSystem
+{
+
+ /*******************************************************************************
+ ** A body that returns a value and may throw.
+ *******************************************************************************/
+ @FunctionalInterface
+ public interface Body
+ {
+ /*******************************************************************************
+ ** Run the body.
+ *******************************************************************************/
+ T run() throws QException;
+ }
+
+
+
+ /*******************************************************************************
+ ** Utility class.
+ *******************************************************************************/
+ private ClusterSystem()
+ {
+ }
+
+
+
+ /*******************************************************************************
+ ** Run in the current instance as the system user.
+ *******************************************************************************/
+ public static T call(Body body) throws QException
+ {
+ QInstance qInstance = QContext.getQInstance();
+ if(qInstance == null)
+ {
+ throw (new QException("The Cluster qbit needs a QContext with an instance"));
+ }
+ return (call(qInstance, body));
+ }
+
+
+
+ /*******************************************************************************
+ ** Run in the given instance as the system user (no transaction).
+ *******************************************************************************/
+ public static T call(QInstance qInstance, Body body) throws QException
+ {
+ CapturedContext original = QContext.capture();
+ try
+ {
+ QContext.init(qInstance, new QSystemUserSession(qInstance));
+ return (body.run());
+ }
+ finally
+ {
+ if(original.qInstance() == null)
+ {
+ QContext.clear();
+ }
+ else
+ {
+ QContext.init(original);
+ }
+ }
+ }
+}
diff --git a/src/main/java/com/kingsrook/qbits/cluster/widgets/ClusterNodesWidgetRenderer.java b/src/main/java/com/kingsrook/qbits/cluster/widgets/ClusterNodesWidgetRenderer.java
new file mode 100644
index 0000000..0cec7e7
--- /dev/null
+++ b/src/main/java/com/kingsrook/qbits/cluster/widgets/ClusterNodesWidgetRenderer.java
@@ -0,0 +1,218 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2026. Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed 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
+ *
+ * https://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 com.kingsrook.qbits.cluster.widgets;
+
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import com.kingsrook.qbits.cluster.model.ClusterTables;
+import com.kingsrook.qbits.cluster.model.NodeStatus;
+import com.kingsrook.qqq.backend.core.actions.dashboard.widgets.AbstractWidgetRenderer;
+import com.kingsrook.qqq.backend.core.actions.tables.ConditionalUpdateAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.cluster.QNodeRoles;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.widgets.RenderWidgetInput;
+import com.kingsrook.qqq.backend.core.model.actions.widgets.RenderWidgetOutput;
+import com.kingsrook.qqq.backend.core.model.dashboard.widgets.TableData;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.utils.StringUtils;
+
+
+/*******************************************************************************
+ ** The Nodes widget (a standard TABLE widget, full width): every node that is
+ ** up, draining or lost, grouped by role (a node with several roles appears
+ ** under each), with its status and how long ago it last sent a heartbeat.
+ ** Stopped nodes are left out; they stay in the Nodes table until purged.
+ ** Reads in the viewer's own session (the widget needs the cluster read
+ ** permission, which also reads the table).
+ *******************************************************************************/
+public class ClusterNodesWidgetRenderer extends AbstractWidgetRenderer
+{
+ public static final String NAME = "clusterNodes";
+
+ static final String ROLE = "role";
+ static final String NODE = "node";
+ static final String STATUS = "status";
+ static final String HEARTBEAT = "heartbeat";
+ static final String WHERE = "where";
+ static final String VERSION = "version";
+
+
+
+ /*******************************************************************************
+ ** Render the table.
+ *******************************************************************************/
+ @Override
+ public RenderWidgetOutput render(RenderWidgetInput input) throws QException
+ {
+ List nodes = new QueryAction().execute(new QueryInput(ClusterTables.NODE)
+ .withFilter(new QQueryFilter(new QFilterCriteria(ClusterTables.STATUS, QCriteriaOperator.NOT_EQUALS, NodeStatus.STOPPED.name()))
+ .withOrderBy(new QFilterOrderBy(ClusterTables.NODE_ID)))).getRecords();
+ Instant now = new ConditionalUpdateAction().databaseNow(ClusterTables.NODE);
+
+ List