You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/01/11 09:30:44 UTC
[GitHub] [beam] daria-malkova opened a new pull request #16477: [BEAM-13560] Split builder into several builder for each step of pipeline execution
daria-malkova opened a new pull request #16477:
URL: https://github.com/apache/beam/pull/16477
Split builder into several builders for each step of pipeline execution
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
`ValidatesRunner` compliance status (on master branch)
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/badge/icon?subject=V1+Streaming">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/badge/icon?subject=V1+Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/badge/icon?subject=V2+Streaming">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/badge/icon?subject=Java+8">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/badge/icon?subject=Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/badge/icon?subject=Portable+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/badge/icon?subject=Portable">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/badge/icon?subject=Structured+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/badge/icon?subject=ValCont">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
</tbody>
</table>
Examples testing status on various runners
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/badge/icon?subject=V1+Java11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
Post-Commit SDK/Transform Integration Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Go</th>
<th>Java</th>
<th>Python</th>
</tr>
</thead>
<tbody>
<tr>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/badge/icon?subject=3.6">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/badge/icon?subject=3.7">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/badge/icon?subject=3.8">
</a>
</td>
</tr>
</tbody>
</table>
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>---</th>
<th>Java</th>
<th>Python</th>
<th>Go</th>
<th>Website</th>
<th>Whitespace</th>
<th>Typescript</th>
</tr>
</thead>
<tbody>
<tr>
<td>Non-portable</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/badge/icon?subject=Tests">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/badge/icon?subject=Lint">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon?subject=Docker">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon?subject=Docs">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Portable</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] daria-malkova commented on a change in pull request #16477: [BEAM-13560][Playground] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
daria-malkova commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r783735862
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,134 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(lc *fs_tool.LifeCycle, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// SetupValidatorBuilder return executor with set args for validator
+func SetupValidatorBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// SetupPreparatorBuilder return executor with set args for preparator
+func SetupPreparatorBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
- WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// SetupCompilerBuilder return executor with set args for compiler
+func SetupCompilerBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
WithArgs(executorConfig.CompileArgs).
WithFileName(lc.GetAbsoluteSourceFilePath()).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(lc.GetAbsoluteSourceFolderPath()))
+ }
+ return &builder
+}
+
+// SetupRunBuilder return executor with set args for runner
+func SetupRunBuilder(lc *fs_tool.LifeCycle, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
Review comment:
Done
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder_test.go
##########
@@ -23,100 +23,293 @@ import (
"beam.apache.org/playground/backend/internal/utils"
"fmt"
"github.com/google/uuid"
+ "reflect"
"strings"
"testing"
)
-func TestSetupExecutor(t *testing.T) {
+var lc *fs_tool.LifeCycle
+var sdkEnv *environment.BeamEnvs
+
+func TestMain(m *testing.M) {
+ setup()
+ m.Run()
+}
+
+func setup() {
pipelineId := uuid.New()
- sdk := pb.Sdk_SDK_JAVA
- lc, err := fs_tool.NewLifeCycle(sdk, pipelineId, "")
- if err != nil {
- t.Error(err)
- }
- pipelineOptions := ""
+ sdk := pb.Sdk_SDK_PYTHON
+ lc, _ = fs_tool.NewLifeCycle(sdk, pipelineId, "")
executorConfig := &environment.ExecutorConfig{
CompileCmd: "MOCK_COMPILE_CMD",
- RunCmd: "MOCK_RUN_CMD",
- TestCmd: "MOCK_TEST_CMD",
CompileArgs: []string{"MOCK_COMPILE_ARG"},
- RunArgs: []string{"MOCK_RUN_ARG"},
- TestArgs: []string{"MOCK_TEST_ARG"},
}
+ sdkEnv = environment.NewBeamEnvs(sdk, executorConfig, "", 0)
+}
+
+func TestSetupValidatorBuilder(t *testing.T) {
+ vals, err := utils.GetValidators(sdkEnv.ApacheBeamSdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
panic(err)
}
+ wantExecutor := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(vals)
- srcFilePath := lc.GetAbsoluteSourceFilePath()
+ wrongSdkEnv := environment.NewBeamEnvs(pb.Sdk_SDK_UNSPECIFIED, sdkEnv.ExecutorConfig, "", 0)
- sdkEnv := environment.NewBeamEnvs(sdk, executorConfig, "", 0)
- val, err := utils.GetValidators(sdk, srcFilePath)
- if err != nil {
- panic(err)
+ type args struct {
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ wantErr bool
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected validator builder.
+ name: "Test correct validator builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ wantErr: false,
+ },
+ {
+ // Test case with calling Setup with incorrect SDK.
+ // As a result, want to receive an error.
+ name: "incorrect sdk",
+ args: args{
+ lc: lc,
+ sdkEnv: wrongSdkEnv,
+ },
+ want: nil,
+ wantErr: true,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, err := SetupValidatorBuilder(tt.args.lc, tt.args.sdkEnv)
+ if (err != nil) != tt.wantErr {
+ t.Errorf("SetupValidatorBuilder() error = %v, wantErr %v", err, tt.wantErr)
+ return
+ }
+ if err != nil && !reflect.DeepEqual(got, tt.want) {
+ t.Errorf("SetupValidatorBuilder() got = %v, want %v", got, tt.want)
+ return
+ }
+ if err == nil && !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupValidatorBuilder() got = %v\n, want %v", got.Build(), tt.want.Build())
+ }
+ })
}
- prep, err := utils.GetPreparators(sdk, srcFilePath)
+}
+
+func TestSetupPreparatorBuilder(t *testing.T) {
+ prep, err := utils.GetPreparators(sdkEnv.ApacheBeamSdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
panic(err)
}
-
wantExecutor := executors.NewExecutorBuilder().
- WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
- WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
- WithSdkPreparators(prep).
- WithCompiler().
- WithCommand(executorConfig.CompileCmd).
- WithArgs(executorConfig.CompileArgs).
- WithFileName(srcFilePath).
- WithRunner().
- WithCommand(executorConfig.RunCmd).
- WithArgs(executorConfig.RunArgs).
- WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
- ExecutorBuilder
+ WithSdkPreparators(prep)
+
+ wrongSdkEnv := environment.NewBeamEnvs(pb.Sdk_SDK_UNSPECIFIED, sdkEnv.ExecutorConfig, "", 0)
type args struct {
- lc *fs_tool.LifeCycle
- pipelineOptions string
- sdkEnv *environment.BeamEnvs
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
}
tests := []struct {
name string
args args
want *executors.ExecutorBuilder
wantErr bool
}{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected preparator builder.
+ name: "Test correct preparator builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ wantErr: false,
+ },
{
// Test case with calling Setup with incorrect SDK.
// As a result, want to receive an error.
- name: "incorrect sdk",
- args: args{lc, pipelineOptions, environment.NewBeamEnvs(pb.Sdk_SDK_UNSPECIFIED, executorConfig, "", 0)},
+ name: "incorrect sdk",
+ args: args{
+ lc: lc,
+ sdkEnv: wrongSdkEnv,
+ },
want: nil,
wantErr: true,
},
- {
- // Test case with calling Setup with correct SDK.
- // As a result, want to receive an expected builder.
- name: "correct sdk",
- args: args{lc, pipelineOptions, sdkEnv},
- want: &wantExecutor,
- wantErr: false,
- },
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- got, err := SetupExecutorBuilder(tt.args.lc, tt.args.pipelineOptions, tt.args.sdkEnv)
+ got, err := SetupPreparatorBuilder(tt.args.lc, tt.args.sdkEnv)
if (err != nil) != tt.wantErr {
- t.Errorf("SetupExecutorBuilder() error = %v, wantErr %v", err, tt.wantErr)
+ t.Errorf("SetupPreparatorBuilder() error = %v, wantErr %v", err, tt.wantErr)
return
}
- if err == nil && fmt.Sprint(got.Build()) != fmt.Sprint(tt.want.Build()) {
- t.Errorf("SetupExecutorBuilder() got = %v\n, want %v", got.Build(), tt.want.Build())
+ if err != nil && !reflect.DeepEqual(got, tt.want) {
+ t.Errorf("SetupPreparatorBuilder() got = %v, want %v", got, tt.want)
+ return
+ }
+ if err == nil && !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupPreparatorBuilder() got = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func TestSetupCompilerBuilder(t *testing.T) {
+ wantExecutor := executors.NewExecutorBuilder().
+ WithCompiler().
+ WithCommand(sdkEnv.ExecutorConfig.CompileCmd).
+ WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
+ WithArgs(sdkEnv.ExecutorConfig.CompileArgs).
+ WithFileName(lc.GetAbsoluteSourceFilePath())
+
+ type args struct {
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected compiler builder.
+ name: "Test correct compiler builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got := SetupCompilerBuilder(tt.args.lc, tt.args.sdkEnv)
+ if !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupCompilerBuilder() = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func TestSetupRunBuilder(t *testing.T) {
+ wantExecutor := executors.NewExecutorBuilder().
+ WithRunner().
+ WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
+ WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
+ WithCommand(sdkEnv.ExecutorConfig.RunCmd).
+ WithArgs(sdkEnv.ExecutorConfig.RunArgs).
+ WithPipelineOptions(strings.Split("", " "))
+
+ type args struct {
+ lc *fs_tool.LifeCycle
+ pipelineOptions string
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected run builder.
+ name: "Test correct run builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, _ := SetupRunBuilder(tt.args.lc, tt.args.pipelineOptions, tt.args.sdkEnv)
+ if !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupRunBuilder() got = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func TestSetupTestBuilder(t *testing.T) {
+ wantExecutor := executors.NewExecutorBuilder().
+ WithTestRunner().
+ WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
+ WithCommand(sdkEnv.ExecutorConfig.TestCmd).
+ WithArgs(sdkEnv.ExecutorConfig.TestArgs).
+ WithWorkingDir(lc.GetAbsoluteSourceFolderPath())
+
+ type args struct {
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected test builder.
+ name: "Test correct test builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, _ := SetupTestBuilder(tt.args.lc, tt.args.sdkEnv)
+ if !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupTestBuilder() got = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func Test_replaceLogPlaceholder(t *testing.T) {
+ type args struct {
+ lc *fs_tool.LifeCycle
+ executorConfig *environment.ExecutorConfig
+ }
+ tests := []struct {
+ name string
+ args args
+ want []string
+ }{
+ {
+ name: "",
Review comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] daria-malkova commented on a change in pull request #16477: [BEAM-13560][Playground] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
daria-malkova commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r784027376
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
Review comment:
Let's do it at the next PR, it is not very critical right now
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
Review comment:
same
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
Review comment:
same
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #16477: [BEAM-13560][Playground] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #16477:
URL: https://github.com/apache/beam/pull/16477#issuecomment-1013427791
ah this LGTM. Thanks all!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] daria-malkova commented on pull request #16477: [BEAM-13560][Playground] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
daria-malkova commented on pull request #16477:
URL: https://github.com/apache/beam/pull/16477#issuecomment-1012210927
R: @pabloem
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] daria-malkova commented on a change in pull request #16477: [BEAM-13560] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
daria-malkova commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r782907002
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +157,121 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func CompileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, executor executors.Executor, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return executors.Executor{}, true
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor = executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return executors.Executor{}, true
+ }
+ if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
+ err := <-errorChannel
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
+ return executors.Executor{}, true
+ } // Compile step is finished and code is compiled
+ if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
+ return executors.Executor{}, true
+ }
+ }
+ return executor, false
+}
+
+func PrepareStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupPreparatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return executors.Executor{}, false
Review comment:
Didn't notice, you are right, fixed.
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +157,121 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func CompileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, executor executors.Executor, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return executors.Executor{}, true
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor = executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return executors.Executor{}, true
Review comment:
Make sense. Done.
##########
File path: playground/backend/internal/fs_tool/java_fs.go
##########
@@ -36,8 +36,7 @@ func newJavaLifeCycle(pipelineId uuid.UUID, pipelinesFolder string) *LifeCycle {
}
// executableName returns name that should be executed (HelloWorld for HelloWorld.class for java SDK)
-func executableName(pipelineId uuid.UUID, pipelinesFolder string) (string, error) {
- baseFileFolder := filepath.Join(pipelinesFolder, pipelineId.String())
+func executableName(baseFileFolder string) (string, error) {
binFileFolder := filepath.Join(baseFileFolder, compiledFolderName)
dirEntries, err := os.ReadDir(binFileFolder)
Review comment:
Good suggestion, let's fix it later when DTO will be merged
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -61,117 +60,65 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
DeleteFolders(pipelineId, lc)
}(lc)
- errorChannel := make(chan error, 1)
- successChannel := make(chan bool, 1)
cancelChannel := make(chan bool, 1)
- stopReadLogsChannel := make(chan bool, 1)
- finishReadLogsChannel := make(chan bool, 1)
+
var validationResults sync.Map
go cancelCheck(pipelineLifeCycleCtx, pipelineId, cancelChannel, cacheService)
- executorBuilder, err := builder.SetupExecutorBuilder(lc, utils.ReduceWhiteSpacesToSinge(pipelineOptions), sdkEnv)
- if err != nil {
- _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ executor, done := ValidateStep(ctx, cacheService, lc, pipelineId, sdkEnv, pipelineLifeCycleCtx, &validationResults, cancelChannel)
+ if done {
return
}
- executor := executorBuilder.Build()
- // Validate
- logger.Infof("%s: Validate() ...\n", pipelineId)
- validateFunc := executor.Validate()
- // Run validate function
- go validateFunc(successChannel, errorChannel, &validationResults)
- // Start of the monitoring of background tasks (validate function/cancellation/timeout)
- ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
- if err != nil {
- return
- }
- if !ok {
- // Validate step is finished, but code isn't valid
- err := <-errorChannel
- _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.ValidationOutput, cacheService, "Validate", pb.Status_STATUS_VALIDATION_ERROR)
- return
- }
- // Validate step is finished and code is valid
- if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Validate", pb.Status_STATUS_PREPARING); err != nil {
+ executor, done = PrepareStep(ctx, cacheService, lc, pipelineId, sdkEnv, pipelineLifeCycleCtx, &validationResults, cancelChannel)
+ if done {
return
}
- // Prepare
- logger.Infof("%s: Prepare() ...\n", pipelineId)
- prepareFunc := executor.Prepare()
- // Run prepare function
- go prepareFunc(successChannel, errorChannel, &validationResults)
+ // Check if is unit test
+ validateIsUnitTest, _ := validationResults.Load(validators.UnitTestValidatorName)
+ isUnitTest := validateIsUnitTest.(bool)
- // Start of the monitoring of background tasks (prepare function/cancellation/timeout)
- ok, err = reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
- if err != nil {
- return
- }
- if !ok {
- // Prepare step is finished, but code couldn't be prepared (some error during prepare step)
- err := <-errorChannel
- _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.PreparationOutput, cacheService, "Prepare", pb.Status_STATUS_PREPARATION_ERROR)
- return
- }
- // Prepare step is finished and code is prepared
- if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Prepare", pb.Status_STATUS_COMPILING); err != nil {
+ executor, done = CompileStep(ctx, cacheService, lc, pipelineId, sdkEnv, isUnitTest, pipelineLifeCycleCtx, executor, cancelChannel)
+ if done {
return
}
- // Check if unit test
- validateIsUnitTest, _ := validationResults.Load(validators.UnitTestValidatorName)
- isUnitTest := validateIsUnitTest.(bool)
+ // Run/RunTest
+ RunStep(ctx, cacheService, lc, pipelineId, isUnitTest, sdkEnv, pipelineOptions, pipelineLifeCycleCtx, cancelChannel)
+}
- // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
- if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
- if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
- return
- }
- } else { // in case of Java, Go (not unit test), Scala - need compile step
- // Compile
- if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_JAVA {
- executor = executorBuilder.WithCompiler().
- WithFileName(builder.GetFileNameFromFolder(lc.GetAbsoluteSourceFolderPath())).Build() // Need changed name for unit tests
- }
- logger.Infof("%s: Compile() ...\n", pipelineId)
- compileCmd := executor.Compile(pipelineLifeCycleCtx)
- var compileError bytes.Buffer
- var compileOutput bytes.Buffer
- runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+func RunStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, isUnitTest bool, sdkEnv *environment.BeamEnvs, pipelineOptions string, pipelineLifeCycleCtx context.Context, cancelChannel chan bool) {
+ errorChannel, successChannel := createStatusChannels()
+ stopReadLogsChannel := make(chan bool, 1)
+ finishReadLogsChannel := make(chan bool, 1)
- // Start of the monitoring of background tasks (compile step/cancellation/timeout)
- ok, err = reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
- if err != nil {
- return
- }
- if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
- err := <-errorChannel
- _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
- return
- } // Compile step is finished and code is compiled
- if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
- return
- }
+ executorBuilder := executors.NewExecutorBuilder()
+ err := error(nil)
+ if isUnitTest {
+ executorBuilder, err = builder.SetupTestBuilder(lc, sdkEnv)
+ } else {
+ executorBuilder, err = builder.SetupRunBuilder(lc, utils.ReduceWhiteSpacesToSinge(pipelineOptions), sdkEnv)
}
-
- // Run
- if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_JAVA {
- executor, err = setJavaExecutableFile(lc, pipelineId, cacheService, pipelineLifeCycleCtx, executorBuilder, filepath.Join(appEnv.WorkingDir(), appEnv.PipelinesFolder()))
- if err != nil {
- return
- }
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return
}
- logger.Infof("%s: Run() ...\n", pipelineId)
- runCmd := getExecuteCmd(&validationResults, &executor, pipelineLifeCycleCtx)
+
+ executor := executorBuilder.Build()
+ logger.Infof("%s: Run()/Test() ...\n", pipelineId)
+ runCmd := getExecuteCmd(isUnitTest, &executor, pipelineLifeCycleCtx)
var runError bytes.Buffer
runOutput := streaming.RunOutputWriter{Ctx: pipelineLifeCycleCtx, CacheService: cacheService, PipelineId: pipelineId}
go readLogFile(pipelineLifeCycleCtx, ctx, cacheService, lc.GetAbsoluteLogFilePath(), pipelineId, stopReadLogsChannel, finishReadLogsChannel)
if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO {
// For go SDK all logs are placed to stdErr.
file, err := os.Create(lc.GetAbsoluteLogFilePath())
+ if err == nil {
+ runCmdWithOutput(runCmd, &runOutput, file, successChannel, errorChannel)
+ }
Review comment:
Done
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +157,121 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func CompileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, executor executors.Executor, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return executors.Executor{}, true
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor = executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return executors.Executor{}, true
+ }
+ if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
+ err := <-errorChannel
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
+ return executors.Executor{}, true
+ } // Compile step is finished and code is compiled
+ if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
+ return executors.Executor{}, true
+ }
+ }
+ return executor, false
+}
+
+func PrepareStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupPreparatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return executors.Executor{}, false
+ }
+ executor := executorBuilder.Build()
+ // Prepare
+ logger.Infof("%s: Prepare() ...\n", pipelineId)
+ prepareFunc := executor.Prepare()
+ // Run prepare function
+ go prepareFunc(successChannel, errorChannel, validationResults)
+
+ // Start of the monitoring of background tasks (prepare function/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return executors.Executor{}, true
+ }
+ if !ok {
+ err := <-errorChannel
+ // Prepare step is finished, but code couldn't be prepared (some error during prepare step)
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.PreparationOutput, cacheService, "Prepare", pb.Status_STATUS_PREPARATION_ERROR)
+ return executors.Executor{}, true
+ }
+ // Prepare step is finished and code is prepared
+ if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Prepare", pb.Status_STATUS_COMPILING); err != nil {
+ return executors.Executor{}, true
+ }
+ return executor, false
+}
+
+func ValidateStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupValidatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return executors.Executor{}, false
Review comment:
Done
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -61,117 +60,65 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
DeleteFolders(pipelineId, lc)
}(lc)
- errorChannel := make(chan error, 1)
- successChannel := make(chan bool, 1)
cancelChannel := make(chan bool, 1)
- stopReadLogsChannel := make(chan bool, 1)
- finishReadLogsChannel := make(chan bool, 1)
+
var validationResults sync.Map
go cancelCheck(pipelineLifeCycleCtx, pipelineId, cancelChannel, cacheService)
- executorBuilder, err := builder.SetupExecutorBuilder(lc, utils.ReduceWhiteSpacesToSinge(pipelineOptions), sdkEnv)
- if err != nil {
- _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ executor, done := ValidateStep(ctx, cacheService, lc, pipelineId, sdkEnv, pipelineLifeCycleCtx, &validationResults, cancelChannel)
+ if done {
return
}
- executor := executorBuilder.Build()
- // Validate
- logger.Infof("%s: Validate() ...\n", pipelineId)
- validateFunc := executor.Validate()
- // Run validate function
- go validateFunc(successChannel, errorChannel, &validationResults)
- // Start of the monitoring of background tasks (validate function/cancellation/timeout)
- ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
- if err != nil {
- return
- }
- if !ok {
- // Validate step is finished, but code isn't valid
- err := <-errorChannel
- _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.ValidationOutput, cacheService, "Validate", pb.Status_STATUS_VALIDATION_ERROR)
- return
- }
- // Validate step is finished and code is valid
- if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Validate", pb.Status_STATUS_PREPARING); err != nil {
+ executor, done = PrepareStep(ctx, cacheService, lc, pipelineId, sdkEnv, pipelineLifeCycleCtx, &validationResults, cancelChannel)
+ if done {
return
}
- // Prepare
- logger.Infof("%s: Prepare() ...\n", pipelineId)
- prepareFunc := executor.Prepare()
- // Run prepare function
- go prepareFunc(successChannel, errorChannel, &validationResults)
+ // Check if is unit test
+ validateIsUnitTest, _ := validationResults.Load(validators.UnitTestValidatorName)
+ isUnitTest := validateIsUnitTest.(bool)
- // Start of the monitoring of background tasks (prepare function/cancellation/timeout)
- ok, err = reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
- if err != nil {
- return
- }
- if !ok {
- // Prepare step is finished, but code couldn't be prepared (some error during prepare step)
- err := <-errorChannel
- _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.PreparationOutput, cacheService, "Prepare", pb.Status_STATUS_PREPARATION_ERROR)
- return
- }
- // Prepare step is finished and code is prepared
- if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Prepare", pb.Status_STATUS_COMPILING); err != nil {
+ executor, done = CompileStep(ctx, cacheService, lc, pipelineId, sdkEnv, isUnitTest, pipelineLifeCycleCtx, executor, cancelChannel)
+ if done {
return
}
- // Check if unit test
- validateIsUnitTest, _ := validationResults.Load(validators.UnitTestValidatorName)
- isUnitTest := validateIsUnitTest.(bool)
+ // Run/RunTest
+ RunStep(ctx, cacheService, lc, pipelineId, isUnitTest, sdkEnv, pipelineOptions, pipelineLifeCycleCtx, cancelChannel)
+}
- // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
- if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
- if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
- return
- }
- } else { // in case of Java, Go (not unit test), Scala - need compile step
- // Compile
- if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_JAVA {
- executor = executorBuilder.WithCompiler().
- WithFileName(builder.GetFileNameFromFolder(lc.GetAbsoluteSourceFolderPath())).Build() // Need changed name for unit tests
- }
- logger.Infof("%s: Compile() ...\n", pipelineId)
- compileCmd := executor.Compile(pipelineLifeCycleCtx)
- var compileError bytes.Buffer
- var compileOutput bytes.Buffer
- runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+func RunStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, isUnitTest bool, sdkEnv *environment.BeamEnvs, pipelineOptions string, pipelineLifeCycleCtx context.Context, cancelChannel chan bool) {
Review comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ilya-kozyrev commented on a change in pull request #16477: [BEAM-13560] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
ilya-kozyrev commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r783053929
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +154,122 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func compileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ var executor = executors.Executor{}
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor := executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return nil
+ }
+ if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
+ err := <-errorChannel
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
+ return nil
+ } // Compile step is finished and code is compiled
+ if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ }
+ return &executor
+}
+
+func prepareStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupPreparatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return nil
+ }
+ executor := executorBuilder.Build()
+ // Prepare
Review comment:
do we need this comment?
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,134 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(lc *fs_tool.LifeCycle, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// SetupValidatorBuilder return executor with set args for validator
+func SetupValidatorBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// SetupPreparatorBuilder return executor with set args for preparator
+func SetupPreparatorBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
- WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// SetupCompilerBuilder return executor with set args for compiler
+func SetupCompilerBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
Review comment:
What about naming it just Compiler? Then we could use it like that: `setup_tools.builder.Compiler()`
and the same for all other builders in setup_tools
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] AydarZaynutdinov commented on a change in pull request #16477: [BEAM-13560] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
AydarZaynutdinov commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r782149597
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -61,117 +60,65 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
DeleteFolders(pipelineId, lc)
}(lc)
- errorChannel := make(chan error, 1)
- successChannel := make(chan bool, 1)
cancelChannel := make(chan bool, 1)
- stopReadLogsChannel := make(chan bool, 1)
- finishReadLogsChannel := make(chan bool, 1)
+
var validationResults sync.Map
go cancelCheck(pipelineLifeCycleCtx, pipelineId, cancelChannel, cacheService)
- executorBuilder, err := builder.SetupExecutorBuilder(lc, utils.ReduceWhiteSpacesToSinge(pipelineOptions), sdkEnv)
- if err != nil {
- _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ executor, done := ValidateStep(ctx, cacheService, lc, pipelineId, sdkEnv, pipelineLifeCycleCtx, &validationResults, cancelChannel)
+ if done {
return
}
- executor := executorBuilder.Build()
- // Validate
- logger.Infof("%s: Validate() ...\n", pipelineId)
- validateFunc := executor.Validate()
- // Run validate function
- go validateFunc(successChannel, errorChannel, &validationResults)
- // Start of the monitoring of background tasks (validate function/cancellation/timeout)
- ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
- if err != nil {
- return
- }
- if !ok {
- // Validate step is finished, but code isn't valid
- err := <-errorChannel
- _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.ValidationOutput, cacheService, "Validate", pb.Status_STATUS_VALIDATION_ERROR)
- return
- }
- // Validate step is finished and code is valid
- if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Validate", pb.Status_STATUS_PREPARING); err != nil {
+ executor, done = PrepareStep(ctx, cacheService, lc, pipelineId, sdkEnv, pipelineLifeCycleCtx, &validationResults, cancelChannel)
+ if done {
return
}
- // Prepare
- logger.Infof("%s: Prepare() ...\n", pipelineId)
- prepareFunc := executor.Prepare()
- // Run prepare function
- go prepareFunc(successChannel, errorChannel, &validationResults)
+ // Check if is unit test
+ validateIsUnitTest, _ := validationResults.Load(validators.UnitTestValidatorName)
+ isUnitTest := validateIsUnitTest.(bool)
- // Start of the monitoring of background tasks (prepare function/cancellation/timeout)
- ok, err = reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
- if err != nil {
- return
- }
- if !ok {
- // Prepare step is finished, but code couldn't be prepared (some error during prepare step)
- err := <-errorChannel
- _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.PreparationOutput, cacheService, "Prepare", pb.Status_STATUS_PREPARATION_ERROR)
- return
- }
- // Prepare step is finished and code is prepared
- if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Prepare", pb.Status_STATUS_COMPILING); err != nil {
+ executor, done = CompileStep(ctx, cacheService, lc, pipelineId, sdkEnv, isUnitTest, pipelineLifeCycleCtx, executor, cancelChannel)
+ if done {
return
}
- // Check if unit test
- validateIsUnitTest, _ := validationResults.Load(validators.UnitTestValidatorName)
- isUnitTest := validateIsUnitTest.(bool)
+ // Run/RunTest
+ RunStep(ctx, cacheService, lc, pipelineId, isUnitTest, sdkEnv, pipelineOptions, pipelineLifeCycleCtx, cancelChannel)
+}
- // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
- if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
- if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
- return
- }
- } else { // in case of Java, Go (not unit test), Scala - need compile step
- // Compile
- if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_JAVA {
- executor = executorBuilder.WithCompiler().
- WithFileName(builder.GetFileNameFromFolder(lc.GetAbsoluteSourceFolderPath())).Build() // Need changed name for unit tests
- }
- logger.Infof("%s: Compile() ...\n", pipelineId)
- compileCmd := executor.Compile(pipelineLifeCycleCtx)
- var compileError bytes.Buffer
- var compileOutput bytes.Buffer
- runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+func RunStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, isUnitTest bool, sdkEnv *environment.BeamEnvs, pipelineOptions string, pipelineLifeCycleCtx context.Context, cancelChannel chan bool) {
+ errorChannel, successChannel := createStatusChannels()
+ stopReadLogsChannel := make(chan bool, 1)
+ finishReadLogsChannel := make(chan bool, 1)
- // Start of the monitoring of background tasks (compile step/cancellation/timeout)
- ok, err = reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
- if err != nil {
- return
- }
- if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
- err := <-errorChannel
- _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
- return
- } // Compile step is finished and code is compiled
- if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
- return
- }
+ executorBuilder := executors.NewExecutorBuilder()
+ err := error(nil)
+ if isUnitTest {
+ executorBuilder, err = builder.SetupTestBuilder(lc, sdkEnv)
+ } else {
+ executorBuilder, err = builder.SetupRunBuilder(lc, utils.ReduceWhiteSpacesToSinge(pipelineOptions), sdkEnv)
}
-
- // Run
- if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_JAVA {
- executor, err = setJavaExecutableFile(lc, pipelineId, cacheService, pipelineLifeCycleCtx, executorBuilder, filepath.Join(appEnv.WorkingDir(), appEnv.PipelinesFolder()))
- if err != nil {
- return
- }
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return
}
- logger.Infof("%s: Run() ...\n", pipelineId)
- runCmd := getExecuteCmd(&validationResults, &executor, pipelineLifeCycleCtx)
+
+ executor := executorBuilder.Build()
+ logger.Infof("%s: Run()/Test() ...\n", pipelineId)
+ runCmd := getExecuteCmd(isUnitTest, &executor, pipelineLifeCycleCtx)
var runError bytes.Buffer
runOutput := streaming.RunOutputWriter{Ctx: pipelineLifeCycleCtx, CacheService: cacheService, PipelineId: pipelineId}
go readLogFile(pipelineLifeCycleCtx, ctx, cacheService, lc.GetAbsoluteLogFilePath(), pipelineId, stopReadLogsChannel, finishReadLogsChannel)
if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO {
// For go SDK all logs are placed to stdErr.
file, err := os.Create(lc.GetAbsoluteLogFilePath())
+ if err == nil {
+ runCmdWithOutput(runCmd, &runOutput, file, successChannel, errorChannel)
+ }
Review comment:
Why do we need this one? We have the same code on the `126-129` lines
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +157,121 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func CompileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, executor executors.Executor, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return executors.Executor{}, true
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor = executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return executors.Executor{}, true
Review comment:
Why do we need to return `executors.Executor{}, true`? Maybe we could change it to `nil`, instead of `executors.Executor{}, true` and do something like:
```
...
executor := CompileStep(...)
if executor is nil {
return
}
...
```
Using this approach we do not need to use an additional boolean value.
##########
File path: playground/backend/internal/fs_tool/java_fs.go
##########
@@ -36,8 +36,7 @@ func newJavaLifeCycle(pipelineId uuid.UUID, pipelinesFolder string) *LifeCycle {
}
// executableName returns name that should be executed (HelloWorld for HelloWorld.class for java SDK)
-func executableName(pipelineId uuid.UUID, pipelinesFolder string) (string, error) {
- baseFileFolder := filepath.Join(pipelinesFolder, pipelineId.String())
+func executableName(baseFileFolder string) (string, error) {
binFileFolder := filepath.Join(baseFileFolder, compiledFolderName)
dirEntries, err := os.ReadDir(binFileFolder)
Review comment:
We could throw `/path/to/workingDir/pipelinesFolder/{pipelineId}/bin` intead of `/path/to/workingDir/pipelinesFolder/{pipelineId}`
```suggestion
func executableName(execFileFolder string) (string, error) {
dirEntries, err := os.ReadDir(execFileFolder)
```
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +157,121 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func CompileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, executor executors.Executor, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return executors.Executor{}, true
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor = executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return executors.Executor{}, true
+ }
+ if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
+ err := <-errorChannel
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
+ return executors.Executor{}, true
+ } // Compile step is finished and code is compiled
+ if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
+ return executors.Executor{}, true
+ }
+ }
+ return executor, false
+}
+
+func PrepareStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupPreparatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return executors.Executor{}, false
Review comment:
Why you return `executors.Executor{}, false` but in all other place `executors.Executor{}, true`?
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +157,121 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func CompileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, executor executors.Executor, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return executors.Executor{}, true
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor = executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return executors.Executor{}, true
+ }
+ if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
+ err := <-errorChannel
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
+ return executors.Executor{}, true
+ } // Compile step is finished and code is compiled
+ if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
+ return executors.Executor{}, true
+ }
+ }
+ return executor, false
+}
+
+func PrepareStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupPreparatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return executors.Executor{}, false
+ }
+ executor := executorBuilder.Build()
+ // Prepare
+ logger.Infof("%s: Prepare() ...\n", pipelineId)
+ prepareFunc := executor.Prepare()
+ // Run prepare function
+ go prepareFunc(successChannel, errorChannel, validationResults)
+
+ // Start of the monitoring of background tasks (prepare function/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return executors.Executor{}, true
+ }
+ if !ok {
+ err := <-errorChannel
+ // Prepare step is finished, but code couldn't be prepared (some error during prepare step)
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.PreparationOutput, cacheService, "Prepare", pb.Status_STATUS_PREPARATION_ERROR)
+ return executors.Executor{}, true
+ }
+ // Prepare step is finished and code is prepared
+ if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Prepare", pb.Status_STATUS_COMPILING); err != nil {
+ return executors.Executor{}, true
+ }
+ return executor, false
+}
+
+func ValidateStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) (executors.Executor, bool) {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupValidatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return executors.Executor{}, false
Review comment:
ditto
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -61,117 +60,65 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
DeleteFolders(pipelineId, lc)
}(lc)
- errorChannel := make(chan error, 1)
- successChannel := make(chan bool, 1)
cancelChannel := make(chan bool, 1)
- stopReadLogsChannel := make(chan bool, 1)
- finishReadLogsChannel := make(chan bool, 1)
+
var validationResults sync.Map
go cancelCheck(pipelineLifeCycleCtx, pipelineId, cancelChannel, cacheService)
- executorBuilder, err := builder.SetupExecutorBuilder(lc, utils.ReduceWhiteSpacesToSinge(pipelineOptions), sdkEnv)
- if err != nil {
- _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ executor, done := ValidateStep(ctx, cacheService, lc, pipelineId, sdkEnv, pipelineLifeCycleCtx, &validationResults, cancelChannel)
+ if done {
return
}
- executor := executorBuilder.Build()
- // Validate
- logger.Infof("%s: Validate() ...\n", pipelineId)
- validateFunc := executor.Validate()
- // Run validate function
- go validateFunc(successChannel, errorChannel, &validationResults)
- // Start of the monitoring of background tasks (validate function/cancellation/timeout)
- ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
- if err != nil {
- return
- }
- if !ok {
- // Validate step is finished, but code isn't valid
- err := <-errorChannel
- _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.ValidationOutput, cacheService, "Validate", pb.Status_STATUS_VALIDATION_ERROR)
- return
- }
- // Validate step is finished and code is valid
- if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Validate", pb.Status_STATUS_PREPARING); err != nil {
+ executor, done = PrepareStep(ctx, cacheService, lc, pipelineId, sdkEnv, pipelineLifeCycleCtx, &validationResults, cancelChannel)
+ if done {
return
}
- // Prepare
- logger.Infof("%s: Prepare() ...\n", pipelineId)
- prepareFunc := executor.Prepare()
- // Run prepare function
- go prepareFunc(successChannel, errorChannel, &validationResults)
+ // Check if is unit test
+ validateIsUnitTest, _ := validationResults.Load(validators.UnitTestValidatorName)
+ isUnitTest := validateIsUnitTest.(bool)
- // Start of the monitoring of background tasks (prepare function/cancellation/timeout)
- ok, err = reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
- if err != nil {
- return
- }
- if !ok {
- // Prepare step is finished, but code couldn't be prepared (some error during prepare step)
- err := <-errorChannel
- _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.PreparationOutput, cacheService, "Prepare", pb.Status_STATUS_PREPARATION_ERROR)
- return
- }
- // Prepare step is finished and code is prepared
- if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Prepare", pb.Status_STATUS_COMPILING); err != nil {
+ executor, done = CompileStep(ctx, cacheService, lc, pipelineId, sdkEnv, isUnitTest, pipelineLifeCycleCtx, executor, cancelChannel)
+ if done {
return
}
- // Check if unit test
- validateIsUnitTest, _ := validationResults.Load(validators.UnitTestValidatorName)
- isUnitTest := validateIsUnitTest.(bool)
+ // Run/RunTest
+ RunStep(ctx, cacheService, lc, pipelineId, isUnitTest, sdkEnv, pipelineOptions, pipelineLifeCycleCtx, cancelChannel)
+}
- // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
- if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
- if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
- return
- }
- } else { // in case of Java, Go (not unit test), Scala - need compile step
- // Compile
- if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_JAVA {
- executor = executorBuilder.WithCompiler().
- WithFileName(builder.GetFileNameFromFolder(lc.GetAbsoluteSourceFolderPath())).Build() // Need changed name for unit tests
- }
- logger.Infof("%s: Compile() ...\n", pipelineId)
- compileCmd := executor.Compile(pipelineLifeCycleCtx)
- var compileError bytes.Buffer
- var compileOutput bytes.Buffer
- runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+func RunStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, isUnitTest bool, sdkEnv *environment.BeamEnvs, pipelineOptions string, pipelineLifeCycleCtx context.Context, cancelChannel chan bool) {
Review comment:
Why does this method name start with a capital letter?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] daria-malkova commented on a change in pull request #16477: [BEAM-13560][Playground] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
daria-malkova commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r784023314
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithCommand(executorConfig.RunCmd).
WithArgs(executorConfig.RunArgs).
WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
ExecutorBuilder
switch sdk {
- case pb.Sdk_SDK_JAVA: // Executable name for java class will be known after compilation
- args := make([]string, 0)
- for _, arg := range executorConfig.RunArgs {
- if strings.Contains(arg, javaLogConfigFilePlaceholder) {
- logConfigFilePath := filepath.Join(paths.AbsoluteBaseFolderPath, javaLogConfigFileName)
- arg = strings.Replace(arg, javaLogConfigFilePlaceholder, logConfigFilePath, 1)
- }
- args = append(args, arg)
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ args := replaceLogPlaceholder(paths, executorConfig)
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
}
- builder = builder.WithRunner().WithArgs(args).ExecutorBuilder
- builder = builder.WithTestRunner().WithWorkingDir(paths.AbsoluteBaseFolderPath).ExecutorBuilder //change directory for unit test
+ builder = builder.
+ WithRunner().
+ WithArgs(args).
+ WithExecutableFileName(className).
+ ExecutorBuilder
case pb.Sdk_SDK_GO: //go run command is executable file itself
builder = builder.
- WithExecutableFileName("").
WithRunner().
Review comment:
we need it
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] KhaninArtur commented on a change in pull request #16477: [BEAM-13560][Playground] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
KhaninArtur commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r783816072
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
Review comment:
Why do you pass here the pointer? I think we don't want to change anything in `paths`, let's remove `*`, what do you think?
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
Review comment:
ditto
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithCommand(executorConfig.RunCmd).
WithArgs(executorConfig.RunArgs).
WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
ExecutorBuilder
switch sdk {
- case pb.Sdk_SDK_JAVA: // Executable name for java class will be known after compilation
- args := make([]string, 0)
- for _, arg := range executorConfig.RunArgs {
- if strings.Contains(arg, javaLogConfigFilePlaceholder) {
- logConfigFilePath := filepath.Join(paths.AbsoluteBaseFolderPath, javaLogConfigFileName)
- arg = strings.Replace(arg, javaLogConfigFilePlaceholder, logConfigFilePath, 1)
- }
- args = append(args, arg)
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ args := replaceLogPlaceholder(paths, executorConfig)
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
}
- builder = builder.WithRunner().WithArgs(args).ExecutorBuilder
- builder = builder.WithTestRunner().WithWorkingDir(paths.AbsoluteBaseFolderPath).ExecutorBuilder //change directory for unit test
+ builder = builder.
+ WithRunner().
Review comment:
We already have `WithRunner()` at the beginning of the building of the executor, should we call it once more here?
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
Review comment:
ditto
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithCommand(executorConfig.RunCmd).
WithArgs(executorConfig.RunArgs).
WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
ExecutorBuilder
switch sdk {
- case pb.Sdk_SDK_JAVA: // Executable name for java class will be known after compilation
- args := make([]string, 0)
- for _, arg := range executorConfig.RunArgs {
- if strings.Contains(arg, javaLogConfigFilePlaceholder) {
- logConfigFilePath := filepath.Join(paths.AbsoluteBaseFolderPath, javaLogConfigFileName)
- arg = strings.Replace(arg, javaLogConfigFilePlaceholder, logConfigFilePath, 1)
- }
- args = append(args, arg)
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ args := replaceLogPlaceholder(paths, executorConfig)
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
}
- builder = builder.WithRunner().WithArgs(args).ExecutorBuilder
- builder = builder.WithTestRunner().WithWorkingDir(paths.AbsoluteBaseFolderPath).ExecutorBuilder //change directory for unit test
+ builder = builder.
+ WithRunner().
+ WithArgs(args).
+ WithExecutableFileName(className).
+ ExecutorBuilder
case pb.Sdk_SDK_GO: //go run command is executable file itself
builder = builder.
- WithExecutableFileName("").
WithRunner().
- WithCommand(paths.AbsoluteExecutableFilePath).ExecutorBuilder
+ WithExecutableFileName("").
+ WithCommand(paths.AbsoluteExecutableFilePath).
+ ExecutorBuilder
case pb.Sdk_SDK_PYTHON:
- builder = *builder.WithExecutableFileName(paths.AbsoluteExecutableFilePath)
- case pb.Sdk_SDK_SCIO:
- return nil, fmt.Errorf("SCIO is not supported yet")
- default:
- return nil, fmt.Errorf("incorrect sdk: %s", sdkEnv.ApacheBeamSdk)
+ builder = builder.
+ WithRunner().
+ WithExecutableFileName(paths.AbsoluteExecutableFilePath).
+ ExecutorBuilder
+ }
+ return &builder, nil
+}
+
+// TestRunner return executor with set args for runner
+func TestRunner(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
+ WithTestRunner().
+ WithExecutableFileName(paths.AbsoluteExecutableFilePath).
+ WithCommand(executorConfig.TestCmd).
+ WithArgs(executorConfig.TestArgs).
+ WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
+ }
+ builder = builder.WithTestRunner().
+ WithExecutableFileName(className).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
+ ExecutorBuilder //change directory for unit test
}
return &builder, nil
}
-// GetFileNameFromFolder return a name of the first file in a specified folder
-func GetFileNameFromFolder(folderAbsolutePath, extension string) string {
- files, _ := filepath.Glob(fmt.Sprintf("%s/*%s", folderAbsolutePath, extension))
+// replaceLogPlaceholder replaces placeholder for log for JAVA SDK
+func replaceLogPlaceholder(paths *fs_tool.LifeCyclePaths, executorConfig *environment.ExecutorConfig) []string {
Review comment:
ditto
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
Review comment:
Maybe, we can also remove `*` from `sdkEnv`?
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithCommand(executorConfig.RunCmd).
WithArgs(executorConfig.RunArgs).
WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
ExecutorBuilder
switch sdk {
- case pb.Sdk_SDK_JAVA: // Executable name for java class will be known after compilation
- args := make([]string, 0)
- for _, arg := range executorConfig.RunArgs {
- if strings.Contains(arg, javaLogConfigFilePlaceholder) {
- logConfigFilePath := filepath.Join(paths.AbsoluteBaseFolderPath, javaLogConfigFileName)
- arg = strings.Replace(arg, javaLogConfigFilePlaceholder, logConfigFilePath, 1)
- }
- args = append(args, arg)
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ args := replaceLogPlaceholder(paths, executorConfig)
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
}
- builder = builder.WithRunner().WithArgs(args).ExecutorBuilder
- builder = builder.WithTestRunner().WithWorkingDir(paths.AbsoluteBaseFolderPath).ExecutorBuilder //change directory for unit test
+ builder = builder.
+ WithRunner().
+ WithArgs(args).
+ WithExecutableFileName(className).
+ ExecutorBuilder
case pb.Sdk_SDK_GO: //go run command is executable file itself
builder = builder.
- WithExecutableFileName("").
WithRunner().
- WithCommand(paths.AbsoluteExecutableFilePath).ExecutorBuilder
+ WithExecutableFileName("").
+ WithCommand(paths.AbsoluteExecutableFilePath).
+ ExecutorBuilder
case pb.Sdk_SDK_PYTHON:
- builder = *builder.WithExecutableFileName(paths.AbsoluteExecutableFilePath)
- case pb.Sdk_SDK_SCIO:
- return nil, fmt.Errorf("SCIO is not supported yet")
- default:
- return nil, fmt.Errorf("incorrect sdk: %s", sdkEnv.ApacheBeamSdk)
+ builder = builder.
+ WithRunner().
Review comment:
ditto
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithCommand(executorConfig.RunCmd).
WithArgs(executorConfig.RunArgs).
WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
ExecutorBuilder
switch sdk {
- case pb.Sdk_SDK_JAVA: // Executable name for java class will be known after compilation
- args := make([]string, 0)
- for _, arg := range executorConfig.RunArgs {
- if strings.Contains(arg, javaLogConfigFilePlaceholder) {
- logConfigFilePath := filepath.Join(paths.AbsoluteBaseFolderPath, javaLogConfigFileName)
- arg = strings.Replace(arg, javaLogConfigFilePlaceholder, logConfigFilePath, 1)
- }
- args = append(args, arg)
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ args := replaceLogPlaceholder(paths, executorConfig)
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
}
- builder = builder.WithRunner().WithArgs(args).ExecutorBuilder
- builder = builder.WithTestRunner().WithWorkingDir(paths.AbsoluteBaseFolderPath).ExecutorBuilder //change directory for unit test
+ builder = builder.
+ WithRunner().
+ WithArgs(args).
+ WithExecutableFileName(className).
+ ExecutorBuilder
case pb.Sdk_SDK_GO: //go run command is executable file itself
builder = builder.
- WithExecutableFileName("").
WithRunner().
- WithCommand(paths.AbsoluteExecutableFilePath).ExecutorBuilder
+ WithExecutableFileName("").
+ WithCommand(paths.AbsoluteExecutableFilePath).
+ ExecutorBuilder
case pb.Sdk_SDK_PYTHON:
- builder = *builder.WithExecutableFileName(paths.AbsoluteExecutableFilePath)
- case pb.Sdk_SDK_SCIO:
- return nil, fmt.Errorf("SCIO is not supported yet")
- default:
- return nil, fmt.Errorf("incorrect sdk: %s", sdkEnv.ApacheBeamSdk)
+ builder = builder.
+ WithRunner().
+ WithExecutableFileName(paths.AbsoluteExecutableFilePath).
+ ExecutorBuilder
+ }
+ return &builder, nil
+}
+
+// TestRunner return executor with set args for runner
+func TestRunner(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
Review comment:
ditto
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
Review comment:
ditto
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithCommand(executorConfig.RunCmd).
WithArgs(executorConfig.RunArgs).
WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
ExecutorBuilder
switch sdk {
- case pb.Sdk_SDK_JAVA: // Executable name for java class will be known after compilation
- args := make([]string, 0)
- for _, arg := range executorConfig.RunArgs {
- if strings.Contains(arg, javaLogConfigFilePlaceholder) {
- logConfigFilePath := filepath.Join(paths.AbsoluteBaseFolderPath, javaLogConfigFileName)
- arg = strings.Replace(arg, javaLogConfigFilePlaceholder, logConfigFilePath, 1)
- }
- args = append(args, arg)
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ args := replaceLogPlaceholder(paths, executorConfig)
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
}
- builder = builder.WithRunner().WithArgs(args).ExecutorBuilder
- builder = builder.WithTestRunner().WithWorkingDir(paths.AbsoluteBaseFolderPath).ExecutorBuilder //change directory for unit test
+ builder = builder.
+ WithRunner().
+ WithArgs(args).
+ WithExecutableFileName(className).
+ ExecutorBuilder
case pb.Sdk_SDK_GO: //go run command is executable file itself
builder = builder.
- WithExecutableFileName("").
WithRunner().
Review comment:
ditto
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] daria-malkova commented on a change in pull request #16477: [BEAM-13560][Playground] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
daria-malkova commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r784027688
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithCommand(executorConfig.RunCmd).
WithArgs(executorConfig.RunArgs).
WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
ExecutorBuilder
switch sdk {
- case pb.Sdk_SDK_JAVA: // Executable name for java class will be known after compilation
- args := make([]string, 0)
- for _, arg := range executorConfig.RunArgs {
- if strings.Contains(arg, javaLogConfigFilePlaceholder) {
- logConfigFilePath := filepath.Join(paths.AbsoluteBaseFolderPath, javaLogConfigFileName)
- arg = strings.Replace(arg, javaLogConfigFilePlaceholder, logConfigFilePath, 1)
- }
- args = append(args, arg)
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ args := replaceLogPlaceholder(paths, executorConfig)
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
}
- builder = builder.WithRunner().WithArgs(args).ExecutorBuilder
- builder = builder.WithTestRunner().WithWorkingDir(paths.AbsoluteBaseFolderPath).ExecutorBuilder //change directory for unit test
+ builder = builder.
+ WithRunner().
+ WithArgs(args).
+ WithExecutableFileName(className).
+ ExecutorBuilder
case pb.Sdk_SDK_GO: //go run command is executable file itself
builder = builder.
- WithExecutableFileName("").
WithRunner().
- WithCommand(paths.AbsoluteExecutableFilePath).ExecutorBuilder
+ WithExecutableFileName("").
+ WithCommand(paths.AbsoluteExecutableFilePath).
+ ExecutorBuilder
case pb.Sdk_SDK_PYTHON:
- builder = *builder.WithExecutableFileName(paths.AbsoluteExecutableFilePath)
- case pb.Sdk_SDK_SCIO:
- return nil, fmt.Errorf("SCIO is not supported yet")
- default:
- return nil, fmt.Errorf("incorrect sdk: %s", sdkEnv.ApacheBeamSdk)
+ builder = builder.
+ WithRunner().
Review comment:
same
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithCommand(executorConfig.RunCmd).
WithArgs(executorConfig.RunArgs).
WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
ExecutorBuilder
switch sdk {
- case pb.Sdk_SDK_JAVA: // Executable name for java class will be known after compilation
- args := make([]string, 0)
- for _, arg := range executorConfig.RunArgs {
- if strings.Contains(arg, javaLogConfigFilePlaceholder) {
- logConfigFilePath := filepath.Join(paths.AbsoluteBaseFolderPath, javaLogConfigFileName)
- arg = strings.Replace(arg, javaLogConfigFilePlaceholder, logConfigFilePath, 1)
- }
- args = append(args, arg)
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ args := replaceLogPlaceholder(paths, executorConfig)
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
}
- builder = builder.WithRunner().WithArgs(args).ExecutorBuilder
- builder = builder.WithTestRunner().WithWorkingDir(paths.AbsoluteBaseFolderPath).ExecutorBuilder //change directory for unit test
+ builder = builder.
+ WithRunner().
+ WithArgs(args).
+ WithExecutableFileName(className).
+ ExecutorBuilder
case pb.Sdk_SDK_GO: //go run command is executable file itself
builder = builder.
- WithExecutableFileName("").
WithRunner().
- WithCommand(paths.AbsoluteExecutableFilePath).ExecutorBuilder
+ WithExecutableFileName("").
+ WithCommand(paths.AbsoluteExecutableFilePath).
+ ExecutorBuilder
case pb.Sdk_SDK_PYTHON:
- builder = *builder.WithExecutableFileName(paths.AbsoluteExecutableFilePath)
- case pb.Sdk_SDK_SCIO:
- return nil, fmt.Errorf("SCIO is not supported yet")
- default:
- return nil, fmt.Errorf("incorrect sdk: %s", sdkEnv.ApacheBeamSdk)
+ builder = builder.
+ WithRunner().
+ WithExecutableFileName(paths.AbsoluteExecutableFilePath).
+ ExecutorBuilder
+ }
+ return &builder, nil
+}
+
+// TestRunner return executor with set args for runner
+func TestRunner(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
+ WithTestRunner().
+ WithExecutableFileName(paths.AbsoluteExecutableFilePath).
+ WithCommand(executorConfig.TestCmd).
+ WithArgs(executorConfig.TestArgs).
+ WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
+ }
+ builder = builder.WithTestRunner().
+ WithExecutableFileName(className).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
+ ExecutorBuilder //change directory for unit test
}
return &builder, nil
}
-// GetFileNameFromFolder return a name of the first file in a specified folder
-func GetFileNameFromFolder(folderAbsolutePath, extension string) string {
- files, _ := filepath.Glob(fmt.Sprintf("%s/*%s", folderAbsolutePath, extension))
+// replaceLogPlaceholder replaces placeholder for log for JAVA SDK
+func replaceLogPlaceholder(paths *fs_tool.LifeCyclePaths, executorConfig *environment.ExecutorConfig) []string {
Review comment:
same
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithCommand(executorConfig.RunCmd).
WithArgs(executorConfig.RunArgs).
WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
ExecutorBuilder
switch sdk {
- case pb.Sdk_SDK_JAVA: // Executable name for java class will be known after compilation
- args := make([]string, 0)
- for _, arg := range executorConfig.RunArgs {
- if strings.Contains(arg, javaLogConfigFilePlaceholder) {
- logConfigFilePath := filepath.Join(paths.AbsoluteBaseFolderPath, javaLogConfigFileName)
- arg = strings.Replace(arg, javaLogConfigFilePlaceholder, logConfigFilePath, 1)
- }
- args = append(args, arg)
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ args := replaceLogPlaceholder(paths, executorConfig)
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
}
- builder = builder.WithRunner().WithArgs(args).ExecutorBuilder
- builder = builder.WithTestRunner().WithWorkingDir(paths.AbsoluteBaseFolderPath).ExecutorBuilder //change directory for unit test
+ builder = builder.
+ WithRunner().
+ WithArgs(args).
+ WithExecutableFileName(className).
+ ExecutorBuilder
case pb.Sdk_SDK_GO: //go run command is executable file itself
builder = builder.
- WithExecutableFileName("").
WithRunner().
- WithCommand(paths.AbsoluteExecutableFilePath).ExecutorBuilder
+ WithExecutableFileName("").
+ WithCommand(paths.AbsoluteExecutableFilePath).
+ ExecutorBuilder
case pb.Sdk_SDK_PYTHON:
- builder = *builder.WithExecutableFileName(paths.AbsoluteExecutableFilePath)
- case pb.Sdk_SDK_SCIO:
- return nil, fmt.Errorf("SCIO is not supported yet")
- default:
- return nil, fmt.Errorf("incorrect sdk: %s", sdkEnv.ApacheBeamSdk)
+ builder = builder.
+ WithRunner().
+ WithExecutableFileName(paths.AbsoluteExecutableFilePath).
+ ExecutorBuilder
+ }
+ return &builder, nil
+}
+
+// TestRunner return executor with set args for runner
+func TestRunner(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
Review comment:
same
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
Review comment:
next PR
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem merged pull request #16477: [BEAM-13560][Playground] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
pabloem merged pull request #16477:
URL: https://github.com/apache/beam/pull/16477
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] daria-malkova commented on pull request #16477: [BEAM-13560] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
daria-malkova commented on pull request #16477:
URL: https://github.com/apache/beam/pull/16477#issuecomment-1010879852
@KhaninArtur @ilya-kozyrev
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] KhaninArtur commented on a change in pull request #16477: [BEAM-13560] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
KhaninArtur commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r782935851
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +154,122 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func compileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ var executor = executors.Executor{}
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor := executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return nil
+ }
+ if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
+ err := <-errorChannel
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
+ return nil
+ } // Compile step is finished and code is compiled
+ if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ }
+ return &executor
+}
+
+func prepareStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupPreparatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return nil
+ }
+ executor := executorBuilder.Build()
+ // Prepare
+ logger.Infof("%s: Prepare() ...\n", pipelineId)
+ prepareFunc := executor.Prepare()
+ // Run prepare function
+ go prepareFunc(successChannel, errorChannel, validationResults)
+
+ // Start of the monitoring of background tasks (prepare function/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return nil
+ }
+ if !ok {
+ err := <-errorChannel
+ // Prepare step is finished, but code couldn't be prepared (some error during prepare step)
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.PreparationOutput, cacheService, "Prepare", pb.Status_STATUS_PREPARATION_ERROR)
+ return nil
+ }
+ // Prepare step is finished and code is prepared
+ if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Prepare", pb.Status_STATUS_COMPILING); err != nil {
+ return nil
+ }
+ return &executor
+}
+
+func validateStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupValidatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return nil
+ }
+ executor := executorBuilder.Build()
+ // Validate
+ logger.Infof("%s: Validate() ...\n", pipelineId)
+ validateFunc := executor.Validate()
+ // Run validate function
Review comment:
ditto
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder_test.go
##########
@@ -23,100 +23,293 @@ import (
"beam.apache.org/playground/backend/internal/utils"
"fmt"
"github.com/google/uuid"
+ "reflect"
"strings"
"testing"
)
-func TestSetupExecutor(t *testing.T) {
+var lc *fs_tool.LifeCycle
+var sdkEnv *environment.BeamEnvs
+
+func TestMain(m *testing.M) {
+ setup()
+ m.Run()
+}
+
+func setup() {
pipelineId := uuid.New()
- sdk := pb.Sdk_SDK_JAVA
- lc, err := fs_tool.NewLifeCycle(sdk, pipelineId, "")
- if err != nil {
- t.Error(err)
- }
- pipelineOptions := ""
+ sdk := pb.Sdk_SDK_PYTHON
+ lc, _ = fs_tool.NewLifeCycle(sdk, pipelineId, "")
executorConfig := &environment.ExecutorConfig{
CompileCmd: "MOCK_COMPILE_CMD",
- RunCmd: "MOCK_RUN_CMD",
- TestCmd: "MOCK_TEST_CMD",
CompileArgs: []string{"MOCK_COMPILE_ARG"},
- RunArgs: []string{"MOCK_RUN_ARG"},
- TestArgs: []string{"MOCK_TEST_ARG"},
}
+ sdkEnv = environment.NewBeamEnvs(sdk, executorConfig, "", 0)
+}
+
+func TestSetupValidatorBuilder(t *testing.T) {
+ vals, err := utils.GetValidators(sdkEnv.ApacheBeamSdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
panic(err)
}
+ wantExecutor := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(vals)
- srcFilePath := lc.GetAbsoluteSourceFilePath()
+ wrongSdkEnv := environment.NewBeamEnvs(pb.Sdk_SDK_UNSPECIFIED, sdkEnv.ExecutorConfig, "", 0)
- sdkEnv := environment.NewBeamEnvs(sdk, executorConfig, "", 0)
- val, err := utils.GetValidators(sdk, srcFilePath)
- if err != nil {
- panic(err)
+ type args struct {
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ wantErr bool
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected validator builder.
+ name: "Test correct validator builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ wantErr: false,
+ },
+ {
+ // Test case with calling Setup with incorrect SDK.
+ // As a result, want to receive an error.
+ name: "incorrect sdk",
+ args: args{
+ lc: lc,
+ sdkEnv: wrongSdkEnv,
+ },
+ want: nil,
+ wantErr: true,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, err := SetupValidatorBuilder(tt.args.lc, tt.args.sdkEnv)
+ if (err != nil) != tt.wantErr {
+ t.Errorf("SetupValidatorBuilder() error = %v, wantErr %v", err, tt.wantErr)
+ return
+ }
+ if err != nil && !reflect.DeepEqual(got, tt.want) {
+ t.Errorf("SetupValidatorBuilder() got = %v, want %v", got, tt.want)
+ return
+ }
+ if err == nil && !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupValidatorBuilder() got = %v\n, want %v", got.Build(), tt.want.Build())
+ }
+ })
}
- prep, err := utils.GetPreparators(sdk, srcFilePath)
+}
+
+func TestSetupPreparatorBuilder(t *testing.T) {
+ prep, err := utils.GetPreparators(sdkEnv.ApacheBeamSdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
panic(err)
}
-
wantExecutor := executors.NewExecutorBuilder().
- WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
- WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
- WithSdkPreparators(prep).
- WithCompiler().
- WithCommand(executorConfig.CompileCmd).
- WithArgs(executorConfig.CompileArgs).
- WithFileName(srcFilePath).
- WithRunner().
- WithCommand(executorConfig.RunCmd).
- WithArgs(executorConfig.RunArgs).
- WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
- ExecutorBuilder
+ WithSdkPreparators(prep)
+
+ wrongSdkEnv := environment.NewBeamEnvs(pb.Sdk_SDK_UNSPECIFIED, sdkEnv.ExecutorConfig, "", 0)
type args struct {
- lc *fs_tool.LifeCycle
- pipelineOptions string
- sdkEnv *environment.BeamEnvs
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
}
tests := []struct {
name string
args args
want *executors.ExecutorBuilder
wantErr bool
}{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected preparator builder.
+ name: "Test correct preparator builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ wantErr: false,
+ },
{
// Test case with calling Setup with incorrect SDK.
// As a result, want to receive an error.
- name: "incorrect sdk",
- args: args{lc, pipelineOptions, environment.NewBeamEnvs(pb.Sdk_SDK_UNSPECIFIED, executorConfig, "", 0)},
+ name: "incorrect sdk",
+ args: args{
+ lc: lc,
+ sdkEnv: wrongSdkEnv,
+ },
want: nil,
wantErr: true,
},
- {
- // Test case with calling Setup with correct SDK.
- // As a result, want to receive an expected builder.
- name: "correct sdk",
- args: args{lc, pipelineOptions, sdkEnv},
- want: &wantExecutor,
- wantErr: false,
- },
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- got, err := SetupExecutorBuilder(tt.args.lc, tt.args.pipelineOptions, tt.args.sdkEnv)
+ got, err := SetupPreparatorBuilder(tt.args.lc, tt.args.sdkEnv)
if (err != nil) != tt.wantErr {
- t.Errorf("SetupExecutorBuilder() error = %v, wantErr %v", err, tt.wantErr)
+ t.Errorf("SetupPreparatorBuilder() error = %v, wantErr %v", err, tt.wantErr)
return
}
- if err == nil && fmt.Sprint(got.Build()) != fmt.Sprint(tt.want.Build()) {
- t.Errorf("SetupExecutorBuilder() got = %v\n, want %v", got.Build(), tt.want.Build())
+ if err != nil && !reflect.DeepEqual(got, tt.want) {
+ t.Errorf("SetupPreparatorBuilder() got = %v, want %v", got, tt.want)
+ return
+ }
+ if err == nil && !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupPreparatorBuilder() got = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func TestSetupCompilerBuilder(t *testing.T) {
+ wantExecutor := executors.NewExecutorBuilder().
+ WithCompiler().
+ WithCommand(sdkEnv.ExecutorConfig.CompileCmd).
+ WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
+ WithArgs(sdkEnv.ExecutorConfig.CompileArgs).
+ WithFileName(lc.GetAbsoluteSourceFilePath())
+
+ type args struct {
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected compiler builder.
+ name: "Test correct compiler builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got := SetupCompilerBuilder(tt.args.lc, tt.args.sdkEnv)
+ if !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupCompilerBuilder() = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func TestSetupRunBuilder(t *testing.T) {
+ wantExecutor := executors.NewExecutorBuilder().
+ WithRunner().
+ WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
+ WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
+ WithCommand(sdkEnv.ExecutorConfig.RunCmd).
+ WithArgs(sdkEnv.ExecutorConfig.RunArgs).
+ WithPipelineOptions(strings.Split("", " "))
+
+ type args struct {
+ lc *fs_tool.LifeCycle
+ pipelineOptions string
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected run builder.
+ name: "Test correct run builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, _ := SetupRunBuilder(tt.args.lc, tt.args.pipelineOptions, tt.args.sdkEnv)
+ if !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupRunBuilder() got = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func TestSetupTestBuilder(t *testing.T) {
+ wantExecutor := executors.NewExecutorBuilder().
+ WithTestRunner().
+ WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
+ WithCommand(sdkEnv.ExecutorConfig.TestCmd).
+ WithArgs(sdkEnv.ExecutorConfig.TestArgs).
+ WithWorkingDir(lc.GetAbsoluteSourceFolderPath())
+
+ type args struct {
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected test builder.
+ name: "Test correct test builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, _ := SetupTestBuilder(tt.args.lc, tt.args.sdkEnv)
+ if !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupTestBuilder() got = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func Test_replaceLogPlaceholder(t *testing.T) {
+ type args struct {
+ lc *fs_tool.LifeCycle
+ executorConfig *environment.ExecutorConfig
+ }
+ tests := []struct {
+ name string
+ args args
+ want []string
+ }{
+ {
+ name: "",
Review comment:
Could you please give a name for this test?
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,134 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(lc *fs_tool.LifeCycle, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// SetupValidatorBuilder return executor with set args for validator
+func SetupValidatorBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// SetupPreparatorBuilder return executor with set args for preparator
+func SetupPreparatorBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
- WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// SetupCompilerBuilder return executor with set args for compiler
+func SetupCompilerBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
WithArgs(executorConfig.CompileArgs).
WithFileName(lc.GetAbsoluteSourceFilePath()).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(lc.GetAbsoluteSourceFolderPath()))
+ }
+ return &builder
+}
+
+// SetupRunBuilder return executor with set args for runner
+func SetupRunBuilder(lc *fs_tool.LifeCycle, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
Review comment:
Shouldn't we move this to the `pb.Sdk_SDK_PYTHON` case to keep consistency?
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +154,122 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func compileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ var executor = executors.Executor{}
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor := executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return nil
+ }
+ if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
+ err := <-errorChannel
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
+ return nil
+ } // Compile step is finished and code is compiled
+ if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ }
+ return &executor
+}
+
+func prepareStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupPreparatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return nil
+ }
+ executor := executorBuilder.Build()
+ // Prepare
+ logger.Infof("%s: Prepare() ...\n", pipelineId)
+ prepareFunc := executor.Prepare()
+ // Run prepare function
+ go prepareFunc(successChannel, errorChannel, validationResults)
+
+ // Start of the monitoring of background tasks (prepare function/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return nil
+ }
+ if !ok {
+ err := <-errorChannel
+ // Prepare step is finished, but code couldn't be prepared (some error during prepare step)
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.PreparationOutput, cacheService, "Prepare", pb.Status_STATUS_PREPARATION_ERROR)
+ return nil
+ }
+ // Prepare step is finished and code is prepared
+ if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Prepare", pb.Status_STATUS_COMPILING); err != nil {
+ return nil
+ }
+ return &executor
+}
+
+func validateStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupValidatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return nil
+ }
+ executor := executorBuilder.Build()
+ // Validate
Review comment:
This comment is "extremely useful", let's remove it๐
##########
File path: playground/backend/internal/code_processing/code_processing_test.go
##########
@@ -532,64 +526,7 @@ func TestGetLastIndex(t *testing.T) {
}
}
-func Test_setJavaExecutableFile(t *testing.T) {
Review comment:
I believe, we also need to test all introduced functions like `validateStep`, `compileStep`, etc.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] daria-malkova commented on a change in pull request #16477: [BEAM-13560] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
daria-malkova commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r783072505
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder_test.go
##########
@@ -23,100 +23,293 @@ import (
"beam.apache.org/playground/backend/internal/utils"
"fmt"
"github.com/google/uuid"
+ "reflect"
"strings"
"testing"
)
-func TestSetupExecutor(t *testing.T) {
+var lc *fs_tool.LifeCycle
+var sdkEnv *environment.BeamEnvs
+
+func TestMain(m *testing.M) {
+ setup()
+ m.Run()
+}
+
+func setup() {
pipelineId := uuid.New()
- sdk := pb.Sdk_SDK_JAVA
- lc, err := fs_tool.NewLifeCycle(sdk, pipelineId, "")
- if err != nil {
- t.Error(err)
- }
- pipelineOptions := ""
+ sdk := pb.Sdk_SDK_PYTHON
+ lc, _ = fs_tool.NewLifeCycle(sdk, pipelineId, "")
executorConfig := &environment.ExecutorConfig{
CompileCmd: "MOCK_COMPILE_CMD",
- RunCmd: "MOCK_RUN_CMD",
- TestCmd: "MOCK_TEST_CMD",
CompileArgs: []string{"MOCK_COMPILE_ARG"},
- RunArgs: []string{"MOCK_RUN_ARG"},
- TestArgs: []string{"MOCK_TEST_ARG"},
}
+ sdkEnv = environment.NewBeamEnvs(sdk, executorConfig, "", 0)
+}
+
+func TestSetupValidatorBuilder(t *testing.T) {
+ vals, err := utils.GetValidators(sdkEnv.ApacheBeamSdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
panic(err)
}
+ wantExecutor := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(vals)
- srcFilePath := lc.GetAbsoluteSourceFilePath()
+ wrongSdkEnv := environment.NewBeamEnvs(pb.Sdk_SDK_UNSPECIFIED, sdkEnv.ExecutorConfig, "", 0)
- sdkEnv := environment.NewBeamEnvs(sdk, executorConfig, "", 0)
- val, err := utils.GetValidators(sdk, srcFilePath)
- if err != nil {
- panic(err)
+ type args struct {
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ wantErr bool
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected validator builder.
+ name: "Test correct validator builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ wantErr: false,
+ },
+ {
+ // Test case with calling Setup with incorrect SDK.
+ // As a result, want to receive an error.
+ name: "incorrect sdk",
+ args: args{
+ lc: lc,
+ sdkEnv: wrongSdkEnv,
+ },
+ want: nil,
+ wantErr: true,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, err := SetupValidatorBuilder(tt.args.lc, tt.args.sdkEnv)
+ if (err != nil) != tt.wantErr {
+ t.Errorf("SetupValidatorBuilder() error = %v, wantErr %v", err, tt.wantErr)
+ return
+ }
+ if err != nil && !reflect.DeepEqual(got, tt.want) {
+ t.Errorf("SetupValidatorBuilder() got = %v, want %v", got, tt.want)
+ return
+ }
+ if err == nil && !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupValidatorBuilder() got = %v\n, want %v", got.Build(), tt.want.Build())
+ }
+ })
}
- prep, err := utils.GetPreparators(sdk, srcFilePath)
+}
+
+func TestSetupPreparatorBuilder(t *testing.T) {
+ prep, err := utils.GetPreparators(sdkEnv.ApacheBeamSdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
panic(err)
}
-
wantExecutor := executors.NewExecutorBuilder().
- WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
- WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
- WithSdkPreparators(prep).
- WithCompiler().
- WithCommand(executorConfig.CompileCmd).
- WithArgs(executorConfig.CompileArgs).
- WithFileName(srcFilePath).
- WithRunner().
- WithCommand(executorConfig.RunCmd).
- WithArgs(executorConfig.RunArgs).
- WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
- ExecutorBuilder
+ WithSdkPreparators(prep)
+
+ wrongSdkEnv := environment.NewBeamEnvs(pb.Sdk_SDK_UNSPECIFIED, sdkEnv.ExecutorConfig, "", 0)
type args struct {
- lc *fs_tool.LifeCycle
- pipelineOptions string
- sdkEnv *environment.BeamEnvs
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
}
tests := []struct {
name string
args args
want *executors.ExecutorBuilder
wantErr bool
}{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected preparator builder.
+ name: "Test correct preparator builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ wantErr: false,
+ },
{
// Test case with calling Setup with incorrect SDK.
// As a result, want to receive an error.
- name: "incorrect sdk",
- args: args{lc, pipelineOptions, environment.NewBeamEnvs(pb.Sdk_SDK_UNSPECIFIED, executorConfig, "", 0)},
+ name: "incorrect sdk",
+ args: args{
+ lc: lc,
+ sdkEnv: wrongSdkEnv,
+ },
want: nil,
wantErr: true,
},
- {
- // Test case with calling Setup with correct SDK.
- // As a result, want to receive an expected builder.
- name: "correct sdk",
- args: args{lc, pipelineOptions, sdkEnv},
- want: &wantExecutor,
- wantErr: false,
- },
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
- got, err := SetupExecutorBuilder(tt.args.lc, tt.args.pipelineOptions, tt.args.sdkEnv)
+ got, err := SetupPreparatorBuilder(tt.args.lc, tt.args.sdkEnv)
if (err != nil) != tt.wantErr {
- t.Errorf("SetupExecutorBuilder() error = %v, wantErr %v", err, tt.wantErr)
+ t.Errorf("SetupPreparatorBuilder() error = %v, wantErr %v", err, tt.wantErr)
return
}
- if err == nil && fmt.Sprint(got.Build()) != fmt.Sprint(tt.want.Build()) {
- t.Errorf("SetupExecutorBuilder() got = %v\n, want %v", got.Build(), tt.want.Build())
+ if err != nil && !reflect.DeepEqual(got, tt.want) {
+ t.Errorf("SetupPreparatorBuilder() got = %v, want %v", got, tt.want)
+ return
+ }
+ if err == nil && !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupPreparatorBuilder() got = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func TestSetupCompilerBuilder(t *testing.T) {
+ wantExecutor := executors.NewExecutorBuilder().
+ WithCompiler().
+ WithCommand(sdkEnv.ExecutorConfig.CompileCmd).
+ WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
+ WithArgs(sdkEnv.ExecutorConfig.CompileArgs).
+ WithFileName(lc.GetAbsoluteSourceFilePath())
+
+ type args struct {
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected compiler builder.
+ name: "Test correct compiler builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got := SetupCompilerBuilder(tt.args.lc, tt.args.sdkEnv)
+ if !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupCompilerBuilder() = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func TestSetupRunBuilder(t *testing.T) {
+ wantExecutor := executors.NewExecutorBuilder().
+ WithRunner().
+ WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
+ WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
+ WithCommand(sdkEnv.ExecutorConfig.RunCmd).
+ WithArgs(sdkEnv.ExecutorConfig.RunArgs).
+ WithPipelineOptions(strings.Split("", " "))
+
+ type args struct {
+ lc *fs_tool.LifeCycle
+ pipelineOptions string
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected run builder.
+ name: "Test correct run builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, _ := SetupRunBuilder(tt.args.lc, tt.args.pipelineOptions, tt.args.sdkEnv)
+ if !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupRunBuilder() got = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func TestSetupTestBuilder(t *testing.T) {
+ wantExecutor := executors.NewExecutorBuilder().
+ WithTestRunner().
+ WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
+ WithCommand(sdkEnv.ExecutorConfig.TestCmd).
+ WithArgs(sdkEnv.ExecutorConfig.TestArgs).
+ WithWorkingDir(lc.GetAbsoluteSourceFolderPath())
+
+ type args struct {
+ lc *fs_tool.LifeCycle
+ sdkEnv *environment.BeamEnvs
+ }
+ tests := []struct {
+ name string
+ args args
+ want *executors.ExecutorBuilder
+ }{
+ {
+ // Test case with calling Setup with correct data.
+ // As a result, want to receive an expected test builder.
+ name: "Test correct test builder",
+ args: args{
+ lc: lc,
+ sdkEnv: sdkEnv,
+ },
+ want: &wantExecutor.ExecutorBuilder,
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, _ := SetupTestBuilder(tt.args.lc, tt.args.sdkEnv)
+ if !reflect.DeepEqual(fmt.Sprint(got.Build()), fmt.Sprint(tt.want.Build())) {
+ t.Errorf("SetupTestBuilder() got = %v, want %v", got.Build(), tt.want.Build())
+ }
+ })
+ }
+}
+
+func Test_replaceLogPlaceholder(t *testing.T) {
+ type args struct {
+ lc *fs_tool.LifeCycle
+ executorConfig *environment.ExecutorConfig
+ }
+ tests := []struct {
+ name string
+ args args
+ want []string
+ }{
+ {
+ name: "",
Review comment:
I will :)
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +154,122 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func compileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ var executor = executors.Executor{}
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor := executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return nil
+ }
+ if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
+ err := <-errorChannel
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
+ return nil
+ } // Compile step is finished and code is compiled
+ if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ }
+ return &executor
+}
+
+func prepareStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupPreparatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return nil
+ }
+ executor := executorBuilder.Build()
+ // Prepare
+ logger.Infof("%s: Prepare() ...\n", pipelineId)
+ prepareFunc := executor.Prepare()
+ // Run prepare function
+ go prepareFunc(successChannel, errorChannel, validationResults)
+
+ // Start of the monitoring of background tasks (prepare function/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return nil
+ }
+ if !ok {
+ err := <-errorChannel
+ // Prepare step is finished, but code couldn't be prepared (some error during prepare step)
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.PreparationOutput, cacheService, "Prepare", pb.Status_STATUS_PREPARATION_ERROR)
+ return nil
+ }
+ // Prepare step is finished and code is prepared
+ if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Prepare", pb.Status_STATUS_COMPILING); err != nil {
+ return nil
+ }
+ return &executor
+}
+
+func validateStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupValidatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return nil
+ }
+ executor := executorBuilder.Build()
+ // Validate
Review comment:
Done
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +154,122 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func compileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ var executor = executors.Executor{}
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor := executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return nil
+ }
+ if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
+ err := <-errorChannel
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
+ return nil
+ } // Compile step is finished and code is compiled
+ if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ }
+ return &executor
+}
+
+func prepareStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupPreparatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return nil
+ }
+ executor := executorBuilder.Build()
+ // Prepare
Review comment:
No, done
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,134 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(lc *fs_tool.LifeCycle, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// SetupValidatorBuilder return executor with set args for validator
+func SetupValidatorBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// SetupPreparatorBuilder return executor with set args for preparator
+func SetupPreparatorBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, lc.GetAbsoluteSourceFilePath())
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(lc.GetAbsoluteExecutableFilePath()).
- WithWorkingDir(lc.GetAbsoluteBaseFolderPath()).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// SetupCompilerBuilder return executor with set args for compiler
+func SetupCompilerBuilder(lc *fs_tool.LifeCycle, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
Review comment:
Done
##########
File path: playground/backend/internal/code_processing/code_processing.go
##########
@@ -210,30 +154,122 @@ func Process(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycl
_ = processRunSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, stopReadLogsChannel, finishReadLogsChannel)
}
+func compileStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, isUnitTest bool, pipelineLifeCycleCtx context.Context, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ var executor = executors.Executor{}
+ // This condition is used for cases when the playground doesn't compile source files. For the Python code and the Go Unit Tests
+ if sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_PYTHON || (sdkEnv.ApacheBeamSdk == pb.Sdk_SDK_GO && isUnitTest) {
+ if err := processCompileSuccess(pipelineLifeCycleCtx, []byte(""), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ } else { // in case of Java, Go (not unit test), Scala - need compile step
+ // Compile
+ executorBuilder := builder.SetupCompilerBuilder(lc, sdkEnv)
+ executor := executorBuilder.Build()
+ logger.Infof("%s: Compile() ...\n", pipelineId)
+ compileCmd := executor.Compile(pipelineLifeCycleCtx)
+ var compileError bytes.Buffer
+ var compileOutput bytes.Buffer
+ runCmdWithOutput(compileCmd, &compileOutput, &compileError, successChannel, errorChannel)
+
+ // Start of the monitoring of background tasks (compile step/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return nil
+ }
+ if !ok { // Compile step is finished, but code couldn't be compiled (some typos for example)
+ err := <-errorChannel
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, compileError.Bytes(), pipelineId, cache.CompileOutput, cacheService, "Compile", pb.Status_STATUS_COMPILE_ERROR)
+ return nil
+ } // Compile step is finished and code is compiled
+ if err := processCompileSuccess(pipelineLifeCycleCtx, compileOutput.Bytes(), pipelineId, cacheService); err != nil {
+ return nil
+ }
+ }
+ return &executor
+}
+
+func prepareStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupPreparatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return nil
+ }
+ executor := executorBuilder.Build()
+ // Prepare
+ logger.Infof("%s: Prepare() ...\n", pipelineId)
+ prepareFunc := executor.Prepare()
+ // Run prepare function
+ go prepareFunc(successChannel, errorChannel, validationResults)
+
+ // Start of the monitoring of background tasks (prepare function/cancellation/timeout)
+ ok, err := reconcileBackgroundTask(pipelineLifeCycleCtx, ctx, pipelineId, cacheService, cancelChannel, successChannel)
+ if err != nil {
+ return nil
+ }
+ if !ok {
+ err := <-errorChannel
+ // Prepare step is finished, but code couldn't be prepared (some error during prepare step)
+ _ = processErrorWithSavingOutput(pipelineLifeCycleCtx, err, []byte(err.Error()), pipelineId, cache.PreparationOutput, cacheService, "Prepare", pb.Status_STATUS_PREPARATION_ERROR)
+ return nil
+ }
+ // Prepare step is finished and code is prepared
+ if err := processSuccess(pipelineLifeCycleCtx, pipelineId, cacheService, "Prepare", pb.Status_STATUS_COMPILING); err != nil {
+ return nil
+ }
+ return &executor
+}
+
+func validateStep(ctx context.Context, cacheService cache.Cache, lc *fs_tool.LifeCycle, pipelineId uuid.UUID, sdkEnv *environment.BeamEnvs, pipelineLifeCycleCtx context.Context, validationResults *sync.Map, cancelChannel chan bool) *executors.Executor {
+ errorChannel, successChannel := createStatusChannels()
+ executorBuilder, err := builder.SetupValidatorBuilder(lc, sdkEnv)
+ if err != nil {
+ _ = processSetupError(err, pipelineId, cacheService, pipelineLifeCycleCtx)
+ return nil
+ }
+ executor := executorBuilder.Build()
+ // Validate
+ logger.Infof("%s: Validate() ...\n", pipelineId)
+ validateFunc := executor.Validate()
+ // Run validate function
Review comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] daria-malkova commented on a change in pull request #16477: [BEAM-13560] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
daria-malkova commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r783223963
##########
File path: playground/backend/internal/code_processing/code_processing_test.go
##########
@@ -532,64 +526,7 @@ func TestGetLastIndex(t *testing.T) {
}
}
-func Test_setJavaExecutableFile(t *testing.T) {
Review comment:
Done
##########
File path: playground/backend/internal/fs_tool/java_fs.go
##########
@@ -36,8 +36,7 @@ func newJavaLifeCycle(pipelineId uuid.UUID, pipelinesFolder string) *LifeCycle {
}
// executableName returns name that should be executed (HelloWorld for HelloWorld.class for java SDK)
-func executableName(pipelineId uuid.UUID, pipelinesFolder string) (string, error) {
- baseFileFolder := filepath.Join(pipelinesFolder, pipelineId.String())
+func executableName(baseFileFolder string) (string, error) {
binFileFolder := filepath.Join(baseFileFolder, compiledFolderName)
dirEntries, err := os.ReadDir(binFileFolder)
Review comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] daria-malkova commented on a change in pull request #16477: [BEAM-13560][Playground] Split builder into several builders for each step of the pipeline execution
Posted by GitBox <gi...@apache.org>.
daria-malkova commented on a change in pull request #16477:
URL: https://github.com/apache/beam/pull/16477#discussion_r784023100
##########
File path: playground/backend/internal/setup_tools/builder/setup_builder.go
##########
@@ -31,73 +31,139 @@ const (
javaLogConfigFilePlaceholder = "{logConfigFile}"
)
-// SetupExecutorBuilder return executor with set args for validator, preparator, compiler and runner
-func SetupExecutorBuilder(paths fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+// Validator return executor with set args for validator
+func Validator(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
sdk := sdkEnv.ApacheBeamSdk
-
- if sdk == pb.Sdk_SDK_JAVA {
- pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
- }
-
val, err := utils.GetValidators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
+ builder := executors.NewExecutorBuilder().
+ WithValidator().
+ WithSdkValidators(val).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Preparer return executor with set args for preparator
+func Preparer(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
prep, err := utils.GetPreparators(sdk, paths.AbsoluteSourceFilePath)
if err != nil {
return nil, err
}
- executorConfig := sdkEnv.ExecutorConfig
builder := executors.NewExecutorBuilder().
- WithExecutableFileName(paths.AbsoluteExecutableFilePath).
- WithWorkingDir(paths.AbsoluteBaseFolderPath).
- WithValidator().
- WithSdkValidators(val).
WithPreparator().
WithSdkPreparators(prep).
+ ExecutorBuilder
+ return &builder, err
+}
+
+// Compiler return executor with set args for compiler
+func Compiler(paths *fs_tool.LifeCyclePaths, sdkEnv *environment.BeamEnvs) *executors.ExecutorBuilder {
+ sdk := sdkEnv.ApacheBeamSdk
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithCompiler().
WithCommand(executorConfig.CompileCmd).
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithArgs(executorConfig.CompileArgs).
WithFileName(paths.AbsoluteSourceFilePath).
+ ExecutorBuilder
+
+ switch sdk {
+ case pb.Sdk_SDK_JAVA:
+ builder.
+ WithCompiler().
+ WithFileName(GetFirstFileFromFolder(paths.AbsoluteSourceFileFolderPath))
+ }
+ return &builder
+}
+
+// Runner return executor with set args for runner
+func Runner(paths *fs_tool.LifeCyclePaths, pipelineOptions string, sdkEnv *environment.BeamEnvs) (*executors.ExecutorBuilder, error) {
+ sdk := sdkEnv.ApacheBeamSdk
+
+ if sdk == pb.Sdk_SDK_JAVA {
+ pipelineOptions = utils.ReplaceSpacesWithEquals(pipelineOptions)
+ }
+ executorConfig := sdkEnv.ExecutorConfig
+ builder := executors.NewExecutorBuilder().
WithRunner().
+ WithWorkingDir(paths.AbsoluteBaseFolderPath).
WithCommand(executorConfig.RunCmd).
WithArgs(executorConfig.RunArgs).
WithPipelineOptions(strings.Split(pipelineOptions, " ")).
- WithTestRunner().
- WithCommand(executorConfig.TestCmd).
- WithArgs(executorConfig.TestArgs).
- WithWorkingDir(paths.AbsoluteSourceFileFolderPath).
ExecutorBuilder
switch sdk {
- case pb.Sdk_SDK_JAVA: // Executable name for java class will be known after compilation
- args := make([]string, 0)
- for _, arg := range executorConfig.RunArgs {
- if strings.Contains(arg, javaLogConfigFilePlaceholder) {
- logConfigFilePath := filepath.Join(paths.AbsoluteBaseFolderPath, javaLogConfigFileName)
- arg = strings.Replace(arg, javaLogConfigFilePlaceholder, logConfigFilePath, 1)
- }
- args = append(args, arg)
+ case pb.Sdk_SDK_JAVA: // Executable name for java class is known after compilation
+ args := replaceLogPlaceholder(paths, executorConfig)
+ className, err := paths.ExecutableName(paths.AbsoluteExecutableFileFolderPath)
+ if err != nil {
+ return nil, fmt.Errorf("no executable file name found for JAVA pipeline at %s", paths.AbsoluteExecutableFileFolderPath)
}
- builder = builder.WithRunner().WithArgs(args).ExecutorBuilder
- builder = builder.WithTestRunner().WithWorkingDir(paths.AbsoluteBaseFolderPath).ExecutorBuilder //change directory for unit test
+ builder = builder.
+ WithRunner().
Review comment:
Yes
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org