[AIRFLOW-6260] Drive _cmd config option by env var

This improves the ability to configure AirFlow
using Kubernetes best practices. You can provide
for exemple AIRFLOW__CORE__SQL_ALCHEMY_CONN_CMD
referencing a shell script that computes the
connection string using Kubernetes secrets.
And that script can be provided to the container
using a configmap.

Adding a unit test to check that an option that
should NOT be overriden by a command is correctly
only read from the configuration.

airflow/configuration.py

@@ -184,6 +184,12 @@ def _get_env_var_option(self, section, key):
         env_var = self._env_var_name(section, key)
         if env_var in os.environ:
             return expand_env_var(os.environ[env_var])
+        # alternatively AIRFLOW__{SECTION}__{KEY}_CMD (for a command)
+        env_var_cmd = env_var + '_CMD'
+        if env_var_cmd in os.environ:
+            # if this is a valid command key...
+            if (section, key) in self.as_command_stdout:
+                return run_command(os.environ[env_var_cmd])
 
     def _get_cmd_option(self, section, key):
         fallback_key = key + '_cmd'

docs/howto/set-config.rst

@@ -58,11 +58,19 @@ The following config options support this ``_cmd`` version:
 * ``bind_password`` in ``[ldap]`` section
 * ``git_password`` in ``[kubernetes]`` section
 
+The ``_cmd`` config options can also be set using a corresponding environment variable
+the same way the usual config options can. For example: 
+
+.. code-block:: bash
+
+    export AIRFLOW__CORE__SQL_ALCHEMY_CONN_CMD=bash_command_to_run
+
 The idea behind this is to not store passwords on boxes in plain text files.
 
 The universal order of precedence for all configuration options is as follows:
 
 #. set as an environment variable
+#. set as a command environment variable
 #. set in ``airflow.cfg``
 #. command in ``airflow.cfg``
 #. Airflow's built in defaults

tests/test_configuration.py

@@ -31,7 +31,9 @@
 
 @unittest.mock.patch.dict('os.environ', {
     'AIRFLOW__TESTSECTION__TESTKEY': 'testvalue',
-    'AIRFLOW__TESTSECTION__TESTPERCENT': 'with%percent'
+    'AIRFLOW__TESTSECTION__TESTPERCENT': 'with%percent',
+    'AIRFLOW__TESTCMDENV__ITSACOMMAND_CMD': 'echo -n "OK"',
+    'AIRFLOW__TESTCMDENV__NOTACOMMAND_CMD': 'echo -n "NOT OK"'
 })
 class TestConf(unittest.TestCase):
 
@@ -421,3 +423,21 @@ def test_deprecated_funcs(self):
             with mock.patch('airflow.configuration.{}'.format(func)):
                 with self.assertWarns(DeprecationWarning):
                     getattr(configuration, func)()
+
+    def test_command_from_env(self):
+        TEST_CMDENV_CONFIG = '''[testcmdenv]
+itsacommand = NOT OK
+notacommand = OK
+'''
+        test_cmdenv_conf = AirflowConfigParser()
+        test_cmdenv_conf.read_string(TEST_CMDENV_CONFIG)
+        test_cmdenv_conf.as_command_stdout.add(('testcmdenv', 'itsacommand'))
+        with unittest.mock.patch.dict('os.environ'):
+            # AIRFLOW__TESTCMDENV__ITSACOMMAND_CMD maps to ('testcmdenv', 'itsacommand') in
+            # as_command_stdout and therefore should return 'OK' from the environment variable's
+            # echo command, and must not return 'NOT OK' from the configuration
+            self.assertEqual(test_cmdenv_conf.get('testcmdenv', 'itsacommand'), 'OK')
+            # AIRFLOW__TESTCMDENV__NOTACOMMAND_CMD maps to no entry in as_command_stdout and therefore
+            # the option should return 'OK' from the configuration, and must not return 'NOT OK' from
+            # the environement variable's echo command
+            self.assertEqual(test_cmdenv_conf.get('testcmdenv', 'notacommand'), 'OK')