This is an automated email from the ASF dual-hosted git repository. Abacn pushed a commit to branch restore-load in repository https://gitbox.apache.org/repos/asf/beam.git
commit 452a4abd48e43eee6071742cdb60b8a78c5cb6f0 Author: Yi Hu <[email protected]> AuthorDate: Thu Aug 13 15:34:14 2026 -0400 debug only --- .../workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml | 18 ++++++++---------- .test-infra/dataproc/flink_cluster.sh | 6 ++++++ sdks/go/pkg/beam/runner.go | 1 + sdks/go/pkg/beam/runners/universal/universal.go | 1 + sdks/go/pkg/beam/x/beamx/run.go | 9 +++++++-- 5 files changed, 23 insertions(+), 12 deletions(-) diff --git a/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml b/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml index 731c5b21274..66ef991a3c5 100644 --- a/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml +++ b/.github/workflows/beam_LoadTests_Go_GBK_Flink_Batch.yml @@ -48,13 +48,13 @@ env: INFLUXDB_USER: ${{ secrets.INFLUXDB_USER }} INFLUXDB_USER_PASSWORD: ${{ secrets.INFLUXDB_USER_PASSWORD }} GCLOUD_REGION: us-central1 - CLUSTER_NAME: beam-loadtests-go-gbk-flink-batch-${{ github.run_id }} + CLUSTER_NAME: beam-loadtests-go-sideinput-flink-batch-31733216613 GCS_BUCKET: gs://beam-flink-cluster FLINK_TASKMANAGER_SLOTS: 2 DETACHED_MODE: true HARNESS_IMAGES_TO_PULL: gcr.io/apache-beam-testing/beam-sdk/beam_go_sdk:latest JOB_SERVER_IMAGE: gcr.io/apache-beam-testing/beam_portability/beam_flink_job_server:latest-flink2.2 - ARTIFACTS_DIR: gs://beam-flink-cluster/beam-loadtests-go-gbk-flink-batch-${{ github.run_id }} + ARTIFACTS_DIR: gs://beam-flink-cluster/beam-loadtests-go-sideinput-flink-batch-31733216613 jobs: beam_LoadTests_Go_GBK_Flink_Batch: @@ -97,11 +97,9 @@ jobs: ${{ github.workspace }}/.github/workflows/load-tests-pipeline-options/go_GBK_Flink_Batch_Reiteration_10KB.txt arguments: | --job_name=load-tests-go-flink-batch-gbk-$(date '+%m%d%H%M%S' --utc) - - name: Start Flink with parallelism 5 - env: - FLINK_NUM_WORKERS: 5 + - name: Connect to existing Flink cluster run: | - cd ${{ github.workspace }}/.test-infra/dataproc; ./flink_cluster.sh create + cd ${{ github.workspace }}/.test-infra/dataproc; ./flink_cluster.sh tunnel # The env variables are created and populated in the test-arguments-action as "<github.job>_test_arguments_<argument_file_paths_index>" - name: run GBK Flink Batch Go Load Test 1 (10 b records) timeout-minutes: 120 @@ -162,7 +160,7 @@ jobs: -PloadTest.mainClass=group_by_key \ -Prunner=FlinkRunner \ '-PloadTest.args=${{ env.beam_LoadTests_Go_GBK_Flink_Batch_test_arguments_5 }}' \ - - name: Teardown Flink - if: always() - run: | - ${{ github.workspace }}/.test-infra/dataproc/flink_cluster.sh delete + # - name: Teardown Flink + # if: always() + # run: | + # ${{ github.workspace }}/.test-infra/dataproc/flink_cluster.sh delete diff --git a/.test-infra/dataproc/flink_cluster.sh b/.test-infra/dataproc/flink_cluster.sh index e146aa4b6ce..17180ced89d 100755 --- a/.test-infra/dataproc/flink_cluster.sh +++ b/.test-infra/dataproc/flink_cluster.sh @@ -183,6 +183,12 @@ function create() { start_tunnel } +# Connects to an existing Flink cluster and opens an SSH tunnel. +function tunnel() { + get_leader + start_tunnel +} + # Resizes an active Flink cluster. function resize() { if [[ -z "$FLINK_NUM_WORKERS" ]]; then diff --git a/sdks/go/pkg/beam/runner.go b/sdks/go/pkg/beam/runner.go index c9747da602e..d3557b24f1a 100644 --- a/sdks/go/pkg/beam/runner.go +++ b/sdks/go/pkg/beam/runner.go @@ -39,6 +39,7 @@ func RegisterRunner(name string, fn func(ctx context.Context, p *Pipeline) (Pipe // to define a "runner" with no default as a flag to let users control runner // selection. func Run(ctx context.Context, runner string, p *Pipeline) (PipelineResult, error) { + log.Infof(ctx, "Executing pipeline with runner: %v", runner) fn, ok := runners[runner] if !ok { log.Exitf(ctx, "Runner %v not registered. Forgot to _ import it?", runner) diff --git a/sdks/go/pkg/beam/runners/universal/universal.go b/sdks/go/pkg/beam/runners/universal/universal.go index 25325b8fe9c..36a3f45a8aa 100644 --- a/sdks/go/pkg/beam/runners/universal/universal.go +++ b/sdks/go/pkg/beam/runners/universal/universal.go @@ -60,6 +60,7 @@ func Execute(ctx context.Context, p *beam.Pipeline) (beam.PipelineResult, error) if err != nil { return nil, err } + log.Infof(ctx, "Universal runner connecting to endpoint: %v", endpoint) edges, _, err := p.Build() if err != nil { diff --git a/sdks/go/pkg/beam/x/beamx/run.go b/sdks/go/pkg/beam/x/beamx/run.go index ff3583917b7..885728ab230 100644 --- a/sdks/go/pkg/beam/x/beamx/run.go +++ b/sdks/go/pkg/beam/x/beamx/run.go @@ -20,6 +20,7 @@ import ( "context" "github.com/apache/beam/sdks/v2/go/pkg/beam" + "github.com/apache/beam/sdks/v2/go/pkg/beam/log" "github.com/apache/beam/sdks/v2/go/pkg/beam/runners" // common runner flag. // Import the reflection-optimized runtime. @@ -54,7 +55,9 @@ func getRunner() string { // defaults to the prism runner, but all beam-distributed runners and textio // filesystems are implicitly registered. func Run(ctx context.Context, p *beam.Pipeline) error { - _, err := beam.Run(ctx, getRunner(), p) + r := getRunner() + log.Infof(ctx, "beamx: using runner %q (flag value %q)", r, *runner) + _, err := beam.Run(ctx, r, p) return err } @@ -62,5 +65,7 @@ func Run(ctx context.Context, p *beam.Pipeline) error { // flag "runner". Returns a beam.PipelineResult objects, which can be // accessed to query the pipeline's metrics. func RunWithMetrics(ctx context.Context, p *beam.Pipeline) (beam.PipelineResult, error) { - return beam.Run(ctx, getRunner(), p) + r := getRunner() + log.Infof(ctx, "beamx: using runner %q (flag value %q)", r, *runner) + return beam.Run(ctx, r, p) }
