Register object-storage caches with ClusterCaches (multi-node) #27
No files matched your search
@@ -31,6 +31,8 @@ All notable changes to this project are documented here. The format follows Keep
|
||||
- None for the health migration; both Test connection record actions, including on-demand write tests, are retained.
|
||||
|
||||
### Changed
|
||||
- Multi-node deployments: the runtime caches are registered with QQQ `ClusterCaches`. Shared (cleared on every node): the bucket rows' presign limits (`qbit-object-storage.bucket-row-limits`, on a bucket row insert or update), the enabled header rules (`qbit-object-storage.header-rules`, on a header rule insert, update or delete), both through the new `ConfigRowChangeCustomizer`, and the capability statuses (`qbit-object-storage.capabilities`: a probe result recorded on one node clears that capability on the others, which reload it from the capability row; before, other nodes kept the old status until restart). The 15 s TTLs remain the safety net. Node-local, with reasons: the alias and connection row-id maps, the S3 client map, the process-permission map, the provider defaults, the running-scan guard and the per-pod presign rate limiter (`StorageBucket.PRESIGN_COUNTS`; per-node rate limits are accepted for now). `ObjectStorageClusterCachesTest` audits the package for unregistered static caches and covers remote and local invalidation.
|
||||
- Builds against QQQ `4.1.0-SNAPSHOT` (`qqq.snapshot.version`) for `ClusterCaches` until the next QQQ RC ships; a release must pin that RC.
|
||||
- Read the qbit metadata version from the Maven-filtered artifact version so release metadata matches the build and help bundle.
|
||||
- Rechecked the released RC.8 JARs: core still has no S3 session-token/provider hook or file-reference interpreter. Retained both compatibility shims and moved their review deadline to RC.9 under issue #10.
|
||||
- Raised the QQQ BOM baseline to released `4.1.0-RC.8` for the central health API.
|
||||
|
||||
@@ -46,6 +46,7 @@ import com.kingsrook.qbits.objectstorage.internal.PostPolicySigner;
|
||||
import com.kingsrook.qbits.objectstorage.internal.ProtectionRules;
|
||||
import com.kingsrook.qbits.objectstorage.internal.RateLimiter;
|
||||
import com.kingsrook.qqq.backend.core.actions.permissions.CorePermissionEnforcer;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.utils.StringUtils;
|
||||
@@ -96,7 +97,8 @@ public final class StorageBucket
|
||||
{
|
||||
private static final QLogger LOG = QLogger.getLogger(StorageBucket.class);
|
||||
|
||||
private static final Map<String, long[]> PRESIGN_COUNTS = new ConcurrentHashMap<>();
|
||||
private static final Map<String, long[]> PRESIGN_COUNTS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
|
||||
"per-pod presign rate limiter; per-node rate limits are accepted for now (cluster HA spec, owner decision 4)");
|
||||
|
||||
private final StorageBucketConfig bucket;
|
||||
private final ConnectionRuntime connection;
|
||||
|
||||
@@ -43,6 +43,7 @@ import com.kingsrook.qbits.objectstorage.internal.ConfigSync;
|
||||
import com.kingsrook.qbits.objectstorage.internal.Db;
|
||||
import com.kingsrook.qbits.objectstorage.internal.ObjectStorageRuntime;
|
||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
|
||||
@@ -78,7 +79,8 @@ public final class ObjectIndex
|
||||
|
||||
public static final int PAGE_SIZE = 1000;
|
||||
|
||||
private static final Set<String> RUNNING = java.util.concurrent.ConcurrentHashMap.newKeySet();
|
||||
private static final Set<String> RUNNING = ClusterCaches.nodeLocal(java.util.concurrent.ConcurrentHashMap.newKeySet(),
|
||||
"the buckets this node is scanning now (an in-flight guard, not cached data)");
|
||||
|
||||
/** Test seam: simulate a crash after this many pages of a scan (so resume can be tested). */
|
||||
static volatile Integer failAfterPages = null;
|
||||
|
||||
@@ -30,6 +30,7 @@ import com.kingsrook.qbits.objectstorage.api.Capability;
|
||||
import com.kingsrook.qbits.objectstorage.api.CapabilityStatus;
|
||||
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
|
||||
@@ -47,7 +48,11 @@ public final class CapabilityStore
|
||||
public static final Set<Capability> CONNECTION_SCOPE = EnumSet.of(Capability.LIST_V2, Capability.PRESIGNED_GET, Capability.PRESIGNED_PUT,
|
||||
Capability.PRESIGNED_POST, Capability.USAGE_ADMIN_API, Capability.QUOTA_ADMIN_API);
|
||||
|
||||
private static final Map<String, CapabilityStatus> CACHE = new ConcurrentHashMap<>();
|
||||
/** Cluster cache name of the capability statuses (a probe result recorded on one node clears that key on every node). */
|
||||
public static final String CACHE_NAME = "qbit-object-storage.capabilities";
|
||||
|
||||
private static final Map<String, CapabilityStatus> CACHE = ClusterCaches.shared(CACHE_NAME, new ConcurrentHashMap<>(),
|
||||
CapabilityStore::clearCache, CapabilityStore::clearKey);
|
||||
private static final Set<String> LOADED = ConcurrentHashMap.newKeySet();
|
||||
|
||||
|
||||
@@ -62,7 +67,7 @@ public final class CapabilityStore
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Drop everything cached.
|
||||
** Drop everything cached on this node.
|
||||
*******************************************************************************/
|
||||
public static void clearCache()
|
||||
{
|
||||
@@ -72,6 +77,28 @@ public final class CapabilityStore
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Drop one cached status on this node; the next read reloads the missing
|
||||
** statuses from the capability rows (cached ones are kept).
|
||||
*******************************************************************************/
|
||||
static void clearKey(String key)
|
||||
{
|
||||
CACHE.remove(key);
|
||||
LOADED.clear();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Whether a status is cached on this node (tests).
|
||||
*******************************************************************************/
|
||||
static boolean isCached(StorageBucketConfig bucket, Capability capability)
|
||||
{
|
||||
return (CACHE.containsKey(scopeKey(bucket, capability)));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The status of a capability for a bucket alias.
|
||||
*******************************************************************************/
|
||||
@@ -112,10 +139,24 @@ public final class CapabilityStore
|
||||
|
||||
/*******************************************************************************
|
||||
** Record one probe result (cache and, when the tables are there, the row).
|
||||
** The row is written first, then the key is cleared on every node (they
|
||||
** reload it from the row), then cached here.
|
||||
*******************************************************************************/
|
||||
public static void record(StorageBucketConfig bucket, Capability capability, CapabilityStatus status, String detail)
|
||||
{
|
||||
CACHE.put(scopeKey(bucket, capability), status);
|
||||
String key = scopeKey(bucket, capability);
|
||||
persist(bucket, capability, status, detail);
|
||||
ClusterCaches.invalidateKey(CACHE_NAME, key);
|
||||
CACHE.put(key, status);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Write one probe result to its capability row; never throws.
|
||||
*******************************************************************************/
|
||||
private static void persist(StorageBucketConfig bucket, Capability capability, CapabilityStatus status, String detail)
|
||||
{
|
||||
if(!Db.available(ObjectStorageTables.BUCKET_CAPABILITY))
|
||||
{
|
||||
return;
|
||||
|
||||
+95
@@ -0,0 +1,95 @@
|
||||
/*
|
||||
* QQQ - Low-code Application Framework for Engineers.
|
||||
* Copyright (C) 2021-2025. Kingsrook, LLC
|
||||
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
|
||||
* contact@kingsrook.com
|
||||
* https://github.com/Kingsrook/
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qbits.objectstorage.internal;
|
||||
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
|
||||
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Post insert, update and delete of the editable config rows that the runtime
|
||||
** caches: a bucket row (presign limits, ConfigSync) or a header rule
|
||||
** (DefaultObjectHeaderPolicy). Clears that cache on every node, so an admin's
|
||||
** edit takes effect everywhere at once; the short TTLs remain the safety net.
|
||||
*******************************************************************************/
|
||||
public class ConfigRowChangeCustomizer implements TableCustomizerInterface
|
||||
{
|
||||
|
||||
/*******************************************************************************
|
||||
** {@inheritDoc}
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public List<QRecord> postInsert(InsertInput insertInput, List<QRecord> records) throws QException
|
||||
{
|
||||
invalidate(insertInput.getTableName());
|
||||
return (records);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** {@inheritDoc}
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public List<QRecord> postUpdate(UpdateInput updateInput, List<QRecord> records, Optional<List<QRecord>> oldRecordList) throws QException
|
||||
{
|
||||
invalidate(updateInput.getTableName());
|
||||
return (records);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** {@inheritDoc}
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public List<QRecord> postDelete(DeleteInput deleteInput, List<QRecord> records) throws QException
|
||||
{
|
||||
invalidate(deleteInput.getTableName());
|
||||
return (records);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Clear the cache that holds the table's rows, on every node.
|
||||
*******************************************************************************/
|
||||
static void invalidate(String tableName)
|
||||
{
|
||||
if(ObjectStorageTables.BUCKET.equals(tableName))
|
||||
{
|
||||
ClusterCaches.invalidate(ConfigSync.ROW_LIMITS_CACHE_NAME);
|
||||
}
|
||||
else if(ObjectStorageTables.HEADER_RULE.equals(tableName))
|
||||
{
|
||||
ClusterCaches.invalidate(DefaultObjectHeaderPolicy.CACHE_NAME);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -30,6 +30,7 @@ import com.kingsrook.qbits.objectstorage.config.CredentialsMode;
|
||||
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
||||
import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
@@ -48,10 +49,16 @@ public final class ConfigSync
|
||||
{
|
||||
private static final QLogger LOG = QLogger.getLogger(ConfigSync.class);
|
||||
|
||||
private static final Map<String, Integer> BUCKET_IDS = new ConcurrentHashMap<>();
|
||||
private static final Map<String, Integer> CONNECTION_IDS = new ConcurrentHashMap<>();
|
||||
/** Cluster cache name of the bucket rows' presign limits (cleared on every node when a bucket row changes). */
|
||||
public static final String ROW_LIMITS_CACHE_NAME = "qbit-object-storage.bucket-row-limits";
|
||||
|
||||
private static final Map<String, RowLimits> ROW_LIMITS = new ConcurrentHashMap<>();
|
||||
private static final Map<String, Integer> BUCKET_IDS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
|
||||
"bucket alias to row id, written by the config sync; bucket rows cannot be inserted or deleted through the UI or API, so the ids are the same on every node");
|
||||
private static final Map<String, Integer> CONNECTION_IDS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
|
||||
"connection name to row id, written by the config sync; connection rows cannot be inserted or deleted through the UI or API, so the ids are the same on every node");
|
||||
|
||||
private static final Map<String, RowLimits> ROW_LIMITS = ClusterCaches.shared(ROW_LIMITS_CACHE_NAME, new ConcurrentHashMap<>(),
|
||||
ConfigSync::clearRowLimits, alias -> ConfigSync.ROW_LIMITS.remove(alias));
|
||||
private static final long ROW_LIMIT_CACHE_MILLIS = 15_000L;
|
||||
|
||||
private static volatile boolean synced = false;
|
||||
@@ -313,7 +320,8 @@ public final class ConfigSync
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Forget the cached row limits (after a row edit in a test; they also expire).
|
||||
** Forget the cached row limits on this node (a bucket row edit clears them on
|
||||
** every node through ConfigRowChangeCustomizer; they also expire).
|
||||
*******************************************************************************/
|
||||
public static void clearRowLimits()
|
||||
{
|
||||
|
||||
+23
-9
@@ -23,12 +23,14 @@ package com.kingsrook.qbits.objectstorage.internal;
|
||||
|
||||
import java.util.Comparator;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.regex.Pattern;
|
||||
import com.kingsrook.qbits.objectstorage.api.ObjectHeaderPolicy;
|
||||
import com.kingsrook.qbits.objectstorage.api.ObjectHeaders;
|
||||
import com.kingsrook.qbits.objectstorage.config.BucketRole;
|
||||
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.utils.StringUtils;
|
||||
@@ -44,19 +46,31 @@ public class DefaultObjectHeaderPolicy implements ObjectHeaderPolicy
|
||||
{
|
||||
private static final QLogger LOG = QLogger.getLogger(DefaultObjectHeaderPolicy.class);
|
||||
|
||||
/** Cluster cache name of the enabled header rules (cleared on every node when a rule changes). */
|
||||
public static final String CACHE_NAME = "qbit-object-storage.header-rules";
|
||||
|
||||
private static final long CACHE_MILLIS = 15_000L;
|
||||
|
||||
private static volatile List<QRecord> cachedRules;
|
||||
private static volatile long cachedAt;
|
||||
private static final AtomicReference<CachedRules> CACHED = ClusterCaches.shared(CACHE_NAME, new AtomicReference<>(), DefaultObjectHeaderPolicy::clearCache, null);
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Drop cached rules (after the rules change).
|
||||
** The enabled rules and when they were read.
|
||||
*******************************************************************************/
|
||||
private record CachedRules(List<QRecord> rules, long readAt)
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Drop the cached rules on this node (a rule edit clears them on every node
|
||||
** through ConfigRowChangeCustomizer).
|
||||
*******************************************************************************/
|
||||
public static void clearCache()
|
||||
{
|
||||
cachedRules = null;
|
||||
CACHED.set(null);
|
||||
}
|
||||
|
||||
|
||||
@@ -119,11 +133,12 @@ public class DefaultObjectHeaderPolicy implements ObjectHeaderPolicy
|
||||
*******************************************************************************/
|
||||
private static List<QRecord> rules()
|
||||
{
|
||||
List<QRecord> rules = cachedRules;
|
||||
if(rules != null && System.currentTimeMillis() - cachedAt < CACHE_MILLIS)
|
||||
CachedRules cached = CACHED.get();
|
||||
if(cached != null && System.currentTimeMillis() - cached.readAt() < CACHE_MILLIS)
|
||||
{
|
||||
return (rules);
|
||||
return (cached.rules());
|
||||
}
|
||||
List<QRecord> rules;
|
||||
if(!Db.available(ObjectStorageTables.HEADER_RULE))
|
||||
{
|
||||
return (List.of());
|
||||
@@ -133,8 +148,7 @@ public class DefaultObjectHeaderPolicy implements ObjectHeaderPolicy
|
||||
rules = Db.asSystem(() -> Db.query(ObjectStorageTables.HEADER_RULE, "enabled", true));
|
||||
rules.sort(Comparator.comparing((QRecord r) -> r.getValueInteger("priority") == null ? 100 : r.getValueInteger("priority"))
|
||||
.thenComparing(r -> r.getValueInteger("id") == null ? 0 : r.getValueInteger("id")));
|
||||
cachedRules = rules;
|
||||
cachedAt = System.currentTimeMillis();
|
||||
CACHED.set(new CachedRules(rules, System.currentTimeMillis()));
|
||||
return (rules);
|
||||
}
|
||||
catch(Exception e)
|
||||
|
||||
@@ -33,6 +33,7 @@ import com.kingsrook.qbits.objectstorage.api.StorageAlertSink;
|
||||
import com.kingsrook.qbits.objectstorage.api.StorageChangeListener;
|
||||
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
||||
import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
@@ -43,9 +44,11 @@ import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
||||
*******************************************************************************/
|
||||
public final class ObjectStorageRuntime
|
||||
{
|
||||
private static final Map<String, ConnectionRuntime> CONNECTIONS = new ConcurrentHashMap<>();
|
||||
private static final Map<String, ConnectionRuntime> CONNECTIONS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
|
||||
"this node's S3 clients, built from the code config at first use; row overrides apply when a client is built, the same on every node");
|
||||
private static final List<StorageChangeListener> LISTENERS = new CopyOnWriteArrayList<>();
|
||||
private static final Map<String, String> PROCESS_PERMISSIONS = new ConcurrentHashMap<>();
|
||||
private static final Map<String, String> PROCESS_PERMISSIONS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
|
||||
"built from the qbit config at startup, the same on every node");
|
||||
|
||||
private static volatile ObjectStorageQBitConfig config;
|
||||
private static volatile ObjectHeaderPolicy headerPolicy;
|
||||
|
||||
@@ -26,6 +26,7 @@ import java.util.Map;
|
||||
import com.kingsrook.qbits.objectstorage.api.Capability;
|
||||
import com.kingsrook.qbits.objectstorage.api.CapabilityStatus;
|
||||
import com.kingsrook.qbits.objectstorage.config.ProviderType;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||
import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.PARTIAL;
|
||||
import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.SUPPORTED;
|
||||
import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.UNKNOWN;
|
||||
@@ -39,7 +40,8 @@ import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.UNSUPPORTED
|
||||
*******************************************************************************/
|
||||
public final class ProviderDefaults
|
||||
{
|
||||
private static final Map<ProviderType, Map<Capability, CapabilityStatus>> TABLE = new EnumMap<>(ProviderType.class);
|
||||
private static final Map<ProviderType, Map<Capability, CapabilityStatus>> TABLE = ClusterCaches.nodeLocal(new EnumMap<>(ProviderType.class),
|
||||
"provider defaults from code, the same on every node");
|
||||
|
||||
static
|
||||
{
|
||||
|
||||
@@ -25,6 +25,7 @@ import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig;
|
||||
import com.kingsrook.qbits.objectstorage.internal.ConfigRowChangeCustomizer;
|
||||
import com.kingsrook.qbits.objectstorage.permissions.ObjectStoragePermissionChecker;
|
||||
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
|
||||
import com.kingsrook.qqq.backend.core.actions.dashboard.widgets.ChildRecordListRenderer;
|
||||
@@ -564,7 +565,9 @@ public final class ObjectStorageTables
|
||||
.withField(ro("orphanedCount", QFieldType.LONG))
|
||||
.withField(ro("orphanedBytes", QFieldType.LONG))
|
||||
.withField(ro("missingCount", QFieldType.LONG))
|
||||
.withField(ro("lastReferenceScanDate", QFieldType.DATE_TIME)), KEY_TYPE_BUCKET, "id", false));
|
||||
.withField(ro("lastReferenceScanDate", QFieldType.DATE_TIME)), KEY_TYPE_BUCKET, "id", false)
|
||||
.withCustomizer(TableCustomizers.POST_INSERT_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class))
|
||||
.withCustomizer(TableCustomizers.POST_UPDATE_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class)));
|
||||
}
|
||||
|
||||
|
||||
@@ -701,7 +704,10 @@ public final class ObjectStorageTables
|
||||
.withField(str("cacheControl", 300))
|
||||
.withField(str("contentDisposition", 300))
|
||||
.withField(new QFieldMetaData("priority", QFieldType.INTEGER).withDefaultValue(100))
|
||||
.withField(new QFieldMetaData("enabled", QFieldType.BOOLEAN).withDefaultValue(true)), KEY_TYPE_BUCKET, "bucketId", true));
|
||||
.withField(new QFieldMetaData("enabled", QFieldType.BOOLEAN).withDefaultValue(true)), KEY_TYPE_BUCKET, "bucketId", true)
|
||||
.withCustomizer(TableCustomizers.POST_INSERT_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class))
|
||||
.withCustomizer(TableCustomizers.POST_UPDATE_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class))
|
||||
.withCustomizer(TableCustomizers.POST_DELETE_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class)));
|
||||
}
|
||||
|
||||
|
||||
|
||||
+240
@@ -0,0 +1,240 @@
|
||||
/*
|
||||
* QQQ - Low-code Application Framework for Engineers.
|
||||
* Copyright (C) 2021-2026. Kingsrook, LLC
|
||||
* 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
|
||||
* contact@kingsrook.com
|
||||
* https://github.com/Kingsrook/
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* https://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package com.kingsrook.qbits.objectstorage.internal;
|
||||
|
||||
|
||||
import java.util.EnumSet;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
import java.util.function.Consumer;
|
||||
import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig;
|
||||
import com.kingsrook.qbits.objectstorage.api.Capability;
|
||||
import com.kingsrook.qbits.objectstorage.api.CapabilityStatus;
|
||||
import com.kingsrook.qbits.objectstorage.config.BucketRole;
|
||||
import com.kingsrook.qbits.objectstorage.config.ProviderType;
|
||||
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
||||
import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||
import com.kingsrook.qbits.objectstorage.testsupport.StorageTestSupport;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterBus;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterBusInterface;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterBusMetaData;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterCacheAudit;
|
||||
import com.kingsrook.qqq.backend.core.cluster.ClusterEvent;
|
||||
import com.kingsrook.qqq.backend.core.cluster.QNodeRoles;
|
||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** The Object Storage caches are correct on several nodes: every static cache
|
||||
** is registered with ClusterCaches, a probe result recorded on one node clears
|
||||
** that capability on the others, and an edited bucket row or header rule
|
||||
** clears its cache on every node.
|
||||
*******************************************************************************/
|
||||
class ObjectStorageClusterCachesTest
|
||||
{
|
||||
/** the "wire": events every node's transport sees */
|
||||
static final List<Consumer<ClusterEvent>> WIRE = new CopyOnWriteArrayList<>();
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A transport over the static WIRE, so a "remote" publish reaches us.
|
||||
*******************************************************************************/
|
||||
public static class WireBus implements ClusterBusInterface
|
||||
{
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public void publish(ClusterEvent event)
|
||||
{
|
||||
WIRE.forEach(listener -> listener.accept(event));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public AutoCloseable subscribe(Consumer<ClusterEvent> listener)
|
||||
{
|
||||
WIRE.add(listener);
|
||||
return (() -> WIRE.remove(listener));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A synced instance whose buckets never touch a real store, on the wire as
|
||||
** node-a.
|
||||
*******************************************************************************/
|
||||
@BeforeEach
|
||||
void setUp() throws Exception
|
||||
{
|
||||
System.setProperty("os.unit.ak", "ak");
|
||||
System.setProperty("os.unit.sk", "sk");
|
||||
StorageTestSupport.setUp(new ObjectStorageQBitConfig().withBackendName(StorageTestSupport.MEMORY).withBucketScoping(false).withOpenWhenAnonymous(true).withAdminApp(true)
|
||||
.withAllowedSecretRefPatterns(List.of("prop.os.unit.*"))
|
||||
.withConnection(new StorageConnectionConfig("main").withProviderType(ProviderType.MINIO).withEndpoint("http://localhost:1").withRegion("us-east-1")
|
||||
.withAccessKeyRef("${prop.os.unit.ak}").withSecretKeyRef("${prop.os.unit.sk}"))
|
||||
.withBucket(new StorageBucketConfig("media").withConnection("main").withBucketName("media-bucket").withRole(BucketRole.PRIVATE).withIndexed(false)),
|
||||
instance -> instance.withSupplementalMetaData(new ClusterBusMetaData().withBus(new QCodeReference(WireBus.class))));
|
||||
QNodeRoles.set(new QNodeRoles.Settings(EnumSet.allOf(QNodeRoles.Role.class), true, true, true, null, "node-a"));
|
||||
ClusterBus.connect(QContext.getQInstance());
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
**
|
||||
*******************************************************************************/
|
||||
@AfterEach
|
||||
void tearDown()
|
||||
{
|
||||
ClusterBus.close(QContext.getQInstance());
|
||||
WIRE.clear();
|
||||
QNodeRoles.set(null);
|
||||
StorageTestSupport.tearDown();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Every static Memoization in the QBit is registered (shared or node-local).
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testNoUnregisteredStaticCaches() throws Exception
|
||||
{
|
||||
assertThat(ClusterCacheAudit.unregisteredStaticCaches("com.kingsrook.qbits.objectstorage")).isEmpty();
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Another node's probe result clears the capability here; the next read
|
||||
** reloads it from the capability row.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testRemoteProbeResultClearsCapability() throws Exception
|
||||
{
|
||||
StorageBucketConfig bucket = ObjectStorageRuntime.bucketConfig("media");
|
||||
CapabilityStore.record(bucket, Capability.VERSIONING, CapabilityStatus.UNSUPPORTED, "probe here");
|
||||
assertThat(CapabilityStore.isCached(bucket, Capability.VERSIONING)).isTrue();
|
||||
|
||||
////////////////////////////////////////////////////////////
|
||||
// node-b probes again and writes SUPPORTED to the row //
|
||||
////////////////////////////////////////////////////////////
|
||||
Integer bucketId = ConfigSync.bucketId("media");
|
||||
Db.asSystem(() ->
|
||||
{
|
||||
QRecord row = Db.first(ObjectStorageTables.BUCKET_CAPABILITY, "bucketId", bucketId, "capability", Capability.VERSIONING.name());
|
||||
Db.update(ObjectStorageTables.BUCKET_CAPABILITY, List.of(new QRecord().withValue("id", row.getValue("id")).withValue("status", CapabilityStatus.SUPPORTED.name())));
|
||||
return (null);
|
||||
});
|
||||
ClusterEvent remote = new ClusterEvent(ClusterEvent.TYPE_MEMO_INVALIDATE, CapabilityStore.CACHE_NAME, "b:media:" + Capability.VERSIONING);
|
||||
remote.setOriginNodeId("node-b");
|
||||
WIRE.forEach(listener -> listener.accept(remote));
|
||||
|
||||
assertThat(CapabilityStore.isCached(bucket, Capability.VERSIONING)).isFalse();
|
||||
assertThat(CapabilityStore.status("media", Capability.VERSIONING)).isEqualTo(CapabilityStatus.SUPPORTED);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A probe result recorded here tells the other nodes, and stays cached here.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testProbeResultPublishesInvalidation()
|
||||
{
|
||||
List<ClusterEvent> sent = sentEvents();
|
||||
StorageBucketConfig bucket = ObjectStorageRuntime.bucketConfig("media");
|
||||
CapabilityStore.record(bucket, Capability.VERSIONING, CapabilityStatus.SUPPORTED, "probe here");
|
||||
|
||||
assertThat(CapabilityStore.status("media", Capability.VERSIONING)).isEqualTo(CapabilityStatus.SUPPORTED);
|
||||
assertThat(sent).anySatisfy(event -> assertInvalidation(event, CapabilityStore.CACHE_NAME, "b:media:VERSIONING"));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** An edited bucket row takes effect at once and tells the other nodes.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testBucketRowEditInvalidatesRowLimitsEverywhere() throws Exception
|
||||
{
|
||||
StorageBucketConfig bucket = ObjectStorageRuntime.bucketConfig("media");
|
||||
assertThat(ConfigSync.effectiveMaxPresignGetSeconds(bucket)).isEqualTo(3600);
|
||||
|
||||
List<ClusterEvent> sent = sentEvents();
|
||||
Db.update(ObjectStorageTables.BUCKET, List.of(new QRecord().withValue("id", ConfigSync.bucketId("media")).withValue("maxPresignGetSeconds", 60)));
|
||||
|
||||
assertThat(ConfigSync.effectiveMaxPresignGetSeconds(bucket)).as("no wait for the 15 s cache").isEqualTo(60);
|
||||
assertThat(sent).anySatisfy(event -> assertInvalidation(event, ConfigSync.ROW_LIMITS_CACHE_NAME, null));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** An edited header rule tells the other nodes.
|
||||
*******************************************************************************/
|
||||
@Test
|
||||
void testHeaderRuleEditInvalidatesRulesEverywhere() throws Exception
|
||||
{
|
||||
List<ClusterEvent> sent = sentEvents();
|
||||
StorageTestSupport.insert(ObjectStorageTables.HEADER_RULE, new QRecord().withValue("name", "all").withValue("keyGlob", "**").withValue("cacheControl", "no-store"));
|
||||
assertThat(sent).anySatisfy(event -> assertInvalidation(event, DefaultObjectHeaderPolicy.CACHE_NAME, null));
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Collect what this node puts on the wire.
|
||||
*******************************************************************************/
|
||||
private static List<ClusterEvent> sentEvents()
|
||||
{
|
||||
List<ClusterEvent> sent = new CopyOnWriteArrayList<>();
|
||||
WIRE.add(sent::add);
|
||||
return (sent);
|
||||
}
|
||||
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** A memo invalidation from node-a for a cache (and key).
|
||||
*******************************************************************************/
|
||||
private static void assertInvalidation(ClusterEvent event, String name, String key)
|
||||
{
|
||||
assertThat(event.getType()).isEqualTo(ClusterEvent.TYPE_MEMO_INVALIDATE);
|
||||
assertThat(event.getName()).isEqualTo(name);
|
||||
assertThat(event.getKey()).isEqualTo(key);
|
||||
assertThat(event.getOriginNodeId()).isEqualTo("node-a");
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user