From 27a931397049fcb56fe83236608594063d1a91d1 Mon Sep 17 00:00:00 2001 From: Jacek Lewandowski Date: Tue, 10 Oct 2023 12:09:44 +0200 Subject: [PATCH] CEP-15: Accord metrics Patch by Jacek Lewandowski, reviewed by Caleb Rackliffe, David Capwell and Henrik Ingo for CASSANDRA-18580 --- modules/accord | 2 +- .../metrics/AccordClientRequestMetrics.java | 4 - .../cassandra/metrics/AccordMetrics.java | 312 ++++++++++++++++++ .../metrics/AccordStateCacheMetrics.java | 50 +++ .../cassandra/metrics/CacheAccessMetrics.java | 73 ++++ .../cassandra/metrics/CacheMetrics.java | 2 +- .../cassandra/metrics/CacheSizeMetrics.java | 63 ++++ .../cassandra/metrics/RatioGaugeSet.java | 121 +++++++ .../service/accord/AccordCommandStore.java | 44 ++- .../service/accord/AccordCommandStores.java | 39 ++- .../service/accord/AccordService.java | 3 +- .../service/accord/AccordStateCache.java | 133 ++++---- .../service/accord/api/AccordAgent.java | 8 + .../test/accord/AccordMetricsTest.java | 273 +++++++++++++++ .../CompactionAccordIteratorsTest.java | 6 +- .../io/sstable/LargePartitionsTest.java | 6 +- .../service/accord/AccordCommandTest.java | 4 +- .../service/accord/AccordStateCacheTest.java | 179 ++++++++-- .../service/accord/AccordTestUtils.java | 6 +- .../service/accord/async/AsyncLoaderTest.java | 4 +- .../accord/async/AsyncOperationTest.java | 8 +- 21 files changed, 1217 insertions(+), 123 deletions(-) create mode 100644 src/java/org/apache/cassandra/metrics/AccordMetrics.java create mode 100644 src/java/org/apache/cassandra/metrics/AccordStateCacheMetrics.java create mode 100644 src/java/org/apache/cassandra/metrics/CacheAccessMetrics.java create mode 100644 src/java/org/apache/cassandra/metrics/CacheSizeMetrics.java create mode 100644 src/java/org/apache/cassandra/metrics/RatioGaugeSet.java create mode 100644 test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java diff --git a/modules/accord b/modules/accord index b1befa3cc0..0419858bd1 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit b1befa3cc0a8496451bb48ec3bb1c0f56b8c7653 +Subproject commit 0419858bd1f6761f08fd1369477f7c142f5bbb4f diff --git a/src/java/org/apache/cassandra/metrics/AccordClientRequestMetrics.java b/src/java/org/apache/cassandra/metrics/AccordClientRequestMetrics.java index c95c3bd11f..a9d1f28c47 100644 --- a/src/java/org/apache/cassandra/metrics/AccordClientRequestMetrics.java +++ b/src/java/org/apache/cassandra/metrics/AccordClientRequestMetrics.java @@ -19,20 +19,17 @@ package org.apache.cassandra.metrics; import com.codahale.metrics.Histogram; -import com.codahale.metrics.Meter; import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; public class AccordClientRequestMetrics extends ClientRequestMetrics { - public final Meter preempts; public final Histogram keySize; public AccordClientRequestMetrics(String scope) { super(scope); - preempts = Metrics.meter(factory.createMetricName("Preempts")); keySize = Metrics.histogram(factory.createMetricName("KeySizeHistogram"), false); } @@ -40,7 +37,6 @@ public class AccordClientRequestMetrics extends ClientRequestMetrics public void release() { super.release(); - Metrics.remove(factory.createMetricName("Preempts")); Metrics.remove(factory.createMetricName("KeySizeHistogram")); } } diff --git a/src/java/org/apache/cassandra/metrics/AccordMetrics.java b/src/java/org/apache/cassandra/metrics/AccordMetrics.java new file mode 100644 index 0000000000..5601b9fa40 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/AccordMetrics.java @@ -0,0 +1,312 @@ +/* + * 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.metrics; + +import java.lang.reflect.Field; +import java.util.concurrent.TimeUnit; + +import accord.api.EventsListener; +import accord.local.Command; +import accord.primitives.Deps; +import accord.primitives.PartialDeps; +import accord.primitives.Timestamp; +import accord.primitives.TxnId; +import com.codahale.metrics.Counting; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.Timer; +import org.apache.cassandra.service.accord.AccordService; + +import static java.util.concurrent.TimeUnit.MICROSECONDS; +import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; + +public class AccordMetrics +{ + public final static AccordMetrics readMetrics = new AccordMetrics("ro"); + public final static AccordMetrics writeMetrics = new AccordMetrics("rw"); + + public static final String COMMIT_LATENCY = "CommitLatency"; + public static final String EXECUTE_LATENCY = "ExecuteLatency"; + public static final String APPLY_LATENCY = "ApplyLatency"; + public static final String APPLY_DURATION = "ApplyDuration"; + public static final String PARTIAL_DEPENDENCIES = "PartialDependencies"; + public static final String PROGRESS_LOG_SIZE = "ProgressLogSize"; + + public static final String DEPENDENCIES = "Dependencies"; + public static final String FAST_PATHS = "FastPaths"; + public static final String SLOW_PATHS = "SlowPaths"; + public static final String PREEMPTS = "Preempts"; + public static final String TIMEOUTS = "Timeouts"; + public static final String INVALIDATIONS = "Invalidations"; + public static final String RECOVERY_DELAY = "RecoveryDelay"; + public static final String RECOVERY_TIME = "RecoveryTime"; + public static final String FAST_PATH_TO_TOTAL = "FastPathToTotal"; + public static final String ACCORD_REPLICA = "accord-replica"; + public static final String ACCORD_COORDINATOR = "accord-coordinator"; + + /** + * The time between start on the coordinator and commit on this replica. + */ + public final Timer commitLatency; + + /** + * The time between start on the coordinator and execution on this replica. + */ + public final Timer executeLatency; + + /** + * The time between start on the coordinator and application on this replica. + */ + public final Timer applyLatency; + + /** + * Duration of applying changes. + */ + public final Timer applyDuration; + + /** + * A histogram of the number of dependencies per partial transaction at this replica. + */ + public final Histogram partialDependencies; + + public final Meter progressLogSize; + + /** + * A histogram of the number of dependencies per transaction at this coordinator. + */ + public final Histogram dependencies; + + /** + * The number of fast path transactions executed on this coordinator. + */ + public final Meter fastPaths; + + /** + * The number of slow path transactions executed on this coordinator. + */ + public final Meter slowPaths; + + /** + * The number of preempted transactions on this coordinator. + */ + public final Meter preempts; + + /** + * The number of timed out transactions on this coordinator. + */ + public final Meter timeouts; + + /** + * The number of invalidated transactions on this coordinator. + */ + public final Meter invalidations; + + /** + * The time between the start of the transaction and the start of the recovery, if the transaction is recovered. + */ + public final Timer recoveryDelay; + + /** + * The time between the start of the recovery and the execution of the transaction, if the transaction is recovered. + */ + public final Timer recoveryDuration; + + /** + * The ratio of the number of fast path transactions to the total number of transactions. + */ + public final RatioGaugeSet fastPathToTotal; + + private AccordMetrics(String scope) + { + DefaultNameFactory replica = new DefaultNameFactory(ACCORD_REPLICA, scope); + commitLatency = Metrics.timer(replica.createMetricName(COMMIT_LATENCY)); + executeLatency = Metrics.timer(replica.createMetricName(EXECUTE_LATENCY)); + applyLatency = Metrics.timer(replica.createMetricName(APPLY_LATENCY)); + applyDuration = Metrics.timer(replica.createMetricName(APPLY_DURATION)); + partialDependencies = Metrics.histogram(replica.createMetricName(PARTIAL_DEPENDENCIES), true); + progressLogSize = Metrics.meter(replica.createMetricName(PROGRESS_LOG_SIZE)); + + DefaultNameFactory coordinator = new DefaultNameFactory(ACCORD_COORDINATOR, scope); + dependencies = Metrics.histogram(coordinator.createMetricName(DEPENDENCIES), true); + fastPaths = Metrics.meter(coordinator.createMetricName(FAST_PATHS)); + slowPaths = Metrics.meter(coordinator.createMetricName(SLOW_PATHS)); + preempts = Metrics.meter(coordinator.createMetricName(PREEMPTS)); + timeouts = Metrics.meter(coordinator.createMetricName(TIMEOUTS)); + invalidations = Metrics.meter(coordinator.createMetricName(INVALIDATIONS)); + recoveryDelay = Metrics.timer(coordinator.createMetricName(RECOVERY_DELAY)); + recoveryDuration = Metrics.timer(coordinator.createMetricName(RECOVERY_TIME)); + fastPathToTotal = new RatioGaugeSet(fastPaths, RatioGaugeSet.sum(fastPaths, slowPaths), coordinator, FAST_PATH_TO_TOTAL + ".%s"); + } + + @Override + public String toString() + { + StringBuilder builder = new StringBuilder(); + builder.append("AccordMetrics ["); + + try + { + for (Field f : getClass().getDeclaredFields()) + { + f.setAccessible(true); + if (Counting.class.isAssignableFrom(f.getType())) + { + Counting metric = (Counting) f.get(this); + builder.append(String.format("%s: count=%d, ", f.getName(), metric.getCount())); + } + } + } + catch (IllegalAccessException e) + { + throw new RuntimeException(e); + } + builder.append("]"); + return builder.toString(); + } + + public static class Listener implements EventsListener + { + public final static Listener instance = new Listener(AccordMetrics.readMetrics, AccordMetrics.writeMetrics); + + private final AccordMetrics readMetrics; + private final AccordMetrics writeMetrics; + + public Listener(AccordMetrics readMetrics, AccordMetrics writeMetrics) + { + this.readMetrics = readMetrics; + this.writeMetrics = writeMetrics; + } + + private AccordMetrics forTransaction(TxnId txnId) + { + if (txnId.isWrite()) + return writeMetrics; + else if (txnId.isRead()) + return readMetrics; + else + return null; + } + + @Override + public void onCommitted(Command cmd) + { + long now = AccordService.uniqueNow(); + AccordMetrics metrics = forTransaction(cmd.txnId()); + if (metrics != null) + { + long trxTimestamp = cmd.txnId().hlc(); + metrics.commitLatency.update(now - trxTimestamp, TimeUnit.MICROSECONDS); + } + } + + @Override + public void onExecuted(Command cmd) + { + long now = AccordService.uniqueNow(); + AccordMetrics metrics = forTransaction(cmd.txnId()); + if (metrics != null) + { + Timestamp trxTimestamp = cmd.txnId(); + metrics.executeLatency.update(now - trxTimestamp.hlc(), TimeUnit.MICROSECONDS); + PartialDeps deps = cmd.partialDeps(); + metrics.partialDependencies.update(deps != null ? deps.txnIdCount() : 0); + } + } + + @Override + public void onApplied(Command cmd, long applyStartTimestamp) + { + long now = AccordService.uniqueNow(); + AccordMetrics metrics = forTransaction(cmd.txnId()); + if (metrics != null) + { + Timestamp trxTimestamp = cmd.txnId(); + metrics.applyLatency.update(now - trxTimestamp.hlc(), TimeUnit.MICROSECONDS); + metrics.applyDuration.update(now - applyStartTimestamp, TimeUnit.MICROSECONDS); + } + } + + @Override + public void onFastPathTaken(TxnId txnId, Deps deps) + { + AccordMetrics metrics = forTransaction(txnId); + if (metrics != null) + { + metrics.fastPaths.mark(); + metrics.dependencies.update(deps.txnIdCount()); + } + } + + @Override + public void onSlowPathTaken(TxnId txnId, Deps deps) + { + AccordMetrics metrics = forTransaction(txnId); + if (metrics != null) + { + metrics.slowPaths.mark(); + metrics.dependencies.update(deps.txnIdCount()); + } + } + + @Override + public void onRecover(TxnId txnId, Timestamp recoveryTimestamp) + { + AccordMetrics metrics = forTransaction(txnId); + if (metrics != null) + { + long now = AccordService.uniqueNow(); + + metrics.recoveryDuration.update(now - recoveryTimestamp.hlc(), MICROSECONDS); + metrics.recoveryDelay.update(recoveryTimestamp.hlc() - txnId.hlc(), MICROSECONDS); + } + } + + @Override + public void onPreempted(TxnId txnId) + { + AccordMetrics metrics = forTransaction(txnId); + if (metrics != null) + metrics.preempts.mark(); + } + + @Override + public void onTimeout(TxnId txnId) + { + AccordMetrics metrics = forTransaction(txnId); + if (metrics != null) + metrics.timeouts.mark(); + } + + @Override + public void onInvalidated(TxnId txnId) + { + AccordMetrics metrics = forTransaction(txnId); + if (metrics != null) + metrics.invalidations.mark(); + } + + @Override + public void onProgressLogSizeChange(TxnId txnId, int delta) + { + AccordMetrics metrics = forTransaction(txnId); + if (metrics != null) + metrics.progressLogSize.mark(delta); + } + } +} diff --git a/src/java/org/apache/cassandra/metrics/AccordStateCacheMetrics.java b/src/java/org/apache/cassandra/metrics/AccordStateCacheMetrics.java new file mode 100644 index 0000000000..fd4308a356 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/AccordStateCacheMetrics.java @@ -0,0 +1,50 @@ +/* + * 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.metrics; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import com.codahale.metrics.Histogram; + +import static org.apache.cassandra.metrics.CacheMetrics.TYPE_NAME; +import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; + +public class AccordStateCacheMetrics extends CacheAccessMetrics +{ + public static final String OBJECT_SIZE = "ObjectSize"; + + public final Histogram objectSize; + + private final Map, CacheAccessMetrics> instanceMetrics = new ConcurrentHashMap<>(2); + + private final String type; + + public AccordStateCacheMetrics(String type) + { + super(new DefaultNameFactory(TYPE_NAME, type)); + objectSize = Metrics.histogram(factory.createMetricName(OBJECT_SIZE), false); + this.type = type; + } + + public CacheAccessMetrics forInstance(Class klass) + { + return instanceMetrics.computeIfAbsent(klass, k -> new CacheAccessMetrics(new DefaultNameFactory(TYPE_NAME, String.format("%s-%s", type, k.getSimpleName())))); + } +} diff --git a/src/java/org/apache/cassandra/metrics/CacheAccessMetrics.java b/src/java/org/apache/cassandra/metrics/CacheAccessMetrics.java new file mode 100644 index 0000000000..59d76a9f49 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/CacheAccessMetrics.java @@ -0,0 +1,73 @@ +/* + * 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.metrics; + +import com.google.common.annotations.VisibleForTesting; + +import com.codahale.metrics.Meter; + +import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; + +public class CacheAccessMetrics +{ + /** + * Total number of cache hits + */ + public final Meter hits; + + /** + * Total number of cache misses + */ + public final Meter misses; + + /** + * Total number of cache requests + */ + public final Meter requests; + + public final RatioGaugeSet hitRate; + + public final RatioGaugeSet missRate; + + protected final MetricNameFactory factory; + + public CacheAccessMetrics(MetricNameFactory factory) + { + this.factory = factory; + + this.hits = Metrics.meter(factory.createMetricName("Hits")); + this.misses = Metrics.meter(factory.createMetricName("Misses")); + this.requests = Metrics.meter(factory.createMetricName("Requests")); + + this.hitRate = new RatioGaugeSet(hits, requests, factory, "%sHitRate"); + this.missRate = new RatioGaugeSet(misses, requests, factory, "%sMissRate"); + } + + @VisibleForTesting + public void reset() + { + // No actual reset happens. The Meter counter is put to zero but will not reset the moving averages + // It rather injects a weird value into them. + // This method is being only used by CacheMetricsTest and CachingBench so fixing this issue was acknowledged + // but not considered mandatory to be fixed now (CASSANDRA-16228) + hits.mark(-hits.getCount()); + misses.mark(-misses.getCount()); + requests.mark(-requests.getCount()); + } +} diff --git a/src/java/org/apache/cassandra/metrics/CacheMetrics.java b/src/java/org/apache/cassandra/metrics/CacheMetrics.java index 574b0f065c..13a59f88e6 100644 --- a/src/java/org/apache/cassandra/metrics/CacheMetrics.java +++ b/src/java/org/apache/cassandra/metrics/CacheMetrics.java @@ -38,7 +38,7 @@ public class CacheMetrics extends AbstractCacheMetrics public final Gauge entries; /** - * Create metrics for given cache. + * Create metrics for the given cache supporting entity. * * @param type Type of Cache to identify metrics * @param cache Weighted Cache to measure metrics diff --git a/src/java/org/apache/cassandra/metrics/CacheSizeMetrics.java b/src/java/org/apache/cassandra/metrics/CacheSizeMetrics.java new file mode 100644 index 0000000000..fb34cfcc19 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/CacheSizeMetrics.java @@ -0,0 +1,63 @@ +/* + * 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.metrics; + +import com.codahale.metrics.Gauge; +import org.apache.cassandra.cache.CacheSize; + +import static org.apache.cassandra.metrics.CacheMetrics.TYPE_NAME; +import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; + +public class CacheSizeMetrics +{ + public static final String CAPACITY = "Capacity"; + public static final String SIZE = "Size"; + public static final String ENTRIES = "Entries"; + /** + * Cache capacity in bytes + */ + public final Gauge capacity; + + /** + * Total size of cache, in bytes + */ + public final Gauge size; + + /** + * Total number of cache entries + */ + public final Gauge entries; + + /** + * Create metrics for the given cache supporting entity. + * + * @param type Type of Cache to identify metrics. + * @param cache Cache to measure metrics + */ + public CacheSizeMetrics(String type, CacheSize cache) + { + this(new DefaultNameFactory(TYPE_NAME, type), cache); + } + + public CacheSizeMetrics(MetricNameFactory factory, CacheSize cache) + { + capacity = Metrics.register(factory.createMetricName(CAPACITY), cache::capacity); + size = Metrics.register(factory.createMetricName(SIZE), cache::weightedSize); + entries = Metrics.register(factory.createMetricName(ENTRIES), cache::size); + } +} diff --git a/src/java/org/apache/cassandra/metrics/RatioGaugeSet.java b/src/java/org/apache/cassandra/metrics/RatioGaugeSet.java new file mode 100644 index 0000000000..057a69011b --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/RatioGaugeSet.java @@ -0,0 +1,121 @@ +/* + * 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.metrics; + +import java.util.function.DoubleSupplier; +import java.util.function.ToDoubleFunction; + +import com.codahale.metrics.Metered; +import com.codahale.metrics.RatioGauge; +import org.apache.cassandra.metrics.CassandraMetricsRegistry.MetricName; + +import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; + +public class RatioGaugeSet +{ + public static final String ONE_MINUTE = "OneMinute"; + public static final String FIVE_MINUTE = "FiveMinute"; + public static final String FIFTEEN_MINUTE = "FifteenMinute"; + public static final String MEAN_RATIO = ""; + + public final RatioGauge oneMinute; + public final RatioGauge fiveMinute; + public final RatioGauge fifteenMinute; + public final RatioGauge mean; + + public RatioGaugeSet(Metered numerator, Metered denominator, MetricNameFactory factory, String namePattern) + { + this.oneMinute = ratioGauge(factory.createMetricName(String.format(namePattern, ONE_MINUTE)), numerator::getOneMinuteRate, denominator::getOneMinuteRate); + this.fiveMinute = ratioGauge(factory.createMetricName(String.format(namePattern, FIVE_MINUTE)), numerator::getFiveMinuteRate, denominator::getFiveMinuteRate); + this.fifteenMinute = ratioGauge(factory.createMetricName(String.format(namePattern, FIFTEEN_MINUTE)), numerator::getFifteenMinuteRate, denominator::getFifteenMinuteRate); + this.mean = ratioGauge(factory.createMetricName(String.format(namePattern, MEAN_RATIO)), numerator::getCount, denominator::getCount); + } + + private static RatioGauge ratioGauge(DoubleSupplier numerator, DoubleSupplier denominator) + { + return new RatioGauge() + { + protected Ratio getRatio() + { + return Ratio.of(numerator.getAsDouble(), denominator.getAsDouble()); + } + }; + } + + private RatioGauge ratioGauge(MetricName name, DoubleSupplier numerator, DoubleSupplier denominator) + { + return Metrics.register(name, ratioGauge(numerator, denominator)); + } + + public static Metered sum(Metered... meters) + { + return new SummingMeter(meters); + } + + private static class SummingMeter implements Metered + { + private final Metered[] meters; + + public SummingMeter(Metered... meters) + { + this.meters = meters; + } + + @Override + public long getCount() + { + long count = 0; + for (Metered meter : meters) + count += meter.getCount(); + return count; + } + + private double getRate(ToDoubleFunction rateSupplier) + { + double rate = 0; + for (Metered meter : meters) + rate += rateSupplier.applyAsDouble(meter); + return rate; + } + + @Override + public double getMeanRate() + { + return getRate(Metered::getMeanRate); + } + + @Override + public double getFifteenMinuteRate() + { + return getRate(Metered::getFifteenMinuteRate); + } + + @Override + public double getFiveMinuteRate() + { + return getRate(Metered::getFiveMinuteRate); + } + + @Override + public double getOneMinuteRate() + { + return getRate(Metered::getOneMinuteRate); + } + } +} diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 1913b3213f..c9a3224b9a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -30,12 +30,10 @@ import java.util.concurrent.TimeUnit; import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; - import javax.annotation.Nullable; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableSet; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -72,10 +70,12 @@ import accord.utils.ReducingRangeMap; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; import accord.utils.async.Observable; +import org.apache.cassandra.cache.CacheSize; import org.apache.cassandra.concurrent.ExecutorPlus; -import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.concurrent.Stage; +import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.metrics.AccordStateCacheMetrics; import org.apache.cassandra.service.accord.api.PartitionKey; import org.apache.cassandra.service.accord.async.AsyncOperation; import org.apache.cassandra.service.accord.async.ExecutionOrder; @@ -85,7 +85,7 @@ import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory; -public class AccordCommandStore extends CommandStore +public class AccordCommandStore extends CommandStore implements CacheSize { private static final Logger logger = LoggerFactory.getLogger(AccordCommandStore.class); @@ -124,9 +124,10 @@ public class AccordCommandStore extends CommandStore DataStore dataStore, ProgressLog.Factory progressLogFactory, EpochUpdateHolder epochUpdateHolder, - AccordJournal journal) + AccordJournal journal, + AccordStateCacheMetrics cacheMetrics) { - this(id, time, agent, dataStore, progressLogFactory, epochUpdateHolder, journal, Stage.READ.executor(), Stage.MUTATION.executor()); + this(id, time, agent, dataStore, progressLogFactory, epochUpdateHolder, journal, Stage.READ.executor(), Stage.MUTATION.executor(), cacheMetrics); } @VisibleForTesting @@ -138,7 +139,8 @@ public class AccordCommandStore extends CommandStore EpochUpdateHolder epochUpdateHolder, AccordJournal journal, ExecutorPlus loadExecutor, - ExecutorPlus saveExecutor) + ExecutorPlus saveExecutor, + AccordStateCacheMetrics cacheMetrics) { super(id, time, agent, dataStore, progressLogFactory, epochUpdateHolder); this.journal = journal; @@ -146,7 +148,7 @@ public class AccordCommandStore extends CommandStore executor = executorFactory().sequential(CommandStore.class.getSimpleName() + '[' + id + ']'); executionOrder = new ExecutionOrder(); threadId = getThreadId(executor); - stateCache = new AccordStateCache(loadExecutor, saveExecutor, 8 << 20); + stateCache = new AccordStateCache(loadExecutor, saveExecutor, 8 << 20, cacheMetrics); commandCache = stateCache.instance(TxnId.class, TxnId.class, @@ -181,10 +183,10 @@ public class AccordCommandStore extends CommandStore executor.execute(this::loadRangesToCommands); } - static Factory factory(AccordJournal journal) + static Factory factory(AccordJournal journal, AccordStateCacheMetrics cacheMetrics) { return (id, time, agent, dataStore, progressLogFactory, rangesForEpoch) -> - new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, rangesForEpoch, journal); + new AccordCommandStore(id, time, agent, dataStore, progressLogFactory, rangesForEpoch, journal, cacheMetrics); } private void loadRangesToCommands() @@ -250,15 +252,29 @@ public class AccordCommandStore extends CommandStore return Thread.currentThread().getId() == threadId; } - public void setCacheSize(long bytes) + @Override + public void setCapacity(long bytes) { checkInStoreThread(); - stateCache.setMaxSize(bytes); + stateCache.setCapacity(bytes); } - public long getCacheSize() + @Override + public long capacity() { - return stateCache.getMaxSize(); + return stateCache.capacity(); + } + + @Override + public int size() + { + return stateCache.size(); + } + + @Override + public long weightedSize() + { + return stateCache.weightedSize(); } public void checkInStoreThread() diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java index bbae3c3d54..d5f1ed3b35 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java @@ -30,15 +30,24 @@ import accord.local.ShardDistributor; import accord.primitives.Range; import accord.topology.Topology; import accord.utils.RandomSource; +import org.apache.cassandra.cache.CacheSize; +import org.apache.cassandra.metrics.AccordStateCacheMetrics; +import org.apache.cassandra.metrics.CacheSizeMetrics; import org.apache.cassandra.service.accord.api.AccordRoutingKey; -public class AccordCommandStores extends CommandStores +public class AccordCommandStores extends CommandStores implements CacheSize { + public static final String ACCORD_STATE_CACHE = "accord-state-cache"; + + private final CacheSizeMetrics cacheSizeMetrics; + private long cacheSize; + AccordCommandStores(NodeTimeService time, Agent agent, DataStore store, RandomSource random, ShardDistributor shardDistributor, ProgressLog.Factory progressLogFactory, AccordJournal journal) { - super(time, agent, store, random, shardDistributor, progressLogFactory, AccordCommandStore.factory(journal)); - setCacheSize(maxCacheSize()); + super(time, agent, store, random, shardDistributor, progressLogFactory, AccordCommandStore.factory(journal, new AccordStateCacheMetrics(ACCORD_STATE_CACHE))); + setCapacity(maxCacheSize()); + this.cacheSizeMetrics = new CacheSizeMetrics(ACCORD_STATE_CACHE, this); } static Factory factory(AccordJournal journal) @@ -67,21 +76,37 @@ public class AccordCommandStores extends CommandStores return false; } - private long cacheSize; - - synchronized void setCacheSize(long bytes) + public synchronized void setCapacity(long bytes) { cacheSize = bytes; refreshCacheSizes(); } + @Override + public long capacity() + { + return cacheSize; + } + + @Override + public int size() + { + return unsafeFoldLeft(0, (size, commandStore) -> size + ((AccordCommandStore) commandStore).size()); + } + + @Override + public long weightedSize() + { + return unsafeFoldLeft(0L, (size, commandStore) -> size + ((AccordCommandStore) commandStore).weightedSize()); + } + synchronized void refreshCacheSizes() { if (count() == 0) return; long perStore = cacheSize / count(); // TODO (low priority, safety): we might transiently breach our limit if we increase one store before decreasing another - forEach(commandStore -> ((AccordSafeCommandStore) commandStore).commandStore().setCacheSize(perStore)); + forEach(commandStore -> ((AccordSafeCommandStore) commandStore).commandStore().setCapacity(perStore)); } private static long maxCacheSize() diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index ae6901118d..95ef73f0e7 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -299,7 +299,6 @@ public class AccordService implements IAccordService, Shutdownable } if (cause instanceof Preempted) { - metrics.preempts.mark(); //TODO need to improve // Coordinator "could" query the accord state to see whats going on but that doesn't exist yet. // Protocol also doesn't have a way to denote "unknown" outcome, so using a timeout as the closest match @@ -372,7 +371,7 @@ public class AccordService implements IAccordService, Shutdownable { long bytes = kb << 10; AccordCommandStores commandStores = (AccordCommandStores) node.commandStores(); - commandStores.setCacheSize(bytes); + commandStores.setCapacity(bytes); } @Override diff --git a/src/java/org/apache/cassandra/service/accord/AccordStateCache.java b/src/java/org/apache/cassandra/service/accord/AccordStateCache.java index 15faee4c00..97bf5b1586 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordStateCache.java +++ b/src/java/org/apache/cassandra/service/accord/AccordStateCache.java @@ -26,13 +26,15 @@ import java.util.function.ToLongFunction; import java.util.stream.Stream; import com.google.common.annotations.VisibleForTesting; - import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.utils.IntrusiveLinkedList; import accord.utils.async.AsyncChains; +import org.apache.cassandra.cache.CacheSize; import org.apache.cassandra.concurrent.ExecutorPlus; +import org.apache.cassandra.metrics.AccordStateCacheMetrics; +import org.apache.cassandra.metrics.CacheAccessMetrics; import org.apache.cassandra.service.accord.AccordCachingState.Status; import static accord.utils.Invariants.checkState; @@ -49,7 +51,7 @@ import static org.apache.cassandra.service.accord.AccordCachingState.Status.SAVI * Supports dynamic object sizes. After each acquire/free cycle, the cacheable objects size is recomputed to * account for data added/removed during txn processing if it's modified flag is set */ -public class AccordStateCache extends IntrusiveLinkedList> +public class AccordStateCache extends IntrusiveLinkedList> implements CacheSize { private static final Logger logger = LoggerFactory.getLogger(AccordStateCache.class); @@ -63,13 +65,6 @@ public class AccordStateCache extends IntrusiveLinkedList> cache = new HashMap<>(); private final HashMap, Instance> instances = new HashMap<>(); @@ -78,22 +73,27 @@ public class AccordStateCache extends IntrusiveLinkedList void maybeUpdateSize(AccordCachingState node, ToLongFunction estimator) { if (node.shouldUpdateSize()) - bytesCached += ((AccordCachingState) node).estimatedSizeOnHeapDelta((ToLongFunction) estimator); + { + long delta = ((AccordCachingState) node).estimatedSizeOnHeapDelta((ToLongFunction) estimator); + bytesCached += delta; + instanceForNode(node).bytesCached += delta; + } } /* @@ -174,6 +178,8 @@ public class AccordStateCache extends IntrusiveLinkedList instance = instanceForNode(node); + instance.bytesCached -= node.lastQueriedEstimatedSizeOnHeap; if (node.status() == LOADED && VALIDATE_LOAD_ON_EVICT) instanceForNode(node).validateLoadEvicted(node); @@ -181,6 +187,8 @@ public class AccordStateCache extends IntrusiveLinkedList self = cache.remove(node.key()); + if (self != null) + instance.itemsCached--; checkState(self == node, "Leaked node detected; was attempting to remove %s but cache had %s", node, self); } else @@ -212,7 +220,7 @@ public class AccordStateCache extends IntrusiveLinkedList> + public class Instance> implements CacheSize { private final Class keyClass; private final Function, S> safeRefFactory; @@ -220,7 +228,11 @@ public class AccordStateCache extends IntrusiveLinkedList saveFunction; private final BiFunction validateFunction; private final ToLongFunction heapEstimator; - private final Stats stats = new Stats(); + private long bytesCached; + private int itemsCached; + + @VisibleForTesting + final CacheAccessMetrics instanceMetrics; public Instance( Class keyClass, @@ -236,6 +248,7 @@ public class AccordStateCache extends IntrusiveLinkedList> stream() @@ -280,8 +293,11 @@ public class AccordStateCache extends IntrusiveLinkedList node = new AccordCachingState<>(key); node.load(loadExecutor, loadFunction); node.references++; - cache.put(key, node); + + if (cache.put(key, node) == null) + itemsCached++; maybeUpdateSize(node, heapEstimator); + metrics.objectSize.update(node.lastQueriedEstimatedSizeOnHeap); maybeEvictSomeNodes(); return node; } @@ -413,37 +429,22 @@ public class AccordStateCache extends IntrusiveLinkedList { + i.itemsCached = 0; + i.bytesCached = 0; + i.instanceMetrics.reset(); + }); //noinspection StatementWithEmptyBody while (null != poll()); } @@ -504,14 +536,14 @@ public class AccordStateCache extends IntrusiveLinkedList node = cache.get(key); return node != null ? node.references : 0; } - - public long cacheQueries() - { - return stats.queries; - } - - public long cacheHits() - { - return stats.hits; - } - - public long cacheMisses() - { - return stats.misses; - } } diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index 3bfaed2d50..e01587e61b 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -24,6 +24,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.api.Agent; +import accord.api.EventsListener; import accord.api.Result; import accord.local.Command; import accord.local.Node; @@ -33,6 +34,7 @@ import accord.primitives.Timestamp; import accord.primitives.Txn; import accord.primitives.TxnId; import org.apache.cassandra.service.accord.AccordService; +import org.apache.cassandra.metrics.AccordMetrics; import org.apache.cassandra.service.accord.txn.TxnQuery; import org.apache.cassandra.service.accord.txn.TxnRead; import org.apache.cassandra.utils.JVMStabilityInspector; @@ -101,4 +103,10 @@ public class AccordAgent implements Agent { return new Txn.InMemory(kind, keysOrRanges, TxnRead.EMPTY, TxnQuery.ALL, null); } + + @Override + public EventsListener metricsEventsListener() + { + return AccordMetrics.Listener.instance; + } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java new file mode 100644 index 0000000000..4db878da47 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordMetricsTest.java @@ -0,0 +1,273 @@ +/* + * 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.distributed.test.accord; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.function.Function; + +import com.google.common.base.Throwables; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.IMessageFilters; +import org.apache.cassandra.exceptions.ReadTimeoutException; +import org.apache.cassandra.exceptions.WriteTimeoutException; +import org.apache.cassandra.metrics.AccordMetrics; +import org.apache.cassandra.metrics.DefaultNameFactory; +import org.apache.cassandra.metrics.RatioGaugeSet; +import org.apache.cassandra.net.Verb; +import org.apache.cassandra.service.accord.AccordService; +import org.apache.cassandra.service.accord.exceptions.ReadPreemptedException; +import org.apache.cassandra.service.accord.exceptions.WritePreemptedException; +import org.assertj.core.data.Offset; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; + + +public class AccordMetricsTest extends AccordTestBase +{ + private static final Logger logger = LoggerFactory.getLogger(AccordMetricsTest.class); + + @Override + protected Logger logger() + { + return logger; + } + + @BeforeClass + public static void setupClass() throws IOException + { + AccordTestBase.setupClass(); + SHARED_CLUSTER.forEach(node -> node.runOnInstance(() -> AccordService.instance().setCacheSize(0))); + for (int i = 0; i < SHARED_CLUSTER.size(); i++) // initialize metrics + logger.trace(SHARED_CLUSTER.get(i + 1).callOnInstance(() -> AccordMetrics.readMetrics.toString() + AccordMetrics.writeMetrics.toString())); + } + + String writeCql() + { + return "BEGIN TRANSACTION\n" + + " LET val = (SELECT v FROM " + currentTable + " WHERE k=? AND c=?);\n" + + " SELECT val.v;\n" + + " UPDATE " + currentTable + " SET v = v + 1 WHERE k=? AND c=?;\n" + + "COMMIT TRANSACTION"; + } + + String readCql() + { + return "BEGIN TRANSACTION\n" + + " LET val = (SELECT v FROM " + currentTable + " WHERE k=? AND c=?);\n" + + " SELECT val.v;\n" + + "COMMIT TRANSACTION"; + } + + Map> countingMetrics0; + + @Before + public void beforeTest() + { + SHARED_CLUSTER.filters().reset(); + SHARED_CLUSTER.schemaChange("CREATE TABLE " + currentTable + " (k int, c int, v int, PRIMARY KEY (k, c))"); + SHARED_CLUSTER.coordinator(1).execute("INSERT INTO " + currentTable + " (k, c, v) VALUES (0, 0, 0)", ConsistencyLevel.ALL); + } + + @Test + public void testRegularMetrics() throws Exception + { + countingMetrics0 = getMetrics(); + SHARED_CLUSTER.coordinator(1).executeWithResult(writeCql(), ConsistencyLevel.ALL, 0, 0, 0, 0); + assertCoordinatorMetrics(0, "rw", 1, 0, 0, 0, 0); + assertCoordinatorMetrics(1, "rw", 0, 0, 0, 0, 0); + assertReplicaMetrics(0, "rw", 1, 1, 1); + assertReplicaMetrics(1, "rw", 1, 1, 1); + assertZeroMetrics("ro"); + + countingMetrics0 = getMetrics(); + SHARED_CLUSTER.coordinator(1).executeWithResult(readCql(), ConsistencyLevel.ALL, 0, 0); + assertCoordinatorMetrics(0, "ro", 1, 0, 0, 0, 0); + assertCoordinatorMetrics(1, "ro", 0, 0, 0, 0, 0); + assertReplicaMetrics(0, "ro", 1, 1, 0); + assertReplicaMetrics(1, "ro", 1, 1, 0); + assertZeroMetrics("rw"); + } + + @Test + public void testPreemptionMetrics() + { + IMessageFilters.Filter commitFilter1 = SHARED_CLUSTER.filters().outbound().verbs(Verb.ACCORD_COMMIT_REQ.id).from(1).to(1).drop(); + IMessageFilters.Filter commitFilter2 = SHARED_CLUSTER.filters().outbound().verbs(Verb.ACCORD_COMMIT_REQ.id).from(1).to(2).drop(); + commitFilter1.on(); + commitFilter2.on(); + + countingMetrics0 = getMetrics(); + try + { + SHARED_CLUSTER.coordinator(1).executeWithResult(writeCql(), ConsistencyLevel.ALL, 0, 0, 0, 0); + fail("expected to fail"); + } + catch (RuntimeException ex) + { + assertThat(Throwables.getCausalChain(ex).stream().map(t -> t.getClass().getName())).contains(WritePreemptedException.class.getName()); + } + + assertCoordinatorMetrics(0, "rw", 1, 0, 1, 0, 0); + assertCoordinatorMetrics(1, "rw", 0, 0, 0, 0, 1); + assertReplicaMetrics(0, "rw", 1, 1, 1); + assertReplicaMetrics(1, "rw", 1, 1, 1); + + assertZeroMetrics("ro"); + + countingMetrics0 = getMetrics(); + try + { + SHARED_CLUSTER.coordinator(1).executeWithResult(readCql(), ConsistencyLevel.ALL, 0, 0); + fail("expected to fail"); + } + catch (RuntimeException ex) + { + assertThat(Throwables.getCausalChain(ex).stream().map(t -> t.getClass().getName())).contains(ReadPreemptedException.class.getName()); + } + + assertCoordinatorMetrics(0, "ro", 1, 0, 1, 0, 0); + assertCoordinatorMetrics(1, "ro", 0, 0, 0, 0, 1); + assertReplicaMetrics(0, "ro", 1, 1, 0); + assertReplicaMetrics(1, "ro", 1, 1, 0); + + assertZeroMetrics("rw"); + } + + @Test + public void testTimeoutMetrics() + { + IMessageFilters.Filter preAcceptFilter = SHARED_CLUSTER.filters().outbound().verbs(Verb.ACCORD_PRE_ACCEPT_REQ.id).from(1).to(2).drop(); + preAcceptFilter.on(); + + countingMetrics0 = getMetrics(); + try + { + SHARED_CLUSTER.coordinator(1).executeWithResult(readCql(), ConsistencyLevel.ALL, 0, 0); + fail("expected to fail"); + } + catch (RuntimeException ex) + { + assertThat(Throwables.getCausalChain(ex).stream().map(t -> t.getClass().getName())).contains(ReadTimeoutException.class.getName()); + } + + assertCoordinatorMetrics(0, "ro", 0, 0, 0, 1, 0); + assertCoordinatorMetrics(1, "ro", 0, 0, 0, 0, 0); + assertReplicaMetrics(0, "ro", 0, 0, 0); + assertReplicaMetrics(1, "ro", 0, 0, 0); + + assertZeroMetrics("rw"); + + countingMetrics0 = getMetrics(); + try + { + SHARED_CLUSTER.coordinator(1).executeWithResult(writeCql(), ConsistencyLevel.ALL, 0, 0, 0, 0); + fail("expected to fail"); + } + catch (RuntimeException ex) + { + assertThat(Throwables.getCausalChain(ex).stream().map(t -> t.getClass().getName())).contains(WriteTimeoutException.class.getName()); + } + + assertCoordinatorMetrics(0, "rw", 0, 0, 0, 1, 0); + assertCoordinatorMetrics(1, "rw", 0, 0, 0, 0, 0); + assertReplicaMetrics(0, "rw", 0, 0, 0); + assertReplicaMetrics(1, "rw", 0, 0, 0); + + assertZeroMetrics("ro"); + } + + private void assertZeroMetrics(String scope) + { + for (int i = 0; i < SHARED_CLUSTER.size(); i++) + { + assertCoordinatorMetrics(i, scope, 0, 0, 0, 0, 0); + assertReplicaMetrics(i, scope, 0, 0, 0); + } + } + + private void assertCoordinatorMetrics(int node, String scope, long fastPaths, long slowPaths, long preempts, long timeouts, long recoveries) + { + DefaultNameFactory nameFactory = new DefaultNameFactory(AccordMetrics.ACCORD_COORDINATOR, scope); + Map metrics = diff(countingMetrics0).get(node); + logger.info("Metrics for node {} / {}: {}", node, scope, metrics); + Function metric = n -> metrics.get(nameFactory.createMetricName(n).getMetricName()); + assertThat(metric.apply(AccordMetrics.FAST_PATHS)).isEqualTo(fastPaths); + assertThat(metric.apply(AccordMetrics.SLOW_PATHS)).isEqualTo(slowPaths); + assertThat(metric.apply(AccordMetrics.PREEMPTS)).isEqualTo(preempts); + assertThat(metric.apply(AccordMetrics.TIMEOUTS)).isEqualTo(timeouts); + assertThat(metric.apply(AccordMetrics.RECOVERY_DELAY)).isEqualTo(recoveries); + assertThat(metric.apply(AccordMetrics.RECOVERY_TIME)).isEqualTo(recoveries); + assertThat(metric.apply(AccordMetrics.DEPENDENCIES)).isEqualTo(fastPaths + slowPaths); + + if ((fastPaths + slowPaths) > 0) + { + String fastPathToTotalName = nameFactory.createMetricName(AccordMetrics.FAST_PATH_TO_TOTAL + "." + RatioGaugeSet.MEAN_RATIO).getMetricName(); + assertThat((double) SHARED_CLUSTER.get(1).metrics().getGauge(fastPathToTotalName)).isEqualTo((double) fastPaths / (double) (fastPaths + slowPaths), Offset.offset(0.01d)); + } + } + + private void assertReplicaMetrics(int node, String scope, long commits, long executions, long applications) + { + DefaultNameFactory nameFactory = new DefaultNameFactory(AccordMetrics.ACCORD_REPLICA, scope); + Map metrics = diff(countingMetrics0).get(node); + Function metric = n -> metrics.get(nameFactory.createMetricName(n).getMetricName()); + assertThat(metric.apply(AccordMetrics.COMMIT_LATENCY)).isEqualTo(commits); + assertThat(metric.apply(AccordMetrics.EXECUTE_LATENCY)).isEqualTo(executions); + assertThat(metric.apply(AccordMetrics.APPLY_LATENCY)).isEqualTo(applications); + assertThat(metric.apply(AccordMetrics.APPLY_DURATION)).isEqualTo(applications); + assertThat(metric.apply(AccordMetrics.PARTIAL_DEPENDENCIES)).isEqualTo(executions); + } + + private Map> getMetrics() + { + Map> metrics = new HashMap<>(); + for (int i = 0; i < SHARED_CLUSTER.size(); i++) + metrics.put(i, SHARED_CLUSTER.get(i + 1).metrics().getCounters(name -> name.startsWith("org.apache.cassandra.metrics.accord-"))); + return metrics; + } + + private Map> diff(Map> prev) + { + Map> curr = getMetrics(); + Map> diff = new HashMap<>(); + for (int i = 0; i < SHARED_CLUSTER.size(); i++) + { + Map prevNode = prev.get(i); + Map currNode = curr.get(i); + Map diffNode = new HashMap<>(); + for (Map.Entry currEntry : currNode.entrySet()) + { + Long prevVal = prevNode.get(currEntry.getKey()); + if (prevVal != null) + diffNode.put(currEntry.getKey(), currEntry.getValue() - prevVal); + } + diff.put(i, diffNode); + } + return diff; + } +} diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java index c09989eb0a..33472cf09f 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java @@ -399,9 +399,9 @@ public class CompactionAccordIteratorsTest { commandStore.executeBlocking(() -> { // clear cache and wait for post-eviction writes to complete - long cacheSize = commandStore.getCacheSize(); - commandStore.setCacheSize(0); - commandStore.setCacheSize(cacheSize); + long cacheSize = commandStore.capacity(); + commandStore.setCapacity(0); + commandStore.setCapacity(cacheSize); commandStore.cache().awaitSaveResults(); }); commands.forceBlockingFlush(FlushReason.UNIT_TESTS); diff --git a/test/unit/org/apache/cassandra/io/sstable/LargePartitionsTest.java b/test/unit/org/apache/cassandra/io/sstable/LargePartitionsTest.java index 17fb0f28ec..a4c1b8608d 100644 --- a/test/unit/org/apache/cassandra/io/sstable/LargePartitionsTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/LargePartitionsTest.java @@ -132,10 +132,10 @@ public class LargePartitionsTest extends CQLTester { CacheMetrics metrics = CacheService.instance.keyCache.getMetrics(); System.out.println("Key cache metrics " + title + ": capacity:" + metrics.capacity.getValue() + - " size:"+metrics.size.getValue()+ + " size:" + metrics.size.getValue() + " entries:" + metrics.entries.getValue() + - " hit-rate:"+metrics.hitRate.getValue() + - " one-min-rate:"+metrics.oneMinuteHitRate.getValue()); + " hit-rate:" + metrics.hitRate.getValue() + + " one-min-rate:" + metrics.hitRate.getValue()); } @Test diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java index 9dcc72b84d..a13f17d49e 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandTest.java @@ -89,7 +89,7 @@ public class AccordCommandTest public void basicCycleTest() throws Throwable { AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl"); - getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.setCacheSize(0))); + getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.setCapacity(0))); TxnId txnId = txnId(1, clock.incrementAndGet(), 1); Txn txn = createWriteTxn(1); @@ -173,7 +173,7 @@ public class AccordCommandTest public void computeDeps() throws Throwable { AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl"); - getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.setCacheSize(0))); + getUninterruptibly(commandStore.execute(PreLoadContext.empty(), unused -> commandStore.setCapacity(0))); TxnId txnId1 = txnId(1, clock.incrementAndGet(), 1); Txn txn = createWriteTxn(2); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordStateCacheTest.java b/test/unit/org/apache/cassandra/service/accord/AccordStateCacheTest.java index 0ac350b92f..5fc6f2305f 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordStateCacheTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordStateCacheTest.java @@ -17,55 +17,66 @@ */ package org.apache.cassandra.service.accord; +import java.util.UUID; + import org.junit.Assert; +import org.junit.Before; import org.junit.Test; import accord.utils.async.AsyncChain; +import org.apache.cassandra.cache.CacheSize; import org.apache.cassandra.concurrent.ManualExecutor; +import org.apache.cassandra.metrics.AccordStateCacheMetrics; +import org.apache.cassandra.metrics.CacheAccessMetrics; import org.apache.cassandra.service.accord.AccordCachingState.Status; import static org.apache.cassandra.service.accord.AccordTestUtils.testLoad; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; public class AccordStateCacheTest { private static final long DEFAULT_NODE_SIZE = nodeSize(0); + private AccordStateCacheMetrics cacheMetrics; - private static class SafeString implements AccordSafeState + private static abstract class TestSafeState implements AccordSafeState { - private boolean invalidated = false; - private final AccordCachingState global; - private String original = null; + protected boolean invalidated = false; + protected final AccordCachingState global; + private T original = null; - public SafeString(AccordCachingState global) + public TestSafeState(AccordCachingState global) { this.global = global; } - public AccordCachingState global() + public AccordCachingState global() { return global; } @Override - public String key() + public T key() { return global.key(); } @Override - public String current() + public T current() { return global.get(); } @Override - public void set(String update) + public void set(T update) { global.set(update); } @Override - public String original() + public T original() { return original; } @@ -118,6 +129,22 @@ public class AccordStateCacheTest } } + private static class SafeString extends TestSafeState + { + public SafeString(AccordCachingState global) + { + super(global); + } + } + + private static class SafeInt extends TestSafeState + { + public SafeInt(AccordCachingState global) + { + super(global); + } + } + private static long emptyNodeSize() { return AccordCachingState.EMPTY_SIZE; @@ -131,15 +158,35 @@ public class AccordStateCacheTest private static void assertCacheState(AccordStateCache cache, int referenced, int total, long bytes) { Assert.assertEquals(referenced, cache.numReferencedEntries()); - Assert.assertEquals(total, cache.totalNumEntries()); - Assert.assertEquals(bytes, cache.bytesCached()); + Assert.assertEquals(total, cache.size()); + Assert.assertEquals(bytes, cache.weightedSize()); + } + + private void assertCacheMetrics(CacheAccessMetrics metrics, int hits, int misses, int requests) + { + Assert.assertEquals(hits, metrics.hits.getCount()); + Assert.assertEquals(misses, metrics.misses.getCount()); + Assert.assertEquals(requests, metrics.requests.getCount()); + if (metrics instanceof AccordStateCacheMetrics) + { + AccordStateCacheMetrics ascMetrics = (AccordStateCacheMetrics) metrics; + Assert.assertEquals(misses, ascMetrics.objectSize.getCount()); + assertThat(ascMetrics.objectSize.getSnapshot().getMax()).isGreaterThanOrEqualTo(DEFAULT_NODE_SIZE); + } + } + + @Before + public void before() + { + String type = String.format("%s-%s", AccordCommandStores.ACCORD_STATE_CACHE, UUID.randomUUID()); + cacheMetrics = new AccordStateCacheMetrics(type); } @Test public void testAcquisitionAndRelease() { ManualExecutor executor = new ManualExecutor(); - AccordStateCache cache = new AccordStateCache(executor, executor,500); + AccordStateCache cache = new AccordStateCache(executor, executor, 500, cacheMetrics); AccordStateCache.Instance instance = cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length); assertCacheState(cache, 0, 0, 0); @@ -162,13 +209,56 @@ public class AccordStateCacheTest Assert.assertSame(safeString1.global, cache.head()); Assert.assertSame(safeString2.global, cache.tail()); + + assertCacheMetrics(cache.metrics, 0, 2, 2); + assertCacheMetrics(instance.instanceMetrics, 0, 2, 2); + } + + @Test + public void testCachingMetricsWithTwoInstances() + { + ManualExecutor executor = new ManualExecutor(); + AccordStateCache cache = new AccordStateCache(executor, executor, 500, cacheMetrics); + AccordStateCache.Instance stringInstance = + cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true,String::length); + AccordStateCache.Instance intInstance = + cache.instance(Integer.class, Integer.class, SafeInt::new, key -> key, (original, current) -> null, (k, v) -> true,ignored -> Integer.BYTES); + assertCacheState(cache, 0, 0, 0); + + SafeString safeString1 = stringInstance.acquire("1"); + testLoad(executor, safeString1, "1"); + stringInstance.release(safeString1); + SafeString safeString2 = stringInstance.acquire("2"); + testLoad(executor, safeString2, "2"); + stringInstance.release(safeString2); + + SafeInt safeInt1 = intInstance.acquire(3); + testLoad(executor, safeInt1, 3); + intInstance.release(safeInt1); + SafeInt safeInt2 = intInstance.acquire(4); + testLoad(executor, safeInt2, 4); + intInstance.release(safeInt2); + SafeInt safeInt3 = intInstance.acquire(5); + testLoad(executor, safeInt3, 5); + intInstance.release(safeInt3); + + assertCacheState(cache, 0, 5, nodeSize(Integer.BYTES) * 3 + nodeSize(1) * 2); + assertThat(stringInstance.size()).isEqualTo(2); + assertThat(stringInstance.weightedSize()).isEqualTo(nodeSize(1) * 2); + assertThat(stringInstance.capacity()).isEqualTo(cache.capacity()); + assertThat(intInstance.size()).isEqualTo(3); + assertThat(intInstance.weightedSize()).isEqualTo(nodeSize(Integer.BYTES) * 3); + assertThat(intInstance.capacity()).isEqualTo(cache.capacity()); + + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(() -> stringInstance.setCapacity(123)); + assertThatExceptionOfType(UnsupportedOperationException.class).isThrownBy(() -> intInstance.setCapacity(123)); } @Test public void testRotation() { ManualExecutor executor = new ManualExecutor(); - AccordStateCache cache = new AccordStateCache(executor, executor, DEFAULT_NODE_SIZE * 5); + AccordStateCache cache = new AccordStateCache(executor, executor, DEFAULT_NODE_SIZE * 5, cacheMetrics); AccordStateCache.Instance instance = cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length); assertCacheState(cache, 0, 0, 0); @@ -187,11 +277,15 @@ public class AccordStateCacheTest Assert.assertSame(items[0].global, cache.head()); Assert.assertSame(items[2].global, cache.tail()); assertCacheState(cache, 0, 3, nodeSize(1) * 3); + assertCacheMetrics(cache.metrics, 0, 3, 3); + assertCacheMetrics(instance.instanceMetrics, 0, 3, 3); SafeString safeString = instance.acquire("1"); Assert.assertEquals(Status.LOADED, safeString.globalStatus()); assertCacheState(cache, 1, 3, nodeSize(1) * 3); + assertCacheMetrics(cache.metrics, 1, 3, 4); + assertCacheMetrics(instance.instanceMetrics, 1, 3, 4); // releasing item should return it to the tail instance.release(safeString); @@ -204,7 +298,7 @@ public class AccordStateCacheTest public void testEvictionOnAcquire() { ManualExecutor executor = new ManualExecutor(); - AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 5); + AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 5, cacheMetrics); AccordStateCache.Instance instance = cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length); assertCacheState(cache, 0, 0, 0); @@ -222,6 +316,8 @@ public class AccordStateCacheTest assertCacheState(cache, 0, 5, nodeSize(1) * 5); Assert.assertSame(items[0].global, cache.head()); Assert.assertSame(items[4].global, cache.tail()); + assertCacheMetrics(cache.metrics, 0, 5, 5); + assertCacheMetrics(instance.instanceMetrics, 0, 5, 5); SafeString safeString = instance.acquire("5"); Assert.assertTrue(instance.isReferenced(safeString.key())); @@ -232,19 +328,23 @@ public class AccordStateCacheTest Assert.assertSame(items[4].global, cache.tail()); Assert.assertFalse(cache.keyIsCached("0")); Assert.assertFalse(cache.keyIsReferenced("0")); + assertCacheMetrics(cache.metrics, 0, 6, 6); + assertCacheMetrics(instance.instanceMetrics, 0, 6, 6); testLoad(executor, safeString, "5"); instance.release(safeString); assertCacheState(cache, 0, 5, nodeSize(1) * 5); Assert.assertSame(items[1].global, cache.head()); Assert.assertSame(safeString.global, cache.tail()); + assertCacheMetrics(cache.metrics, 0, 6, 6); + assertCacheMetrics(instance.instanceMetrics, 0, 6, 6); } @Test public void testEvictionOnRelease() { ManualExecutor executor = new ManualExecutor(); - AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 4); + AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 4, cacheMetrics); AccordStateCache.Instance instance = cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length); assertCacheState(cache, 0, 0, 0); @@ -259,16 +359,22 @@ public class AccordStateCacheTest } assertCacheState(cache, 5, 5, nodeSize(0) * 5); + assertCacheMetrics(cache.metrics, 0, 5, 5); + assertCacheMetrics(instance.instanceMetrics, 0, 5, 5); Assert.assertNull(cache.head()); Assert.assertNull(cache.tail()); instance.release(items[2]); assertCacheState(cache, 4, 4, nodeSize(0) * 4); + assertCacheMetrics(cache.metrics, 0, 5, 5); + assertCacheMetrics(instance.instanceMetrics, 0, 5, 5); Assert.assertNull(cache.head()); Assert.assertNull(cache.tail()); instance.release(items[4]); assertCacheState(cache, 3, 4, nodeSize(0) * 3 + nodeSize(1)); + assertCacheMetrics(cache.metrics, 0, 5, 5); + assertCacheMetrics(instance.instanceMetrics, 0, 5, 5); Assert.assertSame(items[4].global, cache.head()); Assert.assertSame(items[4].global, cache.tail()); } @@ -277,7 +383,7 @@ public class AccordStateCacheTest public void testMultiAcquireRelease() { ManualExecutor executor = new ManualExecutor(); - AccordStateCache cache = new AccordStateCache(executor, executor, DEFAULT_NODE_SIZE * 4); + AccordStateCache cache = new AccordStateCache(executor, executor, DEFAULT_NODE_SIZE * 4, cacheMetrics); AccordStateCache.Instance instance = cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length); assertCacheState(cache, 0, 0, 0); @@ -285,6 +391,8 @@ public class AccordStateCacheTest SafeString safeString1 = instance.acquire("0"); testLoad(executor, safeString1, "0"); Assert.assertEquals(Status.LOADED, safeString1.globalStatus()); + assertCacheMetrics(cache.metrics, 0, 1, 1); + assertCacheMetrics(instance.instanceMetrics, 0, 1, 1); Assert.assertEquals(1, cache.references("0")); assertCacheState(cache, 1, 1, nodeSize(0)); @@ -293,6 +401,8 @@ public class AccordStateCacheTest Assert.assertEquals(Status.LOADED, safeString1.globalStatus()); Assert.assertEquals(2, cache.references("0")); assertCacheState(cache, 1, 1, nodeSize(0)); + assertCacheMetrics(cache.metrics, 1, 1, 2); + assertCacheMetrics(instance.instanceMetrics, 1, 1, 2); instance.release(safeString1); assertCacheState(cache, 1, 1, nodeSize(1)); @@ -304,7 +414,7 @@ public class AccordStateCacheTest public void evictionBlockedOnSaving() { ManualExecutor executor = new ManualExecutor(); - AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 3 + nodeSize(3)); + AccordStateCache cache = new AccordStateCache(executor, executor, nodeSize(1) * 3 + nodeSize(3), cacheMetrics); AccordStateCache.Instance instance = cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length); assertCacheState(cache, 0, 0, 0); @@ -324,9 +434,11 @@ public class AccordStateCacheTest } assertCacheState(cache, 0, 4, nodeSize(1) * 3 + nodeSize(3)); + assertCacheMetrics(cache.metrics, 0, 4, 4); + assertCacheMetrics(instance.instanceMetrics, 0, 4, 4); // force cache eviction - cache.setMaxSize(0); + cache.setCapacity(0); // all should have been evicted except 0 assertCacheState(cache, 0, 1, nodeSize(2)); @@ -341,7 +453,7 @@ public class AccordStateCacheTest public void testUpdates() { ManualExecutor executor = new ManualExecutor(); - AccordStateCache cache = new AccordStateCache(executor, executor, 500); + AccordStateCache cache = new AccordStateCache(executor, executor, 500, cacheMetrics); AccordStateCache.Instance instance = cache.instance(String.class, String.class, SafeString::new, key -> key, (original, current) -> null, (k, v) -> true, String::length); assertCacheState(cache, 0, 0, 0); @@ -360,5 +472,32 @@ public class AccordStateCacheTest assertCacheState(cache, 0, 1, nodeSize(3)); Assert.assertSame(safeString.global, cache.head()); Assert.assertSame(safeString.global, cache.tail()); + + assertCacheMetrics(cache.metrics, 0, 1, 1); + assertCacheMetrics(instance.instanceMetrics, 0, 1, 1); + } + + private CacheSize mockCacheSize(long capacity, long size, int entries) + { + CacheSize cacheSize = mock(CacheSize.class); + when(cacheSize.capacity()).thenReturn(capacity); + when(cacheSize.weightedSize()).thenReturn(size); + when(cacheSize.size()).thenReturn(entries); + return cacheSize; + } + + @Test + public void testAccorStateCacheMetrics() + { + CacheAccessMetrics stringInstance1 = cacheMetrics.forInstance(String.class); + CacheAccessMetrics stringInstance1Dup = cacheMetrics.forInstance(String.class); + CacheAccessMetrics stringInstance2 = cacheMetrics.forInstance(String.class); + CacheAccessMetrics integerInstance1 = cacheMetrics.forInstance(Integer.class); + CacheAccessMetrics integerInstance2 = cacheMetrics.forInstance(Integer.class); + + assertThat(stringInstance1).isSameAs(stringInstance1Dup); + assertThat(stringInstance1).isSameAs(stringInstance2); + assertThat(integerInstance1).isSameAs(integerInstance2); + assertThat(stringInstance1).isNotSameAs(integerInstance1); } } diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index 192aeacfc1..4e21b5096c 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java @@ -77,6 +77,7 @@ import org.apache.cassandra.cql3.QueryProcessor; import org.apache.cassandra.cql3.statements.TransactionStatement; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.metrics.AccordStateCacheMetrics; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.ClientState; @@ -370,7 +371,8 @@ public class AccordTestUtils holder, journal, loadExecutor, - saveExecutor); + saveExecutor, + new AccordStateCacheMetrics(AccordCommandStores.ACCORD_STATE_CACHE + System.currentTimeMillis())); holder.set(result); result.updateRangesForEpoch(); return result; @@ -389,7 +391,7 @@ public class AccordTestUtils Node.Id node = new Id(1); Topology topology = new Topology(1, new Shard(range, Lists.newArrayList(node), Sets.newHashSet(node), Collections.emptySet())); AccordCommandStore store = createAccordCommandStore(node, now, topology, loadExecutor, saveExecutor); - store.execute(PreLoadContext.empty(), safeStore -> ((AccordCommandStore)safeStore.commandStore()).setCacheSize(1 << 20)); + store.execute(PreLoadContext.empty(), safeStore -> ((AccordCommandStore)safeStore.commandStore()).setCapacity(1 << 20)); return store; } diff --git a/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java b/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java index 66a0d00784..e1f5e317f0 100644 --- a/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java +++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncLoaderTest.java @@ -84,7 +84,7 @@ public class AsyncLoaderTest AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl", executor, executor); AccordStateCache.Instance commandCache = commandStore.commandCache(); - commandStore.executeBlocking(() -> commandStore.setCacheSize(1024)); + commandStore.executeBlocking(() -> commandStore.setCapacity(1024)); AccordStateCache.Instance cfkCache = commandStore.commandsForKeyCache(); TxnId txnId = txnId(1, clock.incrementAndGet(), 1); @@ -229,7 +229,7 @@ public class AsyncLoaderTest ManualExecutor executor = new ManualExecutor(); AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl", executor, executor); - commandStore.executor().submit(() -> commandStore.setCacheSize(1024)).get(); + commandStore.executor().submit(() -> commandStore.setCapacity(1024)).get(); AccordStateCache.Instance commandCache = commandStore.commandCache(); AccordStateCache.Instance cfkCache = commandStore.commandsForKeyCache(); TxnId txnId = txnId(1, clock.incrementAndGet(), 1); diff --git a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java index 14d4f5d409..086c9537fe 100644 --- a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java +++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java @@ -223,9 +223,9 @@ public class AsyncOperationTest // clear cache commandStore.executeBlocking(() -> { - long cacheSize = commandStore.getCacheSize(); - commandStore.setCacheSize(0); - commandStore.setCacheSize(cacheSize); + long cacheSize = commandStore.capacity(); + commandStore.setCapacity(0); + commandStore.setCapacity(cacheSize); commandStore.cache().awaitSaveResults(); }); @@ -312,7 +312,7 @@ public class AsyncOperationTest // all txn use the same key; 0 Keys keys = keys(Schema.instance.getTableMetadata("ks", "tbl"), 0); AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl"); - commandStore.executeBlocking(() -> commandStore.setCacheSize(0)); + commandStore.executeBlocking(() -> commandStore.setCapacity(0)); Gen txnIdGen = rs -> txnId(1, clock.incrementAndGet(), 1); qt().withPure(false).withExamples(50).forAll(Gens.random(), Gens.lists(txnIdGen).ofSizeBetween(1, 10)).check((rs, ids) -> {