tvalentyn commented on code in PR #26220:
URL: https://github.com/apache/beam/pull/26220#discussion_r1167108375
##########
sdks/python/apache_beam/options/pipeline_options.py:
##########
@@ -876,6 +876,16 @@ def validate(self, validator):
return errors
+ def lookup_dataflow_service_option(self, key, default=None):
Review Comment:
I would prefer lookup_dataflow_service_option have the same implementation
w.r.t. defaults as lookup_experiment to be consistent. You have changed the
logic. If preexisting logic doesn't work for you, let's name it as a
specialized a helper function `get_cloud_debugger_service_name`.
##########
sdks/python/apache_beam/runners/worker/sdk_worker_main.py:
##########
@@ -175,20 +175,20 @@ def create_harness(environment, dry_run=False):
def main(unused_argv):
"""Main entry point for SDK Fn Harness."""
- fn_log_handler, sdk_harness, sdk_pipeline_options =
create_harness(os.environ)
- experiments = sdk_pipeline_options.view_as(DebugOptions).experiments or []
- dataflow_service_options = (
- sdk_pipeline_options.view_as(GoogleCloudOptions).dataflow_service_options
- or [])
- if (_ENABLE_GOOGLE_CLOUD_PROFILER in experiments) or (
- _ENABLE_GOOGLE_CLOUD_PROFILER in dataflow_service_options):
+ (fn_log_handler, sdk_harness,
+ sdk_pipeline_options) = create_harness(os.environ)
+ experiments = (sdk_pipeline_options.view_as(DebugOptions).experiments or [])
+ service_name = sdk_pipeline_options.view_as(
Review Comment:
```suggestion
gcp_profiler_service_name = sdk_pipeline_options.view_as(
```
##########
sdks/python/apache_beam/options/pipeline_options.py:
##########
@@ -876,6 +876,16 @@ def validate(self, validator):
return errors
+ def lookup_dataflow_service_option(self, key, default=None):
+ if not self.dataflow_service_options:
+ return None
+ elif key in self.dataflow_service_options:
+ return default
+ for service_name in self.dataflow_service_options:
Review Comment:
`option_name` is would be a better name for this varilable. The service is
Dataflow in this case.
##########
sdks/python/apache_beam/options/pipeline_options.py:
##########
@@ -876,6 +876,16 @@ def validate(self, validator):
return errors
+ def lookup_dataflow_service_option(self, key, default=None):
Review Comment:
also, let's add unit tests for this helper.
##########
sdks/python/apache_beam/runners/worker/sdk_worker_main.py:
##########
@@ -175,20 +175,20 @@ def create_harness(environment, dry_run=False):
def main(unused_argv):
"""Main entry point for SDK Fn Harness."""
- fn_log_handler, sdk_harness, sdk_pipeline_options =
create_harness(os.environ)
- experiments = sdk_pipeline_options.view_as(DebugOptions).experiments or []
- dataflow_service_options = (
- sdk_pipeline_options.view_as(GoogleCloudOptions).dataflow_service_options
- or [])
- if (_ENABLE_GOOGLE_CLOUD_PROFILER in experiments) or (
- _ENABLE_GOOGLE_CLOUD_PROFILER in dataflow_service_options):
+ (fn_log_handler, sdk_harness,
+ sdk_pipeline_options) = create_harness(os.environ)
+ experiments = (sdk_pipeline_options.view_as(DebugOptions).experiments or [])
+ service_name = sdk_pipeline_options.view_as(
+ GoogleCloudOptions).lookup_dataflow_service_option(
+ _ENABLE_GOOGLE_CLOUD_PROFILER, default=os.environ["JOB_NAME"])
+
+ if ((_ENABLE_GOOGLE_CLOUD_PROFILER in experiments) or service_name):
Review Comment:
I think it makes sense to move this code into a helper function, for
example: `_start_profiler_if_requested()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
