[GitHub] [airflow] asaf400 commented on a change in pull request #19027: Fix for DockerOperator Xcoms functionality

2021-10-17 Thread GitBox 


asaf400 commented on a change in pull request #19027:
URL: https://github.com/apache/airflow/pull/19027#discussion_r730596500



##
File path: airflow/providers/docker/operators/docker.py
##
@@ -304,21 +304,24 @@ def _run_image_with_mounts(self, target_mounts, 
add_tmp_variable: bool) -> Optio
 working_dir=self.working_dir,
 tty=self.tty,
 )
-lines = self.cli.attach(container=self.container['Id'], stdout=True, 
stderr=True, stream=True)
+logstream = self.cli.attach(container=self.container['Id'], 
stdout=True, stderr=True, stream=True)
 try:
 self.cli.start(self.container['Id'])
 
-line = ''
+log_chunk = ''
 res_lines = []
 return_value = None
-for line in lines:
-if hasattr(line, 'decode'):
-# Note that lines returned can also be byte sequences so 
we have to handle decode here
-line = line.decode('utf-8')
-line = line.strip()
-res_lines.append(line)
-self.log.info(line)
+for log_chunk in logstream:
+if hasattr(log_chunk, 'decode'):
+# Note that log_chunk returned can also be byte sequences 
so we have to handle decode here
+log_chunk = log_chunk.decode('utf-8')
+log_chunk = log_chunk.strip()
+res_lines.append(log_chunk)
+self.log.info(log_chunk)
 result = self.cli.wait(self.container['Id'])
+# after container has exited, grab the entire log ignoring the 
chunked log stream that was used with attach
+# self.cli.logs uses docker's /containers/{id}/logs, while 
self.cli.attach uses /containers/{id}/attach
+lines = self.cli.logs(container=self.container['Id'], stdout=True, 
stderr=True, stream=True)

Review comment:
   Yeah, tail looks great, I didn't know it was an argument of logs(), even 
though it was right there in the docs..




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] asaf400 commented on a change in pull request #19027: Fix for DockerOperator Xcoms functionality

2021-10-17 Thread GitBox 


asaf400 commented on a change in pull request #19027:
URL: https://github.com/apache/airflow/pull/19027#discussion_r730596676



##
File path: airflow/providers/docker/operators/docker.py
##
@@ -304,21 +304,24 @@ def _run_image_with_mounts(self, target_mounts, 
add_tmp_variable: bool) -> Optio
 working_dir=self.working_dir,
 tty=self.tty,
 )
-lines = self.cli.attach(container=self.container['Id'], stdout=True, 
stderr=True, stream=True)
+logstream = self.cli.attach(container=self.container['Id'], 
stdout=True, stderr=True, stream=True)
 try:
 self.cli.start(self.container['Id'])
 
-line = ''
+log_chunk = ''
 res_lines = []
 return_value = None
-for line in lines:
-if hasattr(line, 'decode'):
-# Note that lines returned can also be byte sequences so 
we have to handle decode here
-line = line.decode('utf-8')
-line = line.strip()
-res_lines.append(line)
-self.log.info(line)
+for log_chunk in logstream:
+if hasattr(log_chunk, 'decode'):
+# Note that log_chunk returned can also be byte sequences 
so we have to handle decode here
+log_chunk = log_chunk.decode('utf-8')
+log_chunk = log_chunk.strip()
+res_lines.append(log_chunk)
+self.log.info(log_chunk)
 result = self.cli.wait(self.container['Id'])
+# after container has exited, grab the entire log ignoring the 
chunked log stream that was used with attach
+# self.cli.logs uses docker's /containers/{id}/logs, while 
self.cli.attach uses /containers/{id}/attach
+lines = self.cli.logs(container=self.container['Id'], stdout=True, 
stderr=True, stream=True)

Review comment:
   ```suggestion
   
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] asaf400 commented on a change in pull request #19027: Fix for DockerOperator Xcoms functionality

2021-10-17 Thread GitBox 


asaf400 commented on a change in pull request #19027:
URL: https://github.com/apache/airflow/pull/19027#discussion_r730597633



##
File path: airflow/providers/docker/operators/docker.py
##
@@ -328,7 +331,7 @@ def _run_image_with_mounts(self, target_mounts, 
add_tmp_variable: bool) -> Optio
 if self.retrieve_output:
 ret = return_value
 elif self.do_xcom_push:
-ret = self._get_return_value_from_logs(res_lines, line)
+ret = self._get_return_value_from_logs(lines, lines[-1])

Review comment:
   @uranusjr 
   
   so you were thinking a doing flavor of this, with the previous suggested 
edit (removing lines from earlier in the code) ?
   ```suggestion
   ret = self.cli.logs(container=self.container['Id'], 
stdout=True, stderr=True, stream=True) if self.xcom_all else 
self.cli.logs(container=self.container['Id'], stdout=True, stderr=True, 
stream=True, tail=1)
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] asaf400 commented on a change in pull request #19027: Fix for DockerOperator Xcoms functionality

2021-10-17 Thread GitBox 


asaf400 commented on a change in pull request #19027:
URL: https://github.com/apache/airflow/pull/19027#discussion_r730597633



##
File path: airflow/providers/docker/operators/docker.py
##
@@ -328,7 +331,7 @@ def _run_image_with_mounts(self, target_mounts, 
add_tmp_variable: bool) -> Optio
 if self.retrieve_output:
 ret = return_value
 elif self.do_xcom_push:
-ret = self._get_return_value_from_logs(res_lines, line)
+ret = self._get_return_value_from_logs(lines, lines[-1])

Review comment:
   @uranusjr 
   
   so you were thinking a doing flavor of this, with the previous suggested 
edit (removing lines from earlier in the code) ?
   ```suggestion
   return self.cli.logs(container=self.container['Id'], 
stdout=True, stderr=True, stream=True) if self.xcom_all else 
self.cli.logs(container=self.container['Id'], stdout=True, stderr=True, 
stream=True, tail=1)
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] asaf400 commented on a change in pull request #19027: Fix for DockerOperator Xcoms functionality

2021-10-17 Thread GitBox 


asaf400 commented on a change in pull request #19027:
URL: https://github.com/apache/airflow/pull/19027#discussion_r730598411



##
File path: airflow/providers/docker/operators/docker.py
##
@@ -328,7 +331,7 @@ def _run_image_with_mounts(self, target_mounts, 
add_tmp_variable: bool) -> Optio
 if self.retrieve_output:
 ret = return_value
 elif self.do_xcom_push:
-ret = self._get_return_value_from_logs(res_lines, line)
+ret = self._get_return_value_from_logs(lines, lines[-1])
 return ret

Review comment:
   @uranusjr 
   
   so you were thinking a doing flavor of this, with the previous suggested 
edit (removing lines from earlier in the code) ?
   ```suggestion
   return self.cli.logs(container=self.container['Id'], 
stdout=True, stderr=True, stream=True) if self.xcom_all else 
self.cli.logs(container=self.container['Id'], stdout=True, stderr=True, 
stream=True, tail=1)
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] asaf400 commented on a change in pull request #19027: Fix for DockerOperator Xcoms functionality

2021-10-17 Thread GitBox 


asaf400 commented on a change in pull request #19027:
URL: https://github.com/apache/airflow/pull/19027#discussion_r730598411



##
File path: airflow/providers/docker/operators/docker.py
##
@@ -328,7 +331,7 @@ def _run_image_with_mounts(self, target_mounts, 
add_tmp_variable: bool) -> Optio
 if self.retrieve_output:
 ret = return_value
 elif self.do_xcom_push:
-ret = self._get_return_value_from_logs(res_lines, line)
+ret = self._get_return_value_from_logs(lines, lines[-1])
 return ret

Review comment:
   @uranusjr 
   
   so you were thinking a doing flavor of this, 
   with the previous suggested edit (removing lines from earlier in the code) ?
   
   it kind of looks bad, because it's too long, but it should work,
   Feel free to give another suggestion to make it prettier, I'm missing the 
way to prettify it, if that what you meant by removing 
_get_return_value_from_logs
   
   ```suggestion
   return self.cli.logs(container=self.container['Id'], 
stdout=True, stderr=True, stream=True) if self.xcom_all else 
self.cli.logs(container=self.container['Id'], stdout=True, stderr=True, 
stream=True, tail=1)
   ```




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] asaf400 commented on a change in pull request #19027: Fix for DockerOperator Xcoms functionality

2021-10-18 Thread GitBox 


asaf400 commented on a change in pull request #19027:
URL: https://github.com/apache/airflow/pull/19027#discussion_r731258267



##
File path: airflow/providers/docker/operators/docker.py
##
@@ -328,7 +331,7 @@ def _run_image_with_mounts(self, target_mounts, 
add_tmp_variable: bool) -> Optio
 if self.retrieve_output:
 ret = return_value
 elif self.do_xcom_push:
-ret = self._get_return_value_from_logs(res_lines, line)
+ret = self._get_return_value_from_logs(lines, lines[-1])
 return ret

Review comment:
   Maybe, I tested out adding random \n into one of the bash 'echo' 
(without -e) commands in my example in the issue discussion,
   I tried using logstream.split('\n') but it had undesired results, as some of 
the random `\n`'s were converted to newlines in the output, but I forgot to 
test splitlines() I guess




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] asaf400 commented on a change in pull request #19027: Fix for DockerOperator Xcoms functionality

2021-10-18 Thread GitBox 


asaf400 commented on a change in pull request #19027:
URL: https://github.com/apache/airflow/pull/19027#discussion_r731258638



##
File path: airflow/providers/docker/operators/docker.py
##
@@ -328,7 +331,7 @@ def _run_image_with_mounts(self, target_mounts, 
add_tmp_variable: bool) -> Optio
 if self.retrieve_output:
 ret = return_value
 elif self.do_xcom_push:
-ret = self._get_return_value_from_logs(res_lines, line)
+ret = self._get_return_value_from_logs(lines, lines[-1])
 return ret

Review comment:
   I'll test these suggestions shortly




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] asaf400 commented on a change in pull request #19027: Fix for DockerOperator Xcoms functionality

2021-10-18 Thread GitBox 


asaf400 commented on a change in pull request #19027:
URL: https://github.com/apache/airflow/pull/19027#discussion_r731265880



##
File path: airflow/providers/docker/operators/docker.py
##
@@ -328,7 +331,7 @@ def _run_image_with_mounts(self, target_mounts, 
add_tmp_variable: bool) -> Optio
 if self.retrieve_output:
 ret = return_value
 elif self.do_xcom_push:
-ret = self._get_return_value_from_logs(res_lines, line)
+ret = self._get_return_value_from_logs(lines, lines[-1])
 return ret

Review comment:
   I guess this small test help understand what I meant
   
   ```python
   #!/usr/bin/python3
   
   str = r"""this is
   an 'herestring'
   with some lines,
   but now\n there's even more \n lines, but are they real?\n
   string example\nwow!!!"""
   print (str.splitlines())
   ```
   
   In this case, the difference between an expected results and an unexpected 
results lays in the `r`,
   without the r""", splitlines would also process the "inline `\n`'s"
   
   
   




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

[GitHub] [airflow] asaf400 commented on a change in pull request #19027: Fix for DockerOperator Xcoms functionality

2021-10-18 Thread GitBox 


asaf400 commented on a change in pull request #19027:
URL: https://github.com/apache/airflow/pull/19027#discussion_r731265880



##
File path: airflow/providers/docker/operators/docker.py
##
@@ -328,7 +331,7 @@ def _run_image_with_mounts(self, target_mounts, 
add_tmp_variable: bool) -> Optio
 if self.retrieve_output:
 ret = return_value
 elif self.do_xcom_push:
-ret = self._get_return_value_from_logs(res_lines, line)
+ret = self._get_return_value_from_logs(lines, lines[-1])
 return ret

Review comment:
   I guess this small test help understand what I meant
   
   ```python
   #!/usr/bin/python3
   
   str = r"""this is
   an 'herestring'
   with some lines,
   but now\n there's even more \n lines, but are they real?\n
   string example\nwow!!!"""
   print (str.splitlines())
   ```
   
   In this case, the difference between an expected results and an unexpected 
results lays in the `r`,
   without the r""", splitlines would also process the "inline `\n`'s"
   
   I'll test how logstream would handle splitlines 




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org