We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Numeric environment variable configuration is broken
I recently found that a pipeline was throwing this error:
```
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/distributed/utils.py", line 623, in log_errors
    yield
  File "/usr/local/lib/python3.6/site-packages/distributed/scheduler.py", line 1591, in remove_worker
    if ts.suspicious > self.allowed_failures:
TypeError: '>' not supported between instances of 'int' and 'str'
```
The environment variable `DASK_ALLOWED_FAILURES=20` had been set. In a new terminal I ran the following:
```pya
>>> LocalCluster().scheduler.allowed_failures
'20'
```
There appears to be nothing in the code which ensures that the type of a configuration variable conforms to the expected type.

I'd be happy to submit a PR to fix this but I'm not really sure what form it should take. Some ideas:
1. Remove environment variable configuration completely (it doesn't seem to be documented?)
2. Use `pyyaml` to interpret environment variables the same way yaml interprets them
    - (introduces are hard dependency on `pyyaml`)
3. Approximate the above by just automatically casting all env var config variables to ints where possible
4. Make a `get_config` which casts its return value to be the same type as the provided default; replace `config.get` everywhere. e.g.
```py
def get_config(key, default=None):
    if default is not None:
        return type(default)(config.get(key, default=default))
    return config.get(key)
```
5. Change the `config` dict to be an instance of a subclass of dict which has `get()` overridden to behave as suggested above
6. Change `config.py` to use the example `yaml` file to correctly cast all config variables
7. Cast variables correctly at the point of `get` everywhere
    - e.g. `ALLOWED_FAILURES = int(config.get('allowed-failures', 3))`

I think I'd probably go with 3, which seems like the lowest impact solution.

Of course, the workaround for now is to just use a yaml configuration file.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `distributed/config.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 from contextlib import contextmanager
4 import logging
5 import logging.config
6 import os
7 import sys
8 import warnings
9 
10 from .compatibility import FileExistsError, logging_names
11 
12 logger = logging.getLogger(__name__)
13 
14 config = {}
15 
16 
17 def ensure_config_file(source, destination):
18     if not os.path.exists(destination):
19         import shutil
20         if not os.path.exists(os.path.dirname(destination)):
21             try:
22                 os.mkdir(os.path.dirname(destination))
23             except FileExistsError:
24                 pass
25         # Atomically create destination.  Parallel testing discovered
26         # a race condition where a process can be busy creating the
27         # destination while another process reads an empty config file.
28         tmp = '%s.tmp.%d' % (destination, os.getpid())
29         shutil.copy(source, tmp)
30         try:
31             os.rename(tmp, destination)
32         except OSError:
33             os.remove(tmp)
34 
35 
36 def determine_config_file():
37     path = os.environ.get('DASK_CONFIG')
38     if path:
39         if (os.path.exists(path) and
40                 (os.path.isfile(path) or os.path.islink(path))):
41             return path
42         warnings.warn("DASK_CONFIG set to '%s' but file does not exist "
43                       "or is not a regular file" % (path,),
44                       UserWarning)
45 
46     dirname = os.path.dirname(__file__)
47     default_path = os.path.join(dirname, 'config.yaml')
48     path = os.path.join(os.path.expanduser('~'), '.dask', 'config.yaml')
49 
50     try:
51         ensure_config_file(default_path, path)
52     except EnvironmentError as e:
53         warnings.warn("Could not write default config file to '%s'. "
54                       "Received error %s" % (path, e),
55                       UserWarning)
56 
57     return path if os.path.exists(path) else default_path
58 
59 
60 def load_config_file(config, path):
61     with open(path) as f:
62         text = f.read()
63         config.update(yaml.load(text) or {})
64 
65 
66 def load_env_vars(config):
67     for name, value in os.environ.items():
68         if name.startswith('DASK_'):
69             varname = name[5:].lower().replace('_', '-')
70             config[varname] = value
71 
72 
73 def _initialize_logging_old_style(config):
74     """
75     Initialize logging using the "old-style" configuration scheme, e.g.:
76         {
77         'logging': {
78             'distributed': 'info',
79             'tornado': 'critical',
80             'tornado.application': 'error',
81             }
82         }
83     """
84     loggers = {  # default values
85         'distributed': 'info',
86         'distributed.client': 'warning',
87         'bokeh': 'critical',
88         'tornado': 'critical',
89         'tornado.application': 'error',
90     }
91     loggers.update(config.get('logging', {}))
92 
93     handler = logging.StreamHandler(sys.stderr)
94     handler.setFormatter(logging.Formatter(log_format))
95     for name, level in loggers.items():
96         if isinstance(level, str):
97             level = logging_names[level.upper()]
98         logger = logging.getLogger(name)
99         logger.setLevel(level)
100         logger.handlers[:] = []
101         logger.addHandler(handler)
102         logger.propagate = False
103 
104 
105 def _initialize_logging_new_style(config):
106     """
107     Initialize logging using logging's "Configuration dictionary schema".
108     (ref.: https://docs.python.org/2/library/logging.config.html#logging-config-dictschema)
109     """
110     logging.config.dictConfig(config['logging'])
111 
112 
113 def _initialize_logging_file_config(config):
114     """
115     Initialize logging using logging's "Configuration file format".
116     (ref.: https://docs.python.org/2/library/logging.config.html#configuration-file-format)
117     """
118     logging.config.fileConfig(config['logging-file-config'], disable_existing_loggers=False)
119 
120 
121 def initialize_logging(config):
122     if 'logging-file-config' in config:
123         if 'logging' in config:
124             raise RuntimeError("Config options 'logging-file-config' and 'logging' are mutually exclusive.")
125         _initialize_logging_file_config(config)
126     else:
127         log_config = config.get('logging', {})
128         if 'version' in log_config:
129             # logging module mandates version to be an int
130             log_config['version'] = int(log_config['version'])
131             _initialize_logging_new_style(config)
132         else:
133             _initialize_logging_old_style(config)
134 
135 
136 @contextmanager
137 def set_config(arg=None, **kwargs):
138     if arg and not kwargs:
139         kwargs = arg
140     old = {}
141     for key in kwargs:
142         if key in config:
143             old[key] = config[key]
144 
145     for key, value in kwargs.items():
146         config[key] = value
147 
148     try:
149         yield
150     finally:
151         for key in kwargs:
152             if key in old:
153                 config[key] = old[key]
154             else:
155                 del config[key]
156 
157 
158 try:
159     import yaml
160 except ImportError:
161     pass
162 else:
163     path = determine_config_file()
164     load_config_file(config, path)
165 
166 load_env_vars(config)
167 
168 log_format = config.get('log-format', '%(name)s - %(levelname)s - %(message)s')
169 
170 initialize_logging(config)
171
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/distributed/config.py b/distributed/config.py
--- a/distributed/config.py
+++ b/distributed/config.py
@@ -67,7 +67,24 @@
     for name, value in os.environ.items():
         if name.startswith('DASK_'):
             varname = name[5:].lower().replace('_', '-')
-            config[varname] = value
+            config[varname] = _parse_env_value(value)
+
+
+def _parse_env_value(value):
+    """ Convert a string to an integer, float or boolean (in that order) if possible. """
+    bools = {
+        'true': True,
+        'false': False
+    }
+    try:
+        return int(value)
+    except ValueError:
+        pass
+    try:
+        return float(value)
+    except ValueError:
+        pass
+    return bools.get(value.lower(), value)
 
 
 def _initialize_logging_old_style(config):

{"golden_diff": "diff --git a/distributed/config.py b/distributed/config.py\n--- a/distributed/config.py\n+++ b/distributed/config.py\n@@ -67,7 +67,24 @@\n     for name, value in os.environ.items():\n         if name.startswith('DASK_'):\n             varname = name[5:].lower().replace('_', '-')\n-            config[varname] = value\n+            config[varname] = _parse_env_value(value)\n+\n+\n+def _parse_env_value(value):\n+    \"\"\" Convert a string to an integer, float or boolean (in that order) if possible. \"\"\"\n+    bools = {\n+        'true': True,\n+        'false': False\n+    }\n+    try:\n+        return int(value)\n+    except ValueError:\n+        pass\n+    try:\n+        return float(value)\n+    except ValueError:\n+        pass\n+    return bools.get(value.lower(), value)\n \n \n def _initialize_logging_old_style(config):\n", "issue": "Numeric environment variable configuration is broken\nI recently found that a pipeline was throwing this error:\r\n```\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.6/site-packages/distributed/utils.py\", line 623, in log_errors\r\n    yield\r\n  File \"/usr/local/lib/python3.6/site-packages/distributed/scheduler.py\", line 1591, in remove_worker\r\n    if ts.suspicious > self.allowed_failures:\r\nTypeError: '>' not supported between instances of 'int' and 'str'\r\n```\r\nThe environment variable `DASK_ALLOWED_FAILURES=20` had been set. In a new terminal I ran the following:\r\n```pya\r\n>>> LocalCluster().scheduler.allowed_failures\r\n'20'\r\n```\r\nThere appears to be nothing in the code which ensures that the type of a configuration variable conforms to the expected type.\r\n\r\nI'd be happy to submit a PR to fix this but I'm not really sure what form it should take. Some ideas:\r\n1. Remove environment variable configuration completely (it doesn't seem to be documented?)\r\n2. Use `pyyaml` to interpret environment variables the same way yaml interprets them\r\n    - (introduces are hard dependency on `pyyaml`)\r\n3. Approximate the above by just automatically casting all env var config variables to ints where possible\r\n4. Make a `get_config` which casts its return value to be the same type as the provided default; replace `config.get` everywhere. e.g.\r\n```py\r\ndef get_config(key, default=None):\r\n    if default is not None:\r\n        return type(default)(config.get(key, default=default))\r\n    return config.get(key)\r\n```\r\n5. Change the `config` dict to be an instance of a subclass of dict which has `get()` overridden to behave as suggested above\r\n6. Change `config.py` to use the example `yaml` file to correctly cast all config variables\r\n7. Cast variables correctly at the point of `get` everywhere\r\n    - e.g. `ALLOWED_FAILURES = int(config.get('allowed-failures', 3))`\r\n\r\nI think I'd probably go with 3, which seems like the lowest impact solution.\r\n\r\nOf course, the workaround for now is to just use a yaml configuration file.\r\n\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom contextlib import contextmanager\nimport logging\nimport logging.config\nimport os\nimport sys\nimport warnings\n\nfrom .compatibility import FileExistsError, logging_names\n\nlogger = logging.getLogger(__name__)\n\nconfig = {}\n\n\ndef ensure_config_file(source, destination):\n    if not os.path.exists(destination):\n        import shutil\n        if not os.path.exists(os.path.dirname(destination)):\n            try:\n                os.mkdir(os.path.dirname(destination))\n            except FileExistsError:\n                pass\n        # Atomically create destination.  Parallel testing discovered\n        # a race condition where a process can be busy creating the\n        # destination while another process reads an empty config file.\n        tmp = '%s.tmp.%d' % (destination, os.getpid())\n        shutil.copy(source, tmp)\n        try:\n            os.rename(tmp, destination)\n        except OSError:\n            os.remove(tmp)\n\n\ndef determine_config_file():\n    path = os.environ.get('DASK_CONFIG')\n    if path:\n        if (os.path.exists(path) and\n                (os.path.isfile(path) or os.path.islink(path))):\n            return path\n        warnings.warn(\"DASK_CONFIG set to '%s' but file does not exist \"\n                      \"or is not a regular file\" % (path,),\n                      UserWarning)\n\n    dirname = os.path.dirname(__file__)\n    default_path = os.path.join(dirname, 'config.yaml')\n    path = os.path.join(os.path.expanduser('~'), '.dask', 'config.yaml')\n\n    try:\n        ensure_config_file(default_path, path)\n    except EnvironmentError as e:\n        warnings.warn(\"Could not write default config file to '%s'. \"\n                      \"Received error %s\" % (path, e),\n                      UserWarning)\n\n    return path if os.path.exists(path) else default_path\n\n\ndef load_config_file(config, path):\n    with open(path) as f:\n        text = f.read()\n        config.update(yaml.load(text) or {})\n\n\ndef load_env_vars(config):\n    for name, value in os.environ.items():\n        if name.startswith('DASK_'):\n            varname = name[5:].lower().replace('_', '-')\n            config[varname] = value\n\n\ndef _initialize_logging_old_style(config):\n    \"\"\"\n    Initialize logging using the \"old-style\" configuration scheme, e.g.:\n        {\n        'logging': {\n            'distributed': 'info',\n            'tornado': 'critical',\n            'tornado.application': 'error',\n            }\n        }\n    \"\"\"\n    loggers = {  # default values\n        'distributed': 'info',\n        'distributed.client': 'warning',\n        'bokeh': 'critical',\n        'tornado': 'critical',\n        'tornado.application': 'error',\n    }\n    loggers.update(config.get('logging', {}))\n\n    handler = logging.StreamHandler(sys.stderr)\n    handler.setFormatter(logging.Formatter(log_format))\n    for name, level in loggers.items():\n        if isinstance(level, str):\n            level = logging_names[level.upper()]\n        logger = logging.getLogger(name)\n        logger.setLevel(level)\n        logger.handlers[:] = []\n        logger.addHandler(handler)\n        logger.propagate = False\n\n\ndef _initialize_logging_new_style(config):\n    \"\"\"\n    Initialize logging using logging's \"Configuration dictionary schema\".\n    (ref.: https://docs.python.org/2/library/logging.config.html#logging-config-dictschema)\n    \"\"\"\n    logging.config.dictConfig(config['logging'])\n\n\ndef _initialize_logging_file_config(config):\n    \"\"\"\n    Initialize logging using logging's \"Configuration file format\".\n    (ref.: https://docs.python.org/2/library/logging.config.html#configuration-file-format)\n    \"\"\"\n    logging.config.fileConfig(config['logging-file-config'], disable_existing_loggers=False)\n\n\ndef initialize_logging(config):\n    if 'logging-file-config' in config:\n        if 'logging' in config:\n            raise RuntimeError(\"Config options 'logging-file-config' and 'logging' are mutually exclusive.\")\n        _initialize_logging_file_config(config)\n    else:\n        log_config = config.get('logging', {})\n        if 'version' in log_config:\n            # logging module mandates version to be an int\n            log_config['version'] = int(log_config['version'])\n            _initialize_logging_new_style(config)\n        else:\n            _initialize_logging_old_style(config)\n\n\n@contextmanager\ndef set_config(arg=None, **kwargs):\n    if arg and not kwargs:\n        kwargs = arg\n    old = {}\n    for key in kwargs:\n        if key in config:\n            old[key] = config[key]\n\n    for key, value in kwargs.items():\n        config[key] = value\n\n    try:\n        yield\n    finally:\n        for key in kwargs:\n            if key in old:\n                config[key] = old[key]\n            else:\n                del config[key]\n\n\ntry:\n    import yaml\nexcept ImportError:\n    pass\nelse:\n    path = determine_config_file()\n    load_config_file(config, path)\n\nload_env_vars(config)\n\nlog_format = config.get('log-format', '%(name)s - %(levelname)s - %(message)s')\n\ninitialize_logging(config)\n", "path": "distributed/config.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom contextlib import contextmanager\nimport logging\nimport logging.config\nimport os\nimport sys\nimport warnings\n\nfrom .compatibility import FileExistsError, logging_names\n\nlogger = logging.getLogger(__name__)\n\nconfig = {}\n\n\ndef ensure_config_file(source, destination):\n    if not os.path.exists(destination):\n        import shutil\n        if not os.path.exists(os.path.dirname(destination)):\n            try:\n                os.mkdir(os.path.dirname(destination))\n            except FileExistsError:\n                pass\n        # Atomically create destination.  Parallel testing discovered\n        # a race condition where a process can be busy creating the\n        # destination while another process reads an empty config file.\n        tmp = '%s.tmp.%d' % (destination, os.getpid())\n        shutil.copy(source, tmp)\n        try:\n            os.rename(tmp, destination)\n        except OSError:\n            os.remove(tmp)\n\n\ndef determine_config_file():\n    path = os.environ.get('DASK_CONFIG')\n    if path:\n        if (os.path.exists(path) and\n                (os.path.isfile(path) or os.path.islink(path))):\n            return path\n        warnings.warn(\"DASK_CONFIG set to '%s' but file does not exist \"\n                      \"or is not a regular file\" % (path,),\n                      UserWarning)\n\n    dirname = os.path.dirname(__file__)\n    default_path = os.path.join(dirname, 'config.yaml')\n    path = os.path.join(os.path.expanduser('~'), '.dask', 'config.yaml')\n\n    try:\n        ensure_config_file(default_path, path)\n    except EnvironmentError as e:\n        warnings.warn(\"Could not write default config file to '%s'. \"\n                      \"Received error %s\" % (path, e),\n                      UserWarning)\n\n    return path if os.path.exists(path) else default_path\n\n\ndef load_config_file(config, path):\n    with open(path) as f:\n        text = f.read()\n        config.update(yaml.load(text) or {})\n\n\ndef load_env_vars(config):\n    for name, value in os.environ.items():\n        if name.startswith('DASK_'):\n            varname = name[5:].lower().replace('_', '-')\n            config[varname] = _parse_env_value(value)\n\n\ndef _parse_env_value(value):\n    \"\"\" Convert a string to an integer, float or boolean (in that order) if possible. \"\"\"\n    bools = {\n        'true': True,\n        'false': False\n    }\n    try:\n        return int(value)\n    except ValueError:\n        pass\n    try:\n        return float(value)\n    except ValueError:\n        pass\n    return bools.get(value.lower(), value)\n\n\ndef _initialize_logging_old_style(config):\n    \"\"\"\n    Initialize logging using the \"old-style\" configuration scheme, e.g.:\n        {\n        'logging': {\n            'distributed': 'info',\n            'tornado': 'critical',\n            'tornado.application': 'error',\n            }\n        }\n    \"\"\"\n    loggers = {  # default values\n        'distributed': 'info',\n        'distributed.client': 'warning',\n        'bokeh': 'critical',\n        'tornado': 'critical',\n        'tornado.application': 'error',\n    }\n    loggers.update(config.get('logging', {}))\n\n    handler = logging.StreamHandler(sys.stderr)\n    handler.setFormatter(logging.Formatter(log_format))\n    for name, level in loggers.items():\n        if isinstance(level, str):\n            level = logging_names[level.upper()]\n        logger = logging.getLogger(name)\n        logger.setLevel(level)\n        logger.handlers[:] = []\n        logger.addHandler(handler)\n        logger.propagate = False\n\n\ndef _initialize_logging_new_style(config):\n    \"\"\"\n    Initialize logging using logging's \"Configuration dictionary schema\".\n    (ref.: https://docs.python.org/2/library/logging.config.html#logging-config-dictschema)\n    \"\"\"\n    logging.config.dictConfig(config['logging'])\n\n\ndef _initialize_logging_file_config(config):\n    \"\"\"\n    Initialize logging using logging's \"Configuration file format\".\n    (ref.: https://docs.python.org/2/library/logging.config.html#configuration-file-format)\n    \"\"\"\n    logging.config.fileConfig(config['logging-file-config'], disable_existing_loggers=False)\n\n\ndef initialize_logging(config):\n    if 'logging-file-config' in config:\n        if 'logging' in config:\n            raise RuntimeError(\"Config options 'logging-file-config' and 'logging' are mutually exclusive.\")\n        _initialize_logging_file_config(config)\n    else:\n        log_config = config.get('logging', {})\n        if 'version' in log_config:\n            # logging module mandates version to be an int\n            log_config['version'] = int(log_config['version'])\n            _initialize_logging_new_style(config)\n        else:\n            _initialize_logging_old_style(config)\n\n\n@contextmanager\ndef set_config(arg=None, **kwargs):\n    if arg and not kwargs:\n        kwargs = arg\n    old = {}\n    for key in kwargs:\n        if key in config:\n            old[key] = config[key]\n\n    for key, value in kwargs.items():\n        config[key] = value\n\n    try:\n        yield\n    finally:\n        for key in kwargs:\n            if key in old:\n                config[key] = old[key]\n            else:\n                del config[key]\n\n\ntry:\n    import yaml\nexcept ImportError:\n    pass\nelse:\n    path = determine_config_file()\n    load_config_file(config, path)\n\nload_env_vars(config)\n\nlog_format = config.get('log-format', '%(name)s - %(levelname)s - %(message)s')\n\ninitialize_logging(config)\n", "path": "distributed/config.py"}]}