diff --git a/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationManager.java b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationManager.java index 5a0d6fc9b..c0ab2a6e9 100644 --- a/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationManager.java +++ b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationManager.java @@ -44,12 +44,20 @@ public class ConfigurationManager private volatile ConfigurationOverlaySnapshot cachedBaseSnapshot; /** - * @param provider the configuration provider for fetching overlays + * @param provider the configuration provider for fetching overlays * @param baseTemplatePath path to the base cassandra.yaml template, or {@code null} for an empty base + * @param configurationStore path to the configuration store directory + * @param failurePolicy the failure policy to apply when the provider is unavailable */ - public ConfigurationManager(ConfigurationProvider provider, @Nullable Path baseTemplatePath) + public ConfigurationManager(ConfigurationProvider provider, + @Nullable Path baseTemplatePath, + Path configurationStore, + FailurePolicy failurePolicy) { - this.provider = Objects.requireNonNull(provider, "provider must not be null"); + Objects.requireNonNull(provider, "provider must not be null"); + Objects.requireNonNull(configurationStore, "configurationStore must not be null"); + Objects.requireNonNull(failurePolicy, "failurePolicy must not be null"); + this.provider = FailurePolicyWrapper.wrap(provider, configurationStore, failurePolicy); this.baseTemplatePath = baseTemplatePath; } @@ -70,6 +78,12 @@ public ConfigurationOverlaySnapshot getEffectiveConfiguration(InstanceMetadata i { providerSnapshot = provider.getOverlay(instance); } + catch (ConfigurationManagerException e) + { + // Preserve the subtype (e.g. ConfigurationProviderUnavailableException) so handlers can + // map it to the appropriate HTTP status (e.g. 503) rather than a generic 500. + throw e; + } catch (Exception e) { throw new ConfigurationManagerException( diff --git a/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationProvider.java b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationProvider.java index e639eef7c..bfb116494 100644 --- a/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationProvider.java +++ b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationProvider.java @@ -29,6 +29,10 @@ * Configuration Manager. Implementations may persist overlays locally (files), remotely * (etcd, Consul, HTTP APIs), or in-memory (for testing). * + *

Implementations that cannot reach their backing store may throw + * {@link ConfigurationProviderUnavailableException} to signal an outage explicitly; see + * {@link FailurePolicyWrapper#isUnavailable(Throwable)} for how other exception types are classified. + * *

The provider stores version-agnostic overlays and does not perform version-specific * validation or merge logic. Validation against a version-aware schema and computing updated * overlays (via {@link ConfigurationPatchApplier}) are the responsibility of the diff --git a/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationProviderUnavailableException.java b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationProviderUnavailableException.java new file mode 100644 index 000000000..6560ebde7 --- /dev/null +++ b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationProviderUnavailableException.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.sidecar.configmanagement; + +/** + * Thrown when a configuration operation is rejected because the + * {@link ConfigurationProvider} is unavailable and the configured + * {@link FailurePolicy} does not allow the operation to proceed. + */ +public class ConfigurationProviderUnavailableException extends ConfigurationManagerException +{ + public ConfigurationProviderUnavailableException(String message, Throwable cause) + { + super(message, cause); + } +} diff --git a/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/FailurePolicy.java b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/FailurePolicy.java new file mode 100644 index 000000000..6e28c1922 --- /dev/null +++ b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/FailurePolicy.java @@ -0,0 +1,48 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.sidecar.configmanagement; + +/** + * Governs behavior when the {@link ConfigurationProvider} is unreachable. + */ +public enum FailurePolicy +{ + /** + * All operations fall back to the cached configuration. + * Reads use the cached overlay; writes update the cached overlay. + * + *

Warning: writes performed against the cache while the provider is + * unavailable are local-only. Once the provider recovers, subsequent reads return + * the delegate's value, which overwrites the local cache. Any writes made during the + * outage are therefore lost and are not reconciled back to the delegate. + */ + CACHED_READ_WRITE, + + /** + * Reads fall back to the cached overlay; writes are rejected + * with {@link ConfigurationProviderUnavailableException}. + */ + CACHED_READ_ONLY, + + /** + * All operations fail when the provider is unavailable, throwing + * {@link ConfigurationProviderUnavailableException} so handlers can surface HTTP 503. + */ + FAIL +} diff --git a/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/FailurePolicyWrapper.java b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/FailurePolicyWrapper.java new file mode 100644 index 000000000..08ec29f5f --- /dev/null +++ b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/FailurePolicyWrapper.java @@ -0,0 +1,231 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.sidecar.configmanagement; + +import java.nio.file.Path; +import java.util.Objects; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** + * A decorator around {@link ConfigurationProvider} that applies a configurable + * {@link FailurePolicy} when the downstream provider is unavailable. + * + *

On successful calls to the delegate, the result is cached locally via a + * {@link FileBasedConfigurationProvider} writing to {@code cached_overlay.json}. + * When the delegate fails, behavior is governed by the configured failure policy. + * + *

Only failures indicating the delegate could not be reached activate the failure policy; any + * other failure is propagated to the caller unchanged. See {@link #isUnavailable(Throwable)}. + */ +public class FailurePolicyWrapper implements ConfigurationProvider +{ + private static final Logger LOGGER = LoggerFactory.getLogger(FailurePolicyWrapper.class); + static final String CACHED_OVERLAY_FILE_NAME = "cached_overlay.json"; + + private final ConfigurationProvider delegate; + private final FileBasedConfigurationProvider cache; + private final FailurePolicy failurePolicy; + + public FailurePolicyWrapper(ConfigurationProvider delegate, + Path cacheDir, + FailurePolicy failurePolicy) + { + this.delegate = Objects.requireNonNull(delegate, "delegate must not be null"); + this.failurePolicy = Objects.requireNonNull(failurePolicy, "failurePolicy must not be null"); + Objects.requireNonNull(cacheDir, "cacheDir must not be null"); + this.cache = new FileBasedConfigurationProvider(cacheDir, CACHED_OVERLAY_FILE_NAME); + } + + /** + * Wraps a provider with failure policy handling. If the delegate is already a + * {@link FileBasedConfigurationProvider}, it is returned directly (local providers + * cannot be "unavailable" in the network sense). + * + * @param delegate the downstream configuration provider + * @param cacheDir path to the configuration store directory for caching + * @param failurePolicy the failure policy to apply + * @return the delegate directly if it is file-based, or a wrapped instance otherwise + */ + public static ConfigurationProvider wrap(ConfigurationProvider delegate, + Path cacheDir, + FailurePolicy failurePolicy) + { + if (delegate instanceof FileBasedConfigurationProvider) + { + return delegate; + } + return new FailurePolicyWrapper(delegate, cacheDir, failurePolicy); + } + + @Override + @Nullable + public ConfigurationOverlaySnapshot getOverlay(InstanceMetadata instance) + { + ConfigurationOverlaySnapshot result; + try + { + result = delegate.getOverlay(instance); + } + catch (Exception e) + { + return handleReadFailure(instance, e); + } + + // A null result means the overlay was deleted upstream. Cache an empty overlay (rather + // than leaving the stale entry) so a deleted overlay is not resurrected from cache during + // a later outage. An empty overlay merges to a no-op, so it is equivalent to no overlay. + updateCache(instance, result != null ? result : ConfigurationOverlaySnapshot.emptySnapshot()); + return result; + } + + @Override + public boolean storeOverlay(InstanceMetadata instance, + @Nullable String originalHash, + @NotNull ConfigurationOverlaySnapshot newSnapshot) + { + boolean stored; + try + { + stored = delegate.storeOverlay(instance, originalHash, newSnapshot); + } + catch (Exception e) + { + return handleWriteFailure(instance, originalHash, newSnapshot, e); + } + + if (stored) + { + updateCache(instance, newSnapshot); + } + return stored; + } + + /** + * @param failure the exception thrown by the delegate + * @return {@code true} if the failure indicates the provider could not be reached + */ + static boolean isUnavailable(Throwable failure) + { + // The provider's explicit outage signal. + if (failure instanceof ConfigurationProviderUnavailableException) + { + return true; + } + + // Domain errors and provider bugs must be surfaced to the client. Anything else — IOException, + // socket and timeout failures, provider client exceptions — is considered an outage. + return !(failure instanceof ConfigurationManagerException + || failure instanceof IllegalArgumentException + || failure instanceof NullPointerException + || failure instanceof IllegalStateException + || failure instanceof UnsupportedOperationException); + } + + private static void rethrowUnlessUnavailable(Exception failure) + { + if (failure instanceof RuntimeException && !isUnavailable(failure)) + { + throw (RuntimeException) failure; + } + } + + @NotNull + private ConfigurationOverlaySnapshot handleReadFailure(InstanceMetadata instance, Exception cause) + { + rethrowUnlessUnavailable(cause); + + if (failurePolicy == FailurePolicy.FAIL) + { + throw new ConfigurationProviderUnavailableException( + "Configuration provider is unavailable; reads are rejected under FAIL policy", cause); + } + LOGGER.warn("Configuration provider unavailable for instance {}; falling back to cached overlay", + instance.id(), cause); + + ConfigurationOverlaySnapshot cached = cache.getOverlay(instance); + if (cached == null) + { + // null means no read has been cached yet, so there is nothing to serve; a cached empty + // overlay would instead mean the provider has no overlay for this instance. + throw new ConfigurationProviderUnavailableException( + "Configuration provider is unavailable and no overlay is cached for instance " + + instance.id(), cause); + } + return cached; + } + + private boolean handleWriteFailure(InstanceMetadata instance, + @Nullable String originalHash, + @NotNull ConfigurationOverlaySnapshot newSnapshot, + Exception cause) + { + rethrowUnlessUnavailable(cause); + + switch (failurePolicy) + { + case FAIL: + throw new ConfigurationProviderUnavailableException( + "Configuration provider is unavailable; writes are rejected under FAIL policy", cause); + case CACHED_READ_ONLY: + throw new ConfigurationProviderUnavailableException( + "Configuration provider is unavailable; writes are rejected under CACHED_READ_ONLY policy", + cause); + case CACHED_READ_WRITE: + LOGGER.warn("Configuration provider unavailable for instance {}; writing to cached overlay", + instance.id(), cause); + return cache.storeOverlay(instance, originalHash, newSnapshot); + default: + throw new IllegalStateException("Unknown failure policy: " + failurePolicy); + } + } + + /** + * Best-effort update of the local cache after a successful delegate operation. + * Cache failures (e.g. disk full, permission errors, corrupt cache file) are logged + * and swallowed so they do not fail an operation the delegate already completed. + */ + private void updateCache(InstanceMetadata instance, ConfigurationOverlaySnapshot snapshot) + { + try + { + ConfigurationOverlaySnapshot cached = cache.getOverlay(instance); + String cachedHash = cached != null ? cached.hash() : null; + // Skip the disk write when the cache already holds the same content. + if (snapshot.hash().equals(cachedHash)) + { + return; + } + if (!cache.storeOverlay(instance, cachedHash, snapshot)) + { + LOGGER.debug("Cache update skipped for instance {} due to concurrent modification", instance.id()); + } + } + catch (Exception e) + { + LOGGER.warn("Failed to update cached overlay for instance {}; delegate operation succeeded", + instance.id(), e); + } + } +} diff --git a/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/FileBasedConfigurationProvider.java b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/FileBasedConfigurationProvider.java index b9d48d295..e45dd6946 100644 --- a/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/FileBasedConfigurationProvider.java +++ b/server/src/main/java/org/apache/cassandra/sidecar/configmanagement/FileBasedConfigurationProvider.java @@ -44,14 +44,21 @@ */ public class FileBasedConfigurationProvider implements ConfigurationProvider { - private static final String CONFIG_FILE_NAME = "overlay.json"; + static final String DEFAULT_OVERLAY_FILE_NAME = "overlay.json"; private final Path configDir; + private final String configFileName; private final ConcurrentHashMap overlays = new ConcurrentHashMap<>(); public FileBasedConfigurationProvider(Path configDir) + { + this(configDir, DEFAULT_OVERLAY_FILE_NAME); + } + + public FileBasedConfigurationProvider(Path configDir, String configFileName) { this.configDir = Objects.requireNonNull(configDir, "configDir must not be null"); + this.configFileName = Objects.requireNonNull(configFileName, "configFileName must not be null"); } @Override @@ -93,7 +100,7 @@ public boolean storeOverlay(InstanceMetadata instance, @Nullable private ConfigurationOverlaySnapshot readFromDisk(InstanceMetadata instance) { - Path configFile = resolveInstanceDir(instance).resolve(CONFIG_FILE_NAME); + Path configFile = resolveInstanceDir(instance).resolve(configFileName); if (!Files.exists(configFile)) { return null; @@ -112,7 +119,7 @@ private ConfigurationOverlaySnapshot readFromDisk(InstanceMetadata instance) private void writeToDisk(InstanceMetadata instance, ConfigurationOverlaySnapshot snapshot) { Path instanceDir = resolveInstanceDir(instance); - Path configFile = instanceDir.resolve(CONFIG_FILE_NAME); + Path configFile = instanceDir.resolve(configFileName); Path tempFile = null; try { diff --git a/server/src/test/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationManagerTest.java b/server/src/test/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationManagerTest.java index 719f6b0f4..be3b9b2c5 100644 --- a/server/src/test/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationManagerTest.java +++ b/server/src/test/java/org/apache/cassandra/sidecar/configmanagement/ConfigurationManagerTest.java @@ -32,10 +32,12 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import io.vertx.core.json.JsonObject; import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata; @@ -53,6 +55,9 @@ class ConfigurationManagerTest { private static final Path BASE_TEMPLATE = Paths.get("src/test/resources/configmanagement/cassandra_latest.yaml"); + @TempDir + Path tempDir; + private InMemoryConfigurationProvider provider; @BeforeEach @@ -64,7 +69,7 @@ void setUp() @Test void testGetEffectiveConfigurationNoOverlay() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); ConfigurationOverlaySnapshot result = manager.getEffectiveConfiguration(instance); @@ -92,7 +97,7 @@ void testGetEffectiveConfigurationWithOverlay() ConfigurationOverlaySnapshot snapshot = new ConfigurationOverlaySnapshot(Instant.now(), overlay); provider.storeOverlay(instance, null, snapshot); - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); ConfigurationOverlaySnapshot result = manager.getEffectiveConfiguration(instance); // Overlay values take precedence @@ -125,19 +130,19 @@ public boolean storeOverlay(InstanceMetadata instance, String originalHash, } }; - ConfigurationManager manager = new ConfigurationManager(failingProvider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(failingProvider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); assertThatThrownBy(() -> manager.getEffectiveConfiguration(instance)) - .isInstanceOf(ConfigurationManagerException.class) - .hasMessageContaining("Failed to retrieve configuration overlay from provider") + .isInstanceOf(ConfigurationProviderUnavailableException.class) + .hasMessageContaining("FAIL policy") .hasCauseInstanceOf(UncheckedIOException.class); } @Test void testGetEffectiveConfigurationNullBaseTemplateNoOverlay() { - ConfigurationManager manager = new ConfigurationManager(provider, null); + ConfigurationManager manager = new ConfigurationManager(provider, null, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); ConfigurationOverlaySnapshot result = manager.getEffectiveConfiguration(instance); @@ -161,7 +166,7 @@ void testGetEffectiveConfigurationNullBaseTemplateWithOverlay() ConfigurationOverlaySnapshot snapshot = new ConfigurationOverlaySnapshot(overlayTime, overlay); provider.storeOverlay(instance, null, snapshot); - ConfigurationManager manager = new ConfigurationManager(provider, null); + ConfigurationManager manager = new ConfigurationManager(provider, null, tempDir, FailurePolicy.FAIL); ConfigurationOverlaySnapshot result = manager.getEffectiveConfiguration(instance); assertThat(result.configuration().cassandraYaml().getInteger("concurrent_reads")).isEqualTo(128); @@ -172,7 +177,7 @@ void testGetEffectiveConfigurationNullBaseTemplateWithOverlay() @Test void testGetEffectiveConfigurationCachesBaseSnapshot() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); ConfigurationOverlaySnapshot first = manager.getEffectiveConfiguration(instance); @@ -184,7 +189,7 @@ void testGetEffectiveConfigurationCachesBaseSnapshot() @Test void testPatchAddTopLevelKey() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); ConfigurationOverlaySnapshot baseEffective = manager.getEffectiveConfiguration(instance); @@ -205,7 +210,7 @@ void testPatchAddTopLevelKey() @Test void testPatchAddJvmOpt() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); String baseHash = manager.getEffectiveConfiguration(instance).hash(); @@ -230,7 +235,7 @@ void testPatchConflictingBooleanJvmOptRejected() CassandraConfigurationOverlay initial = new CassandraConfigurationOverlay(null, jvmOpts); provider.storeOverlay(instance, null, new ConfigurationOverlaySnapshot(Instant.now(), initial)); - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); String effectiveHash = manager.getEffectiveConfiguration(instance).hash(); List ops = List.of( @@ -257,7 +262,7 @@ void testPatchReturnedConfigMatchesStoredConfig() CassandraConfigurationOverlay initial = new CassandraConfigurationOverlay(null, jvmOpts); provider.storeOverlay(instance, null, new ConfigurationOverlaySnapshot(Instant.now(), initial)); - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); String effectiveHash = manager.getEffectiveConfiguration(instance).hash(); List ops = List.of( @@ -285,7 +290,7 @@ void testPatchRemoveTopLevelKeyFromOverlay() CassandraConfigurationOverlay initial = new CassandraConfigurationOverlay(initialYaml, null); provider.storeOverlay(instance, null, new ConfigurationOverlaySnapshot(Instant.now(), initial)); - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); String effectiveHash = manager.getEffectiveConfiguration(instance).hash(); List ops = List.of( @@ -301,7 +306,7 @@ void testPatchRemoveTopLevelKeyFromOverlay() @Test void testPatchRemoveTemplateOnlyKeyFails() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); String baseHash = manager.getEffectiveConfiguration(instance).hash(); @@ -317,7 +322,7 @@ void testPatchRemoveTemplateOnlyKeyFails() @Test void testPatchReplaceExistingKey() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); String baseHash = manager.getEffectiveConfiguration(instance).hash(); @@ -334,7 +339,7 @@ void testPatchReplaceExistingKey() @Test void testPatchReplaceAbsentKeyFails() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); String baseHash = manager.getEffectiveConfiguration(instance).hash(); @@ -350,7 +355,7 @@ void testPatchReplaceAbsentKeyFails() @Test void testPatchTestMatchingValue() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); String baseHash = manager.getEffectiveConfiguration(instance).hash(); @@ -374,7 +379,7 @@ void testPatchTestMismatchRejectsEntirePatch() CassandraConfigurationOverlay initial = new CassandraConfigurationOverlay(initialYaml, null); provider.storeOverlay(instance, null, new ConfigurationOverlaySnapshot(Instant.now(), initial)); - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); String effectiveHash = manager.getEffectiveConfiguration(instance).hash(); List ops = List.of( @@ -402,7 +407,7 @@ void testPatchConflictStaleHash() CassandraConfigurationOverlay initial = new CassandraConfigurationOverlay(initialYaml, null); provider.storeOverlay(instance, null, new ConfigurationOverlaySnapshot(Instant.now(), initial)); - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); String actualHash = manager.getEffectiveConfiguration(instance).hash(); String staleHash = "sha256:0000000000000000000000000000000000000000000000000000000000000000"; @@ -450,7 +455,7 @@ public boolean storeOverlay(InstanceMetadata inst, String originalHash, } }; - ConfigurationManager manager = new ConfigurationManager(rejectingProvider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(rejectingProvider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); String effectiveHash = manager.getEffectiveConfiguration(instance).hash(); List ops = List.of( @@ -495,7 +500,7 @@ public boolean storeOverlay(InstanceMetadata inst, String originalHash, } }; - ConfigurationManager manager = new ConfigurationManager(conflictingProvider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(conflictingProvider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); String effectiveHash = manager.getEffectiveConfiguration(instance).hash(); List ops = List.of( @@ -530,17 +535,19 @@ public boolean storeOverlay(InstanceMetadata instance, String originalHash, } }; - ConfigurationManager manager = new ConfigurationManager(failingProvider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(failingProvider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); List ops = List.of( new ConfigurationPatchOperation(ConfigurationPatchOperation.Op.ADD, "/configuration/cassandraYaml/concurrent_reads", 128)); + // The patch path reads the current overlay first, so an unavailable provider surfaces the + // ConfigurationProviderUnavailableException from the read before any write is attempted. assertThatThrownBy(() -> manager.patchConfiguration(instance, "sha256:abc", ops)) - .isInstanceOf(ConfigurationManagerException.class) + .isInstanceOf(ConfigurationProviderUnavailableException.class) .isNotInstanceOf(ConfigurationConflictException.class) - .hasMessageContaining("Failed to patch configuration") + .hasMessageContaining("FAIL policy") .hasCauseInstanceOf(UncheckedIOException.class); } @@ -548,7 +555,7 @@ public boolean storeOverlay(InstanceMetadata instance, String originalHash, void testPatchConcurrentSameInstance() throws Exception { InstanceMetadata instance = mockInstance(1); - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); String baseHash = manager.getEffectiveConfiguration(instance).hash(); int threadCount = 10; @@ -596,7 +603,7 @@ void testPatchConcurrentSameInstance() throws Exception @Test void testPatchDuplicatePathsRejected() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); String baseHash = manager.getEffectiveConfiguration(instance).hash(); @@ -614,7 +621,7 @@ void testPatchDuplicatePathsRejected() @Test void testPatchInvalidPathFormat() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); String baseHash = manager.getEffectiveConfiguration(instance).hash(); @@ -630,7 +637,7 @@ void testPatchInvalidPathFormat() @Test void testPatchEmptyOperationsRejected() { - ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE); + ConfigurationManager manager = new ConfigurationManager(provider, BASE_TEMPLATE, tempDir, FailurePolicy.FAIL); InstanceMetadata instance = mockInstance(1); String baseHash = manager.getEffectiveConfiguration(instance).hash(); @@ -639,10 +646,147 @@ void testPatchEmptyOperationsRejected() .hasMessageContaining("must not be empty"); } + // -- Failure policy integration tests -- + + @Test + void testGetEffectiveConfigurationCachedReadOnlyFallsBackToCache() + { + ToggleableProvider delegate = new ToggleableProvider(); + InstanceMetadata instance = mockInstance(1); + + ConfigurationManager manager = new ConfigurationManager(delegate, BASE_TEMPLATE, tempDir, + FailurePolicy.CACHED_READ_ONLY); + + // Seed the overlay and fetch to populate cache + JsonObject yaml = new JsonObject().put("concurrent_reads", 128); + CassandraConfigurationOverlay overlay = new CassandraConfigurationOverlay(yaml, null); + delegate.storeOverlay(instance, null, new ConfigurationOverlaySnapshot(Instant.now(), overlay)); + ConfigurationOverlaySnapshot first = manager.getEffectiveConfiguration(instance); + assertThat(first.configuration().cassandraYaml().getInteger("concurrent_reads")).isEqualTo(128); + + // Fail the delegate — should fall back to cached overlay + delegate.setFailing(true); + ConfigurationOverlaySnapshot cached = manager.getEffectiveConfiguration(instance); + assertThat(cached.configuration().cassandraYaml().getInteger("concurrent_reads")).isEqualTo(128); + assertThat(cached.configuration().cassandraYaml().getString("cluster_name")).isEqualTo("Test Cluster"); + } + + @Test + void testGetEffectiveConfigurationFailPolicyThrows() + { + ToggleableProvider delegate = new ToggleableProvider(); + delegate.setFailing(true); + InstanceMetadata instance = mockInstance(1); + + ConfigurationManager manager = new ConfigurationManager(delegate, BASE_TEMPLATE, tempDir, + FailurePolicy.FAIL); + + assertThatThrownBy(() -> manager.getEffectiveConfiguration(instance)) + .isInstanceOf(ConfigurationProviderUnavailableException.class) + .hasMessageContaining("FAIL policy"); + } + + @Test + void testPatchCachedReadOnlyRejectsWriteWhenProviderUnavailable() + { + ToggleableProvider delegate = new ToggleableProvider(); + InstanceMetadata instance = mockInstance(1); + + ConfigurationManager manager = new ConfigurationManager(delegate, BASE_TEMPLATE, tempDir, + FailurePolicy.CACHED_READ_ONLY); + + // Get the base hash while delegate is up + String baseHash = manager.getEffectiveConfiguration(instance).hash(); + + // Fail delegate, then patch should fail + delegate.setFailing(true); + + List ops = List.of( + new ConfigurationPatchOperation(ConfigurationPatchOperation.Op.ADD, + "/configuration/cassandraYaml/concurrent_reads", 64)); + assertThatThrownBy(() -> manager.patchConfiguration(instance, baseHash, ops)) + .isInstanceOf(ConfigurationManagerException.class); + } + + @Test + void testPatchCachedReadWriteFallsBackToCache() + { + ToggleableProvider delegate = new ToggleableProvider(); + InstanceMetadata instance = mockInstance(1); + + ConfigurationManager manager = new ConfigurationManager(delegate, BASE_TEMPLATE, tempDir, + FailurePolicy.CACHED_READ_WRITE); + + // Get the base hash while delegate is up, populate cache + String baseHash = manager.getEffectiveConfiguration(instance).hash(); + + // Fail delegate, patch should still succeed via cache + delegate.setFailing(true); + List ops = List.of( + new ConfigurationPatchOperation(ConfigurationPatchOperation.Op.ADD, + "/configuration/cassandraYaml/concurrent_reads", 64)); + ConfigurationOverlaySnapshot result = manager.patchConfiguration(instance, baseHash, ops); + + assertThat(result.configuration().cassandraYaml().getInteger("concurrent_reads")).isEqualTo(64); + assertThat(result.configuration().cassandraYaml().getString("cluster_name")).isEqualTo("Test Cluster"); + } + + @Test + void testGetEffectiveConfigurationCachedReadOnlyNoCacheIsUnavailable() + { + ToggleableProvider delegate = new ToggleableProvider(); + delegate.setFailing(true); + InstanceMetadata instance = mockInstance(1); + + ConfigurationManager manager = new ConfigurationManager(delegate, BASE_TEMPLATE, tempDir, + FailurePolicy.CACHED_READ_ONLY); + + // No cache exists and the provider is down: the overlay is unknown, so the base template + // alone is not the effective configuration and must not be passed off as it. + assertThatThrownBy(() -> manager.getEffectiveConfiguration(instance)) + .isInstanceOf(ConfigurationProviderUnavailableException.class) + .hasMessageContaining("no overlay is cached"); + } + private static InstanceMetadata mockInstance(int id) { InstanceMetadata instance = mock(InstanceMetadata.class); when(instance.id()).thenReturn(id); return instance; } + + /** + * A provider that can be toggled between success and failure modes. + */ + private static class ToggleableProvider implements ConfigurationProvider + { + private final InMemoryConfigurationProvider inner = new InMemoryConfigurationProvider(); + private final AtomicBoolean failing = new AtomicBoolean(false); + + void setFailing(boolean fail) + { + failing.set(fail); + } + + @Override + public ConfigurationOverlaySnapshot getOverlay(InstanceMetadata instance) + { + if (failing.get()) + { + throw new UncheckedIOException(new IOException("provider unavailable")); + } + return inner.getOverlay(instance); + } + + @Override + public boolean storeOverlay(InstanceMetadata instance, String originalHash, + @NotNull ConfigurationOverlaySnapshot newSnapshot) + { + if (failing.get()) + { + throw new UncheckedIOException(new IOException("provider unavailable")); + } + return inner.storeOverlay(instance, originalHash, newSnapshot); + } + } } diff --git a/server/src/test/java/org/apache/cassandra/sidecar/configmanagement/FailurePolicyWrapperTest.java b/server/src/test/java/org/apache/cassandra/sidecar/configmanagement/FailurePolicyWrapperTest.java new file mode 100644 index 000000000..506cf0973 --- /dev/null +++ b/server/src/test/java/org/apache/cassandra/sidecar/configmanagement/FailurePolicyWrapperTest.java @@ -0,0 +1,620 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.sidecar.configmanagement; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.file.Path; +import java.time.Instant; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import io.vertx.core.json.JsonObject; +import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Tests for {@link FailurePolicyWrapper} + */ +class FailurePolicyWrapperTest +{ + @TempDir + Path tempDir; + + private InstanceMetadata instance; + + @BeforeEach + void setUp() + { + instance = mockInstance(1); + } + + // -- Factory method tests -- + + @Test + void testWrapReturnsUnwrappedFileBasedProvider() + { + FileBasedConfigurationProvider fileBased = new FileBasedConfigurationProvider(tempDir); + ConfigurationProvider result = FailurePolicyWrapper.wrap(fileBased, tempDir, FailurePolicy.CACHED_READ_ONLY); + assertThat(result).isSameAs(fileBased); + } + + @Test + void testWrapReturnsWrapperForNonFileBased() + { + InMemoryConfigurationProvider inMemory = new InMemoryConfigurationProvider(); + ConfigurationProvider result = FailurePolicyWrapper.wrap(inMemory, tempDir, FailurePolicy.CACHED_READ_ONLY); + assertThat(result).isInstanceOf(FailurePolicyWrapper.class); + } + + // -- Happy path tests (delegate works) -- + + @Test + void testGetOverlayDelegatesSuccessfully() + { + InMemoryConfigurationProvider delegate = new InMemoryConfigurationProvider(); + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 64); + delegate.storeOverlay(instance, null, snapshot); + + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + ConfigurationOverlaySnapshot result = wrapper.getOverlay(instance); + + assertThat(result).isNotNull(); + assertThat(result.configuration().cassandraYaml().getInteger("concurrent_reads")).isEqualTo(64); + } + + @Test + void testGetOverlayPopulatesCache() + { + InMemoryConfigurationProvider delegate = new InMemoryConfigurationProvider(); + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 64); + delegate.storeOverlay(instance, null, snapshot); + + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + wrapper.getOverlay(instance); + + Path cachedFile = tempDir.resolve("1").resolve(FailurePolicyWrapper.CACHED_OVERLAY_FILE_NAME); + assertThat(cachedFile).isRegularFile(); + } + + @Test + void testGetOverlaySkipsCacheWriteWhenContentUnchanged() + { + InMemoryConfigurationProvider delegate = new InMemoryConfigurationProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + + // First fetch populates the cache with a snapshot stamped at t1 + Instant t1 = Instant.parse("2020-01-01T00:00:00Z"); + ConfigurationOverlaySnapshot first = snapshotAt(t1, "concurrent_reads", 64); + delegate.storeOverlay(instance, null, first); + wrapper.getOverlay(instance); + + // Second fetch returns identical content (same hash) but a newer lastModified (t2) + Instant t2 = Instant.parse("2021-06-15T12:00:00Z"); + ConfigurationOverlaySnapshot second = snapshotAt(t2, "concurrent_reads", 64); + delegate.storeOverlay(instance, first.hash(), second); + assertThat(second.hash()).isEqualTo(first.hash()); // hash is content-only, ignores lastModified + wrapper.getOverlay(instance); + + // Because the content hash was unchanged, the cache write was skipped: the persisted snapshot + // still carries t1 rather than being rewritten with t2. + ConfigurationOverlaySnapshot cached = + new FileBasedConfigurationProvider(tempDir, FailurePolicyWrapper.CACHED_OVERLAY_FILE_NAME) + .getOverlay(instance); + assertThat(cached).isNotNull(); + assertThat(cached.lastModified()).isEqualTo(t1); + } + + @Test + void testGetOverlayReturnsNullWhenDelegateReturnsNull() + { + InMemoryConfigurationProvider delegate = new InMemoryConfigurationProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + + assertThat(wrapper.getOverlay(instance)).isNull(); + } + + @Test + void testStoreOverlayDelegatesSuccessfully() + { + InMemoryConfigurationProvider delegate = new InMemoryConfigurationProvider(); + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 64); + + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + boolean stored = wrapper.storeOverlay(instance, null, snapshot); + + assertThat(stored).isTrue(); + assertThat(delegate.getOverlay(instance)).isNotNull(); + } + + @Test + void testStoreOverlayUpdatesCacheOnSuccess() + { + InMemoryConfigurationProvider delegate = new InMemoryConfigurationProvider(); + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 64); + + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + wrapper.storeOverlay(instance, null, snapshot); + + Path cachedFile = tempDir.resolve("1").resolve(FailurePolicyWrapper.CACHED_OVERLAY_FILE_NAME); + assertThat(cachedFile).isRegularFile(); + } + + @Test + void testStoreOverlayDoesNotUpdateCacheOnConflict() + { + InMemoryConfigurationProvider delegate = new InMemoryConfigurationProvider(); + ConfigurationOverlaySnapshot initial = createSnapshot("concurrent_reads", 32); + delegate.storeOverlay(instance, null, initial); + + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + ConfigurationOverlaySnapshot update = createSnapshot("concurrent_reads", 64); + + boolean stored = wrapper.storeOverlay(instance, "sha256:stale", update); + + assertThat(stored).isFalse(); + assertThat(tempDir.resolve("1").resolve(FailurePolicyWrapper.CACHED_OVERLAY_FILE_NAME)).doesNotExist(); + } + + // -- FAIL policy tests -- + + @Test + void testGetOverlayFailPolicyThrowsProviderUnavailable() + { + ConfigurationProvider failing = failingProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(failing, tempDir, FailurePolicy.FAIL); + + assertThatThrownBy(() -> wrapper.getOverlay(instance)) + .isInstanceOf(ConfigurationProviderUnavailableException.class) + .hasMessageContaining("FAIL policy") + .hasCauseInstanceOf(UncheckedIOException.class); + } + + @Test + void testStoreOverlayFailPolicyThrowsProviderUnavailable() + { + ConfigurationProvider failing = failingProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(failing, tempDir, FailurePolicy.FAIL); + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 64); + + assertThatThrownBy(() -> wrapper.storeOverlay(instance, null, snapshot)) + .isInstanceOf(ConfigurationProviderUnavailableException.class) + .hasMessageContaining("FAIL policy") + .hasCauseInstanceOf(UncheckedIOException.class); + } + + // -- CACHED_READ_ONLY policy tests -- + + @Test + void testGetOverlayReadOnlyReturnsCachedWhenDelegateThrows() + { + ToggleableProvider delegate = new ToggleableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 64); + delegate.storeOverlay(instance, null, snapshot); + wrapper.getOverlay(instance); + + delegate.setFailing(true); + ConfigurationOverlaySnapshot cached = wrapper.getOverlay(instance); + + assertThat(cached).isNotNull(); + assertThat(cached.configuration().cassandraYaml().getInteger("concurrent_reads")).isEqualTo(64); + } + + @Test + void testGetOverlayReadOnlyThrowsWhenNoCacheAndDelegateThrows() + { + ConfigurationProvider failing = failingProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(failing, tempDir, FailurePolicy.CACHED_READ_ONLY); + + assertThatThrownBy(() -> wrapper.getOverlay(instance)) + .isInstanceOf(ConfigurationProviderUnavailableException.class) + .hasMessageContaining("no overlay is cached") + .hasCauseInstanceOf(UncheckedIOException.class); + } + + @Test + void testStoreOverlayReadOnlyRejectsWriteWhenDelegateThrows() + { + ToggleableProvider delegate = new ToggleableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + + delegate.setFailing(true); + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 64); + + assertThatThrownBy(() -> wrapper.storeOverlay(instance, null, snapshot)) + .isInstanceOf(ConfigurationProviderUnavailableException.class) + .hasMessageContaining("CACHED_READ_ONLY") + .hasCauseInstanceOf(UncheckedIOException.class); + } + + // -- CACHED_READ_WRITE policy tests -- + + @Test + void testGetOverlayReadWriteReturnsCachedWhenDelegateThrows() + { + ToggleableProvider delegate = new ToggleableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_WRITE); + + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 64); + delegate.storeOverlay(instance, null, snapshot); + wrapper.getOverlay(instance); + + delegate.setFailing(true); + ConfigurationOverlaySnapshot cached = wrapper.getOverlay(instance); + + assertThat(cached).isNotNull(); + assertThat(cached.configuration().cassandraYaml().getInteger("concurrent_reads")).isEqualTo(64); + } + + @Test + void testStoreOverlayReadWriteWritesToCacheWhenDelegateThrows() + { + ToggleableProvider delegate = new ToggleableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_WRITE); + + delegate.setFailing(true); + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 64); + + boolean stored = wrapper.storeOverlay(instance, null, snapshot); + + assertThat(stored).isTrue(); + Path cachedFile = tempDir.resolve("1").resolve(FailurePolicyWrapper.CACHED_OVERLAY_FILE_NAME); + assertThat(cachedFile).isRegularFile(); + } + + @Test + void testStoreOverlayReadWriteReturnsFalseOnCacheHashMismatch() + { + ToggleableProvider delegate = new ToggleableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_WRITE); + + // Seed the cache with an initial value via a successful call + ConfigurationOverlaySnapshot initial = createSnapshot("concurrent_reads", 32); + delegate.storeOverlay(instance, null, initial); + wrapper.getOverlay(instance); + + delegate.setFailing(true); + ConfigurationOverlaySnapshot update = createSnapshot("concurrent_reads", 64); + + // originalHash doesn't match the cached overlay's hash + boolean stored = wrapper.storeOverlay(instance, "sha256:stale", update); + assertThat(stored).isFalse(); + } + + // -- Cache behavior across transitions -- + + @Test + void testCacheUpdatedAfterSuccessfulGetThenServesOnFailure() + { + ToggleableProvider delegate = new ToggleableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + + // Store initial, fetch to populate cache + ConfigurationOverlaySnapshot v1 = createSnapshot("concurrent_reads", 32); + delegate.storeOverlay(instance, null, v1); + wrapper.getOverlay(instance); + + // Update to v2, fetch to update cache + ConfigurationOverlaySnapshot v2 = createSnapshot("concurrent_reads", 64); + delegate.storeOverlay(instance, v1.hash(), v2); + wrapper.getOverlay(instance); + + // Fail delegate, should get v2 from cache + delegate.setFailing(true); + ConfigurationOverlaySnapshot cached = wrapper.getOverlay(instance); + + assertThat(cached).isNotNull(); + assertThat(cached.configuration().cassandraYaml().getInteger("concurrent_reads")).isEqualTo(64); + } + + @Test + void testCacheUpdatedAfterSuccessfulStoreThenServesOnFailure() + { + ToggleableProvider delegate = new ToggleableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 128); + wrapper.storeOverlay(instance, null, snapshot); + + delegate.setFailing(true); + ConfigurationOverlaySnapshot cached = wrapper.getOverlay(instance); + + assertThat(cached).isNotNull(); + assertThat(cached.configuration().cassandraYaml().getInteger("concurrent_reads")).isEqualTo(128); + } + + @Test + void testCacheStoresEmptyOverlayWhenDelegateReturnsNull() + { + // A null delegate result means no overlay exists. The wrapper caches an empty overlay so a + // deleted overlay is not resurrected from cache during a later outage. + ControllableProvider delegate = new ControllableProvider(); + delegate.set(null); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + + assertThat(wrapper.getOverlay(instance)).isNull(); + + // An empty overlay is written to the cache + Path cachedFile = tempDir.resolve("1").resolve(FailurePolicyWrapper.CACHED_OVERLAY_FILE_NAME); + assertThat(cachedFile).isRegularFile(); + ConfigurationOverlaySnapshot cached = + new FileBasedConfigurationProvider(tempDir, FailurePolicyWrapper.CACHED_OVERLAY_FILE_NAME) + .getOverlay(instance); + assertThat(cached).isNotNull(); + assertThat(cached.configuration().cassandraYaml()).isEqualTo(new JsonObject()); + assertThat(cached.configuration().extraJvmOpts()).isEmpty(); + } + + @Test + void testDeletedUpstreamOverlayNotResurrectedFromCacheDuringOutage() + { + ControllableProvider delegate = new ControllableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + + // Seed the cache with a real overlay via a successful read + delegate.set(createSnapshot("concurrent_reads", 64)); + assertThat(wrapper.getOverlay(instance)).isNotNull(); + + // Overlay is deleted upstream: a successful read now returns null and overwrites the cache + delegate.set(null); + assertThat(wrapper.getOverlay(instance)).isNull(); + + // During a later outage, the cache serves the empty overlay, not the stale deleted value + delegate.setFailing(true); + ConfigurationOverlaySnapshot cached = wrapper.getOverlay(instance); + assertThat(cached).isNotNull(); + assertThat(cached.configuration().cassandraYaml().getInteger("concurrent_reads")).isNull(); + } + + @Test + void testStoreOverlayReadWriteWithExtraJvmOpts() + { + ToggleableProvider delegate = new ToggleableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_WRITE); + + delegate.setFailing(true); + + JsonObject yaml = new JsonObject().put("concurrent_reads", 64); + Map jvmOpts = new LinkedHashMap<>(); + jvmOpts.put("-Xmx", "4G"); + CassandraConfigurationOverlay overlay = new CassandraConfigurationOverlay(yaml, jvmOpts); + ConfigurationOverlaySnapshot snapshot = new ConfigurationOverlaySnapshot(Instant.now(), overlay); + + boolean stored = wrapper.storeOverlay(instance, null, snapshot); + assertThat(stored).isTrue(); + + ConfigurationOverlaySnapshot cached = wrapper.getOverlay(instance); + assertThat(cached).isNotNull(); + assertThat(cached.configuration().extraJvmOpts()).containsEntry("-Xmx", "4G"); + assertThat(cached.configuration().cassandraYaml().getInteger("concurrent_reads")).isEqualTo(64); + } + + // -- Failure classification tests -- + + @Test + void testGetOverlayPropagatesRejectionInsteadOfFallingBackToCache() + { + ControllableProvider delegate = new ControllableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_ONLY); + + // Seed the cache with a successful read so a fallback would have something to return + delegate.set(createSnapshot("concurrent_reads", 64)); + assertThat(wrapper.getOverlay(instance)).isNotNull(); + + delegate.setFailure(new IllegalArgumentException("unsupported instance")); + + assertThatThrownBy(() -> wrapper.getOverlay(instance)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("unsupported instance"); + } + + @Test + void testGetOverlayFailPolicyPropagatesRejectionUnwrapped() + { + // A rejection must not be relabeled as an outage, which would surface as a 503 + ControllableProvider delegate = new ControllableProvider(); + delegate.setFailure(new IllegalArgumentException("unsupported instance")); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.FAIL); + + assertThatThrownBy(() -> wrapper.getOverlay(instance)) + .isInstanceOf(IllegalArgumentException.class) + .isNotInstanceOf(ConfigurationProviderUnavailableException.class); + } + + @Test + void testStoreOverlayReadWriteDoesNotCacheRejectedWrite() + { + ControllableProvider delegate = new ControllableProvider(); + FailurePolicyWrapper wrapper = new FailurePolicyWrapper(delegate, tempDir, FailurePolicy.CACHED_READ_WRITE); + + delegate.setFailure(new IllegalArgumentException("invalid overlay")); + ConfigurationOverlaySnapshot snapshot = createSnapshot("concurrent_reads", 64); + + assertThatThrownBy(() -> wrapper.storeOverlay(instance, null, snapshot)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("invalid overlay"); + + // The rejected snapshot was neither persisted nor reported as stored + assertThat(tempDir.resolve("1").resolve(FailurePolicyWrapper.CACHED_OVERLAY_FILE_NAME)).doesNotExist(); + } + + @Test + void testIsUnavailableClassification() + { + assertThat(FailurePolicyWrapper.isUnavailable(new UncheckedIOException(new IOException("down")))).isTrue(); + assertThat(FailurePolicyWrapper.isUnavailable(new ProviderClientException("down"))).isTrue(); + assertThat(FailurePolicyWrapper.isUnavailable( + new ConfigurationProviderUnavailableException("down", null))).isTrue(); + + assertThat(FailurePolicyWrapper.isUnavailable(new IllegalArgumentException("bad"))).isFalse(); + assertThat(FailurePolicyWrapper.isUnavailable(new NullPointerException())).isFalse(); + assertThat(FailurePolicyWrapper.isUnavailable(new IllegalStateException("bug"))).isFalse(); + assertThat(FailurePolicyWrapper.isUnavailable(new UnsupportedOperationException("nope"))).isFalse(); + assertThat(FailurePolicyWrapper.isUnavailable( + new ConfigurationConflictException("sha256:a", "sha256:b"))).isFalse(); + assertThat(FailurePolicyWrapper.isUnavailable( + new ConfigurationPatchException("bad patch", null))).isFalse(); + } + + // -- Helpers -- + + private static ConfigurationOverlaySnapshot createSnapshot(String field, int value) + { + JsonObject yaml = new JsonObject().put(field, value); + CassandraConfigurationOverlay overlay = new CassandraConfigurationOverlay(yaml, null); + return new ConfigurationOverlaySnapshot(Instant.now(), overlay); + } + + private static ConfigurationOverlaySnapshot snapshotAt(Instant lastModified, String field, int value) + { + JsonObject yaml = new JsonObject().put(field, value); + CassandraConfigurationOverlay overlay = new CassandraConfigurationOverlay(yaml, null); + return new ConfigurationOverlaySnapshot(lastModified, overlay); + } + + private static InstanceMetadata mockInstance(int id) + { + InstanceMetadata inst = mock(InstanceMetadata.class); + when(inst.id()).thenReturn(id); + return inst; + } + + private static ConfigurationProvider failingProvider() + { + return new ConfigurationProvider() + { + @Override + public ConfigurationOverlaySnapshot getOverlay(InstanceMetadata instance) + { + throw new UncheckedIOException(new IOException("provider unavailable")); + } + + @Override + public boolean storeOverlay(InstanceMetadata instance, String originalHash, + ConfigurationOverlaySnapshot newSnapshot) + { + throw new UncheckedIOException(new IOException("provider unavailable")); + } + }; + } + + /** + * A provider backed by {@link InMemoryConfigurationProvider} that can be toggled to fail. + */ + private static class ToggleableProvider implements ConfigurationProvider + { + private final InMemoryConfigurationProvider inner = new InMemoryConfigurationProvider(); + private final AtomicBoolean failing = new AtomicBoolean(false); + + void setFailing(boolean fail) + { + failing.set(fail); + } + + @Override + public ConfigurationOverlaySnapshot getOverlay(InstanceMetadata instance) + { + if (failing.get()) + { + throw new UncheckedIOException(new IOException("provider unavailable")); + } + return inner.getOverlay(instance); + } + + @Override + public boolean storeOverlay(InstanceMetadata instance, String originalHash, + ConfigurationOverlaySnapshot newSnapshot) + { + if (failing.get()) + { + throw new UncheckedIOException(new IOException("provider unavailable")); + } + return inner.storeOverlay(instance, originalHash, newSnapshot); + } + } + + /** + * A provider whose current overlay can be set directly (including to {@code null} to simulate an + * upstream deletion) and which can be toggled to fail, for testing cache invalidation behavior. + */ + private static class ControllableProvider implements ConfigurationProvider + { + private volatile ConfigurationOverlaySnapshot current; + private volatile RuntimeException failure; + + void set(ConfigurationOverlaySnapshot snapshot) + { + this.current = snapshot; + } + + void setFailing(boolean fail) + { + this.failure = fail ? new UncheckedIOException(new IOException("provider unavailable")) : null; + } + + void setFailure(RuntimeException failure) + { + this.failure = failure; + } + + @Override + public ConfigurationOverlaySnapshot getOverlay(InstanceMetadata instance) + { + if (failure != null) + { + throw failure; + } + return current; + } + + @Override + public boolean storeOverlay(InstanceMetadata instance, String originalHash, + ConfigurationOverlaySnapshot newSnapshot) + { + if (failure != null) + { + throw failure; + } + this.current = newSnapshot; + return true; + } + } + + /** + * Stands in for a remote provider's own client exception type, which the wrapper cannot enumerate. + */ + private static class ProviderClientException extends RuntimeException + { + ProviderClientException(String message) + { + super(message); + } + } +}