Merge branch 'google:main' into main

Author: krykania

dumper/app/build.gradle

@@ -128,6 +128,8 @@ dependencies {
     implementation libs.logback.classic
     implementation libs.jcl.over.slf4j
 
+    implementation libs.async.profiler
+
     runtimeOnly libs.postgresql
     runtimeOnly libs.snowflake.jdbc
     runtimeOnly libs.redshift.jdbc

dumper/app/gradle.lockfile

@@ -4,6 +4,7 @@
 ch.qos.logback:logback-classic:1.3.16=compileClasspath,runtimeClasspath,sources,testCompileClasspath,testFixturesRuntimeClasspath,testRuntimeClasspath
 ch.qos.logback:logback-core:1.3.16=compileClasspath,runtimeClasspath,testCompileClasspath,testFixturesRuntimeClasspath,testRuntimeClasspath
 com.amazon.redshift:redshift-jdbc42:2.1.0.32=runtimeClasspath,testFixturesRuntimeClasspath,testRuntimeClasspath
+tools.profiler:async-profiler:4.2=compileClasspath,runtimeClasspath,testCompileClasspath,testFixturesRuntimeClasspath,testRuntimeClasspath
 com.amazonaws:aws-java-sdk-cloudwatch:1.12.791=compileClasspath,runtimeClasspath,testCompileClasspath,testFixturesRuntimeClasspath,testRuntimeClasspath
 com.amazonaws:aws-java-sdk-core:1.12.791=compileClasspath,runtimeClasspath,testCompileClasspath,testFixturesRuntimeClasspath,testRuntimeClasspath
 com.amazonaws:aws-java-sdk-redshift:1.12.791=compileClasspath,runtimeClasspath,testCompileClasspath,testFixturesRuntimeClasspath,testRuntimeClasspath

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/Main.java

@@ -20,7 +20,14 @@
 
 import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.nio.file.*;
+import java.util.Collections;
+import java.util.Map;
 import java.util.Objects;
+import one.profiler.AsyncProfiler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -51,13 +58,21 @@ private static void printErrorMessages(Throwable e) {
 
   public static void main(String... args) throws Exception {
     try {
+      AsyncProfiler asyncProfiler = AsyncProfiler.getInstance();
+      asyncProfiler.execute("start,event=cpu,interval=10ms");
+
       StartUpMetaInfoProcessor.printMetaInfo();
 
       if (args.length == 0) {
         args = new String[] {"--help"};
       }
 
-      MetadataDumper metadataDumper = new MetadataDumper(args);
+      MetadataDumper metadataDumper =
+          new MetadataDumper(
+              (outputFileLocation) -> {
+                stopProfileAndAppendToZip(asyncProfiler, outputFileLocation);
+              },
+              args);
 
       if (!metadataDumper.run()) {
         System.exit(1);
@@ -74,4 +89,29 @@ public static void main(String... args) throws Exception {
       System.exit(1);
     }
   }
+
+  private static void stopProfileAndAppendToZip(
+      AsyncProfiler asyncProfiler, String outputFileLocation) {
+    try {
+      File tempFlameGraph = File.createTempFile("flamegraph", ".html");
+      String stopCommand = "stop,output=flamegraph,file=" + tempFlameGraph.getAbsolutePath();
+      asyncProfiler.execute(stopCommand);
+
+      moveFileToZip(outputFileLocation, tempFlameGraph, "flamegraph.html");
+    } catch (Exception ignored) {
+    }
+  }
+
+  private static void moveFileToZip(String zipFile, File file, String entryName)
+      throws IOException {
+    Map<String, String> env = Collections.singletonMap("create", "false");
+    URI zipUri = URI.create("jar:" + Paths.get(zipFile).toUri());
+
+    try (FileSystem zipFs = FileSystems.newFileSystem(zipUri, env)) {
+      Path pathInZip = zipFs.getPath(entryName);
+
+      Files.copy(file.toPath(), pathInZip, StandardCopyOption.REPLACE_EXISTING);
+      Files.deleteIfExists(file.toPath());
+    }
+  }
 }

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/MetadataDumper.java

@@ -61,17 +61,24 @@ public class MetadataDumper {
   private static final Pattern GCS_PATH_PATTERN =
       Pattern.compile("gs://(?<bucket>[^/]+)/(?<path>.*)");
 
-  private TelemetryProcessor telemetryProcessor;
-  private ConnectorArguments connectorArguments;
+  private final TelemetryProcessor telemetryProcessor;
+  private final ConnectorArguments connectorArguments;
+  private final ShutdownHook shutdownHook;
 
   public MetadataDumper(String... args) throws Exception {
+    this((zipPath) -> {}, args);
+  }
+
+  public MetadataDumper(ShutdownHook shutdownHook, String... args) throws Exception {
     this.connectorArguments = new ConnectorArguments(JsonResponseFile.addResponseFiles(args));
     telemetryProcessor =
         new TelemetryProcessor(
             TelemetryStrategyFactory.createStrategy(connectorArguments.isTelemetryOn()));
     if (connectorArguments.saveResponseFile()) {
       JsonResponseFile.save(connectorArguments);
     }
+
+    this.shutdownHook = shutdownHook;
   }
 
   public boolean run() throws Exception {
@@ -158,6 +165,8 @@ protected boolean run(@Nonnull Connector connector) throws Exception {
           connectorArguments, state, stopwatch, requiredTaskSucceeded);
       telemetryProcessor.processTelemetry(fileSystem);
     } finally {
+      shutdownHook.shutdown(outputFileLocation);
+
       // We must do this in finally after the ZipFileSystem has been closed.
       File outputFile = new File(outputFileLocation);
       if (outputFile.isFile()) {
@@ -301,4 +310,9 @@ private void logFinalSummary(
           linePrinter.println("Dumper execution: " + stateToPrint);
         });
   }
+
+  @FunctionalInterface
+  public interface ShutdownHook {
+    void shutdown(String outputzip);
+  }
 }

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/cloudera/manager/AbstractClouderaTimeSeriesTask.java

@@ -18,6 +18,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.base.Preconditions;
+import com.google.edwmigration.dumper.application.dumper.task.TaskCategory;
 import java.net.URI;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
@@ -33,12 +34,14 @@ abstract class AbstractClouderaTimeSeriesTask extends AbstractClouderaManagerTas
   private final ZonedDateTime startDate;
   private final ZonedDateTime endDate;
   private final TimeSeriesAggregation tsAggregation;
+  private final TaskCategory taskCategory;
 
   public AbstractClouderaTimeSeriesTask(
       @Nonnull String targetPath,
       @Nonnull ZonedDateTime startDate,
       @Nonnull ZonedDateTime endDate,
-      @Nonnull TimeSeriesAggregation tsAggregation) {
+      @Nonnull TimeSeriesAggregation tsAggregation,
+      @Nonnull TaskCategory taskCategory) {
     super(targetPath);
     Preconditions.checkNotNull(targetPath, "Target path must be not null.");
     Preconditions.checkState(!targetPath.isEmpty(), "Target file path must be not empty.");
@@ -50,6 +53,13 @@ public AbstractClouderaTimeSeriesTask(
     this.startDate = startDate;
     this.endDate = endDate;
     this.tsAggregation = tsAggregation;
+    this.taskCategory = taskCategory;
+  }
+
+  @Nonnull
+  @Override
+  public TaskCategory getCategory() {
+    return taskCategory;
   }
 
   protected JsonNode requestTimeSeriesChart(ClouderaManagerHandle handle, String query)

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/cloudera/manager/ClouderaAPIHostsTask.java

@@ -80,7 +80,11 @@ protected void doRun(
           hosts.add(ClouderaHostDTO.create(apiHost.getId(), apiHost.getName()));
         }
       }
-      handle.initHostsIfNull(hosts);
+      if (hosts.isEmpty()) {
+        throw new MetadataDumperUsageException(
+            "No hosts were found in any of the initialized Cloudera clusters.");
+      }
+      handle.initHosts(hosts);
     }
   }
 }

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/cloudera/manager/ClouderaCMFHostsTask.java

@@ -20,15 +20,11 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.io.ByteSink;
 import com.google.edwmigration.dumper.application.dumper.connector.cloudera.manager.ClouderaManagerHandle.ClouderaClusterDTO;
-import com.google.edwmigration.dumper.application.dumper.connector.cloudera.manager.ClouderaManagerHandle.ClouderaHostDTO;
-import com.google.edwmigration.dumper.application.dumper.connector.cloudera.manager.dto.CMFHostDTO;
-import com.google.edwmigration.dumper.application.dumper.connector.cloudera.manager.dto.CMFHostListDTO;
 import com.google.edwmigration.dumper.application.dumper.task.TaskCategory;
 import com.google.edwmigration.dumper.application.dumper.task.TaskRunContext;
 import java.io.Writer;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
 import java.util.List;
 import javax.annotation.Nonnull;
 import org.apache.http.client.methods.CloseableHttpResponse;
@@ -67,7 +63,6 @@ protected void doRun(
     }
 
     final URI baseURI = handle.getBaseURI();
-    List<ClouderaHostDTO> hosts = new ArrayList<>();
     try (Writer writer = sink.asCharSink(StandardCharsets.UTF_8).openBufferedStream()) {
       for (ClouderaClusterDTO cluster : clusters) {
         if (cluster.getId() == null) {
@@ -95,14 +90,7 @@ protected void doRun(
         String stringifiedHosts = hostsJson.toString();
         writer.write(stringifiedHosts);
         writer.write('\n');
-
-        CMFHostListDTO apiHosts = parseJsonStringToObject(stringifiedHosts, CMFHostListDTO.class);
-        for (CMFHostDTO apiHost : apiHosts.getHosts()) {
-          hosts.add(ClouderaHostDTO.create(apiHost.getId(), apiHost.getName()));
-        }
       }
     }
-
-    handle.initHostsIfNull(hosts);
   }
 }

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/cloudera/manager/ClouderaClusterCPUChartTask.java

@@ -20,6 +20,7 @@
 import com.google.common.io.ByteSink;
 import com.google.edwmigration.dumper.application.dumper.MetadataDumperUsageException;
 import com.google.edwmigration.dumper.application.dumper.connector.cloudera.manager.ClouderaManagerHandle.ClouderaClusterDTO;
+import com.google.edwmigration.dumper.application.dumper.task.TaskCategory;
 import com.google.edwmigration.dumper.application.dumper.task.TaskRunContext;
 import java.io.Writer;
 import java.nio.charset.StandardCharsets;
@@ -40,13 +41,16 @@
  * language.
  */
 public class ClouderaClusterCPUChartTask extends AbstractClouderaTimeSeriesTask {
-  private static final Logger logger = LoggerFactory.getLogger(ClouderaCMFHostsTask.class);
+  private static final Logger logger = LoggerFactory.getLogger(ClouderaClusterCPUChartTask.class);
   private static final String TS_CPU_QUERY_TEMPLATE =
       "SELECT cpu_percent_across_hosts WHERE entityName = \"%s\" AND category = CLUSTER";
 
   public ClouderaClusterCPUChartTask(
-      ZonedDateTime startDate, ZonedDateTime endDate, TimeSeriesAggregation tsAggregation) {
-    super("cluster-cpu.jsonl", startDate, endDate, tsAggregation);
+      ZonedDateTime startDate,
+      ZonedDateTime endDate,
+      TimeSeriesAggregation tsAggregation,
+      TaskCategory taskCategory) {
+    super("cluster-cpu.jsonl", startDate, endDate, tsAggregation, taskCategory);
   }
 
   @Override

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/cloudera/manager/ClouderaConnectorVerifier.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2022-2025 Google LLC
+ * Copyright 2013-2021 CompilerWorks
+ *
+ * Licensed 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 com.google.edwmigration.dumper.application.dumper.connector.cloudera.manager;
+
+import com.google.edwmigration.dumper.application.dumper.ConnectorArguments;
+import java.io.IOException;
+import javax.annotation.Nonnull;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.util.EntityUtils;
+
+/** Utility class for verifying the preconditions and configuration of a Cloudera Connector. */
+public final class ClouderaConnectorVerifier {
+
+  private ClouderaConnectorVerifier() {}
+
+  /**
+   * Verifies the validity of the connector configuration and environment.
+   *
+   * <p>This involves checking connectivity to the API, verifying the existence of the specified
+   * cluster (if provided), and ensuring other preconditions are met.
+   *
+   * @param handle The handle to the Cloudera Manager API.
+   * @param arguments The connector arguments containing target cluster details.
+   * @throws ClouderaConnectorException If verification fails due to missing resources, API errors,
+   *     or connectivity issues.
+   */
+  public static void verify(
+      @Nonnull ClouderaManagerHandle handle, @Nonnull ConnectorArguments arguments)
+      throws ClouderaConnectorException {
+    verifyClusterExists(handle, arguments.getCluster());
+  }
+
+  private static void verifyClusterExists(ClouderaManagerHandle handle, String clusterName)
+      throws ClouderaConnectorException {
+    if (clusterName == null) {
+      return;
+    }
+    String endpoint = String.format("%s/clusters/%s", handle.getApiURI(), clusterName);
+    HttpGet httpGet = new HttpGet(endpoint);
+
+    try (CloseableHttpResponse response = handle.getHttpClient().execute(httpGet)) {
+
+      int statusCode = response.getStatusLine().getStatusCode();
+
+      if (statusCode == 404) {
+        throw new ClouderaConnectorException(
+            String.format("Specified cluster '%s' not found.", clusterName));
+      }
+
+      if (!isHttpStatusSuccess(statusCode)) {
+        String errorMsg = EntityUtils.toString(response.getEntity());
+        throw new ClouderaConnectorException(
+            String.format(
+                "Unexpected API error checking cluster '%s'. Code: %d. Message: %s",
+                clusterName, statusCode, errorMsg));
+      }
+    } catch (IOException e) {
+      throw new ClouderaConnectorException(
+          String.format(
+              "Failed to communicate with Cloudera Manager API while checking cluster '%s'.",
+              clusterName),
+          e);
+    }
+  }
+
+  private static boolean isHttpStatusSuccess(int statusCode) {
+    return statusCode >= 200 && statusCode < 300;
+  }
+}

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/cloudera/manager/ClouderaHostRAMChartTask.java

@@ -20,6 +20,7 @@
 import com.google.common.io.ByteSink;
 import com.google.edwmigration.dumper.application.dumper.MetadataDumperUsageException;
 import com.google.edwmigration.dumper.application.dumper.connector.cloudera.manager.ClouderaManagerHandle.ClouderaHostDTO;
+import com.google.edwmigration.dumper.application.dumper.task.TaskCategory;
 import com.google.edwmigration.dumper.application.dumper.task.TaskRunContext;
 import java.io.Writer;
 import java.nio.charset.StandardCharsets;
@@ -40,14 +41,17 @@
  */
 public class ClouderaHostRAMChartTask extends AbstractClouderaTimeSeriesTask {
 
-  private static final Logger logger = LoggerFactory.getLogger(ClouderaCMFHostsTask.class);
+  private static final Logger logger = LoggerFactory.getLogger(ClouderaHostRAMChartTask.class);
 
   private static final String TS_RAM_QUERY_TEMPLATE =
       "select swap_used, physical_memory_used, physical_memory_total, physical_memory_cached, physical_memory_buffers where entityName = \"%s\"";
 
   public ClouderaHostRAMChartTask(
-      ZonedDateTime startDate, ZonedDateTime endDate, TimeSeriesAggregation tsAggregation) {
-    super("host-ram.jsonl", startDate, endDate, tsAggregation);
+      ZonedDateTime startDate,
+      ZonedDateTime endDate,
+      TimeSeriesAggregation tsAggregation,
+      TaskCategory taskCategory) {
+    super("host-ram.jsonl", startDate, endDate, tsAggregation, taskCategory);
   }
 
   @Override