You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2017/09/15 20:53:06 UTC
[1/2] beam git commit: [BEAM-2956] Attempt to correctly report the
Dataflow distribution in GCP related modules.
Repository: beam
Updated Branches:
refs/heads/master 979923a4e -> 2693bd726
[BEAM-2956] Attempt to correctly report the Dataflow distribution in GCP related modules.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/3a2b4182
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/3a2b4182
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/3a2b4182
Branch: refs/heads/master
Commit: 3a2b41826bfba77f576d1bb5f6fc780a9c47b1bb
Parents: 979923a
Author: Luke Cwik <lc...@google.com>
Authored: Wed Sep 13 17:00:04 2017 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Fri Sep 15 13:52:32 2017 -0700
----------------------------------------------------------------------
.../beam/runners/dataflow/DataflowRunner.java | 7 +-
.../runners/dataflow/DataflowRunnerInfo.java | 91 +++++++++++++++-----
.../dataflow/DataflowRunnerInfoTest.java | 9 ++
.../runners/dataflow/DataflowRunnerTest.java | 5 +-
4 files changed, 83 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/3a2b4182/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
index 422fd11..d5a9845 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java
@@ -132,7 +132,6 @@ import org.apache.beam.sdk.util.CoderUtils;
import org.apache.beam.sdk.util.InstanceBuilder;
import org.apache.beam.sdk.util.MimeTypes;
import org.apache.beam.sdk.util.NameUtils;
-import org.apache.beam.sdk.util.ReleaseInfo;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PBegin;
@@ -533,14 +532,14 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> {
Job newJob = jobSpecification.getJob();
newJob.setClientRequestId(requestId);
- ReleaseInfo releaseInfo = ReleaseInfo.getReleaseInfo();
- String version = releaseInfo.getVersion();
+ DataflowRunnerInfo dataflowRunnerInfo = DataflowRunnerInfo.getDataflowRunnerInfo();
+ String version = dataflowRunnerInfo.getVersion();
checkState(
!version.equals("${pom.version}"),
"Unable to submit a job to the Dataflow service with unset version ${pom.version}");
System.out.println("Dataflow SDK version: " + version);
- newJob.getEnvironment().setUserAgent((Map) releaseInfo.getProperties());
+ newJob.getEnvironment().setUserAgent((Map) dataflowRunnerInfo.getProperties());
// The Dataflow Service may write to the temporary directory directly, so
// must be verified.
if (!isNullOrEmpty(options.getGcpTempLocation())) {
http://git-wip-us.apache.org/repos/asf/beam/blob/3a2b4182/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunnerInfo.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunnerInfo.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunnerInfo.java
index 12b3f38..69e4f46 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunnerInfo.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunnerInfo.java
@@ -19,23 +19,66 @@ package org.apache.beam.runners.dataflow;
import static com.google.common.base.Preconditions.checkState;
+import com.google.common.collect.ImmutableMap;
import java.io.IOException;
import java.io.InputStream;
+import java.util.Map;
import java.util.Properties;
+import org.apache.beam.sdk.util.ReleaseInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Populates versioning and other information for {@link DataflowRunner}.
*/
-public final class DataflowRunnerInfo {
+public final class DataflowRunnerInfo extends ReleaseInfo {
private static final Logger LOG = LoggerFactory.getLogger(DataflowRunnerInfo.class);
- private static final String PROPERTIES_PATH =
+ private static final String APACHE_BEAM_DISTRIBUTION_PROPERTIES_PATH =
"/org/apache/beam/runners/dataflow/dataflow.properties";
+ private static final String DATAFLOW_DISTRIBUTION_PROPERTIES_PATH =
+ "/org/apache/beam/runners/dataflow/dataflow-distribution.properties";
+ private static final String FNAPI_ENVIRONMENT_MAJOR_VERSION_KEY =
+ "fnapi.environment.major.version";
+ private static final String LEGACY_ENVIRONMENT_MAJOR_VERSION_KEY =
+ "legacy.environment.major.version";
+ private static final String CONTAINER_VERSION_KEY = "container.version";
private static class LazyInit {
- private static final DataflowRunnerInfo INSTANCE = new DataflowRunnerInfo(PROPERTIES_PATH);
+ private static final DataflowRunnerInfo INSTANCE;
+ static {
+ Properties properties;
+ try {
+ properties = load(DATAFLOW_DISTRIBUTION_PROPERTIES_PATH);
+ if (properties == null) {
+ properties = load(APACHE_BEAM_DISTRIBUTION_PROPERTIES_PATH);
+ }
+ if (properties == null) {
+ // Print a warning if we can not load either the Dataflow distribution properties
+ // or the
+ LOG.warn("Dataflow runner properties resource not found.");
+ properties = new Properties();
+ }
+ } catch (IOException e) {
+ LOG.warn("Error loading Dataflow runner properties resource: ", e);
+ properties = new Properties();
+ }
+
+ // Inherit the name and version from the Apache Beam distribution if this isn't
+ // the Dataflow distribution.
+ if (!properties.containsKey("name")) {
+ properties.setProperty("name", ReleaseInfo.getReleaseInfo().getName());
+ }
+ if (!properties.containsKey("version")) {
+ properties.setProperty("version", ReleaseInfo.getReleaseInfo().getVersion());
+ }
+ copyFromSystemProperties("java.vendor", properties);
+ copyFromSystemProperties("java.version", properties);
+ copyFromSystemProperties("os.arch", properties);
+ copyFromSystemProperties("os.name", properties);
+ copyFromSystemProperties("os.version", properties);
+ INSTANCE = new DataflowRunnerInfo(ImmutableMap.copyOf((Map) properties));
+ }
}
/**
@@ -45,20 +88,12 @@ public final class DataflowRunnerInfo {
return LazyInit.INSTANCE;
}
- private Properties properties;
-
- private static final String FNAPI_ENVIRONMENT_MAJOR_VERSION_KEY =
- "fnapi.environment.major.version";
- private static final String LEGACY_ENVIRONMENT_MAJOR_VERSION_KEY =
- "legacy.environment.major.version";
- private static final String CONTAINER_VERSION_KEY = "container.version";
-
/** Provides the legacy environment's major version number. */
public String getLegacyEnvironmentMajorVersion() {
checkState(
properties.containsKey(LEGACY_ENVIRONMENT_MAJOR_VERSION_KEY),
"Unknown legacy environment major version");
- return properties.getProperty(LEGACY_ENVIRONMENT_MAJOR_VERSION_KEY);
+ return properties.get(LEGACY_ENVIRONMENT_MAJOR_VERSION_KEY);
}
/** Provides the FnAPI environment's major version number. */
@@ -66,7 +101,7 @@ public final class DataflowRunnerInfo {
checkState(
properties.containsKey(FNAPI_ENVIRONMENT_MAJOR_VERSION_KEY),
"Unknown FnAPI environment major version");
- return properties.getProperty(FNAPI_ENVIRONMENT_MAJOR_VERSION_KEY);
+ return properties.get(FNAPI_ENVIRONMENT_MAJOR_VERSION_KEY);
}
/** Provides the container version that will be used for constructing harness image paths. */
@@ -74,21 +109,33 @@ public final class DataflowRunnerInfo {
checkState(
properties.containsKey(CONTAINER_VERSION_KEY),
"Unknown container version");
- return properties.getProperty(CONTAINER_VERSION_KEY);
+ return properties.get(CONTAINER_VERSION_KEY);
+ }
+
+ public Map<String, String> getProperties() {
+ return ImmutableMap.copyOf((Map) properties);
}
- private DataflowRunnerInfo(String resourcePath) {
- properties = new Properties();
+ private final Map<String, String> properties;
+ private DataflowRunnerInfo(Map<String, String> properties) {
+ this.properties = properties;
+ }
- try (InputStream in = DataflowRunnerInfo.class.getResourceAsStream(PROPERTIES_PATH)) {
+ private static Properties load(String path) throws IOException {
+ Properties properties = new Properties();
+ try (InputStream in = DataflowRunnerInfo.class.getResourceAsStream(path)) {
if (in == null) {
- LOG.warn("Dataflow runner properties resource not found: {}", resourcePath);
- return;
+ return null;
}
-
properties.load(in);
- } catch (IOException e) {
- LOG.warn("Error loading Dataflow runner properties resource: ", e);
+ }
+ return properties;
+ }
+
+ private static void copyFromSystemProperties(String property, Properties properties) {
+ String value = System.getProperty(property);
+ if (value != null) {
+ properties.setProperty(property, value);
}
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/3a2b4182/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerInfoTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerInfoTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerInfoTest.java
index 3502040..051f37e 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerInfoTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerInfoTest.java
@@ -19,6 +19,7 @@ package org.apache.beam.runners.dataflow;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.not;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
@@ -26,6 +27,9 @@ import org.junit.Test;
/**
* Tests for {@link DataflowRunnerInfo}.
+ *
+ * <p>Note that tests for checking that the Dataflow distribution correctly loads overridden
+ * properties is contained within the Dataflow distribution.
*/
public class DataflowRunnerInfoTest {
@@ -50,5 +54,10 @@ public class DataflowRunnerInfoTest {
"container version invalid",
info.getContainerVersion(),
not(containsString("$")));
+
+ for (String property
+ : new String[]{ "java.vendor", "java.version", "os.arch", "os.name", "os.version"}) {
+ assertEquals(System.getProperty(property), info.getProperties().get(property));
+ }
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/3a2b4182/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
index 761ce89..f1e3805 100644
--- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
+++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java
@@ -109,7 +109,6 @@ import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.transforms.windowing.Sessions;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.util.GcsUtil;
-import org.apache.beam.sdk.util.ReleaseInfo;
import org.apache.beam.sdk.util.gcsfs.GcsPath;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
@@ -549,10 +548,10 @@ public class DataflowRunnerTest implements Serializable {
cloudDataflowDataset,
workflowJob.getEnvironment().getDataset());
assertEquals(
- ReleaseInfo.getReleaseInfo().getName(),
+ DataflowRunnerInfo.getDataflowRunnerInfo().getName(),
workflowJob.getEnvironment().getUserAgent().get("name"));
assertEquals(
- ReleaseInfo.getReleaseInfo().getVersion(),
+ DataflowRunnerInfo.getDataflowRunnerInfo().getVersion(),
workflowJob.getEnvironment().getUserAgent().get("version"));
}
[2/2] beam git commit: [BEAM-2956] Attempt to correctly report the
Dataflow distribution in GCP related modules.
Posted by lc...@apache.org.
[BEAM-2956] Attempt to correctly report the Dataflow distribution in GCP related modules.
This closes #3849
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/2693bd72
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/2693bd72
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/2693bd72
Branch: refs/heads/master
Commit: 2693bd726fa4baffdeda04e68f912b251b5d24e2
Parents: 979923a 3a2b418
Author: Luke Cwik <lc...@google.com>
Authored: Fri Sep 15 13:52:58 2017 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Fri Sep 15 13:52:58 2017 -0700
----------------------------------------------------------------------
.../beam/runners/dataflow/DataflowRunner.java | 7 +-
.../runners/dataflow/DataflowRunnerInfo.java | 91 +++++++++++++++-----
.../dataflow/DataflowRunnerInfoTest.java | 9 ++
.../runners/dataflow/DataflowRunnerTest.java | 5 +-
4 files changed, 83 insertions(+), 29 deletions(-)
----------------------------------------------------------------------