You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vp...@apache.org on 2024/09/17 07:17:17 UTC
(ignite-3) 01/01: Added tracing for single get command.
This is an automated email from the ASF dual-hosted git repository.
vpyatkov pushed a commit to branch ignite-23199
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 6f2387fea8846cc724e9bec4b913177755d1e0f2
Author: Vladislav Pyatkov <vl...@gmail.com>
AuthorDate: Tue Sep 17 10:12:42 2024 +0300
Added tracing for single get command.
---
.../handler/ClientInboundMessageHandler.java | 31 +++++++++++-----
.../requests/table/ClientTupleGetRequest.java | 5 +--
.../ignite/internal/client/ReliableChannel.java | 7 ++--
.../client/table/ClientKeyValueBinaryView.java | 13 ++++++-
.../ignite/internal/client/table/ClientTable.java | 12 +++++--
.../ignite/internal/replicator/ReplicaService.java | 3 ++
.../ignite/internal/table/AbstractTableView.java | 3 +-
.../internal/table/RecordBinaryViewImpl.java | 4 ++-
.../replicator/PartitionReplicaListener.java | 42 ++++++++++++++--------
.../distributed/storage/InternalTableImpl.java | 14 +++++---
10 files changed, 95 insertions(+), 39 deletions(-)
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
index 000c696ad9..f6fee0e372 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
@@ -17,7 +17,10 @@
package org.apache.ignite.client.handler;
+import static org.apache.ignite.internal.tracing.Instrumentation.mark;
+import static org.apache.ignite.internal.tracing.Instrumentation.measure;
import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.firstNotNull;
import static org.apache.ignite.lang.ErrorGroups.Client.HANDSHAKE_HEADER_ERR;
import static org.apache.ignite.lang.ErrorGroups.Client.PROTOCOL_COMPATIBILITY_ERR;
@@ -304,6 +307,8 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
/** {@inheritDoc} */
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
+ mark("channelReadMark");
+
ByteBuf byteBuf = (ByteBuf) msg;
// Each inbound handler in a pipeline has to release the received messages.
@@ -311,7 +316,7 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
metrics.bytesReceivedAdd(byteBuf.readableBytes() + ClientMessageCommon.HEADER_SIZE);
// Packer buffer is released by Netty on send, or by inner exception handlers below.
- var packer = getPacker(ctx.alloc());
+ var packer = measure(() -> getPacker(ctx.alloc()), "getPacker");
switch (state) {
case STATE_BEFORE_HANDSHAKE:
@@ -327,7 +332,9 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
case STATE_HANDSHAKE_RESPONSE_SENT:
assert clientContext != null : "Client context != null";
- processOperation(ctx, unpacker, packer);
+ processOperation(ctx, unpacker, packer).thenRun(() -> {
+ mark("channelReadEndMark");
+ });
break;
@@ -582,14 +589,14 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
return new ClientMessagePacker(alloc.buffer());
}
- private void processOperation(ChannelHandlerContext ctx, ClientMessageUnpacker in, ClientMessagePacker out) {
+ private CompletableFuture processOperation(ChannelHandlerContext ctx, ClientMessageUnpacker in, ClientMessagePacker out) {
long requestId = -1;
int opCode = -1;
metrics.requestsActiveIncrement();
try {
- opCode = in.unpackInt();
- requestId = in.unpackLong();
+ opCode = measure(() -> in.unpackInt(), "readOpId");
+ requestId = measure(() -> in.unpackLong(), "readReqId");
if (LOG.isTraceEnabled()) {
LOG.trace("Client request started [id=" + requestId + ", op=" + opCode
@@ -617,12 +624,14 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
metrics.requestsProcessedIncrement();
metrics.requestsActiveDecrement();
+
+ return nullCompletedFuture();
} else {
var reqId = requestId;
var op = opCode;
- fut.whenComplete((Object res, Object err) -> {
- in.close();
+ return fut.whenComplete((Object res, Object err) -> {
+ measure(() -> in.close(), "closeMsgIn");
metrics.requestsActiveDecrement();
if (err != null) {
@@ -631,8 +640,8 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
metrics.requestsFailedIncrement();
} else {
- out.setLong(observableTimestampIdx, observableTimestamp(out));
- write(out, ctx);
+ measure(() -> out.setLong(observableTimestampIdx, observableTimestamp(out)), "writeObservableTimestamp");
+ measure(() -> write(out, ctx), "writeCtx");
metrics.requestsProcessedIncrement();
@@ -650,6 +659,8 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
writeError(requestId, opCode, t, ctx, false);
metrics.requestsFailedIncrement();
+
+ return nullCompletedFuture();
}
}
@@ -659,6 +670,8 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
int opCode,
long requestId
) throws IgniteInternalCheckedException {
+ mark("processOperation:" + opCode);
+
switch (opCode) {
case ClientOp.HEARTBEAT:
return null;
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java
index f80d2c5d48..de5dc08055 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java
@@ -20,6 +20,7 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTx;
+import static org.apache.ignite.internal.tracing.Instrumentation.measure;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
@@ -48,8 +49,8 @@ public class ClientTupleGetRequest {
ClientResourceRegistry resources
) {
return readTableAsync(in, tables).thenCompose(table -> {
- var tx = readTx(in, out, resources);
- return readTuple(in, table, true).thenCompose(keyTuple -> {
+ var tx = measure(() -> readTx(in, out, resources), "readTx");
+ return measure(() -> readTuple(in, table, true), "readTuple").thenCompose(keyTuple -> {
return table.recordView().getAsync(tx, keyTuple)
.thenAccept(t -> ClientTableCommon.writeTupleOrNil(out, t, TuplePart.KEY_AND_VAL, table.schemaView()));
});
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java b/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java
index ce56bb44a6..3053af74fe 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.client;
import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.apache.ignite.internal.tracing.Instrumentation.measure;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
import static org.apache.ignite.lang.ErrorGroups.Client.CLUSTER_ID_MISMATCH_ERR;
@@ -243,11 +244,11 @@ public final class ReliableChannel implements AutoCloseable {
@Nullable RetryPolicy retryPolicyOverride,
boolean expectNotifications
) {
- return ClientFutureUtils.doWithRetryAsync(
- () -> getChannelAsync(preferredNodeName)
+ return ClientFutureUtils.doWithRetryAsync(() -> measure(() -> getChannelAsync(preferredNodeName), "getChannelAsync")
.thenCompose(ch -> serviceAsyncInternal(opCode, payloadWriter, payloadReader, expectNotifications, ch)),
null,
- ctx -> shouldRetry(opCode, ctx, retryPolicyOverride));
+ ctx -> shouldRetry(opCode, ctx, retryPolicyOverride)
+ );
}
/**
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
index 5b63c3620c..c79809aad1 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
@@ -17,6 +17,9 @@
package org.apache.ignite.internal.client.table;
+import static org.apache.ignite.internal.tracing.Instrumentation.end;
+import static org.apache.ignite.internal.tracing.Instrumentation.mark;
+import static org.apache.ignite.internal.tracing.Instrumentation.start;
import static org.apache.ignite.internal.util.CompletableFutures.emptyCollectionCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.emptyMapCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
@@ -76,7 +79,15 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
/** {@inheritDoc} */
@Override
public Tuple get(@Nullable Transaction tx, Tuple key) {
- return sync(getAsync(tx, key));
+ start(false);
+ mark("kvClientGetMark");
+
+ Tuple res = sync(getAsync(tx, key));
+
+ mark("kvClientGetEndMark");
+ end();
+
+ return res;
}
/** {@inheritDoc} */
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
index 9818661563..1963c04919 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.client.table;
+import static org.apache.ignite.internal.tracing.Instrumentation.measure;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.lang.ErrorGroups.Client.CONNECTION_ERR;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
@@ -446,12 +447,17 @@ public class ClientTable implements Table {
: getPartitionAssignment();
// Wait for schema and partition assignment.
- CompletableFuture.allOf(schemaFut, partitionsFut)
+ measure(() -> CompletableFuture.allOf(schemaFut, partitionsFut), "getSchemaAndAssignment")
.thenCompose(v -> {
ClientSchema schema = schemaFut.getNow(null);
String txPreferredNodeName = getPreferredNodeName(provider, partitionsFut.getNow(null), schema);
- return ClientLazyTransaction.ensureStarted(tx, ch, txPreferredNodeName).thenCompose(unused -> {
+ CompletableFuture<ClientTransaction> ensureTxStateFut = measure(
+ () -> ClientLazyTransaction.ensureStarted(tx, ch, txPreferredNodeName),
+ "ensureStartedTxState"
+ );
+
+ return ensureTxStateFut.thenCompose(unused -> {
// Update preferred node name after starting the transaction.
// All operations for a given explicit transaction should go to the same node (tx coordinator).
String opPreferredNodeName = getPreferredNodeName(provider, partitionsFut.getNow(null), schema);
@@ -467,7 +473,7 @@ public class ClientTable implements Table {
})
// Read resulting schema and the rest of the response.
- .thenCompose(t -> loadSchemaAndReadData(t, reader))
+ .thenCompose(t -> measure(() -> loadSchemaAndReadData(t, reader), "loadSchemaAndReadData"))
.whenComplete((res, err) -> {
if (err == null) {
fut.complete(res);
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaService.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaService.java
index b9c76dfeb6..11a45c4947 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaService.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaService.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.replicator;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+import static org.apache.ignite.internal.tracing.Instrumentation.mark;
import static org.apache.ignite.internal.util.ExceptionUtils.matchAny;
import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
import static org.apache.ignite.internal.util.ExceptionUtils.withCause;
@@ -132,6 +133,8 @@ public class ReplicaService {
* @see ReplicationTimeoutException If the response could not be received due to a timeout.
*/
private <R> CompletableFuture<R> sendToReplica(String targetNodeConsistentId, ReplicaRequest req) {
+ mark("sendToReplica:" + targetNodeConsistentId);
+
CompletableFuture<R> res = new CompletableFuture<>();
messagingService.invoke(
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/AbstractTableView.java b/modules/table/src/main/java/org/apache/ignite/internal/table/AbstractTableView.java
index 2778dbf84b..c89102fe30 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/AbstractTableView.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/AbstractTableView.java
@@ -43,6 +43,7 @@ import org.apache.ignite.internal.table.criteria.SqlSerializer.Builder;
import org.apache.ignite.internal.table.distributed.replicator.IncompatibleSchemaVersionException;
import org.apache.ignite.internal.table.distributed.replicator.InternalSchemaVersionMismatchException;
import org.apache.ignite.internal.table.distributed.schema.SchemaVersions;
+import org.apache.ignite.internal.tracing.Instrumentation;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.lang.AsyncCursor;
import org.apache.ignite.lang.Cursor;
@@ -138,7 +139,7 @@ abstract class AbstractTableView<R> implements CriteriaQuerySource<R> {
? schemaVersions.schemaVersionAtNow(tbl.tableId())
: schemaVersions.schemaVersionAt(((InternalTransaction) tx).startTimestamp(), tbl.tableId());
- return schemaVersionFuture
+ return Instrumentation.measure(() -> schemaVersionFuture,"readSchemaVersion")
.thenCompose(schemaVersion -> action.act(schemaVersion)
.handle((res, ex) -> {
if (ex == null) {
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/RecordBinaryViewImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/RecordBinaryViewImpl.java
index 4295dba0e3..5ebf8360f0 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/RecordBinaryViewImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/RecordBinaryViewImpl.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.table;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.lang.IgniteExceptionMapperUtil.convertToPublicFuture;
+import static org.apache.ignite.internal.tracing.Instrumentation.measure;
import static org.apache.ignite.internal.util.CompletableFutures.trueCompletedFuture;
import static org.apache.ignite.internal.util.ViewUtils.checkKeysForNulls;
import static org.apache.ignite.internal.util.ViewUtils.sync;
@@ -95,7 +96,8 @@ public class RecordBinaryViewImpl extends AbstractTableView<Tuple> implements Re
Objects.requireNonNull(keyRec);
return doOperation(tx, (schemaVersion) -> {
- Row keyRow = marshal(keyRec, schemaVersion, true); // Convert to portable format to pass TX/storage layer.
+ // Convert to portable format to pass TX/storage layer.
+ Row keyRow = measure(() -> marshal(keyRec, schemaVersion, true), "marshal");
return tbl.get(keyRow, (InternalTransaction) tx).thenApply(row -> wrap(row, schemaVersion));
});
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
index 3936743bb3..399814e7b6 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
@@ -37,6 +37,7 @@ import static org.apache.ignite.internal.table.distributed.replicator.RemoteReso
import static org.apache.ignite.internal.table.distributed.replicator.ReplicatorUtils.beginRwTxTs;
import static org.apache.ignite.internal.table.distributed.replicator.ReplicatorUtils.latestIndexDescriptorInBuildingStatus;
import static org.apache.ignite.internal.table.distributed.replicator.ReplicatorUtils.rwTxActiveCatalogVersion;
+import static org.apache.ignite.internal.tracing.Instrumentation.measure;
import static org.apache.ignite.internal.tx.TransactionIds.beginTimestamp;
import static org.apache.ignite.internal.tx.TxState.ABANDONED;
import static org.apache.ignite.internal.tx.TxState.ABORTED;
@@ -466,7 +467,7 @@ public class PartitionReplicaListener implements ReplicaListener {
@Override
public CompletableFuture<ReplicaResult> invoke(ReplicaRequest request, String senderId) {
- return ensureReplicaIsPrimary(request)
+ return measure(() -> ensureReplicaIsPrimary(request), "ensureReplicaIsPrimary")
.thenCompose(res -> processRequest(request, res.get1(), senderId, res.get2()))
.thenApply(res -> {
if (res instanceof ReplicaResult) {
@@ -520,9 +521,12 @@ public class PartitionReplicaListener implements ReplicaListener {
return processGetEstimatedSizeRequest();
}
- @Nullable HybridTimestamp opTs = getTxOpTimestamp(request);
- @Nullable HybridTimestamp opTsIfDirectRo = (request instanceof ReadOnlyDirectReplicaRequest) ? opTs : null;
- @Nullable HybridTimestamp txTs = getTxStartTimestamp(request);
+ @Nullable HybridTimestamp opTs = measure(() -> getTxOpTimestamp(request), "getTxOpTimestamp");
+ @Nullable HybridTimestamp opTsIfDirectRo = measure(
+ () -> (request instanceof ReadOnlyDirectReplicaRequest) ? opTs : null,
+ "getOpTsIfDirectRo"
+ );
+ @Nullable HybridTimestamp txTs = measure(() -> getTxStartTimestamp(request), "getTxStartTimestamp");
if (txTs == null) {
txTs = opTsIfDirectRo;
}
@@ -550,8 +554,12 @@ public class PartitionReplicaListener implements ReplicaListener {
}
};
- return schemaSyncService.waitForMetadataCompleteness(opTs).thenRun(validateClo).thenCompose(ignored ->
- processOperationRequestWithTxRwCounter(senderId, request, isPrimary, opTsIfDirectRo, leaseStartTime));
+ return measure(
+ () -> schemaSyncService.waitForMetadataCompleteness(opTs).thenRun(validateClo),
+ "waitForMetadataCompleteness"
+ ).thenCompose(ignored ->
+ processOperationRequestWithTxRwCounter(senderId, request, isPrimary, opTsIfDirectRo, leaseStartTime)
+ );
}
private CompletableFuture<Long> processGetEstimatedSizeRequest() {
@@ -2104,19 +2112,21 @@ public class PartitionReplicaListener implements ReplicaListener {
// It means that there exists one and only one non-empty readResult for any read timestamp for the given key.
// Which in turn means that if we have found non empty readResult during PK index iteration
// we can proceed with readResult resolution and stop the iteration.
- try (Cursor<RowId> cursor = getFromPkIndex(pk)) {
+ try (Cursor<RowId> cursor = measure(() -> getFromPkIndex(pk), "getFromPkIndex")) {
// TODO https://issues.apache.org/jira/browse/IGNITE-18767 scan of multiple write intents should not be needed
List<ReadResult> writeIntents = new ArrayList<>();
List<ReadResult> regularEntries = new ArrayList<>();
for (RowId rowId : cursor) {
- ReadResult readResult = mvDataStorage.read(rowId, ts);
+ ReadResult readResult = measure(() -> mvDataStorage.read(rowId, ts), "readResultFromCursor");
- if (readResult.isWriteIntent()) {
- writeIntents.add(readResult);
- } else if (!readResult.isEmpty()) {
- regularEntries.add(readResult);
- }
+ measure(() -> {
+ if (readResult.isWriteIntent()) {
+ writeIntents.add(readResult);
+ } else if (!readResult.isEmpty()) {
+ regularEntries.add(readResult);
+ }
+ }, "checkReadResult");
}
// Nothing found in the storage, return null.
@@ -2126,7 +2136,9 @@ public class PartitionReplicaListener implements ReplicaListener {
if (writeIntents.isEmpty()) {
// No write intents, then return the committed value. We already know that regularEntries is not empty.
- return completedFuture(regularEntries.get(0).binaryRow());
+ var bo = measure(() -> regularEntries.get(0).binaryRow(), "readBinaryResult");
+
+ return completedFuture(bo);
} else {
ReadResult writeIntent = writeIntents.get(0);
@@ -2950,7 +2962,7 @@ public class PartitionReplicaListener implements ReplicaListener {
ReadOnlyDirectSingleRowReplicaRequest request,
HybridTimestamp opStartTimestamp
) {
- BinaryTuple primaryKey = resolvePk(request.primaryKey());
+ BinaryTuple primaryKey = measure(() -> resolvePk(request.primaryKey()), "resolvePk");
HybridTimestamp readTimestamp = opStartTimestamp;
if (request.requestType() != RO_GET) {
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
index 39900f36e9..e1d77d855f 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
@@ -43,6 +43,7 @@ import static org.apache.ignite.internal.partition.replicator.network.replicatio
import static org.apache.ignite.internal.partition.replicator.network.replication.RequestType.RW_UPSERT_ALL;
import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toTablePartitionIdMessage;
import static org.apache.ignite.internal.table.distributed.storage.RowBatch.allResultFutures;
+import static org.apache.ignite.internal.tracing.Instrumentation.measure;
import static org.apache.ignite.internal.util.CompletableFutures.completedOrFailedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.emptyListCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
@@ -750,18 +751,23 @@ public class InternalTableImpl implements InternalTable {
BinaryRowEx row,
BiFunction<TablePartitionId, Long, ReplicaRequest> op
) {
- InternalTransaction tx = txManager.begin(observableTimestampTracker, true);
+ InternalTransaction tx = measure(() -> txManager.begin(observableTimestampTracker, true), "startImplicitTx");
int partId = partitionId(row);
TablePartitionId tablePartitionId = new TablePartitionId(tableId, partId);
- CompletableFuture<R> fut = awaitPrimaryReplica(tablePartitionId, tx.startTimestamp())
+ CompletableFuture<R> fut = measure(() -> awaitPrimaryReplica(tablePartitionId, tx.startTimestamp()), "awaitPrimaryReplica")
.thenCompose(primaryReplica -> {
try {
ClusterNode node = getClusterNode(primaryReplica);
- return replicaSvc.invoke(node, op.apply(tablePartitionId, enlistmentConsistencyToken(primaryReplica)));
+ ReplicaRequest request = measure(
+ () -> op.apply(tablePartitionId, enlistmentConsistencyToken(primaryReplica)),
+ "prepareRequest"
+ );
+
+ return replicaSvc.invoke(node, request);
} catch (Throwable e) {
throw new TransactionException(
INTERNAL_ERR,
@@ -840,7 +846,7 @@ public class InternalTableImpl implements InternalTable {
}); // Preserve failed state.
}
- return tx.finish(true, clock.now()).thenApply(ignored -> r);
+ return measure(() -> tx.finish(true, clock.now()), "finishTx").thenApply(ignored -> r);
}).thenCompose(identity());
}