diff --git a/cdap-runtime-ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocServerlessProvisioner.java b/cdap-runtime-ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocServerlessProvisioner.java new file mode 100644 index 000000000000..5055b6986a56 --- /dev/null +++ b/cdap-runtime-ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocServerlessProvisioner.java @@ -0,0 +1,119 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * 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 io.cdap.cdap.runtime.spi.provisioner.dataproc; + +import io.cdap.cdap.runtime.spi.provisioner.Cluster; +import io.cdap.cdap.runtime.spi.provisioner.ClusterStatus; +import io.cdap.cdap.runtime.spi.provisioner.PollingStrategies; +import io.cdap.cdap.runtime.spi.provisioner.PollingStrategy; +import io.cdap.cdap.runtime.spi.provisioner.ProvisionerContext; +import io.cdap.cdap.runtime.spi.provisioner.ProvisionerSpecification; +import io.cdap.cdap.runtime.spi.runtimejob.DataprocClusterInfo; +import io.cdap.cdap.runtime.spi.runtimejob.DataprocServerlessRuntimeJobManager; +import io.cdap.cdap.runtime.spi.runtimejob.RuntimeJobManager; +import java.util.Collections; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Provisioner for executing jobs on Dataproc Serverless. + */ +public class DataprocServerlessProvisioner extends AbstractDataprocProvisioner { + + private static final Logger LOG = LoggerFactory.getLogger(DataprocServerlessProvisioner.class); + + private static final ProvisionerSpecification SPEC = new ProvisionerSpecification( + "gcp-dataproc-serverless", "Dataproc Serverless", + "Execute Spark jobs as serverless workloads on GCP Dataproc Serverless."); + + private static final String CLUSTER_NAME = "dataproc-serverless-mock"; + + public DataprocServerlessProvisioner() { + super(SPEC); + } + + @Override + public void validateProperties(Map properties) { + // Validates the properties (re-uses existing DataprocConf checks) + DataprocConf.create(properties); + } + + @Override + protected String getClusterName(ProvisionerContext context) { + return CLUSTER_NAME; + } + + @Override + public Cluster createCluster(ProvisionerContext context) throws Exception { + LOG.warn("TEST_LOG: Entering DataprocServerlessProvisioner.createCluster. Returning mock RUNNING cluster."); + // No-op for cluster creation. Return a mock cluster that is already in RUNNING state. + Map properties = createContextProperties(context); + return new Cluster(CLUSTER_NAME, ClusterStatus.RUNNING, Collections.emptyList(), properties); + } + + @Override + protected void doDeleteCluster(ProvisionerContext context, Cluster cluster, DataprocConf conf) { + LOG.warn("TEST_LOG: Entering DataprocServerlessProvisioner.doDeleteCluster. This is a mock no-op deletion."); + // No-op for cluster deletion. + } + + @Override + public ClusterStatus getClusterStatus(ProvisionerContext context, Cluster cluster) { + ClusterStatus status = cluster.getStatus(); + LOG.warn("TEST_LOG: DataprocServerlessProvisioner.getClusterStatus. Current status: {}", status); + // If the cluster status is DELETING, report that it is deleted (NOT_EXISTS). + return status == ClusterStatus.DELETING ? ClusterStatus.NOT_EXISTS : status; + } + + @Override + public Cluster getClusterDetail(ProvisionerContext context, Cluster cluster) { + return new Cluster(cluster, getClusterStatus(context, cluster)); + } + + @Override + public PollingStrategy getPollingStrategy(ProvisionerContext context, Cluster cluster) { + // Fixed polling strategy since there is no cluster creation time. + return PollingStrategies.fixedInterval(0, TimeUnit.SECONDS); + } + + @Override + public Optional getRuntimeJobManager(ProvisionerContext context) { + LOG.warn("TEST_LOG: Entering DataprocServerlessProvisioner.getRuntimeJobManager."); + Map properties = createContextProperties(context); + DataprocConf conf = DataprocConf.create(properties); + + try { + String clusterName = getClusterName(context); + String projectId = conf.getProjectId(); + String region = conf.getRegion(); + String bucket = conf.getGcsBucket() != null ? conf.getGcsBucket() : properties.get("bucket"); + + LOG.warn("TEST_LOG: Creating DataprocServerlessRuntimeJobManager for project: {}, region: {}, bucket: {}", + projectId, region, bucket); + return Optional.of(new DataprocServerlessRuntimeJobManager( + new DataprocClusterInfo(context, clusterName, conf.getDataprocCredentials(), + getRootUrl(conf), projectId, region, bucket, getCommonDataprocLabels(context)), + Collections.unmodifiableMap(properties), context.getCDAPVersionInfo())); + } catch (Exception e) { + LOG.warn("TEST_LOG: Exception while initializing DataprocServerlessRuntimeJobManager: ", e); + throw new RuntimeException("Error while getting credentials for Dataproc Serverless. ", e); + } + } +} diff --git a/cdap-runtime-ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/runtimejob/DataprocServerlessRuntimeJobManager.java b/cdap-runtime-ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/runtimejob/DataprocServerlessRuntimeJobManager.java new file mode 100644 index 000000000000..32e2e2d7a32f --- /dev/null +++ b/cdap-runtime-ext-dataproc/src/main/java/io/cdap/cdap/runtime/spi/runtimejob/DataprocServerlessRuntimeJobManager.java @@ -0,0 +1,703 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * 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 io.cdap.cdap.runtime.spi.runtimejob; + +import com.google.api.gax.core.CredentialsProvider; +import com.google.api.gax.core.FixedCredentialsProvider; +import com.google.api.gax.rpc.ApiException; +import com.google.api.gax.rpc.StatusCode; +import com.google.auth.oauth2.GoogleCredentials; +import com.google.cloud.WriteChannel; +import com.google.cloud.dataproc.v1.Batch; +import com.google.cloud.dataproc.v1.BatchControllerClient; +import com.google.cloud.dataproc.v1.BatchControllerSettings; +import com.google.cloud.dataproc.v1.CreateBatchRequest; +import com.google.cloud.dataproc.v1.DeleteBatchRequest; +import com.google.cloud.dataproc.v1.EnvironmentConfig; +import com.google.cloud.dataproc.v1.ExecutionConfig; +import com.google.cloud.dataproc.v1.GetBatchRequest; +import com.google.cloud.dataproc.v1.RuntimeConfig; +import com.google.cloud.dataproc.v1.SparkBatch; +import com.google.cloud.http.HttpTransportOptions; +import com.google.cloud.storage.Blob; +import com.google.cloud.storage.BlobId; +import com.google.cloud.storage.BlobInfo; +import com.google.cloud.storage.Bucket; +import com.google.cloud.storage.BucketInfo; +import com.google.cloud.storage.Storage; +import com.google.cloud.storage.StorageException; +import com.google.cloud.storage.StorageOptions; +import com.google.cloud.storage.StorageRetryStrategy; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Joiner; +import com.google.common.base.Strings; +import com.google.common.io.ByteStreams; +import io.cdap.cdap.api.exception.ErrorCategory; +import io.cdap.cdap.api.exception.ErrorCategory.ErrorCategoryEnum; +import io.cdap.cdap.api.exception.ErrorCodeType; +import io.cdap.cdap.api.exception.ErrorType; +import io.cdap.cdap.api.exception.ErrorUtils; +import io.cdap.cdap.api.exception.ErrorUtils.ActionErrorPair; +import io.cdap.cdap.api.exception.ProgramFailureException; +import io.cdap.cdap.runtime.spi.CacheableLocalFile; +import io.cdap.cdap.runtime.spi.ProgramRunInfo; +import io.cdap.cdap.runtime.spi.VersionInfo; +import io.cdap.cdap.runtime.spi.common.DataprocMetric; +import io.cdap.cdap.runtime.spi.common.DataprocUtils; +import io.cdap.cdap.runtime.spi.provisioner.ProvisionerContext; +import io.cdap.cdap.runtime.spi.provisioner.dataproc.DataprocRuntimeException; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.net.HttpURLConnection; +import java.net.URI; +import java.nio.channels.Channels; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; +import org.apache.twill.api.LocalFile; +import org.apache.twill.filesystem.LocalLocationFactory; +import org.apache.twill.filesystem.Location; +import org.apache.twill.filesystem.LocationFactory; +import org.apache.twill.internal.Constants; +import org.apache.twill.internal.DefaultLocalFile; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.threeten.bp.Duration; + +/** + * Dataproc Serverless runtime job manager. + */ +public class DataprocServerlessRuntimeJobManager implements RuntimeJobManager { + + private static final Logger LOG = LoggerFactory.getLogger(DataprocServerlessRuntimeJobManager.class); + + // Dataproc job properties + public static final String CDAP_RUNTIME_NAMESPACE = "cdap.runtime.namespace"; + public static final String CDAP_RUNTIME_APPLICATION = "cdap.runtime.application"; + public static final String CDAP_RUNTIME_VERSION = "cdap.runtime.version"; + public static final String CDAP_RUNTIME_PROGRAM_TYPE = "cdap.runtime.program.type"; + public static final String CDAP_RUNTIME_PROGRAM = "cdap.runtime.program"; + public static final String CDAP_RUNTIME_RUNID = "cdap.runtime.runid"; + + private static final String GCS_DOC_URL = + "https://cloud.google.com/storage/docs/json_api/v1/status-codes"; + + private final ProvisionerContext provisionerContext; + private final GoogleCredentials credentials; + private final String endpoint; + private final String projectId; + private final String region; + private final String bucket; + private final Map labels; + private final Map provisionerProperties; + private final VersionInfo cdapVersionInfo; + + private volatile Storage storageClient; + private volatile BatchControllerClient batchControllerClient; + + private static final List artifactsCacheablePerCDAPVersion = new ArrayList<>( + Arrays.asList(Constants.Files.TWILL_JAR, Constants.Files.LAUNCHER_JAR) + ); + private static final int SNAPSHOT_EXPIRE_DAYS = 7; + private static final int EXPIRE_DAYS = 730; + + /** + * Constructs a new DataprocServerlessRuntimeJobManager. + */ + public DataprocServerlessRuntimeJobManager(DataprocClusterInfo clusterInfo, + Map provisionerProperties, VersionInfo cdapVersionInfo) { + this.provisionerContext = clusterInfo.getProvisionerContext(); + this.credentials = clusterInfo.getCredentials(); + this.endpoint = clusterInfo.getEndpoint(); + this.projectId = clusterInfo.getProjectId(); + this.region = clusterInfo.getRegion(); + this.bucket = clusterInfo.getBucket(); + this.labels = clusterInfo.getLabels(); + this.provisionerProperties = provisionerProperties; + this.cdapVersionInfo = cdapVersionInfo; + } + + /** + * Retrieves the Storage client, creating it if it doesn't already exist. + */ + @VisibleForTesting + public Storage getStorageClient() { + Storage client = storageClient; + if (client != null) { + return client; + } + + synchronized (this) { + client = storageClient; + if (client != null) { + return client; + } + + int gcsHttpRequestConnectionTimeout = Integer.parseInt(provisionerProperties.getOrDefault( + DataprocUtils.GCS_HTTP_REQUEST_CONNECTION_TIMEOUT_MILLIS, + DataprocUtils.GCS_HTTP_REQUEST_CONNECTION_TIMEOUT_MILLIS_DEFAULT + )); + int gcsHttpRequestReadTimeout = Integer.parseInt(provisionerProperties.getOrDefault( + DataprocUtils.GCS_HTTP_REQUEST_READ_TIMEOUT_MILLIS, + DataprocUtils.GCS_HTTP_REQUEST_READ_TIMEOUT_MILLIS_DEFAULT + )); + int gcsHttpRequestTotalTimeout = Integer.parseInt(provisionerProperties.getOrDefault( + DataprocUtils.GCS_HTTP_REQUEST_TOTAL_TIMEOUT_MINS, + DataprocUtils.GCS_HTTP_REQUEST_TOTAL_TIMEOUT_MINS_DEFAULT + )); + + HttpTransportOptions transportOptions = StorageOptions.getDefaultHttpTransportOptions() + .toBuilder() + .setConnectTimeout(gcsHttpRequestConnectionTimeout) + .setReadTimeout(gcsHttpRequestReadTimeout) + .build(); + + this.storageClient = client = StorageOptions.newBuilder() + .setStorageRetryStrategy(StorageRetryStrategy.getUniformStorageRetryStrategy()) + .setProjectId(projectId) + .setCredentials(credentials) + .setRetrySettings(StorageOptions.getDefaultRetrySettings().toBuilder() + .setTotalTimeout(Duration.ofMinutes(gcsHttpRequestTotalTimeout)).build()) + .setTransportOptions(transportOptions) + .build() + .getService(); + } + return client; + } + + private BatchControllerClient getBatchControllerClient() throws IOException { + BatchControllerClient client = batchControllerClient; + if (client != null) { + return client; + } + + synchronized (this) { + client = batchControllerClient; + if (client != null) { + return client; + } + + CredentialsProvider credentialsProvider = FixedCredentialsProvider.create(credentials); + this.batchControllerClient = client = BatchControllerClient.create( + BatchControllerSettings.newBuilder().setCredentialsProvider(credentialsProvider) + .setEndpoint(String.format("%s-%s", region, endpoint)).build()); + } + return client; + } + + @Override + public void launch(RuntimeJobInfo runtimeJobInfo) throws Exception { + String bucket = DataprocUtils.getBucketName(this.bucket); + ProgramRunInfo runInfo = runtimeJobInfo.getProgramRunInfo(); + String batchId = getBatchId(runInfo); + + LOG.warn("TEST_LOG: Entering DataprocServerlessRuntimeJobManager.launch. runId: {}, batchId: {}, " + + "namespace: {}, app: {}, program: {}", + runInfo.getRun(), batchId, runInfo.getNamespace(), runInfo.getApplication(), runInfo.getProgram()); + + boolean gcsCacheEnabled = Boolean.parseBoolean( + provisionerContext.getProperties().getOrDefault(DataprocUtils.GCS_CACHE_ENABLED, "true")) + || !validateDeleteLifecycle(bucket, runInfo.getRun()); + + LOG.warn("TEST_LOG: Launch parameters - gcsCacheEnabled: {}, projectId: {}, region: {}, bucket: {}", + gcsCacheEnabled, projectId, region, bucket); + + File tempDir = DataprocUtils.CACHE_DIR_PATH.toFile(); + boolean disableLocalCaching = Boolean.parseBoolean( + provisionerContext.getProperties().getOrDefault(DataprocUtils.LOCAL_CACHE_DISABLED, "false")); + + String runRootPath = getPath(DataprocUtils.CDAP_GCS_ROOT, runInfo.getRun()); + String cacheRootPath = getPath(DataprocUtils.CDAP_GCS_ROOT, DataprocUtils.CDAP_CACHED_ARTIFACTS); + + String cdapVersion; + if (cdapVersionInfo.isSnapshot()) { + cdapVersion = String.format("%s.%s.%s-SNAPSHOT", cdapVersionInfo.getMajor(), + cdapVersionInfo.getMinor(), cdapVersionInfo.getFix()); + } else { + cdapVersion = String.format("%s.%s.%s", cdapVersionInfo.getMajor(), + cdapVersionInfo.getMinor(), cdapVersionInfo.getFix()); + } + + LaunchMode launchMode = LaunchMode.valueOf( + provisionerProperties.getOrDefault("launchMode", LaunchMode.CLIENT.name()).toUpperCase()); + + DataprocMetric.Builder submitJobMetric = + DataprocMetric.builder("provisioner.submitJob.response.count") + .setRegion(region) + .setLaunchMode(launchMode); + + try { + if (disableLocalCaching) { + tempDir = Files.createTempDirectory("dataproc.launcher").toFile(); + } + List localFiles = getRuntimeLocalFiles(runtimeJobInfo.getLocalizeFiles(), tempDir); + LOG.warn("TEST_LOG: Prepared local files for upload. Total files count: {}", localFiles.size()); + + List> uploadFutures = new ArrayList<>(); + for (LocalFile fileToUpload : localFiles) { + boolean cacheable = gcsCacheEnabled && fileToUpload instanceof CacheableLocalFile; + String targetFilePath = getPath(cacheable ? cacheRootPath : runRootPath, fileToUpload.getName()); + String targetFilePathWithVersion = getPath(cacheRootPath, cdapVersion, fileToUpload.getName()); + + LOG.warn("TEST_LOG: Scheduling upload for file: {}, cacheable: {}, target path: {}", + fileToUpload.getName(), cacheable, targetFilePath); + + if (gcsCacheEnabled && artifactsCacheablePerCDAPVersion.contains(fileToUpload.getName())) { + uploadFutures.add( + provisionerContext.execute( + () -> uploadCacheableFile(bucket, targetFilePathWithVersion, fileToUpload)) + .toCompletableFuture()); + } else { + if (cacheable) { + uploadFutures.add( + provisionerContext.execute( + () -> uploadCacheableFile(bucket, targetFilePath, fileToUpload)) + .toCompletableFuture()); + } else { + uploadFutures.add(provisionerContext.execute( + () -> uploadFile(bucket, targetFilePath, fileToUpload, false)) + .toCompletableFuture()); + } + } + } + + List uploadedFiles = new ArrayList<>(); + for (Future uploadFuture : uploadFutures) { + uploadedFiles.add(uploadFuture.get()); + } + LOG.warn("TEST_LOG: Completed upload of all files. Total uploaded count: {}", uploadedFiles.size()); + + CreateBatchRequest request = getCreateBatchRequest(runtimeJobInfo, uploadedFiles, launchMode); + LOG.warn("TEST_LOG: Created Batch request. batchId: {}, parent: {}, mainClass: {}", + request.getBatchId(), request.getParent(), request.getBatch().getSparkBatch().getMainClass()); + + try { + LOG.warn("TEST_LOG: Submitting Spark Batch creation to Dataproc Serverless API."); + getBatchControllerClient().createBatchAsync(request); + LOG.warn("TEST_LOG: Submitted successfully. Spark Batch runId: {}", runInfo.getRun()); + } catch (Exception ex) { + LOG.warn("TEST_LOG: Exception while submitting Batch creation to Dataproc: ", ex); + throw ex; + } + DataprocUtils.emitMetric(provisionerContext, submitJobMetric.build()); + } catch (Exception e) { + LOG.warn("TEST_LOG: Exception in launch method: ", e); + String errorReason = String.format("Error while launching serverless job %s.", getBatchId(runInfo)); + DataprocUtils.deleteGcsPath(getStorageClient(), bucket, runRootPath); + DataprocUtils.emitMetric(provisionerContext, submitJobMetric.setException(e).build()); + + ErrorCategory errorCategory = new ErrorCategory(ErrorCategoryEnum.STARTING); + if (e instanceof ApiException) { + int statusCode = ((ApiException) e).getStatusCode().getCode().getHttpStatusCode(); + ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(statusCode); + throw new DataprocRuntimeException.Builder() + .withCause(e) + .withErrorCategory(errorCategory) + .withErrorMessage(e.getMessage()) + .withErrorReason(DataprocUtils.getErrorReason(errorReason, e)) + .withErrorType(pair.getErrorType()) + .withErrorCodeType(ErrorCodeType.HTTP) + .withErrorCode(String.valueOf(statusCode)) + .withDependency(true) + .build(); + } + throw new DataprocRuntimeException.Builder() + .withErrorMessage(e.getMessage()) + .withErrorReason(errorReason) + .withErrorCategory(errorCategory) + .withCause(e) + .build(); + } finally { + if (disableLocalCaching) { + DataprocUtils.deleteDirectoryContents(tempDir); + } + } + } + + @Override + public Optional getDetail(ProgramRunInfo programRunInfo) throws Exception { + String batchId = getBatchId(programRunInfo); + LOG.warn("TEST_LOG: Entering DataprocServerlessRuntimeJobManager.getDetail for runId: {}, batchId: {}", + programRunInfo.getRun(), batchId); + try { + Batch batch = getBatchControllerClient().getBatch(GetBatchRequest.newBuilder() + .setName(String.format("projects/%s/locations/%s/batches/%s", projectId, region, batchId)) + .build()); + RuntimeJobStatus jobStatus = getJobStatus(batch); + String statusDetails = getJobStatusDetails(batch); + LOG.warn("TEST_LOG: getDetail status response - batchId: {}, status: {}, details: {}", + batchId, jobStatus, statusDetails); + return Optional.of(new DataprocRuntimeJobDetail( + getProgramRunInfo(batch), + jobStatus, + statusDetails, + batchId)); + } catch (ApiException e) { + LOG.warn("TEST_LOG: ApiException in getDetail for batchId: {}, statusCode: {}", + batchId, e.getStatusCode().getCode()); + if (e.getStatusCode().getCode() != StatusCode.Code.NOT_FOUND) { + int code = e.getStatusCode().getCode().getHttpStatusCode(); + ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(code); + String errorReason = String.format("%s Unable to get details for serverless batch %s. %s", + code, batchId, pair.getCorrectiveAction()); + throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategoryEnum.OTHERS), + errorReason, e.getMessage(), pair.getErrorType(), true, ErrorCodeType.HTTP, + String.valueOf(code), null, e); + } + LOG.debug("Dataproc serverless batch {} does not exist.", batchId); + } + return Optional.empty(); + } + + @Override + public void stop(ProgramRunInfo programRunInfo) throws Exception { + String batchId = getBatchId(programRunInfo); + LOG.warn("TEST_LOG: Entering DataprocServerlessRuntimeJobManager.stop for runId: {}, batchId: {}", + programRunInfo.getRun(), batchId); + try { + getBatchControllerClient().deleteBatch(DeleteBatchRequest.newBuilder() + .setName(String.format("projects/%s/locations/%s/batches/%s", projectId, region, batchId)) + .build()); + LOG.warn("TEST_LOG: Successfully requested deletion of serverless batch: {}", batchId); + } catch (ApiException e) { + LOG.warn("TEST_LOG: ApiException in stop for batchId: {}, statusCode: {}", batchId, e.getStatusCode().getCode()); + if (e.getStatusCode().getCode() != StatusCode.Code.FAILED_PRECONDITION) { + throw new Exception(String.format("Error occurred while stopping serverless batch %s.", batchId), e); + } + LOG.debug("Serverless batch {} is already deleted or stopping.", batchId); + } + } + + @Override + public void kill(RuntimeJobDetail jobDetail) throws Exception { + if (jobDetail != null) { + stop(jobDetail.getRunInfo()); + } + } + + @Override + public void close() { + BatchControllerClient client = this.batchControllerClient; + if (client != null) { + client.close(); + } + } + + private List getRuntimeLocalFiles(Collection runtimeLocalFiles, + File tempDir) throws Exception { + LocationFactory locationFactory = new LocalLocationFactory(tempDir); + List localFiles = new ArrayList<>(runtimeLocalFiles); + localFiles.add(getTwillJar(locationFactory)); + localFiles.add(getLauncherJar(locationFactory)); + localFiles.sort(Comparator.comparingLong(LocalFile::getSize).reversed()); + return localFiles; + } + + private LocalFile getTwillJar(LocationFactory locationFactory) throws IOException { + Location location = locationFactory.create(Constants.Files.TWILL_JAR); + if (location.exists()) { + return DataprocJarUtil.getLocalFile(location, true); + } + return DataprocJarUtil.getTwillJar(locationFactory); + } + + private LocalFile getLauncherJar(LocationFactory locationFactory) throws IOException { + Location location = locationFactory.create(Constants.Files.LAUNCHER_JAR); + if (location.exists()) { + return DataprocJarUtil.getLocalFile(location, false); + } + return DataprocJarUtil.getLauncherJar(locationFactory); + } + + private boolean validateDeleteLifecycle(String bucketName, String run) { + Storage storage = getStorageClient(); + Bucket bucket = storage.get(bucketName); + for (BucketInfo.LifecycleRule rule : bucket.getLifecycleRules()) { + if (rule.getAction() == null || rule.getCondition() == null + || rule.getCondition().getDaysSinceCustomTime() == null) { + continue; + } + if (rule.getAction() instanceof BucketInfo.LifecycleRule.DeleteLifecycleAction + && rule.getCondition().getDaysSinceCustomTime() > 0) { + if (!provisionerContext.getProperties() + .containsKey(DataprocUtils.ARTIFACTS_COMPUTE_HASH_TIME_BUCKET_DAYS)) { + return true; + } + try { + int timeBucketDays = Integer.parseInt( + provisionerContext.getProperties() + .get(DataprocUtils.ARTIFACTS_COMPUTE_HASH_TIME_BUCKET_DAYS)); + return rule.getCondition().getDaysSinceCustomTime() > timeBucketDays; + } catch (NumberFormatException e) { + return false; + } + } + } + return false; + } + + private LocalFile uploadCacheableFile(String bucket, String targetFilePath, LocalFile localFile) throws IOException { + Storage storage = getStorageClient(); + BlobId blobId = BlobId.of(bucket, targetFilePath); + Blob blob = storage.get(blobId); + LocalFile result; + + if (blob != null && blob.exists()) { + if (artifactsCacheablePerCDAPVersion.contains(localFile.getName()) + && (blob.getUpdateTime() < cdapVersionInfo.getBuildTime())) { + BlobInfo newBlobInfo = blob.toBuilder().setCustomTime(getCustomTime()).build(); + try { + uploadToGcsUtil(localFile, storage, targetFilePath, newBlobInfo, + Storage.BlobWriteOption.generationMatch(), + Storage.BlobWriteOption.metagenerationMatch()); + } catch (StorageException e) { + if (e.getCode() != HttpURLConnection.HTTP_PRECON_FAILED) { + ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(e.getCode()); + String errorReason = String.format("%s Unable to upload file %s to GCS bucket gs://%s. %s", + e.getCode(), localFile.getURI(), bucket, pair.getCorrectiveAction()); + throw ErrorUtils.getProgramFailureException( + new ErrorCategory(ErrorCategoryEnum.STARTING), errorReason, e.getMessage(), + pair.getErrorType(), true, ErrorCodeType.HTTP, String.valueOf(e.getCode()), + GCS_DOC_URL, e); + } + } + } + result = new DefaultLocalFile(localFile.getName(), + URI.create(String.format("gs://%s/%s", bucket, targetFilePath)), + localFile.getLastModified(), localFile.getSize(), + localFile.isArchive(), localFile.getPattern()); + } else { + result = uploadFile(bucket, targetFilePath, localFile, true); + } + return result; + } + + /** + * Uploads a file to a GCS bucket, optionally caching it. + */ + public LocalFile uploadFile(String bucket, String targetFilePath, LocalFile localFile, + boolean cacheable) throws IOException { + BlobId blobId = BlobId.of(bucket, targetFilePath); + String contentType = "application/octet-stream"; + BlobInfo.Builder blobInfoBuilder = BlobInfo.newBuilder(blobId); + if (cacheable) { + long customTime = System.currentTimeMillis(); + if (artifactsCacheablePerCDAPVersion.contains(localFile.getName())) { + customTime = getCustomTime(); + } + blobInfoBuilder.setCustomTime(customTime); + } + BlobInfo blobInfo = blobInfoBuilder.setContentType(contentType).build(); + Storage storage = getStorageClient(); + Bucket bucketObj = storage.get(bucket); + + if (bucketObj == null) { + String error = String.format("GCS Bucket '%s' does not exist", bucket); + throw new ProgramFailureException.Builder() + .withErrorCategory(new ErrorCategory(ErrorCategoryEnum.STARTING)) + .withErrorReason(error) + .withErrorMessage(error) + .withErrorType(ErrorType.USER) + .build(); + } + + try { + uploadToGcsUtil(localFile, storage, targetFilePath, blobInfo, Storage.BlobWriteOption.doesNotExist()); + } catch (StorageException e) { + if (e.getCode() != HttpURLConnection.HTTP_PRECON_FAILED) { + ActionErrorPair pair = ErrorUtils.getActionErrorByStatusCode(e.getCode()); + String errorReason = String.format("%s Unable to upload file %s to GCS bucket gs://%s. %s", + e.getCode(), localFile.getURI(), bucket, pair.getCorrectiveAction()); + throw ErrorUtils.getProgramFailureException(new ErrorCategory(ErrorCategoryEnum.STARTING), + errorReason, e.getMessage(), pair.getErrorType(), true, + ErrorCodeType.HTTP, String.valueOf(e.getCode()), GCS_DOC_URL, e); + } + if (!cacheable) { + Blob existingBlob = storage.get(blobId); + BlobInfo newBlobInfo = BlobInfo.newBuilder(existingBlob.getBlobId()).setContentType(contentType).build(); + uploadToGcsUtil(localFile, storage, targetFilePath, newBlobInfo); + } + } + + return new DefaultLocalFile(localFile.getName(), + URI.create(String.format("gs://%s/%s", bucket, targetFilePath)), + localFile.getLastModified(), localFile.getSize(), + localFile.isArchive(), localFile.getPattern()); + } + + private long getCustomTime() { + return cdapVersionInfo.getBuildTime() + + TimeUnit.DAYS.toMillis(cdapVersionInfo.isSnapshot() ? SNAPSHOT_EXPIRE_DAYS : EXPIRE_DAYS); + } + + /** + * Helper utility method to copy local files directly into GCS. + */ + public void uploadToGcsUtil(LocalFile localFile, Storage storage, String targetFilePath, BlobInfo blobInfo, + Storage.BlobWriteOption... blobWriteOptions) throws IOException { + try (InputStream inputStream = openStream(localFile.getURI()); + WriteChannel writer = storage.writer(blobInfo, blobWriteOptions)) { + ByteStreams.copy(inputStream, Channels.newOutputStream(writer)); + } + } + + private InputStream openStream(URI uri) throws IOException { + if ("file".equals(uri.getScheme())) { + return Files.newInputStream(new File(uri).toPath()); + } + LocationFactory locationFactory = provisionerContext.getLocationFactory(); + if (locationFactory.getHomeLocation().toURI().getScheme().equals(uri.getScheme())) { + return locationFactory.create(uri).getInputStream(); + } + if ("gs".equals(uri.getScheme())) { + BlobId blobId = BlobId.of(uri.getAuthority(), uri.getPath().substring(1)); + Storage client = StorageOptions.getDefaultInstance().getService(); + return Channels.newInputStream(client.get(blobId).reader()); + } + return uri.toURL().openStream(); + } + + private CreateBatchRequest getCreateBatchRequest(RuntimeJobInfo runtimeJobInfo, + List localFiles, + LaunchMode launchMode) throws IOException { + List jarUris = new ArrayList<>(); + List fileUris = new ArrayList<>(); + for (LocalFile localFile : localFiles) { + if (localFile.getName().endsWith("jar")) { + jarUris.add(localFile.getURI().toString()); + } else { + fileUris.add(localFile.getURI().toString()); + } + } + + Map sparkProperties = new LinkedHashMap<>(); + sparkProperties.putAll(getProperties(runtimeJobInfo)); + // Prepend CDAP launcher jar to the classpath settings for Spark + sparkProperties.put("spark.driver.extraClassPath", "./" + Constants.Files.LAUNCHER_JAR); + sparkProperties.put("spark.executor.extraClassPath", "./" + Constants.Files.LAUNCHER_JAR); + + String applicationJarLocalizedName = runtimeJobInfo.getArguments().get(Constants.Files.APPLICATION_JAR); + SparkBatch sparkBatch = SparkBatch.newBuilder() + .setMainClass(DataprocJobMain.class.getName()) + .addAllArgs(DataprocRuntimeJobManager.getArguments( + runtimeJobInfo, localFiles, provisionerContext.getSparkCompat().getCompat(), + applicationJarLocalizedName, launchMode)) + .addAllJarFileUris(jarUris) + .addAllFileUris(fileUris) + .build(); + + RuntimeConfig runtimeConfig = RuntimeConfig.newBuilder() + .putAllProperties(sparkProperties) + .build(); + + Batch.Builder batchBuilder = Batch.newBuilder() + .setSparkBatch(sparkBatch) + .setRuntimeConfig(runtimeConfig) + .putAllLabels(labels); + + ExecutionConfig.Builder execConfigBuilder = ExecutionConfig.newBuilder(); + String serviceAccount = provisionerProperties.get("serviceAccount"); + String subnet = provisionerProperties.get("subnet"); + + if (!Strings.isNullOrEmpty(serviceAccount)) { + execConfigBuilder.setServiceAccount(serviceAccount); + } + if (!Strings.isNullOrEmpty(subnet)) { + execConfigBuilder.setSubnetworkUri(subnet); + } + + batchBuilder.setEnvironmentConfig(EnvironmentConfig.newBuilder() + .setExecutionConfig(execConfigBuilder.build()) + .build()); + + return CreateBatchRequest.newBuilder() + .setParent(String.format("projects/%s/locations/%s", projectId, region)) + .setBatchId(getBatchId(runtimeJobInfo.getProgramRunInfo())) + .setBatch(batchBuilder.build()) + .build(); + } + + private Map getProperties(RuntimeJobInfo runtimeJobInfo) { + ProgramRunInfo runInfo = runtimeJobInfo.getProgramRunInfo(); + Map properties = new LinkedHashMap<>(); + properties.put(CDAP_RUNTIME_NAMESPACE, runInfo.getNamespace()); + properties.put(CDAP_RUNTIME_APPLICATION, runInfo.getApplication()); + properties.put(CDAP_RUNTIME_VERSION, runInfo.getVersion()); + properties.put(CDAP_RUNTIME_PROGRAM, runInfo.getProgram()); + properties.put(CDAP_RUNTIME_PROGRAM_TYPE, runInfo.getProgramType()); + properties.put(CDAP_RUNTIME_RUNID, runInfo.getRun()); + return properties; + } + + private ProgramRunInfo getProgramRunInfo(Batch batch) { + Map properties = batch.getRuntimeConfig().getPropertiesMap(); + return new ProgramRunInfo.Builder() + .setNamespace(properties.get(CDAP_RUNTIME_NAMESPACE)) + .setApplication(properties.get(CDAP_RUNTIME_APPLICATION)) + .setVersion(properties.get(CDAP_RUNTIME_VERSION)) + .setProgramType(properties.get(CDAP_RUNTIME_PROGRAM_TYPE)) + .setProgram(properties.get(CDAP_RUNTIME_PROGRAM)) + .setRun(properties.get(CDAP_RUNTIME_RUNID)) + .build(); + } + + private RuntimeJobStatus getJobStatus(Batch batch) { + Batch.State state = batch.getState(); + switch (state) { + case STATE_UNSPECIFIED: + case PENDING: + return RuntimeJobStatus.STARTING; + case RUNNING: + return RuntimeJobStatus.RUNNING; + case CANCELLING: + return RuntimeJobStatus.STOPPING; + case CANCELLED: + return RuntimeJobStatus.STOPPED; + case SUCCEEDED: + return RuntimeJobStatus.COMPLETED; + case FAILED: + return RuntimeJobStatus.FAILED; + default: + throw new IllegalStateException( + String.format("Unsupported state %s of serverless batch %s.", state, batch.getName())); + } + } + + @Nullable + private String getJobStatusDetails(Batch batch) { + return batch.getStateMessage(); + } + + private String getPath(String... pathSubComponents) { + return Joiner.on("/").join(pathSubComponents); + } + + public static String getBatchId(ProgramRunInfo runInfo) { + // Generate a valid batch id (starts with letter, lowercase letters, numbers and hyphens only, max 63 chars) + return "cdap-" + runInfo.getRun(); + } +} diff --git a/cdap-runtime-ext-dataproc/src/main/resources/META-INF/services/io.cdap.cdap.runtime.spi.provisioner.Provisioner b/cdap-runtime-ext-dataproc/src/main/resources/META-INF/services/io.cdap.cdap.runtime.spi.provisioner.Provisioner index 581186e1f676..5f74f51cab76 100644 --- a/cdap-runtime-ext-dataproc/src/main/resources/META-INF/services/io.cdap.cdap.runtime.spi.provisioner.Provisioner +++ b/cdap-runtime-ext-dataproc/src/main/resources/META-INF/services/io.cdap.cdap.runtime.spi.provisioner.Provisioner @@ -16,3 +16,5 @@ io.cdap.cdap.runtime.spi.provisioner.dataproc.DataprocProvisioner io.cdap.cdap.runtime.spi.provisioner.dataproc.ExistingDataprocProvisioner +io.cdap.cdap.runtime.spi.provisioner.dataproc.DataprocServerlessProvisioner + diff --git a/cdap-runtime-ext-dataproc/src/main/resources/gcp-dataproc-serverless.json b/cdap-runtime-ext-dataproc/src/main/resources/gcp-dataproc-serverless.json new file mode 100644 index 000000000000..d1e880c7c122 --- /dev/null +++ b/cdap-runtime-ext-dataproc/src/main/resources/gcp-dataproc-serverless.json @@ -0,0 +1,135 @@ +{ + "icon": { + "type": "inline", + "arguments": { + "data": "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAFUAAABVCAYAAAA49akaAAAOU0lEQVR4Xu2deXRU1R3Hv3febEkmewiEnUAQRaniUtuq1SyEsBSxDepxgR4Rat2ONSBJEAeSSdh6WihuqK3K0SqporIJWVuxQkux0oJKQAsBIfskmSSzvttzJyaTyczkvTfvTUg43L9g3u/+7u9+3u/d+7trCC4nxQkQxTVeVojLUEPgBJehDimolJJUk3Usp9EP43nHoHl5KpWGuhzW+oqVYadDwNOtUvHKGo1UdUDPr6aUPAXQiFAZLl8vaSeE/u4Wq+o5o5Hw8vV5NCgK9XYjNXB6vhoUI5Q0MqS6CC64rKqUKiOxKFWOYlCzjVTbpHN9ByBeKeMGUE9jnI0bWWIkdiXKVAxperFzOwWylTDqYuggQElZrnqBEmUrAnXGBjrB5XCdAlG+jVaikqJ0UFBOw03cv4x8K0q+HyFFoKYVO04AJEWuMRc/PzlRnstdIdcO2VBT1zqzCKV75BoyWPJTQmZVrFDvlWOPLKjZ2ZRrmu5sBBAtx4hBlrcl7og6vqSEuIK1SxbUtGLHGgDPBlv4IM5XUJ6rWRWsfUFDzTI2Rtk1kU0ghAu28EGbj1KX1tEWt9cY3xqMjUFDTSuyVwC4I5hCh0ieyvI8bWowtgYFNa3IdhVAjgVT4NDKQ6eW5+mOS7U5SKiOGgI6WmphQ02egpwtz9OMkWq3ZKgZxbZsSsl2qQUNVXlC6ILSXF2JFPslQWXj+2atvQnAIJ59klJ9UbLtsXZtnJR5AUlQM4rsmyjoE6JMuYSECMjm0jztk2KrJBrq3I00odNurwWgEqv8EpLjw7Ta4TtzSIOYOomGmm6yfQaCm8UoVUpGrwGmjVUh3gDYncB5M8Xxc1Qp9dL0UBwsy9f9SEwmUVAz19unu5z8v8QoVEJmVBzBmp9rMCbBd9rL6QIqj7vw2z1OuBSdrxe2nFOrrt+3XHtESFIU1PSizlqAJAopU+L5/Bs5PJKmBhGwzOoAHtpqR13rQHourSvLCxsuVE9BqBmmzsWUkFeEFCnxPHMah5zZatGqHE4ge7MV7TbBaojWKSRIKH24ND/s1f7k+rUmazPVOSy2FgA6ocLkPg/TAjue0oHz0w2euuBEuI4gKdZ3muFMgxMPbXUAZMD6T5vGoIve+wSxBapzv1DTiju3EYr75QITkz9ntgbMU3un9w6248V9rWDtKEtR4Sq8vDTBB+5DLzTgtDkcZODAbivLC3tQMtRMU3uSC6pzA7VEsitHD53GY+bfjlux6p1mH7s1HMGuvOHQaTz+8PHnnVi7wwy1PnpgwLKlF/Cj9uVHnPcHNqCnphd1fglgihgvU0Jm/wq9V+c0t7gWbZ3+u/eFtxvwy9TInmKbLTzmr2chNAYOLPBVWV7YlaKhpq913ALe+YkSsMTo0HDAnuX6HlFKgYw153s++746bpqkw/oH43p+7rBRzDJd6Pn/gHmsSn1r2QrNgb72+XoqpSS9uKMJIDFigCglU5ob5qUqq/A8Ou3+w6V7fmLArzKjeuRbOnjMW+uBOnAeS81lueFxIMTLUB+o6UUdOQA2KAVLrJ6+UHcd7sDGj8w+2VUE2JmbhAi9x/TK/3Zi9Xbf9neAPHZZWV74xt6GekG93VhnUGsMzDrxwaJYagJypXnensrEX9rXincOeHbjqDmC5x9OwBWjevVoAB7d2oBjNf43lwwAWKfTYYmtMib2GOoFNd3UvgsgsxXiJEmNP6hMgdVOcfSMHQYdweRRWqj7hKP1rS5kb/D+9L0LJlDro0K8lEZ3l+VHzOkutwdqerE5GbzmZCh2AoqhW5oXLkbMS4anwIINF9DQJreyaHGKPpVA5JpXlxnzDDPRANXWcBOhEyTVTKINUqDwPLHmxDicvOERbEFqw5FRZfvikHqhpBZaZREVk7coQXbMAgr2hspDKYuURGRZ46PnSvha8e8ACqdMpoQRLeZpV/qzhY5K9nXLN1R2smzXIBSMnf2+ozAvTjOcweaQW6x6IR0yEf7gslFr2egNOnBfvrSEOtyyxKeExJL2o9WFQ1VY5QJTIW5rnWfZiUFOfO+tWy6YAZ14XgZx5sX4nW5gM6/lXbGsIOALzZ1/IPJbwS0i6yXIGgORlWCVA9tYRCGq3jJoDnpwTiznXR/ida2Wd1o6DFjy/1wz2bzEpRGBrSHqhxQmCi751RwhqN6Q4A4e1DyRg8kjvWLX7ud1JsW5HMyqOdohqbxUHS+Ei6aY2ke9VzLsPXqY0z9Okd33+NQGVsZDlmnE6mO5LCNiZ1be48PTr9TjT0H97yxZsGFjBpQYJVSMZhW2DwlP394Ga1g/UniCbAL/4USSWZsYEbG8PnbBi9bsNAecRepoXpaYNmadmuNtUOiBtql5DcNMkDuOHqdwBck0Tj4PVLrBZpmCgetpbgmez43HbVN+hLpNhIdqOQ214fk//7a0yTQGpcff+hJKQ9v4x4QTF94Zh4nD/odHpBh7jEjzP3CGVCE/thjpllBZbliSCY7Mt/aTWDh6L/nABzZbAIzC5YCmhS9xxqrm6LWRx6k0T1VizIAwC9fVC0QWVBSX9p3CtCvnZ8fjxFP8e6i8368juWncO7dZA69tdbSwRWs71b5olJiUyxv1q0wpaZqpCMKKaMpLDpkXhko+sCEFVqYD7b4vGolRWef+1Y/DYI43aV+BMvQMLN/tdCelRFozH8u4RVfTHPSXOMLWepIBiY3+m+MNlBrB2tHdyuihOnneApxQpI7Vga059U39Qb5ioR+F9w7zWqHrnZ8CefK0W5nbeDXVUvBqbFw9HrME7anzslVocOxNwQdStUgpYApzanx/lGfszBWyWivAqxWap0q7W4JmfeZZIWBmffd2JlW/Xg0HrTs/Mj8PM6d4jZH9Qh0VzWP9gIsYn+o9Pmb4G9zTgOZ+XxLx1Z95orxfx+bdW/OaPdUItjFiwlKp431kqpj3D1LoLoIrMp77xiAFJsZ7Oh8WLCzexU5a+6cWlSZgyWtvzoAtq1yFnrZogZ148Mq4V3r257I06HD7Z6beMu26OxOOzPetabMJmrilwLOxRIqaNJbtL86N851OZEjbzr9HoFJn5/zAnEmFabric6e9/M06/LPaf4XZxMnLjyT11IOFQJmrz2DujQY8NostAQk6lFvgjmcDnzZnI7H3nvFs/mbrX7MKhDvD7pL7aQqcDoct8Mw/UzCj0JxDQWSvUe1ZEQU2Xu9OdxbXoKXdf4/LmFUUjvMix3rnCL3/EKzW7ET+W/V45ddJXsDv3nAWdS3+w6Vp4/XYtNizDcoNdY14qIHaWAK6bP/KmMBrVO5aUUoyTK1st7Ss1dQPl3l7at62Oneb6i9dOUaHF5YKn2ZnPfqGHY0o+6Ld3QntXz0WbN2qO+0+bMHGD9hZOd+0ZekITB3j2b3EXsw9G33bX6Fvoo/Hmkvzo4RXU5nS9MLWWwh4Wev+ry2NxJheAX1dS1clKPu2+6RXHx2JiUmeNrXvczbr9MHBVmzZ0+QeHXWnp+fFY86Nnk0V7PeC7fWoONrupeLeW6OxJDPW67cX9jah5NNgjkl52lgK1a1lK6NErPt/X3RGoVnWDpWfXqVB/nzvdadDJzrx3J/rYHN0kWHh1MoFCbhtauBO6MuzNqx4sxZsNNQ3seZhZ/5Ynzb3dJ0Dlf9pd3tx5nURGBbtvTjscLFP/3TAzRpC3uq2XRf1VemqePE7VFgmtpeK5x2y9lK9/3QUDL3W57uN/V+dAy6eInmENuDAwNzucsP8+lz/9xosSo3BwlRpLdWad+vd0GUk6nBwoz9dP9ZvONNvv5pR0LwNhAS9629CIoeXFhtE996sksyLtuxuwkf/aBNd5+V3JSCrT6wbKPOblWb8qdx3k4bowtyC5K0q0/iAXAT3pzpbW2TtT71uvBqmeyN81uv9VaLsCwvWvd8Q1GeZ/oMI5NyZEHCkxaKJVW/X4cg3Vmn8fKVt0CCmyjghoCLBCDDDZF4MSmXtpDbogYLsMLBevu/ECut4vjprQ/F79agRmFAWosHi2R+mhOOOayIwJqFr5HXqgh3lRy3497eyYbr1UYIlfy1M7peHIFSmKKOwme1TlLfnn1IQZyuunaDD+EStu0lgEA+f6oT9+45LCNrFfk4I6ioLk+Xv+e/qtMzTeUrln06hFA4ra00GxQqO5HdEVNwNlQXjBDmI8tQub236DFDgHNVQBUvIoarCZFHnyERDnbuxNcFmddZSJU78DTGwFITnedeIT4pT6sW4t2io37etm6DU2dQhBZZsqTIlPy4GqDvgEivI5LKNx7RmdZJyp6iHBFjSPkzTGVdinCr6djVJULu8tTEbUPC8/yAHS1Tk7sqCZEn3G0iGysDOKGisoQTK3UwxWMESnK0qnCR5+T4oqFlFjVe5eCh7h8pgBKuiV1cWpEiuZ1BQ3d5a2FABEGVv+xlUYGlVpSklqPoFDZXdS+XiaJPim9sGA1hKXHotjdtrTAlmwlVa7983UsgsqF9DQZS/Qe0igyWgpoqiySulREa9ZYP2VHeIlb2da5mWGpq7/i4aWNKSUH0kvqRkgfDpjADUZUF1t62mpizwfGhupbwoYOmcyqLJu4P1UsnBf6CCZhTUnwAQmvtTBxAsAamuKEqZLAeoglDNE0Adobvpd2DAUo7HpLK1k91noeQk2Z9/d+GZBQ3bKWjo7qQOMVgC8peKohRF7FcMKrtdrYWr/w4gobs9PURgeaApUetIkjK+78+TFYPKCmHbhnQcqaagwjsjgv2+FAdLaqk2ZlLvA7vBmtadT1GoTCn7ixR/VzWuhop/CjREdwIqAJYH7VAR8vuEE1+skhM++XsBikPtXUhq4XfjOGiGcbxT8XLYmRNnh/QBD+HUVAVSX2FKHjp/O0Xup3Mp5Ffcgy4FKHLrcBmqXIJ+8l+GGgKo/wfsUyhPEm1t6gAAAABJRU5ErkJggg==" + } + }, + "configuration-groups": [ + { + "label": "Cluster Information", + "properties": [ + { + "widget-type": "textbox", + "label": "Project ID", + "name": "projectId", + "description": "Google Cloud Project ID, which uniquely identifies your project. You can find it on the Dashboard in the Cloud Platform Console. If the system is running on Google Cloud Platform, this can be left blank or set to 'auto-detect' and the system's project id will be used.", + "widget-attributes": { + "placeholder": "Specify your GCP Project ID" + } + }, + { + "widget-type": "select", + "label": "Region", + "name": "region", + "description": "The region where the Dataproc Serverless batch job will be executed.", + "widget-attributes": { + "values": [ + "auto-detect", + "africa-south1", + "asia-east1", + "asia-east2", + "asia-northeast1", + "asia-northeast2", + "asia-northeast3", + "asia-south1", + "asia-south2", + "asia-southeast1", + "asia-southeast2", + "australia-southeast1", + "australia-southeast2", + "europe-central2", + "europe-north1", + "europe-north2", + "europe-southwest1", + "europe-west1", + "europe-west10", + "europe-west12", + "europe-west2", + "europe-west3", + "europe-west4", + "europe-west6", + "europe-west8", + "europe-west9", + "me-central1", + "me-central2", + "me-west1", + "northamerica-northeast1", + "northamerica-northeast2", + "northamerica-south1", + "southamerica-east1", + "southamerica-west1", + "us-central1", + "us-central2", + "us-east1", + "us-east4", + "us-east5", + "us-south1", + "us-west1", + "us-west2", + "us-west3", + "us-west4", + "us-west8" + ], + "default": "us-east1", + "size": "medium" + } + }, + { + "widget-type": "securekey-textarea", + "label": "Service Account Key", + "name": "accountKey", + "description": "A service account key used to authenticate with GCP. Paste the contents of the service account key JSON file. If the system is running on Google Cloud Platform, this can be left blank or set to 'auto-detect'.", + "widget-attributes": { + "placeholder": "Specify the GCP service account credentials" + } + }, + { + "widget-type": "textbox", + "label": "Execution Service Account Email", + "name": "serviceAccount", + "description": "The GCP Service Account email that will run the Dataproc Serverless batch workload.", + "widget-attributes": { + "placeholder": "e.g. my-sa-email@my-project.iam.gserviceaccount.com" + } + }, + { + "widget-type": "textbox", + "label": "Subnetwork URI", + "name": "subnet", + "description": "The subnetwork to run Dataproc Serverless compute workloads inside. Required if the default network is not used or Private Google Access is enabled.", + "widget-attributes": { + "placeholder": "e.g. projects/project-id/regions/region-name/subnetworks/subnet-name" + } + }, + { + "widget-type": "textbox", + "label": "Staging Bucket", + "name": "gcsBucket", + "description": "Google Cloud Storage bucket used to stage job dependencies and config files for running serverless pipelines.", + "widget-attributes": { + "size": "medium" + } + } + ] + }, + { + "label": "Advanced", + "properties": [ + { + "widget-type": "keyvalue", + "label": "Common Labels", + "name": "labels", + "description": "Specifies labels for Dataproc Serverless batch jobs.", + "widget-attributes": { + "showDelimiter": "false", + "delimiter": ";", + "kv-delimiter": "|", + "key-placeholder": "Key", + "value-placeholder": "Value" + } + } + ] + } + ] +} diff --git a/cdap-runtime-ext-dataproc/src/test/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocServerlessRuntimeJobManagerTest.java b/cdap-runtime-ext-dataproc/src/test/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocServerlessRuntimeJobManagerTest.java new file mode 100644 index 000000000000..0d29efeec795 --- /dev/null +++ b/cdap-runtime-ext-dataproc/src/test/java/io/cdap/cdap/runtime/spi/provisioner/dataproc/DataprocServerlessRuntimeJobManagerTest.java @@ -0,0 +1,133 @@ +/* + * Copyright © 2026 Cask Data, Inc. + * + * 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 io.cdap.cdap.runtime.spi.provisioner.dataproc; + +import com.google.cloud.dataproc.v1.Batch; +import com.google.cloud.dataproc.v1.RuntimeConfig; +import com.google.common.collect.ImmutableMap; +import io.cdap.cdap.runtime.spi.ProgramRunInfo; +import io.cdap.cdap.runtime.spi.runtimejob.DataprocServerlessRuntimeJobManager; +import io.cdap.cdap.runtime.spi.runtimejob.RuntimeJobInfo; +import java.util.Collection; +import java.util.Collections; +import java.util.Map; +import java.util.UUID; +import java.util.regex.Pattern; +import org.apache.twill.api.LocalFile; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +/** + * Tests for DataprocServerlessRuntimeJobManager. + */ +public class DataprocServerlessRuntimeJobManagerTest { + + private static final Pattern BATCH_ID_PATTERN = Pattern.compile("^[a-z][a-z0-9-]{3,62}$"); + + private static RuntimeJobInfo runtimeJobInfo; + + @BeforeClass + public static void setUp() { + runtimeJobInfo = + new RuntimeJobInfo() { + private final ProgramRunInfo runInfo = + new ProgramRunInfo.Builder() + .setNamespace("namespace") + .setApplication("application") + .setVersion("1.0") + .setProgramType("workflow") + .setProgram("program") + .setRun(UUID.randomUUID().toString()) + .build(); + + @Override + public Collection getLocalizeFiles() { + return Collections.emptyList(); + } + + @Override + public String getRuntimeJobClassname() { + return "io.cdap.cdap.runtime.spi.runtimejob.DataprocJobMain"; + } + + @Override + public ProgramRunInfo getProgramRunInfo() { + return runInfo; + } + + @Override + public Map getJvmProperties() { + return ImmutableMap.of("key", "val"); + } + }; + } + + @Test + public void getBatchIdValidationTest() { + ProgramRunInfo runInfo = + new ProgramRunInfo.Builder() + .setNamespace("NAMESPACE-Upper") + .setApplication("AppWith_Underscore$$$") + .setVersion("1.0") + .setProgramType("workflow") + .setProgram("program") + .setRun(UUID.randomUUID().toString()) + .build(); + + String batchId = DataprocServerlessRuntimeJobManager.getBatchId(runInfo); + + // Batch ID must match Dataproc rules: starts with a letter, max 63 chars, only a-z, 0-9 and hyphens. + Assert.assertTrue("Batch ID '" + batchId + "' is invalid.", BATCH_ID_PATTERN.matcher(batchId).matches()); + Assert.assertTrue(batchId.startsWith("cdap-")); + Assert.assertEquals(41, batchId.length()); // "cdap-" (5) + UUID (36) + } + + @Test + public void getPropertiesTest() throws Exception { + ProgramRunInfo runInfo = runtimeJobInfo.getProgramRunInfo(); + // Simulate mapping in RuntimeJobManager + // Set properties + Batch batch = Batch.newBuilder() + .setRuntimeConfig(RuntimeConfig.newBuilder() + .putAllProperties(ImmutableMap.of( + DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_NAMESPACE, runInfo.getNamespace(), + DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_APPLICATION, runInfo.getApplication(), + DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_VERSION, runInfo.getVersion(), + DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_PROGRAM_TYPE, runInfo.getProgramType(), + DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_PROGRAM, runInfo.getProgram(), + DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_RUNID, runInfo.getRun() + )).build()) + .build(); + + // Map properties from RuntimeConfig + Map properties = batch.getRuntimeConfig().getPropertiesMap(); + + Assert.assertEquals(runInfo.getNamespace(), + properties.get(DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_NAMESPACE)); + Assert.assertEquals(runInfo.getApplication(), + properties.get(DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_APPLICATION)); + Assert.assertEquals(runInfo.getVersion(), + properties.get(DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_VERSION)); + Assert.assertEquals(runInfo.getProgramType(), + properties.get(DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_PROGRAM_TYPE)); + Assert.assertEquals(runInfo.getProgram(), + properties.get(DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_PROGRAM)); + Assert.assertEquals(runInfo.getRun(), + properties.get(DataprocServerlessRuntimeJobManager.CDAP_RUNTIME_RUNID)); + } +}