From 51b8f1ea7a3f6230f27c531d822688390f3ef359 Mon Sep 17 00:00:00 2001 From: Nicholas Padilla Date: Wed, 29 Jul 2026 05:06:48 -0700 Subject: [PATCH 1/7] feat: Ignite cluster observer and REPLICATED vertx metadata caches Diagnose and optionally act on the orphaned-node / split-brain failure modes while structures is on continuum 2.6.x (vertx 4): - VertxClusterCacheConfiguration: REPLICATED FULL_SYNC template for the __vertx.* caches (continuum collects CacheConfiguration beans). With the default PARTITIONED 0-backup caches, a failed node's entries are destroyed with the topology change before vertx-ignite's cleanup listener runs, which breaks the cleanup election and leaves stale subscriptions that fail with 'Not a member of the cluster' - IgniteClusterObserver: logs membership changes with topology context, segmentation events, and stale routing state after node departures (nodeInfo/subs inspection via binary reads, no vertx-ignite compile dependency); watches the server topology against structures.cluster.observer.minimumClusterSize with arm-after-join, grace period, and startup quorum timeout - observe-only by default: shutdown decisions are logged with an [observe-only] prefix; set structures.cluster.observer.shutdownEnabled =true to actually exit for orchestrator replacement Co-Authored-By: Claude Fable 5 --- .../config/IgniteClusterObserver.java | 372 ++++++++++++++++++ .../VertxClusterCacheConfiguration.java | 38 ++ 2 files changed, 410 insertions(+) create mode 100644 structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java create mode 100644 structures-core/src/main/java/org/kinotic/structures/internal/config/VertxClusterCacheConfiguration.java diff --git a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java new file mode 100644 index 00000000..6079ac24 --- /dev/null +++ b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java @@ -0,0 +1,372 @@ +package org.kinotic.structures.internal.config; + +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import lombok.extern.slf4j.Slf4j; +import org.apache.ignite.Ignite; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.binary.BinaryObject; +import org.apache.ignite.events.DiscoveryEvent; +import org.apache.ignite.events.Event; +import org.apache.ignite.events.EventType; +import org.apache.ignite.lang.IgnitePredicate; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.stereotype.Component; + +import javax.cache.Cache; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Observes Ignite cluster membership from the structures side and diagnoses the + * orphaned-node / split-brain failure modes, logging evidence of each: + * + * + * By default this component only observes and logs. Set + * structures.cluster.observer.shutdownEnabled=true to also shut the process down (non-zero + * exit, so the orchestrator starts a fresh instance) when this node is segmented or stays + * below the minimum cluster size beyond the grace period. minimumClusterSize should be a + * majority of the replica count (floor(n/2)+1). This is a diagnostic port of continuum 3.x's + * IgniteOrphanedNodeGuard for use while structures is on continuum 2.6.x. + */ +@Slf4j +@Component +@ConditionalOnProperty(value = "continuum.disableClustering", havingValue = "false", matchIfMissing = true) +public class IgniteClusterObserver { + + private static final int EXIT_CODE = 1; + private static final long TOPOLOGY_POLL_MS = 10_000L; + private static final long STALE_ROUTE_CHECK_DELAY_MS = 15_000L; + private static final int MAX_STALE_ADDRESSES_LOGGED = 10; + + private final Ignite ignite; + private final ConfigurableApplicationContext applicationContext; + + @Value("${structures.cluster.observer.minimumClusterSize:1}") + private int minimumClusterSize; + + @Value("${structures.cluster.observer.orphanGracePeriodMs:60000}") + private long orphanGracePeriodMs; + + @Value("${structures.cluster.observer.startupQuorumTimeoutMs:300000}") + private long startupQuorumTimeoutMs; + + /** + * Master switch for taking action. False (default) = observe and log only, no outward + * behavior change. True = shut the process down on segmentation or sustained loss of + * the minimum cluster size, so the orchestrator can start a fresh instance. + */ + @Value("${structures.cluster.observer.shutdownEnabled:false}") + private boolean shutdownEnabled; + + @Value("${structures.cluster.observer.shutdownWatchdogTimeoutMs:30000}") + private long shutdownWatchdogTimeoutMs; + + private final AtomicBoolean shutdownInitiated = new AtomicBoolean(false); + private volatile boolean closed = false; + private volatile boolean armed = false; + private volatile boolean observeOnlyReported = false; + private volatile long belowMinimumSinceNanos = -1; + private volatile long startedAtNanos = -1; + private volatile int lastObservedServerNodes = -1; + private volatile boolean topologyUnavailableLogged = false; + + private IgnitePredicate membershipListener; + private IgnitePredicate segmentationListener; + private ScheduledExecutorService scheduler; + + public IgniteClusterObserver(Ignite ignite, ConfigurableApplicationContext applicationContext) { + this.ignite = ignite; + this.applicationContext = applicationContext; + } + + @PostConstruct + public void start() { + scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "structures-cluster-observer"); + thread.setDaemon(true); + return thread; + }); + + // Membership diagnostics: log joins/departures with topology context and, after a + // departure, verify the vertx routing caches were actually cleaned up + membershipListener = event -> { + DiscoveryEvent discoveryEvent = (DiscoveryEvent) event; + String eventNodeId = discoveryEvent.eventNode().id().toString(); + long topologyVersion = discoveryEvent.topologyVersion(); + int serverNodes = safeServerTopologySize(); + switch (event.type()) { + case EventType.EVT_NODE_JOINED -> + log.info("Cluster node joined: {} (topologyVersion={}, serverNodes={})", + eventNodeId, topologyVersion, serverNodes); + case EventType.EVT_NODE_LEFT -> + log.info("Cluster node left: {} (topologyVersion={}, serverNodes={})", + eventNodeId, topologyVersion, serverNodes); + case EventType.EVT_NODE_FAILED -> + log.warn("Cluster node FAILED: {} (topologyVersion={}, serverNodes={})", + eventNodeId, topologyVersion, serverNodes); + default -> { /* not registered for others */ } + } + if (event.type() == EventType.EVT_NODE_LEFT || event.type() == EventType.EVT_NODE_FAILED) { + // Check after vertx-ignite's cleanup listener has had ample time to run + scheduler.schedule(() -> reportStaleRoutingState(eventNodeId), + STALE_ROUTE_CHECK_DELAY_MS, TimeUnit.MILLISECONDS); + } + return true; + }; + ignite.events().localListen(membershipListener, + EventType.EVT_NODE_JOINED, + EventType.EVT_NODE_LEFT, + EventType.EVT_NODE_FAILED); + + // Segmentation is always logged; a segmented server node can never rejoin without + // a restart, so with shutdownEnabled we exit for a fresh instance + segmentationListener = event -> { + actOn("Ignite node was segmented from the cluster. " + + "Segmented server nodes cannot rejoin without a restart."); + return false; // one shot + }; + ignite.events().localListen(segmentationListener, EventType.EVT_NODE_SEGMENTED); + + log.info("Ignite cluster observer started: minimumClusterSize={}, orphanGracePeriodMs={}, " + + "startupQuorumTimeoutMs={}, shutdownEnabled={}", + minimumClusterSize, orphanGracePeriodMs, startupQuorumTimeoutMs, shutdownEnabled); + + if (minimumClusterSize > 1) { + startedAtNanos = System.nanoTime(); + scheduler.scheduleWithFixedDelay(this::checkTopologySafely, + 0, + TOPOLOGY_POLL_MS, + TimeUnit.MILLISECONDS); + } + } + + @PreDestroy + public void stop() { + closed = true; + // A normal context shutdown must never be escalated to a JVM halt by an in-flight poll + shutdownInitiated.set(true); + if (scheduler != null) { + scheduler.shutdownNow(); + } + if (membershipListener != null) { + try { + ignite.events().stopLocalListen(membershipListener, + EventType.EVT_NODE_JOINED, + EventType.EVT_NODE_LEFT, + EventType.EVT_NODE_FAILED); + } catch (Exception e) { + log.debug("Could not remove membership listener during shutdown", e); + } + } + if (segmentationListener != null) { + try { + ignite.events().stopLocalListen(segmentationListener, EventType.EVT_NODE_SEGMENTED); + } catch (Exception e) { + log.debug("Could not remove segmentation listener during shutdown", e); + } + } + } + + /** + * Inspect the vertx-ignite routing caches after a node departed and log any stale + * state left behind - the direct evidence of the cleanup-election failure that + * produces "Not a member of the cluster" event bus send errors. + */ + private void reportStaleRoutingState(String departedNodeId) { + if (closed) { + return; + } + try { + IgniteCache nodeInfoCache = ignite.cache("__vertx.nodeInfo"); + boolean nodeInfoPresent = nodeInfoCache != null && nodeInfoCache.containsKey(departedNodeId); + + int staleSubs = 0; + List staleAddresses = new ArrayList<>(); + // Read the subs cache in binary form to avoid a compile-time dependency on + // vertx-ignite's IgniteRegistrationInfo (field names match its writeBinary) + IgniteCache subsCache = ignite.cache("__vertx.subs"); + if (subsCache != null) { + for (Cache.Entry entry : subsCache.withKeepBinary()) { + if (entry.getKey() instanceof BinaryObject key + && departedNodeId.equals(key.field("nodeId"))) { + staleSubs++; + if (staleAddresses.size() < MAX_STALE_ADDRESSES_LOGGED) { + staleAddresses.add(key.field("address")); + } + } + } + } + + if (nodeInfoPresent || staleSubs > 0) { + log.warn("Stale routing state remains for departed node {}: nodeInfoStillPresent={}, " + + "staleSubscriptionEntries={}, sampleAddresses={}. Event bus sends to these " + + "addresses can fail with 'Not a member of the cluster' until handlers re-register.", + departedNodeId, nodeInfoPresent, staleSubs, staleAddresses); + } else { + log.debug("Routing caches are clean after departure of node {}", departedNodeId); + } + } catch (Exception e) { + log.debug("Could not inspect routing caches after departure of node {}", departedNodeId, e); + } + } + + /** + * An exception escaping a scheduled task silently cancels all future executions - never + * let that happen + */ + private void checkTopologySafely() { + try { + checkTopology(); + } catch (Throwable t) { + log.error("Unexpected error in cluster observer check", t); + } + } + + private void checkTopology() { + if (closed || shutdownInitiated.get()) { + return; + } + + int serverNodes; + try { + serverNodes = ignite.cluster().forServers().nodes().size(); + } catch (Exception e) { + lastObservedServerNodes = 0; + if (!topologyUnavailableLogged) { + topologyUnavailableLogged = true; + log.warn("Ignite topology is not queryable", e); + } + return; + } + topologyUnavailableLogged = false; + + lastObservedServerNodes = serverNodes; + + if (log.isDebugEnabled()) { + log.debug("Topology poll: serverNodes={}, minimum={}, armed={}, belowMinimumForMs={}", + serverNodes, minimumClusterSize, armed, + belowMinimumSinceNanos == -1 ? 0 : elapsedMs(belowMinimumSinceNanos)); + } + + if (serverNodes >= minimumClusterSize) { + if (!armed) { + armed = true; + log.info("Cluster observer armed: server topology reached {} nodes", serverNodes); + } else if (belowMinimumSinceNanos != -1) { + log.info("Server topology recovered to {} nodes after {} ms below minimum", + serverNodes, elapsedMs(belowMinimumSinceNanos)); + } + belowMinimumSinceNanos = -1; + observeOnlyReported = false; + return; + } + + // Below the minimum. Normal startup never trips this (arm-after-join), but a node + // that NEVER reaches the minimum likely started into an ongoing partition; Ignite + // topologies never merge once formed. + if (!armed) { + if (startupQuorumTimeoutMs > 0 && elapsedMs(startedAtNanos) >= startupQuorumTimeoutMs) { + actOn("Server topology never reached the minimum cluster size of " + + minimumClusterSize + " within " + startupQuorumTimeoutMs + + " ms of startup. This node likely started into an ongoing partition " + + "and would run split-brained."); + } + return; + } + + if (belowMinimumSinceNanos == -1) { + belowMinimumSinceNanos = System.nanoTime(); + log.warn("Server topology dropped to {} nodes (minimum {}). Action in {} ms unless it recovers " + + "(shutdownEnabled={})", + serverNodes, minimumClusterSize, orphanGracePeriodMs, shutdownEnabled); + return; + } + + long belowForMs = elapsedMs(belowMinimumSinceNanos); + if (belowForMs >= orphanGracePeriodMs) { + actOn("Server topology has been below the minimum cluster size of " + + minimumClusterSize + " for " + belowForMs + + " ms. This node is likely orphaned from the cluster."); + } + } + + private void actOn(String reason) { + if (closed) { + return; + } + + if (!shutdownEnabled) { + // Report once per below-minimum episode so the log stays readable + if (!observeOnlyReported) { + observeOnlyReported = true; + log.error("[observe-only] Cluster observer would shut this node down: {}", reason); + } + return; + } + + if (!shutdownInitiated.compareAndSet(false, true)) { + return; + } + + log.error("Shutting down so the orchestrator can start a fresh instance: {}", reason); + + Thread watchdog = new Thread(() -> { + try { + Thread.sleep(shutdownWatchdogTimeoutMs); + } catch (InterruptedException ignored) { + return; + } + log.error("Graceful shutdown did not complete within {} ms, halting JVM", shutdownWatchdogTimeoutMs); + Runtime.getRuntime().halt(EXIT_CODE); + }, "structures-cluster-shutdown-watchdog"); + watchdog.setDaemon(true); + watchdog.start(); + + // Never shut down on an Ignite thread: closing the context stops Ignite, which + // would deadlock waiting on the very thread we are running on + Thread shutdown = new Thread(() -> { + try { + System.exit(SpringApplication.exit(applicationContext, () -> EXIT_CODE)); + } catch (Throwable t) { + log.error("Error during graceful shutdown, halting JVM", t); + Runtime.getRuntime().halt(EXIT_CODE); + } + }, "structures-cluster-shutdown"); + shutdown.setDaemon(false); + shutdown.start(); + } + + private int safeServerTopologySize() { + try { + return ignite.cluster().forServers().nodes().size(); + } catch (Exception e) { + return -1; + } + } + + // Monotonic elapsed time: wall-clock can step forward under NTP corrections or VM + // pauses and would count that step against the grace periods + private static long elapsedMs(long sinceNanos) { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - sinceNanos); + } +} diff --git a/structures-core/src/main/java/org/kinotic/structures/internal/config/VertxClusterCacheConfiguration.java b/structures-core/src/main/java/org/kinotic/structures/internal/config/VertxClusterCacheConfiguration.java new file mode 100644 index 00000000..fc81c691 --- /dev/null +++ b/structures-core/src/main/java/org/kinotic/structures/internal/config/VertxClusterCacheConfiguration.java @@ -0,0 +1,38 @@ +package org.kinotic.structures.internal.config; + +import org.apache.ignite.cache.CacheMode; +import org.apache.ignite.cache.CacheWriteSynchronizationMode; +import org.apache.ignite.configuration.CacheConfiguration; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * Cache configuration for the vertx-ignite cluster metadata caches. + *

+ * vertx-ignite stores its cluster routing metadata (__vertx.subs, __vertx.nodeInfo) in caches + * created with getOrCreateCache and no explicit configuration, which yields PARTITIONED with + * 0 backups. When a node fails, every entry primary-owned by it is destroyed with the topology + * change BEFORE vertx-ignite's cleanup listener runs. That silently loses healthy routing data + * AND breaks the cleanup election (nodeInfoMap.remove returns false on every survivor, so + * cleanSubs is skipped), leaving stale subscriptions that surface as + * "Not a member of the cluster" event bus send failures. + *

+ * Continuum collects CacheConfiguration beans into the IgniteConfiguration, and a name ending + * in '*' acts as an Ignite cache template, so this REPLICATED template applies to all + * __vertx.* caches: every node keeps a full copy, nothing is lost on node failure, and the + * cleanup election is deterministic. These maps are tiny and low-write, so the replication + * cost is negligible. + */ +@Configuration +@ConditionalOnProperty(value = "continuum.disableClustering", havingValue = "false", matchIfMissing = true) +public class VertxClusterCacheConfiguration { + + @Bean + public CacheConfiguration vertxClusterCacheTemplate() { + CacheConfiguration template = new CacheConfiguration<>("__vertx.*"); + template.setCacheMode(CacheMode.REPLICATED); + template.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC); + return template; + } +} From 36d3363299f384d758ad846b05633f7b2920aba0 Mon Sep 17 00:00:00 2001 From: Nicholas Padilla Date: Wed, 29 Jul 2026 08:41:29 -0700 Subject: [PATCH 2/7] fix: correct observer mechanism claims and config binding Review of the initial commit showed the cache-config change was based on a wrong premise and would have hurt: - drop VertxClusterCacheConfiguration entirely. Continuum 2.6.0 already contributes a "*" template (PARTITIONED, backups=1, PRIMARY_SYNC) covering the __vertx.* caches, so they were never the 0-backup caches the fix assumed. Ignite matches the longest wildcard, so the narrower __vertx.* template would have silently replaced PRIMARY_SYNC with FULL_SYNC, making every event bus handler registration block on acks from every node - and templates only apply at cache creation, so it would have been inert on rolling deploys regardless. - state the stale-routing mechanism as UNCONFIRMED in the javadoc; the observer exists to capture evidence, not to assume a cause - accept both camelCase and kebab-case property spellings: @Value has no relaxed binding, so idiomatic Boot config would have silently left the observer disabled - always log segmentation, never suppressed by the below-minimum report flag (the one-shot listener unsubscribes, so a swallowed log was lost forever), and skip shutdown under the development profile to match continuum's NoOpFailureHandler - remove dead lastObservedServerNodes state (its accessors were not ported) Co-Authored-By: Claude Fable 5 --- .../config/IgniteClusterObserver.java | 52 ++++++++++++------- .../VertxClusterCacheConfiguration.java | 38 -------------- 2 files changed, 32 insertions(+), 58 deletions(-) delete mode 100644 structures-core/src/main/java/org/kinotic/structures/internal/config/VertxClusterCacheConfiguration.java diff --git a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java index 6079ac24..7ef2d611 100644 --- a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java +++ b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java @@ -14,6 +14,7 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.core.env.Environment; import org.springframework.stereotype.Component; import javax.cache.Cache; @@ -30,12 +31,15 @@ *

    *
  • membership changes (join/left/failed) with topology context
    • *
  • segmentation events (a segmented server node can never rejoin without a restart)
    • - *
  • stale vertx routing state after a node departs: with PARTITIONED 0-backup - * __vertx.* caches the departed node's nodeInfo entry can be destroyed with its - * partition before vertx-ignite's cleanup listener runs, the cleanup election then - * no-ops on every survivor, and stale __vertx.subs entries remain - the source of - * "Not a member of the cluster" event bus send failures - * (see {@link VertxClusterCacheConfiguration} for the fix)
    • + *
  • stale vertx routing state after a node departs. vertx-ignite cleans up a + * departed node's subscriptions only on the single survivor whose + * nodeInfoMap.remove(id) returns true; if that entry is already gone, no node runs + * cleanSubs and stale __vertx.subs entries remain, which is what produces + * "Not a member of the cluster" event bus send failures. Whether that is what + * happens here is UNCONFIRMED - continuum contributes a "*" cache template + * (PARTITIONED, backups=1, PRIMARY_SYNC) covering the __vertx.* caches, so entries + * are not lost outright on a single node failure. This observer exists to capture + * the evidence rather than assume a mechanism.
    • *
  • server topology staying below structures.cluster.observer.minimumClusterSize - * the split-brain condition Ignite cannot detect by design (group splits keep a healthy * ring on each side; a restart into a partition forms a fresh singleton topology)
    • @@ -60,14 +64,15 @@ public class IgniteClusterObserver { private final Ignite ignite; private final ConfigurableApplicationContext applicationContext; + private final Environment environment; - @Value("${structures.cluster.observer.minimumClusterSize:1}") + @Value("${structures.cluster.observer.minimumClusterSize:${structures.cluster.observer.minimum-cluster-size:1}}") private int minimumClusterSize; - @Value("${structures.cluster.observer.orphanGracePeriodMs:60000}") + @Value("${structures.cluster.observer.orphanGracePeriodMs:${structures.cluster.observer.orphan-grace-period-ms:60000}}") private long orphanGracePeriodMs; - @Value("${structures.cluster.observer.startupQuorumTimeoutMs:300000}") + @Value("${structures.cluster.observer.startupQuorumTimeoutMs:${structures.cluster.observer.startup-quorum-timeout-ms:300000}}") private long startupQuorumTimeoutMs; /** @@ -75,10 +80,10 @@ public class IgniteClusterObserver { * behavior change. True = shut the process down on segmentation or sustained loss of * the minimum cluster size, so the orchestrator can start a fresh instance. */ - @Value("${structures.cluster.observer.shutdownEnabled:false}") + @Value("${structures.cluster.observer.shutdownEnabled:${structures.cluster.observer.shutdown-enabled:false}}") private boolean shutdownEnabled; - @Value("${structures.cluster.observer.shutdownWatchdogTimeoutMs:30000}") + @Value("${structures.cluster.observer.shutdownWatchdogTimeoutMs:${structures.cluster.observer.shutdown-watchdog-timeout-ms:30000}}") private long shutdownWatchdogTimeoutMs; private final AtomicBoolean shutdownInitiated = new AtomicBoolean(false); @@ -87,16 +92,18 @@ public class IgniteClusterObserver { private volatile boolean observeOnlyReported = false; private volatile long belowMinimumSinceNanos = -1; private volatile long startedAtNanos = -1; - private volatile int lastObservedServerNodes = -1; private volatile boolean topologyUnavailableLogged = false; private IgnitePredicate membershipListener; private IgnitePredicate segmentationListener; private ScheduledExecutorService scheduler; - public IgniteClusterObserver(Ignite ignite, ConfigurableApplicationContext applicationContext) { + public IgniteClusterObserver(Ignite ignite, + ConfigurableApplicationContext applicationContext, + Environment environment) { this.ignite = ignite; this.applicationContext = applicationContext; + this.environment = environment; } @PostConstruct @@ -138,11 +145,19 @@ public void start() { EventType.EVT_NODE_LEFT, EventType.EVT_NODE_FAILED); - // Segmentation is always logged; a segmented server node can never rejoin without - // a restart, so with shutdownEnabled we exit for a fresh instance + // Segmentation is ALWAYS logged (never suppressed by the below-minimum report + // flag - it is the single most important diagnostic this component produces). + // Shutdown is skipped under the development profile, matching the NoOpFailureHandler + // continuum installs there so a sleeping laptop does not kill the local server. + boolean development = environment.matchesProfiles("development"); segmentationListener = event -> { - actOn("Ignite node was segmented from the cluster. " - + "Segmented server nodes cannot rejoin without a restart."); + String reason = "Ignite node was segmented from the cluster. " + + "Segmented server nodes cannot rejoin without a restart."; + log.error("Node segmentation detected: {} (shutdownEnabled={}, development={})", + reason, shutdownEnabled, development); + if (!development) { + actOn(reason); + } return false; // one shot }; ignite.events().localListen(segmentationListener, EventType.EVT_NODE_SEGMENTED); @@ -251,7 +266,6 @@ private void checkTopology() { try { serverNodes = ignite.cluster().forServers().nodes().size(); } catch (Exception e) { - lastObservedServerNodes = 0; if (!topologyUnavailableLogged) { topologyUnavailableLogged = true; log.warn("Ignite topology is not queryable", e); @@ -260,8 +274,6 @@ private void checkTopology() { } topologyUnavailableLogged = false; - lastObservedServerNodes = serverNodes; - if (log.isDebugEnabled()) { log.debug("Topology poll: serverNodes={}, minimum={}, armed={}, belowMinimumForMs={}", serverNodes, minimumClusterSize, armed, diff --git a/structures-core/src/main/java/org/kinotic/structures/internal/config/VertxClusterCacheConfiguration.java b/structures-core/src/main/java/org/kinotic/structures/internal/config/VertxClusterCacheConfiguration.java deleted file mode 100644 index fc81c691..00000000 --- a/structures-core/src/main/java/org/kinotic/structures/internal/config/VertxClusterCacheConfiguration.java +++ /dev/null @@ -1,38 +0,0 @@ -package org.kinotic.structures.internal.config; - -import org.apache.ignite.cache.CacheMode; -import org.apache.ignite.cache.CacheWriteSynchronizationMode; -import org.apache.ignite.configuration.CacheConfiguration; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; - -/** - * Cache configuration for the vertx-ignite cluster metadata caches. - *

    - * vertx-ignite stores its cluster routing metadata (__vertx.subs, __vertx.nodeInfo) in caches - * created with getOrCreateCache and no explicit configuration, which yields PARTITIONED with - * 0 backups. When a node fails, every entry primary-owned by it is destroyed with the topology - * change BEFORE vertx-ignite's cleanup listener runs. That silently loses healthy routing data - * AND breaks the cleanup election (nodeInfoMap.remove returns false on every survivor, so - * cleanSubs is skipped), leaving stale subscriptions that surface as - * "Not a member of the cluster" event bus send failures. - *

    - * Continuum collects CacheConfiguration beans into the IgniteConfiguration, and a name ending - * in '*' acts as an Ignite cache template, so this REPLICATED template applies to all - * __vertx.* caches: every node keeps a full copy, nothing is lost on node failure, and the - * cleanup election is deterministic. These maps are tiny and low-write, so the replication - * cost is negligible. - */ -@Configuration -@ConditionalOnProperty(value = "continuum.disableClustering", havingValue = "false", matchIfMissing = true) -public class VertxClusterCacheConfiguration { - - @Bean - public CacheConfiguration vertxClusterCacheTemplate() { - CacheConfiguration template = new CacheConfiguration<>("__vertx.*"); - template.setCacheMode(CacheMode.REPLICATED); - template.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC); - return template; - } -} From 279d57ce4451475304ff3dc31d87719b55391322 Mon Sep 17 00:00:00 2001 From: Nicholas Padilla Date: Wed, 29 Jul 2026 10:36:03 -0700 Subject: [PATCH 3/7] refactor: make the cluster observer purely diagnostic Remove every action path so this can be deployed anywhere clustering is on with no behavioral risk: no shutdown, no exit codes, no readiness gating. The only configuration left is structures.cluster.observer.minimumClusterSize (default 1, watchdog off), which controls logging alone. Review fixes carried in: - the stale-route inspection now runs a node-LOCAL bounded ScanQuery on its own thread instead of a cluster-wide iteration on the shared scheduler thread, so it adds no distributed query load during a failure and can never stall topology polling - it is sampled at 5s/20s/60s after a departure, so an in-progress cleanup is distinguishable from a real leak; only the final sample warns - a failed inspection logs at WARN, and a clean result at INFO, so absence of evidence can never be read as evidence of absence - listeners are deregistered before the executors stop, so a departure during shutdown no longer throws RejectedExecutionException into Ignite's discovery thread - segmentation is logged unconditionally and never acted on; continuum's FailureHandler owns what happens to the process Co-Authored-By: Claude Fable 5 --- .../config/IgniteClusterObserver.java | 345 +++++++++--------- 1 file changed, 164 insertions(+), 181 deletions(-) diff --git a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java index 7ef2d611..d078c316 100644 --- a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java +++ b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java @@ -6,15 +6,14 @@ import org.apache.ignite.Ignite; import org.apache.ignite.IgniteCache; import org.apache.ignite.binary.BinaryObject; +import org.apache.ignite.cache.query.QueryCursor; +import org.apache.ignite.cache.query.ScanQuery; import org.apache.ignite.events.DiscoveryEvent; import org.apache.ignite.events.Event; import org.apache.ignite.events.EventType; import org.apache.ignite.lang.IgnitePredicate; import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.core.env.Environment; import org.springframework.stereotype.Component; import javax.cache.Cache; @@ -26,96 +25,81 @@ import java.util.concurrent.atomic.AtomicBoolean; /** - * Observes Ignite cluster membership from the structures side and diagnoses the - * orphaned-node / split-brain failure modes, logging evidence of each: + * Purely diagnostic observer of Ignite cluster membership. It never changes behavior: + * it does not shut anything down, does not gate readiness, and takes no action of any + * kind - it only logs what it sees, so it is safe to run everywhere clustering is on. + *

    + * What it records: *

      - *
    • membership changes (join/left/failed) with topology context
      • - *
    • segmentation events (a segmented server node can never rejoin without a restart)
      • - *
    • stale vertx routing state after a node departs. vertx-ignite cleans up a - * departed node's subscriptions only on the single survivor whose - * nodeInfoMap.remove(id) returns true; if that entry is already gone, no node runs - * cleanSubs and stale __vertx.subs entries remain, which is what produces - * "Not a member of the cluster" event bus send failures. Whether that is what - * happens here is UNCONFIRMED - continuum contributes a "*" cache template - * (PARTITIONED, backups=1, PRIMARY_SYNC) covering the __vertx.* caches, so entries - * are not lost outright on a single node failure. This observer exists to capture - * the evidence rather than assume a mechanism.
      • - *
    • server topology staying below structures.cluster.observer.minimumClusterSize - - * the split-brain condition Ignite cannot detect by design (group splits keep a healthy - * ring on each side; a restart into a partition forms a fresh singleton topology)
      • + *
    • membership changes (join/left/failed) with topology version and server count
      • + *
    • segmentation events - a segmented server node can never rejoin without a restart, + * so this is the highest value line it produces. Note continuum's non-development + * FailureHandler halts the JVM on the same thread right after listeners are notified, + * so this log may be the last thing the process writes.
      • + *
    • stale vertx routing state after a node departs. vertx-ignite cleans up a departed + * node's subscriptions only on the single survivor whose nodeInfoMap.remove(id) returns + * true; if that entry is already gone, no node runs cleanSubs and stale __vertx.subs + * entries remain, which is what produces "Not a member of the cluster" event bus send + * failures. Whether that is what happens here is UNCONFIRMED - continuum contributes a + * "*" cache template (PARTITIONED, backups=1, PRIMARY_SYNC) covering the __vertx.* + * caches, so entries are not lost outright on a single node failure. This observer + * exists to capture evidence rather than assume a mechanism. Because vertx-ignite's + * cleanup removes entries one at a time and can legitimately take a while, the check is + * sampled several times after a departure so an in-progress cleanup is distinguishable + * from a leak; only the final sample warns.
      • + *
    • server topology below structures.cluster.observer.minimumClusterSize (the only + * configuration this class has, default 1 = topology watchdog off). That is the + * split-brain condition Ignite cannot detect by design: group splits keep a healthy ring + * on each side, and a restart into a partition forms a fresh singleton topology. Set it + * to a majority of the replica count (floor(n/2)+1) to have those episodes logged.
      • *
    - * - * By default this component only observes and logs. Set - * structures.cluster.observer.shutdownEnabled=true to also shut the process down (non-zero - * exit, so the orchestrator starts a fresh instance) when this node is segmented or stays - * below the minimum cluster size beyond the grace period. minimumClusterSize should be a - * majority of the replica count (floor(n/2)+1). This is a diagnostic port of continuum 3.x's - * IgniteOrphanedNodeGuard for use while structures is on continuum 2.6.x. + * Inspections run on their own thread, are bounded, and read only node-local cache + * partitions, so they add no cluster-wide query load during a failure. */ @Slf4j @Component @ConditionalOnProperty(value = "continuum.disableClustering", havingValue = "false", matchIfMissing = true) public class IgniteClusterObserver { - private static final int EXIT_CODE = 1; private static final long TOPOLOGY_POLL_MS = 10_000L; - private static final long STALE_ROUTE_CHECK_DELAY_MS = 15_000L; + private static final long ORPHAN_REPORT_AFTER_MS = 60_000L; private static final int MAX_STALE_ADDRESSES_LOGGED = 10; + private static final int MAX_ENTRIES_SCANNED = 50_000; + /** Sampled repeatedly so a slow cleanup is not reported as a leak; only the last warns */ + private static final long[] STALE_ROUTE_SAMPLE_DELAYS_MS = {5_000L, 20_000L, 60_000L}; private final Ignite ignite; - private final ConfigurableApplicationContext applicationContext; - private final Environment environment; - - @Value("${structures.cluster.observer.minimumClusterSize:${structures.cluster.observer.minimum-cluster-size:1}}") - private int minimumClusterSize; - - @Value("${structures.cluster.observer.orphanGracePeriodMs:${structures.cluster.observer.orphan-grace-period-ms:60000}}") - private long orphanGracePeriodMs; - - @Value("${structures.cluster.observer.startupQuorumTimeoutMs:${structures.cluster.observer.startup-quorum-timeout-ms:300000}}") - private long startupQuorumTimeoutMs; /** - * Master switch for taking action. False (default) = observe and log only, no outward - * behavior change. True = shut the process down on segmentation or sustained loss of - * the minimum cluster size, so the orchestrator can start a fresh instance. + * Minimum number of server nodes expected in the topology. When above 1, episodes + * below it are logged (with duration). Purely informational - nothing is ever shut + * down. Set it to a majority of the replica count, e.g. 2 for 3 replicas. */ - @Value("${structures.cluster.observer.shutdownEnabled:${structures.cluster.observer.shutdown-enabled:false}}") - private boolean shutdownEnabled; - - @Value("${structures.cluster.observer.shutdownWatchdogTimeoutMs:${structures.cluster.observer.shutdown-watchdog-timeout-ms:30000}}") - private long shutdownWatchdogTimeoutMs; + @Value("${structures.cluster.observer.minimumClusterSize:${structures.cluster.observer.minimum-cluster-size:1}}") + private int minimumClusterSize; - private final AtomicBoolean shutdownInitiated = new AtomicBoolean(false); private volatile boolean closed = false; private volatile boolean armed = false; - private volatile boolean observeOnlyReported = false; + private volatile boolean belowMinimumReported = false; private volatile long belowMinimumSinceNanos = -1; - private volatile long startedAtNanos = -1; - private volatile boolean topologyUnavailableLogged = false; + private final AtomicBoolean inspectionInProgress = new AtomicBoolean(false); private IgnitePredicate membershipListener; private IgnitePredicate segmentationListener; private ScheduledExecutorService scheduler; + private ScheduledExecutorService inspector; - public IgniteClusterObserver(Ignite ignite, - ConfigurableApplicationContext applicationContext, - Environment environment) { + public IgniteClusterObserver(Ignite ignite) { this.ignite = ignite; - this.applicationContext = applicationContext; - this.environment = environment; } @PostConstruct public void start() { - scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { - Thread thread = new Thread(runnable, "structures-cluster-observer"); - thread.setDaemon(true); - return thread; - }); + scheduler = newDaemonScheduler("structures-cluster-observer"); + // Inspections get their own thread: a cache read during a partition can block for + // a long time, and it must never stall topology polling + inspector = newDaemonScheduler("structures-cluster-inspector"); - // Membership diagnostics: log joins/departures with topology context and, after a - // departure, verify the vertx routing caches were actually cleaned up membershipListener = event -> { DiscoveryEvent discoveryEvent = (DiscoveryEvent) event; String eventNodeId = discoveryEvent.eventNode().id().toString(); @@ -134,9 +118,7 @@ public void start() { default -> { /* not registered for others */ } } if (event.type() == EventType.EVT_NODE_LEFT || event.type() == EventType.EVT_NODE_FAILED) { - // Check after vertx-ignite's cleanup listener has had ample time to run - scheduler.schedule(() -> reportStaleRoutingState(eventNodeId), - STALE_ROUTE_CHECK_DELAY_MS, TimeUnit.MILLISECONDS); + scheduleStaleRouteSamples(eventNodeId); } return true; }; @@ -145,44 +127,36 @@ public void start() { EventType.EVT_NODE_LEFT, EventType.EVT_NODE_FAILED); - // Segmentation is ALWAYS logged (never suppressed by the below-minimum report - // flag - it is the single most important diagnostic this component produces). - // Shutdown is skipped under the development profile, matching the NoOpFailureHandler - // continuum installs there so a sleeping laptop does not kill the local server. - boolean development = environment.matchesProfiles("development"); + // Logged, never acted on. Continuum's FailureHandler decides what happens to the + // process; this line is the evidence that segmentation is what happened. segmentationListener = event -> { - String reason = "Ignite node was segmented from the cluster. " - + "Segmented server nodes cannot rejoin without a restart."; - log.error("Node segmentation detected: {} (shutdownEnabled={}, development={})", - reason, shutdownEnabled, development); - if (!development) { - actOn(reason); - } + log.error("Node segmentation detected: this Ignite node was segmented from the cluster. " + + "Segmented server nodes cannot rejoin without a restart (serverNodes={})", + safeServerTopologySize()); return false; // one shot }; ignite.events().localListen(segmentationListener, EventType.EVT_NODE_SEGMENTED); - log.info("Ignite cluster observer started: minimumClusterSize={}, orphanGracePeriodMs={}, " - + "startupQuorumTimeoutMs={}, shutdownEnabled={}", - minimumClusterSize, orphanGracePeriodMs, startupQuorumTimeoutMs, shutdownEnabled); - if (minimumClusterSize > 1) { - startedAtNanos = System.nanoTime(); scheduler.scheduleWithFixedDelay(this::checkTopologySafely, 0, TOPOLOGY_POLL_MS, TimeUnit.MILLISECONDS); + log.info("Ignite cluster observer started (diagnostic only): minimumClusterSize={}", + minimumClusterSize); + } else { + log.info("Ignite cluster observer started (diagnostic only): membership and routing " + + "diagnostics active, topology watchdog off " + + "(set structures.cluster.observer.minimumClusterSize above 1 to enable it)"); } } @PreDestroy public void stop() { closed = true; - // A normal context shutdown must never be escalated to a JVM halt by an in-flight poll - shutdownInitiated.set(true); - if (scheduler != null) { - scheduler.shutdownNow(); - } + // Deregister listeners BEFORE stopping the executors, otherwise a departure arriving + // in between would schedule onto a terminated executor and throw + // RejectedExecutionException back into Ignite's discovery notification thread if (membershipListener != null) { try { ignite.events().stopLocalListen(membershipListener, @@ -200,48 +174,105 @@ public void stop() { log.debug("Could not remove segmentation listener during shutdown", e); } } + if (scheduler != null) { + scheduler.shutdownNow(); + } + if (inspector != null) { + inspector.shutdownNow(); + } + } + + private void scheduleStaleRouteSamples(String departedNodeId) { + if (closed) { + return; + } + for (int i = 0; i < STALE_ROUTE_SAMPLE_DELAYS_MS.length; i++) { + boolean finalSample = i == STALE_ROUTE_SAMPLE_DELAYS_MS.length - 1; + long delay = STALE_ROUTE_SAMPLE_DELAYS_MS[i]; + try { + inspector.schedule(() -> reportStaleRoutingState(departedNodeId, delay, finalSample), + delay, TimeUnit.MILLISECONDS); + } catch (Exception e) { + // Executor already stopping; nothing to diagnose + log.debug("Could not schedule routing state inspection", e); + return; + } + } } /** - * Inspect the vertx-ignite routing caches after a node departed and log any stale - * state left behind - the direct evidence of the cleanup-election failure that - * produces "Not a member of the cluster" event bus send errors. + * Inspect this node's LOCAL partitions of the vertx routing caches for entries that + * still reference a departed node. Local-only by design: it costs nothing beyond a + * node-local iteration, adds no distributed query load while the cluster is already + * rebalancing, and every surviving node logs its own view, which together cover the + * cluster. */ - private void reportStaleRoutingState(String departedNodeId) { + private void reportStaleRoutingState(String departedNodeId, long afterMs, boolean finalSample) { if (closed) { return; } + // Never let overlapping departures stack up inspections + if (!inspectionInProgress.compareAndSet(false, true)) { + log.debug("Skipping routing state inspection for {}, another inspection is running", + departedNodeId); + return; + } try { IgniteCache nodeInfoCache = ignite.cache("__vertx.nodeInfo"); boolean nodeInfoPresent = nodeInfoCache != null && nodeInfoCache.containsKey(departedNodeId); int staleSubs = 0; + int scanned = 0; + boolean truncated = false; List staleAddresses = new ArrayList<>(); - // Read the subs cache in binary form to avoid a compile-time dependency on - // vertx-ignite's IgniteRegistrationInfo (field names match its writeBinary) IgniteCache subsCache = ignite.cache("__vertx.subs"); if (subsCache != null) { - for (Cache.Entry entry : subsCache.withKeepBinary()) { - if (entry.getKey() instanceof BinaryObject key - && departedNodeId.equals(key.field("nodeId"))) { - staleSubs++; - if (staleAddresses.size() < MAX_STALE_ADDRESSES_LOGGED) { - staleAddresses.add(key.field("address")); + // Local scan, binary form: no cluster-wide query, and no compile-time + // dependency on vertx-ignite's IgniteRegistrationInfo (the binary field + // names match its writeBinary implementation) + ScanQuery query = new ScanQuery<>(); + query.setLocal(true); + try (QueryCursor> cursor + = subsCache.withKeepBinary().query(query)) { + for (Cache.Entry entry : cursor) { + if (++scanned > MAX_ENTRIES_SCANNED) { + truncated = true; + break; + } + if (entry.getKey() instanceof BinaryObject key + && departedNodeId.equals(key.field("nodeId"))) { + staleSubs++; + if (staleAddresses.size() < MAX_STALE_ADDRESSES_LOGGED) { + staleAddresses.add(key.field("address")); + } } } } } - if (nodeInfoPresent || staleSubs > 0) { - log.warn("Stale routing state remains for departed node {}: nodeInfoStillPresent={}, " - + "staleSubscriptionEntries={}, sampleAddresses={}. Event bus sends to these " - + "addresses can fail with 'Not a member of the cluster' until handlers re-register.", - departedNodeId, nodeInfoPresent, staleSubs, staleAddresses); + if (!nodeInfoPresent && staleSubs == 0) { + log.info("Routing caches are clean (local view) {} ms after departure of node {}: " + + "scannedLocalSubs={}", afterMs, departedNodeId, scanned); + } else if (finalSample) { + log.warn("Stale routing state remains {} ms after departure of node {}: " + + "nodeInfoStillPresent={}, staleLocalSubscriptionEntries={}, " + + "sampleAddresses={}, scannedLocalSubs={}, truncated={}. Event bus sends to " + + "these addresses can fail with 'Not a member of the cluster' until the " + + "handlers re-register.", + afterMs, departedNodeId, nodeInfoPresent, staleSubs, staleAddresses, + scanned, truncated); } else { - log.debug("Routing caches are clean after departure of node {}", departedNodeId); + log.info("Routing cleanup still in progress {} ms after departure of node {}: " + + "nodeInfoStillPresent={}, staleLocalSubscriptionEntries={}", + afterMs, departedNodeId, nodeInfoPresent, staleSubs); } } catch (Exception e) { - log.debug("Could not inspect routing caches after departure of node {}", departedNodeId, e); + // WARN, not DEBUG: a failed inspection must never be mistaken for a clean result + log.warn("Could not inspect routing caches {} ms after departure of node {}; " + + "no conclusion can be drawn about stale routing state", + afterMs, departedNodeId, e); + } finally { + inspectionInProgress.set(false); } } @@ -258,21 +289,15 @@ private void checkTopologySafely() { } private void checkTopology() { - if (closed || shutdownInitiated.get()) { + if (closed) { return; } - int serverNodes; - try { - serverNodes = ignite.cluster().forServers().nodes().size(); - } catch (Exception e) { - if (!topologyUnavailableLogged) { - topologyUnavailableLogged = true; - log.warn("Ignite topology is not queryable", e); - } + int serverNodes = safeServerTopologySize(); + if (serverNodes < 0) { + log.warn("Ignite topology is not queryable"); return; } - topologyUnavailableLogged = false; if (log.isDebugEnabled()) { log.debug("Topology poll: serverNodes={}, minimum={}, armed={}, belowMinimumForMs={}", @@ -285,87 +310,45 @@ private void checkTopology() { armed = true; log.info("Cluster observer armed: server topology reached {} nodes", serverNodes); } else if (belowMinimumSinceNanos != -1) { - log.info("Server topology recovered to {} nodes after {} ms below minimum", - serverNodes, elapsedMs(belowMinimumSinceNanos)); + log.info("Server topology recovered to {} nodes after {} ms below the minimum of {}", + serverNodes, elapsedMs(belowMinimumSinceNanos), minimumClusterSize); } belowMinimumSinceNanos = -1; - observeOnlyReported = false; + belowMinimumReported = false; return; } - // Below the minimum. Normal startup never trips this (arm-after-join), but a node - // that NEVER reaches the minimum likely started into an ongoing partition; Ignite - // topologies never merge once formed. + // Below the minimum. Startup (nodes joining one by one) is not interesting, so only + // report once the topology has actually reached the minimum at least once. if (!armed) { - if (startupQuorumTimeoutMs > 0 && elapsedMs(startedAtNanos) >= startupQuorumTimeoutMs) { - actOn("Server topology never reached the minimum cluster size of " - + minimumClusterSize + " within " + startupQuorumTimeoutMs - + " ms of startup. This node likely started into an ongoing partition " - + "and would run split-brained."); - } return; } if (belowMinimumSinceNanos == -1) { belowMinimumSinceNanos = System.nanoTime(); - log.warn("Server topology dropped to {} nodes (minimum {}). Action in {} ms unless it recovers " - + "(shutdownEnabled={})", - serverNodes, minimumClusterSize, orphanGracePeriodMs, shutdownEnabled); + log.warn("Server topology dropped to {} nodes, below the minimum of {}. This node may be " + + "orphaned from the cluster; watching", + serverNodes, minimumClusterSize); return; } long belowForMs = elapsedMs(belowMinimumSinceNanos); - if (belowForMs >= orphanGracePeriodMs) { - actOn("Server topology has been below the minimum cluster size of " - + minimumClusterSize + " for " + belowForMs - + " ms. This node is likely orphaned from the cluster."); + // One escalation per episode so a long orphan does not flood the log + if (!belowMinimumReported && belowForMs >= ORPHAN_REPORT_AFTER_MS) { + belowMinimumReported = true; + log.error("Server topology has been at {} nodes, below the minimum of {}, for {} ms. " + + "This node is very likely orphaned or split-brained and would need a restart " + + "to rejoin the cluster", + serverNodes, minimumClusterSize, belowForMs); } } - private void actOn(String reason) { - if (closed) { - return; - } - - if (!shutdownEnabled) { - // Report once per below-minimum episode so the log stays readable - if (!observeOnlyReported) { - observeOnlyReported = true; - log.error("[observe-only] Cluster observer would shut this node down: {}", reason); - } - return; - } - - if (!shutdownInitiated.compareAndSet(false, true)) { - return; - } - - log.error("Shutting down so the orchestrator can start a fresh instance: {}", reason); - - Thread watchdog = new Thread(() -> { - try { - Thread.sleep(shutdownWatchdogTimeoutMs); - } catch (InterruptedException ignored) { - return; - } - log.error("Graceful shutdown did not complete within {} ms, halting JVM", shutdownWatchdogTimeoutMs); - Runtime.getRuntime().halt(EXIT_CODE); - }, "structures-cluster-shutdown-watchdog"); - watchdog.setDaemon(true); - watchdog.start(); - - // Never shut down on an Ignite thread: closing the context stops Ignite, which - // would deadlock waiting on the very thread we are running on - Thread shutdown = new Thread(() -> { - try { - System.exit(SpringApplication.exit(applicationContext, () -> EXIT_CODE)); - } catch (Throwable t) { - log.error("Error during graceful shutdown, halting JVM", t); - Runtime.getRuntime().halt(EXIT_CODE); - } - }, "structures-cluster-shutdown"); - shutdown.setDaemon(false); - shutdown.start(); + private ScheduledExecutorService newDaemonScheduler(String threadName) { + return Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, threadName); + thread.setDaemon(true); + return thread; + }); } private int safeServerTopologySize() { @@ -377,7 +360,7 @@ private int safeServerTopologySize() { } // Monotonic elapsed time: wall-clock can step forward under NTP corrections or VM - // pauses and would count that step against the grace periods + // pauses and would misreport how long a node has been below the minimum private static long elapsedMs(long sinceNanos) { return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - sinceNanos); } From 8525311204d50262024c277188b5f9dd7def01ab Mon Sep 17 00:00:00 2001 From: Nicholas Padilla Date: Wed, 29 Jul 2026 17:44:33 -0700 Subject: [PATCH 4/7] fix: make observer evidence trustworthy and plumb its config Review found the diagnostics could emit false all-clears, which is worse than no diagnostic. Every finding addressed: - never log an unqualified "clean" result. Missing vertx caches, truncated scans and unreadable keys now produce an explicit "Inconclusive routing inspection ... NOT an all-clear" warning, and every line states the coverage it achieved - read primary AND backup partitions via localEntries instead of a primary-only local ScanQuery: a departure triggers rebalancing, and a backup-held entry is exactly what a primary-only scan would miss - log measured elapsed time per sample instead of the nominal schedule, so a sample that runs late while the cluster is unhealthy does not report a timestamp it did not observe - restore the never-reached-minimum diagnostic (log only): a node that restarts into an ongoing partition forms its own singleton topology that Ignite never merges, and previously produced no output at all - hand membership logging off to the observer thread; Ignite notifies listeners inline on the monitored discovery worker, where a blocked log write could trip SYSTEM_WORKER_BLOCKED and the failure handler - throttle the "topology is not queryable" warning (it could repeat every 10s forever), report the underlying exception instead of a bare serverNodes=-1 sentinel, fix the scan-cap off-by-one, and log shutdown-interrupted inspections at DEBUG so ordinary deploys stay quiet - plumb structures.cluster.observer.minimumClusterSize through the Helm configmap and values so the topology reporting is reachable in k8s Co-Authored-By: Claude Fable 5 --- .../structures-server-config-map.yaml | 4 + helm/structures/values.yaml | 6 + .../config/IgniteClusterObserver.java | 308 +++++++++++------- 3 files changed, 209 insertions(+), 109 deletions(-) diff --git a/helm/structures/templates/structures-server-config-map.yaml b/helm/structures/templates/structures-server-config-map.yaml index cf610375..a0e4ccf7 100644 --- a/helm/structures/templates/structures-server-config-map.yaml +++ b/helm/structures/templates/structures-server-config-map.yaml @@ -50,6 +50,10 @@ data: # Reference: structures-core/src/test/resources/docker-compose/cluster-test-compose.yml CONTINUUM_CLUSTER_DISABLE_CLUSTERING: "{{ .Values.continuum.disableClustering | default "false" }}" + # Diagnostic cluster observer (logging only, never shuts anything down). Above 1 it also + # logs episodes where the server topology is below this count, which is the split-brain + # condition Ignite cannot detect. Set to a majority of replicaCount, e.g. 2 for 3. + STRUCTURES_CLUSTER_OBSERVER_MINIMUMCLUSTERSIZE: "{{ .Values.continuum.cluster.observerMinimumClusterSize | default "1" }}" CONTINUUM_CLUSTER_DISCOVERY_TYPE: "{{ .Values.continuum.cluster.discoveryType | default "LOCAL" | upper }}" CONTINUUM_CLUSTER_JOIN_TIMEOUT_MS: "{{ .Values.continuum.cluster.joinTimeoutMs | default "0" }}" CONTINUUM_CLUSTER_DISCOVERY_PORT: "{{ .Values.continuum.cluster.discoveryPort | default "47500" }}" diff --git a/helm/structures/values.yaml b/helm/structures/values.yaml index 24133441..4241156c 100644 --- a/helm/structures/values.yaml +++ b/helm/structures/values.yaml @@ -83,6 +83,12 @@ continuum: kubernetesIncludeNotReadyAddresses: false kubernetesMasterUrl: "" kubernetesAccountToken: "" + # Diagnostic cluster observer: logging only, it never shuts anything down. + # 1 (default) logs membership changes, segmentation and post-departure vertx routing + # state. Above 1 it additionally logs episodes where the server topology is below this + # count - the split-brain condition Ignite cannot detect on its own. Set it to a + # majority of replicaCount (floor(n/2)+1), e.g. 2 for 3 replicas. + observerMinimumClusterSize: 1 # TODO: adjust for this later # OIDC security service configuration diff --git a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java index d078c316..4a4f0909 100644 --- a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java +++ b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java @@ -6,8 +6,7 @@ import org.apache.ignite.Ignite; import org.apache.ignite.IgniteCache; import org.apache.ignite.binary.BinaryObject; -import org.apache.ignite.cache.query.QueryCursor; -import org.apache.ignite.cache.query.ScanQuery; +import org.apache.ignite.cache.CachePeekMode; import org.apache.ignite.events.DiscoveryEvent; import org.apache.ignite.events.Event; import org.apache.ignite.events.EventType; @@ -20,22 +19,25 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; /** * Purely diagnostic observer of Ignite cluster membership. It never changes behavior: * it does not shut anything down, does not gate readiness, and takes no action of any * kind - it only logs what it sees, so it is safe to run everywhere clustering is on. *

    + * All work is handed to its own daemon threads; nothing but the handoff runs on Ignite's + * discovery worker, because stalling that worker would trip Ignite's own failure handler. + *

    * What it records: *

      *
    • membership changes (join/left/failed) with topology version and server count
      • *
    • segmentation events - a segmented server node can never rejoin without a restart, * so this is the highest value line it produces. Note continuum's non-development - * FailureHandler halts the JVM on the same thread right after listeners are notified, - * so this log may be the last thing the process writes.
      • + * FailureHandler halts the JVM shortly after listeners are notified, so this log may be + * the last thing the process writes. *
    • stale vertx routing state after a node departs. vertx-ignite cleans up a departed * node's subscriptions only on the single survivor whose nodeInfoMap.remove(id) returns * true; if that entry is already gone, no node runs cleanSubs and stale __vertx.subs @@ -43,18 +45,18 @@ * failures. Whether that is what happens here is UNCONFIRMED - continuum contributes a * "*" cache template (PARTITIONED, backups=1, PRIMARY_SYNC) covering the __vertx.* * caches, so entries are not lost outright on a single node failure. This observer - * exists to capture evidence rather than assume a mechanism. Because vertx-ignite's - * cleanup removes entries one at a time and can legitimately take a while, the check is - * sampled several times after a departure so an in-progress cleanup is distinguishable - * from a leak; only the final sample warns.
      • + * exists to capture evidence rather than assume a mechanism. Cleanup removes entries one + * at a time and can legitimately take a while, so the check is sampled several times + * after a departure; only the final sample warns. Every result states the coverage it + * achieved, and a scan that could not cover the data never reports "clean" - an + * inconclusive result must never be mistaken for an all-clear. *
    • server topology below structures.cluster.observer.minimumClusterSize (the only - * configuration this class has, default 1 = topology watchdog off). That is the + * configuration this class has, default 1 = topology reporting off). That is the * split-brain condition Ignite cannot detect by design: group splits keep a healthy ring - * on each side, and a restart into a partition forms a fresh singleton topology. Set it - * to a majority of the replica count (floor(n/2)+1) to have those episodes logged.
      • + * on each side, and a restart into a partition forms a fresh singleton topology that + * Ignite never merges. Set it to a majority of the replica count (floor(n/2)+1) to have + * those episodes logged, including a node that never reaches the minimum at all. *
    - * Inspections run on their own thread, are bounded, and read only node-local cache - * partitions, so they add no cluster-wide query load during a failure. */ @Slf4j @Component @@ -62,7 +64,9 @@ public class IgniteClusterObserver { private static final long TOPOLOGY_POLL_MS = 10_000L; - private static final long ORPHAN_REPORT_AFTER_MS = 60_000L; + private static final long REPORT_AFTER_BELOW_MINIMUM_MS = 60_000L; + private static final long REPORT_NEVER_REACHED_MINIMUM_MS = 300_000L; + private static final long UNQUERYABLE_REPORT_INTERVAL_MS = 600_000L; private static final int MAX_STALE_ADDRESSES_LOGGED = 10; private static final int MAX_ENTRIES_SCANNED = 50_000; /** Sampled repeatedly so a slow cleanup is not reported as a leak; only the last warns */ @@ -81,8 +85,10 @@ public class IgniteClusterObserver { private volatile boolean closed = false; private volatile boolean armed = false; private volatile boolean belowMinimumReported = false; + private volatile boolean neverReachedMinimumReported = false; private volatile long belowMinimumSinceNanos = -1; - private final AtomicBoolean inspectionInProgress = new AtomicBoolean(false); + private volatile long startedAtNanos = -1; + private volatile long lastUnqueryableReportNanos = -1; private IgnitePredicate membershipListener; private IgnitePredicate segmentationListener; @@ -95,29 +101,22 @@ public IgniteClusterObserver(Ignite ignite) { @PostConstruct public void start() { + startedAtNanos = System.nanoTime(); scheduler = newDaemonScheduler("structures-cluster-observer"); // Inspections get their own thread: a cache read during a partition can block for // a long time, and it must never stall topology polling inspector = newDaemonScheduler("structures-cluster-inspector"); + // The listener captures the event and hands off immediately. Ignite notifies local + // listeners inline on the discovery worker, which is a monitored critical worker: + // logging (and any cluster call) there risks tripping SYSTEM_WORKER_BLOCKED. membershipListener = event -> { DiscoveryEvent discoveryEvent = (DiscoveryEvent) event; String eventNodeId = discoveryEvent.eventNode().id().toString(); long topologyVersion = discoveryEvent.topologyVersion(); - int serverNodes = safeServerTopologySize(); - switch (event.type()) { - case EventType.EVT_NODE_JOINED -> - log.info("Cluster node joined: {} (topologyVersion={}, serverNodes={})", - eventNodeId, topologyVersion, serverNodes); - case EventType.EVT_NODE_LEFT -> - log.info("Cluster node left: {} (topologyVersion={}, serverNodes={})", - eventNodeId, topologyVersion, serverNodes); - case EventType.EVT_NODE_FAILED -> - log.warn("Cluster node FAILED: {} (topologyVersion={}, serverNodes={})", - eventNodeId, topologyVersion, serverNodes); - default -> { /* not registered for others */ } - } - if (event.type() == EventType.EVT_NODE_LEFT || event.type() == EventType.EVT_NODE_FAILED) { + int eventType = event.type(); + submit(scheduler, () -> logMembershipChange(eventType, eventNodeId, topologyVersion)); + if (eventType == EventType.EVT_NODE_LEFT || eventType == EventType.EVT_NODE_FAILED) { scheduleStaleRouteSamples(eventNodeId); } return true; @@ -128,11 +127,12 @@ public void start() { EventType.EVT_NODE_FAILED); // Logged, never acted on. Continuum's FailureHandler decides what happens to the - // process; this line is the evidence that segmentation is what happened. + // process. Logged inline rather than handed off: the JVM is likely to be halted + // moments from now, and an off-thread log would never be written. segmentationListener = event -> { log.error("Node segmentation detected: this Ignite node was segmented from the cluster. " - + "Segmented server nodes cannot rejoin without a restart (serverNodes={})", - safeServerTopologySize()); + + "Segmented server nodes cannot rejoin without a restart. {}", + describeServerTopology()); return false; // one shot }; ignite.events().localListen(segmentationListener, EventType.EVT_NODE_SEGMENTED); @@ -146,7 +146,7 @@ public void start() { minimumClusterSize); } else { log.info("Ignite cluster observer started (diagnostic only): membership and routing " - + "diagnostics active, topology watchdog off " + + "diagnostics active, topology reporting off " + "(set structures.cluster.observer.minimumClusterSize above 1 to enable it)"); } } @@ -182,97 +182,135 @@ public void stop() { } } + private void logMembershipChange(int eventType, String eventNodeId, long topologyVersion) { + switch (eventType) { + case EventType.EVT_NODE_JOINED -> + log.info("Cluster node joined: {} (topologyVersion={}, {})", + eventNodeId, topologyVersion, describeServerTopology()); + case EventType.EVT_NODE_LEFT -> + log.info("Cluster node left: {} (topologyVersion={}, {})", + eventNodeId, topologyVersion, describeServerTopology()); + case EventType.EVT_NODE_FAILED -> + log.warn("Cluster node FAILED: {} (topologyVersion={}, {})", + eventNodeId, topologyVersion, describeServerTopology()); + default -> { /* not registered for others */ } + } + } + private void scheduleStaleRouteSamples(String departedNodeId) { if (closed) { return; } + long departedAtNanos = System.nanoTime(); for (int i = 0; i < STALE_ROUTE_SAMPLE_DELAYS_MS.length; i++) { boolean finalSample = i == STALE_ROUTE_SAMPLE_DELAYS_MS.length - 1; long delay = STALE_ROUTE_SAMPLE_DELAYS_MS[i]; - try { - inspector.schedule(() -> reportStaleRoutingState(departedNodeId, delay, finalSample), - delay, TimeUnit.MILLISECONDS); - } catch (Exception e) { - // Executor already stopping; nothing to diagnose - log.debug("Could not schedule routing state inspection", e); - return; - } + submit(inspector, + () -> reportStaleRoutingState(departedNodeId, departedAtNanos, finalSample), + delay); } } /** - * Inspect this node's LOCAL partitions of the vertx routing caches for entries that - * still reference a departed node. Local-only by design: it costs nothing beyond a - * node-local iteration, adds no distributed query load while the cluster is already - * rebalancing, and every surviving node logs its own view, which together cover the - * cluster. + * Inspect this node's local partitions of the vertx routing caches for entries that + * still reference a departed node. Local by design: it costs nothing beyond a + * node-local iteration and adds no distributed query load while the cluster is already + * rebalancing. Primary AND backup partitions are read, because a departure triggers + * rebalancing and an entry this node holds only as a backup is exactly the kind a + * primary-only scan would miss. Every log line states the coverage achieved so an + * inconclusive scan can never be read as an all-clear. */ - private void reportStaleRoutingState(String departedNodeId, long afterMs, boolean finalSample) { + private void reportStaleRoutingState(String departedNodeId, long departedAtNanos, boolean finalSample) { if (closed) { return; } - // Never let overlapping departures stack up inspections - if (!inspectionInProgress.compareAndSet(false, true)) { - log.debug("Skipping routing state inspection for {}, another inspection is running", - departedNodeId); - return; - } + // Real elapsed time, not the nominal schedule: a sample can run late when the + // cluster is unhealthy, which is exactly when the timestamp matters + long afterMs = elapsedMs(departedAtNanos); try { IgniteCache nodeInfoCache = ignite.cache("__vertx.nodeInfo"); - boolean nodeInfoPresent = nodeInfoCache != null && nodeInfoCache.containsKey(departedNodeId); + IgniteCache subsCache = ignite.cache("__vertx.subs"); + + if (nodeInfoCache == null || subsCache == null) { + log.warn("Inconclusive routing inspection {} ms after departure of node {}: " + + "vertx caches are not available on this node " + + "(nodeInfoCachePresent={}, subsCachePresent={}). No conclusion can be " + + "drawn about stale routing state.", + afterMs, departedNodeId, nodeInfoCache != null, subsCache != null); + return; + } + + boolean nodeInfoPresent = nodeInfoCache.containsKey(departedNodeId); int staleSubs = 0; int scanned = 0; + int nonBinaryKeys = 0; boolean truncated = false; List staleAddresses = new ArrayList<>(); - IgniteCache subsCache = ignite.cache("__vertx.subs"); - if (subsCache != null) { - // Local scan, binary form: no cluster-wide query, and no compile-time - // dependency on vertx-ignite's IgniteRegistrationInfo (the binary field - // names match its writeBinary implementation) - ScanQuery query = new ScanQuery<>(); - query.setLocal(true); - try (QueryCursor> cursor - = subsCache.withKeepBinary().query(query)) { - for (Cache.Entry entry : cursor) { - if (++scanned > MAX_ENTRIES_SCANNED) { - truncated = true; - break; - } - if (entry.getKey() instanceof BinaryObject key - && departedNodeId.equals(key.field("nodeId"))) { - staleSubs++; - if (staleAddresses.size() < MAX_STALE_ADDRESSES_LOGGED) { - staleAddresses.add(key.field("address")); - } + // localEntries with explicit peek modes rather than a ScanQuery: it is + // node-local (no distributed query while the cluster is rebalancing) AND it + // includes backup partitions. A plain scan query only covers partitions this + // node is primary for, which is exactly where a departure's stale entries can + // hide during rebalancing. Binary form avoids a compile-time dependency on + // vertx-ignite's IgniteRegistrationInfo (the field names match its writeBinary). + for (Cache.Entry entry : subsCache.withKeepBinary() + .localEntries(CachePeekMode.PRIMARY, + CachePeekMode.BACKUP)) { + if (scanned >= MAX_ENTRIES_SCANNED) { + truncated = true; + break; + } + scanned++; + if (entry.getKey() instanceof BinaryObject key) { + if (departedNodeId.equals(key.field("nodeId"))) { + staleSubs++; + if (staleAddresses.size() < MAX_STALE_ADDRESSES_LOGGED) { + staleAddresses.add(key.field("address")); } } + } else { + nonBinaryKeys++; } } - if (!nodeInfoPresent && staleSubs == 0) { - log.info("Routing caches are clean (local view) {} ms after departure of node {}: " - + "scannedLocalSubs={}", afterMs, departedNodeId, scanned); - } else if (finalSample) { - log.warn("Stale routing state remains {} ms after departure of node {}: " - + "nodeInfoStillPresent={}, staleLocalSubscriptionEntries={}, " - + "sampleAddresses={}, scannedLocalSubs={}, truncated={}. Event bus sends to " - + "these addresses can fail with 'Not a member of the cluster' until the " - + "handlers re-register.", - afterMs, departedNodeId, nodeInfoPresent, staleSubs, staleAddresses, - scanned, truncated); + // An unexamined remainder or unreadable keys mean the scan cannot support an + // all-clear, so say so rather than implying the caches are clean + boolean conclusive = !truncated && nonBinaryKeys == 0; + String coverage = String.format( + "localEntriesScanned=%d (primary+backup), truncated=%b, unreadableKeys=%d", + scanned, truncated, nonBinaryKeys); + + if (nodeInfoPresent || staleSubs > 0) { + if (finalSample) { + log.warn("Stale routing state remains {} ms after departure of node {}: " + + "nodeInfoStillPresent={}, staleLocalSubscriptionEntries={}, " + + "sampleAddresses={}, {}. Event bus sends to these addresses can fail " + + "with 'Not a member of the cluster' until the handlers re-register.", + afterMs, departedNodeId, nodeInfoPresent, staleSubs, staleAddresses, coverage); + } else { + log.info("Routing cleanup still in progress {} ms after departure of node {}: " + + "nodeInfoStillPresent={}, staleLocalSubscriptionEntries={}, {}", + afterMs, departedNodeId, nodeInfoPresent, staleSubs, coverage); + } + } else if (conclusive) { + log.info("Routing caches are clean (local view) {} ms after departure of node {}: {}", + afterMs, departedNodeId, coverage); } else { - log.info("Routing cleanup still in progress {} ms after departure of node {}: " - + "nodeInfoStillPresent={}, staleLocalSubscriptionEntries={}", - afterMs, departedNodeId, nodeInfoPresent, staleSubs); + log.warn("Inconclusive routing inspection {} ms after departure of node {}: no stale " + + "entries seen, but the scan did not cover all local entries so this is NOT " + + "an all-clear ({})", + afterMs, departedNodeId, coverage); } } catch (Exception e) { - // WARN, not DEBUG: a failed inspection must never be mistaken for a clean result - log.warn("Could not inspect routing caches {} ms after departure of node {}; " - + "no conclusion can be drawn about stale routing state", - afterMs, departedNodeId, e); - } finally { - inspectionInProgress.set(false); + if (closed || Thread.currentThread().isInterrupted()) { + // Ordinary shutdown interrupted the read; not a cluster problem + log.debug("Routing inspection for node {} aborted during shutdown", departedNodeId, e); + } else { + // WARN, not DEBUG: a failed inspection must never be mistaken for a clean result + log.warn("Could not inspect routing caches {} ms after departure of node {}; " + + "no conclusion can be drawn about stale routing state", + afterMs, departedNodeId, e); + } } } @@ -293,11 +331,14 @@ private void checkTopology() { return; } - int serverNodes = safeServerTopologySize(); - if (serverNodes < 0) { - log.warn("Ignite topology is not queryable"); + int serverNodes; + try { + serverNodes = ignite.cluster().forServers().nodes().size(); + } catch (Exception e) { + reportUnqueryableTopology(e); return; } + lastUnqueryableReportNanos = -1; if (log.isDebugEnabled()) { log.debug("Topology poll: serverNodes={}, minimum={}, armed={}, belowMinimumForMs={}", @@ -318,9 +359,20 @@ private void checkTopology() { return; } - // Below the minimum. Startup (nodes joining one by one) is not interesting, so only - // report once the topology has actually reached the minimum at least once. + // Never reached the minimum. Normal startup climbs to it within seconds; a node + // that stays here has very likely started into an ongoing partition and formed its + // own singleton topology, which Ignite never merges back. Report it once - this is + // the split-brain case that produces no segmentation event at all. if (!armed) { + if (!neverReachedMinimumReported + && elapsedMs(startedAtNanos) >= REPORT_NEVER_REACHED_MINIMUM_MS) { + neverReachedMinimumReported = true; + log.error("Server topology has never reached the minimum of {} since startup {} ms ago " + + "(currently {} nodes). This node may have started into an ongoing partition " + + "and formed its own topology, which Ignite cannot merge; it would need a " + + "restart to join the real cluster", + minimumClusterSize, elapsedMs(startedAtNanos), serverNodes); + } return; } @@ -334,7 +386,7 @@ private void checkTopology() { long belowForMs = elapsedMs(belowMinimumSinceNanos); // One escalation per episode so a long orphan does not flood the log - if (!belowMinimumReported && belowForMs >= ORPHAN_REPORT_AFTER_MS) { + if (!belowMinimumReported && belowForMs >= REPORT_AFTER_BELOW_MINIMUM_MS) { belowMinimumReported = true; log.error("Server topology has been at {} nodes, below the minimum of {}, for {} ms. " + "This node is very likely orphaned or split-brained and would need a restart " @@ -343,6 +395,52 @@ private void checkTopology() { } } + /** + * The topology can stay unqueryable indefinitely (a stopped Ignite node in a live JVM), + * so this is throttled: an unbounded repeat would bury the evidence this class exists + * to produce. + */ + private void reportUnqueryableTopology(Exception cause) { + if (lastUnqueryableReportNanos == -1 + || elapsedMs(lastUnqueryableReportNanos) >= UNQUERYABLE_REPORT_INTERVAL_MS) { + lastUnqueryableReportNanos = System.nanoTime(); + log.warn("Ignite topology is not queryable; cluster observations are unavailable " + + "(further occurrences logged at most every {} ms)", + UNQUERYABLE_REPORT_INTERVAL_MS, cause); + } + } + + /** + * Server topology description for log context. Includes the failure reason rather than + * a bare sentinel, so a line that could not read the topology says why. + */ + private String describeServerTopology() { + try { + return "serverNodes=" + ignite.cluster().forServers().nodes().size(); + } catch (Exception e) { + return "serverNodes=unknown (" + e.getClass().getSimpleName() + ": " + e.getMessage() + ")"; + } + } + + private void submit(ScheduledExecutorService executor, Runnable task) { + submit(executor, task, 0); + } + + private void submit(ScheduledExecutorService executor, Runnable task, long delayMs) { + try { + executor.schedule(() -> { + try { + task.run(); + } catch (Throwable t) { + log.error("Unexpected error in cluster observer task", t); + } + }, delayMs, TimeUnit.MILLISECONDS); + } catch (RejectedExecutionException e) { + // Executor already stopping; nothing to diagnose + log.debug("Cluster observer task not scheduled, observer is shutting down"); + } + } + private ScheduledExecutorService newDaemonScheduler(String threadName) { return Executors.newSingleThreadScheduledExecutor(runnable -> { Thread thread = new Thread(runnable, threadName); @@ -351,14 +449,6 @@ private ScheduledExecutorService newDaemonScheduler(String threadName) { }); } - private int safeServerTopologySize() { - try { - return ignite.cluster().forServers().nodes().size(); - } catch (Exception e) { - return -1; - } - } - // Monotonic elapsed time: wall-clock can step forward under NTP corrections or VM // pauses and would misreport how long a node has been below the minimum private static long elapsedMs(long sinceNanos) { From 07c7c8c89ce72dc73972e87adb9240d9143ee96f Mon Sep 17 00:00:00 2001 From: Nicholas Padilla Date: Thu, 30 Jul 2026 07:03:13 -0700 Subject: [PATCH 5/7] fix: harden the cluster observer for long-running production use Final pre-release pass. The observer must be able to run for days and still be believed, so every reported number is now defensible and every Ignite interaction is bounded. Counting and ownership: - count an entry as stale only on the node whose affinity function says it currently owns that partition as PRIMARY. That restores the once-per-entry invariant (localEntries(PRIMARY,BACKUP) had started double counting across nodes) and excludes partitions being rebalanced away, which still hold their old contents and produced phantom findings during rolling restarts. Backup-held copies are counted and reported separately, never mixed into the summable total. Bounded and leak-free: - every Ignite read is time-bounded (async containsKey with timeout, a deadline inside the scan), so the partition-map-exchange stall this class exists to document can no longer silence it - resolve cache handles via cacheNames() so an unknown name cannot trigger a blocking cluster-wide dynamic cache start - close the scan iterator on every path, including the truncation break - shutdown() instead of shutdownNow(): interrupting a thread inside an Ignite operation risked this node leaving as FAILED rather than LEFT Reporting fidelity: - never-reached-minimum now warns first and only escalates to ERROR once slow cluster formation is implausible, repeats at a low rate so a days-old one-shot line is not the only record, and is explicitly retracted when the topology later reaches the minimum - membership and segmentation are logged inline again, from data carried on the event itself (no cluster calls on the discovery thread), so the lines survive a JVM halt moments later - the nodeInfo and subs halves are independent again: a missing subs cache no longer costs the nodeInfo verdict, the direct evidence of a failed cleanup election - throttle the catch-all error path Configuration moved into StructuresProperties as cluster-observer.*, so it ships with configuration metadata and validation like every other structures setting, and the Helm value is a top-level clusterObserver block (the env var binding was verified against Spring's relaxed binder). Co-Authored-By: Claude Fable 5 --- .../structures-server-config-map.yaml | 8 +- helm/structures/values.yaml | 15 +- .../api/config/ClusterObserverProperties.java | 65 +++ .../api/config/StructuresProperties.java | 5 + .../config/IgniteClusterObserver.java | 542 +++++++++++------- 5 files changed, 431 insertions(+), 204 deletions(-) create mode 100644 structures-core/src/main/java/org/kinotic/structures/api/config/ClusterObserverProperties.java diff --git a/helm/structures/templates/structures-server-config-map.yaml b/helm/structures/templates/structures-server-config-map.yaml index a0e4ccf7..b42be038 100644 --- a/helm/structures/templates/structures-server-config-map.yaml +++ b/helm/structures/templates/structures-server-config-map.yaml @@ -39,6 +39,10 @@ data: STRUCTURES_ENABLE_STATIC_FILE_SERVER: "{{ .Values.properties.structures.enableStaticFileServer }}" STRUCTURES_INITIALIZE_WITH_SAMPLE_DATA: "{{ .Values.properties.structures.initializeWithSampleData }}" STRUCTURES_TENANT_ID_FIELD_NAME: "{{ .Values.properties.structures.tenantIdFieldName }}" + # Diagnostic cluster observer (logging only, never shuts anything down). Above 1 it also + # logs episodes where the server topology is below this count, which is the split-brain + # condition Ignite cannot detect. Set to a majority of replicaCount, e.g. 2 for 3. + STRUCTURES_CLUSTER_OBSERVER_MINIMUM_CLUSTER_SIZE: "{{ .Values.clusterObserver.minimumClusterSize | default "1" }}" {{- if .Values.evictionTracking.enabled }} # Eviction tracking - path includes ${POD_NAME} which Spring resolves from the POD_NAME env var STRUCTURES_CACHE_EVICTION_CSV_PATH: "{{ .Values.evictionTracking.mountPath | default "/eviction-data" }}/evictions-${POD_NAME}.csv" @@ -50,10 +54,6 @@ data: # Reference: structures-core/src/test/resources/docker-compose/cluster-test-compose.yml CONTINUUM_CLUSTER_DISABLE_CLUSTERING: "{{ .Values.continuum.disableClustering | default "false" }}" - # Diagnostic cluster observer (logging only, never shuts anything down). Above 1 it also - # logs episodes where the server topology is below this count, which is the split-brain - # condition Ignite cannot detect. Set to a majority of replicaCount, e.g. 2 for 3. - STRUCTURES_CLUSTER_OBSERVER_MINIMUMCLUSTERSIZE: "{{ .Values.continuum.cluster.observerMinimumClusterSize | default "1" }}" CONTINUUM_CLUSTER_DISCOVERY_TYPE: "{{ .Values.continuum.cluster.discoveryType | default "LOCAL" | upper }}" CONTINUUM_CLUSTER_JOIN_TIMEOUT_MS: "{{ .Values.continuum.cluster.joinTimeoutMs | default "0" }}" CONTINUUM_CLUSTER_DISCOVERY_PORT: "{{ .Values.continuum.cluster.discoveryPort | default "47500" }}" diff --git a/helm/structures/values.yaml b/helm/structures/values.yaml index 4241156c..b779f3bc 100644 --- a/helm/structures/values.yaml +++ b/helm/structures/values.yaml @@ -83,12 +83,15 @@ continuum: kubernetesIncludeNotReadyAddresses: false kubernetesMasterUrl: "" kubernetesAccountToken: "" - # Diagnostic cluster observer: logging only, it never shuts anything down. - # 1 (default) logs membership changes, segmentation and post-departure vertx routing - # state. Above 1 it additionally logs episodes where the server topology is below this - # count - the split-brain condition Ignite cannot detect on its own. Set it to a - # majority of replicaCount (floor(n/2)+1), e.g. 2 for 3 replicas. - observerMinimumClusterSize: 1 + +# Diagnostic Ignite cluster observer (logging only - it never shuts anything down). +# Binds to structures.cluster-observer.* in StructuresProperties. +clusterObserver: + # 1 (default): membership, segmentation and post-departure vertx routing diagnostics. + # Above 1: additionally logs episodes where the server topology is below this count, + # the split-brain condition Ignite cannot detect on its own. Set it to a majority of + # replicaCount (floor(n/2)+1), e.g. 2 for 3 replicas. + minimumClusterSize: 1 # TODO: adjust for this later # OIDC security service configuration diff --git a/structures-core/src/main/java/org/kinotic/structures/api/config/ClusterObserverProperties.java b/structures-core/src/main/java/org/kinotic/structures/api/config/ClusterObserverProperties.java new file mode 100644 index 00000000..00b9ea81 --- /dev/null +++ b/structures-core/src/main/java/org/kinotic/structures/api/config/ClusterObserverProperties.java @@ -0,0 +1,65 @@ +package org.kinotic.structures.api.config; + +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; +import lombok.experimental.Accessors; + +/** + * Configuration for the diagnostic Ignite cluster observer. + *

    + * The observer never changes behavior - it only logs - so these settings affect what is + * reported and how often, never what the node does. + * + * @see StructuresProperties + */ +@Getter +@Setter +@Accessors(chain = true) +@NoArgsConstructor +public class ClusterObserverProperties { + + /** + * Minimum number of server nodes expected in the topology. When above 1, episodes where + * the topology is below this count are logged, which is the split-brain condition Ignite + * cannot detect on its own. Set it to a majority of the replica count (floor(n/2)+1), + * e.g. 2 for 3 replicas. The default of 1 leaves topology reporting off; membership, + * segmentation and routing diagnostics are always active. + */ + private Integer minimumClusterSize = 1; + + /** + * How long the topology may stay below {@link #getMinimumClusterSize()} before it is + * reported. Reported once per episode, with an explicit recovery line if it resolves. + */ + private Long reportBelowMinimumAfterMs = 60_000L; + + /** + * How long a node may run without the topology ever reaching + * {@link #getMinimumClusterSize()} before that is reported as a warning. Slow cluster + * formation is normal, so this only warns; see + * {@link #getEscalateNeverReachedMinimumAfterMs()} for the point at which it is treated + * as a real problem. + */ + private Long reportNeverReachedMinimumAfterMs = 300_000L; + + /** + * How long a node may run without the topology ever reaching + * {@link #getMinimumClusterSize()} before it is reported as an error. By this point slow + * startup is no longer a plausible explanation and the node has most likely formed its + * own topology, which Ignite can never merge. + */ + private Long escalateNeverReachedMinimumAfterMs = 900_000L; + + /** + * Upper bound on any single Ignite read the observer performs. Bounded so a cluster + * hang (the very condition being diagnosed) can never stall the observer itself. + */ + private Long inspectionTimeoutMs = 10_000L; + + /** + * Maximum number of local cache entries examined per inspection. A scan that hits this + * cap is reported as inconclusive rather than clean. + */ + private Integer maxEntriesScanned = 50_000; +} diff --git a/structures-core/src/main/java/org/kinotic/structures/api/config/StructuresProperties.java b/structures-core/src/main/java/org/kinotic/structures/api/config/StructuresProperties.java index a0f12302..5cb3f010 100644 --- a/structures-core/src/main/java/org/kinotic/structures/api/config/StructuresProperties.java +++ b/structures-core/src/main/java/org/kinotic/structures/api/config/StructuresProperties.java @@ -126,6 +126,11 @@ public class StructuresProperties { */ private ClusterEvictionProperties clusterEviction = new ClusterEvictionProperties(); + /** + * Diagnostic cluster observer configuration (logging only) + */ + private ClusterObserverProperties clusterObserver = new ClusterObserverProperties(); + public boolean hasElasticUsernameAndPassword(){ return elasticUsername != null && !elasticUsername.isBlank() && elasticPassword != null && !elasticPassword.isBlank(); diff --git a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java index 4a4f0909..84be200b 100644 --- a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java +++ b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java @@ -7,16 +7,21 @@ import org.apache.ignite.IgniteCache; import org.apache.ignite.binary.BinaryObject; import org.apache.ignite.cache.CachePeekMode; +import org.apache.ignite.cache.affinity.Affinity; +import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.events.DiscoveryEvent; import org.apache.ignite.events.Event; import org.apache.ignite.events.EventType; import org.apache.ignite.lang.IgnitePredicate; -import org.springframework.beans.factory.annotation.Value; +import org.kinotic.structures.api.config.ClusterObserverProperties; +import org.kinotic.structures.api.config.StructuresProperties; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Component; import javax.cache.Cache; import java.util.ArrayList; +import java.util.Collection; +import java.util.Iterator; import java.util.List; import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; @@ -26,18 +31,21 @@ /** * Purely diagnostic observer of Ignite cluster membership. It never changes behavior: * it does not shut anything down, does not gate readiness, and takes no action of any - * kind - it only logs what it sees, so it is safe to run everywhere clustering is on. + * kind - it only logs what it sees, so it is safe to run continuously in production. *

    - * All work is handed to its own daemon threads; nothing but the handoff runs on Ignite's - * discovery worker, because stalling that worker would trip Ignite's own failure handler. + * It is built to run for days and stay trustworthy: every Ignite read it performs is + * time-bounded, every repeated condition is throttled, and any result it cannot fully + * substantiate is reported as inconclusive rather than as an all-clear. A diagnostic that + * cries wolf is worse than none, so the reporting rules are deliberately conservative. *

    * What it records: *

      - *
    • membership changes (join/left/failed) with topology version and server count
      • + *
    • membership changes (join/left/failed) with topology version and server count, + * logged inline from the event itself so the line survives a JVM halt moments later
      • *
    • segmentation events - a segmented server node can never rejoin without a restart, - * so this is the highest value line it produces. Note continuum's non-development - * FailureHandler halts the JVM shortly after listeners are notified, so this log may be - * the last thing the process writes.
      • + * so this is the highest value line it produces. Continuum's non-development + * FailureHandler halts the JVM shortly after listeners are notified, so this may be the + * last thing the process writes. *
    • stale vertx routing state after a node departs. vertx-ignite cleans up a departed * node's subscriptions only on the single survivor whose nodeInfoMap.remove(id) returns * true; if that entry is already gone, no node runs cleanSubs and stale __vertx.subs @@ -45,79 +53,95 @@ * failures. Whether that is what happens here is UNCONFIRMED - continuum contributes a * "*" cache template (PARTITIONED, backups=1, PRIMARY_SYNC) covering the __vertx.* * caches, so entries are not lost outright on a single node failure. This observer - * exists to capture evidence rather than assume a mechanism. Cleanup removes entries one - * at a time and can legitimately take a while, so the check is sampled several times - * after a departure; only the final sample warns. Every result states the coverage it - * achieved, and a scan that could not cover the data never reports "clean" - an - * inconclusive result must never be mistaken for an all-clear.
      • - *
    • server topology below structures.cluster.observer.minimumClusterSize (the only - * configuration this class has, default 1 = topology reporting off). That is the - * split-brain condition Ignite cannot detect by design: group splits keep a healthy ring - * on each side, and a restart into a partition forms a fresh singleton topology that - * Ignite never merges. Set it to a majority of the replica count (floor(n/2)+1) to have - * those episodes logged, including a node that never reaches the minimum at all.
      • + * exists to capture evidence rather than assume a mechanism. + *
    • server topology below the configured minimum cluster size - the split-brain + * condition Ignite cannot detect by design, since group splits keep a healthy ring on + * each side and a restart into a partition forms a fresh singleton topology that Ignite + * never merges.
      • *
    + * Counting rules for the routing inspection, so numbers from different pods can be + * compared and summed: an entry is counted as stale by the node that currently owns its + * partition as PRIMARY, verified through the affinity function rather than inferred from + * local storage. That keeps each entry counted exactly once cluster-wide and excludes + * partitions in the process of being rebalanced away, which still hold their old contents + * and would otherwise produce phantom findings during a rolling restart. Entries held + * only as backups are counted and reported separately, never mixed into the primary total. + * + * @see ClusterObserverProperties */ @Slf4j @Component @ConditionalOnProperty(value = "continuum.disableClustering", havingValue = "false", matchIfMissing = true) public class IgniteClusterObserver { + private static final String NODE_INFO_CACHE = "__vertx.nodeInfo"; + private static final String SUBS_CACHE = "__vertx.subs"; private static final long TOPOLOGY_POLL_MS = 10_000L; - private static final long REPORT_AFTER_BELOW_MINIMUM_MS = 60_000L; - private static final long REPORT_NEVER_REACHED_MINIMUM_MS = 300_000L; private static final long UNQUERYABLE_REPORT_INTERVAL_MS = 600_000L; + private static final long UNEXPECTED_ERROR_REPORT_INTERVAL_MS = 600_000L; + private static final long NEVER_REACHED_REPORT_INTERVAL_MS = 3_600_000L; private static final int MAX_STALE_ADDRESSES_LOGGED = 10; - private static final int MAX_ENTRIES_SCANNED = 50_000; - /** Sampled repeatedly so a slow cleanup is not reported as a leak; only the last warns */ + /** Sampled repeatedly so a slow cleanup is not reported as a leak; only the final sample warns */ private static final long[] STALE_ROUTE_SAMPLE_DELAYS_MS = {5_000L, 20_000L, 60_000L}; private final Ignite ignite; - - /** - * Minimum number of server nodes expected in the topology. When above 1, episodes - * below it are logged (with duration). Purely informational - nothing is ever shut - * down. Set it to a majority of the replica count, e.g. 2 for 3 replicas. - */ - @Value("${structures.cluster.observer.minimumClusterSize:${structures.cluster.observer.minimum-cluster-size:1}}") - private int minimumClusterSize; + private final ClusterObserverProperties properties; private volatile boolean closed = false; private volatile boolean armed = false; private volatile boolean belowMinimumReported = false; - private volatile boolean neverReachedMinimumReported = false; private volatile long belowMinimumSinceNanos = -1; private volatile long startedAtNanos = -1; private volatile long lastUnqueryableReportNanos = -1; + private volatile long lastUnexpectedErrorReportNanos = -1; + private volatile long lastNeverReachedReportNanos = -1; + private volatile boolean neverReachedEscalated = false; private IgnitePredicate membershipListener; private IgnitePredicate segmentationListener; private ScheduledExecutorService scheduler; private ScheduledExecutorService inspector; - public IgniteClusterObserver(Ignite ignite) { + public IgniteClusterObserver(Ignite ignite, StructuresProperties structuresProperties) { this.ignite = ignite; + this.properties = structuresProperties.getClusterObserver(); } @PostConstruct public void start() { startedAtNanos = System.nanoTime(); scheduler = newDaemonScheduler("structures-cluster-observer"); - // Inspections get their own thread: a cache read during a partition can block for - // a long time, and it must never stall topology polling + // Inspections get their own thread so a slow cache read can never delay topology polling inspector = newDaemonScheduler("structures-cluster-inspector"); - // The listener captures the event and hands off immediately. Ignite notifies local - // listeners inline on the discovery worker, which is a monitored critical worker: - // logging (and any cluster call) there risks tripping SYSTEM_WORKER_BLOCKED. + // Logged inline, from data carried on the event itself. No cluster calls here: Ignite + // notifies listeners on its monitored discovery worker, and this line must be on disk + // before a segmentation halt can discard it. membershipListener = event -> { - DiscoveryEvent discoveryEvent = (DiscoveryEvent) event; - String eventNodeId = discoveryEvent.eventNode().id().toString(); - long topologyVersion = discoveryEvent.topologyVersion(); - int eventType = event.type(); - submit(scheduler, () -> logMembershipChange(eventType, eventNodeId, topologyVersion)); - if (eventType == EventType.EVT_NODE_LEFT || eventType == EventType.EVT_NODE_FAILED) { - scheduleStaleRouteSamples(eventNodeId); + try { + DiscoveryEvent discoveryEvent = (DiscoveryEvent) event; + String eventNodeId = discoveryEvent.eventNode().id().toString(); + long topologyVersion = discoveryEvent.topologyVersion(); + int serverNodes = countServers(discoveryEvent.topologyNodes()); + switch (event.type()) { + case EventType.EVT_NODE_JOINED -> + log.info("Cluster node joined: {} (topologyVersion={}, serverNodes={})", + eventNodeId, topologyVersion, serverNodes); + case EventType.EVT_NODE_LEFT -> { + log.info("Cluster node left: {} (topologyVersion={}, serverNodes={})", + eventNodeId, topologyVersion, serverNodes); + scheduleStaleRouteSamples(eventNodeId); + } + case EventType.EVT_NODE_FAILED -> { + log.warn("Cluster node FAILED: {} (topologyVersion={}, serverNodes={})", + eventNodeId, topologyVersion, serverNodes); + scheduleStaleRouteSamples(eventNodeId); + } + default -> { /* not registered for others */ } + } + } catch (Throwable t) { + // Never let a diagnostic throw into Ignite's discovery thread + log.warn("Error handling cluster membership event", t); } return true; }; @@ -127,36 +151,41 @@ public void start() { EventType.EVT_NODE_FAILED); // Logged, never acted on. Continuum's FailureHandler decides what happens to the - // process. Logged inline rather than handed off: the JVM is likely to be halted - // moments from now, and an off-thread log would never be written. + // process; inline for the same durability reason as above. segmentationListener = event -> { - log.error("Node segmentation detected: this Ignite node was segmented from the cluster. " - + "Segmented server nodes cannot rejoin without a restart. {}", - describeServerTopology()); + try { + DiscoveryEvent discoveryEvent = (DiscoveryEvent) event; + log.error("Node segmentation detected: this Ignite node was segmented from the " + + "cluster and cannot rejoin without a restart " + + "(topologyVersion={}, serverNodesVisible={})", + discoveryEvent.topologyVersion(), countServers(discoveryEvent.topologyNodes())); + } catch (Throwable t) { + log.error("Node segmentation detected: this Ignite node was segmented from the " + + "cluster and cannot rejoin without a restart", t); + } return false; // one shot }; ignite.events().localListen(segmentationListener, EventType.EVT_NODE_SEGMENTED); - if (minimumClusterSize > 1) { + if (minimumClusterSize() > 1) { scheduler.scheduleWithFixedDelay(this::checkTopologySafely, 0, TOPOLOGY_POLL_MS, TimeUnit.MILLISECONDS); log.info("Ignite cluster observer started (diagnostic only): minimumClusterSize={}", - minimumClusterSize); + minimumClusterSize()); } else { - log.info("Ignite cluster observer started (diagnostic only): membership and routing " - + "diagnostics active, topology reporting off " - + "(set structures.cluster.observer.minimumClusterSize above 1 to enable it)"); + log.info("Ignite cluster observer started (diagnostic only): membership, segmentation " + + "and routing diagnostics active, topology reporting off " + + "(set structures.cluster-observer.minimum-cluster-size above 1 to enable it)"); } } @PreDestroy public void stop() { closed = true; - // Deregister listeners BEFORE stopping the executors, otherwise a departure arriving - // in between would schedule onto a terminated executor and throw - // RejectedExecutionException back into Ignite's discovery notification thread + // Deregister listeners BEFORE stopping the executors, so a departure arriving in + // between cannot schedule onto a terminated executor and throw back into Ignite if (membershipListener != null) { try { ignite.events().stopLocalListen(membershipListener, @@ -174,26 +203,15 @@ public void stop() { log.debug("Could not remove segmentation listener during shutdown", e); } } + // shutdown(), never shutdownNow(): interrupting a thread inside an Ignite cache + // operation can tear down in-flight futures and make this node leave as FAILED + // rather than LEFT. Tasks observe `closed` and return promptly, and both executors + // are daemons, so nothing can hold the JVM open. if (scheduler != null) { - scheduler.shutdownNow(); + scheduler.shutdown(); } if (inspector != null) { - inspector.shutdownNow(); - } - } - - private void logMembershipChange(int eventType, String eventNodeId, long topologyVersion) { - switch (eventType) { - case EventType.EVT_NODE_JOINED -> - log.info("Cluster node joined: {} (topologyVersion={}, {})", - eventNodeId, topologyVersion, describeServerTopology()); - case EventType.EVT_NODE_LEFT -> - log.info("Cluster node left: {} (topologyVersion={}, {})", - eventNodeId, topologyVersion, describeServerTopology()); - case EventType.EVT_NODE_FAILED -> - log.warn("Cluster node FAILED: {} (topologyVersion={}, {})", - eventNodeId, topologyVersion, describeServerTopology()); - default -> { /* not registered for others */ } + inspector.shutdown(); } } @@ -205,20 +223,21 @@ private void scheduleStaleRouteSamples(String departedNodeId) { for (int i = 0; i < STALE_ROUTE_SAMPLE_DELAYS_MS.length; i++) { boolean finalSample = i == STALE_ROUTE_SAMPLE_DELAYS_MS.length - 1; long delay = STALE_ROUTE_SAMPLE_DELAYS_MS[i]; - submit(inspector, - () -> reportStaleRoutingState(departedNodeId, departedAtNanos, finalSample), - delay); + try { + inspector.schedule(() -> reportStaleRoutingState(departedNodeId, departedAtNanos, finalSample), + delay, TimeUnit.MILLISECONDS); + } catch (RejectedExecutionException e) { + log.debug("Routing inspection not scheduled, observer is shutting down"); + return; + } } } /** - * Inspect this node's local partitions of the vertx routing caches for entries that - * still reference a departed node. Local by design: it costs nothing beyond a - * node-local iteration and adds no distributed query load while the cluster is already - * rebalancing. Primary AND backup partitions are read, because a departure triggers - * rebalancing and an entry this node holds only as a backup is exactly the kind a - * primary-only scan would miss. Every log line states the coverage achieved so an - * inconclusive scan can never be read as an all-clear. + * Inspect this node's local view of the vertx routing caches for entries that still + * reference a departed node. Local reads only, so this adds no distributed query load + * while the cluster is already rebalancing, and every Ignite call is time-bounded so a + * cluster hang cannot silence the observer. */ private void reportStaleRoutingState(String departedNodeId, long departedAtNanos, boolean finalSample) { if (closed) { @@ -228,83 +247,63 @@ private void reportStaleRoutingState(String departedNodeId, long departedAtNanos // cluster is unhealthy, which is exactly when the timestamp matters long afterMs = elapsedMs(departedAtNanos); try { - IgniteCache nodeInfoCache = ignite.cache("__vertx.nodeInfo"); - IgniteCache subsCache = ignite.cache("__vertx.subs"); - - if (nodeInfoCache == null || subsCache == null) { - log.warn("Inconclusive routing inspection {} ms after departure of node {}: " - + "vertx caches are not available on this node " - + "(nodeInfoCachePresent={}, subsCachePresent={}). No conclusion can be " - + "drawn about stale routing state.", - afterMs, departedNodeId, nodeInfoCache != null, subsCache != null); + // cacheNames() is a local metadata read; ignite.cache() on an unknown name can + // otherwise trigger a blocking cluster-wide dynamic cache start + Collection cacheNames = ignite.cacheNames(); + boolean nodeInfoAvailable = cacheNames.contains(NODE_INFO_CACHE); + boolean subsAvailable = cacheNames.contains(SUBS_CACHE); + + if (!nodeInfoAvailable && !subsAvailable) { + log.warn("Inconclusive routing inspection {} ms after departure of node {}: neither " + + "{} nor {} exists on this node, so no conclusion can be drawn about stale " + + "routing state", afterMs, departedNodeId, NODE_INFO_CACHE, SUBS_CACHE); return; } - boolean nodeInfoPresent = nodeInfoCache.containsKey(departedNodeId); - - int staleSubs = 0; - int scanned = 0; - int nonBinaryKeys = 0; - boolean truncated = false; - List staleAddresses = new ArrayList<>(); - // localEntries with explicit peek modes rather than a ScanQuery: it is - // node-local (no distributed query while the cluster is rebalancing) AND it - // includes backup partitions. A plain scan query only covers partitions this - // node is primary for, which is exactly where a departure's stale entries can - // hide during rebalancing. Binary form avoids a compile-time dependency on - // vertx-ignite's IgniteRegistrationInfo (the field names match its writeBinary). - for (Cache.Entry entry : subsCache.withKeepBinary() - .localEntries(CachePeekMode.PRIMARY, - CachePeekMode.BACKUP)) { - if (scanned >= MAX_ENTRIES_SCANNED) { - truncated = true; - break; - } - scanned++; - if (entry.getKey() instanceof BinaryObject key) { - if (departedNodeId.equals(key.field("nodeId"))) { - staleSubs++; - if (staleAddresses.size() < MAX_STALE_ADDRESSES_LOGGED) { - staleAddresses.add(key.field("address")); - } - } - } else { - nonBinaryKeys++; - } - } + // Each half is independent: a missing subs cache must not cost us the nodeInfo + // verdict, which is the direct evidence of a failed cleanup election + Boolean nodeInfoPresent = nodeInfoAvailable ? checkNodeInfoPresent(departedNodeId) : null; + SubsScanResult subs = subsAvailable ? scanSubs(departedNodeId) : null; - // An unexamined remainder or unreadable keys mean the scan cannot support an - // all-clear, so say so rather than implying the caches are clean - boolean conclusive = !truncated && nonBinaryKeys == 0; String coverage = String.format( - "localEntriesScanned=%d (primary+backup), truncated=%b, unreadableKeys=%d", - scanned, truncated, nonBinaryKeys); - - if (nodeInfoPresent || staleSubs > 0) { + "nodeInfoChecked=%s, subsScanned=%s", + nodeInfoPresent != null ? "yes" : "no (unavailable or timed out)", + subs != null ? subs.describe() : "no (cache unavailable)"); + + boolean sawStale = Boolean.TRUE.equals(nodeInfoPresent) + || (subs != null && (subs.primaryStale > 0 || subs.backupStale > 0)); + boolean conclusive = nodeInfoPresent != null && subs != null && subs.conclusive(); + + if (sawStale) { + String detail = String.format( + "nodeInfoStillPresent=%s, staleSubsOwnedHere=%d, staleSubsBackupCopiesHere=%d, " + + "sampleAddresses=%s, %s", + nodeInfoPresent, subs != null ? subs.primaryStale : -1, + subs != null ? subs.backupStale : -1, + subs != null ? subs.addresses : List.of(), coverage); if (finalSample) { - log.warn("Stale routing state remains {} ms after departure of node {}: " - + "nodeInfoStillPresent={}, staleLocalSubscriptionEntries={}, " - + "sampleAddresses={}, {}. Event bus sends to these addresses can fail " - + "with 'Not a member of the cluster' until the handlers re-register.", - afterMs, departedNodeId, nodeInfoPresent, staleSubs, staleAddresses, coverage); + log.warn("Stale routing state remains {} ms after departure of node {}: {}. " + + "staleSubsOwnedHere counts only partitions this node currently owns as " + + "primary, so it is safe to sum across pods; backup copies are reported " + + "separately and duplicate another pod's primary count. Event bus sends " + + "to these addresses can fail with 'Not a member of the cluster' until " + + "the handlers re-register.", + afterMs, departedNodeId, detail); } else { - log.info("Routing cleanup still in progress {} ms after departure of node {}: " - + "nodeInfoStillPresent={}, staleLocalSubscriptionEntries={}, {}", - afterMs, departedNodeId, nodeInfoPresent, staleSubs, coverage); + log.info("Routing cleanup still in progress {} ms after departure of node {}: {}", + afterMs, departedNodeId, detail); } } else if (conclusive) { log.info("Routing caches are clean (local view) {} ms after departure of node {}: {}", afterMs, departedNodeId, coverage); } else { log.warn("Inconclusive routing inspection {} ms after departure of node {}: no stale " - + "entries seen, but the scan did not cover all local entries so this is NOT " - + "an all-clear ({})", - afterMs, departedNodeId, coverage); + + "entries seen, but the inspection did not complete fully so this is NOT an " + + "all-clear ({})", afterMs, departedNodeId, coverage); } } catch (Exception e) { - if (closed || Thread.currentThread().isInterrupted()) { - // Ordinary shutdown interrupted the read; not a cluster problem - log.debug("Routing inspection for node {} aborted during shutdown", departedNodeId, e); + if (closed) { + log.debug("Routing inspection for node {} abandoned during shutdown", departedNodeId, e); } else { // WARN, not DEBUG: a failed inspection must never be mistaken for a clean result log.warn("Could not inspect routing caches {} ms after departure of node {}; " @@ -314,15 +313,93 @@ private void reportStaleRoutingState(String departedNodeId, long departedAtNanos } } + /** + * @return TRUE/FALSE if the check completed, or null if it could not be completed + * within the configured timeout - never guess, the caller reports it as inconclusive + */ + private Boolean checkNodeInfoPresent(String departedNodeId) { + try { + IgniteCache cache = ignite.cache(NODE_INFO_CACHE); + if (cache == null) { + return null; + } + // Async with an explicit timeout: containsKey on a PARTITIONED cache is a + // distributed read that can block on partition map exchange indefinitely, + // which is precisely the condition being diagnosed + return cache.containsKeyAsync(departedNodeId) + .get(inspectionTimeoutMs(), TimeUnit.MILLISECONDS); + } catch (Exception e) { + log.debug("nodeInfo lookup for departed node {} did not complete", departedNodeId, e); + return null; + } + } + + private SubsScanResult scanSubs(String departedNodeId) { + SubsScanResult result = new SubsScanResult(); + IgniteCache cache = ignite.cache(SUBS_CACHE); + if (cache == null) { + return null; + } + Affinity affinity = ignite.affinity(SUBS_CACHE); + ClusterNode localNode = ignite.cluster().localNode(); + long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(inspectionTimeoutMs()); + int maxEntries = maxEntriesScanned(); + + Iterable> entries = + cache.withKeepBinary().localEntries(CachePeekMode.PRIMARY, CachePeekMode.BACKUP); + Iterator> iterator = entries.iterator(); + try { + while (iterator.hasNext()) { + if (result.scanned >= maxEntries) { + result.truncated = true; + break; + } + if (System.nanoTime() > deadlineNanos) { + result.timedOut = true; + break; + } + Cache.Entry entry = iterator.next(); + result.scanned++; + if (!(entry.getKey() instanceof BinaryObject key)) { + result.unreadableKeys++; + continue; + } + if (!departedNodeId.equals(key.field("nodeId"))) { + continue; + } + // Ownership from the affinity function, not from the fact the data is here: + // partitions being rebalanced away still hold their old contents, and counting + // those would produce phantom findings during every rolling restart + if (affinity.isPrimary(localNode, key)) { + result.primaryStale++; + if (result.addresses.size() < MAX_STALE_ADDRESSES_LOGGED) { + result.addresses.add(String.valueOf(key.field("address"))); + } + } else if (affinity.isBackup(localNode, key)) { + result.backupStale++; + } + // Entries in partitions this node no longer owns are deliberately ignored + } + } finally { + closeQuietly(iterator); + } + return result; + } + /** * An exception escaping a scheduled task silently cancels all future executions - never - * let that happen + * let that happen, and never let a persistent failure flood the log either */ private void checkTopologySafely() { try { checkTopology(); } catch (Throwable t) { - log.error("Unexpected error in cluster observer check", t); + if (lastUnexpectedErrorReportNanos == -1 + || elapsedMs(lastUnexpectedErrorReportNanos) >= UNEXPECTED_ERROR_REPORT_INTERVAL_MS) { + lastUnexpectedErrorReportNanos = System.nanoTime(); + log.error("Unexpected error in cluster observer check (further occurrences logged at " + + "most every {} ms)", UNEXPECTED_ERROR_REPORT_INTERVAL_MS, t); + } } } @@ -340,6 +417,8 @@ private void checkTopology() { } lastUnqueryableReportNanos = -1; + int minimumClusterSize = minimumClusterSize(); + if (log.isDebugEnabled()) { log.debug("Topology poll: serverNodes={}, minimum={}, armed={}, belowMinimumForMs={}", serverNodes, minimumClusterSize, armed, @@ -349,7 +428,19 @@ private void checkTopology() { if (serverNodes >= minimumClusterSize) { if (!armed) { armed = true; - log.info("Cluster observer armed: server topology reached {} nodes", serverNodes); + long formationMs = elapsedMs(startedAtNanos); + if (lastNeverReachedReportNanos != -1) { + // Explicitly retract the earlier report so days of logs are not left with a + // scary line that later turned out to be slow startup + log.info("Server topology reached the minimum of {} ({} nodes) after {} ms; the " + + "earlier report about never reaching the minimum is resolved", + minimumClusterSize, serverNodes, formationMs); + } else { + log.info("Cluster observer armed: server topology reached {} nodes after {} ms", + serverNodes, formationMs); + } + lastNeverReachedReportNanos = -1; + neverReachedEscalated = false; } else if (belowMinimumSinceNanos != -1) { log.info("Server topology recovered to {} nodes after {} ms below the minimum of {}", serverNodes, elapsedMs(belowMinimumSinceNanos), minimumClusterSize); @@ -359,39 +450,60 @@ private void checkTopology() { return; } - // Never reached the minimum. Normal startup climbs to it within seconds; a node - // that stays here has very likely started into an ongoing partition and formed its - // own singleton topology, which Ignite never merges back. Report it once - this is - // the split-brain case that produces no segmentation event at all. if (!armed) { - if (!neverReachedMinimumReported - && elapsedMs(startedAtNanos) >= REPORT_NEVER_REACHED_MINIMUM_MS) { - neverReachedMinimumReported = true; - log.error("Server topology has never reached the minimum of {} since startup {} ms ago " - + "(currently {} nodes). This node may have started into an ongoing partition " - + "and formed its own topology, which Ignite cannot merge; it would need a " - + "restart to join the real cluster", - minimumClusterSize, elapsedMs(startedAtNanos), serverNodes); - } + reportNeverReachedMinimum(serverNodes, minimumClusterSize); return; } if (belowMinimumSinceNanos == -1) { belowMinimumSinceNanos = System.nanoTime(); log.warn("Server topology dropped to {} nodes, below the minimum of {}. This node may be " - + "orphaned from the cluster; watching", - serverNodes, minimumClusterSize); + + "orphaned from the cluster; watching", serverNodes, minimumClusterSize); return; } long belowForMs = elapsedMs(belowMinimumSinceNanos); // One escalation per episode so a long orphan does not flood the log - if (!belowMinimumReported && belowForMs >= REPORT_AFTER_BELOW_MINIMUM_MS) { + if (!belowMinimumReported && belowForMs >= reportBelowMinimumAfterMs()) { belowMinimumReported = true; log.error("Server topology has been at {} nodes, below the minimum of {}, for {} ms. " + "This node is very likely orphaned or split-brained and would need a restart " - + "to rejoin the cluster", - serverNodes, minimumClusterSize, belowForMs); + + "to rejoin the cluster", serverNodes, minimumClusterSize, belowForMs); + } + } + + /** + * A node that has not yet reached the minimum may simply be starting into a cluster + * that is still forming, which on Kubernetes can legitimately take minutes. So this + * warns first, escalates to an error only once slow startup is no longer plausible, + * and repeats at a low rate so a long-running pod's current state is always visible in + * recent logs rather than only in a one-shot line from days ago. + */ + private void reportNeverReachedMinimum(int serverNodes, int minimumClusterSize) { + long runningMs = elapsedMs(startedAtNanos); + if (runningMs < reportNeverReachedMinimumAfterMs()) { + return; + } + boolean escalate = runningMs >= escalateNeverReachedMinimumAfterMs(); + boolean firstEscalation = escalate && !neverReachedEscalated; + if (!firstEscalation + && lastNeverReachedReportNanos != -1 + && elapsedMs(lastNeverReachedReportNanos) < NEVER_REACHED_REPORT_INTERVAL_MS) { + return; + } + lastNeverReachedReportNanos = System.nanoTime(); + if (escalate) { + neverReachedEscalated = true; + log.error("Server topology has never reached the minimum of {} in the {} ms since startup " + + "(currently {} nodes). Cluster formation should be long complete, so this node " + + "has most likely formed its own topology - which Ignite can never merge - and " + + "would need a restart to join the real cluster", + minimumClusterSize, runningMs, serverNodes); + } else { + log.warn("Server topology has not yet reached the minimum of {} in the {} ms since startup " + + "(currently {} nodes). This is normal while a cluster is still forming; it will " + + "be reported as an error if it persists past {} ms", + minimumClusterSize, runningMs, serverNodes, escalateNeverReachedMinimumAfterMs()); } } @@ -410,34 +522,26 @@ private void reportUnqueryableTopology(Exception cause) { } } - /** - * Server topology description for log context. Includes the failure reason rather than - * a bare sentinel, so a line that could not read the topology says why. - */ - private String describeServerTopology() { - try { - return "serverNodes=" + ignite.cluster().forServers().nodes().size(); - } catch (Exception e) { - return "serverNodes=unknown (" + e.getClass().getSimpleName() + ": " + e.getMessage() + ")"; + private static int countServers(Collection nodes) { + if (nodes == null) { + return -1; } + int count = 0; + for (ClusterNode node : nodes) { + if (!node.isClient()) { + count++; + } + } + return count; } - private void submit(ScheduledExecutorService executor, Runnable task) { - submit(executor, task, 0); - } - - private void submit(ScheduledExecutorService executor, Runnable task, long delayMs) { - try { - executor.schedule(() -> { - try { - task.run(); - } catch (Throwable t) { - log.error("Unexpected error in cluster observer task", t); - } - }, delayMs, TimeUnit.MILLISECONDS); - } catch (RejectedExecutionException e) { - // Executor already stopping; nothing to diagnose - log.debug("Cluster observer task not scheduled, observer is shutting down"); + private static void closeQuietly(Iterator iterator) { + if (iterator instanceof AutoCloseable closeable) { + try { + closeable.close(); + } catch (Exception e) { + log.debug("Could not close cache iterator", e); + } } } @@ -450,8 +554,58 @@ private ScheduledExecutorService newDaemonScheduler(String threadName) { } // Monotonic elapsed time: wall-clock can step forward under NTP corrections or VM - // pauses and would misreport how long a node has been below the minimum + // pauses and would misreport how long a condition has lasted private static long elapsedMs(long sinceNanos) { return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - sinceNanos); } + + private int minimumClusterSize() { + return properties.getMinimumClusterSize() != null ? properties.getMinimumClusterSize() : 1; + } + + private long reportBelowMinimumAfterMs() { + return properties.getReportBelowMinimumAfterMs() != null + ? properties.getReportBelowMinimumAfterMs() : 60_000L; + } + + private long reportNeverReachedMinimumAfterMs() { + return properties.getReportNeverReachedMinimumAfterMs() != null + ? properties.getReportNeverReachedMinimumAfterMs() : 300_000L; + } + + private long escalateNeverReachedMinimumAfterMs() { + return properties.getEscalateNeverReachedMinimumAfterMs() != null + ? properties.getEscalateNeverReachedMinimumAfterMs() : 900_000L; + } + + private long inspectionTimeoutMs() { + return properties.getInspectionTimeoutMs() != null ? properties.getInspectionTimeoutMs() : 10_000L; + } + + private int maxEntriesScanned() { + return properties.getMaxEntriesScanned() != null ? properties.getMaxEntriesScanned() : 50_000; + } + + /** + * Result of a local __vertx.subs scan. primaryStale is the cluster-summable count; + * backupStale duplicates another node's primary count and is kept separate. + */ + private static final class SubsScanResult { + private int scanned; + private int primaryStale; + private int backupStale; + private int unreadableKeys; + private boolean truncated; + private boolean timedOut; + private final List addresses = new ArrayList<>(); + + private boolean conclusive() { + return !truncated && !timedOut && unreadableKeys == 0; + } + + private String describe() { + return String.format("localEntriesScanned=%d, truncated=%b, timedOut=%b, unreadableKeys=%d", + scanned, truncated, timedOut, unreadableKeys); + } + } } From a93925a09d308578aa66a84768d668b4ac39c94b Mon Sep 17 00:00:00 2001 From: Nicholas Padilla Date: Thu, 30 Jul 2026 09:21:33 -0700 Subject: [PATCH 6/7] fix: make the routing inspection tolerant instead of precise Chasing an exact, cluster-summable stale-entry count kept trading one defect for another, and precision was never what this needs: a transient failure just means the next sample retries, while a real problem shows up as the same complaint repeating. So the inspection is now defensive by construction and self-contained - it cannot affect clustering. - inspections run on a small bounded pool instead of a single thread, so a read wedged inside Ignite can no longer silence every later inspection for the life of the pod; when all threads are busy the skipped inspection is logged, which is itself evidence of a hang - drop the per-entry affinity ownership check. It could silently discard entries when the topology moved mid-scan (producing exactly the false all-clear the class forbids) and left backup-only findings with an empty address list. Counts are now plainly this node's local view and documented as overlapping across pods - catch Throwable, not Exception, around inspections: an Error from Ignite internals iterating a partition being evicted would otherwise vanish into the executor as a silence that reads like a clean result - always report why an inspection was incomplete, at INFO for early samples (expected during a rebalance) and WARN on the final one, so a failure is never invisible at production log levels nor alarming when it is routine - report the below-minimum condition hourly rather than once per process, so a node orphaned days ago is still visible in a recent log window - stop resetting the unqueryable-topology throttle on success, which made it useless against an alternating failure pattern - move the Helm value under properties.structures.clusterObserver, where every other STRUCTURES_* setting lives, so an override placed there is not silently ignored Co-Authored-By: Claude Fable 5 --- .../structures-server-config-map.yaml | 2 +- helm/structures/values.yaml | 17 +- .../config/IgniteClusterObserver.java | 309 ++++++++++++------ 3 files changed, 209 insertions(+), 119 deletions(-) diff --git a/helm/structures/templates/structures-server-config-map.yaml b/helm/structures/templates/structures-server-config-map.yaml index b42be038..36d7e938 100644 --- a/helm/structures/templates/structures-server-config-map.yaml +++ b/helm/structures/templates/structures-server-config-map.yaml @@ -42,7 +42,7 @@ data: # Diagnostic cluster observer (logging only, never shuts anything down). Above 1 it also # logs episodes where the server topology is below this count, which is the split-brain # condition Ignite cannot detect. Set to a majority of replicaCount, e.g. 2 for 3. - STRUCTURES_CLUSTER_OBSERVER_MINIMUM_CLUSTER_SIZE: "{{ .Values.clusterObserver.minimumClusterSize | default "1" }}" + STRUCTURES_CLUSTER_OBSERVER_MINIMUM_CLUSTER_SIZE: "{{ .Values.properties.structures.clusterObserver.minimumClusterSize | default "1" }}" {{- if .Values.evictionTracking.enabled }} # Eviction tracking - path includes ${POD_NAME} which Spring resolves from the POD_NAME env var STRUCTURES_CACHE_EVICTION_CSV_PATH: "{{ .Values.evictionTracking.mountPath | default "/eviction-data" }}/evictions-${POD_NAME}.csv" diff --git a/helm/structures/values.yaml b/helm/structures/values.yaml index b779f3bc..5db41c1f 100644 --- a/helm/structures/values.yaml +++ b/helm/structures/values.yaml @@ -53,6 +53,14 @@ properties: apiKey: "noop" baseUrl: "https://api.x.ai" model: "grok-4" + # Diagnostic Ignite cluster observer (logging only - it never shuts anything down). + # Binds to structures.cluster-observer.* in StructuresProperties. + clusterObserver: + # 1 (default): membership, segmentation and post-departure vertx routing diagnostics. + # Above 1: additionally logs episodes where the server topology is below this count, + # the split-brain condition Ignite cannot detect on its own. Set it to a majority of + # replicaCount (floor(n/2)+1), e.g. 2 for 3 replicas. + minimumClusterSize: 1 continuum_gateway: stomp: @@ -84,15 +92,6 @@ continuum: kubernetesMasterUrl: "" kubernetesAccountToken: "" -# Diagnostic Ignite cluster observer (logging only - it never shuts anything down). -# Binds to structures.cluster-observer.* in StructuresProperties. -clusterObserver: - # 1 (default): membership, segmentation and post-departure vertx routing diagnostics. - # Above 1: additionally logs episodes where the server topology is below this count, - # the split-brain condition Ignite cannot detect on its own. Set it to a majority of - # replicaCount (floor(n/2)+1), e.g. 2 for 3 replicas. - minimumClusterSize: 1 - # TODO: adjust for this later # OIDC security service configuration oidc: diff --git a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java index 84be200b..c52216aa 100644 --- a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java +++ b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java @@ -7,7 +7,6 @@ import org.apache.ignite.IgniteCache; import org.apache.ignite.binary.BinaryObject; import org.apache.ignite.cache.CachePeekMode; -import org.apache.ignite.cache.affinity.Affinity; import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.events.DiscoveryEvent; import org.apache.ignite.events.Event; @@ -23,9 +22,11 @@ import java.util.Collection; import java.util.Iterator; import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** @@ -59,13 +60,12 @@ * each side and a restart into a partition forms a fresh singleton topology that Ignite * never merges. * - * Counting rules for the routing inspection, so numbers from different pods can be - * compared and summed: an entry is counted as stale by the node that currently owns its - * partition as PRIMARY, verified through the affinity function rather than inferred from - * local storage. That keeps each entry counted exactly once cluster-wide and excludes - * partitions in the process of being rebalanced away, which still hold their old contents - * and would otherwise produce phantom findings during a rolling restart. Entries held - * only as backups are counted and reported separately, never mixed into the primary total. + * Reading the routing inspection output: counts are this node's LOCAL view - primary and + * backup copies, including partitions being rebalanced - so the same entry can appear in + * more than one pod's log. Correlate the reported addresses across pods rather than + * summing the counts. Precision is deliberately traded for robustness: a failed or partial + * inspection is reported and retried by the next sample, so transient conditions resolve + * themselves and it is a complaint that keeps repeating which indicates a real problem. * * @see ClusterObserverProperties */ @@ -80,6 +80,12 @@ public class IgniteClusterObserver { private static final long UNQUERYABLE_REPORT_INTERVAL_MS = 600_000L; private static final long UNEXPECTED_ERROR_REPORT_INTERVAL_MS = 600_000L; private static final long NEVER_REACHED_REPORT_INTERVAL_MS = 3_600_000L; + private static final long BELOW_MINIMUM_REPORT_INTERVAL_MS = 3_600_000L; + // A wedged Ignite read must never silence later inspections, so inspections run on a + // small bounded pool: at worst a hung cluster consumes these threads and subsequent + // inspections are skipped with a log line, which is itself evidence of the hang + private static final int INSPECTOR_THREADS = 2; + private static final int INSPECTOR_QUEUE_DEPTH = 8; private static final int MAX_STALE_ADDRESSES_LOGGED = 10; /** Sampled repeatedly so a slow cleanup is not reported as a leak; only the final sample warns */ private static final long[] STALE_ROUTE_SAMPLE_DELAYS_MS = {5_000L, 20_000L, 60_000L}; @@ -89,7 +95,7 @@ public class IgniteClusterObserver { private volatile boolean closed = false; private volatile boolean armed = false; - private volatile boolean belowMinimumReported = false; + private volatile long lastBelowMinimumReportNanos = -1; private volatile long belowMinimumSinceNanos = -1; private volatile long startedAtNanos = -1; private volatile long lastUnqueryableReportNanos = -1; @@ -100,7 +106,7 @@ public class IgniteClusterObserver { private IgnitePredicate membershipListener; private IgnitePredicate segmentationListener; private ScheduledExecutorService scheduler; - private ScheduledExecutorService inspector; + private ThreadPoolExecutor inspector; public IgniteClusterObserver(Ignite ignite, StructuresProperties structuresProperties) { this.ignite = ignite; @@ -111,8 +117,23 @@ public IgniteClusterObserver(Ignite ignite, StructuresProperties structuresPrope public void start() { startedAtNanos = System.nanoTime(); scheduler = newDaemonScheduler("structures-cluster-observer"); - // Inspections get their own thread so a slow cache read can never delay topology polling - inspector = newDaemonScheduler("structures-cluster-inspector"); + // Inspections run on their own bounded pool so a slow or wedged cache read can + // neither delay topology polling nor prevent later inspections from running. If + // every thread is stuck, further inspections are skipped and logged - a skipped + // inspection is itself a signal that cluster reads are hanging. + inspector = new ThreadPoolExecutor(INSPECTOR_THREADS, INSPECTOR_THREADS, + 0L, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(INSPECTOR_QUEUE_DEPTH), + runnable -> { + Thread thread = new Thread(runnable, "structures-cluster-inspector"); + thread.setDaemon(true); + return thread; + }, + (runnable, executor) -> log.warn( + "Skipping a routing inspection: previous inspections are " + + "still running, which usually means Ignite reads are " + + "blocked (activeInspections={})", + ((ThreadPoolExecutor) executor).getActiveCount())); // Logged inline, from data carried on the event itself. No cluster calls here: Ignite // notifies listeners on its monitored discovery worker, and this line must be on disk @@ -224,8 +245,13 @@ private void scheduleStaleRouteSamples(String departedNodeId) { boolean finalSample = i == STALE_ROUTE_SAMPLE_DELAYS_MS.length - 1; long delay = STALE_ROUTE_SAMPLE_DELAYS_MS[i]; try { - inspector.schedule(() -> reportStaleRoutingState(departedNodeId, departedAtNanos, finalSample), - delay, TimeUnit.MILLISECONDS); + scheduler.schedule(() -> { + try { + inspector.execute(() -> reportStaleRoutingState(departedNodeId, departedAtNanos, finalSample)); + } catch (RejectedExecutionException e) { + log.debug("Routing inspection not dispatched, observer is shutting down"); + } + }, delay, TimeUnit.MILLISECONDS); } catch (RejectedExecutionException e) { log.debug("Routing inspection not scheduled, observer is shutting down"); return; @@ -235,9 +261,16 @@ private void scheduleStaleRouteSamples(String departedNodeId) { /** * Inspect this node's local view of the vertx routing caches for entries that still - * reference a departed node. Local reads only, so this adds no distributed query load - * while the cluster is already rebalancing, and every Ignite call is time-bounded so a - * cluster hang cannot silence the observer. + * reference a departed node. + *

    + * Deliberately tolerant rather than precise. Reads are local and time-bounded, and any + * failure - a timeout, a cursor invalidated by the rebalance a departure triggers, a + * cache handle that is not ready - is reported and then simply retried by the next + * sample or the next departure. Transient problems resolve themselves that way; + * a genuinely stuck cluster shows up as the same complaint repeating, which is the + * signal worth acting on. Counts are this node's local view only (primary and backup + * copies, and entries in partitions being rebalanced), so the same entry can appear in + * more than one pod's log - correlate across pods rather than summing. */ private void reportStaleRoutingState(String departedNodeId, long departedAtNanos, boolean finalSample) { if (closed) { @@ -247,108 +280,125 @@ private void reportStaleRoutingState(String departedNodeId, long departedAtNanos // cluster is unhealthy, which is exactly when the timestamp matters long afterMs = elapsedMs(departedAtNanos); try { - // cacheNames() is a local metadata read; ignite.cache() on an unknown name can - // otherwise trigger a blocking cluster-wide dynamic cache start + // cacheNames() is a local metadata read; it also keeps us from asking for a + // cache this node has never seen Collection cacheNames = ignite.cacheNames(); boolean nodeInfoAvailable = cacheNames.contains(NODE_INFO_CACHE); boolean subsAvailable = cacheNames.contains(SUBS_CACHE); if (!nodeInfoAvailable && !subsAvailable) { - log.warn("Inconclusive routing inspection {} ms after departure of node {}: neither " - + "{} nor {} exists on this node, so no conclusion can be drawn about stale " - + "routing state", afterMs, departedNodeId, NODE_INFO_CACHE, SUBS_CACHE); + logInspectionIncomplete(finalSample, afterMs, departedNodeId, + "neither " + NODE_INFO_CACHE + " nor " + SUBS_CACHE + + " exists on this node", null); return; } - // Each half is independent: a missing subs cache must not cost us the nodeInfo - // verdict, which is the direct evidence of a failed cleanup election - Boolean nodeInfoPresent = nodeInfoAvailable ? checkNodeInfoPresent(departedNodeId) : null; - SubsScanResult subs = subsAvailable ? scanSubs(departedNodeId) : null; - - String coverage = String.format( - "nodeInfoChecked=%s, subsScanned=%s", - nodeInfoPresent != null ? "yes" : "no (unavailable or timed out)", - subs != null ? subs.describe() : "no (cache unavailable)"); - - boolean sawStale = Boolean.TRUE.equals(nodeInfoPresent) - || (subs != null && (subs.primaryStale > 0 || subs.backupStale > 0)); - boolean conclusive = nodeInfoPresent != null && subs != null && subs.conclusive(); - - if (sawStale) { + // Each half is independent: a failure in one must not cost the other. The + // nodeInfo verdict in particular is the direct evidence of a failed cleanup + // election, so it is reported whenever it can be obtained. + InspectionOutcome nodeInfo = nodeInfoAvailable + ? checkNodeInfoPresent(departedNodeId) + : InspectionOutcome.unavailable("cache not present on this node"); + InspectionOutcome subs = subsAvailable + ? scanSubs(departedNodeId) + : InspectionOutcome.unavailable("cache not present on this node"); + + boolean nodeInfoStale = Boolean.TRUE.equals(nodeInfo.value); + SubsScanResult scan = subs.value; + int staleSubs = scan != null ? scan.staleEntries : 0; + + String coverage = String.format("nodeInfoCheck=%s, subsScan=%s", + nodeInfo.describe(), + scan != null ? scan.describe() : subs.describe()); + + if (nodeInfoStale || staleSubs > 0) { String detail = String.format( - "nodeInfoStillPresent=%s, staleSubsOwnedHere=%d, staleSubsBackupCopiesHere=%d, " - + "sampleAddresses=%s, %s", - nodeInfoPresent, subs != null ? subs.primaryStale : -1, - subs != null ? subs.backupStale : -1, - subs != null ? subs.addresses : List.of(), coverage); + "nodeInfoStillPresent=%s, staleSubscriptionEntriesHere=%d, sampleAddresses=%s, %s", + nodeInfoStale, staleSubs, scan != null ? scan.addresses : List.of(), coverage); if (finalSample) { - log.warn("Stale routing state remains {} ms after departure of node {}: {}. " - + "staleSubsOwnedHere counts only partitions this node currently owns as " - + "primary, so it is safe to sum across pods; backup copies are reported " - + "separately and duplicate another pod's primary count. Event bus sends " - + "to these addresses can fail with 'Not a member of the cluster' until " - + "the handlers re-register.", + log.warn("Stale routing state remains {} ms after departure of node {}: {}. Counts are " + + "this node's local view and can overlap other pods, so correlate rather " + + "than sum. Event bus sends to these addresses can fail with 'Not a member " + + "of the cluster' until the handlers re-register.", afterMs, departedNodeId, detail); } else { log.info("Routing cleanup still in progress {} ms after departure of node {}: {}", afterMs, departedNodeId, detail); } - } else if (conclusive) { + } else if (nodeInfo.complete() && subs.complete()) { log.info("Routing caches are clean (local view) {} ms after departure of node {}: {}", afterMs, departedNodeId, coverage); } else { - log.warn("Inconclusive routing inspection {} ms after departure of node {}: no stale " - + "entries seen, but the inspection did not complete fully so this is NOT an " - + "all-clear ({})", afterMs, departedNodeId, coverage); + logInspectionIncomplete(finalSample, afterMs, departedNodeId, + "no stale entries seen, but the inspection did not complete " + + "fully so this is NOT an all-clear (" + coverage + ")", + null); } - } catch (Exception e) { + } catch (Throwable t) { + // Throwable, not Exception: an Error from Ignite internals (an iterator over a + // partition being evicted, for instance) must not vanish into the executor, + // leaving a silence that reads like a clean result if (closed) { - log.debug("Routing inspection for node {} abandoned during shutdown", departedNodeId, e); + log.debug("Routing inspection for node {} abandoned during shutdown", departedNodeId, t); } else { - // WARN, not DEBUG: a failed inspection must never be mistaken for a clean result - log.warn("Could not inspect routing caches {} ms after departure of node {}; " - + "no conclusion can be drawn about stale routing state", - afterMs, departedNodeId, e); + logInspectionIncomplete(finalSample, afterMs, departedNodeId, + "the inspection failed", t); } } } /** - * @return TRUE/FALSE if the check completed, or null if it could not be completed - * within the configured timeout - never guess, the caller reports it as inconclusive + * Incomplete inspections are expected while a cluster is rebalancing, so early samples + * report at INFO and only the final sample warns. Either way the reason is always + * visible at production log levels: the whole point is that a failed check is never + * mistaken for a clean one. */ - private Boolean checkNodeInfoPresent(String departedNodeId) { + private void logInspectionIncomplete(boolean finalSample, long afterMs, String departedNodeId, + String reason, Throwable cause) { + if (finalSample) { + log.warn("Inconclusive routing inspection {} ms after departure of node {}: {}", + afterMs, departedNodeId, reason, cause); + } else { + log.info("Routing inspection incomplete {} ms after departure of node {}: {} " + + "(will retry on the next sample)", + afterMs, departedNodeId, reason, cause); + } + } + + private InspectionOutcome checkNodeInfoPresent(String departedNodeId) { try { IgniteCache cache = ignite.cache(NODE_INFO_CACHE); if (cache == null) { - return null; + return InspectionOutcome.unavailable("cache handle not available"); } // Async with an explicit timeout: containsKey on a PARTITIONED cache is a // distributed read that can block on partition map exchange indefinitely, // which is precisely the condition being diagnosed - return cache.containsKeyAsync(departedNodeId) - .get(inspectionTimeoutMs(), TimeUnit.MILLISECONDS); - } catch (Exception e) { - log.debug("nodeInfo lookup for departed node {} did not complete", departedNodeId, e); - return null; + Boolean present = cache.containsKeyAsync(departedNodeId) + .get(inspectionTimeoutMs(), TimeUnit.MILLISECONDS); + return InspectionOutcome.complete(present); + } catch (Throwable t) { + return InspectionOutcome.failed(describeFailure(t)); } } - private SubsScanResult scanSubs(String departedNodeId) { + private InspectionOutcome scanSubs(String departedNodeId) { SubsScanResult result = new SubsScanResult(); - IgniteCache cache = ignite.cache(SUBS_CACHE); - if (cache == null) { - return null; - } - Affinity affinity = ignite.affinity(SUBS_CACHE); - ClusterNode localNode = ignite.cluster().localNode(); - long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(inspectionTimeoutMs()); - int maxEntries = maxEntriesScanned(); - - Iterable> entries = - cache.withKeepBinary().localEntries(CachePeekMode.PRIMARY, CachePeekMode.BACKUP); - Iterator> iterator = entries.iterator(); + Iterator> iterator = null; try { + IgniteCache cache = ignite.cache(SUBS_CACHE); + if (cache == null) { + return InspectionOutcome.unavailable("cache handle not available"); + } + long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(inspectionTimeoutMs()); + int maxEntries = maxEntriesScanned(); + + // Local reads only: no distributed query while the cluster is rebalancing. + // Both peek modes are used deliberately - a departure's stale entries can sit + // in either, and over-reporting on one pod is preferable to missing them. + iterator = cache.withKeepBinary() + .localEntries(CachePeekMode.PRIMARY, CachePeekMode.BACKUP) + .iterator(); while (iterator.hasNext()) { if (result.scanned >= maxEntries) { result.truncated = true; @@ -364,26 +414,27 @@ private SubsScanResult scanSubs(String departedNodeId) { result.unreadableKeys++; continue; } - if (!departedNodeId.equals(key.field("nodeId"))) { - continue; - } - // Ownership from the affinity function, not from the fact the data is here: - // partitions being rebalanced away still hold their old contents, and counting - // those would produce phantom findings during every rolling restart - if (affinity.isPrimary(localNode, key)) { - result.primaryStale++; + if (departedNodeId.equals(key.field("nodeId"))) { + result.staleEntries++; if (result.addresses.size() < MAX_STALE_ADDRESSES_LOGGED) { result.addresses.add(String.valueOf(key.field("address"))); } - } else if (affinity.isBackup(localNode, key)) { - result.backupStale++; } - // Entries in partitions this node no longer owns are deliberately ignored } + return InspectionOutcome.complete(result); + } catch (Throwable t) { + // A cursor invalidated by partition eviction is normal during the rebalance a + // departure triggers; report what was gathered and let the next sample retry + result.failure = describeFailure(t); + return InspectionOutcome.partial(result, result.failure); } finally { closeQuietly(iterator); } - return result; + } + + private static String describeFailure(Throwable t) { + return t.getClass().getSimpleName() + + (t.getMessage() != null ? ": " + t.getMessage() : ""); } /** @@ -415,8 +466,6 @@ private void checkTopology() { reportUnqueryableTopology(e); return; } - lastUnqueryableReportNanos = -1; - int minimumClusterSize = minimumClusterSize(); if (log.isDebugEnabled()) { @@ -446,7 +495,7 @@ private void checkTopology() { serverNodes, elapsedMs(belowMinimumSinceNanos), minimumClusterSize); } belowMinimumSinceNanos = -1; - belowMinimumReported = false; + lastBelowMinimumReportNanos = -1; return; } @@ -463,12 +512,18 @@ private void checkTopology() { } long belowForMs = elapsedMs(belowMinimumSinceNanos); - // One escalation per episode so a long orphan does not flood the log - if (!belowMinimumReported && belowForMs >= reportBelowMinimumAfterMs()) { - belowMinimumReported = true; + if (belowForMs < reportBelowMinimumAfterMs()) { + return; + } + // Repeated hourly rather than once per episode: a node orphaned days ago must still + // be visible in a recent log window, not only in one line from when it happened + if (lastBelowMinimumReportNanos == -1 + || elapsedMs(lastBelowMinimumReportNanos) >= BELOW_MINIMUM_REPORT_INTERVAL_MS) { + lastBelowMinimumReportNanos = System.nanoTime(); log.error("Server topology has been at {} nodes, below the minimum of {}, for {} ms. " + "This node is very likely orphaned or split-brained and would need a restart " - + "to rejoin the cluster", serverNodes, minimumClusterSize, belowForMs); + + "to rejoin the cluster (repeated at most every {} ms while it persists)", + serverNodes, minimumClusterSize, belowForMs, BELOW_MINIMUM_REPORT_INTERVAL_MS); } } @@ -587,25 +642,61 @@ private int maxEntriesScanned() { } /** - * Result of a local __vertx.subs scan. primaryStale is the cluster-summable count; - * backupStale duplicates another node's primary count and is kept separate. + * Outcome of one half of an inspection: whether it completed, and if not, why. Nothing + * is ever inferred from a failure - an incomplete half simply prevents an all-clear. + */ + private record InspectionOutcome(T value, boolean complete, String detail) { + + private static InspectionOutcome complete(T value) { + return new InspectionOutcome<>(value, true, "ok"); + } + + private static InspectionOutcome partial(T value, String detail) { + return new InspectionOutcome<>(value, false, detail); + } + + private static InspectionOutcome unavailable(String detail) { + return new InspectionOutcome<>(null, false, detail); + } + + private static InspectionOutcome failed(String detail) { + return new InspectionOutcome<>(null, false, detail); + } + + private String describe() { + return detail; + } + } + + /** + * What a local __vertx.subs scan saw. Counts are this node's local view (primary and + * backup copies, and partitions mid-rebalance), so they can overlap other pods. */ private static final class SubsScanResult { private int scanned; - private int primaryStale; - private int backupStale; + private int staleEntries; private int unreadableKeys; private boolean truncated; private boolean timedOut; + private String failure; private final List addresses = new ArrayList<>(); - private boolean conclusive() { - return !truncated && !timedOut && unreadableKeys == 0; - } - private String describe() { - return String.format("localEntriesScanned=%d, truncated=%b, timedOut=%b, unreadableKeys=%d", - scanned, truncated, timedOut, unreadableKeys); + StringBuilder sb = new StringBuilder(); + sb.append("localEntriesScanned=").append(scanned); + if (truncated) { + sb.append(", truncated=true"); + } + if (timedOut) { + sb.append(", timedOut=true"); + } + if (unreadableKeys > 0) { + sb.append(", unreadableKeys=").append(unreadableKeys); + } + if (failure != null) { + sb.append(", failed=").append(failure); + } + return sb.toString(); } } } From 60bc7c7834f787c4645fa49ed6fedbd84b968ea4 Mon Sep 17 00:00:00 2001 From: Nicholas Padilla Date: Thu, 30 Jul 2026 09:39:13 -0700 Subject: [PATCH 7/7] feat: make observer log cadence configurable The config surface was arbitrary: when a condition starts being reported was a property, but how often it repeats - and the routing sample schedule - were hardcoded. Since this ships in an immutable release and may run for days, promote the three tunables most likely to need adjusting in a specific environment: - repeatBelowMinimumEveryMs and repeatNeverReachedEveryMs (both 1h), pairing with the existing report-after settings - staleRouteSampleDelaysMs (5s/20s/60s): if cleanup legitimately takes longer than the window in a given environment, every final sample would warn, and that can now be widened without a release. Falls back to the defaults if configured empty, so the inspection can never be silently disabled Defaults are unchanged, so behavior is identical unless set. The remaining constants (poll interval, throttles, pool sizing, address sample limit) stay internal. Co-Authored-By: Claude Fable 5 --- .../api/config/ClusterObserverProperties.java | 24 +++++++++++ .../config/IgniteClusterObserver.java | 40 ++++++++++++++----- 2 files changed, 54 insertions(+), 10 deletions(-) diff --git a/structures-core/src/main/java/org/kinotic/structures/api/config/ClusterObserverProperties.java b/structures-core/src/main/java/org/kinotic/structures/api/config/ClusterObserverProperties.java index 00b9ea81..b6f93650 100644 --- a/structures-core/src/main/java/org/kinotic/structures/api/config/ClusterObserverProperties.java +++ b/structures-core/src/main/java/org/kinotic/structures/api/config/ClusterObserverProperties.java @@ -1,5 +1,7 @@ package org.kinotic.structures.api.config; +import java.util.List; + import lombok.Getter; import lombok.NoArgsConstructor; import lombok.Setter; @@ -34,6 +36,13 @@ public class ClusterObserverProperties { */ private Long reportBelowMinimumAfterMs = 60_000L; + /** + * How often the below-minimum condition is repeated while it persists. Repeating + * matters on long-running pods: a node orphaned days ago must still be visible in a + * recent log window, not only in a single line from when it happened. + */ + private Long repeatBelowMinimumEveryMs = 3_600_000L; + /** * How long a node may run without the topology ever reaching * {@link #getMinimumClusterSize()} before that is reported as a warning. Slow cluster @@ -51,6 +60,21 @@ public class ClusterObserverProperties { */ private Long escalateNeverReachedMinimumAfterMs = 900_000L; + /** + * How often the never-reached-minimum condition is repeated while it persists, for the + * same reason as {@link #getRepeatBelowMinimumEveryMs()}. + */ + private Long repeatNeverReachedEveryMs = 3_600_000L; + + /** + * Delays, in milliseconds after a node departs, at which the vertx routing caches are + * inspected for entries still referencing it. Sampling repeatedly distinguishes a + * cleanup that is merely slow from one that never completes; only the last sample + * warns. Widen the final delay if cleanup in your environment legitimately takes + * longer than the default window. + */ + private List staleRouteSampleDelaysMs = List.of(5_000L, 20_000L, 60_000L); + /** * Upper bound on any single Ignite read the observer performs. Bounded so a cluster * hang (the very condition being diagnosed) can never stall the observer itself. diff --git a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java index c52216aa..aa0806b8 100644 --- a/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java +++ b/structures-core/src/main/java/org/kinotic/structures/internal/config/IgniteClusterObserver.java @@ -79,16 +79,13 @@ public class IgniteClusterObserver { private static final long TOPOLOGY_POLL_MS = 10_000L; private static final long UNQUERYABLE_REPORT_INTERVAL_MS = 600_000L; private static final long UNEXPECTED_ERROR_REPORT_INTERVAL_MS = 600_000L; - private static final long NEVER_REACHED_REPORT_INTERVAL_MS = 3_600_000L; - private static final long BELOW_MINIMUM_REPORT_INTERVAL_MS = 3_600_000L; // A wedged Ignite read must never silence later inspections, so inspections run on a // small bounded pool: at worst a hung cluster consumes these threads and subsequent // inspections are skipped with a log line, which is itself evidence of the hang private static final int INSPECTOR_THREADS = 2; private static final int INSPECTOR_QUEUE_DEPTH = 8; private static final int MAX_STALE_ADDRESSES_LOGGED = 10; - /** Sampled repeatedly so a slow cleanup is not reported as a leak; only the final sample warns */ - private static final long[] STALE_ROUTE_SAMPLE_DELAYS_MS = {5_000L, 20_000L, 60_000L}; + private static final List DEFAULT_STALE_ROUTE_SAMPLE_DELAYS_MS = List.of(5_000L, 20_000L, 60_000L); private final Ignite ignite; private final ClusterObserverProperties properties; @@ -241,9 +238,10 @@ private void scheduleStaleRouteSamples(String departedNodeId) { return; } long departedAtNanos = System.nanoTime(); - for (int i = 0; i < STALE_ROUTE_SAMPLE_DELAYS_MS.length; i++) { - boolean finalSample = i == STALE_ROUTE_SAMPLE_DELAYS_MS.length - 1; - long delay = STALE_ROUTE_SAMPLE_DELAYS_MS[i]; + List sampleDelays = staleRouteSampleDelaysMs(); + for (int i = 0; i < sampleDelays.size(); i++) { + boolean finalSample = i == sampleDelays.size() - 1; + long delay = sampleDelays.get(i); try { scheduler.schedule(() -> { try { @@ -518,12 +516,12 @@ private void checkTopology() { // Repeated hourly rather than once per episode: a node orphaned days ago must still // be visible in a recent log window, not only in one line from when it happened if (lastBelowMinimumReportNanos == -1 - || elapsedMs(lastBelowMinimumReportNanos) >= BELOW_MINIMUM_REPORT_INTERVAL_MS) { + || elapsedMs(lastBelowMinimumReportNanos) >= repeatBelowMinimumEveryMs()) { lastBelowMinimumReportNanos = System.nanoTime(); log.error("Server topology has been at {} nodes, below the minimum of {}, for {} ms. " + "This node is very likely orphaned or split-brained and would need a restart " + "to rejoin the cluster (repeated at most every {} ms while it persists)", - serverNodes, minimumClusterSize, belowForMs, BELOW_MINIMUM_REPORT_INTERVAL_MS); + serverNodes, minimumClusterSize, belowForMs, repeatBelowMinimumEveryMs()); } } @@ -543,7 +541,7 @@ private void reportNeverReachedMinimum(int serverNodes, int minimumClusterSize) boolean firstEscalation = escalate && !neverReachedEscalated; if (!firstEscalation && lastNeverReachedReportNanos != -1 - && elapsedMs(lastNeverReachedReportNanos) < NEVER_REACHED_REPORT_INTERVAL_MS) { + && elapsedMs(lastNeverReachedReportNanos) < repeatNeverReachedEveryMs()) { return; } lastNeverReachedReportNanos = System.nanoTime(); @@ -633,6 +631,28 @@ private long escalateNeverReachedMinimumAfterMs() { ? properties.getEscalateNeverReachedMinimumAfterMs() : 900_000L; } + private long repeatBelowMinimumEveryMs() { + return properties.getRepeatBelowMinimumEveryMs() != null + ? properties.getRepeatBelowMinimumEveryMs() : 3_600_000L; + } + + private long repeatNeverReachedEveryMs() { + return properties.getRepeatNeverReachedEveryMs() != null + ? properties.getRepeatNeverReachedEveryMs() : 3_600_000L; + } + + /** + * Always returns at least one delay: an empty or missing list would otherwise silently + * disable the routing inspection entirely + */ + private List staleRouteSampleDelaysMs() { + List configured = properties.getStaleRouteSampleDelaysMs(); + if (configured == null || configured.isEmpty()) { + return DEFAULT_STALE_ROUTE_SAMPLE_DELAYS_MS; + } + return configured; + } + private long inspectionTimeoutMs() { return properties.getInspectionTimeoutMs() != null ? properties.getInspectionTimeoutMs() : 10_000L; }