diff --git a/providers/src/airflow/providers/edge/CHANGELOG.rst b/providers/src/airflow/providers/edge/CHANGELOG.rst index 6f2fda5db9a63..def788fcf6410 100644 --- a/providers/src/airflow/providers/edge/CHANGELOG.rst +++ b/providers/src/airflow/providers/edge/CHANGELOG.rst @@ -27,6 +27,14 @@ Changelog --------- +0.5.5pre0 +......... + +Misc +~~~~ + +* ``Fixed reading none UTF-8 signs in log file.`` + 0.5.4pre0 ......... @@ -36,7 +44,6 @@ Misc * ``Fix SIGINT handling of child processes. Ensure graceful shutdown when SIGINT in received (not killing working tasks).`` * ``Fix SIGTERM handling of child processes. Ensure all childs are terminated on SIGTERM.`` - 0.5.3pre0 ......... diff --git a/providers/src/airflow/providers/edge/__init__.py b/providers/src/airflow/providers/edge/__init__.py index 996e57a229c6a..20e4f706d05ff 100644 --- a/providers/src/airflow/providers/edge/__init__.py +++ b/providers/src/airflow/providers/edge/__init__.py @@ -29,7 +29,7 @@ __all__ = ["__version__"] -__version__ = "0.5.4pre0" +__version__ = "0.5.5pre0" if packaging.version.parse(packaging.version.parse(airflow_version).base_version) < packaging.version.parse( "2.10.0" diff --git a/providers/src/airflow/providers/edge/cli/edge_command.py b/providers/src/airflow/providers/edge/cli/edge_command.py index be9a739dc29a0..4d0b46d74e79c 100644 --- a/providers/src/airflow/providers/edge/cli/edge_command.py +++ b/providers/src/airflow/providers/edge/cli/edge_command.py @@ -248,19 +248,24 @@ def check_running_jobs(self) -> None: logger.error("Job failed: %s", job.edge_job) EdgeJob.set_state(job.edge_job.key, TaskInstanceState.FAILED) if job.logfile.exists() and job.logfile.stat().st_size > job.logsize: - with job.logfile.open("r") as logfile: + with job.logfile.open("rb") as logfile: push_log_chunk_size = conf.getint("edge", "push_log_chunk_size") logfile.seek(job.logsize, os.SEEK_SET) + read_data = logfile.read() + job.logsize += len(read_data) + # backslashreplace to keep not decoded characters and not raising exception + log_data = read_data.decode(errors="backslashreplace") while True: - logdata = logfile.read(push_log_chunk_size) - if not logdata: + chunk_data = log_data[:push_log_chunk_size] + log_data = log_data[push_log_chunk_size:] + if not chunk_data: break + EdgeLogs.push_logs( task=job.edge_job.key, log_chunk_time=datetime.now(), - log_chunk_data=logdata, + log_chunk_data=chunk_data, ) - job.logsize += len(logdata) def heartbeat(self) -> None: """Report liveness state of worker to central site with stats.""" diff --git a/providers/src/airflow/providers/edge/provider.yaml b/providers/src/airflow/providers/edge/provider.yaml index fd0e55fd6f636..9ae4ea660a0e1 100644 --- a/providers/src/airflow/providers/edge/provider.yaml +++ b/providers/src/airflow/providers/edge/provider.yaml @@ -27,7 +27,7 @@ source-date-epoch: 1729683247 # note that those versions are maintained by release manager - do not update them manually versions: - - 0.5.4pre0 + - 0.5.5pre0 dependencies: - apache-airflow>=2.10.0 diff --git a/providers/tests/edge/cli/test_edge_command.py b/providers/tests/edge/cli/test_edge_command.py index f340813024387..43041a69c0518 100644 --- a/providers/tests/edge/cli/test_edge_command.py +++ b/providers/tests/edge/cli/test_edge_command.py @@ -234,15 +234,16 @@ def test_check_running_jobs_log_push_increment(self, mock_push_logs, worker_with def test_check_running_jobs_log_push_chunks(self, mock_push_logs, worker_with_job: _EdgeWorkerCli): job = worker_with_job.jobs[0] job.process.generated_returncode = None - job.logfile.write_text("log1log2log3") + job.logfile.write_bytes("log1log2ülog3".encode("latin-1")) with conf_vars({("edge", "api_url"): "https://mock.server", ("edge", "push_log_chunk_size"): "4"}): worker_with_job.check_running_jobs() assert len(worker_with_job.jobs) == 1 calls = mock_push_logs.call_args_list - len(calls) == 3 + assert len(calls) == 4 assert calls[0] == call(task=job.edge_job.key, log_chunk_time=datetime.now(), log_chunk_data="log1") assert calls[1] == call(task=job.edge_job.key, log_chunk_time=datetime.now(), log_chunk_data="log2") - assert calls[2] == call(task=job.edge_job.key, log_chunk_time=datetime.now(), log_chunk_data="log3") + assert calls[2] == call(task=job.edge_job.key, log_chunk_time=datetime.now(), log_chunk_data="\\xfc") + assert calls[3] == call(task=job.edge_job.key, log_chunk_time=datetime.now(), log_chunk_data="log3") @pytest.mark.parametrize( "drain, jobs, expected_state",