[AIRFLOW-6885] Delete worker on success

[dimberman]

Users now have the option to only delete worker pods when they are successful

---

Issue link: AIRFLOW-6885

Make sure to mark the boxes below before creating PR: [x]

• Description above provides context of the change
• Commit message/PR title starts with [AIRFLOW-NNNN]. AIRFLOW-NNNN = JIRA ID^*
• Unit tests coverage for changes (not needed for documentation changes)
• Commits follow "How to write a good git commit message"
• Relevant documentation is updated including usage instructions.
• I will engage committers as explained in Contribution Workflow Example.

^* For document-only changes commit message can start with [AIRFLOW-XXXX].

---

In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.

[ashb]

Like the feature idea, but what do you think of using a single (existing) config option with possible values True, False, and only_on_success.

This may be a bag idea because of the config API, but might be more user friendly?

[kaxil]

Travis failed:

tests/executors/test_kubernetes_executor.py ........F
_______________ TestKubernetesExecutor.test_change_state_failed ________________
self = <tests.executors.test_kubernetes_executor.TestKubernetesExecutor testMethod=test_change_state_failed>
mock_delete_pod = <MagicMock name='delete_pod' id='140527588788768'>
mock_get_kube_client = <MagicMock name='get_kube_client' id='140527589891936'>
mock_kubernetes_job_watcher = <MagicMock name='KubernetesJobWatcher' id='140527589720472'>
    @mock.patch('airflow.executors.kubernetes_executor.KubernetesJobWatcher')
    @mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
    @mock.patch('airflow.executors.kubernetes_executor.AirflowKubernetesScheduler.delete_pod')
    def test_change_state_failed(self, mock_delete_pod, mock_get_kube_client, mock_kubernetes_job_watcher):
        executor = KubernetesExecutor()
        executor.start()
        test_time = timezone.utcnow()
        key = ('dag_id', 'task_id', test_time, 'try_number3')
        executor._change_state(key, State.FAILED, 'pod_id', 'default')
        self.assertTrue(executor.event_buffer[key] == State.FAILED)
>       mock_delete_pod.assert_not_called()
tests/executors/test_kubernetes_executor.py:266: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
_mock_self = <MagicMock name='delete_pod' id='140527588788768'>
    def assert_not_called(_mock_self):
        """assert that the mock was never called.
        """
        self = _mock_self
        if self.call_count != 0:
            msg = ("Expected '%s' to not have been called. Called %s times.%s"
                   % (self._mock_name or 'mock',
                      self.call_count,
                      self._calls_repr()))
>           raise AssertionError(msg)
E           AssertionError: Expected 'delete_pod' to not have been called. Called 1 times.
E           Calls: [call('pod_id', 'default')].
/usr/local/lib/python3.6/site-packages/mock/mock.py:871: AssertionError
----------------------------- Captured stdout call -----------------------------
[%(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
tests/executors/test_kubernetes_executor.py F
_________ TestKubernetesExecutor.test_change_state_failed_pod_deletion _________
self = <tests.executors.test_kubernetes_executor.TestKubernetesExecutor testMethod=test_change_state_failed_pod_deletion>
mock_delete_pod = <MagicMock name='delete_pod' id='140527588755552'>
mock_get_kube_client = <MagicMock name='get_kube_client' id='140527588795280'>
mock_kubernetes_job_watcher = <MagicMock name='KubernetesJobWatcher' id='140527588672288'>
    @mock.patch('airflow.executors.kubernetes_executor.KubernetesJobWatcher')
    @mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
    @mock.patch('airflow.executors.kubernetes_executor.AirflowKubernetesScheduler.delete_pod')
    def test_change_state_failed_pod_deletion(self, mock_delete_pod, mock_get_kube_client,
                                              mock_kubernetes_job_watcher):
        executor = KubernetesExecutor()
        executor.kube_config.delete_worker_pods_on_success = True
        executor.start()
        key = ('dag_id', 'task_id', 'ex_time', 'try_number2')
        executor._change_state(key, State.FAILED, 'pod_id', 'test-namespace')
        self.assertTrue(executor.event_buffer[key] == State.FAILED)
>       mock_delete_pod.assert_not_called()
tests/executors/test_kubernetes_executor.py:296: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
_mock_self = <MagicMock name='delete_pod' id='140527588755552'>
    def assert_not_called(_mock_self):
        """assert that the mock was never called.
        """
        self = _mock_self
        if self.call_count != 0:
            msg = ("Expected '%s' to not have been called. Called %s times.%s"
                   % (self._mock_name or 'mock',
                      self.call_count,
                      self._calls_repr()))
>           raise AssertionError(msg)
E           AssertionError: Expected 'delete_pod' to not have been called. Called 1 times.
E           Calls: [call('pod_id', 'test-namespace')].
/usr/local/lib/python3.6/site-packages/mock/mock.py:871: AssertionError