We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Occasional "KeyError: 'cpu'" error message in dscheduler logs
I am not sure how to reproduce this error but it seems that when I start a new cluster I sometimes get the following error in the logs of the scheduler:

```
[...]
distributed.scheduler - INFO - Register 10.112.2.4:53995
distributed.utils - ERROR - 'cpu'
Traceback (most recent call last):
  File "/work/miniconda/lib/python3.5/site-packages/distributed/utils.py", line 205, in log_errors
    yield
  File "/work/miniconda/lib/python3.5/site-packages/distributed/bokeh/status/server_lifecycle.py", line 59, in workers
    resource_append(messages['workers']['plot-data'], msg)
  File "/work/miniconda/lib/python3.5/site-packages/distributed/bokeh/worker_monitor.py", line 72, in resource_append
    lists[k].append(mean(pluck(k, L)) / 100)
  File "/work/miniconda/lib/python3.5/site-packages/distributed/bokeh/worker_monitor.py", line 87, in mean
    seq = list(seq)
KeyError: 'cpu'
distributed.scheduler - INFO - Starting worker compute stream, 10.112.18.3:58457
distributed.scheduler - INFO - Starting worker compute stream, 10.112.19.3:52639
distributed.scheduler - INFO - Starting worker compute stream, 10.112.5.4:35957
distributed.scheduler - INFO - Starting worker compute stream, 10.112.1.3:47780
distributed.scheduler - INFO - Starting worker compute stream, 10.112.8.4:36129
distributed.scheduler - INFO - Starting worker compute stream, 10.112.0.3:37636
distributed.scheduler - INFO - Starting worker compute stream, 10.112.16.4:48440
distributed.scheduler - INFO - Starting worker compute stream, 10.112.14.4:48649
distributed.scheduler - INFO - Starting worker compute stream, 10.112.4.3:43642
distributed.scheduler - INFO - Starting worker compute stream, 10.112.7.4:44042
distributed.scheduler - INFO - Starting worker compute stream, 10.112.3.3:49770
distributed.scheduler - INFO - Starting worker compute stream, 10.112.10.4:47595
distributed.scheduler - INFO - Starting worker compute stream, 10.112.2.3:37964
distributed.scheduler - INFO - Starting worker compute stream, 10.112.11.3:41369
distributed.scheduler - INFO - Starting worker compute stream, 10.112.9.3:54121
distributed.scheduler - INFO - Starting worker compute stream, 10.112.19.4:45283
distributed.scheduler - INFO - Starting worker compute stream, 10.112.14.3:41628
distributed.scheduler - INFO - Starting worker compute stream, 10.112.17.3:53510
distributed.scheduler - INFO - Starting worker compute stream, 10.112.2.4:55194
distributed.scheduler - INFO - Starting worker compute stream, 10.112.6.3:50701
bokeh.util.tornado - ERROR - Error thrown from periodic callback: KeyError('cpu',)
distributed.core - INFO - Connection from 10.112.16.3:39481 to Scheduler
```

This does seem to cause any problem using the cluster though.


--- 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/bokeh/worker_monitor.py`
Content:
```
1 from __future__ import print_function, division, absolute_import
2 
3 from collections import defaultdict
4 from itertools import chain
5 
6 from toolz import pluck
7 
8 from ..utils import ignoring
9 
10 with ignoring(ImportError):
11     from bokeh.models import (ColumnDataSource, DataRange1d, Range1d,
12             NumeralTickFormatter, LinearAxis)
13     from bokeh.palettes import Spectral9
14     from bokeh.plotting import figure
15 
16 
17 def resource_profile_plot(width=600, height=300):
18     names = ['time', 'cpu', 'memory-percent', 'network-send', 'network-recv']
19     source = ColumnDataSource({k: [] for k in names})
20 
21     x_range = DataRange1d(follow='end', follow_interval=30000, range_padding=0)
22     y_range = Range1d(0, 1)
23     p = figure(width=width, height=height, x_axis_type='datetime',
24                responsive=True, tools='xpan,xwheel_zoom,box_zoom,resize,reset',
25                x_range=x_range, y_range=y_range)
26     p.line(x='time', y='memory-percent', line_width=2, line_alpha=0.8,
27            color=Spectral9[7], legend='Memory', source=source)
28     p.line(x='time', y='cpu', line_width=2, line_alpha=0.8,
29            color=Spectral9[0], legend='CPU', source=source)
30     p.legend[0].location = 'top_left'
31     p.yaxis[0].formatter = NumeralTickFormatter(format="0 %")
32     p.min_border_right = 10
33     p.extra_y_ranges = {"network": DataRange1d(bounds=(0, None))}
34     p.add_layout(LinearAxis(y_range_name="network",
35                             axis_label="Throughput (MB/s)"),
36                  'right')
37     p.yaxis.axis_label_text_font_size = "10pt"
38 
39     p.line(x='time', y='network-send', line_width=2, line_alpha=0.8,
40            color=Spectral9[2], legend='Network Send', source=source,
41            y_range_name="network")
42     p.line(x='time', y='network-recv', line_width=2, line_alpha=0.8,
43            color=Spectral9[3], legend='Network Recv', source=source,
44            y_range_name="network")
45 
46     return source, p
47 
48 
49 def resource_profile_update(source, worker_buffer, times_buffer):
50     data = defaultdict(list)
51 
52     workers = sorted(list(set(chain(*list(w.keys() for w in worker_buffer)))))
53 
54     for name in ['cpu', 'memory-percent', 'network-send', 'network-recv']:
55         data[name] = [[msg[w][name] if w in msg and name in msg[w] else 'null'
56                        for msg in worker_buffer]
57                        for w in workers]
58 
59     data['workers'] = workers
60     data['times'] = [[t * 1000 if w in worker_buffer[i] else 'null'
61                       for i, t in enumerate(times_buffer)]
62                       for w in workers]
63 
64     source.data.update(data)
65 
66 
67 def resource_append(lists, msg):
68     L = list(msg.values())
69     if not L:
70         return
71     for k in ['cpu', 'memory-percent']:
72         lists[k].append(mean(pluck(k, L)) / 100)
73 
74     lists['time'].append(mean(pluck('time', L)) * 1000)
75     if len(lists['time']) >= 2:
76         t1, t2 = lists['time'][-2], lists['time'][-1]
77         interval = (t2 - t1) / 1000
78     else:
79         interval = 0.5
80     send = mean(pluck('network-send', L, 0))
81     lists['network-send'].append(send / 2**20 / (interval or 0.5))
82     recv = mean(pluck('network-recv', L, 0))
83     lists['network-recv'].append(recv / 2**20 / (interval or 0.5))
84 
85 
86 def mean(seq):
87     seq = list(seq)
88     return sum(seq) / len(seq)
89
```
--- 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/bokeh/worker_monitor.py b/distributed/bokeh/worker_monitor.py
--- a/distributed/bokeh/worker_monitor.py
+++ b/distributed/bokeh/worker_monitor.py
@@ -68,8 +68,11 @@
     L = list(msg.values())
     if not L:
         return
-    for k in ['cpu', 'memory-percent']:
-        lists[k].append(mean(pluck(k, L)) / 100)
+    try:
+        for k in ['cpu', 'memory-percent']:
+            lists[k].append(mean(pluck(k, L)) / 100)
+    except KeyError:  # initial messages sometimes lack resource data
+        return        # this is safe to skip
 
     lists['time'].append(mean(pluck('time', L)) * 1000)
     if len(lists['time']) >= 2:

{"golden_diff": "diff --git a/distributed/bokeh/worker_monitor.py b/distributed/bokeh/worker_monitor.py\n--- a/distributed/bokeh/worker_monitor.py\n+++ b/distributed/bokeh/worker_monitor.py\n@@ -68,8 +68,11 @@\n     L = list(msg.values())\n     if not L:\n         return\n-    for k in ['cpu', 'memory-percent']:\n-        lists[k].append(mean(pluck(k, L)) / 100)\n+    try:\n+        for k in ['cpu', 'memory-percent']:\n+            lists[k].append(mean(pluck(k, L)) / 100)\n+    except KeyError:  # initial messages sometimes lack resource data\n+        return        # this is safe to skip\n \n     lists['time'].append(mean(pluck('time', L)) * 1000)\n     if len(lists['time']) >= 2:\n", "issue": "Occasional \"KeyError: 'cpu'\" error message in dscheduler logs\nI am not sure how to reproduce this error but it seems that when I start a new cluster I sometimes get the following error in the logs of the scheduler:\n\n```\n[...]\ndistributed.scheduler - INFO - Register 10.112.2.4:53995\ndistributed.utils - ERROR - 'cpu'\nTraceback (most recent call last):\n  File \"/work/miniconda/lib/python3.5/site-packages/distributed/utils.py\", line 205, in log_errors\n    yield\n  File \"/work/miniconda/lib/python3.5/site-packages/distributed/bokeh/status/server_lifecycle.py\", line 59, in workers\n    resource_append(messages['workers']['plot-data'], msg)\n  File \"/work/miniconda/lib/python3.5/site-packages/distributed/bokeh/worker_monitor.py\", line 72, in resource_append\n    lists[k].append(mean(pluck(k, L)) / 100)\n  File \"/work/miniconda/lib/python3.5/site-packages/distributed/bokeh/worker_monitor.py\", line 87, in mean\n    seq = list(seq)\nKeyError: 'cpu'\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.18.3:58457\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.19.3:52639\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.5.4:35957\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.1.3:47780\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.8.4:36129\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.0.3:37636\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.16.4:48440\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.14.4:48649\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.4.3:43642\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.7.4:44042\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.3.3:49770\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.10.4:47595\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.2.3:37964\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.11.3:41369\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.9.3:54121\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.19.4:45283\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.14.3:41628\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.17.3:53510\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.2.4:55194\ndistributed.scheduler - INFO - Starting worker compute stream, 10.112.6.3:50701\nbokeh.util.tornado - ERROR - Error thrown from periodic callback: KeyError('cpu',)\ndistributed.core - INFO - Connection from 10.112.16.3:39481 to Scheduler\n```\n\nThis does seem to cause any problem using the cluster though.\n\n", "before_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom collections import defaultdict\nfrom itertools import chain\n\nfrom toolz import pluck\n\nfrom ..utils import ignoring\n\nwith ignoring(ImportError):\n    from bokeh.models import (ColumnDataSource, DataRange1d, Range1d,\n            NumeralTickFormatter, LinearAxis)\n    from bokeh.palettes import Spectral9\n    from bokeh.plotting import figure\n\n\ndef resource_profile_plot(width=600, height=300):\n    names = ['time', 'cpu', 'memory-percent', 'network-send', 'network-recv']\n    source = ColumnDataSource({k: [] for k in names})\n\n    x_range = DataRange1d(follow='end', follow_interval=30000, range_padding=0)\n    y_range = Range1d(0, 1)\n    p = figure(width=width, height=height, x_axis_type='datetime',\n               responsive=True, tools='xpan,xwheel_zoom,box_zoom,resize,reset',\n               x_range=x_range, y_range=y_range)\n    p.line(x='time', y='memory-percent', line_width=2, line_alpha=0.8,\n           color=Spectral9[7], legend='Memory', source=source)\n    p.line(x='time', y='cpu', line_width=2, line_alpha=0.8,\n           color=Spectral9[0], legend='CPU', source=source)\n    p.legend[0].location = 'top_left'\n    p.yaxis[0].formatter = NumeralTickFormatter(format=\"0 %\")\n    p.min_border_right = 10\n    p.extra_y_ranges = {\"network\": DataRange1d(bounds=(0, None))}\n    p.add_layout(LinearAxis(y_range_name=\"network\",\n                            axis_label=\"Throughput (MB/s)\"),\n                 'right')\n    p.yaxis.axis_label_text_font_size = \"10pt\"\n\n    p.line(x='time', y='network-send', line_width=2, line_alpha=0.8,\n           color=Spectral9[2], legend='Network Send', source=source,\n           y_range_name=\"network\")\n    p.line(x='time', y='network-recv', line_width=2, line_alpha=0.8,\n           color=Spectral9[3], legend='Network Recv', source=source,\n           y_range_name=\"network\")\n\n    return source, p\n\n\ndef resource_profile_update(source, worker_buffer, times_buffer):\n    data = defaultdict(list)\n\n    workers = sorted(list(set(chain(*list(w.keys() for w in worker_buffer)))))\n\n    for name in ['cpu', 'memory-percent', 'network-send', 'network-recv']:\n        data[name] = [[msg[w][name] if w in msg and name in msg[w] else 'null'\n                       for msg in worker_buffer]\n                       for w in workers]\n\n    data['workers'] = workers\n    data['times'] = [[t * 1000 if w in worker_buffer[i] else 'null'\n                      for i, t in enumerate(times_buffer)]\n                      for w in workers]\n\n    source.data.update(data)\n\n\ndef resource_append(lists, msg):\n    L = list(msg.values())\n    if not L:\n        return\n    for k in ['cpu', 'memory-percent']:\n        lists[k].append(mean(pluck(k, L)) / 100)\n\n    lists['time'].append(mean(pluck('time', L)) * 1000)\n    if len(lists['time']) >= 2:\n        t1, t2 = lists['time'][-2], lists['time'][-1]\n        interval = (t2 - t1) / 1000\n    else:\n        interval = 0.5\n    send = mean(pluck('network-send', L, 0))\n    lists['network-send'].append(send / 2**20 / (interval or 0.5))\n    recv = mean(pluck('network-recv', L, 0))\n    lists['network-recv'].append(recv / 2**20 / (interval or 0.5))\n\n\ndef mean(seq):\n    seq = list(seq)\n    return sum(seq) / len(seq)\n", "path": "distributed/bokeh/worker_monitor.py"}], "after_files": [{"content": "from __future__ import print_function, division, absolute_import\n\nfrom collections import defaultdict\nfrom itertools import chain\n\nfrom toolz import pluck\n\nfrom ..utils import ignoring\n\nwith ignoring(ImportError):\n    from bokeh.models import (ColumnDataSource, DataRange1d, Range1d,\n            NumeralTickFormatter, LinearAxis)\n    from bokeh.palettes import Spectral9\n    from bokeh.plotting import figure\n\n\ndef resource_profile_plot(width=600, height=300):\n    names = ['time', 'cpu', 'memory-percent', 'network-send', 'network-recv']\n    source = ColumnDataSource({k: [] for k in names})\n\n    x_range = DataRange1d(follow='end', follow_interval=30000, range_padding=0)\n    y_range = Range1d(0, 1)\n    p = figure(width=width, height=height, x_axis_type='datetime',\n               responsive=True, tools='xpan,xwheel_zoom,box_zoom,resize,reset',\n               x_range=x_range, y_range=y_range)\n    p.line(x='time', y='memory-percent', line_width=2, line_alpha=0.8,\n           color=Spectral9[7], legend='Memory', source=source)\n    p.line(x='time', y='cpu', line_width=2, line_alpha=0.8,\n           color=Spectral9[0], legend='CPU', source=source)\n    p.legend[0].location = 'top_left'\n    p.yaxis[0].formatter = NumeralTickFormatter(format=\"0 %\")\n    p.min_border_right = 10\n    p.extra_y_ranges = {\"network\": DataRange1d(bounds=(0, None))}\n    p.add_layout(LinearAxis(y_range_name=\"network\",\n                            axis_label=\"Throughput (MB/s)\"),\n                 'right')\n    p.yaxis.axis_label_text_font_size = \"10pt\"\n\n    p.line(x='time', y='network-send', line_width=2, line_alpha=0.8,\n           color=Spectral9[2], legend='Network Send', source=source,\n           y_range_name=\"network\")\n    p.line(x='time', y='network-recv', line_width=2, line_alpha=0.8,\n           color=Spectral9[3], legend='Network Recv', source=source,\n           y_range_name=\"network\")\n\n    return source, p\n\n\ndef resource_profile_update(source, worker_buffer, times_buffer):\n    data = defaultdict(list)\n\n    workers = sorted(list(set(chain(*list(w.keys() for w in worker_buffer)))))\n\n    for name in ['cpu', 'memory-percent', 'network-send', 'network-recv']:\n        data[name] = [[msg[w][name] if w in msg and name in msg[w] else 'null'\n                       for msg in worker_buffer]\n                       for w in workers]\n\n    data['workers'] = workers\n    data['times'] = [[t * 1000 if w in worker_buffer[i] else 'null'\n                      for i, t in enumerate(times_buffer)]\n                      for w in workers]\n\n    source.data.update(data)\n\n\ndef resource_append(lists, msg):\n    L = list(msg.values())\n    if not L:\n        return\n    try:\n        for k in ['cpu', 'memory-percent']:\n            lists[k].append(mean(pluck(k, L)) / 100)\n    except KeyError:  # initial messages sometimes lack resource data\n        return        # this is safe to skip\n\n    lists['time'].append(mean(pluck('time', L)) * 1000)\n    if len(lists['time']) >= 2:\n        t1, t2 = lists['time'][-2], lists['time'][-1]\n        interval = (t2 - t1) / 1000\n    else:\n        interval = 0.5\n    send = mean(pluck('network-send', L, 0))\n    lists['network-send'].append(send / 2**20 / (interval or 0.5))\n    recv = mean(pluck('network-recv', L, 0))\n    lists['network-recv'].append(recv / 2**20 / (interval or 0.5))\n\n\ndef mean(seq):\n    seq = list(seq)\n    return sum(seq) / len(seq)\n", "path": "distributed/bokeh/worker_monitor.py"}]}