You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by al...@apache.org on 2019/02/14 00:55:20 UTC
[beam] branch master updated: [BEAM-6664] Temporarily convert
dataflowKmsKey flag to experimental flag.
This is an automated email from the ASF dual-hosted git repository.
altay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 7b45e1b [BEAM-6664] Temporarily convert dataflowKmsKey flag to experimental flag.
new 61eb4ed Merge pull request #7835 from udim/cmek-df
7b45e1b is described below
commit 7b45e1b5df96b1e3a37ab5dc77eb11cdefaa9064
Author: Udi Meiri <eh...@google.com>
AuthorDate: Wed Feb 13 14:12:29 2019 -0800
[BEAM-6664] Temporarily convert dataflowKmsKey flag to experimental flag.
---
runners/google-cloud-dataflow-java/build.gradle | 2 +-
.../beam/runners/dataflow/DataflowPipelineTranslator.java | 6 ++++++
.../org/apache/beam/sdk/options/ExperimentalOptions.java | 13 +++++++++++++
sdks/python/apache_beam/io/gcp/gcsio_integration_test.py | 8 +++++++-
sdks/python/apache_beam/options/pipeline_options.py | 6 ++++++
sdks/python/apache_beam/runners/dataflow/dataflow_runner.py | 5 +++++
sdks/python/scripts/run_integration_test.sh | 9 +++++----
7 files changed, 43 insertions(+), 6 deletions(-)
diff --git a/runners/google-cloud-dataflow-java/build.gradle b/runners/google-cloud-dataflow-java/build.gradle
index c7e126b..c871296 100644
--- a/runners/google-cloud-dataflow-java/build.gradle
+++ b/runners/google-cloud-dataflow-java/build.gradle
@@ -321,7 +321,7 @@ task googleCloudPlatformLegacyWorkerKmsIntegrationTest(type: Test) {
])
include '**/*KmsKeyIT.class'
- exclude '**/BigQueryKmsKeyIT.java' // Only needs to run on direct runner.
+ exclude '**/BigQueryKmsKeyIT.class' // Only needs to run on direct runner.
maxParallelForks 4
classpath = configurations.googleCloudPlatformIntegrationTest
testClassesDirs = files(project(":beam-sdks-java-io-google-cloud-platform").sourceSets.test.output.classesDirs)
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java
index 503460c..e653e87 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java
@@ -73,6 +73,7 @@ import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.IterableCoder;
import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
import org.apache.beam.sdk.io.Read;
+import org.apache.beam.sdk.options.ExperimentalOptions;
import org.apache.beam.sdk.options.StreamingOptions;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.runners.TransformHierarchy;
@@ -402,6 +403,11 @@ public class DataflowPipelineTranslator {
if (options.getServiceAccount() != null) {
environment.setServiceAccountEmail(options.getServiceAccount());
}
+ // TODO(BEAM-6664): Remove once Dataflow supports --dataflowKmsKey.
+ if (options.getDataflowKmsKey() != null) {
+ ExperimentalOptions.addExperiment(
+ options, String.format("service_default_cmek_config=%s", options.getDataflowKmsKey()));
+ }
pipeline.traverseTopologically(this);
return job;
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ExperimentalOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ExperimentalOptions.java
index 2e5a245..9abf7d6 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ExperimentalOptions.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/ExperimentalOptions.java
@@ -20,6 +20,7 @@ package org.apache.beam.sdk.options;
import java.util.List;
import javax.annotation.Nullable;
import org.apache.beam.sdk.annotations.Experimental;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Lists;
/**
* Apache Beam provides a number of experimental features that can be enabled with this flag. If
@@ -47,4 +48,16 @@ public interface ExperimentalOptions extends PipelineOptions {
List<String> experiments = options.as(ExperimentalOptions.class).getExperiments();
return experiments != null && experiments.contains(experiment);
}
+
+ /** Adds experiment to options if not already present. */
+ static void addExperiment(ExperimentalOptions options, String experiment) {
+ List<String> experiments = options.getExperiments();
+ if (experiments == null) {
+ experiments = Lists.newArrayList();
+ }
+ if (!experiments.contains(experiment)) {
+ experiments.add(experiment);
+ }
+ options.setExperiments(experiments);
+ }
}
diff --git a/sdks/python/apache_beam/io/gcp/gcsio_integration_test.py b/sdks/python/apache_beam/io/gcp/gcsio_integration_test.py
index ff15a6f..18a367d 100644
--- a/sdks/python/apache_beam/io/gcp/gcsio_integration_test.py
+++ b/sdks/python/apache_beam/io/gcp/gcsio_integration_test.py
@@ -85,7 +85,13 @@ class GcsIOIntegrationTest(unittest.TestCase):
src_checksum = self.gcsio.checksum(src)
dst_checksum = self.gcsio.checksum(dst)
self.assertEqual(src_checksum, dst_checksum)
- self.assertEqual(self.gcsio.kms_key(dst), dst_kms_key_name)
+ actual_dst_kms_key = self.gcsio.kms_key(dst)
+ if actual_dst_kms_key is None:
+ self.assertEqual(actual_dst_kms_key, dst_kms_key_name)
+ else:
+ self.assertTrue(actual_dst_kms_key.startswith(dst_kms_key_name),
+ "got: %s, wanted startswith: %s" % (actual_dst_kms_key,
+ dst_kms_key_name))
def _test_copy(self, name, kms_key_name=None,
max_bytes_rewritten_per_call=None, src=None):
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index 71cd795..1cee4a8 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -590,6 +590,12 @@ class DebugOptions(PipelineOptions):
'enabled with this flag. Please sync with the owners of the runner '
'before enabling any experiments.'))
+ def add_experiment(self, experiment):
+ if self.experiments is None:
+ self.experiments = []
+ if experiment not in self.experiments:
+ self.experiments.append(experiment)
+
def lookup_experiment(self, key, default=None):
if not self.experiments:
return default
diff --git a/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py b/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
index 1d24d10..460a5e0 100644
--- a/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
+++ b/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py
@@ -381,6 +381,11 @@ class DataflowRunner(PipelineRunner):
and/or enable_streaming_engine are present. It is recommended you
only set enableStreamingEngine.""")
+ # TODO(BEAM-6664): Remove once Dataflow supports --dataflow_kms_key.
+ if google_cloud_options.dataflow_kms_key is not None:
+ debug_options.add_experiment('service_default_cmek_config=' +
+ google_cloud_options.dataflow_kms_key)
+
self.job = apiclient.Job(options, self.proto_pipeline)
# Dataflow runner requires a KV type for GBK inputs, hence we enforce that
diff --git a/sdks/python/scripts/run_integration_test.sh b/sdks/python/scripts/run_integration_test.sh
index f8f8bef..33e8199 100755
--- a/sdks/python/scripts/run_integration_test.sh
+++ b/sdks/python/scripts/run_integration_test.sh
@@ -71,9 +71,7 @@ NUM_WORKERS=1
SLEEP_SECS=20
STREAMING=false
WORKER_JAR=""
-# Specify "/cryptoKeyVersions/1" suffix for testing simplicity. For this to work
-# in the long term, this key has rotation disabled.
-KMS_KEY_NAME="projects/apache-beam-testing/locations/global/keyRings/beam-it/cryptoKeys/test/cryptoKeyVersions/1"
+KMS_KEY_NAME="projects/apache-beam-testing/locations/global/keyRings/beam-it/cryptoKeys/test"
# Default test (nose) options.
# Default test sets are full integration tests.
@@ -202,7 +200,10 @@ if [[ -z $PIPELINE_OPTS ]]; then
fi
if [[ ! -z "$KMS_KEY_NAME" ]]; then
- opts+=("--kms_key_name=$KMS_KEY_NAME")
+ opts+=(
+ "--kms_key_name=$KMS_KEY_NAME"
+ "--dataflow_kms_key=$KMS_KEY_NAME"
+ )
fi
PIPELINE_OPTS=$(IFS=" " ; echo "${opts[*]}")