[edge] Fixed UnicodeDecodeError during log file upload of Edge worker

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
 .........
 

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"

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."""

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

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",