Dask SSH not working

I'm trying to create a dask cluster between my machines using Dask SSH, but I'm having a hard time making the following work on my host without errors:

```python
cluster = SSHCluster(
["localhost", "vinhtran@vinhtran"],
connect_options={"known_hosts": None},
worker_options={"nthreads": 10},
#scheduler_options={}
)
```

vinhtran@vinhtran is the ssh login to the other machine known_hosts is what I think the keys it uses i'm not sure, but I get the following errors:
(minus the dask cluster, my machines can communicate with each other through ssh without username/passwords by copying over the keys.

Can you help me resolve this issues? I've already tried Kubernetes and Hadoop and failed miserable

```
distributed.deploy.ssh - INFO - distributed.scheduler - INFO - -----------------------------------------------
distributed.deploy.ssh - INFO - distributed.scheduler - INFO - Local Directory: /tmp/scheduler-mwn0mopp
distributed.deploy.ssh - INFO - distributed.scheduler - INFO - -----------------------------------------------
distributed.deploy.ssh - INFO - distributed.scheduler - INFO - Clear task state
distributed.deploy.ssh - INFO - distributed.scheduler - INFO - Scheduler at: tcp://192.168.1.191:8786
Task exception was never retrieved
future: exception=gaierror(-2, 'Name or service not known')>
Traceback (most recent call last):
File "/home/vinhdiesal/anaconda3/lib/python3.7/asyncio/tasks.py", line 630, in _wrap_awaitable
return (yield from awaitable.__await__())
File "/home/vinhdiesal/anaconda3/lib/python3.7/site-packages/distributed/deploy/spec.py", line 50, in _
await self.start()
File "/home/vinhdiesal/anaconda3/lib/python3.7/site-packages/distributed/deploy/ssh.py", line 88, in start
self.connection = await asyncssh.connect(self.address, **self.connect_options)
File "/home/vinhdiesal/anaconda3/lib/python3.7/site-packages/asyncssh/connection.py", line 5696, in connect
conn_factory, 'Opening SSH connection to')
File "/home/vinhdiesal/anaconda3/lib/python3.7/site-packages/asyncssh/connection.py", line 171, in _connect
local_addr=local_addr)
File "/home/vinhdiesal/anaconda3/lib/python3.7/asyncio/base_events.py", line 909, in create_connection
type=socket.SOCK_STREAM, proto=proto, flags=flags, loop=self)
File "/home/vinhdiesal/anaconda3/lib/python3.7/asyncio/base_events.py", line 1286, in _ensure_resolved
proto=proto, flags=flags)
File "/home/vinhdiesal/anaconda3/lib/python3.7/asyncio/base_events.py", line 788, in getaddrinfo
None, getaddr_func, host, port, family, type, proto, flags)
File "/home/vinhdiesal/anaconda3/lib/python3.7/concurrent/futures/thread.py", line 57, in run
result = self.fn(*self.args, **self.kwargs)
File "/home/vinhdiesal/anaconda3/lib/python3.7/socket.py", line 752, in getaddrinfo
for res in _socket.getaddrinfo(host, port, family, type, proto, flags):
socket.gaierror: [Errno -2] Name or service not known
---------------------------------------------------------------------------
gaierror Traceback (most recent call last)
in
2 ["localhost", "vinhtran@vinhtran"],
3 connect_options={"known_hosts": None},
----> 4 worker_options={"nthreads": 10},
5 #scheduler_options={}
6 )

~/anaconda3/lib/python3.7/site-packages/distributed/deploy/ssh.py in SSHCluster(hosts, connect_options, worker_options, scheduler_options, worker_module, **kwargs)
290 for i, host in enumerate(hosts[1:])
291 }
--> 292 return SpecCluster(workers, scheduler, name="SSHCluster", **kwargs)

~/anaconda3/lib/python3.7/site-packages/distributed/deploy/spec.py in __init__(self, workers, scheduler, worker, asynchronous, loop, security, silence_logs, name)
255 self._loop_runner.start()
256 self.sync(self._start)
--> 257 self.sync(self._correct_state)
258
259 async def _start(self):

~/anaconda3/lib/python3.7/site-packages/distributed/deploy/cluster.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
158 return future
159 else:
--> 160 return sync(self.loop, func, *args, **kwargs)
161
162 async def _logs(self, scheduler=True, workers=True):

~/anaconda3/lib/python3.7/site-packages/distributed/utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
346 if error[0]:
347 typ, exc, tb = error[0]
--> 348 raise exc.with_traceback(tb)
349 else:
350 return result[0]

~/anaconda3/lib/python3.7/site-packages/distributed/utils.py in f()
330 if callback_timeout is not None:
331 future = asyncio.wait_for(future, callback_timeout)
--> 332 result[0] = yield future
333 except Exception as exc:
334 error[0] = sys.exc_info()

~/anaconda3/lib/python3.7/site-packages/tornado/gen.py in run(self)
733
734 try:
--> 735 value = future.result()
736 except Exception:
737 exc_info = sys.exc_info()

~/anaconda3/lib/python3.7/site-packages/distributed/deploy/spec.py in _correct_state_internal(self)
333 for w in workers:
334 w._cluster = weakref.ref(self)
--> 335 await w # for tornado gen.coroutine support
336 self.workers.update(dict(zip(to_open, workers)))
337

~/anaconda3/lib/python3.7/site-packages/distributed/deploy/spec.py in _()
48 async with self.lock:
49 if self.status == "created":
---> 50 await self.start()
51 assert self.status == "running"
52 return self

~/anaconda3/lib/python3.7/site-packages/distributed/deploy/ssh.py in start(self)
86 import asyncssh # import now to avoid adding to module startup time
87
---> 88 self.connection = await asyncssh.connect(self.address, **self.connect_options)
89 self.proc = await self.connection.create_process(
90 " ".join(

~/anaconda3/lib/python3.7/site-packages/asyncssh/connection.py in connect(host, port, tunnel, family, flags, local_addr, options, **kwargs)
5694
5695 return await _connect(host, port, loop, tunnel, family, flags, local_addr,
-> 5696 conn_factory, 'Opening SSH connection to')
5697
5698

~/anaconda3/lib/python3.7/site-packages/asyncssh/connection.py in _connect(host, port, loop, tunnel, family, flags, local_addr, conn_factory, msg)
169 _, conn = await loop.create_connection(conn_factory, host, port,
170 family=family, flags=flags,
--> 171 local_addr=local_addr)
172
173 # pylint: disable=broad-except

~/anaconda3/lib/python3.7/asyncio/base_events.py in create_connection(self, protocol_factory, host, port, ssl, family, proto, flags, sock, local_addr, server_hostname, ssl_handshake_timeout)
907 infos = await self._ensure_resolved(
908 (host, port), family=family,
--> 909 type=socket.SOCK_STREAM, proto=proto, flags=flags, loop=self)
910 if not infos:
911 raise OSError('getaddrinfo() returned empty list')

~/anaconda3/lib/python3.7/asyncio/base_events.py in _ensure_resolved(self, address, family, type, proto, flags, loop)
1284 else:
1285 return await loop.getaddrinfo(host, port, family=family, type=type,
-> 1286 proto=proto, flags=flags)
1287
1288 async def _create_server_getaddrinfo(self, host, port, family, flags):

~/anaconda3/lib/python3.7/asyncio/base_events.py in getaddrinfo(self, host, port, family, type, proto, flags)
786
787 return await self.run_in_executor(
--> 788 None, getaddr_func, host, port, family, type, proto, flags)
789
790 async def getnameinfo(self, sockaddr, flags=0):

~/anaconda3/lib/python3.7/concurrent/futures/thread.py in run(self)
55
56 try:
---> 57 result = self.fn(*self.args, **self.kwargs)
58 except BaseException as exc:
59 self.future.set_exception(exc)

~/anaconda3/lib/python3.7/socket.py in getaddrinfo(host, port, family, type, proto, flags)
750 # and socket type values to enum constants.
751 addrlist = []
--> 752 for res in _socket.getaddrinfo(host, port, family, type, proto, flags):
753 af, socktype, proto, canonname, sa = res
754 addrlist.append((_intenum_converter(af, AddressFamily),

gaierror: [Errno -2] Name or service not known
```