JAVA-3148: Implementing OpenTelemetry for Observability

bom/pom.xml

@@ -75,6 +75,11 @@
         <artifactId>java-driver-metrics-microprofile</artifactId>
         <version>4.18.2-SNAPSHOT</version>
       </dependency>
+      <dependency>
+        <groupId>org.apache.cassandra</groupId>
+        <artifactId>java-driver-open-telemetry</artifactId>
+        <version>4.18.2-SNAPSHOT</version>
+      </dependency>
       <dependency>
         <groupId>com.datastax.oss</groupId>
         <artifactId>native-protocol</artifactId>

core/src/main/java/com/datastax/dse/driver/api/core/graph/GraphExecutionInfo.java

@@ -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;
@@ -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();
 
@@ -61,6 +69,12 @@ public interface GraphExecutionInfo {
    */
   int getSuccessfulExecutionIndex();
 
+  /** @return Exception raised by the driver to the application. */
+  @Nullable
+  default Throwable getDriverError() {
+    return null;
+  }
+
   /**
    * The errors encountered on previous coordinators, if any.
    *

core/src/main/java/com/datastax/dse/driver/internal/core/cql/continuous/ContinuousRequestHandlerBase.java

@@ -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;
@@ -627,7 +629,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(error.getCause(), null);
           lock.lock();
           try {
             abort(error.getCause(), false);
@@ -644,7 +646,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(error.getCause(), null);
           sendRequest(statement, null, executionIndex, retryCount, scheduleSpeculativeExecution);
         }
       } else {
@@ -739,7 +741,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}.
      */
@@ -760,15 +763,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(error, response);
             abort(error, false);
           }
         } catch (Throwable t) {
-          trackNodeError(node, t);
+          trackNodeError(t, response);
           abort(t, false);
         }
       } finally {
@@ -831,7 +834,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(null).withServerResponse(result, frame).build();
         if (result instanceof Rows) {
           DseRowsMetadata rowsMetadata = (DseRowsMetadata) ((Rows) result).getMetadata();
           if (columnDefinitions == null) {
@@ -902,7 +906,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);
@@ -911,7 +915,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(error, frame);
           sendRequest(statement, null, executionIndex, retryCount, false);
         } else if (error instanceof QueryValidationException
             || error instanceof FunctionFailureException
@@ -923,7 +927,7 @@ private void processErrorResponse(@NonNull Error errorMessage) {
           NodeMetricUpdater metricUpdater = ((DefaultNode) node).getMetricUpdater();
           metricUpdater.incrementCounter(
               DefaultNodeMetric.OTHER_ERRORS, executionProfile.getName());
-          trackNodeError(node, error);
+          trackNodeError(error, frame);
           abort(error, true);
         } else {
           try {
@@ -1062,7 +1066,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(illegalStateException, null);
                     fatalError = illegalStateException;
                   } else {
                     LOG.trace(
@@ -1081,18 +1085,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(prepareError, null);
                         fatalError = prepareError;
                       }
                     }
                   } else if (exception instanceof RequestThrottlingException) {
-                    trackNodeError(node, exception);
+                    trackNodeError(exception, null);
                     fatalError = exception;
                   }
                   if (fatalError == null) {
                     LOG.trace("[{}] Re-prepare failed, trying next node", logPrefix);
                     recordError(node, exception);
-                    trackNodeError(node, exception);
+                    trackNodeError(exception, null);
                     sendRequest(statement, null, executionIndex, retryCount, false);
                   }
                 }
@@ -1120,18 +1124,18 @@ private void processRetryVerdict(@NonNull RetryVerdict verdict, @NonNull Throwab
       switch (verdict.getRetryDecision()) {
         case RETRY_SAME:
           recordError(node, error);
-          trackNodeError(node, error);
+          trackNodeError(error, null);
           sendRequest(
               verdict.getRetryRequest(statement), node, executionIndex, retryCount + 1, false);
           break;
         case RETRY_NEXT:
           recordError(node, error);
-          trackNodeError(node, error);
+          trackNodeError(error, null);
           sendRequest(
               verdict.getRetryRequest(statement), null, executionIndex, retryCount + 1, false);
           break;
         case RETHROW:
-          trackNodeError(node, error);
+          trackNodeError(error, null);
           abort(error, true);
           break;
         case IGNORE:
@@ -1444,12 +1448,15 @@ private void reenableAutoReadIfNeeded() {
 
     // ERROR HANDLING
 
-    private void trackNodeError(@NonNull Node node, @NonNull Throwable error) {
+    private void trackNodeError(@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(
+                latencyNanos,
+                createExecutionInfo(error).withServerResponse(frame).build(),
+                logPrefix);
       }
     }
 
@@ -1563,21 +1570,25 @@ 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);
+            context.getRequestTracker().onNodeSuccess(nodeLatencyNanos, executionInfo, logPrefix);
           }
-          context
-              .getRequestTracker()
-              .onSuccess(statement, totalLatencyNanos, executionProfile, node, logPrefix);
+          context.getRequestTracker().onSuccess(totalLatencyNanos, executionInfo, logPrefix);
         }
       } else {
         Throwable error = (Throwable) pageOrError;
         if (future.completeExceptionally(error)) {
           context
               .getRequestTracker()
-              .onError(statement, error, totalLatencyNanos, executionProfile, node, logPrefix);
+              .onError(totalLatencyNanos, createExecutionInfo(error).build(), logPrefix);
           if (error instanceof DriverTimeoutException) {
             throttler.signalTimeout(ContinuousRequestHandlerBase.this);
             session
@@ -1591,18 +1602,14 @@ 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(Throwable error) {
+      return DefaultExecutionInfo.builder(
           statement,
           node,
           startedSpeculativeExecutionsCount.get(),
           executionIndex,
+          error,
           errors,
-          pagingState,
-          response,
-          true,
           session,
           context,
           executionProfile);

core/src/main/java/com/datastax/dse/driver/internal/core/graph/GraphExecutionInfoConverter.java

@@ -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;
@@ -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() {
@@ -78,6 +84,12 @@ public int getSuccessfulExecutionIndex() {
         return graphExecutionInfo.getSuccessfulExecutionIndex();
       }
 
+      @Nullable
+      @Override
+      public Throwable getDriverError() {
+        return graphExecutionInfo.getDriverError();
+      }
+
       @NonNull
       @Override
       public List<Entry<Node, Throwable>> getErrors() {
@@ -146,6 +158,11 @@ public GraphStatement<?> getStatement() {
         return (GraphStatement<?>) executionInfo.getRequest();
       }
 
+      @Override
+      public DriverExecutionProfile getExecutionProfile() {
+        return executionInfo.getExecutionProfile();
+      }
+
       @Override
       public Node getCoordinator() {
         return executionInfo.getCoordinator();
@@ -161,6 +178,11 @@ public int getSuccessfulExecutionIndex() {
         return executionInfo.getSuccessfulExecutionIndex();
       }
 
+      @Override
+      public Throwable getDriverError() {
+        return executionInfo.getDriverError();
+      }
+
       @Override
       public List<Entry<Node, Throwable>> getErrors() {
         return executionInfo.getErrors();