[CDAP-21257]Capturing Dataproc operations

cdap-runtime-ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/common/DataprocMetric.java

@@ -16,6 +16,7 @@
 
 package io.cdap.cdap.runtime.spi.common;
 
+import com.google.api.gax.rpc.StatusCode;
 import com.google.common.base.Strings;
 import io.cdap.cdap.runtime.spi.runtimejob.LaunchMode;
 import javax.annotation.Nullable;
@@ -32,21 +33,28 @@
   private final LaunchMode launchMode;
   @Nullable
   private final String imageVersion;
+  @Nullable
+  private final String method;
+  @Nullable
+  private final StatusCode.Code statusCode;
 
   private DataprocMetric(String metricName, String region, @Nullable Exception exception,
-      @Nullable LaunchMode launchMode, @Nullable String imageVersion) {
+      @Nullable LaunchMode launchMode, @Nullable String imageVersion, @Nullable String method,
+                         @Nullable StatusCode.Code statusCode) {
     this.metricName = metricName;
     this.region = region;
     this.exception = exception;
     this.launchMode = launchMode;
     this.imageVersion = imageVersion;
+    this.method = method;
+    this.statusCode = statusCode;
   }
 
   public String getMetricName() {
     return metricName;
   }
 
   @Nullable
   public String getImageVersion() {
     if (!Strings.isNullOrEmpty(imageVersion)) {
       // return major.minor
@@ -72,6 +80,16 @@
     return launchMode;
   }
 
+  @Nullable
+  public StatusCode.Code  getStatusCode() {
+    return statusCode;
+  }
+
+  @Nullable
+  public String getMethod() {
+    return method;
+  }
+
   /**
    * Returns a builder to create a DataprocMetric.
    *
@@ -93,6 +111,10 @@
     private Exception exception;
     @Nullable
     private LaunchMode launchMode;
+    @Nullable
+    private String method;
+    @Nullable
+    private StatusCode.Code statusCode;
 
     private Builder(String metricName) {
       this.metricName = metricName;
@@ -113,11 +135,21 @@
       return this;
     }
 
+    public Builder setStatusCode(@Nullable StatusCode.Code statusCode) {
+      this.statusCode = statusCode;
+      return this;
+    }
+
     public Builder setImageVersion(String imageVersion) {
       this.imageVersion = imageVersion;
       return this;
     }
 
+    public Builder setMethod(@Nullable String method) {
+      this.method = method;
+      return this;
+    }
+
     /**
      * Returns a DataprocMetric.
      *
@@ -128,7 +160,7 @@
         // region should always be set unless there is a bug in the code
         throw new IllegalStateException("Dataproc metric is missing the region");
       }
-      return new DataprocMetric(metricName, region, exception, launchMode, imageVersion);
+      return new DataprocMetric(metricName, region, exception, launchMode, imageVersion, method, statusCode);
     }
   }
 }

cdap-runtime-ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/common/DataprocUtils.java

@@ -140,7 +140,7 @@
           .put(16, 401) // UNAUTHENTICATED
           .build();
 
   public interface ConnectionProvider {
     HttpURLConnection getConnection(URL url) throws IOException;
   }
 
@@ -375,19 +375,7 @@
    * Emit a dataproc metric.
    **/
   public static void emitMetric(ProvisionerContext context, DataprocMetric dataprocMetric) {
-    StatusCode.Code statusCode;
-    Exception e = dataprocMetric.getException();
-    if (e == null) {
-      statusCode = StatusCode.Code.OK;
-    } else {
-      Throwable cause = e.getCause();
-      if (cause instanceof ApiException) {
-        ApiException apiException = (ApiException) cause;
-        statusCode = apiException.getStatusCode().getCode();
-      } else {
-        statusCode = StatusCode.Code.INTERNAL;
-      }
-    }
+    StatusCode.Code statusCode = getCode(dataprocMetric);
     ImmutableMap.Builder<String, String> tags = ImmutableMap.<String, String>builder()
         .put("reg", dataprocMetric.getRegion())
         .put("sc", statusCode.toString());
@@ -397,10 +385,30 @@
     if (!Strings.isNullOrEmpty(dataprocMetric.getImageVersion())) {
       tags.put("imgVer", dataprocMetric.getImageVersion());
     }
+    if (!Strings.isNullOrEmpty(dataprocMetric.getMethod())) {
+      tags.put("method", dataprocMetric.getMethod());
+    }
     ProvisionerMetrics metrics = context.getMetrics(tags.build());
     metrics.count(dataprocMetric.getMetricName(), 1);
   }
 
+  private static StatusCode.Code getCode(DataprocMetric dataprocMetric) {
+    StatusCode.Code statusCode = StatusCode.Code.OK;
+    Exception e = dataprocMetric.getException();
+    if (dataprocMetric.getStatusCode() != null) {
+      statusCode = dataprocMetric.getStatusCode();
+    } else if (e != null) {
+      Throwable cause = e.getCause();
+      if (cause instanceof ApiException) {
+        ApiException apiException = (ApiException) cause;
+        statusCode = apiException.getStatusCode().getCode();
+      } else {
+        statusCode = StatusCode.Code.INTERNAL;
+      }
+    }
+    return statusCode;
+  }
+
   /**
    * Makes a request to the metadata server that lives on the VM, as described at
    * https://cloud.google.com/compute/docs/storing-retrieving-metadata.

cdap-runtime-ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocClient.java

@@ -47,6 +47,7 @@
 import com.google.common.base.Strings;
 import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Iterables;
 import com.google.longrunning.Operation;
 import com.google.longrunning.OperationsClient;
 import com.google.protobuf.Duration;
@@ -822,6 +823,26 @@ String getClusterFailureMsg(String name) throws RetryableProvisionException {
     return "";
   }
 
+  /**
+   * Get the latest operation by cluster name and operation type.
+   */
+  public Optional<Operation> getLatestOperation(String clusterName, String operationType)
+    throws RetryableProvisionException {
+    String operationName = String.format("projects/%s/regions/%s/operations", conf.getProjectId(), conf.getRegion());
+    String filter = String.format("clusterName=\"%s\" AND operationType=\"%s\"", clusterName, operationType);
+
+    try {
+      OperationsClient.ListOperationsPagedResponse response =
+        client.getOperationsClient().listOperations(operationName, filter);
+
+      return Optional.ofNullable(Iterables.getFirst(response.getPage().getValues(), null));
+
+    } catch (ApiException e) {
+      throw DataprocUtils.handleApiException(null, e,
+        "Failed to retrieve latest operation metadata for cluster " + clusterName, errorCategory);
+    }
+  }
+
   /**
    * Get information about the specified cluster. The cluster will not be present if it could not be
    * found.

cdap-runtime-ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocProvisioner.java

@@ -16,16 +16,21 @@
 
 package io.cdap.cdap.runtime.spi.provisioner.dataproc;
 
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.StatusCode;
 import com.google.cloud.dataproc.v1.ClusterOperationMetadata;
 import com.google.cloud.dataproc.v1.ClusterStatus.State;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Stopwatch;
 import com.google.common.collect.ImmutableSet;
+import com.google.longrunning.Operation;
 import io.cdap.cdap.api.exception.ErrorCategory;
 import io.cdap.cdap.api.exception.ErrorType;
 import io.cdap.cdap.runtime.spi.ProgramRunInfo;
 import io.cdap.cdap.runtime.spi.RuntimeMonitorType;
 import io.cdap.cdap.runtime.spi.common.DataprocImageVersion;
+import io.cdap.cdap.runtime.spi.common.DataprocMetric;
 import io.cdap.cdap.runtime.spi.common.DataprocUtils;
 import io.cdap.cdap.runtime.spi.provisioner.Cluster;
 import io.cdap.cdap.runtime.spi.provisioner.ClusterStatus;
@@ -38,17 +43,21 @@
 import io.cdap.cdap.runtime.spi.ssh.SSHKeyPair;
 import io.cdap.cdap.runtime.spi.ssh.SSHPublicKey;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
 import java.util.regex.Pattern;
 import javax.annotation.Nullable;
+
+import io.grpc.Status;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -73,6 +82,9 @@
   //A lock to use for cluster reuse
   private static final String REUSE_LOCK = "reuse";
 
+  private static final Set<ClusterStatus> TERMINAL_STATES =
+    EnumSet.of(ClusterStatus.RUNNING, ClusterStatus.FAILED, ClusterStatus.NOT_EXISTS);
+
   private final DataprocClientFactory clientFactory;
 
   @SuppressWarnings("WeakerAccess")
@@ -417,7 +429,7 @@
    * @param clusterName cluster name
    * @param runLabels map with labels to look for the cluster
    * @throws Exception if wait failed, interrupted or cluster can't be found even after operation
    * finished.
    */
   private static void waitForLabelsUpdateToApply(DataprocClient client, DataprocConf conf,
       Future<?> updateLabelsFuture, String clusterName, Map<String, String> runLabels) throws Exception {
@@ -471,16 +483,21 @@
     DataprocConf conf = DataprocConf.create(createContextProperties(context));
     String clusterName = cluster.getName();
 
-    ClusterStatus status = cluster.getStatus();
+    ClusterStatus previousStatus = cluster.getStatus();
     // if we are skipping the delete, need to avoid checking the real cluster status and pretend like it is deleted.
-    if (conf.isSkipDelete() && status == ClusterStatus.DELETING) {
+    if (conf.isSkipDelete() && previousStatus == ClusterStatus.DELETING) {
       return ClusterStatus.NOT_EXISTS;
     }
 
+    ClusterStatus status;
     try (DataprocClient client = clientFactory.create(conf, context.getErrorCategory())) {
       status = client.getClusterStatus(clusterName);
       DataprocUtils.emitMetric(context, conf.getRegion(), getImageVersion(context, conf),
           "provisioner.clusterStatus.response.count");
+
+      if (hasReachedTerminalState(previousStatus, status)) {
+        emitLroMetric(context, client, conf, previousStatus, clusterName);
+      }
       return status;
     } catch (Exception e) {
       DataprocUtils.emitMetric(context, conf.getRegion(), getImageVersion(context, conf),
@@ -489,6 +506,68 @@
     }
   }
 
+  private boolean hasReachedTerminalState(ClusterStatus previous, ClusterStatus current) {
+    return !TERMINAL_STATES.contains(previous) && TERMINAL_STATES.contains(current);
+  }
+
+  /**
+   * Emits a metric for the outcome of a Long Running Operation (LRO) when a cluster
+   * transitions to a terminal status.
+   *
+   * @param context the provisioner context
+   * @param client the dataproc client
+   * @param conf the dataproc configuration
+   * @param previousStatus the status of the cluster before the transition
+   * @param clusterName the name of the cluster
+   */
+  private void emitLroMetric(ProvisionerContext context, DataprocClient client, DataprocConf conf,
+                             ClusterStatus previousStatus, String clusterName) {
+    String method = getMethod(previousStatus);
+    if (method == null) {
+      return;
+    }
+
+    try {
+      Optional<Operation> operation = client.getLatestOperation(clusterName, method);
+
+      if (!operation.isPresent() || !operation.get().getDone()) {
+        return;
+      }
+
+      Operation op = operation.get();
+      String metricName = "provisioner.operation.response.count";
+      String imageVersion = getImageVersion(context, conf);
+
+      DataprocMetric.Builder builder = DataprocMetric.builder(metricName)
+        .setRegion(conf.getRegion())
+        .setImageVersion(imageVersion)
+        .setMethod(method);
+
+      if (op.hasError()) {
+        Status.Code grpcCode = Status.fromCodeValue(op.getError().getCode()).getCode();
+        StatusCode gaxStatusCode = GrpcStatusCode.of(grpcCode);
+        builder.setStatusCode(gaxStatusCode.getCode());
+      }
+
+      DataprocUtils.emitMetric(context, builder.build());
+    } catch (RetryableProvisionException | RuntimeException ex) {
+      LOG.warn("Failed to retrieve LRO operation metadata for metric emission on cluster {}", clusterName, ex);
+    }
+  }
+
+  @Nullable
+  private String getMethod(ClusterStatus status) {
+    switch (status) {
+      case CREATING:
+        return "CREATE";
+      case DELETING:
+        return "DELETE";
+      default:
+        return null;
+    }
+  }
+
+
   @Override
   public String getClusterFailureMsg(ProvisionerContext context, Cluster cluster) throws Exception {
     DataprocConf conf = DataprocConf.create(createContextProperties(context));