[AIRFLOW-3099] Don't ever warn about missing sections of config (apache#4028)

Rather than looping through and setting each config variable
individually, and having to know which sections are optional and which
aren't, instead we can just call a single function on ConfigParser and
it will read the config from the dict, and more importantly here, never
error about missing sections - it will just create them as needed.

Author: ashb

airflow/bin/cli.py

@@ -19,7 +19,6 @@
 # under the License.
 
 from __future__ import print_function
-from backports.configparser import NoSectionError
 import logging
 
 import os
@@ -488,24 +487,6 @@ def _run(args, dag, ti):
 
 @cli_utils.action_logging
 def run(args, dag=None):
-    # Optional sections won't log an error if they're missing in airflow.cfg.
-    OPTIONAL_AIRFLOW_CFG_SECTIONS = [
-        'atlas',
-        'celery',
-        'celery_broker_transport_options',
-        'dask',
-        'elasticsearch',
-        'github_enterprise',
-        'hive',
-        'kerberos',
-        'kubernetes',
-        'kubernetes_node_selectors',
-        'kubernetes_secrets',
-        'ldap',
-        'lineage',
-        'mesos',
-    ]
-
     if dag:
         args.dag_id = dag.dag_id
 
@@ -519,25 +500,7 @@ def run(args, dag=None):
         if os.path.exists(args.cfg_path):
             os.remove(args.cfg_path)
 
-        # Do not log these properties since some may contain passwords.
-        # This may also set default values for database properties like
-        # core.sql_alchemy_pool_size
-        # core.sql_alchemy_pool_recycle
-        for section, config in conf_dict.items():
-            for option, value in config.items():
-                try:
-                    conf.set(section, option, value)
-                except NoSectionError:
-                    no_section_msg = (
-                        'Section {section} Option {option} '
-                        'does not exist in the config!'
-                    ).format(section=section, option=option)
-
-                    if section in OPTIONAL_AIRFLOW_CFG_SECTIONS:
-                        log.debug(no_section_msg)
-                    else:
-                        log.error(no_section_msg)
-
+        conf.conf.read_dict(conf_dict, source=args.cfg_path)
         settings.configure_vars()
 
     # IMPORTANT, have to use the NullPool, otherwise, each "run" command may leave

airflow/configuration.py

@@ -286,6 +286,10 @@ def read(self, filenames):
         super(AirflowConfigParser, self).read(filenames)
         self._validate()
 
+    def read_dict(self, *args, **kwargs):
+        super(AirflowConfigParser, self).read_dict(*args, **kwargs)
+        self._validate()
+
     def has_option(self, section, option):
         try:
             # Using self.get() to avoid reimplementing the priority order