diff --git a/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java b/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java index f8c6912ef7a3..4d1b440294fa 100644 --- a/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java +++ b/dotCMS/src/enterprise/java/com/dotcms/enterprise/cluster/ClusterFactory.java @@ -56,6 +56,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import org.elasticsearch.common.settings.Settings.Builder; @@ -64,7 +65,14 @@ public class ClusterFactory { private static boolean CLUSTER_INITED=false; - private static List KNOWN_SERVERS=Collections.EMPTY_LIST; + private static List KNOWN_SERVERS=Collections.emptyList(); + + /** + * Consecutive cluster cache-transport rewire failures. Reset to zero on the first + * successful rewire. Surfaced by the cache-transport health check and metrics so a + * persistently failing rewire is visible instead of being logged and forgotten. + */ + private static final AtomicLong REWIRE_FAILURES = new AtomicLong(0); private static final String VERIFY_REQUIRED_TABLES = "SELECT dc.cluster_id, dc.cluster_salt, cs.server_id, cs.cluster_id, cs.name, " @@ -343,8 +351,8 @@ public static synchronized void rewireClusterIfNeeded() { try{ List aliveServers = APILocator.getServerAPI().getAliveServers(); - if (!aliveServers.equals(KNOWN_SERVERS) || !aliveServers - .contains(APILocator.getServerAPI().getCurrentServer()) ) { + if (shouldRewire(aliveServers, KNOWN_SERVERS, + APILocator.getServerAPI().getCurrentServer(), REWIRE_FAILURES.get())) { rewireCluster(); } @@ -352,23 +360,58 @@ public static synchronized void rewireClusterIfNeeded() { catch(Exception e){ Logger.error(ClusterFactory.class, "Unable to rewire cluster:" + e.getMessage()); Logger.error(ClusterFactory.class, "servers:" + KNOWN_SERVERS); - - + + throw new DotRuntimeException(e); } } + + /** + * Decides whether {@link #rewireCluster()} should run on this heartbeat. + * + * Extracted from {@link #rewireClusterIfNeeded()} so it can be unit tested without the + * enterprise statics, {@code APILocator} and license state that method needs. Pure function of + * its arguments; behaviour is unchanged from the inline condition it replaces. + * + * A rewire is needed when any of the following holds: + * + * - {@code pendingFailures > 0} -- a previous rewire failed and must be retried. This is the + * clause added for issue #36803. The membership comparison below only fires when the + * alive-server set *changes*, so without this a rewire that failed while membership then + * settled back to {@code knownServers} would never be attempted again: the transport would + * stay broken, {@code REWIRE_FAILURES} could never return to zero, and the cache-transport + * health check would report unhealthy indefinitely. + * - cluster membership changed since the last successful rewire. + * - this server is missing from the alive set, so its own registration needs redoing. + * + * @param aliveServers servers currently seen as alive + * @param knownServers membership as of the last *successful* rewire + * @param currentServer this server + * @param pendingFailures consecutive rewire failures not yet cleared by a success + */ + @VisibleForTesting + static boolean shouldRewire(final List aliveServers, final List knownServers, + final Server currentServer, final long pendingFailures) { + + return pendingFailures > 0 + || !aliveServers.equals(knownServers) + || !aliveServers.contains(currentServer); + } public static void rewireCluster() throws Exception { if(clusterReady()) { - addMeToCacheIfNeeded(); - KNOWN_SERVERS = APILocator.getServerAPI().getAliveServers(); + if (addMeToCacheIfNeeded()) { + // only remember the alive-server set when the rewire actually succeeded, + // otherwise a failed transport init would never be retried + KNOWN_SERVERS = APILocator.getServerAPI().getAliveServers(); + } }else { Logger.info(ClusterFactory.class, "Cluster not yet active. Not rewiring"); } } - private static void addMeToCacheIfNeeded() throws DotDataException { + private static boolean addMeToCacheIfNeeded() throws DotDataException { if(isEnterprise()) { final Server localServer = APILocator.getServerAPI().getOrCreateMyServer(); @@ -378,16 +421,30 @@ private static void addMeToCacheIfNeeded() throws DotDataException { getImplementationObject()).setCluster(localServer); ((ChainableCacheAdministratorImpl) CacheLocator.getCacheAdministrator(). getImplementationObject()).testCluster(); + REWIRE_FAILURES.set(0); + return true; } catch (Exception e) { - Logger.error(ClusterFactory.class, e.getMessage(), e); + final long failures = REWIRE_FAILURES.incrementAndGet(); + Logger.error(ClusterFactory.class, "Unable to (re)initialize the cluster cache transport" + + " (consecutive failures: " + failures + + "). Cache invalidations are NOT reaching other nodes: " + e.getMessage(), e); + return false; } - - } else { - CacheLocator.getCacheAdministrator().getTransport().shutdown(); + CacheLocator.getCacheAdministrator().getTransport().shutdown(); + return true; } } + + /** + * Consecutive cluster cache-transport rewire failures, zero when the last rewire succeeded. + * + * @see #REWIRE_FAILURES + */ + public static long getRewireFailures() { + return REWIRE_FAILURES.get(); + } private static boolean isEnterprise() { diff --git a/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java b/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java index 6abd9521a38a..098a4fea5c62 100644 --- a/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java +++ b/dotCMS/src/main/java/com/dotcms/cache/transport/PubSubCacheTransport.java @@ -3,6 +3,7 @@ import java.io.Serializable; import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import com.dotcms.cache.transport.CacheTransportTopic.CacheEventType; import com.dotcms.cluster.bean.Server; import com.dotcms.dotpubsub.DotPubSubEvent; @@ -14,7 +15,9 @@ import com.dotmarketing.business.CacheLocator; import com.dotmarketing.business.cache.transport.CacheTransport; import com.dotmarketing.business.cache.transport.CacheTransportException; +import com.dotmarketing.util.Config; import com.dotmarketing.util.Logger; +import com.google.common.annotations.VisibleForTesting; import io.vavr.control.Try; /** @@ -31,19 +34,90 @@ public class PubSubCacheTransport implements CacheTransport { final AtomicBoolean initialized = new AtomicBoolean(false); + final AtomicLong droppedMessages = new AtomicLong(0); + + /** + * Invalidations dropped before this transport was ever initialized. + * + * Boot order guarantees some of these: caches are invalidated by startup tasks and by the + * starter import long before {@code ClusterFactory} wires the cluster and calls + * {@link #init(Server)}, and a node that is alone at that point has nobody to notify anyway. A + * first boot against an empty database was measured at ~2,800 of them. + * + * They are kept apart from {@link #droppedMessages} so the number an operator sees on a + * healthy node is zero rather than thousands of benign startup drops -- a cumulative counter + * dominated by boot noise is worthless as an alerting signal, which is the whole point of + * issue #36803. Reported separately for diagnostics. + */ + final AtomicLong startupDroppedMessages = new AtomicLong(0); + + /** + * Whether {@link #init(Server)} has ever succeeded, which ends the startup accounting for the + * life of this transport. Drops after that point were suffered by a transport that had been + * working and are genuine invalidation loss, so a later re-init must not retire them. + */ + private final AtomicBoolean everInitialized = new AtomicBoolean(false); + + /** + * Sends that were attempted and reported as failed by a synchronous provider. Failures from + * an asynchronous provider are counted by the provider itself and read back in + * {@link #getFailedMessages()}; see {@link DotPubSubProvider#getFailedPublishCount(String)}. + */ + final AtomicLong failedMessages = new AtomicLong(0); + + private final AtomicLong lastDropWarnAt = new AtomicLong(0); + + private final AtomicLong lastFailWarnAt = new AtomicLong(0); + + private static final long DROP_WARN_INTERVAL_MILLIS = + Config.getLongProperty("CACHE_TRANSPORT_DROP_WARN_INTERVAL_MILLIS", 30000); + @Override public boolean requiresAutowiring() { return false; } public PubSubCacheTransport() { - this.pubsub = DotPubSubProviderLocator.provider.get(); - this.topic = new CacheTransportTopic(); + this(DotPubSubProviderLocator.provider.get(), new CacheTransportTopic()); + } + + @VisibleForTesting + PubSubCacheTransport(final DotPubSubProvider pubsub, final CacheTransportTopic topic) { + this.pubsub = pubsub; + this.topic = topic; Logger.debug(this.getClass(), "PubSubCacheTransport"); } + /** + * {@inheritDoc} + * + * {@code synchronized} so the guard below is a genuine test-and-set rather than a + * check-then-act: two threads could otherwise both read {@code initialized == false} and each + * run {@code start()} + {@code subscribe()}, double-subscribing the listener -- the opposite of + * the churn this transport was changed to remove. + * + * Today every path here is already serialized -- {@code init()} is reached only through + * {@code ChainableCacheAdministratorImpl.setCluster()} <- {@code addMeToCacheIfNeeded()} <- + * {@code rewireCluster()} <- {@code ClusterFactory.rewireClusterIfNeeded()}, which is + * {@code static synchronized}, and each of those has exactly one call site. This guards the + * future: {@code rewireCluster()} is {@code public static} and not itself synchronized, so a + * new caller could bypass the lock that currently makes the race unreachable. The method runs + * once per cluster rewire, so the monitor costs nothing. + * + * Note that {@code initialized} is deliberately set only *after* {@code start()} and + * {@code subscribe()} have returned. Marking it up front (for instance via a + * {@code compareAndSet} guard) would leave a thrown {@code start()} permanently flagged as + * initialized: {@link #isInitialized()} would report true, {@link #shouldReinit()} would report + * false, {@code setCluster()} would never retry, and the health check would report a healthy + * transport that never subscribed -- exactly the silent failure issue #36803 removes. + */ @Override - public void init(final Server localServer) throws CacheTransportException { + public synchronized void init(final Server localServer) throws CacheTransportException { + + if (this.initialized.get()) { + Logger.debug(this.getClass(), "PubSubCacheTransport already initialized, skipping re-init"); + return; + } Logger.info(this.getClass(), "initing PubSubCacheTransport"); this.pubsub.start(); @@ -51,21 +125,78 @@ public void init(final Server localServer) throws CacheTransportException { this.initialized.set(true); + retireStartupDrops(); + } + + /** + * On the first successful init only, moves the drops accumulated so far into + * {@link #startupDroppedMessages}, so {@link #getDroppedMessages()} counts only invalidations + * lost while the transport was expected to be carrying them. + * + * Deliberately first-init-only. A transport that came up, went down, dropped invalidations and + * recovered has lost real ones, and retiring those on every re-init would launder genuine loss + * into the benign startup bucket -- exactly the blind spot issue #36803 exists to remove. + * + * The drop-warning throttle is cleared on every init: a genuine drop minutes later must log + * immediately rather than be swallowed because an earlier drop consumed the window. + * + * A {@code send()} racing with this can have its increment land in either bucket. The total is + * preserved either way, and both counters are diagnostics rather than exact accounting. + */ + private void retireStartupDrops() { + + if (this.everInitialized.compareAndSet(false, true)) { + final long startupDrops = this.droppedMessages.getAndSet(0); + if (startupDrops > 0) { + this.startupDroppedMessages.addAndGet(startupDrops); + Logger.info(this.getClass(), "Cache transport initialized. " + startupDrops + + " cache invalidation(s) were dropped before it came up (expected during" + + " startup, when there is no cluster to notify yet)."); + } + } + this.lastDropWarnAt.set(0); } @Override public void send(final String message) throws CacheTransportException { if (!this.initialized.get()) { + final long dropped = this.droppedMessages.incrementAndGet(); + warnThrottled(this.lastDropWarnAt, + "Cache transport is not initialized - dropping cluster cache invalidations. " + + "Other nodes may serve stale content. Total dropped: " + dropped); return; } final DotPubSubEvent event = new DotPubSubEvent.Builder().withTopic(this.topic) .withType(CacheEventType.INVAL.name()).withMessage(message).build(); - this.pubsub.publish(event); + // The boolean matters: every provider signals a failed send by returning false rather + // than throwing, so ignoring it (as this did before issue #36803) made a transport that + // was initialized but failing every publish completely invisible - no drops recorded, + // isInitialized() still true, health checks green, other nodes stale. Asynchronous + // providers cannot answer here and return true immediately; their failures are counted + // provider-side and picked up by getFailedMessages(). + if (!this.pubsub.publish(event)) { + final long failed = this.failedMessages.incrementAndGet(); + warnThrottled(this.lastFailWarnAt, + "Cache transport failed to publish cluster cache invalidations. " + + "Other nodes may serve stale content. Total failed: " + failed); + } } + /** + * Logs at most one warning per {@link #DROP_WARN_INTERVAL_MILLIS} for the given throttle, so + * a sustained failure does not flood the log at invalidation rate. + */ + private void warnThrottled(final AtomicLong lastWarnAt, final String message) { + final long now = System.currentTimeMillis(); + final long lastWarn = lastWarnAt.get(); + if (now - lastWarn > DROP_WARN_INTERVAL_MILLIS && lastWarnAt.compareAndSet(lastWarn, now)) { + Logger.warn(this.getClass(), message); + } + } + @Override public void testCluster() throws CacheTransportException { @@ -104,8 +235,16 @@ public Map validateCacheInCluster(final int maxWaitInMilli return this.topic.readResponses(); } + /** + * {@inheritDoc} + * + * Shares {@link #init(Server)}'s monitor so the two cannot interleave. Without it a shutdown + * landing between {@code init()}'s {@code subscribe()} and its {@code initialized.set(true)} + * would stop the provider and then be overwritten back to initialized, leaving a transport that + * reports itself up with nothing listening. + */ @Override - public void shutdown() throws CacheTransportException { + public synchronized void shutdown() throws CacheTransportException { Logger.debug(this.getClass(), "shutdown()"); this.pubsub.stop(); if (initialized.get()) { @@ -125,6 +264,35 @@ public boolean shouldReinit() { return !initialized.get(); } + /** + * {@inheritDoc} + * + * Counts only drops recorded after the transport was first initialized; see + * {@link #startupDroppedMessages} for the benign pre-init ones. + */ + @Override + public long getDroppedMessages() { + return droppedMessages.get(); + } + + @Override + public long getStartupDroppedMessages() { + return startupDroppedMessages.get(); + } + + /** + * {@inheritDoc} + * + * Sums the failures this transport observed synchronously with those the provider recorded on + * its own thread. Exactly one of the two counts a given attempt: a synchronous provider + * returns the real boolean to {@link #send(String)} and reports 0 here, while an + * asynchronous one returns true immediately and counts the outcome itself. + */ + @Override + public long getFailedMessages() { + return failedMessages.get() + this.pubsub.getFailedPublishCount(this.topic.getTopic()); + } + @Override public CacheTransportInfo getInfo() { diff --git a/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java b/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java index 8921c8d42aa9..80959b6afd96 100644 --- a/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java +++ b/dotCMS/src/main/java/com/dotcms/dotpubsub/DotPubSubProvider.java @@ -70,6 +70,33 @@ default DotPubSubEvent lastEventOut() { default String getProviderName() { return getClass().getSimpleName(); } + + /** + * Number of publish attempts for the given topic that this provider knows to have failed, + * i.e. where {@link #publish(DotPubSubEvent)} ultimately returned {@code false}. + * + * Only meaningful for providers that publish asynchronously and therefore cannot report the + * outcome through their own return value -- {@link QueuingPubSubWrapper} is the one that + * does, returning {@code true} immediately and completing the real publish on another + * thread. Callers that publish synchronously already see the {@code boolean} and should + * count it themselves; they must not add this value on top of their own count for the same + * attempt, or one failure is counted twice. + * + * Counted per topic because a single provider instance is shared by every topic in the JVM + * (cache invalidation, OSGi restart, cluster management), so a JVM-wide total could not be + * attributed to the cache transport. + * + * The topic key is matched case-insensitively, so a caller may pass either + * {@link DotPubSubEvent#getTopic()} (already lowercased by the event builder) or + * {@link DotPubSubTopic#getTopic()} (not normalized) and get the same answer. Implementations + * that key their own storage by topic must normalize it the same way. + * + * @param topic the topic key, as returned by {@link DotPubSubEvent#getTopic()} + * @return the failure count, or 0 for providers that publish synchronously + */ + default long getFailedPublishCount(final String topic) { + return 0; + } diff --git a/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java b/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java index 633ad5907db2..b9f8d1dd5284 100644 --- a/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java +++ b/dotCMS/src/main/java/com/dotcms/dotpubsub/QueuingPubSubWrapper.java @@ -1,7 +1,10 @@ package com.dotcms.dotpubsub; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import com.dotcms.concurrent.DotConcurrentFactory; import com.dotcms.concurrent.DotConcurrentFactory.SubmitterConfig; import com.dotcms.concurrent.DotSubmitter; @@ -22,8 +25,16 @@ public class QueuingPubSubWrapper implements DotPubSubProvider { private final DotPubSubProvider wrappedProvider; private final Cache recentEvents ; - + private final DotSubmitter submitter; + + /** + * Failed publish attempts per topic. Keyed by topic because one provider instance serves + * every topic in the JVM, so a single total could not be attributed to the cache transport. + * + * Keys are normalized through {@link #topicKey(String)}; see that method for why. + */ + private final Map failedByTopic = new ConcurrentHashMap<>(); @@ -98,9 +109,9 @@ public boolean publish(final DotPubSubEvent event) { sent++; recentEvents.put(event.hashCode(), true); - submitter.submit(()->this.wrappedProvider.publish(event)); - - + submitter.submit(()->publishAndRecordOutcome(event)); + + Logger.debug(this.getClass(), ()->"sent/skipped: " + sent + "/" + skipped); Logger.debug(this.getClass(), ()->"active count:" + submitter.getActiveCount()); @@ -109,6 +120,68 @@ public boolean publish(final DotPubSubEvent event) { } + /** + * Runs the real publish on the submitter thread and records the outcome. + * + * {@link #publish(DotPubSubEvent)} has to return before this completes, so it always returns + * {@code true} and the caller never learns whether the send actually worked. Before issue + * #36803 the result of this task was discarded entirely, which meant a transport that was + * initialized but failing every publish -- a dropped database connection, a stopped Redis + * client -- reported no drops, stayed "initialized", and kept every health check green while + * other nodes served stale content. + */ + private void publishAndRecordOutcome(final DotPubSubEvent event) { + try { + if (!this.wrappedProvider.publish(event)) { + recordFailure(event); + } + } catch (Exception e) { + // providers are expected to swallow their own failures and return false; this is + // the belt-and-braces path so an unexpected throw is still counted, not lost + recordFailure(event); + Logger.warnAndDebug(this.getClass(), + "Unable to publish pubsub event " + event + " : " + e.getMessage(), e); + } + } + + private void recordFailure(final DotPubSubEvent event) { + failedByTopic.computeIfAbsent(topicKey(String.valueOf(event.getTopic())), + t -> new AtomicLong(0)).incrementAndGet(); + } + + /** + * Normalizes a topic into the key {@link #failedByTopic} is indexed by. + * + * Writes and reads reach this map by different routes that do not agree on case. + * {@code DotPubSubEvent.Builder.withTopic} lowercases, so anything recorded from an event is + * already lowercase; the read side goes through {@code DotPubSubTopic.getTopic()}, which + * returns {@code String.valueOf(getKey())} with no normalization at all. Those happen to match + * today only because every current key is already lowercase + * ({@code CacheTransportTopic.CACHE_TOPIC} is {@code "dotcache_topic"}). + * + * A future topic key with an uppercase character would make the lookup miss silently, and + * {@link #getFailedPublishCount(String)} would report zero failures while invalidations were + * being lost -- precisely the kind of blind spot issue #36803 exists to remove. Normalizing on + * both sides here keeps the map self-consistent regardless of what a caller passes, rather than + * relying on every caller knowing this convention. + */ + private static String topicKey(final String topic) { + return topic == null ? "null" : topic.toLowerCase(); + } + + /** + * {@inheritDoc} + * + * Returns only this wrapper's own count. The wrapped provider published synchronously from + * the submitter thread, so its {@code false} return has already been counted here -- adding + * {@code wrappedProvider.getFailedPublishCount(topic)} would double count. + */ + @Override + public long getFailedPublishCount(final String topic) { + final AtomicLong failed = failedByTopic.get(topicKey(topic)); + return failed == null ? 0 : failed.get(); + } + @Override public DotPubSubProvider unsubscribe(DotPubSubTopic topic) { diff --git a/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java b/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java new file mode 100644 index 000000000000..fb70c7581d6a --- /dev/null +++ b/dotCMS/src/main/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheck.java @@ -0,0 +1,325 @@ +package com.dotcms.health.checks.cdi; + +import com.dotcms.enterprise.cluster.ClusterFactory; +import com.dotcms.health.config.HealthCheckConfig.HealthCheckMode; +import com.dotcms.health.model.HealthStatus; +import com.dotcms.health.util.HealthCheckBase; +import com.dotmarketing.business.CacheLocator; +import com.dotmarketing.business.cache.transport.CacheTransport; +import com.dotmarketing.business.cache.transport.NullTransport; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import javax.enterprise.context.ApplicationScoped; + +/** + * Health check for the cluster cache-invalidation transport (pub/sub). + * + * A node whose cache transport is not initialized silently drops every cache invalidation it + * tries to send, so other nodes serve stale content while all standard probes report healthy + * (see issue #36803 / incident #36544). This check surfaces that state: + * + * - transport not initialized -> unhealthy, once past the startup grace period + * (see {@link #DEFAULT_INITIALIZATION_GRACE_SECONDS}) + * - cluster rewire failing repeatedly -> unhealthy (see {@link #DEFAULT_REWIRE_FAILURE_THRESHOLD}) + * - dropped and failed invalidation counts exposed in structured data for monitoring + * + * Failed invalidations (initialized transport whose sends are erroring) are reported but do not + * on their own make the check unhealthy: the count is cumulative and never resets, so alarming on + * "greater than zero" would pin a node DOWN forever after one transient publish error. Alert on + * the rate of increase of {@code dotcms.cache.transport.invalidations.failed} instead. + * + * Nodes with no real transport -- {@link NullTransport} via + * {@code CACHE_INVALIDATION_TRANSPORT_CLASS}, or no resolvable cache layer at all -- always + * report healthy, since there is nothing that could be dropping invalidations. + * + * Readiness: by default this check runs in MONITOR_MODE, so it reports degradation but never + * fails readiness probes -- a node that cannot send invalidations can still serve traffic, and + * gating readiness on the transport could deadlock a cold cluster start where the transport + * initializes after traffic begins. Operators who prefer to drain such nodes can opt in with + * {@code health.check.cache-transport.mode=PRODUCTION}. + * + * Configuration Properties: + * - health.check.cache-transport.mode = Safety mode (PRODUCTION, MONITOR_MODE, DISABLED) + * - health.check.cache-transport.rewire-failure-threshold = consecutive rewire failures + * tolerated before reporting unhealthy (default 3) + * - health.check.cache-transport.initialization-grace-period-seconds = how long a + * never-yet-initialized transport is reported as still starting up (default 120) + */ +@ApplicationScoped +public class CacheTransportHealthCheck extends HealthCheckBase { + + /** + * Consecutive rewire failures tolerated before this check reports unhealthy. + * + * A single failure is usually a transient blip: {@code testCluster()} can throw on a + * momentary database hiccup while the transport itself stays initialized and invalidations + * keep flowing. Alarming on the first failure would report DOWN for a working node purely + * from a counter that has not been reset yet. {@code ClusterFactory.rewireClusterIfNeeded()} + * retries on every server heartbeat (60s by default) and zeroes the counter on success, so + * reaching this threshold means the rewire has been failing for minutes, not milliseconds. + */ + private static final int DEFAULT_REWIRE_FAILURE_THRESHOLD = 3; + + /** + * How long a transport that has never been initialized is treated as still starting up. + * + * The transport is initialized late in boot, when {@code ClusterFactory} wires the cluster, so + * every node necessarily reports an uninitialized transport for the first seconds of its life. + * Failing the check there would turn a normal rolling deployment into a DEGRADED overall + * status on every pod start -- noise that costs an operator's trust in the signal long before + * it ever catches the real fault from issue #36803. A node genuinely stuck without a transport + * stays stuck, so waiting this out loses nothing but the first two minutes of detection. + * + * Only applies before the first successful init. Once this check has seen an initialized + * transport, losing it is a regression and is reported immediately. + */ + private static final int DEFAULT_INITIALIZATION_GRACE_SECONDS = 120; + + /** + * Latched the first time an initialized transport is observed, which ends the startup grace + * period for the lifetime of the JVM. + */ + private final AtomicBoolean everInitialized = new AtomicBoolean(false); + + /** + * When the current run of uninitialized observations began, or 0 if the last observation was + * healthy. Used to measure the grace period from the first poll that saw the problem rather + * than from JVM start, so the window is not consumed by a slow boot before this check runs. + */ + private final AtomicLong uninitializedSince = new AtomicLong(0); + + /** + * Transport state resolved once per {@code check()} invocation. + * + * {@code HealthCheckBase.check()} calls {@link #performCheck()} and then + * {@link #buildStructuredData} sequentially on the same thread, so the snapshot is handed + * between them here instead of resolving the transport and re-reading its counters twice. + * Beyond the wasted work, two independent reads let a single health response contradict + * itself -- a message saying the transport is initialized alongside structured data saying + * it is not, because the state moved in between. A ThreadLocal (rather than a field) keeps + * concurrent liveness/readiness polls of this {@code @ApplicationScoped} bean isolated, and + * {@link #buildStructuredData} always clears it. + */ + private static final ThreadLocal CURRENT_SNAPSHOT = new ThreadLocal<>(); + + @Override + protected CheckResult performCheck() throws Exception { + + if (isShutdownInProgress()) { + CURRENT_SNAPSHOT.set(TransportSnapshot.unavailable(0L)); + return new CheckResult(true, 0L, "Cache transport check skipped during shutdown"); + } + + final TransportSnapshot snapshot = snapshot(); + CURRENT_SNAPSHOT.set(snapshot); + + if (!snapshot.present) { + return new CheckResult(true, 0L, + "No cluster cache transport in use (" + snapshot.transportName + ")"); + } + + if (!snapshot.initialized) { + return uninitialized(snapshot); + } + + everInitialized.set(true); + uninitializedSince.set(0); + + final int threshold = getConfigProperty("rewire-failure-threshold", + DEFAULT_REWIRE_FAILURE_THRESHOLD); + + if (snapshot.rewireFailures >= threshold) { + return new CheckResult(false, 0L, + "Cluster rewire has failed " + snapshot.rewireFailures + + " consecutive times (threshold: " + threshold + + ") - the cache transport may be stale"); + } + + return new CheckResult(true, 0L, + "Cache transport initialized (" + snapshot.transportName + + ", dropped invalidations: " + snapshot.dropped + + ", failed invalidations: " + snapshot.failed + + ", consecutive rewire failures: " + snapshot.rewireFailures + ")"); + } + + /** + * Decides whether an uninitialized transport is a node that is still starting up or a node + * that is silently dropping invalidations. See {@link #DEFAULT_INITIALIZATION_GRACE_SECONDS}. + */ + private CheckResult uninitialized(final TransportSnapshot snapshot) { + + final long now = System.currentTimeMillis(); + uninitializedSince.compareAndSet(0, now); + + if (!everInitialized.get()) { + + final long graceMillis = TimeUnit.SECONDS.toMillis(getConfigProperty( + "initialization-grace-period-seconds", DEFAULT_INITIALIZATION_GRACE_SECONDS)); + final long waitingMillis = now - uninitializedSince.get(); + + if (waitingMillis < graceMillis) { + return new CheckResult(true, 0L, + "Cache transport is still initializing (" + waitingMillis / 1000 + + "s of " + graceMillis / 1000 + "s grace period, dropped so far: " + + snapshot.dropped + ")"); + } + + return new CheckResult(false, 0L, + "Cache transport has NOT initialized within " + graceMillis / 1000 + + "s of startup - cluster cache invalidations are being dropped" + + " (dropped so far: " + snapshot.dropped + + ", consecutive rewire failures: " + snapshot.rewireFailures + ")"); + } + + return new CheckResult(false, 0L, + "Cache transport was initialized and is NOT anymore - cluster cache invalidations" + + " are being dropped (dropped so far: " + snapshot.dropped + + ", consecutive rewire failures: " + snapshot.rewireFailures + ")"); + } + + /** + * Reads every piece of transport state this check reports on, exactly once. + */ + private TransportSnapshot snapshot() { + + final long rewireFailures = ClusterFactory.getRewireFailures(); + final CacheTransport transport = getTransport(); + + if (transport == null) { + return TransportSnapshot.unavailable(rewireFailures); + } + + // NullTransport is a deliberate no-op transport (single node, or + // CACHE_INVALIDATION_TRANSPORT_CLASS pointed at it). Its isInitialized() is false + // whenever it has been shut down, which must not be read as dropped invalidations. + final boolean present = !(transport instanceof NullTransport); + + return new TransportSnapshot(transport.getClass().getSimpleName(), present, + transport.isInitialized(), transport.getDroppedMessages(), + transport.getStartupDroppedMessages(), transport.getFailedMessages(), + rewireFailures); + } + + /** + * Resolves the transport through the {@code DotCacheAdministrator} interface. + * + * Deliberately not {@code getImplementationObject()} plus a cast to + * {@code ChainableCacheAdministratorImpl}: {@code getTransport()} is part of the interface + * and {@code CommitListenerCacheWrapper} delegates it straight through, whereas the cast + * throws {@code ClassCastException} for any other administrator implementation -- + * {@code NullCacheAdministrator.getImplementationObject()} returns itself, which is the + * unit-test path. Compare {@code ClusterResource}, which uses the interface method. + */ + private CacheTransport getTransport() { + try { + return CacheLocator.getCacheAdministrator().getTransport(); + } catch (Exception e) { + // cache layer not up yet - nothing to report on + return null; + } + } + + @Override + public String getName() { + return "cache-transport"; + } + + @Override + protected HealthCheckMode getDefaultMode() { + return HealthCheckMode.MONITOR_MODE; + } + + @Override + public int getOrder() { + return 35; // right after the cache check + } + + /** + * Never liveness - a broken transport must not restart pods (that made #36544 worse) + */ + @Override + public boolean isLivenessCheck() { + return false; + } + + /** + * Participates in readiness reporting, but the default MONITOR_MODE means it will not + * fail probes unless an operator opts in to PRODUCTION mode. + */ + @Override + public boolean isReadinessCheck() { + return getMode() != HealthCheckMode.DISABLED; + } + + @Override + public String getDescription() { + return String.format( + "Verifies the cluster cache-invalidation transport is initialized and not dropping messages (Mode: %s)", + getMode().name()); + } + + @Override + protected Map buildStructuredData(CheckResult result, HealthStatus originalStatus, + HealthStatus finalStatus, HealthCheckMode mode) { + try { + TransportSnapshot snapshot = CURRENT_SNAPSHOT.get(); + if (snapshot == null) { + // performCheck() threw before it could snapshot - resolve once here instead + snapshot = snapshot(); + } + + final Map data = new HashMap<>(); + data.put("transport", snapshot.transportName); + data.put("rewireFailures", snapshot.rewireFailures); + if (snapshot.present) { + data.put("initialized", snapshot.initialized); + data.put("droppedInvalidations", snapshot.dropped); + data.put("startupDroppedInvalidations", snapshot.startupDropped); + data.put("failedInvalidations", snapshot.failed); + // Lets a consumer tell "initialized:false but still booting, reported UP" apart + // from "initialized:false and past the grace period, reported DOWN". + data.put("awaitingInitialization", !snapshot.initialized && !everInitialized.get()); + } + return data; + } finally { + CURRENT_SNAPSHOT.remove(); + } + } + + /** + * Immutable, single-read view of the transport state, so the message and the structured + * data of one health response can never disagree. + */ + private static final class TransportSnapshot { + + private static final String NO_TRANSPORT = "none"; + + final String transportName; + /** true only for a real transport that is expected to carry invalidations */ + final boolean present; + final boolean initialized; + final long dropped; + final long startupDropped; + final long failed; + final long rewireFailures; + + TransportSnapshot(final String transportName, final boolean present, final boolean initialized, + final long dropped, final long startupDropped, final long failed, + final long rewireFailures) { + this.transportName = transportName; + this.present = present; + this.initialized = initialized; + this.dropped = dropped; + this.startupDropped = startupDropped; + this.failed = failed; + this.rewireFailures = rewireFailures; + } + + static TransportSnapshot unavailable(final long rewireFailures) { + return new TransportSnapshot(NO_TRANSPORT, false, false, 0L, 0L, 0L, rewireFailures); + } + } +} diff --git a/dotCMS/src/main/java/com/dotcms/health/providers/CoreHealthCheckProvider.java b/dotCMS/src/main/java/com/dotcms/health/providers/CoreHealthCheckProvider.java index 1ad6080490af..8b47f2d116b7 100644 --- a/dotCMS/src/main/java/com/dotcms/health/providers/CoreHealthCheckProvider.java +++ b/dotCMS/src/main/java/com/dotcms/health/providers/CoreHealthCheckProvider.java @@ -3,6 +3,7 @@ import com.dotcms.health.api.HealthCheck; import com.dotcms.health.api.HealthCheckProvider; import com.dotcms.health.checks.cdi.CacheHealthCheck; +import com.dotcms.health.checks.cdi.CacheTransportHealthCheck; import com.dotcms.health.checks.cdi.DatabaseHealthCheck; import com.dotcms.health.checks.cdi.ElasticsearchHealthCheck; import com.dotcms.health.checks.cdi.VelocityHealthCheck; @@ -25,6 +26,7 @@ public List getHealthChecks() { return Arrays.asList( new DatabaseHealthCheck(), new CacheHealthCheck(), + new CacheTransportHealthCheck(), new ElasticsearchHealthCheck(), new VelocityHealthCheck() // Additional dependency health checks can be added here diff --git a/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java b/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java index 8520d852750e..f9d00a0d2711 100644 --- a/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java +++ b/dotCMS/src/main/java/com/dotcms/metrics/binders/CacheMetrics.java @@ -1,9 +1,12 @@ package com.dotcms.metrics.binders; +import com.dotcms.enterprise.cluster.ClusterFactory; import com.dotmarketing.business.CacheLocator; import com.dotmarketing.business.DotCacheAdministrator; import com.dotmarketing.business.cache.provider.CacheProviderStats; import com.dotmarketing.business.cache.provider.CacheStats; +import com.dotmarketing.business.cache.transport.CacheTransport; +import com.dotmarketing.business.cache.transport.NullTransport; import com.dotmarketing.util.Logger; import com.dotmarketing.util.UtilMethods; import io.micrometer.core.instrument.Gauge; @@ -11,6 +14,7 @@ import io.micrometer.core.instrument.binder.MeterBinder; import java.util.List; +import java.util.Optional; /** * Comprehensive metric binder for dotCMS cache-related metrics. @@ -40,6 +44,7 @@ public void bindTo(MeterRegistry registry) { if (cacheAdmin != null) { registerProviderLevelMetrics(registry, cacheAdmin); registerRegionLevelMetrics(registry, cacheAdmin); + registerTransportMetrics(registry); } Logger.info(this, "Comprehensive cache metrics registered successfully"); @@ -49,6 +54,69 @@ public void bindTo(MeterRegistry registry) { } } + /** + * Register cluster cache-transport metrics (issue #36803): silent invalidation drops + * and persistent rewire failures were previously invisible to monitoring. + */ + private void registerTransportMetrics(MeterRegistry registry) { + + Gauge.builder(METRIC_PREFIX + ".transport.invalidations.dropped", this, + m -> activeTransport().map(t -> (double) t.getDroppedMessages()).orElse(0.0)) + .description("Cluster cache invalidations dropped, after the transport first came up," + + " because it was not initialized") + .register(registry); + + Gauge.builder(METRIC_PREFIX + ".transport.invalidations.dropped.startup", this, + m -> activeTransport().map(t -> (double) t.getStartupDroppedMessages()).orElse(0.0)) + .description("Cluster cache invalidations dropped before the transport was first" + + " initialized (expected during startup, not an alerting signal)") + .register(registry); + + Gauge.builder(METRIC_PREFIX + ".transport.invalidations.failed", this, + m -> activeTransport().map(t -> (double) t.getFailedMessages()).orElse(0.0)) + .description("Cluster cache invalidations the transport attempted but failed to publish") + .register(registry); + + // Reported raw, with no startup grace period: a gauge's job is to state the current + // fact, and every node reads 0 here for the first seconds of its life. Alert on it with + // a duration clause (Prometheus "for: 2m") rather than on the instantaneous value, or + // alert on the cache-transport health check, which applies the grace period itself. + Gauge.builder(METRIC_PREFIX + ".transport.initialized", this, + m -> activeTransport().map(t -> t.isInitialized() ? 1.0 : 0.0).orElse(1.0)) + .description("Whether the cluster cache transport is initialized (1) or dropping invalidations (0)") + .register(registry); + + Gauge.builder(METRIC_PREFIX + ".transport.rewire.failures", this, + m -> (double) ClusterFactory.getRewireFailures()) + .description("Consecutive cluster cache-transport rewire failures (reset on success)") + .register(registry); + } + + /** + * The cache transport, but only when it is one that actually carries cluster invalidations. + * + * Returns empty for {@link NullTransport} as well as for an unresolvable transport, so the + * gauges above fall back to their "nothing to report" defaults. This matters for + * {@code transport.initialized}: {@code NullTransport.isInitialized()} is false whenever it + * has been shut down, so without this guard every node configured with no real transport + * would publish {@code transport.initialized=0} and alert as if it were dropping + * invalidations -- while {@code CacheTransportHealthCheck} reported the same node healthy. + * + * Uses the {@code DotCacheAdministrator.getTransport()} interface method rather than + * casting {@code getImplementationObject()} to {@code ChainableCacheAdministratorImpl}, + * which throws {@code ClassCastException} for any other administrator implementation. + */ + private Optional activeTransport() { + try { + final CacheTransport transport = CacheLocator.getCacheAdministrator().getTransport(); + return transport == null || transport instanceof NullTransport + ? Optional.empty() + : Optional.of(transport); + } catch (Exception e) { + return Optional.empty(); + } + } + /** * Register provider-level aggregate metrics. */ diff --git a/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java b/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java index 50ceaa210d56..e4adbedf866b 100644 --- a/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java +++ b/dotCMS/src/main/java/com/dotmarketing/business/cache/transport/CacheTransport.java @@ -56,6 +56,41 @@ public interface CacheTransport { default boolean requiresAutowiring() { return true; } + + /** + * Number of cache invalidation messages this transport has dropped because it was asked to + * send while not initialized, counted only from the first successful {@link #init(Server)} + * onwards. Used by health checks and metrics to surface silent invalidation loss in a cluster. + * + * Drops from before the transport ever came up are reported by + * {@link #getStartupDroppedMessages()} instead: they are an unavoidable consequence of boot + * order and would otherwise dominate this counter permanently. + */ + default long getDroppedMessages() { + return 0; + } + + /** + * Number of cache invalidation messages dropped before this transport was initialized for the + * first time, which is expected during startup and is not an operational problem on its own. + * Kept out of {@link #getDroppedMessages()} so that counter stays usable for alerting. + */ + default long getStartupDroppedMessages() { + return 0; + } + + /** + * Number of cache invalidation messages this transport tried to send and that the underlying + * provider reported as failed -- as opposed to {@link #getDroppedMessages()}, which counts + * messages never attempted because the transport was not initialized. + * + * The distinction is operational: dropped means "the transport is down, fix the transport", + * failed means "the transport believes it is up but sends are erroring". Both lose cluster + * invalidations, but only the first is visible from {@link #isInitialized()}. + */ + default long getFailedMessages() { + return 0; + } public interface CacheTransportInfo extends Serializable { diff --git a/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java b/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java new file mode 100644 index 000000000000..a45cc128f06f --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/cache/transport/PubSubCacheTransportTest.java @@ -0,0 +1,349 @@ +package com.dotcms.cache.transport; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import com.dotcms.dotpubsub.NullDotPubSubProvider; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Test; + +/** + * Unit tests for the silent-invalidation-drop fixes from issue #36803. + */ +public class PubSubCacheTransportTest { + + private PubSubCacheTransport newTransport(final NullDotPubSubProvider provider) { + return new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + } + + /** + * send() before init() must not publish, but must count the dropped message + * instead of discarding it silently. + */ + @Test + public void test_send_before_init_counts_dropped_messages() throws Exception { + final NullDotPubSubProvider provider = new NullDotPubSubProvider(); + final PubSubCacheTransport transport = newTransport(provider); + + assertEquals(0, transport.getDroppedMessages()); + + transport.send("0:testGroup"); + transport.send("0:testGroup"); + + assertEquals(2, transport.getDroppedMessages()); + assertNull("nothing may be published while uninitialized", provider.lastEventOut()); + assertFalse(transport.isInitialized()); + } + + /** + * After init(), send() publishes and the dropped counter stops growing. + */ + @Test + public void test_send_after_init_publishes() throws Exception { + final NullDotPubSubProvider provider = new NullDotPubSubProvider(); + final PubSubCacheTransport transport = newTransport(provider); + + transport.send("dropped-before-init"); + transport.init(null); + assertTrue(transport.isInitialized()); + + transport.send("0:testGroup"); + + assertNotNull(provider.lastEventOut()); + assertEquals("the pre-init drop was retired to the startup counter", + 0, transport.getDroppedMessages()); + assertEquals(1, transport.getStartupDroppedMessages()); + } + + /** + * A successful init() moves the drops that happened before the transport ever came up into the + * startup counter, so getDroppedMessages() reports only invalidations lost while the transport + * was expected to be carrying them. Without this, every node permanently reports the boot + * burst -- measured at ~2,800 on a first boot -- and the counter is useless for alerting. + */ + @Test + public void test_init_retires_startup_drops_from_the_dropped_counter() throws Exception { + final NullDotPubSubProvider provider = new NullDotPubSubProvider(); + final PubSubCacheTransport transport = newTransport(provider); + + transport.send("boot-1"); + transport.send("boot-2"); + transport.send("boot-3"); + assertEquals(3, transport.getDroppedMessages()); + assertEquals(0, transport.getStartupDroppedMessages()); + + transport.init(null); + + assertEquals("drops from before the first init are not operational drops", + 0, transport.getDroppedMessages()); + assertEquals("but they are still reported, for diagnostics", + 3, transport.getStartupDroppedMessages()); + } + + /** + * Drops after the transport has been up once are real invalidation loss: the transport was + * expected to be carrying them. A later re-init must not launder those into the startup + * bucket, which would recreate the blind spot this issue exists to remove. + */ + @Test + public void test_drops_after_first_init_are_not_retired_by_a_later_init() throws Exception { + final NullDotPubSubProvider provider = new NullDotPubSubProvider(); + final PubSubCacheTransport transport = newTransport(provider); + + transport.send("boot-1"); + transport.init(null); + assertEquals(1, transport.getStartupDroppedMessages()); + + // the transport goes down, and invalidations are lost while it is down + transport.shutdown(); + assertFalse(transport.isInitialized()); + transport.send("lost-1"); + transport.send("lost-2"); + assertEquals(2, transport.getDroppedMessages()); + + // recovery must leave the real loss on the books + transport.init(null); + + assertEquals("real drops survive a re-init", 2, transport.getDroppedMessages()); + assertEquals("and are not moved into the startup bucket", + 1, transport.getStartupDroppedMessages()); + } + + /** + * init() is idempotent: calling it on an already-initialized transport is a no-op + * rather than a second pubsub start/subscribe cycle (the rewire loop from #36544). + */ + @Test + public void test_init_is_idempotent() throws Exception { + final CountingProvider provider = new CountingProvider(); + final PubSubCacheTransport transport = + new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + + transport.init(null); + transport.init(null); + transport.init(null); + + assertEquals(1, provider.starts()); + assertTrue(transport.isInitialized()); + + // after shutdown, init() must run again + transport.shutdown(); + assertFalse(transport.isInitialized()); + transport.init(null); + assertEquals(2, provider.starts()); + } + + /** + * The guard in init() must be a test-and-set, not a check-then-act. Threads racing into init() + * all released at once must produce exactly one start()/subscribe() pair; without the + * synchronization, several can read initialized==false and each subscribe the listener, which + * is the double-subscribe the idempotency fix exists to prevent. + */ + @Test + public void test_concurrent_init_starts_and_subscribes_exactly_once() throws Exception { + final int threads = 16; + final CountingProvider provider = new CountingProvider(); + final PubSubCacheTransport transport = + new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + + final CountDownLatch ready = new CountDownLatch(threads); + final CountDownLatch go = new CountDownLatch(1); + final List failures = new ArrayList<>(); + final List workers = new ArrayList<>(); + + for (int i = 0; i < threads; i++) { + final Thread t = new Thread(() -> { + try { + ready.countDown(); + go.await(); + transport.init(null); + } catch (Throwable e) { + synchronized (failures) { + failures.add(e); + } + } + }); + workers.add(t); + t.start(); + } + + ready.await(); + go.countDown(); + for (final Thread t : workers) { + t.join(); + } + + assertTrue("no thread may fail while initializing: " + failures, failures.isEmpty()); + assertEquals("the pubsub provider must be started exactly once", 1, provider.starts()); + assertEquals("and the topic subscribed exactly once", 1, provider.subscribeCount.get()); + assertTrue(transport.isInitialized()); + } + + /** + * A failed init() must stay retryable. This is why initialized is set only after start() and + * subscribe() return, rather than being claimed up front by a compareAndSet guard: marking it + * before the work would leave a thrown start() permanently flagged as initialized, so + * isInitialized() would report a healthy transport that never subscribed and shouldReinit() + * would stop setCluster() from ever retrying it. + */ + @Test + public void test_failed_init_leaves_the_transport_retryable() throws Exception { + final BrokenStartProvider provider = new BrokenStartProvider(); + final PubSubCacheTransport transport = + new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + + try { + transport.init(null); + throw new AssertionError("init() was expected to propagate the provider failure"); + } catch (IllegalStateException expected) { + // the provider could not reach its backend + } + + assertFalse("a failed init must not report itself initialized", transport.isInitialized()); + assertTrue("and must ask to be re-initialized", transport.shouldReinit()); + + // the next cluster rewire retries, and succeeds once the backend is reachable + provider.failing = false; + transport.init(null); + + assertTrue(transport.isInitialized()); + assertEquals("both the failed attempt and the successful retry reached the provider", + 2, provider.startAttempts); + } + + /** + * The scenario the health check could not see before: init() succeeded, so the transport + * reports initialized and counts no drops, but every publish is failing. Providers signal + * that by returning false rather than throwing, and send() used to discard the boolean. + */ + @Test + public void test_send_after_init_counts_failed_publishes() throws Exception { + final FailingProvider provider = new FailingProvider(); + final PubSubCacheTransport transport = + new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + + transport.init(null); + assertTrue(transport.isInitialized()); + + transport.send("0:testGroup"); + transport.send("0:testGroup"); + + assertEquals("a failing publish is not a drop - it was attempted", + 0, transport.getDroppedMessages()); + assertEquals("failed publishes must be counted, not discarded", + 2, transport.getFailedMessages()); + assertTrue("the transport still believes it is initialized", transport.isInitialized()); + } + + /** + * A provider that publishes successfully must not inflate the failure counter. + */ + @Test + public void test_send_after_init_counts_no_failures_when_publish_succeeds() throws Exception { + final NullDotPubSubProvider provider = new NullDotPubSubProvider(); + final PubSubCacheTransport transport = newTransport(provider); + + transport.init(null); + transport.send("0:testGroup"); + + assertEquals(0, transport.getFailedMessages()); + assertEquals(0, transport.getDroppedMessages()); + } + + /** + * An asynchronous provider returns true from publish() before the send has happened, so it + * reports its own failures per topic instead. getFailedMessages() must pick those up, and + * must not double count them against send()'s own synchronous tally. + */ + @Test + public void test_getFailedMessages_includes_async_provider_failures() throws Exception { + final AsyncFailingProvider provider = new AsyncFailingProvider(); + final PubSubCacheTransport transport = + new PubSubCacheTransport(provider, new CacheTransportTopic("fakeServer", provider)); + + transport.init(null); + transport.send("0:testGroup"); + transport.send("0:testGroup"); + + assertEquals("send() saw only the optimistic true, so it counted nothing itself", + 0, transport.failedMessages.get()); + assertEquals("provider-reported failures must still surface", 2, + transport.getFailedMessages()); + } + + private static class CountingProvider extends NullDotPubSubProvider { + final AtomicInteger startCount = new AtomicInteger(0); + final AtomicInteger subscribeCount = new AtomicInteger(0); + + int starts() { + return startCount.get(); + } + + @Override + public com.dotcms.dotpubsub.DotPubSubProvider start() { + startCount.incrementAndGet(); + return this; + } + + @Override + public com.dotcms.dotpubsub.DotPubSubProvider subscribe( + final com.dotcms.dotpubsub.DotPubSubTopic topic) { + subscribeCount.incrementAndGet(); + return this; + } + } + + /** + * Mimics a provider whose backing connection is unavailable, so start() throws rather than + * returning. Flips to succeeding once {@link #failing} is cleared. + */ + private static class BrokenStartProvider extends NullDotPubSubProvider { + boolean failing = true; + int startAttempts = 0; + + @Override + public com.dotcms.dotpubsub.DotPubSubProvider start() { + startAttempts++; + if (failing) { + throw new IllegalStateException("cannot reach the pubsub backend"); + } + return this; + } + } + + /** + * Mimics a synchronous provider whose connection has dropped: publish() returns false and + * never throws, exactly as JDBCPubSubImpl/PostgresPubSubImpl/RedisPubSubImpl do. + */ + private static class FailingProvider extends NullDotPubSubProvider { + @Override + public boolean publish(final com.dotcms.dotpubsub.DotPubSubEvent event) { + return false; + } + } + + /** + * Mimics QueuingPubSubWrapper: publish() returns true immediately and the real outcome is + * reported later through getFailedPublishCount(). + */ + private static class AsyncFailingProvider extends NullDotPubSubProvider { + private long failures = 0; + + @Override + public boolean publish(final com.dotcms.dotpubsub.DotPubSubEvent event) { + failures++; + return true; + } + + @Override + public long getFailedPublishCount(final String topic) { + return failures; + } + } +} diff --git a/dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java b/dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java new file mode 100644 index 000000000000..cb90821098c4 --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/dotpubsub/QueuingPubSubWrapperTest.java @@ -0,0 +1,165 @@ +package com.dotcms.dotpubsub; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.Test; + +/** + * Covers the failure accounting added for issue #36803. + * + * {@link QueuingPubSubWrapper#publish} returns {@code true} before the wrapped provider has + * actually published, so the caller can never see a failure. Previously the result of the + * submitted task was discarded outright, which is what let a transport that was initialized but + * failing every publish look completely healthy. + */ +public class QueuingPubSubWrapperTest { + + private static final String CACHE_TOPIC = "dotcache_topic"; + private static final String OTHER_TOPIC = "osgi_topic"; + + private static final long AWAIT_MILLIS = 5000; + + /** + * Waits for the wrapper's submitter thread to drain, since publish() is asynchronous. + */ + private static void awaitFailures(final QueuingPubSubWrapper wrapper, final String topic, + final long expected) throws InterruptedException { + final long deadline = System.currentTimeMillis() + AWAIT_MILLIS; + while (System.currentTimeMillis() < deadline + && wrapper.getFailedPublishCount(topic) < expected) { + Thread.sleep(25); + } + } + + private static DotPubSubEvent event(final String topic, final String message) { + return new DotPubSubEvent.Builder().withTopic(topic).withMessage(message).build(); + } + + /** + * Method to test: {@link QueuingPubSubWrapper#getFailedPublishCount(String)} + * Given Scenario: the wrapped provider returns false for every publish, as a synchronous + * provider with a dropped connection does. + * Expected Result: each failure is counted even though publish() reported success to the + * caller. + */ + @Test + public void test_failed_publishes_are_counted() throws Exception { + final QueuingPubSubWrapper wrapper = new QueuingPubSubWrapper(new AlwaysFailsProvider()); + + assertTrue("publish() reports success regardless - that is why the count is needed", + wrapper.publish(event(CACHE_TOPIC, "inval-1"))); + assertTrue(wrapper.publish(event(CACHE_TOPIC, "inval-2"))); + + awaitFailures(wrapper, CACHE_TOPIC, 2); + assertEquals(2, wrapper.getFailedPublishCount(CACHE_TOPIC)); + } + + /** + * Method to test: {@link QueuingPubSubWrapper#getFailedPublishCount(String)} + * Given Scenario: a provider that throws instead of returning false. + * Expected Result: the throw is counted rather than lost on the submitter thread. + */ + @Test + public void test_thrown_publish_failures_are_counted() throws Exception { + final QueuingPubSubWrapper wrapper = new QueuingPubSubWrapper(new ThrowingProvider()); + + wrapper.publish(event(CACHE_TOPIC, "inval-throw")); + + awaitFailures(wrapper, CACHE_TOPIC, 1); + assertEquals(1, wrapper.getFailedPublishCount(CACHE_TOPIC)); + } + + /** + * Method to test: {@link QueuingPubSubWrapper#getFailedPublishCount(String)} + * Given Scenario: one provider instance is shared by every topic in the JVM, and a non-cache + * topic fails. + * Expected Result: the failure is attributed to its own topic only, so the cache transport's + * metric is not inflated by OSGi or cluster-management traffic. + */ + @Test + public void test_failures_are_attributed_per_topic() throws Exception { + final QueuingPubSubWrapper wrapper = new QueuingPubSubWrapper(new AlwaysFailsProvider()); + + wrapper.publish(event(OTHER_TOPIC, "osgi-restart")); + + awaitFailures(wrapper, OTHER_TOPIC, 1); + assertEquals(1, wrapper.getFailedPublishCount(OTHER_TOPIC)); + assertEquals("cache topic saw no traffic and must report no failures", + 0, wrapper.getFailedPublishCount(CACHE_TOPIC)); + } + + /** + * Method to test: {@link QueuingPubSubWrapper#getFailedPublishCount(String)} + * Given Scenario: the wrapped provider publishes successfully. + * Expected Result: nothing is counted, and a topic that was never published to reports 0 + * rather than failing on a missing counter. + */ + @Test + public void test_successful_publishes_are_not_counted() throws Exception { + final CountingProvider provider = new CountingProvider(); + final QueuingPubSubWrapper wrapper = new QueuingPubSubWrapper(provider); + + wrapper.publish(event(CACHE_TOPIC, "inval-ok")); + + final long deadline = System.currentTimeMillis() + AWAIT_MILLIS; + while (System.currentTimeMillis() < deadline && provider.published.get() == 0) { + Thread.sleep(25); + } + + assertEquals(1, provider.published.get()); + assertEquals(0, wrapper.getFailedPublishCount(CACHE_TOPIC)); + assertEquals(0, wrapper.getFailedPublishCount("never-used-topic")); + } + + /** + * Method to test: {@link QueuingPubSubWrapper#getFailedPublishCount(String)} + * Given Scenario: the failure is recorded from an event, whose topic + * {@code DotPubSubEvent.Builder.withTopic} has lowercased, but the lookup arrives with the + * topic key un-normalized -- which is what {@code DotPubSubTopic.getTopic()} returns, since it + * is a bare {@code String.valueOf(getKey())}. + * Expected Result: the count is found regardless of case. Without normalization this misses + * silently and reports zero failures while invalidations are being lost, which is the same + * class of blind spot #36803 removes. It happens to work today only because every current + * topic key is already lowercase. + */ + @Test + public void test_failure_lookup_is_case_insensitive() throws Exception { + final QueuingPubSubWrapper wrapper = new QueuingPubSubWrapper(new AlwaysFailsProvider()); + + wrapper.publish(event("DotCache_Topic", "inval-mixed-case")); + + awaitFailures(wrapper, "dotcache_topic", 1); + assertEquals("recorded under the lowercased key the event builder produced", + 1, wrapper.getFailedPublishCount("dotcache_topic")); + assertEquals("and still found when the caller passes the un-normalized topic key", + 1, wrapper.getFailedPublishCount("DotCache_Topic")); + assertEquals("any casing resolves to the same counter", + 1, wrapper.getFailedPublishCount("DOTCACHE_TOPIC")); + } + + private static class AlwaysFailsProvider extends NullDotPubSubProvider { + @Override + public boolean publish(final DotPubSubEvent event) { + return false; + } + } + + private static class ThrowingProvider extends NullDotPubSubProvider { + @Override + public boolean publish(final DotPubSubEvent event) { + throw new IllegalStateException("pubsub connection is gone"); + } + } + + private static class CountingProvider extends NullDotPubSubProvider { + final AtomicInteger published = new AtomicInteger(0); + + @Override + public boolean publish(final DotPubSubEvent event) { + published.incrementAndGet(); + return true; + } + } +} diff --git a/dotCMS/src/test/java/com/dotcms/enterprise/cluster/ClusterFactoryRewireTest.java b/dotCMS/src/test/java/com/dotcms/enterprise/cluster/ClusterFactoryRewireTest.java new file mode 100644 index 000000000000..00040641e35b --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/enterprise/cluster/ClusterFactoryRewireTest.java @@ -0,0 +1,112 @@ +package com.dotcms.enterprise.cluster; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import com.dotcms.cluster.bean.Server; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.junit.Test; + +/** + * Unit tests for {@link ClusterFactory#shouldRewire}, the decision that governs whether a + * heartbeat re-runs the cluster rewire. + * + * This is the logic that guarantees a failed cache-transport init keeps being retried (issue + * #36803). It could not be exercised against a live cluster: {@code ServerHeartbeatJob.execute()} + * calls {@code LicenseUtil.updateLicenseHeartbeat()} before {@code rewireClusterIfNeeded()}, and + * that throws when the database is down, so the only way to move the failure counter is for the + * database to be healthy while {@code setCluster()}/{@code testCluster()} fails -- a window that + * resisted local reproduction. Hence these tests on the extracted predicate. + */ +public class ClusterFactoryRewireTest { + + private static final long NO_PENDING_FAILURES = 0L; + + private static Server server(final String id) { + return Server.builder().withServerId(id).withIpAddress("127.0.0." + id.length()).build(); + } + + private static final Server ME = server("me"); + private static final Server PEER = server("peer"); + + /** + * Method to test: {@link ClusterFactory#shouldRewire} + * Given Scenario: membership is unchanged, this server is in the alive set, and no rewire + * failure is pending -- the steady state of a healthy cluster on every heartbeat. + * Expected Result: no rewire. Rewiring here is what produced the churn measured in #36544. + */ + @Test + public void steadyStateDoesNotRewire() { + final List alive = Arrays.asList(ME, PEER); + + assertFalse(ClusterFactory.shouldRewire(alive, Arrays.asList(ME, PEER), ME, + NO_PENDING_FAILURES)); + } + + /** + * Method to test: {@link ClusterFactory#shouldRewire} + * Given Scenario: a previous rewire failed, so KNOWN_SERVERS was never advanced, and + * membership has since settled back to exactly that stale set. + * Expected Result: rewire anyway. This is the regression guard for the retry clause added in + * #36803 -- the membership comparison alone reports "nothing changed", so without the pending + * failure check the transport would stay broken forever, the failure counter could never + * return to zero, and the cache-transport health check would report unhealthy indefinitely. + */ + @Test + public void pendingFailureForcesRetryEvenWhenMembershipLooksUnchanged() { + final List alive = Arrays.asList(ME, PEER); + final List knownServers = Arrays.asList(ME, PEER); + + // sanity: membership genuinely looks unchanged, so only the failure counter can trigger it + assertFalse("precondition: membership must look identical for this test to mean anything", + ClusterFactory.shouldRewire(alive, knownServers, ME, NO_PENDING_FAILURES)); + + assertTrue(ClusterFactory.shouldRewire(alive, knownServers, ME, 1L)); + assertTrue("still retried after several failures", + ClusterFactory.shouldRewire(alive, knownServers, ME, 7L)); + } + + /** + * Method to test: {@link ClusterFactory#shouldRewire} + * Given Scenario: a node joined since the last successful rewire. + * Expected Result: rewire, so the transport is wired to the new membership. + */ + @Test + public void membershipChangeRewires() { + assertTrue("a server joined", + ClusterFactory.shouldRewire(Arrays.asList(ME, PEER), Collections.singletonList(ME), + ME, NO_PENDING_FAILURES)); + + assertTrue("a server left", + ClusterFactory.shouldRewire(Collections.singletonList(ME), Arrays.asList(ME, PEER), + ME, NO_PENDING_FAILURES)); + } + + /** + * Method to test: {@link ClusterFactory#shouldRewire} + * Given Scenario: this server is absent from the alive set, which is how a node that lost its + * own registration looks. + * Expected Result: rewire, so it registers itself again. + */ + @Test + public void missingSelfRewires() { + final List aliveWithoutMe = Collections.singletonList(PEER); + + assertTrue(ClusterFactory.shouldRewire(aliveWithoutMe, aliveWithoutMe, ME, + NO_PENDING_FAILURES)); + } + + /** + * Method to test: {@link ClusterFactory#shouldRewire} + * Given Scenario: the very first heartbeat, when KNOWN_SERVERS is still the empty list it is + * initialized to and this server is already alive. + * Expected Result: rewire, so the cluster is wired at least once. + */ + @Test + public void firstHeartbeatRewires() { + assertTrue(ClusterFactory.shouldRewire(Collections.singletonList(ME), + Collections.emptyList(), ME, NO_PENDING_FAILURES)); + } +} diff --git a/dotCMS/src/test/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheckTest.java b/dotCMS/src/test/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheckTest.java new file mode 100644 index 000000000000..21e72b87e5cc --- /dev/null +++ b/dotCMS/src/test/java/com/dotcms/health/checks/cdi/CacheTransportHealthCheckTest.java @@ -0,0 +1,280 @@ +package com.dotcms.health.checks.cdi; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; + +import com.dotcms.enterprise.cluster.ClusterFactory; +import com.dotcms.health.config.HealthCheckConfig.HealthCheckMode; +import com.dotcms.health.model.HealthCheckResult; +import com.dotcms.health.model.HealthStatus; +import com.dotmarketing.business.CacheLocator; +import com.dotmarketing.business.DotCacheAdministrator; +import com.dotmarketing.business.cache.transport.CacheTransport; +import com.dotmarketing.business.cache.transport.NullTransport; +import com.dotmarketing.util.Config; +import java.util.Map; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.MockedStatic; + +/** + * Unit tests for {@link CacheTransportHealthCheck}, the probe added by issue #36803 to surface a + * cache transport that is silently dropping cluster invalidations. + * + * The transport and the rewire counter are both reached through statics + * ({@code CacheLocator.getCacheAdministrator()} and {@code ClusterFactory.getRewireFailures()}), + * so they are mocked statically here, in the same style as {@link VelocityHealthCheckTest}. + * + * Tests run the check in PRODUCTION mode unless they are specifically about monitor mode, so an + * unhealthy verdict shows up as DOWN rather than being converted to DEGRADED. + */ +public class CacheTransportHealthCheckTest { + + private static final String MODE_KEY = "health.check.cache-transport.mode"; + private static final String GRACE_KEY = + "health.check.cache-transport.initialization-grace-period-seconds"; + private static final String THRESHOLD_KEY = + "health.check.cache-transport.rewire-failure-threshold"; + + private MockedStatic cacheLocatorMock; + private MockedStatic clusterFactoryMock; + + @Before + public void setUp() { + cacheLocatorMock = mockStatic(CacheLocator.class); + clusterFactoryMock = mockStatic(ClusterFactory.class); + clusterFactoryMock.when(ClusterFactory::getRewireFailures).thenReturn(0L); + Config.setProperty(MODE_KEY, "PRODUCTION"); + } + + @After + public void tearDown() { + cacheLocatorMock.close(); + clusterFactoryMock.close(); + // Restored to the values the production code defaults to, rather than cleared: Config has + // no clear-for-tests hook, and leaving these set would follow the JVM fork into other + // test classes. + Config.setProperty(MODE_KEY, "MONITOR_MODE"); + Config.setProperty(GRACE_KEY, "120"); + Config.setProperty(THRESHOLD_KEY, "3"); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: the node is configured with {@link NullTransport}, the deliberate no-op + * transport for a node that is not part of a cluster. + * Expected Result: healthy. There is nothing that could be dropping invalidations, and the + * per-transport fields are omitted rather than reported as zeros. + */ + @Test + public void nullTransportReportsUp() { + withTransport(new NullTransport()); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.UP, result.status()); + final Map data = data(result); + assertEquals("NullTransport", data.get("transport")); + assertFalse("a no-op transport has no invalidation state worth reporting", + data.containsKey("initialized")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: the cache layer cannot be resolved at all, which is what a probe sees when + * it runs before the cache subsystem is up. + * Expected Result: healthy, not an exception and not a false alarm. + */ + @Test + public void unresolvableCacheLayerReportsUp() { + cacheLocatorMock.when(CacheLocator::getCacheAdministrator) + .thenThrow(new IllegalStateException("cache layer not up yet")); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.UP, result.status()); + assertEquals("none", data(result).get("transport")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: a real transport that has never been initialized, polled inside the startup + * grace period. Every node is in this state for the first seconds of its life, because the + * transport is initialized late in boot when the cluster is wired. + * Expected Result: healthy, flagged as awaiting initialization. Reporting DOWN here would turn + * a normal rolling deployment into a DEGRADED overall status on every pod start. + */ + @Test + public void uninitializedTransportWithinGracePeriodReportsUp() { + Config.setProperty(GRACE_KEY, "120"); + withTransport(transport(false, 7L, 2000L, 0L)); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.UP, result.status()); + final Map data = data(result); + assertEquals(false, data.get("initialized")); + assertEquals(true, data.get("awaitingInitialization")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: the same never-initialized transport, but the grace period has elapsed. + * Expected Result: unhealthy. This is the #36803 fault the probe exists to catch -- a node + * that never wires its transport and drops every invalidation forever. + */ + @Test + public void uninitializedTransportPastGracePeriodReportsDown() { + Config.setProperty(GRACE_KEY, "0"); + withTransport(transport(false, 7L, 2000L, 0L)); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.DOWN, result.status()); + assertEquals(true, data(result).get("awaitingInitialization")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: the transport was initialized on an earlier poll and is not anymore, with a + * long grace period configured. + * Expected Result: unhealthy immediately. The grace period covers startup only; losing a + * transport that had been working is a regression, not a node still booting. + */ + @Test + public void transportLostAfterInitializationReportsDownDespiteGracePeriod() { + Config.setProperty(GRACE_KEY, "3600"); + final CacheTransport transport = transport(true, 0L, 0L, 0L); + withTransport(transport); + final CacheTransportHealthCheck check = new CacheTransportHealthCheck(); + + assertEquals("baseline: a working transport is UP", HealthStatus.UP, check.check().status()); + + when(transport.isInitialized()).thenReturn(false); + + final HealthCheckResult result = check.check(); + assertEquals(HealthStatus.DOWN, result.status()); + assertEquals("the grace period must not apply once a transport has been seen working", + false, data(result).get("awaitingInitialization")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: an initialized transport on a node whose last rewire attempt failed once, + * below the configured threshold. + * Expected Result: healthy. {@code testCluster()} can throw on a momentary database hiccup + * while invalidations keep flowing, so a single stale failure must not report DOWN. + */ + @Test + public void rewireFailuresBelowThresholdReportUp() { + Config.setProperty(THRESHOLD_KEY, "3"); + clusterFactoryMock.when(ClusterFactory::getRewireFailures).thenReturn(1L); + withTransport(transport(true, 0L, 0L, 0L)); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.UP, result.status()); + assertEquals(1L, data(result).get("rewireFailures")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: consecutive rewire failures have reached the threshold while the transport + * still reports itself initialized. + * Expected Result: unhealthy -- the transport may be wired to a stale view of the cluster. + */ + @Test + public void rewireFailuresAtThresholdReportDown() { + Config.setProperty(THRESHOLD_KEY, "3"); + clusterFactoryMock.when(ClusterFactory::getRewireFailures).thenReturn(3L); + withTransport(transport(true, 0L, 0L, 0L)); + + final HealthCheckResult result = new CacheTransportHealthCheck().check(); + + assertEquals(HealthStatus.DOWN, result.status()); + assertEquals(3L, data(result).get("rewireFailures")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#buildStructuredData} + * Given Scenario: a healthy transport carrying non-zero counters. + * Expected Result: every counter monitoring is expected to consume is present, and the drops + * from before the transport came up are reported separately from the operational ones. + */ + @Test + public void structuredDataExposesEveryCounter() { + clusterFactoryMock.when(ClusterFactory::getRewireFailures).thenReturn(2L); + withTransport(transport(true, 5L, 2841L, 11L)); + + final Map data = data(new CacheTransportHealthCheck().check()); + + assertEquals(true, data.get("initialized")); + assertEquals(5L, data.get("droppedInvalidations")); + assertEquals("startup drops are reported apart from operational ones, so the alertable" + + " counter is not permanently dominated by the boot burst", + 2841L, data.get("startupDroppedInvalidations")); + assertEquals(11L, data.get("failedInvalidations")); + assertEquals(2L, data.get("rewireFailures")); + assertEquals(false, data.get("awaitingInitialization")); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#check()} + * Given Scenario: a genuinely broken transport, with the check in its default mode. + * Expected Result: DEGRADED rather than DOWN. The default is MONITOR_MODE precisely so a + * broken transport cannot fail a readiness probe and drain a node that can still serve + * traffic; this test pins that deployment-safety property. + */ + @Test + public void defaultModeIsMonitorModeAndConvertsDownToDegraded() { + final CacheTransportHealthCheck check = new CacheTransportHealthCheck(); + assertEquals("the default must stay MONITOR_MODE", HealthCheckMode.MONITOR_MODE, + check.getDefaultMode()); + + Config.setProperty(MODE_KEY, HealthCheckMode.MONITOR_MODE.name()); + Config.setProperty(GRACE_KEY, "0"); + withTransport(transport(false, 12L, 0L, 0L)); + + final HealthCheckResult result = check.check(); + + assertEquals(HealthStatus.DEGRADED, result.status()); + assertTrue(result.monitorModeApplied()); + } + + /** + * Method to test: {@link CacheTransportHealthCheck#isLivenessCheck()} + * Given Scenario: the probe registration flags are read by the health framework. + * Expected Result: never a liveness check. Restarting pods on transport failure is what + * amplified the incident in #36544. + */ + @Test + public void isNeverALivenessCheck() { + assertFalse(new CacheTransportHealthCheck().isLivenessCheck()); + } + + private void withTransport(final CacheTransport transport) { + final DotCacheAdministrator admin = mock(DotCacheAdministrator.class); + when(admin.getTransport()).thenReturn(transport); + cacheLocatorMock.when(CacheLocator::getCacheAdministrator).thenReturn(admin); + } + + private CacheTransport transport(final boolean initialized, final long dropped, + final long startupDropped, final long failed) { + final CacheTransport transport = mock(CacheTransport.class); + when(transport.isInitialized()).thenReturn(initialized); + when(transport.getDroppedMessages()).thenReturn(dropped); + when(transport.getStartupDroppedMessages()).thenReturn(startupDropped); + when(transport.getFailedMessages()).thenReturn(failed); + return transport; + } + + private Map data(final HealthCheckResult result) { + return result.data().orElseThrow( + () -> new AssertionError("the check must always publish structured data")); + } +}