Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Follow up on #1640 - Add ExecutionInfo to RequestTracker callbacks #1949

Open
wants to merge 2 commits into
base: 4.x
Choose a base branch
from
+572 −235
Open

Follow up on #1640 - Add ExecutionInfo to RequestTracker callbacks #1949

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
43db6f6
Add ExecutionInfo to RequestTracker methods
Jun 5, 2023
495fdda
Add ExecutionInfo to RequestTracker callbacks
lukasz-antoniak Jun 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions core/src/main/java/com/datastax/dse/driver/api/core/graph/GraphExecutionInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,10 @@
package com.datastax.dse.driver.api.core.graph;

import com.datastax.oss.driver.api.core.DefaultProtocolVersion;
import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
import com.datastax.oss.driver.api.core.metadata.Node;
import com.datastax.oss.driver.api.core.specex.SpeculativeExecutionPolicy;
import edu.umd.cs.findbugs.annotations.Nullable;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
Expand All @@ -37,6 +39,12 @@ public interface GraphExecutionInfo {
/** The statement that was executed. */
GraphStatement<?> getStatement();

/** @return Execution profile applied when executing given request. */
@Nullable
default DriverExecutionProfile getExecutionProfile() {
return null;
}

/** The node that was used as a coordinator to successfully complete the query. */
Node getCoordinator();

Expand Down
76 changes: 47 additions & 29 deletions ...va/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,11 +19,13 @@

import com.datastax.dse.driver.api.core.DseProtocolVersion;
import com.datastax.dse.driver.api.core.cql.continuous.ContinuousAsyncResultSet;
import com.datastax.dse.driver.api.core.graph.AsyncGraphResultSet;
import com.datastax.dse.driver.internal.core.DseProtocolFeature;
import com.datastax.dse.driver.internal.core.cql.DseConversions;
import com.datastax.dse.protocol.internal.request.Revise;
import com.datastax.dse.protocol.internal.response.result.DseRowsMetadata;
import com.datastax.oss.driver.api.core.AllNodesFailedException;
import com.datastax.oss.driver.api.core.AsyncPagingIterable;
import com.datastax.oss.driver.api.core.CqlIdentifier;
import com.datastax.oss.driver.api.core.DriverTimeoutException;
import com.datastax.oss.driver.api.core.NodeUnavailableException;
Expand Down Expand Up @@ -626,7 +628,7 @@ public void operationComplete(@NonNull Future<java.lang.Void> future) {
Throwable error = future.cause();
if (error instanceof EncoderException
&& error.getCause() instanceof FrameTooLongException) {
trackNodeError(node, error.getCause());
trackNodeError(node, error.getCause(), null);
lock.lock();
try {
abort(error.getCause(), false);
Expand All @@ -643,7 +645,7 @@ public void operationComplete(@NonNull Future<java.lang.Void> future) {
.getMetricUpdater()
.incrementCounter(DefaultNodeMetric.UNSENT_REQUESTS, executionProfile.getName());
recordError(node, error);
trackNodeError(node, error.getCause());
trackNodeError(node, error.getCause(), null);
sendRequest(statement, null, executionIndex, retryCount, scheduleSpeculativeExecution);
}
} else {
Expand Down Expand Up @@ -738,7 +740,8 @@ private void onPageTimeout(int expectedPage) {
* Invoked when a continuous paging response is received, either a successful or failed one.
*
* <p>Delegates further processing to appropriate methods: {@link #processResultResponse(Result,
* Frame)} if the response was successful, or {@link #processErrorResponse(Error)} if it wasn't.
* Frame)} if the response was successful, or {@link #processErrorResponse(Error, Frame)} if it
* wasn't.
*
* @param response the received {@link Frame}.
*/
Expand All @@ -759,15 +762,15 @@ public void onResponse(@NonNull Frame response) {
processResultResponse((Result) responseMessage, response);
} else if (responseMessage instanceof Error) {
LOG.trace("[{}] Got error response", logPrefix);
processErrorResponse((Error) responseMessage);
processErrorResponse((Error) responseMessage, response);
} else {
IllegalStateException error =
new IllegalStateException("Unexpected response " + responseMessage);
trackNodeError(node, error);
trackNodeError(node, error, response);
abort(error, false);
}
} catch (Throwable t) {
trackNodeError(node, t);
trackNodeError(node, t, response);
abort(t, false);
}
} finally {
Expand Down Expand Up @@ -830,7 +833,8 @@ public void onFailure(@NonNull Throwable error) {
private void processResultResponse(@NonNull Result result, @Nullable Frame frame) {
assert lock.isHeldByCurrentThread();
try {
ExecutionInfo executionInfo = createExecutionInfo(result, frame);
ExecutionInfo executionInfo =
createExecutionInfo().withServerResponse(result, frame).build();
if (result instanceof Rows) {
DseRowsMetadata rowsMetadata = (DseRowsMetadata) ((Rows) result).getMetadata();
if (columnDefinitions == null) {
Expand Down Expand Up @@ -901,7 +905,7 @@ private void processResultResponse(@NonNull Result result, @Nullable Frame frame
* @param errorMessage the error message received.
*/
@SuppressWarnings("GuardedBy") // this method is only called with the lock held
private void processErrorResponse(@NonNull Error errorMessage) {
private void processErrorResponse(@NonNull Error errorMessage, @NonNull Frame frame) {
assert lock.isHeldByCurrentThread();
if (errorMessage instanceof Unprepared) {
processUnprepared((Unprepared) errorMessage);
Expand All @@ -910,7 +914,7 @@ private void processErrorResponse(@NonNull Error errorMessage) {
if (error instanceof BootstrappingException) {
LOG.trace("[{}] {} is bootstrapping, trying next node", logPrefix, node);
recordError(node, error);
trackNodeError(node, error);
trackNodeError(node, error, frame);
sendRequest(statement, null, executionIndex, retryCount, false);
} else if (error instanceof QueryValidationException
|| error instanceof FunctionFailureException
Expand All @@ -922,7 +926,7 @@ private void processErrorResponse(@NonNull Error errorMessage) {
NodeMetricUpdater metricUpdater = ((DefaultNode) node).getMetricUpdater();
metricUpdater.incrementCounter(
DefaultNodeMetric.OTHER_ERRORS, executionProfile.getName());
trackNodeError(node, error);
trackNodeError(node, error, frame);
abort(error, true);
} else {
try {
Expand Down Expand Up @@ -1061,7 +1065,7 @@ private void processUnprepared(@NonNull Unprepared errorMessage) {
+ "This usually happens when you run a 'USE...' query after "
+ "the statement was prepared.",
Bytes.toHexString(idToReprepare), Bytes.toHexString(repreparedId)));
trackNodeError(node, illegalStateException);
trackNodeError(node, illegalStateException, null);
fatalError = illegalStateException;
} else {
LOG.trace(
Expand All @@ -1080,18 +1084,18 @@ private void processUnprepared(@NonNull Unprepared errorMessage) {
|| prepareError instanceof FunctionFailureException
|| prepareError instanceof ProtocolError) {
LOG.trace("[{}] Unrecoverable error on re-prepare, rethrowing", logPrefix);
trackNodeError(node, prepareError);
trackNodeError(node, prepareError, null);
fatalError = prepareError;
}
}
} else if (exception instanceof RequestThrottlingException) {
trackNodeError(node, exception);
trackNodeError(node, exception, null);
fatalError = exception;
}
if (fatalError == null) {
LOG.trace("[{}] Re-prepare failed, trying next node", logPrefix);
recordError(node, exception);
trackNodeError(node, exception);
trackNodeError(node, exception, null);
sendRequest(statement, null, executionIndex, retryCount, false);
}
}
Expand Down Expand Up @@ -1119,18 +1123,18 @@ private void processRetryVerdict(@NonNull RetryVerdict verdict, @NonNull Throwab
switch (verdict.getRetryDecision()) {
case RETRY_SAME:
recordError(node, error);
trackNodeError(node, error);
trackNodeError(node, error, null);
sendRequest(
verdict.getRetryRequest(statement), node, executionIndex, retryCount + 1, false);
break;
case RETRY_NEXT:
recordError(node, error);
trackNodeError(node, error);
trackNodeError(node, error, null);
sendRequest(
verdict.getRetryRequest(statement), null, executionIndex, retryCount + 1, false);
break;
case RETHROW:
trackNodeError(node, error);
trackNodeError(node, error, null);
abort(error, true);
break;
case IGNORE:
Expand Down Expand Up @@ -1443,12 +1447,20 @@ private void reenableAutoReadIfNeeded() {

// ERROR HANDLING

private void trackNodeError(@NonNull Node node, @NonNull Throwable error) {
private void trackNodeError(
@NonNull Node node, @NonNull Throwable error, @Nullable Frame frame) {
if (nodeErrorReported.compareAndSet(false, true)) {
long latencyNanos = System.nanoTime() - this.messageStartTimeNanos;
context
.getRequestTracker()
.onNodeError(this.statement, error, latencyNanos, executionProfile, node, logPrefix);
.onNodeError(
this.statement,
error,
latencyNanos,
executionProfile,
node,
createExecutionInfo().withServerResponse(frame).build(),
logPrefix);
}
}

Expand Down Expand Up @@ -1562,21 +1574,32 @@ private void completeResultSetFuture(
if (resultSetClass.isInstance(pageOrError)) {
if (future.complete(resultSetClass.cast(pageOrError))) {
throttler.signalSuccess(ContinuousRequestHandlerBase.this);

ExecutionInfo executionInfo = null;
if (pageOrError instanceof AsyncPagingIterable) {
executionInfo = ((AsyncPagingIterable<?, ?>) pageOrError).getExecutionInfo();
} else if (pageOrError instanceof AsyncGraphResultSet) {
executionInfo = ((AsyncGraphResultSet) pageOrError).getRequestExecutionInfo();
}

if (nodeSuccessReported.compareAndSet(false, true)) {
context
.getRequestTracker()
.onNodeSuccess(statement, nodeLatencyNanos, executionProfile, node, logPrefix);
.onNodeSuccess(
statement, nodeLatencyNanos, executionProfile, node, executionInfo, logPrefix);
}
context
.getRequestTracker()
.onSuccess(statement, totalLatencyNanos, executionProfile, node, logPrefix);
.onSuccess(
statement, totalLatencyNanos, executionProfile, node, executionInfo, logPrefix);
}
} else {
Throwable error = (Throwable) pageOrError;
if (future.completeExceptionally(error)) {
context
.getRequestTracker()
.onError(statement, error, totalLatencyNanos, executionProfile, node, logPrefix);
.onError(
statement, error, totalLatencyNanos, executionProfile, node, null, logPrefix);
if (error instanceof DriverTimeoutException) {
throttler.signalTimeout(ContinuousRequestHandlerBase.this);
session
Expand All @@ -1590,18 +1613,13 @@ private void completeResultSetFuture(
}

@NonNull
private ExecutionInfo createExecutionInfo(@NonNull Result result, @Nullable Frame response) {
ByteBuffer pagingState =
result instanceof Rows ? ((Rows) result).getMetadata().pagingState : null;
return new DefaultExecutionInfo(
private DefaultExecutionInfo.Builder createExecutionInfo() {
return DefaultExecutionInfo.builder(
statement,
node,
startedSpeculativeExecutionsCount.get(),
executionIndex,
errors,
pagingState,
response,
true,
session,
context,
executionProfile);
Expand Down
11 changes: 11 additions & 0 deletions ...rc/main/java/com/datastax/dse/driver/internal/core/graph/GraphExecutionInfoConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package com.datastax.dse.driver.internal.core.graph;

import com.datastax.dse.driver.api.core.graph.GraphStatement;
import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
import com.datastax.oss.driver.api.core.cql.ExecutionInfo;
import com.datastax.oss.driver.api.core.cql.QueryTrace;
import com.datastax.oss.driver.api.core.cql.Statement;
Expand Down Expand Up @@ -62,6 +63,11 @@ public Statement<?> getStatement() {
throw new ClassCastException("GraphStatement cannot be cast to Statement");
}

@Override
public DriverExecutionProfile getExecutionProfile() {
return graphExecutionInfo.getExecutionProfile();
}

@Nullable
@Override
public Node getCoordinator() {
Expand Down Expand Up @@ -146,6 +152,11 @@ public GraphStatement<?> getStatement() {
return (GraphStatement<?>) executionInfo.getRequest();
}

@Override
public DriverExecutionProfile getExecutionProfile() {
return executionInfo.getExecutionProfile();
}

@Override
public Node getCoordinator() {
return executionInfo.getCoordinator();
Expand Down
Loading