Issue 1:
Description

I got following error

[2021-11-09, 08:25:30 UTC] {base.py:70} INFO - Using connection to: id: rdb_conn_id. Host: rdb, Port: 5432, Schema: configuration, Login: user, Password: ***, extra: {'sslmode': 'allow', 'client_encoding': 'utf8'}
[2021-11-09, 08:25:30 UTC] {taskinstance.py:1703} ERROR - Task failed with exception
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1332, in _run_raw_task
self._execute_task_with_callbacks(context)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1458, in _execute_task_with_callbacks
result = self._execute_task(context, self.task)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1514, in _execute_task
result = execute_callable(context=context)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/operators/python.py", line 151, in execute
return_value = self.execute_callable()
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/operators/python.py", line 162, in execute_callable
return self.python_callable(*self.op_args, **self.op_kwargs)
File "/opt/airflow/dags/repo/dags/run_configuration.py", line 34, in run_job
dagsUtils.run_step_insert_to_temp_table(tenant, job_name, table_name, job_type)
File "/opt/airflow/dags/rev-e3db01f68e7979d71d12ae24008a97065db2144f/dags/utils/dag_util.py", line 106, in run_step_insert_to_temp_table
for df in df_result:
File "/home/airflow/.local/lib/python3.9/site-packages/pandas/io/sql.py", line 1499, in _query_iterator
data = result.fetchmany(chunksize)
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/result.py", line 1316, in fetchmany
self.connection._handle_dbapi_exception(
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1514, in _handle_dbapi_exception
util.raise_(exc_info[1], with_traceback=exc_info[2])
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
raise exception
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/result.py", line 1311, in fetchmany
l = self.process_rows(self._fetchmany_impl(size))
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/result.py", line 1224, in _fetchmany_impl
return self.cursor.fetchmany(size)
UnicodeDecodeError: 'ascii' codec can't decode byte 0xe2 in position 12: ordinal not in range(128)

Try to set extra in airflow connection but it does not work

Use case/motivation

See that airflow/providers/mysql/hooks support getting extra configs from airflow connection #6816 but not yet for postgresql hook

Related issues

No response

Are you willing to submit a PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
Apache Airflow Provider(s)

cncf-kubernetes

Versions of Apache Airflow Providers

apache-airflow-providers-cncf-kubernetes==4.4.0, but it appears in the most recent code I have seen.

Apache Airflow version

apache-airflow==2.4.3

Operating System

Ubuntu

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

What happened

The following function which gets logs from K8 pods occasionally, produces duplicated logs. I believe the reason for this is math.ceil() rounds up the number of seconds and occasionally logs are duplicated due to rounding errors. We see this stream cut out while producing logs then return logs from just before it cut out. An example of that is provided bellow.

def fetch_container_logs(
self, pod: V1Pod, container_name: str, *, follow=False, since_time: DateTime | None = None
) -> PodLoggingStatus:
"""
Follows the logs of container and streams to airflow logging.
Returns when container exits.
"""

def consume_logs(*, since_time: DateTime | None = None, follow: bool = True) -> DateTime | None:
"""
Tries to follow container logs until container completes.
For a long-running container, sometimes the log read may be interrupted
Such errors of this kind are suppressed.

Returns the last timestamp observed in logs.
"""
timestamp = None
try:
logs = self.read_pod_logs(
pod=pod,
container_name=container_name,
timestamps=True,
since_seconds=(
math.ceil((pendulum.now() - since_time).total_seconds()) if since_time else None
),
follow=follow,
)
for raw_line in logs:
line = raw_line.decode('utf-8', errors="backslashreplace")
timestamp, message = self.parse_log_line(line)
self.log.info(message)
except BaseHTTPError as e:
self.log.warning(
"Reading of logs interrupted with error %r; will retry. "
"Set log level to DEBUG for traceback.",
e,
)
self.log.debug(
"Traceback for interrupted logs read for pod %r",
pod.metadata.name,
exc_info=True,
)
return timestamp or since_time

# note: `read_pod_logs` follows the logs, so we shouldn't necessarily *need* to
# loop as we do here. But in a long-running process we might temporarily lose connectivity.
# So the looping logic is there to let us resume following the logs.
last_log_time = since_time
while True:
last_log_time = consume_logs(since_time=last_log_time, follow=follow)
if not self.container_is_running(pod, container_name=container_name):
return PodLoggingStatus(running=False, last_log_time=last_log_time)
if not follow:
return PodLoggingStatus(running=True, last_log_time=last_log_time)
else:
self.log.warning(
'Pod %s log read interrupted but container %s still running',
pod.metadata.name,
container_name,
)
time.sleep(1)

Example log illustrating this issue:

[2022-10-16, 14:45:32 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:32 UTC level=INFO|message= |fileName=aws.py|funcName=send_sqs_message|lineno=349|run_name=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|run_id=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|platform_count=3|group_id=e5b06226-b7a2-4174-9301-b243b54e0df0|schedule_id=e5b06226-b7a2-4174-9301-b243b54e0df0|task_type=xband_collect|event=send_notification|notification_type=sqs|notification_url=https://sqs.us-gov-east-1.amazonaws.com/252649730194/scheduler-pass-status.fifo|notification_message=e5b06226-b7a2-4174-9301-b243b54e0df0:delivered-3ball|notification_dedup_id_id=nfaZZWF-vTRsEa5j4jDyRDxA7Z0=|notification_message_id=f9880a14-1dfe-4bb4-bd36-f05cc322dcff|notification_status_code=200|status=success
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/e5b06226-b7a2-4174-9301-b243b54e0df0
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9/status?status=downloaded
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9|pass_status=downloaded
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9/status?status=Stored
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9|pass_status=Stored
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9/status?status=inspected
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9|pass_status=inspected
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/63ef6f27-97b7-4c16-88e6-26c1a6e2581e/status?status=downloaded
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=63ef6f27-97b7-4c16-88e6-26c1a6e2581e|pass_status=downloaded
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/63ef6f27-97b7-4c16-88e6-26c1a6e2581e/status?status=Stored
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=63ef6f27-97b7-4c16-88e6-26c1a6e2581e|pass_status=Stored
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/63ef6f27-97b7-4c16-88e6-26c1a6e2581e/status?status=inspected
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=63ef6f27-97b7-4c16-88e6-26c1a6e2581e|pass_status=inspected
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/beb79165-3d77-4615-b86d-37bb5b81ae64/status?status=downloaded
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=beb79165-3d77-4615-b86d-37bb5b81ae64|pass_status=downloaded
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/beb79165-3d77-4615-b86d-37bb5b81ae64/status?status=Stored
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=beb79165-3d77-4615-b86d-37bb5b81ae64|pass_status=Stored
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/beb79165-3d77-4615-b86d-37bb5b81ae64/status?status=inspected
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=beb79165-3d77-4615-b86d-37bb5b81ae64|pass_status=inspected
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=promote_processing_run.py|funcName=log_status_time_intervals|lineno=121|run_name=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|run_id=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|platform_count=3|group_id=e5b06226-b7a2-4174-9301-b243b54e0df0|schedule_id=e5b06226-b7a2-4174-9301-b243b54e0df0|task_type=xband_collect|event=collection_end_to_download|collection_end_time=1665887200|downloaded_time=1665904990|latency_minutes=296.5|collection_end_fallback=False|downloaded_fallback=False
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=promote_processing_run.py|funcName=log_status_time_intervals|lineno=133|run_name=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|run_id=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|platform_count=3|group_id=e5b06226-b7a2-4174-9301-b243b54e0df0|schedule_id=e5b06226-b7a2-4174-9301-b243b54e0df0|task_type=xband_collect|event=download_to_stored|downloaded_time=1665904990|stored_time=1665924817|latency_minutes=330.45|downloaded_fallback=False|stored_fallback=False
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=promote_processing_run.py|funcName=log_status_time_intervals|lineno=145|run_name=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|run_id=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|platform_count=3|group_id=e5b06226-b7a2-4174-9301-b243b54e0df0|schedule_id=e5b06226-b7a2-4174-9301-b243b54e0df0|task_type=xband_collect|event=stored_to_inspected|stored_time=1665924817|inspected_time=1665924984|latency_minutes=2.783333333333333|stored_fallback=False|inspected_fallback=False
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=promote_processing_run.py|funcName=log_status_time_intervals|lineno=155|run_name=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|run_id=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|platform_count=3|group_id=e5b06226-b7a2-4174-9301-b243b54e0df0|schedule_id=e5b06226-b7a2-4174-9301-b243b54e0df0|task_type=xband_collect|event=inspected_to_promoted|inspected_time=1665924984|promotion_time=1665931533|latency_minutes=109.15|inspected_fallback=False|status=delivered-3ball
[2022-10-16, 14:45:33 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=common.py|funcName=stop|lineno=47|run_name=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|run_id=e5b06226_geoloc_pipeline_radar_caf_1665924984.629687|platform_count=3|group_id=e5b06226-b7a2-4174-9301-b243b54e0df0|schedule_id=e5b06226-b7a2-4174-9301-b243b54e0df0|task_type=xband_collect|event=ending_heartbeat
[2022-10-16, 14:45:35 UTC] {pod_manager.py:251} WARNING - Pod geoloc-promote-processing-run-bdf89b8515e84aefa0fbb2cbe90b29a6 log read interrupted but container base still running
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/e5b06226-b7a2-4174-9301-b243b54e0df0
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9/status?status=downloaded
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9|pass_status=downloaded
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9/status?status=Stored
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9|pass_status=Stored
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9/status?status=inspected
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=2a3e94a4-0a5c-42aa-ba53-9bbf39fa30c9|pass_status=inspected
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/63ef6f27-97b7-4c16-88e6-26c1a6e2581e/status?status=downloaded
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=63ef6f27-97b7-4c16-88e6-26c1a6e2581e|pass_status=downloaded
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/63ef6f27-97b7-4c16-88e6-26c1a6e2581e/status?status=Stored
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=63ef6f27-97b7-4c16-88e6-26c1a6e2581e|pass_status=Stored
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/63ef6f27-97b7-4c16-88e6-26c1a6e2581e/status?status=inspected
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=63ef6f27-97b7-4c16-88e6-26c1a6e2581e|pass_status=inspected
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/beb79165-3d77-4615-b86d-37bb5b81ae64/status?status=downloaded
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=beb79165-3d77-4615-b86d-37bb5b81ae64|pass_status=downloaded
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/beb79165-3d77-4615-b86d-37bb5b81ae64/status?status=Stored
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=beb79165-3d77-4615-b86d-37bb5b81ae64|pass_status=Stored
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message={} |fileName=api_utils.py|funcName=_check_request_response|lineno=54|event=get_retry_request|status=success|url=https://scheduler.space.prod.gov.he2pi.com/api/schedule/passes/beb79165-3d77-4615-b86d-37bb5b81ae64/status?status=inspected
[2022-10-16, 14:45:36 UTC] {pod_manager.py:226} INFO - 2022-10-16, 14:45:33 UTC level=INFO|message= |fileName=scheduler_api.py|funcName=get_pass_id_status_timestamp|lineno=31|event=pass_id_status_retrieval|status=success|pass_id=beb79165-3d77-4615-b86d-37bb5b81ae64|pass_status=inspected

What you think should happen instead

I believe the following code should fix this issue.

def fetch_container_logs(
self, pod: V1Pod, container_name: str, *, follow=False, since_time: DateTime | None = None
) -> PodLoggingStatus:
"""
Follows the logs of container and streams to airflow logging.
Returns when container exits.
"""

def consume_logs(*, since_time: DateTime | None = None, follow: bool = True) -> DateTime | None:
"""
Tries to follow container logs until container completes.
For a long-running container, sometimes the log read may be interrupted
Such errors of this kind are suppressed.

Returns the last timestamp observed in logs.
"""
timestamp = None
try:
logs = self.read_pod_logs(
pod=pod,
container_name=container_name,
timestamps=True,
since_seconds=(
math.ceil((pendulum.now() - since_time).total_seconds()) if since_time else None
),
follow=follow,
)
for raw_line in logs:
line = raw_line.decode('utf-8', errors="backslashreplace")
tmp_timestamp, message = self.parse_log_line(line)
if tmp_timestamp > timestamp:
timestamp = tmp_timestamp
self.log.info(message)
except BaseHTTPError as e:
self.log.warning(
"Reading of logs interrupted with error %r; will retry. "
"Set log level to DEBUG for traceback.",
e,
)
self.log.debug(
"Traceback for interrupted logs read for pod %r",
pod.metadata.name,
exc_info=True,
)
return timestamp or since_time
``` def fetch_container_logs(
self, pod: V1Pod, container_name: str, *, follow=False, since_time: DateTime | None = None
) -> PodLoggingStatus:
"""
Follows the logs of container and streams to airflow logging.
Returns when container exits.
"""

def consume_logs(*, since_time: DateTime | None = None, follow: bool = True) -> DateTime | None:
"""
Tries to follow container logs until container completes.
For a long-running container, sometimes the log read may be interrupted
Such errors of this kind are suppressed.

Returns the last timestamp observed in logs.
"""
timestamp = None
try:
logs = self.read_pod_logs(
pod=pod,
container_name=container_name,
timestamps=True,
since_seconds=(
math.ceil((pendulum.now() - since_time).total_seconds()) if since_time else None
),
follow=follow,
)
for raw_line in logs:
line = raw_line.decode('utf-8', errors="backslashreplace")
tmp_timestamp, message = self.parse_log_line(line)
if tmp_timestamp > timestamp:
timestamp = tmp_timestamp
self.log.info(message)
except BaseHTTPError as e:
self.log.warning(
"Reading of logs interrupted with error %r; will retry. "
"Set log level to DEBUG for traceback.",
e,
)
self.log.debug(
"Traceback for interrupted logs read for pod %r",
pod.metadata.name,
exc_info=True,
)
return timestamp or since_time

How to reproduce

Please don't close my issue. I don't know how to provide reproducible code for this. I hope that my example, pointing to exactly where the problem is, and my proposed solution is sufficient to get this bug fixed.

Anything else

This bug appears fairly frequently in our set up which has lots of long running processes on K8 pods. It impacts some of our reporting and metrics and we would like to be able to correct it. Please let me know if there is anyway I can be of assistance as you work to correct the issue.

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct