Add running_pod_log_lines config option to KubernetesExecutor

providers/cncf/kubernetes/provider.yaml

@@ -281,6 +281,14 @@ config:
         type: boolean
         example: ~
         default: "False"
+      running_pod_log_lines:
+        description: |
+          Number of lines read from the end of a running task's pod log when the task log
+          is served through the kube API, e.g. when viewing logs of a running task in the UI.
+        version_added: 10.20.0
+        type: integer
+        example: ~
+        default: "100"
       pod_template_file:
         description: |
           Path to the YAML pod file that forms the basis for KubernetesExecutor workers.

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py

@@ -120,6 +120,10 @@ def __init__(self, *args, **kwargs):
         self.task_publish_max_retries = self.conf.getint(
             "kubernetes_executor", "task_publish_max_retries", fallback=0
         )
+        # Shadows the class-level default with the configured value for this instance.
+        self.RUNNING_POD_LOG_LINES = self.conf.getint(
+            "kubernetes_executor", "running_pod_log_lines", fallback=self.RUNNING_POD_LOG_LINES
+        )
         self.completed: dict[tuple[str, str], KubernetesResults] = {}
         self.create_pods_after: datetime | None = None
 

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/get_provider_info.py

@@ -154,6 +154,13 @@ def get_provider_info():
                         "example": None,
                         "default": "False",
                     },
+                    "running_pod_log_lines": {
+                        "description": "Number of lines read from the end of a running task's pod log when the task log\nis served through the kube API, e.g. when viewing logs of a running task in the UI.\n",
+                        "version_added": "10.20.0",
+                        "type": "integer",
+                        "example": None,
+                        "default": "100",
+                    },
                     "pod_template_file": {
                         "description": "Path to the YAML pod file that forms the basis for KubernetesExecutor workers.\n",
                         "version_added": None,

providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py

@@ -640,6 +640,13 @@ def test_running_pod_log_lines(self):
         assert kube_executor.RUNNING_POD_LOG_LINES == 100
         assert kube_executor_2.RUNNING_POD_LOG_LINES == 200
 
+    @conf_vars({("kubernetes_executor", "running_pod_log_lines"): "500"})
+    def test_running_pod_log_lines_from_config(self):
+        kube_executor = KubernetesExecutor()
+
+        assert kube_executor.RUNNING_POD_LOG_LINES == 500
+        assert KubernetesExecutor.RUNNING_POD_LOG_LINES == 100
+
 
 class TestKubernetesExecutor:
     """