[AIRFLOW-6099] Add host name to task runner log

[turbaszek]

Make sure you have checked all steps below.

Jira

• My PR addresses the following Airflow Jira issues and references them in the PR title. For example, "[AIRFLOW-XXX] My Airflow PR"
  • https://issues.apache.org/jira/browse/AIRFLOW-6099

Description

• Here are some details about my PR, including screenshots of any UI changes:
  It would be great to be able to look at log at web UI and see on what host the task was running. This is very helpful especially with celery executor using multiple workers.

[2019-11-29 10:10:41,214] {taskinstance.py:846} INFO - Executing <Task(PubSubTopicCreateOperator): create_topic> on 2019-11-28T00:00:00+00:00
[2019-11-29 10:10:41,215] {base_task_runner.py:125} INFO - Running on host: airflow-worker-75887d4878-4rcx2@-@{"workflow": "example_gcp_pubsub", "task-id": "create_topic", "execution-date": "2019-11-28T00:00:00+00:00"}
[2019-11-29 10:10:41,216] {base_task_runner.py:126} INFO - Running: ['airflow', 'tasks', 'run', 'example_gcp_pubsub', 'create_topic', '2019-11-28T00:00:00+00:00', '--job_id', '2', '--pool', 'default_pool', '--raw', '-sd', '/opt/airflow/airflow/providers/google/cloud/example_dags/example_pubsub.py', '--cfg_path', '/tmp/tmpwrwdlsdv']

Tests

• My PR adds the following unit tests OR does not need testing for this extremely good reason:

Commits

• My commits all reference Jira issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters (not including Jira issue reference)
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• In case of new functionality, my PR adds documentation that describes how to use it.
  • All the public functions and the classes in the PR contain docstrings that explain what it does
  • If you implement backwards incompatible changes, please leave a note in the Updating.md so we can assign it to a appropriate release

[feluelle]

I am re-running the failed check:

=====================================================================

126) ERROR: test_on_kill (tests.task.task_runner.test_standard_task_runner.TestStandardTaskRunner)

----------------------------------------------------------------------

   Traceback (most recent call last):

    tests/task/task_runner/test_standard_task_runner.py line 129 in test_on_kill

      with open(path, "r") as f:

   FileNotFoundError: [Errno 2] No such file or directory: '/tmp/airflow_on_kill'

   -------------------- >> begin captured stdout << ---------------------

   [%(asctime)s] {{%(filename)s:%(lineno)d}} %(levelname)s - %(message)s

   [2019-11-29 11:12:48,056] {test_task_view_type_check.py:49} INFO - class_instance type: <class 'unusual_prefix_5d280a9b385120fec3c40cfe5be04e2f41b6b5e8_test_task_view_type_check.CallableClass'>

   [%(asctime)s] {{%(filename)s:%(lineno)d}} %(levelname)s - %(message)s

   [%(asctime)s] {{%(filename)s:%(lineno)d}} %(levelname)s - %(message)s

   [%(asctime)s] {{%(filename)s:%(lineno)d}} %(levelname)s - %(message)s

   [%(asctime)s] {{%(filename)s:%(lineno)d}} %(levelname)s - %(message)s

   [%(asctime)s] {{%(filename)s:%(lineno)d}} %(levelname)s - %(message)s

   [%(asctime)s] {{%(filename)s:%(lineno)d}} %(levelname)s - %(message)s

   [%(asctime)s] {{%(filename)s:%(lineno)d}} %(levelname)s - %(message)s

   

   --------------------- >> end captured stdout << ----------------------

   -------------------- >> begin captured logging << --------------------

   root: INFO: class_instance type: <class 'unusual_prefix_5d280a9b385120fec3c40cfe5be04e2f41b6b5e8_test_task_view_type_check.CallableClass'>

   --------------------- >> end captured logging << ---------------------

[feluelle]

Would like to see how flaky this test_on_kill actually is.

[turbaszek]

@feluelle I think that extending sleep time in this test seems to work. It's a test that runs a DAG and terminates it, imho there's no guarantee that everything will be done in 3s.

[codecov-io]

Codecov Report

Merging #6688 into master will decrease coverage by 0.31%.
The diff coverage is 100%.

@@            Coverage Diff             @@
##           master    #6688      +/-   ##
==========================================
- Coverage   83.82%   83.51%   -0.32%     
==========================================
  Files         668      668              
  Lines       37543    37557      +14     
==========================================
- Hits        31472    31367     -105     
- Misses       6071     6190     +119

Impacted Files	Coverage Δ
airflow/task/task_runner/base_task_runner.py	73.77% <100%> (+0.88%)	⬆️
airflow/kubernetes/volume_mount.py	44.44% <0%> (-55.56%)	⬇️
airflow/kubernetes/volume.py	52.94% <0%> (-47.06%)	⬇️
airflow/kubernetes/pod_launcher.py	45.25% <0%> (-46.72%)	⬇️
airflow/kubernetes/refresh_config.py	50.98% <0%> (-23.53%)	⬇️
...rflow/contrib/operators/kubernetes_pod_operator.py	78.2% <0%> (-20.52%)	⬇️
airflow/configuration.py	89.13% <0%> (-3.63%)	⬇️
airflow/models/dagbag.py	85.99% <0%> (-0.54%)	⬇️
airflow/jobs/local_task_job.py	90% <0%> (+5%)	⬆️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update e82008d...3561f21. Read the comment docs.