diff --git a/distributed/client.py b/distributed/client.py
index 625d1fb1040..12e5c9c36cc 100644
--- a/distributed/client.py
+++ b/distributed/client.py
@@ -3494,8 +3494,8 @@ def get_scheduler_logs(self, n=None):
         Parameters
         ----------
         n : int
-            Number of logs to retrive.  Maxes out at 10000 by default,
-            confiruable in config.yaml::log-length
+            Number of logs to retrive.  Maxes out at 100000 by default,
+            confiruable in distributed.yaml::transition-log-length
 
         Returns
         -------
@@ -3509,8 +3509,8 @@ def get_worker_logs(self, n=None, workers=None, nanny=False):
         Parameters
         ----------
         n : int
-            Number of logs to retrive.  Maxes out at 10000 by default,
-            confiruable in config.yaml::log-length
+            Number of logs to retrive.  Set to 0 by default,
+            confiruable in distributed.yaml::recent-messages-log-length
         workers : iterable
             List of worker addresses to retrieve.  Gets all workers by default.
         nanny : bool, default False
diff --git a/distributed/distributed-schema.yaml b/distributed/distributed-schema.yaml
index 331fdd10ef0..14bd4ae8cf5 100644
--- a/distributed/distributed-schema.yaml
+++ b/distributed/distributed-schema.yaml
@@ -6,6 +6,13 @@ properties:
       version:
         type: integer
 
+      logging:
+        type: object
+        description: |
+          The logging levels at which various administrative events get logged using the
+          Python standard logging module. Both the logging level and logging handlers
+          are customizable
+
       scheduler:
         type: object
         properties:
diff --git a/distributed/distributed.yaml b/distributed/distributed.yaml
index fb01a168d38..bf31fd9d045 100644
--- a/distributed/distributed.yaml
+++ b/distributed/distributed.yaml
@@ -1,12 +1,12 @@
 distributed:
   version: 2
-  # logging:
-  #   distributed: info
-  #   distributed.client: warning
-  #   bokeh: critical
-  #   # http://stackoverflow.com/questions/21234772/python-tornado-disable-logging-to-stderr
-  #   tornado: critical
-  #   tornado.application: error
+  logging:
+    distributed: info
+    distributed.client: warning
+    bokeh: critical
+    # http://stackoverflow.com/questions/21234772/python-tornado-disable-logging-to-stderr
+    tornado: critical
+    tornado.application: error
 
   scheduler:
     allowed-failures: 3     # number of retries before a task is considered bad
diff --git a/distributed/tests/test_config.py b/distributed/tests/test_config.py
index 74b57b1f011..507dfcfd258 100644
--- a/distributed/tests/test_config.py
+++ b/distributed/tests/test_config.py
@@ -152,7 +152,14 @@ def test_logging_simple():
     """
     Test simple ("old-style") logging configuration.
     """
-    c = {"logging": {"distributed.foo": "info", "distributed.foo.bar": "error"}}
+    c = {
+        "distributed": {
+            "logging": {
+                "distributed.foo": "info",
+                "distributed.foo.bar": "error",
+            }
+        }
+    }
     # Must test using a subprocess to avoid wrecking pre-existing configuration
     with new_config_file(c):
         code = """if 1:
@@ -188,29 +195,31 @@ def test_logging_extended():
     Test extended ("new-style") logging configuration.
     """
     c = {
-        "logging": {
-            "version": "1",
-            "formatters": {
-                "simple": {"format": "%(levelname)s: %(name)s: %(message)s"}
-            },
-            "handlers": {
-                "console": {
-                    "class": "logging.StreamHandler",
-                    "stream": "ext://sys.stderr",
-                    "formatter": "simple",
-                }
-            },
-            "loggers": {
-                "distributed.foo": {
-                    "level": "INFO",
-                    #'handlers': ['console'],
+        "distributed": {
+            "logging": {
+                "version": "1",
+                "formatters": {
+                    "simple": {"format": "%(levelname)s: %(name)s: %(message)s"}
                 },
-                "distributed.foo.bar": {
-                    "level": "ERROR",
-                    #'handlers': ['console'],
+                "handlers": {
+                    "console": {
+                        "class": "logging.StreamHandler",
+                        "stream": "ext://sys.stderr",
+                        "formatter": "simple",
+                    }
                 },
-            },
-            "root": {"level": "WARNING", "handlers": ["console"]},
+                "loggers": {
+                    "distributed.foo": {
+                        "level": "INFO",
+                        #'handlers': ['console'],
+                    },
+                    "distributed.foo.bar": {
+                        "level": "ERROR",
+                        #'handlers': ['console'],
+                    },
+                },
+                "root": {"level": "WARNING", "handlers": ["console"]},
+            }
         }
     }
     # Must test using a subprocess to avoid wrecking pre-existing configuration
@@ -294,17 +303,25 @@ def test_logging_file_config():
     with tempfile.NamedTemporaryFile(mode="w", delete=False) as logging_config:
         logging_config.write(logging_config_contents)
     dask_config = {"logging-file-config": logging_config.name}
-    with new_config_file(dask_config):
-        code = """if 1:
-            import logging
-            from distributed import config
-            foo = logging.getLogger('foo')
-            bar = logging.getLogger('foo.bar')
-            assert logging.INFO == foo.getEffectiveLevel()
-            assert logging.ERROR == bar.getEffectiveLevel()
-            """
-        subprocess.check_call([sys.executable, "-c", code])
-    os.remove(logging_config.name)
+
+    try:
+        with new_config_file(dask_config):
+            code = """if 1:
+                import logging
+                from distributed import config
+
+                foo = logging.getLogger('foo')
+                bar = logging.getLogger('foo.bar')
+
+                expected = logging.INFO, logging.ERROR
+                actual = foo.getEffectiveLevel(), bar.getEffectiveLevel()
+
+                assert expected == actual, "{} != {}".format(expected, actual)
+                """
+            subprocess.check_call([sys.executable, "-c", code])
+
+    finally:
+        os.remove(logging_config.name)
 
 
 def test_schema():
@@ -331,7 +348,7 @@ def test_schema_is_complete():
     with open(schema_fn) as f:
         schema = yaml.safe_load(f)
 
-    skip = {"default-task-durations", "bokeh-application"}
+    skip = {"default-task-durations", "bokeh-application", "logging"}
 
     def test_matches(c, s):
         if set(c) != set(s["properties"]):