Merge pull request #15234 from cdapio/CDAP-19428

[CDAP-19428] Allow configurable GCP Dataproc Oauth Scopes
cdapio · Jul 14, 2023 · 9e79ee8 · 9e79ee8
2 parents a935317 + c091059
commit 9e79ee8
Show file tree

Hide file tree

Showing 4 changed files with 33 additions and 3 deletions.
diff --git a/...-dataproc/src/main/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocClient.java b/...-dataproc/src/main/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocClient.java
@@ -183,7 +183,7 @@ ClusterOperationMetadata createCluster(String name, String imageVersion,
       metadata.putAll(conf.getClusterMetaData());
 
       GceClusterConfig.Builder clusterConfig = GceClusterConfig.newBuilder()
-          .addServiceAccountScopes(DataprocConf.CLOUD_PLATFORM_SCOPE)
+          .addAllServiceAccountScopes(conf.getScopes())
           .setShieldedInstanceConfig(
               ShieldedInstanceConfig.newBuilder()
                   .setEnableSecureBoot(conf.isSecureBootEnabled())

diff --git a/...xt-dataproc/src/main/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocConf.java b/...xt-dataproc/src/main/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocConf.java
@@ -64,6 +64,7 @@ final class DataprocConf {
   // Dataproc will pass it to GCE when creating the GCE cluster.
   // It can be overridden by profile runtime arguments (system.profile.properties.serviceAccount)
   static final String SERVICE_ACCOUNT = "serviceAccount";
+  static final String SCOPES = "scopes";
   static final String ROOT_URL = "root.url";
 
   static final String GCS_BUCKET = "gcsBucket";
@@ -153,6 +154,7 @@ final class DataprocConf {
   private final Map<String, String> clusterMetaData;
   private final Map<String, String> clusterLabels;
   private final List<String> networkTags;
+  private final List<String> scopes;
   private final String initActions;
   private final String autoScalingPolicy;
   private final int idleTtlMinutes;
@@ -199,7 +201,7 @@ private DataprocConf(@Nullable String accountKey, String region, String zone, St
       @Nullable String customImageUri,
       @Nullable Map<String, String> clusterMetaData,
       @Nullable Map<String, String> clusterLabels, List<String> networkTags,
-      @Nullable String initActions, boolean runtimeJobManagerEnabled,
+      List<String> scopes, @Nullable String initActions, boolean runtimeJobManagerEnabled,
       Map<String, String> clusterProperties, @Nullable String autoScalingPolicy, int idleTtlMinutes,
       @Nullable String tokenEndpoint, boolean secureBootEnabled, boolean vTpmEnabled,
       boolean integrityMonitoringEnabled, boolean clusterReuseEnabled,
@@ -213,6 +215,7 @@ private DataprocConf(@Nullable String accountKey, String region, String zone, St
     this.zone = zone;
     this.projectId = projectId;
     this.tempBucket = tempBucket;
+    this.scopes = scopes;
     this.clusterReuseEnabled = clusterReuseEnabled;
     this.clusterReuseThresholdMinutes = clusterReuseThresholdMinutes;
     this.clusterReuseRetryDelayMs = clusterReuseRetryDelayMs;
@@ -426,6 +429,10 @@ List<String> getNetworkTags() {
     return Collections.unmodifiableList(networkTags);
   }
 
+  List<String> getScopes() {
+    return Collections.unmodifiableList(scopes);
+  }
+
   List<String> getInitActions() {
     if (Strings.isNullOrEmpty(initActions)) {
       return Collections.emptyList();
@@ -771,6 +778,15 @@ static DataprocConf create(Map<String, String> properties) {
         properties.getOrDefault(DataprocUtils.TROUBLESHOOTING_DOCS_URL_KEY,
             DataprocUtils.TROUBLESHOOTING_DOCS_URL_DEFAULT);
 
+    final String scopesProperty = String.format("%s,%s",
+        Optional.ofNullable(getString(properties, SCOPES)).orElse(""), CLOUD_PLATFORM_SCOPE);
+    List<String> scopes = Collections.unmodifiableList(
+        Arrays.stream(scopesProperty.split(","))
+            .map(String::trim)
+            .filter(s -> !s.isEmpty())
+            .distinct()
+            .collect(Collectors.toList()));
+
     return new DataprocConf(accountKey, region, zone, projectId, networkHostProjectId, network,
         subnet,
         masterNumNodes, masterCpus, masterMemoryMb, masterDiskGb,
@@ -782,7 +798,7 @@ static DataprocConf create(Map<String, String> properties) {
         stackdriverLoggingEnabled, stackdriverMonitoringEnabled,
         componentGatewayEnabled, skipDelete,
         imageVersion, customImageUri, clusterMetaData, clusterLabels, networkTags,
-        initActions, runtimeJobManagerEnabled, clusterProps, autoScalingPolicy, idleTtl,
+        scopes, initActions, runtimeJobManagerEnabled, clusterProps, autoScalingPolicy, idleTtl,
         tokenEndpoint, secureBootEnabled, vTpmEnabled, integrityMonitoringEnabled,
         clusterReuseEnabled, clusterReuseThresholdMinutes, clusterReuseRetryDelayMs,
         clusterReuseRetryMaxMs, clusterReuseUpdateMaxMs, clusterReuseKey,

diff --git a/cdap-runtime-ext-dataproc/src/main/resources/gcp-dataproc.json b/cdap-runtime-ext-dataproc/src/main/resources/gcp-dataproc.json
@@ -493,6 +493,15 @@
             "placeholder": "projects/<gcp-project-id>/locations/<key-location>/keyRings/<key-ring-name>/cryptoKeys/<key-name>"
           }
         },
+        {
+          "name": "scopes",
+          "label": "OAuth Scopes",
+          "widget-type": "csv",
+          "description": "The OAuth 2.0 scopes that you might need to request to access Google APIs, depending on the level of access you need. Google Cloud Platform Scope (https://www.googleapis.com/auth/cloud-platform) is always included.",
+          "widget-attributes" : {
+            "value-placeholder": ""
+          }
+        },
         {
           "name": "initActions",
           "label": "Initialization Actions",

diff --git a/.../src/test/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocProvisionerTest.java b/.../src/test/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocProvisionerTest.java
@@ -121,6 +121,9 @@ public void testRunKey() throws Exception {
   @Test
   public void testDataprocConf() {
     Map<String, String> props = new HashMap<>();
+    String scopes = String.format("%s,https://www.googleapis.com/auth/drive",
+        DataprocConf.CLOUD_PLATFORM_SCOPE);
+
     props.put(DataprocConf.PROJECT_ID_KEY, "pid");
     props.put("accountKey", "key");
     props.put("region", "region1");
@@ -138,6 +141,7 @@ public void testDataprocConf() {
     props.put("idleTTL", "20");
     props.put("clusterReuseRetryDelayMs", "20");
     props.put("clusterReuseRetryMaxMs", "200");
+    props.put(DataprocConf.SCOPES, scopes);
 
     DataprocConf conf = DataprocConf.create(props);
 
@@ -167,6 +171,7 @@ public void testDataprocConf() {
 
     Assert.assertEquals(20, conf.getClusterReuseRetryDelayMs());
     Assert.assertEquals(200, conf.getClusterReuseRetryMaxMs());
+    Assert.assertEquals(scopes, String.join(",", conf.getScopes()));
   }
 
   @Test