Open paramjeet01 opened 5 months ago
/help
It is unclear where the limitation is coming from. Whether the limitation is coming from this client or some underlying library
@roycaihw: This request has been marked as needing help from a contributor.
Please ensure that the issue body includes answers to the following questions:
For more details on the requirements of such an issue, please see here and ensure that they are met.
If this request no longer meets these requirements, the label can be removed
by commenting with the /remove-help
command.
I'm not sure where the data is lost in between but I'm trying this code as suggested by @paul424 in this issue : https://github.com/kubernetes-client/python-base/issues/190. My understanding is that the data is lost while reading from the cat command,
def _extract_xcom(self, pod: V1Pod):
try:
self.log.info(f'Running command... cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
# Can't use _preload_content=True because that would return str(json.load(response))
client = kubernetes_stream(
self._client.connect_get_namespaced_pod_exec,
pod.metadata.name,
pod.metadata.namespace,
container=PodDefaults.SIDECAR_CONTAINER_NAME,
command=[
'/bin/sh',
'-c',
f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json',
],
stderr=True,
stdin=False,
stdout=True,
tty=False,
_preload_content=False,
_request_timeout=10,
)
client.run_forever(timeout=10)
resp = client.read_all()
self.log.info("Received {} ({}) ({} ... {}))".format(type(resp), len(resp), resp[:64], resp[-64:]))
# validate it's valid json
_ = json.loads(resp)
# Terminate the sidecar
_ = kubernetes_stream(
self._client.connect_get_namespaced_pod_exec,
pod.metadata.name,
pod.metadata.namespace,
container=PodDefaults.SIDECAR_CONTAINER_NAME,
command=[
'/bin/sh',
'-c',
'kill -s SIGINT 1',
],
stderr=True,
stdin=False,
stdout=True,
tty=False,
_preload_content=True,
_request_timeout=10,
)
return resp
except JSONDecodeError:
message = f'Failed to decode json document from pod: {pod.metadata.name}'
self.log.exception(message)
raise AirflowException(message)
except Exception as e:
message = f'Failed to extract xcom from pod: {pod.metadata.name}'
self.log.exception(message)
raise AirflowException(message)
We are facing this issue in airflow and think it's due to python kubernetes-client. Earlier repo issue : https://github.com/kubernetes-client/python-base/issues/190 Airflow issue : https://github.com/apache/airflow/issues/39267 Error log :