diff --git a/.gitignore b/.gitignore
index 1111f0e36b004..876424a9ed42b 100644
--- a/.gitignore
+++ b/.gitignore
@@ -124,3 +124,6 @@ ENV/
 
 # Spark
 rat-results.txt
+
+# IDE History
+.history
diff --git a/airflow/bin/cli.py b/airflow/bin/cli.py
index f568d5dc8ea95..8ca4490064539 100755
--- a/airflow/bin/cli.py
+++ b/airflow/bin/cli.py
@@ -90,7 +90,17 @@ def sigquit_handler(sig, frame):
 
 def setup_logging(filename):
     root = logging.getLogger()
-    handler = logging.FileHandler(filename)
+    log_handler_kargs = None
+    lh_mod_name, lh_cls_name = conf.get('core', 'log_handler_class').rsplit('.', 1)
+    lh_mod = __import__(lh_mod_name, globals(), locals(), [lh_cls_name], 0)
+    log_handler = getattr(lh_mod, lh_cls_name)
+    try:
+        log_handler_kargs = json.loads(conf.get('core', 'log_handler_params'))
+        handler = log_handler(filename, **log_handler_kargs)
+    except ValueError:
+        handler = log_handler(filename)
+    logging.info('log_handler: {}'.format(log_handler))
+    logging.info('log_handler_kargs: {}'.format(log_handler_kargs))
     formatter = logging.Formatter(settings.SIMPLE_LOG_FORMAT)
     handler.setFormatter(formatter)
     root.addHandler(handler)
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index ddd1ba816e84b..7fa4a941e3095 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -55,6 +55,10 @@ log_format_with_pid = [%%(asctime)s] [%%(process)d] {{%%(filename)s:%%(lineno)d}
 log_format_with_thread_name = [%%(asctime)s] {{%%(filename)s:%%(lineno)d}} %%(threadName)s %%(levelname)s - %%(message)s
 simple_log_format = %%(asctime)s %%(levelname)s - %%(message)s
 
+# Log Handler (supports TimedRotatingFileHandler, RotatingFileHandler, WatchedFileHandler, FileHandler or StreamHandler)
+log_handler_class = logging.FileHandler
+log_handler_params = None  # valid json format required or ignored.
+
 # The executor class that airflow should use. Choices include
 # SequentialExecutor, LocalExecutor, CeleryExecutor, DaskExecutor
 executor = SequentialExecutor