You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ch...@apache.org on 2024/03/06 01:02:18 UTC
(beam) 01/01: Debugging Transform Service test suite
This is an automated email from the ASF dual-hosted git repository.
chamikara pushed a commit to branch transform_service_test_suite
in repository https://gitbox.apache.org/repos/asf/beam.git
commit e28b71fe2b5164d1608d4f1c47d560e94a746d48
Author: Chamikara Jayalath <ch...@gmail.com>
AuthorDate: Tue Mar 5 17:01:46 2024 -0800
Debugging Transform Service test suite
---
.../beam_PostCommit_TransformService_Direct.yml | 2 ++
.../org/apache/beam/gradle/BeamModulePlugin.groovy | 2 +-
.../apache_beam/io/gcp/bigtableio_it_test.py | 27 ++++++++++++------
sdks/python/expansion-service-container/boot.go | 14 +++++++++-
sdks/python/scripts/run_transform_service.sh | 32 ++++++++++++++++++++++
5 files changed, 67 insertions(+), 10 deletions(-)
diff --git a/.github/workflows/beam_PostCommit_TransformService_Direct.yml b/.github/workflows/beam_PostCommit_TransformService_Direct.yml
index 1497681ae99..db8fa934f47 100644
--- a/.github/workflows/beam_PostCommit_TransformService_Direct.yml
+++ b/.github/workflows/beam_PostCommit_TransformService_Direct.yml
@@ -87,6 +87,8 @@ jobs:
-Pjava11Home=$JAVA_HOME_11_X64 \
-PuseWheelDistribution \
-PpythonVersion=${{ matrix.python_version }} \
+ --stacktrace \
+ --info \
- name: Archive Python Test Results
uses: actions/upload-artifact@v4
if: failure()
diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
index 6434746fd3a..fe36bb89cdc 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -2890,7 +2890,7 @@ class BeamModulePlugin implements Plugin<Project> {
def cmdArgs = project.project(':sdks:python').mapToArgString(beamPythonTestPipelineOptions)
project.exec {
- environment "EXPANSION_PORT", externalPort
+ environment "TRANSFORM_SERVICE_PORT", externalPort
executable 'sh'
args '-c', ". $envDir/bin/activate && cd $pythonDir && ./scripts/run_integration_test.sh $cmdArgs"
}
diff --git a/sdks/python/apache_beam/io/gcp/bigtableio_it_test.py b/sdks/python/apache_beam/io/gcp/bigtableio_it_test.py
index 13909cded1f..b34e072e6d8 100644
--- a/sdks/python/apache_beam/io/gcp/bigtableio_it_test.py
+++ b/sdks/python/apache_beam/io/gcp/bigtableio_it_test.py
@@ -60,9 +60,10 @@ def instance_prefix(instance):
@pytest.mark.uses_transform_service
@unittest.skipIf(client is None, 'Bigtable dependencies are not installed')
@unittest.skipUnless(
- os.environ.get('EXPANSION_JARS'),
- "EXPANSION_JARS environment var is not provided, "
- "indicating that jars have not been built")
+ os.environ.get('EXPANSION_JARS') or
+ os.environ.get('TRANSFORM_SERVICE_PORT'),
+ "A valid expansion service is not available for executing the "
+ "cross-language test.")
class TestReadFromBigTableIT(unittest.TestCase):
INSTANCE = "bt-read-tests"
TABLE_ID = "test-table"
@@ -92,6 +93,12 @@ class TestReadFromBigTableIT(unittest.TestCase):
self.table = self.instance.table(self.TABLE_ID)
self.table.create()
_LOGGER.info("Created table [%s]", self.table.table_id)
+ if (os.environ.get('TRANSFORM_SERVICE_PORT')):
+ self._transform_service_address = (
+ 'localhost:' + os.environ.get('TRANSFORM_SERVICE_PORT'))
+ else:
+ self._transform_service_address = None
+
def tearDown(self):
try:
@@ -141,7 +148,8 @@ class TestReadFromBigTableIT(unittest.TestCase):
| bigtableio.ReadFromBigtable(
project_id=self.project,
instance_id=self.instance.instance_id,
- table_id=self.table.table_id)
+ table_id=self.table.table_id,
+ expansion_service=self._transform_service_address)
| "Extract cells" >> beam.Map(lambda row: row._cells))
assert_that(cells, equal_to(expected_cells))
@@ -151,9 +159,10 @@ class TestReadFromBigTableIT(unittest.TestCase):
@pytest.mark.uses_transform_service
@unittest.skipIf(client is None, 'Bigtable dependencies are not installed')
@unittest.skipUnless(
- os.environ.get('EXPANSION_JARS'),
- "EXPANSION_JARS environment var is not provided, "
- "indicating that jars have not been built")
+ os.environ.get('EXPANSION_JARS') or
+ os.environ.get('TRANSFORM_SERVICE_PORT'),
+ "A valid expansion service is not available for executing the "
+ "cross-language test.")
class TestWriteToBigtableXlangIT(unittest.TestCase):
# These are integration tests for the cross-language write transform.
INSTANCE = "bt-write-xlang"
@@ -214,9 +223,11 @@ class TestWriteToBigtableXlangIT(unittest.TestCase):
project_id=self.project,
instance_id=self.instance.instance_id,
table_id=self.table.table_id,
- use_cross_language=True))
+ use_cross_language=True,
+ expansion_service=self._transform_service_address))
def test_set_mutation(self):
+# raise Exception("Intentional failure to check trivial passing")
row1: DirectRow = DirectRow('key-1')
row2: DirectRow = DirectRow('key-2')
col_fam = self.table.column_family('col_fam')
diff --git a/sdks/python/expansion-service-container/boot.go b/sdks/python/expansion-service-container/boot.go
index ba56b349c4e..3c6636fd846 100644
--- a/sdks/python/expansion-service-container/boot.go
+++ b/sdks/python/expansion-service-container/boot.go
@@ -24,6 +24,7 @@ import (
"io/ioutil"
"log"
"os"
+// "os/exec"
"path/filepath"
"strconv"
"strings"
@@ -137,7 +138,7 @@ func launchExpansionServiceProcess() error {
if err != nil {
return err
}
- log.Printf("Starting Python expansion service ...")
+ log.Printf("Starting Python expansion service ... 1")
dir := filepath.Join("/opt/apache/beam", venvDirectory)
os.Setenv("VIRTUAL_ENV", dir)
@@ -145,6 +146,17 @@ func launchExpansionServiceProcess() error {
args := []string{"-m", expansionServiceEntrypoint, "-p", strconv.Itoa(*port), "--fully_qualified_name_glob", "*"}
+
+ log.Printf("****** xyz123 executing ls -al:")
+ if err := execx.Execute("/usr/bin/ls", "-al"); err != nil {
+ return fmt.Errorf("Could not execute /usr/bin/ls -al: %s", err)
+ }
+
+ log.Printf("****** xyz123 executing ls -al /:")
+ if err := execx.Execute("/usr/bin/ls", "-al", "/"); err != nil {
+ return fmt.Errorf("Could not execute /usr/bin/ls -al /: %s", err)
+ }
+
if *requirements_file != "" {
log.Printf("Received the requirements file %v", *requirements_file)
updatedRequirementsFileName, err := getUpdatedRequirementsFile(*requirements_file, *dependencies_dir)
diff --git a/sdks/python/scripts/run_transform_service.sh b/sdks/python/scripts/run_transform_service.sh
index b71d67707e0..b1596112555 100755
--- a/sdks/python/scripts/run_transform_service.sh
+++ b/sdks/python/scripts/run_transform_service.sh
@@ -16,6 +16,8 @@
# limitations under the License.
#
+echo "************ xyz123 run_transform_service.sh: 1"
+
read -r -d '' USAGE <<END
Usage: run_expansion_services.sh (start|stop) [options]
Options:
@@ -26,6 +28,8 @@ Options:
--stop [command to stop the transform service for the given group_id]
END
+echo "************ xyz123 run_transform_service.sh: 2"
+
while [[ $# -gt 0 ]]; do
key="$1"
case $key in
@@ -66,6 +70,8 @@ while [[ $# -gt 0 ]]; do
esac
done
+echo "************ xyz123 run_transform_service.sh: 3"
+
FILE_BASE="beam-transform-service"
if [ -n "$GROUP_ID" ]; then
FILE_BASE="$FILE_BASE-$GROUP_ID"
@@ -73,12 +79,35 @@ fi
TEMP_DIR=/tmp
+echo "************ xyz123 run_transform_service.sh: 4"
+
case $STARTSTOP in
start)
+ echo "************ xyz123 in script run_transform_service.sh: starting the service"
echo "Starting the transform service for project $GROUP_ID at port $EXTERNAL_PORT for Beam version $BEAM_VERSION_DOCKER transform service startup jar is $TRANSFORM_SERVICE_LAUNCHER_JAR"
java -jar $TRANSFORM_SERVICE_LAUNCHER_JAR --project_name $GROUP_ID --port $EXTERNAL_PORT --beam_version $BEAM_VERSION_DOCKER --command up >$TEMP_DIR/$FILE_BASE-java1.log 2>&1 </dev/null
+ echo "************ xyz123 in script run_transform_service.sh: DONE starting the service"
;;
stop)
+ echo "************ xyz123 run_transform_service.sh: printing docker logs"
+ temp_output=`docker ps`
+ printf "temp_output: $temp_output"
+ printf "Logs from Controller:\n"
+ temp_output=`docker ps | grep 'controller'`
+ printf "temp_output controller: $temp_output"
+ container=${temp_output%% *}
+ docker logs $container
+ printf "Logs from Java expansion service:\n"
+ temp_output=`docker ps | grep 'java'`
+ printf "temp_output java exp service: $temp_output"
+ container=${temp_output%% *}
+ docker logs $container
+ printf "Logs from Python expansion service:\n"
+ temp_output=`docker ps | grep 'python'`
+ printf "temp_output py exp service: $temp_output"
+ container=${temp_output%% *}
+ docker logs $container
+ echo "************ xyz123 run_transform_service.sh: DONE printing docker logs"
echo "Stopping the transform service for project $GROUP_ID at port $EXTERNAL_PORT for Beam version $BEAM_VERSION_DOCKER transform service startup jar is $TRANSFORM_SERVICE_LAUNCHER_JAR"
java -jar $TRANSFORM_SERVICE_LAUNCHER_JAR --project_name $GROUP_ID --port $EXTERNAL_PORT --beam_version $BEAM_VERSION_DOCKER --command down >$TEMP_DIR/$FILE_BASE-java2.log 2>&1 </dev/null
TRANSFORM_SERVICE_TEMP_DIR=$TEMP_DIR/$GROUP_ID
@@ -88,3 +117,6 @@ case $STARTSTOP in
fi
;;
esac
+
+echo "************ xyz123 run_transform_service.sh: 5"
+