Fix log tailing issues with legacy log view (apache#29496)

Probably we should just chop this view in favor of grid view logging which is the future. But this fixes rendering issues raised here apache#29447 (comment).

What we do, is in log tailing context (which apparently isn't used in grid, and that's why I did not see this in developing trigger logging) we don't add the messages to the log content.  So, whenever log_pos is in metadata we don't add messages.  It means the messages could be a bit stale but that seems ok.  Refreshing the page could fix that.  Longer term, we could update the API so that log content is just content and the messages are themselves returned in the metadata dict.  That's probably the "right" solution ultimately.  But can be saved for another day.  Also resolve the "cannot load lazy instance" issue when invoking the reader logic from this different context.

airflow/utils/log/file_task_handler.py

@@ -340,7 +340,8 @@ def _read(
         if metadata and "log_pos" in metadata:
             previous_chars = metadata["log_pos"]
             logs = logs[previous_chars:]  # Cut off previously passed log test as new tail
-        return messages + logs, {"end_of_log": end_of_log, "log_pos": log_pos}
+        out_message = logs if "log_pos" in (metadata or {}) else messages + logs
+        return out_message, {"end_of_log": end_of_log, "log_pos": log_pos}
 
     @staticmethod
     def _get_pod_namespace(ti: TaskInstance):

airflow/www/views.py

@@ -1523,7 +1523,15 @@ def get_logs_with_metadata(self, session=None):
 
         ti = (
             session.query(models.TaskInstance)
-            .filter_by(dag_id=dag_id, task_id=task_id, execution_date=execution_date, map_index=map_index)
+            .filter(
+                TaskInstance.task_id == task_id,
+                TaskInstance.dag_id == dag_id,
+                TaskInstance.execution_date == execution_date,
+                TaskInstance.map_index == map_index,
+            )
+            .join(TaskInstance.dag_run)
+            .options(joinedload("trigger"))
+            .options(joinedload("trigger.triggerer_job"))
             .first()
         )