diff --git a/debian/control.in b/debian/control.in
index 19f590d06..64b0a4ce0 100644
--- a/debian/control.in
+++ b/debian/control.in
@@ -287,6 +287,7 @@ Depends:
 Suggests:
  python3-netaddr,
  python3-pyparsing,
+ python3-unbound,
 Description: Python 3 bindings for Open vSwitch
  Open vSwitch is a production quality, multilayer, software-based,
  Ethernet virtual switch. It is designed to enable massive network
diff --git a/python/TODO.rst b/python/TODO.rst
index 3a53489f1..acc5461e2 100644
--- a/python/TODO.rst
+++ b/python/TODO.rst
@@ -32,3 +32,10 @@ Python Bindings To-do List
 
   * Support write-only-changed monitor mode (equivalent of
     OVSDB_IDL_WRITE_CHANGED_ONLY).
+
+* socket_util:
+
+  * Add equivalent fuctions to inet_parse_passive, parse_sockaddr_components,
+    et al. to better support using async dns. The reconnect code will
+    currently log a warning when inet_parse_active() returns w/o yet having
+    resolved an address, but will continue to connect and eventually succeed.
diff --git a/python/automake.mk b/python/automake.mk
index d00911828..82a508787 100644
--- a/python/automake.mk
+++ b/python/automake.mk
@@ -16,6 +16,7 @@ ovs_pyfiles = \
 	python/ovs/compat/sortedcontainers/sorteddict.py \
 	python/ovs/compat/sortedcontainers/sortedset.py \
 	python/ovs/daemon.py \
+	python/ovs/dns_resolve.py \
 	python/ovs/db/__init__.py \
 	python/ovs/db/custom_index.py \
 	python/ovs/db/data.py \
@@ -55,6 +56,7 @@ ovs_pyfiles = \
 
 ovs_pytests = \
 	python/ovs/tests/test_decoders.py \
+	python/ovs/tests/test_dns_resolve.py \
 	python/ovs/tests/test_filter.py \
 	python/ovs/tests/test_kv.py \
 	python/ovs/tests/test_list.py \
diff --git a/python/ovs/dns_resolve.py b/python/ovs/dns_resolve.py
new file mode 100644
index 000000000..58638aae4
--- /dev/null
+++ b/python/ovs/dns_resolve.py
@@ -0,0 +1,257 @@
+import collections
+import functools
+import ipaddress
+import os
+import threading
+import time
+import typing
+
+try:
+    import unbound  # type: ignore
+except ImportError:
+    pass
+
+import ovs.vlog
+
+vlog = ovs.vlog.Vlog("dns_resolve")
+
+
+class DNSRequest:
+    INVALID = 0
+    PENDING = 1
+    GOOD = 2
+    ERROR = 3
+
+    def __init__(self, name):
+        self.name = name
+        self.state = self.INVALID
+        self.time = None
+        self.result = None  # set by DNSResolver._callback
+        self.ttl = None
+
+    @property
+    def expired(self):
+        return time.time() > self.time + self.ttl
+
+    @property
+    def is_valid(self):
+        return self.state == self.GOOD and not self.expired
+
+    def __str__(self):
+        return (f"DNSRequest(name={self.name}, state={self.state}, "
+                f"time={self.time}, result={self.result})")
+
+
+class DefaultReqDict(collections.defaultdict):
+    def __init__(self):
+        super().__init__(DNSRequest)
+
+    def __missing__(self, key):
+        ret = self.default_factory(key)
+        self[key] = ret
+        return ret
+
+
+class UnboundException(Exception):
+
+    def __init__(self, message, errno):
+        try:
+            msg = f"{message}: {unbound.ub_strerror(errno)}"
+        except NameError:
+            msg = message
+        super().__init__(msg)
+
+
+def dns_enabled(func):
+    @functools.wraps(func)
+    def wrapper(self, *args, **kwargs):
+        if self.dns_enabled:
+            return func(self, *args, **kwargs)
+        vlog.err("DNS support requires the python unbound library")
+    return wrapper
+
+
+class singleton:
+    def __init__(self, klass):
+        self._klass = klass
+        self._instance = None
+
+    def __call__(self, *args, **kwargs):
+        if self._instance is None:
+            self._instance = self._klass(*args, **kwargs)
+        return self._instance
+
+
+@singleton
+class DNSResolver:
+    def __init__(self, is_daemon: bool = False):
+        """Create a resolver instance
+
+        If is_daemon is true, set the resolver to handle requests
+        asynchronously. The following environment variables are processed:
+
+        OVS_UNBOUND_CONF: The filename for an unbound.conf file
+        OVS_RESOLVE_CONF: A filename to override the system default resolv.conf
+        OVS_HOSTS_FILE: A filename to override the system default hosts file
+
+        In the event that the unbound library is missing or fails to initialize
+        DNS lookup support will be disabled and the resolve() method will
+        return None.
+        """
+        self._is_daemon = is_daemon
+        try:
+            self._ctx = unbound.ub_ctx()
+            self.dns_enabled = True
+        except Exception:
+            # The unbound docs mention that this could thrown an exception
+            # but do not specify what exception that is. This can also
+            # happen with a missing unbound library
+            self.dns_enabled = False
+            vlog.err("Failed to initialize the unbound library")
+            return
+
+        # NOTE(twilson) This cache, like the C version, can grow without bound
+        # and has so cleanup or aging mechanism. Given our usage patterns, this
+        # should not be a problem. But this should not be used to resolve an
+        # unbounded list of addresses in a long-running daemon.
+        self._requests = DefaultReqDict()
+        self._lock = threading.RLock()
+
+        self._ub_call(self._set_unbound_conf)
+
+        # NOTE(twilson) The C version disables DNS in this case. I didn't do
+        # that here since it could still be useful to resolve addresses from
+        # /etc/hosts even w/o resolv.conf
+        self._ub_call(self._set_resolv_conf)
+        self._ub_call(self._set_hosts_file)
+
+        self._ctx.set_async(True)  # Sets threaded behavior for resolve_async()
+
+    def _ub_call(self, fn, *args, **kwargs):
+        """Convert UnboundExceptions into vlog warnings"""
+        try:
+            return fn(*args, **kwargs)
+        except UnboundException as e:
+            vlog.warn(e)
+
+    @dns_enabled
+    def _set_unbound_conf(self):
+        ub_cfg = os.getenv("OVS_UNBOUND_CONF")
+        if ub_cfg:
+            retval = self._ctx.config(ub_cfg)
+            if retval != 0:
+                raise UnboundException(
+                    "Failed to set libunbound context config", retval)
+
+    @dns_enabled
+    def _set_resolv_conf(self):
+        filename = os.getenv("OVS_RESOLV_CONF")
+        # The C lib checks that the file exists and also sets filename to
+        # /etc/resolv.conf on non-Windows, but resolvconf already does this
+        retval = self._ctx.resolvconf(filename)
+        if retval != 0:
+            location = filename or "system default nameserver"
+            raise UnboundException(location, retval)
+
+    @dns_enabled
+    def _set_hosts_file(self):
+        # The C lib doesn't have the ability to set a hosts file, but it is
+        # useful to have, especially for writing tests that don't rely on
+        # network connectivity. hosts(None) uses /etc/hosts.
+        filename = os.getenv("OVS_HOSTS_FILE")
+        retval = self._ctx.hosts(filename)
+        if retval != 0:
+            location = filename or "system default hosts file"
+            raise UnboundException(location, retval)
+
+    @dns_enabled
+    def _callback(self, req: DNSRequest, err: int, result):
+        if err != 0 or (result.qtype == unbound.RR_TYPE_AAAA
+                        and not result.havedata):
+            req.state = req.ERROR
+            vlog.warn(f"{req.name}: failed to resolve")
+            return
+        if result.qtype == unbound.RR_TYPE_A and not result.havedata:
+            self._resolve_async(req, unbound.RR_TYPE_AAAA)
+            return
+        try:
+            ip_str = next(iter(result.data.as_raw_data()))
+            ip = ipaddress.ip_address(ip_str)  # test if IP is valid
+            # NOTE (twilson) For some reason, accessing result data outside of
+            # _callback causes a segfault. So just grab and store what we need.
+            req.result = str(ip)
+            req.ttl = result.ttl
+            req.state = req.GOOD
+            req.time = time.time()
+        except (ValueError, StopIteration):
+            req.state = req.ERROR
+            vlog.err(f"{req.name}: failed to resolve")
+
+    @dns_enabled
+    def _resolve_sync(self, name: str) -> typing.Optional[str]:
+        for qtype in (unbound.RR_TYPE_A, unbound.RR_TYPE_AAAA):
+            err, result = self._ctx.resolve(name, qtype)
+            if err != 0:
+                return None
+            if not result.havedata:
+                continue
+            try:
+                ip = ipaddress.ip_address(
+                    next(iter(result.data.as_raw_data())))
+            except (ValueError, StopIteration):
+                return None
+            return str(ip)
+
+        return None
+
+    @dns_enabled
+    def _resolve_async(self, req: DNSRequest, qtype) -> None:
+        err, res = self._ctx.resolve_async(req.name, req, self._callback,
+                                           qtype)
+        if err != 0:
+            req.state = req.ERROR
+            return None
+
+        req.state = req.PENDING
+        return None
+
+    @dns_enabled
+    def resolve(self, name: str) -> typing.Optional[str]:
+        """Resolve a host name to an IP address
+
+        If the resolver is set to handle requests asynchronously, resolve()
+        should be recalled until it returns a non-None result. Errors will be
+        logged.
+
+        :param name: The host name to resolve
+        :returns: The IP address or None on error or not (yet) found
+        """
+        if not self._is_daemon:
+            return self._resolve_sync(name)
+        with self._lock:
+            retval = self._ctx.process()
+            if retval != 0:
+                vlog.err(f"dns-resolve error: {unbound.ub_strerror(retval)}")
+                return None
+            req = self._requests[name]  # Creates a DNSRequest if not found
+            if req.is_valid:
+                return req.result
+            elif req.state != req.PENDING:
+                self._resolve_async(req, unbound.RR_TYPE_A)
+        return None
+
+
+def resolve(name: str) -> typing.Optional[str]:
+    """Resolve a host name to an IP address
+
+    If a DNSResolver instance has not been instantiated, or if it has been
+    created with is_daemon=False, resolve() will synchronously resolve the
+    hostname. If DNSResolver has been initialized with is_daemon=True, it
+    will instead resolve asynchornously and resolve() will return None until
+    the hostname has been resolved.
+
+    :param name: The host name to resolve
+    :returns: The IP address or None on error or not (yet) found
+    """
+
+    return DNSResolver().resolve(name)
diff --git a/python/ovs/socket_util.py b/python/ovs/socket_util.py
index 7b41dc44b..a26298b75 100644
--- a/python/ovs/socket_util.py
+++ b/python/ovs/socket_util.py
@@ -13,12 +13,14 @@
 # limitations under the License.
 
 import errno
+import ipaddress
 import os
 import os.path
 import random
 import socket
 import sys
 
+from ovs import dns_resolve
 import ovs.fatal_signal
 import ovs.poller
 import ovs.vlog
@@ -216,7 +218,7 @@ def is_valid_ipv4_address(address):
     return True
 
 
-def inet_parse_active(target, default_port):
+def _inet_parse_active(target, default_port):
     address = target.split(":")
     if len(address) >= 2:
         host_name = ":".join(address[0:-1]).lstrip('[').rstrip(']')
@@ -229,9 +231,24 @@ def inet_parse_active(target, default_port):
         host_name = address[0]
     if not host_name:
         raise ValueError("%s: bad peer name format" % target)
+    try:
+        host_name = str(ipaddress.ip_address(host_name))
+    except ValueError:
+        host_name = dns_resolve.resolve(host_name)
+    if not host_name:
+        raise ValueError("%s: bad peer name format" % target)
     return (host_name, port)
 
 
+def inet_parse_active(target, default_port, raises=True):
+    try:
+        return _inet_parse_active(target, default_port)
+    except ValueError:
+        if raises:
+            raise
+        return ("", default_port)
+
+
 def inet_create_socket_active(style, address):
     try:
         is_addr_inet = is_valid_ipv4_address(address[0])
@@ -262,7 +279,7 @@ def inet_connect_active(sock, address, family, dscp):
 
 
 def inet_open_active(style, target, default_port, dscp):
-    address = inet_parse_active(target, default_port)
+    address = inet_parse_active(target, default_port, raises=False)
     family, sock = inet_create_socket_active(style, address)
     if sock is None:
         return family, sock
diff --git a/python/ovs/stream.py b/python/ovs/stream.py
index b32341076..82fbb0d68 100644
--- a/python/ovs/stream.py
+++ b/python/ovs/stream.py
@@ -784,7 +784,7 @@ class SSLStream(Stream):
 
     @staticmethod
     def _open(suffix, dscp):
-        address = ovs.socket_util.inet_parse_active(suffix, 0)
+        address = ovs.socket_util.inet_parse_active(suffix, 0, raises=False)
         family, sock = ovs.socket_util.inet_create_socket_active(
                 socket.SOCK_STREAM, address)
         if sock is None:
diff --git a/python/ovs/tests/test_dns_resolve.py b/python/ovs/tests/test_dns_resolve.py
new file mode 100644
index 000000000..b4f7f9ba2
--- /dev/null
+++ b/python/ovs/tests/test_dns_resolve.py
@@ -0,0 +1,270 @@
+import contextlib
+import ipaddress
+import sys
+import time
+from unittest import mock
+
+import pytest
+
+from ovs import dns_resolve
+from ovs import socket_util
+
+
+HOSTS = [("192.0.2.1", "fake.ip4.domain", "192.0.2.1"),
+         ("2001:db8:2::1", "fake.ip6.domain", "2001:db8:2::1"),
+         ("192.0.2.2", "fake.both.domain", "192.0.2.2"),
+         ("2001:db8:2::2", "fake.both.domain", "192.0.2.2")]
+
+
+def _tmp_file(path, content):
+    path.write_text(content)
+    assert content == path.read_text()
+    return path
+
+
+@pytest.fixture(params=[False, True], ids=["not_daemon", "daemon"])
+def resolver_factory(monkeypatch, tmp_path, request):
+    # Allow delaying the instantiation of the DNSResolver
+    def resolver_factory(hosts=HOSTS):
+        path = tmp_path / "hosts"
+        content = "\n".join(f"{ip}\t{host}" for ip, host, _ in hosts)
+        _tmp_file(path, content)
+
+        with monkeypatch.context() as m:
+            m.setenv("OVS_HOSTS_FILE", str(path))
+            # Test with both is_daemon False and True
+            resolver = dns_resolve.DNSResolver(request.param)
+            assert resolver._is_daemon == request.param
+            return dns_resolve.DNSResolver(request.param)
+
+    yield resolver_factory
+    dns_resolve.DNSResolver._instance = None
+
+
+@contextlib.contextmanager
+def DNSResolver(*args, **kwargs):
+    """Clean up after returning a dns_resolver.DNSResolver
+
+    Since it is a singleton, and pytest runs all tests in the same process,
+    we can't use dns_resolver.DNSResolver directly in these tests. This
+    context manager will reset the singleton at the end of the with block.
+    """
+    resolver = dns_resolve.DNSResolver(*args, **kwargs)
+    try:
+        yield resolver
+    finally:
+        dns_resolve.DNSResolver._instance = None
+
+
+@pytest.fixture
+def unbound_conf(tmp_path):
+    path = tmp_path / "unbound.conf"
+    content = """
+    server:
+        verbosity: 1
+    """
+    return _tmp_file(path, content)
+
+
+@pytest.fixture
+def resolv_conf(tmp_path):
+    path = tmp_path / "resolv.conf"
+    content = "nameserver 127.0.0.1"
+    return _tmp_file(path, content)
+
+
+@pytest.fixture
+def hosts_file(tmp_path):
+    path = tmp_path / "hosts"
+    content = "127.0.0.1\tfakelocalhost.localdomain"
+    return _tmp_file(path, content)
+
+
+@pytest.fixture
+def missing_file(tmp_path):
+    f = tmp_path / "missing_file"
+    assert not f.exists()
+    return f
+
+
+@pytest.fixture(params=[False, True], ids=["with unbound", "without unbound"])
+def missing_unbound(monkeypatch, request):
+    if request.param:
+        monkeypatch.setitem(sys.modules, 'unbound', None)
+        monkeypatch.delitem(dns_resolve.__dict__, "unbound")
+    return request.param
+
+
+def test_missing_unbound(missing_unbound, resolver_factory):
+    resolver = resolver_factory()  # Dont fail even w/o unbound
+    assert resolver.dns_enabled == (not missing_unbound)
+
+
+def test_DNSRequest_defaults():
+    req = dns_resolve.DNSRequest(HOSTS[0][1])
+    assert HOSTS[0][1] == req.name
+    assert req.state == req.INVALID
+    assert req.time == req.result == req.ttl is None
+    assert str(req)
+
+
+def test_DNSResolver_singleton():
+    with DNSResolver(True) as r1:
+        assert r1._is_daemon
+        r2 = dns_resolve.DNSResolver(False)
+        assert r1 == r2
+        assert r1._is_daemon
+
+
+def _resolve(resolver, host, fn=dns_resolve.resolve):
+    """Handle sync/async lookups, giving up if more than 1 second has passed"""
+
+    timeout = 1
+    start = time.time()
+    name = fn(host)
+    if resolver._is_daemon:
+        while name is None:
+            name = fn(host)
+            if name:
+                break
+            time.sleep(0.01)
+            end = time.time()
+            if end - start > timeout:
+                break
+    if name:
+        return name
+    raise LookupError(f"{host} not found")
+
+
+@pytest.mark.parametrize("ip,host,expected", HOSTS)
+def test_resolve_addresses(missing_unbound, resolver_factory, ip, host,
+                           expected):
+    resolver = resolver_factory()
+    if missing_unbound:
+        with pytest.raises(LookupError):
+            _resolve(resolver, host)
+    else:
+        result = _resolve(resolver, host)
+        assert ipaddress.ip_address(expected) == ipaddress.ip_address(result)
+
+
+def test_resolve_unknown_host(missing_unbound, resolver_factory):
+    resolver = resolver_factory()
+    with pytest.raises(LookupError):
+        _resolve(resolver, "fake.notadomain")
+
+
+def test_resolve_process_error():
+    with DNSResolver(True) as resolver:
+        with mock.patch.object(resolver._ctx, "process", return_value=-1):
+            assert resolver.resolve("fake.domain") is None
+
+
+def test_resolve_resolve_error():
+    with DNSResolver(False) as resolver:
+        with mock.patch.object(resolver._ctx, "resolve",
+                               return_value=(-1, None)):
+            assert resolver.resolve("fake.domain") is None
+
+
+def test_resolve_resolve_async_error():
+    with DNSResolver(True) as resolver:
+        with mock.patch.object(resolver._ctx, "resolve_async",
+                               return_value=(-1, None)):
+            with pytest.raises(LookupError):
+                _resolve(resolver, "fake.domain")
+
+
+@pytest.mark.parametrize("file,raises",
+                         [(None, False),
+                          ("missing_file", dns_resolve.UnboundException),
+                          ("unbound_conf", False)])
+def test_set_unbound_conf(monkeypatch, missing_unbound, resolver_factory,
+                          request, file, raises):
+    if file:
+        file = str(request.getfixturevalue(file))
+        monkeypatch.setenv("OVS_UNBOUND_CONF", file)
+    resolver = resolver_factory()  # Doesn't raise
+    if missing_unbound:
+        assert resolver._set_unbound_conf() is None
+        return
+    with mock.patch.object(resolver._ctx, "config",
+                           side_effect=resolver._ctx.config) as c:
+        if raises:
+            with pytest.raises(raises):
+                resolver._set_unbound_conf()
+        else:
+            resolver._set_unbound_conf()
+        if file:
+            c.assert_called_once_with(file)
+        else:
+            c.assert_not_called()
+
+
+@pytest.mark.parametrize("file,raises",
+                         [(None, False),
+                          ("missing_file", dns_resolve.UnboundException),
+                          ("resolv_conf", False)])
+def test_resolv_conf(monkeypatch, missing_unbound, resolver_factory, request,
+                     file, raises):
+    if file:
+        file = str(request.getfixturevalue(file))
+        monkeypatch.setenv("OVS_RESOLV_CONF", file)
+    resolver = resolver_factory()  # Doesn't raise
+    if missing_unbound:
+        assert resolver._set_resolv_conf() is None
+        return
+    with mock.patch.object(resolver._ctx, "resolvconf",
+                           side_effect=resolver._ctx.resolvconf) as c:
+        if raises:
+            with pytest.raises(raises):
+                resolver._set_resolv_conf()
+        else:
+            resolver._set_resolv_conf()
+        c.assert_called_once_with(file)
+
+
+@pytest.mark.parametrize("file,raises",
+                         [(None, False),
+                          ("missing_file", dns_resolve.UnboundException),
+                          ("hosts_file", False)])
+def test_hosts(monkeypatch, missing_unbound, resolver_factory, request, file,
+               raises):
+    if file:
+        file = str(request.getfixturevalue(file))
+        monkeypatch.setenv("OVS_HOSTS_FILE", file)
+    resolver = resolver_factory()  # Doesn't raise
+    if missing_unbound:
+        assert resolver._set_hosts_file() is None
+        return
+    with mock.patch.object(resolver._ctx, "hosts",
+                           side_effect=resolver._ctx.hosts) as c:
+        if raises:
+            with pytest.raises(raises):
+                resolver._set_hosts_file()
+        else:
+            resolver._set_hosts_file()
+        c.assert_called_once_with(file)
+
+
+def test_UnboundException(missing_unbound):
+    with pytest.raises(dns_resolve.UnboundException):
+        raise dns_resolve.UnboundException("Fake exception", -1)
+
+
+@pytest.mark.parametrize("ip,host,expected", HOSTS)
+def test_inet_parse_active(resolver_factory, ip, host, expected):
+    resolver = resolver_factory()
+
+    def fn(name):
+        # Return the same thing _resolve() would so we can call
+        # this multiple times for the is_daemon=True case
+        return socket_util.inet_parse_active(f"{name}:6640", 6640,
+                                             raises=False)[0] or None
+
+    # parsing IPs still works
+    IP = _resolve(resolver, ip, fn)
+    assert ipaddress.ip_address(ip) == ipaddress.ip_address(IP)
+    # parsing hosts works
+    IP = _resolve(resolver, host, fn)
+    assert ipaddress.ip_address(IP) == ipaddress.ip_address(expected)
diff --git a/python/setup.py b/python/setup.py
index 27684c404..bcf832ce9 100644
--- a/python/setup.py
+++ b/python/setup.py
@@ -99,8 +99,7 @@ setup_args = dict(
         'Topic :: System :: Networking',
         'License :: OSI Approved :: Apache Software License',
         'Programming Language :: Python :: 3',
-        'Programming Language :: Python :: 3.4',
-        'Programming Language :: Python :: 3.5',
+        'Programming Language :: Python :: 3.6',
     ],
     ext_modules=[setuptools.Extension("ovs._json",
                                       sources=["ovs/_json.c"],
@@ -110,7 +109,8 @@ setup_args = dict(
     cmdclass={'build_ext': try_build_ext},
     install_requires=['sortedcontainers'],
     extras_require={':sys_platform == "win32"': ['pywin32 >= 1.0'],
-                    'flow': ['netaddr', 'pyparsing']},
+                    'flow': ['netaddr', 'pyparsing'],
+                    'dns': ['unbound']},
 )
 
 try:
diff --git a/rhel/openvswitch-fedora.spec.in b/rhel/openvswitch-fedora.spec.in
index 44899c1ca..343a5716d 100644
--- a/rhel/openvswitch-fedora.spec.in
+++ b/rhel/openvswitch-fedora.spec.in
@@ -113,7 +113,7 @@ Summary: Open vSwitch python3 bindings
 License: ASL 2.0
 BuildArch: noarch
 Requires: python3
-Suggests: python3-netaddr python3-pyparsing
+Suggests: python3-netaddr python3-pyparsing python3-unbound
 %{?python_provide:%python_provide python3-openvswitch = %{version}-%{release}}
 
 %description -n python3-openvswitch
