sanpwc commented on code in PR #6206: URL: https://github.com/apache/ignite-3/pull/6206#discussion_r2200896632
########## modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/ItTransactionMetricsTest.java: ########## @@ -0,0 +1,256 @@ +/* + * 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.ignite.internal.tx; + +import static org.apache.ignite.internal.AssignmentsTestUtils.awaitAssignmentsStabilizationOnDefaultZone; +import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import org.apache.ignite.Ignite; +import org.apache.ignite.internal.ClusterPerClassIntegrationTest; +import org.apache.ignite.internal.metrics.LongMetric; +import org.apache.ignite.internal.metrics.MetricSet; +import org.apache.ignite.internal.tx.metrics.TransactionMetricsSource; +import org.apache.ignite.table.KeyValueView; +import org.apache.ignite.tx.Transaction; +import org.apache.ignite.tx.TransactionOptions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +/** + * Tests transaction metrics. + */ +public class ItTransactionMetricsTest extends ClusterPerClassIntegrationTest { + public static final String TABLE_NAME = "test_table_name"; + + @Override + protected int initialNodes() { + return 2; + } + + @BeforeAll + void createTable() throws Exception { + sql("CREATE TABLE " + TABLE_NAME + " (id INT PRIMARY KEY, val VARCHAR)"); + + awaitAssignmentsStabilizationOnDefaultZone(CLUSTER.aliveNode()); + } + + /** + * Returns a key value view for the table {@link #TABLE_NAME}. + * + * @param nodeIndex Node index to create a key value view. + * @return Key value view. + */ + private KeyValueView<Integer, String> keyValueView(int nodeIndex) { + return keyValueView(CLUSTER.node(nodeIndex)); + } + + /** + * Returns a key value view for the table {@link #TABLE_NAME}. + * + * @param node Node to create a key value view. + * @return Key value view. + */ + private KeyValueView<Integer, String> keyValueView(Ignite node) { + return node.tables().table(TABLE_NAME).keyValueView(Integer.class, String.class); + } + + /** + * Returns a snapshot of transaction metrics for the given node. + * + * @param nodeIndex Node index to capture transaction metrics. + * @return Snapshot of transaction metrics. + */ + private Map<String, Long> metricValues(int nodeIndex) { + var values = new HashMap<String, Long>(); + + MetricSet txMetrics = unwrapIgniteImpl(node(nodeIndex)) + .metricManager() + .metricSnapshot() + .metrics() + .get(TransactionMetricsSource.SOURCE_NAME); + + txMetrics.iterator().forEachRemaining(metric -> { + if (metric instanceof LongMetric) { + values.put(metric.name(), ((LongMetric) metric).value()); + } + }); + + return values; + } + + private static void testMetricValues(Map<String, Long> initial, Map<String, Long> actual, String... ignored) { + assertThat("Number of metrics should be the same.", initial.size(), is(actual.size())); + + var exclude = Set.of(ignored); + + for (Map.Entry<String, Long> e : initial.entrySet()) { + if (!exclude.contains(e.getKey())) { + assertThat("Metric name = " + e.getKey(), actual.get(e.getKey()), is(e.getValue())); + } + } + } + + /** + * Tests that TotalCommits and RoCommits are incremented when a read only transaction successfully committed. + * + * @param implicit {@code true} if a transaction should be implicit and {@code false} otherwise. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testCommitReadOnlyTransaction(boolean implicit) { + Map<String, Long> metrics0 = metricValues(0); + Map<String, Long> metrics1 = metricValues(1); + + Transaction tx = implicit ? null : node(0).transactions().begin(new TransactionOptions().readOnly(true)); + keyValueView(0).get(tx, 12); + if (!implicit) { + tx.commit(); + } + + Map<String, Long> actualMetrics0 = metricValues(0); + Map<String, Long> actualMetrics1 = metricValues(1); + + // Check that there are no updates on the node 1. + testMetricValues(metrics1, actualMetrics1); + + // Check that all transaction metrics ere not changed except TotalCommits and RoCommits. + testMetricValues(metrics0, actualMetrics0, "TotalCommits", "RoCommits"); + + assertThat(actualMetrics0.get("TotalCommits"), is(metrics0.get("TotalCommits") + 1)); + assertThat(actualMetrics0.get("RoCommits"), is(metrics0.get("RoCommits") + 1)); + } + + /** + * Tests that TotalCommits and RwCommits are incremented when a read write transaction successfully committed. + * + * @param implicit {@code true} if a transaction should be implicit and {@code false} otherwise. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testCommitReadWriteTransaction(boolean implicit) { + Map<String, Long> metrics0 = metricValues(0); + Map<String, Long> metrics1 = metricValues(1); + + Transaction tx = implicit ? null : node(0).transactions().begin(); + keyValueView(0).put(tx, 12, "value"); + if (!implicit) { + tx.commit(); + } + + Map<String, Long> actualMetrics0 = metricValues(0); + Map<String, Long> actualMetrics1 = metricValues(1); + + // Check that there are no updates on the node 1. + testMetricValues(metrics1, actualMetrics1); + + // Check that all transaction metrics ere not changed except TotalCommits and RwCommits. + testMetricValues(metrics0, actualMetrics0, "TotalCommits", "RwCommits"); + + assertThat(actualMetrics0.get("TotalCommits"), is(metrics0.get("TotalCommits") + 1)); + assertThat(actualMetrics0.get("RwCommits"), is(metrics0.get("RwCommits") + 1)); + } + + /** + * Tests that TotalCommits and RwCommits/RoCommits are incremented when an "empty" transaction successfully committed. + * Empty means that there are no entries enlisted into the transaction. + */ + @Test + void testCommitEmptyTransaction() { + Map<String, Long> metrics0 = metricValues(0); + Map<String, Long> metrics1 = metricValues(1); + + Transaction rwTx = node(0).transactions().begin(); + rwTx.commit(); + + Transaction roTx = node(0).transactions().begin(new TransactionOptions().readOnly(true)); + roTx.commit(); + + Map<String, Long> actualMetrics0 = metricValues(0); + Map<String, Long> actualMetrics1 = metricValues(1); + + // Check that there are no updates on the node 1. + testMetricValues(metrics1, actualMetrics1); + + // Check that all transaction metrics ere not changed except TotalCommits, RwCommits and RoCommits. + testMetricValues(metrics0, actualMetrics0, "TotalCommits", "RwCommits", "RoCommits"); + + assertThat(actualMetrics0.get("TotalCommits"), is(metrics0.get("TotalCommits") + 2)); + assertThat(actualMetrics0.get("RwCommits"), is(metrics0.get("RwCommits") + 1)); + assertThat(actualMetrics0.get("RoCommits"), is(metrics0.get("RoCommits") + 1)); + } + + /** + * Tests that TotalRollbacks and RwRollbacks/RoRollbacks are incremented when a transaction rolled back. + */ + @Test + void testRollbackTransaction() { + Map<String, Long> metrics0 = metricValues(0); + Map<String, Long> metrics1 = metricValues(1); + + Transaction rwTx = node(0).transactions().begin(); + keyValueView(0).put(rwTx, 12, "value"); + rwTx.rollback(); + + Transaction roTx = node(0).transactions().begin(new TransactionOptions().readOnly(true)); + keyValueView(0).get(roTx, 12); + roTx.rollback(); + + Map<String, Long> actualMetrics0 = metricValues(0); + Map<String, Long> actualMetrics1 = metricValues(1); + + // Check that there are no updates on the node 1. + testMetricValues(metrics1, actualMetrics1); + + // Check that all transaction metrics ere not changed except TotalRollbacks, RwRollbacks and RoRollbacks. + testMetricValues(metrics0, actualMetrics0, "TotalRollbacks", "RwRollbacks", "RoRollbacks"); + + assertThat(actualMetrics0.get("TotalRollbacks"), is(metrics0.get("TotalRollbacks") + 2)); + assertThat(actualMetrics0.get("RwRollbacks"), is(metrics0.get("RwRollbacks") + 1)); + assertThat(actualMetrics0.get("RoRollbacks"), is(metrics0.get("RoRollbacks") + 1)); + } + + /** + * Tests that TotalCommits and RoCommits are incremented when a SQL engine is used. + */ + @Test + void testSqlImplicitTransaction() { + Map<String, Long> metrics0 = metricValues(0); + Map<String, Long> metrics1 = metricValues(1); + + sql(0, "select * from " + TABLE_NAME); Review Comment: What about RW metric and thus RW sql query? ########## modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java: ########## @@ -135,7 +136,12 @@ public CompletableFuture<Void> rollbackTimeoutExceededAsync() { } @Override - public CompletableFuture<Void> finish(boolean commit, HybridTimestamp executionTimestamp, boolean full, boolean timeoutExceeded) { + public CompletableFuture<Void> finish( + boolean commit, + @Nullable HybridTimestamp executionTimestamp, Review Comment: In what cases do we propagate null `executionTimestamp` for RO transactions? ########## modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java: ########## @@ -173,15 +173,15 @@ void finishFull( * @param timestampTracker Observable timestamp tracker is used to determine the read timestamp for read-only transactions. Each client * should pass its own tracker to provide linearizability between read-write and read-only transactions started by this client. * @param commitPartition Partition to store a transaction state. {@code null} if nothing was enlisted into the transaction. - * @param commit {@code true} if a commit requested. + * @param commitIntent {@code true} if a commit requested. Review Comment: It's just a rename? You think this makes the method clearer? ########## modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java: ########## @@ -1148,11 +1158,16 @@ public CompletableFuture<Void> executeWriteIntentSwitchAsync(Runnable runnable) return runAsync(runnable, writeIntentSwitchPool); } - void completeReadOnlyTransactionFuture(TxIdAndTimestamp txIdAndTimestamp, boolean timeoutExceeded) { - finishedTxs.add(1); - + void completeReadOnlyTransactionFuture( + boolean commit, + boolean implicit, Review Comment: Is it used? ########## modules/transactions/src/integrationTest/java/org/apache/ignite/internal/tx/ItTransactionMetricsTest.java: ########## @@ -0,0 +1,256 @@ +/* + * 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.ignite.internal.tx; + +import static org.apache.ignite.internal.AssignmentsTestUtils.awaitAssignmentsStabilizationOnDefaultZone; +import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; + +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import org.apache.ignite.Ignite; +import org.apache.ignite.internal.ClusterPerClassIntegrationTest; +import org.apache.ignite.internal.metrics.LongMetric; +import org.apache.ignite.internal.metrics.MetricSet; +import org.apache.ignite.internal.tx.metrics.TransactionMetricsSource; +import org.apache.ignite.table.KeyValueView; +import org.apache.ignite.tx.Transaction; +import org.apache.ignite.tx.TransactionOptions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +/** + * Tests transaction metrics. + */ +public class ItTransactionMetricsTest extends ClusterPerClassIntegrationTest { + public static final String TABLE_NAME = "test_table_name"; + + @Override + protected int initialNodes() { + return 2; + } + + @BeforeAll + void createTable() throws Exception { + sql("CREATE TABLE " + TABLE_NAME + " (id INT PRIMARY KEY, val VARCHAR)"); + + awaitAssignmentsStabilizationOnDefaultZone(CLUSTER.aliveNode()); + } + + /** + * Returns a key value view for the table {@link #TABLE_NAME}. + * + * @param nodeIndex Node index to create a key value view. + * @return Key value view. + */ + private KeyValueView<Integer, String> keyValueView(int nodeIndex) { + return keyValueView(CLUSTER.node(nodeIndex)); + } + + /** + * Returns a key value view for the table {@link #TABLE_NAME}. + * + * @param node Node to create a key value view. + * @return Key value view. + */ + private KeyValueView<Integer, String> keyValueView(Ignite node) { + return node.tables().table(TABLE_NAME).keyValueView(Integer.class, String.class); + } + + /** + * Returns a snapshot of transaction metrics for the given node. + * + * @param nodeIndex Node index to capture transaction metrics. + * @return Snapshot of transaction metrics. + */ + private Map<String, Long> metricValues(int nodeIndex) { + var values = new HashMap<String, Long>(); + + MetricSet txMetrics = unwrapIgniteImpl(node(nodeIndex)) + .metricManager() + .metricSnapshot() + .metrics() + .get(TransactionMetricsSource.SOURCE_NAME); + + txMetrics.iterator().forEachRemaining(metric -> { + if (metric instanceof LongMetric) { + values.put(metric.name(), ((LongMetric) metric).value()); + } + }); + + return values; + } + + private static void testMetricValues(Map<String, Long> initial, Map<String, Long> actual, String... ignored) { + assertThat("Number of metrics should be the same.", initial.size(), is(actual.size())); + + var exclude = Set.of(ignored); + + for (Map.Entry<String, Long> e : initial.entrySet()) { + if (!exclude.contains(e.getKey())) { + assertThat("Metric name = " + e.getKey(), actual.get(e.getKey()), is(e.getValue())); + } + } + } + + /** + * Tests that TotalCommits and RoCommits are incremented when a read only transaction successfully committed. + * + * @param implicit {@code true} if a transaction should be implicit and {@code false} otherwise. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testCommitReadOnlyTransaction(boolean implicit) { + Map<String, Long> metrics0 = metricValues(0); + Map<String, Long> metrics1 = metricValues(1); + + Transaction tx = implicit ? null : node(0).transactions().begin(new TransactionOptions().readOnly(true)); + keyValueView(0).get(tx, 12); + if (!implicit) { + tx.commit(); + } + + Map<String, Long> actualMetrics0 = metricValues(0); + Map<String, Long> actualMetrics1 = metricValues(1); + + // Check that there are no updates on the node 1. + testMetricValues(metrics1, actualMetrics1); + + // Check that all transaction metrics ere not changed except TotalCommits and RoCommits. + testMetricValues(metrics0, actualMetrics0, "TotalCommits", "RoCommits"); + + assertThat(actualMetrics0.get("TotalCommits"), is(metrics0.get("TotalCommits") + 1)); + assertThat(actualMetrics0.get("RoCommits"), is(metrics0.get("RoCommits") + 1)); + } + + /** + * Tests that TotalCommits and RwCommits are incremented when a read write transaction successfully committed. + * + * @param implicit {@code true} if a transaction should be implicit and {@code false} otherwise. + */ + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testCommitReadWriteTransaction(boolean implicit) { + Map<String, Long> metrics0 = metricValues(0); + Map<String, Long> metrics1 = metricValues(1); + + Transaction tx = implicit ? null : node(0).transactions().begin(); + keyValueView(0).put(tx, 12, "value"); + if (!implicit) { + tx.commit(); + } + + Map<String, Long> actualMetrics0 = metricValues(0); + Map<String, Long> actualMetrics1 = metricValues(1); + + // Check that there are no updates on the node 1. + testMetricValues(metrics1, actualMetrics1); + + // Check that all transaction metrics ere not changed except TotalCommits and RwCommits. + testMetricValues(metrics0, actualMetrics0, "TotalCommits", "RwCommits"); + + assertThat(actualMetrics0.get("TotalCommits"), is(metrics0.get("TotalCommits") + 1)); + assertThat(actualMetrics0.get("RwCommits"), is(metrics0.get("RwCommits") + 1)); + } + + /** + * Tests that TotalCommits and RwCommits/RoCommits are incremented when an "empty" transaction successfully committed. + * Empty means that there are no entries enlisted into the transaction. + */ + @Test + void testCommitEmptyTransaction() { + Map<String, Long> metrics0 = metricValues(0); + Map<String, Long> metrics1 = metricValues(1); + + Transaction rwTx = node(0).transactions().begin(); + rwTx.commit(); + + Transaction roTx = node(0).transactions().begin(new TransactionOptions().readOnly(true)); + roTx.commit(); + + Map<String, Long> actualMetrics0 = metricValues(0); + Map<String, Long> actualMetrics1 = metricValues(1); + + // Check that there are no updates on the node 1. + testMetricValues(metrics1, actualMetrics1); + + // Check that all transaction metrics ere not changed except TotalCommits, RwCommits and RoCommits. + testMetricValues(metrics0, actualMetrics0, "TotalCommits", "RwCommits", "RoCommits"); + + assertThat(actualMetrics0.get("TotalCommits"), is(metrics0.get("TotalCommits") + 2)); + assertThat(actualMetrics0.get("RwCommits"), is(metrics0.get("RwCommits") + 1)); + assertThat(actualMetrics0.get("RoCommits"), is(metrics0.get("RoCommits") + 1)); + } + + /** + * Tests that TotalRollbacks and RwRollbacks/RoRollbacks are incremented when a transaction rolled back. + */ + @Test + void testRollbackTransaction() { + Map<String, Long> metrics0 = metricValues(0); + Map<String, Long> metrics1 = metricValues(1); + + Transaction rwTx = node(0).transactions().begin(); + keyValueView(0).put(rwTx, 12, "value"); + rwTx.rollback(); + + Transaction roTx = node(0).transactions().begin(new TransactionOptions().readOnly(true)); + keyValueView(0).get(roTx, 12); + roTx.rollback(); Review Comment: I'd also check that total RW/RO are incremented in case of rollback caused by exception and timeout. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
