[AIRFLOW-5282] Add default timeout on kubeclient & catch HTTPError

[rolanddb]

Make sure you have checked all steps below.

Jira

• My PR addresses the following Airflow Jira issues:
  • https://issues.apache.org/jira/browse/AIRFLOW-5282

Description

This PR:

• Adds a default _request_timeout in airflow.cfg, so that default installations will not hang indefinitely
• Catches the errors in the scheduler, and retries scheduling the task, in a similar fashion to how ApiExcptions are caught. (

  airflow/airflow/executors/kubernetes_executor.py

  Lines 800 to 803 in 46885cc

  	except ApiException as e:
  	self.log.warning('ApiException when attempting to run task, re-queueing. '
  	'Message: %s' % json.loads(e.body)['message'])
  	self.task_queue.put(task)

  )
• Fixes a bug in the parsing of airflow.cfg. The notes state that a json dict is expected. This is parsed with json.loads, which returns a dict. But in the code, .iteritems() is called on this dict, which cannot work. .items() should be used instead.

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

Code Quality

• Passes flake8

[rolanddb]

@andy-g14 I added the timeout configuration from your unit test as a default. This works at runtime, but tests are complaining about a KeyError.

f70a6aa#diff-2fd5967071561042847ef71c5f8b7e8dR216

Did you ever see this work? I tried various changes, e.g. kube_client_request_args = "hello world" runs fine, but it seems that adding json there breaks things..

[ashb]

Suggested change

	kube_client_request_args = {"_request_timeout" : [60,60] }
	kube_client_request_args = {{"_request_timeout" : [60,60] }}

The problem is this file isn't a plain config file, it's actually a giant python format string.

[ashb]

Setting this in the ~/airflow/airflow.cfg wouldn't have the same problem, it's just this default config template that is the problem.

[andy-g14]

@rolanddb , you will need to provide the value as json string, i.e enclosed in single quotes. it works for me this way
kube_client_request_args = '{"_request_timeout" : [10,100] }'

[ashb]

The problem in the other PR seemed to be around putting a default value in our config template (which is actually a giant python format string so {} -> {{}})

[rolanddb]

The problem in the other PR seemed to be around putting a default value in our config template (which is actually a giant python format string so {} -> {{}})

Ah, I was very confused why it didn't work!

[ashb]

Nice!

Note that this new default config won't change the value for any existing installs, so some people might have to update their configs to get this new change.