You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2021/09/08 01:42:25 UTC
[flink] branch master updated (9136c40 -> 5565590)
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 9136c40 [FLINK-23527][core] Clarify semantics of SourceFunction.cancel() with respect to thread interruptions.
new a6f8358 [FLINK-24137][python] Add more logs during starting up Python worker
new 5565590 [FLINK-24137][python][tests] Upload logs for PyFlink tests
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../pyflink/fn_execution/beam/beam_boot.py | 6 ++++-
flink-python/pyflink/pyflink_gateway_server.py | 26 +++++++++++++++++-----
flink-python/tox.ini | 1 +
tools/ci/test_controller.sh | 3 ++-
4 files changed, 29 insertions(+), 7 deletions(-)
[flink] 02/02: [FLINK-24137][python][tests] Upload logs for PyFlink
tests
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5565590dc27a77330127cdc027d83ccd10400bed
Author: Dian Fu <di...@apache.org>
AuthorDate: Tue Sep 7 20:59:47 2021 +0800
[FLINK-24137][python][tests] Upload logs for PyFlink tests
---
flink-python/pyflink/pyflink_gateway_server.py | 26 +++++++++++++++++++++-----
flink-python/tox.ini | 1 +
tools/ci/test_controller.sh | 3 ++-
3 files changed, 24 insertions(+), 6 deletions(-)
diff --git a/flink-python/pyflink/pyflink_gateway_server.py b/flink-python/pyflink/pyflink_gateway_server.py
index e778644..b137748 100644
--- a/flink-python/pyflink/pyflink_gateway_server.py
+++ b/flink-python/pyflink/pyflink_gateway_server.py
@@ -64,23 +64,39 @@ def find_java_executable():
def construct_log_settings():
templates = [
"-Dlog.file=${flink_log_dir}/flink-${flink_ident_string}-python-${hostname}.log",
- "-Dlog4j.configuration=${flink_conf_dir}/log4j-cli.properties",
- "-Dlog4j.configurationFile=${flink_conf_dir}/log4j-cli.properties",
- "-Dlogback.configurationFile=${flink_conf_dir}/logback.xml"
+ "-Dlog4j.configuration=${log4j_properties}",
+ "-Dlog4j.configurationFile=${log4j_properties}",
+ "-Dlogback.configurationFile=${logback_xml}"
]
flink_home = _find_flink_home()
flink_conf_dir = os.path.join(flink_home, "conf")
- flink_log_dir = os.path.join(flink_home, "log")
+ if "FLINK_LOG_DIR" in os.environ:
+ flink_log_dir = os.environ["FLINK_LOG_DIR"]
+ else:
+ flink_log_dir = os.path.join(flink_home, "log")
+
+ if "LOG4J_PROPERTIES" in os.environ:
+ log4j_properties = os.environ["LOG4J_PROPERTIES"]
+ else:
+ log4j_properties = "%s/log4j-cli.properties" % flink_conf_dir
+
+ if "LOGBACK_XML" in os.environ:
+ logback_xml = os.environ["LOGBACK_XML"]
+ else:
+ logback_xml = "%s/logback.xml" % flink_conf_dir
+
if "FLINK_IDENT_STRING" in os.environ:
flink_ident_string = os.environ["FLINK_IDENT_STRING"]
else:
flink_ident_string = getpass.getuser()
+
hostname = socket.gethostname()
log_settings = []
for template in templates:
log_settings.append(Template(template).substitute(
- flink_conf_dir=flink_conf_dir,
+ log4j_properties=log4j_properties,
+ logback_xml=logback_xml,
flink_log_dir=flink_log_dir,
flink_ident_string=flink_ident_string,
hostname=hostname))
diff --git a/flink-python/tox.ini b/flink-python/tox.ini
index 2cc04f6..475a316 100644
--- a/flink-python/tox.ini
+++ b/flink-python/tox.ini
@@ -33,6 +33,7 @@ deps =
grpcio>=1.17.0,<=1.26.0
grpcio-tools>=1.3.5,<=1.14.2
apache-flink-libraries
+passenv = *
commands =
python --version
# python test
diff --git a/tools/ci/test_controller.sh b/tools/ci/test_controller.sh
index 4f4d0cc..4bd7242 100755
--- a/tools/ci/test_controller.sh
+++ b/tools/ci/test_controller.sh
@@ -77,7 +77,7 @@ source "${HERE}/watchdog.sh"
# Step 1: Rebuild jars and install Flink to local maven repository
# =============================================================================
-LOG4J_PROPERTIES=${HERE}/log4j.properties
+export LOG4J_PROPERTIES=${HERE}/log4j.properties
MVN_LOGGING_OPTIONS="-Dlog.dir=${DEBUG_FILES_OUTPUT_DIR} -Dlog4j.configurationFile=file://$LOG4J_PROPERTIES"
MVN_COMMON_OPTIONS="-Dflink.forkCount=2 -Dflink.forkCountTestPackage=2 -Dfast -Pskip-webui-build $MVN_LOGGING_OPTIONS"
@@ -101,6 +101,7 @@ fi
# =============================================================================
if [ $STAGE == $STAGE_PYTHON ]; then
+ sed -i "s/\(^appender\.file\.fileName = \).*$/\1\$\{sys:log\.file\}/g" ${HERE}/log4j.properties
run_with_watchdog "./flink-python/dev/lint-python.sh" $CALLBACK_ON_TIMEOUT
EXIT_CODE=$?
else
[flink] 01/02: [FLINK-24137][python] Add more logs during starting
up Python worker
Posted by di...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a6f8358166adc04ff20235331a919d82884b3a48
Author: Dian Fu <di...@apache.org>
AuthorDate: Tue Sep 7 20:58:39 2021 +0800
[FLINK-24137][python] Add more logs during starting up Python worker
---
flink-python/pyflink/fn_execution/beam/beam_boot.py | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
diff --git a/flink-python/pyflink/fn_execution/beam/beam_boot.py b/flink-python/pyflink/fn_execution/beam/beam_boot.py
index 3c3dd27..a0f2da0 100644
--- a/flink-python/pyflink/fn_execution/beam/beam_boot.py
+++ b/flink-python/pyflink/fn_execution/beam/beam_boot.py
@@ -74,9 +74,11 @@ if __name__ == "__main__":
check_not_empty(worker_id, "No id provided.")
check_not_empty(provision_endpoint, "No provision endpoint provided.")
- logging.info("Initializing python harness: %s" % " ".join(sys.argv))
+ logging.info("Initializing Python harness: %s" % " ".join(sys.argv))
if 'PYFLINK_LOOPBACK_SERVER_ADDRESS' in os.environ:
+ logging.info("Starting up Python harness in loopback mode.")
+
params = dict(os.environ)
params.update({'SEMI_PERSISTENT_DIRECTORY': semi_persist_dir})
with grpc.insecure_channel(os.environ['PYFLINK_LOOPBACK_SERVER_ADDRESS']) as channel:
@@ -87,6 +89,7 @@ if __name__ == "__main__":
params=params)
client.StartWorker(request)
else:
+ logging.info("Starting up Python harness in a standalone process.")
metadata = [("worker_id", worker_id)]
# read job information from provision stub
@@ -108,6 +111,7 @@ if __name__ == "__main__":
env = dict(os.environ)
if "FLINK_BOOT_TESTING" in os.environ and os.environ["FLINK_BOOT_TESTING"] == "1":
+ logging.info("Shut down Python harness due to FLINK_BOOT_TESTING is set.")
exit(0)
call([python_exec, "-m", "pyflink.fn_execution.beam.beam_sdk_worker_main"],