Add saveProfilesToGcs option; deprecate enableProfilingAgent (#497)

Add saveProfilesToGcs option to DataflowProfilingOptions

This option (if present) is verified to be a writeable GCS path during pipeline submission.

Author: bjchambers

sdk/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowProfilingOptions.java

@@ -27,11 +27,17 @@
 @Hidden
 public interface DataflowProfilingOptions {
 
-  @Description("Whether to periodically dump profiling information to local disk.\n"
-      + "WARNING: Enabling this option may fill local disk with profiling information.")
+  @Description(
+      "This option is deprecated and ignored. Using --saveProfilesToGcs=<GCS path> is preferred.")
   boolean getEnableProfilingAgent();
   void setEnableProfilingAgent(boolean enabled);
 
+  @Description(
+      "When set to a non-empty value, enables recording profiles and saving them to GCS.\n"
+      + "Profiles will continue until the pipeline is stopped or updated without this option.\n")
+  String getSaveProfilesToGcs();
+  void setSaveProfilesToGcs(String gcsPath);
+
   @Description(
       "[INTERNAL] Additional configuration for the profiling agent. Not typically necessary.")
   @Hidden

sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/DataflowPipelineRunner.java

@@ -275,6 +275,14 @@ public static DataflowPipelineRunner fromOptions(PipelineOptions options) {
     if (dataflowOptions.getTempLocation() != null) {
       validator.validateOutputFilePrefixSupported(dataflowOptions.getTempLocation());
     }
+    if (!Strings.isNullOrEmpty(dataflowOptions.getSaveProfilesToGcs())) {
+      validator.validateOutputFilePrefixSupported(dataflowOptions.getSaveProfilesToGcs());
+    }
+    if (dataflowOptions.getEnableProfilingAgent()) {
+      LOG.error("--enableProfilingAgent is no longer supported, and will be ignored. "
+          + "Use --saveProfilesToGcs instead.");
+    }
+
     if (Strings.isNullOrEmpty(dataflowOptions.getTempLocation())) {
       dataflowOptions.setTempLocation(dataflowOptions.getStagingLocation());
     } else if (Strings.isNullOrEmpty(dataflowOptions.getStagingLocation())) {