# Copyright (c) 2021, Dell Inc. or its subsidiaries. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
# See the LICENSE file for details.
#
# This file is part of NVMe STorage Appliance Services (nvme-stas).
#
# Authors: Martin Belanger <Martin.Belanger@dell.com>
#
''' Module that provides a way to retrieve discovered
services from the Avahi daemon over D-Bus.
'''
import socket
import typing
import logging
import functools
import dasbus.error
import dasbus.connection
import dasbus.client.proxy
import dasbus.client.observer
from gi.repository import GLib
from staslib import defs, conf, gutil, iputil
def _txt2dict(txt: list):
'''@param txt: A list of list of integers. The integers are the ASCII value
of printable text characters.
'''
the_dict = dict()
for list_of_chars in txt:
try:
string = functools.reduce(lambda accumulator, c: accumulator + chr(c), list_of_chars, '')
if string.isprintable():
key, val = string.split('=')
if key: # Make sure the key is not an empty string
the_dict[key.lower()] = val
except ValueError:
pass
return the_dict
def _proto2trans(protocol):
'''Return the matching transport for the given protocol.'''
if protocol is None:
return None
protocol = protocol.strip().lower()
if protocol == 'tcp':
return 'tcp'
if protocol in ('roce', 'iwarp', 'rdma'):
return 'rdma'
return None
def mk_service_key(interface, protocol, name, stype, domain):
'''Return a tuple used as a service key (unique identifier)'''
return (interface, protocol, name, stype, domain)
def fmt_service_str(interface, protocol, name, stype, domain, flags): # pylint: disable=too-many-arguments
'''Return service identifier as a string'''
return (
f'interface={interface}:{(socket.if_indextoname(interface) + ","):<9} '
f'protocol={Avahi.protocol_as_string(protocol)}, '
f'stype={stype}, '
f'domain={domain}, '
f'flags={flags}:{(Avahi.result_flags_as_string(flags) + ","):<12} '
f'name={name}'
)
class ValueRange:
'''Implement a range of values with ceiling. Once the ceiling has been
reached, then any further request for a new value will return the
ceiling (i.e last value).'''
def __init__(self, values: list):
self._values = values
self._index = 0
def get_next(self):
'''Get the next value (or last value if ceiling was reached)'''
value = self._values[self._index]
if self._index >= 0:
self._index += 1
if self._index >= len(self._values):
self._index = -1
return value
def reset(self):
'''Reset the range to start from the beginning'''
self._index = 0
# ******************************************************************************
class Service: # pylint: disable=too-many-instance-attributes
'''Object used to keep track of the services discovered from the avahi-daemon'''
interface_name = property(lambda self: self._interface_name)
interface = property(lambda self: self._interface_id)
ip_family = property(lambda self: self._ip_family)
reachable = property(lambda self: self._reachable)
protocol = property(lambda self: self._protocol_id)
key_str = property(lambda self: self._key_str)
domain = property(lambda self: self._domain)
stype = property(lambda self: self._stype)
data = property(lambda self: self._data)
name = property(lambda self: self._name)
key = property(lambda self: self._key)
ip = property(lambda self: self._ip)
def __init__(self, args, identified_cback):
self._identified_cback = identified_cback
self._interface_id = args[0]
self._protocol_id = args[1]
self._name = args[2]
self._stype = args[3]
self._domain = args[4]
self._flags = args[5]
self._ip_family = 4 if self._protocol_id == Avahi.PROTO_INET else 6
self._interface_name = socket.if_indextoname(self._interface_id).strip()
self._protocol_name = Avahi.protocol_as_string(self._protocol_id)
self._flags_str = '(' + Avahi.result_flags_as_string(self._flags) + '),'
self._key = mk_service_key(self._interface_id, self._protocol_id, self._name, self._stype, self._domain)
self._key_str = f'({self._interface_name}, {self._protocol_name}, {self._name}.{self._domain}, {self._stype})'
self._id = fmt_service_str(
self._interface_id, self._protocol_id, self._name, self._stype, self._domain, self._flags
)
self._connect_check_retry_tmo = ValueRange([2, 5, 10, 30, 60, 300, 600])
self._connect_check_retry_tmr = gutil.GTimer(
self._connect_check_retry_tmo.get_next(), self._on_connect_check_retry
)
self._ip = None
self._resolver = None
self._data = {}
self._reachable = False
self._connect_checker = None
def info(self):
'''Return debug info'''
info = self._data
info['reachable'] = str(self._reachable)
return info
def __str__(self):
return self._id
def set_identity(self, transport, address, port, txt): # pylint: disable=too-many-arguments
'''Complete identification and check connectivity (if needed)
Return True if identification is complete. Return False if
we need to check connectivity.
'''
traddr = address.strip()
trsvcid = str(port).strip()
# host-iface permitted for tcp alone and not rdma
host_iface = self._interface_name if transport == 'tcp' else ''
self._data = {
'transport': transport,
'traddr': traddr,
'trsvcid': trsvcid,
# host-iface permitted for tcp alone and not rdma
'host-iface': host_iface,
'subsysnqn': txt.get('nqn', defs.WELL_KNOWN_DISC_NQN).strip()
if conf.NvmeOptions().discovery_supp
else defs.WELL_KNOWN_DISC_NQN,
}
self._ip = iputil.get_ipaddress_obj(traddr, ipv4_mapped_convert=True)
if transport != 'tcp':
self._reachable = True
self._identified_cback()
return
self._connect_check(verbose=True) # Enable verbosity on first attempt
def _connect_check(self, verbose=False):
self._reachable = False
connect_checker = gutil.TcpChecker(
self._data['traddr'],
self._data['trsvcid'],
self._data['host-iface'],
verbose,
self._tcp_connect_check_cback,
)
try:
connect_checker.connect()
except RuntimeError as err:
logging.error('Unable to verify connectivity: %s', err)
connect_checker.close()
connect_checker = None
self._connect_checker = connect_checker
def _tcp_connect_check_cback(self, connected):
if self._connect_checker is not None:
self._connect_checker.close()
self._connect_checker = None
self._reachable = connected
if self._reachable:
self._identified_cback()
else:
# Restart the timer but with incremented timeout
self._connect_check_retry_tmr.start(self._connect_check_retry_tmo.get_next())
def _on_connect_check_retry(self):
self._connect_check()
return GLib.SOURCE_REMOVE
def set_resolver(self, resolver):
'''Set the resolver object'''
self._resolver = resolver
def close(self):
'''Close this object and release all resources'''
if self._connect_checker is not None:
self._connect_checker.close()
self._connect_checker = None
if self._resolver is not None:
try:
self._resolver.Free()
dasbus.client.proxy.disconnect_proxy(self._resolver)
except (AttributeError, dasbus.error.DBusError) as ex:
logging.debug('Service.close() - Failed to Free() resolver. %s', ex)
self._resolver = None
# ******************************************************************************
class Avahi: # pylint: disable=too-many-instance-attributes
'''@brief Avahi Server proxy. Set up the D-Bus connection to the Avahi
daemon and register to be notified when services of a certain
type (stype) are discovered or lost.
'''
DBUS_NAME = 'org.freedesktop.Avahi'
DBUS_INTERFACE_SERVICE_BROWSER = DBUS_NAME + '.ServiceBrowser'
DBUS_INTERFACE_SERVICE_RESOLVER = DBUS_NAME + '.ServiceResolver'
LOOKUP_USE_MULTICAST = 2
IF_UNSPEC = -1
PROTO_INET = 0
PROTO_INET6 = 1
PROTO_UNSPEC = -1
LOOKUP_RESULT_LOCAL = 8 # This record/service resides on and was announced by the local host
LOOKUP_RESULT_CACHED = 1 # This response originates from the cache
LOOKUP_RESULT_STATIC = 32 # The returned data has been defined statically by some configuration option
LOOKUP_RESULT_OUR_OWN = 16 # This service belongs to the same local client as the browser object
LOOKUP_RESULT_WIDE_AREA = 2 # This response originates from wide area DNS
LOOKUP_RESULT_MULTICAST = 4 # This response originates from multicast DNS
result_flags = {
LOOKUP_RESULT_LOCAL: 'local',
LOOKUP_RESULT_CACHED: 'cache',
LOOKUP_RESULT_STATIC: 'static',
LOOKUP_RESULT_OUR_OWN: 'own',
LOOKUP_RESULT_WIDE_AREA: 'wan',
LOOKUP_RESULT_MULTICAST: 'mcast',
}
protos = {PROTO_INET: 'IPv4', PROTO_INET6: 'IPv6', PROTO_UNSPEC: 'uspecified'}
@classmethod
def result_flags_as_string(cls, flags):
'''Convert flags to human-readable string'''
return '+'.join((value for flag, value in Avahi.result_flags.items() if (flags & flag) != 0))
@classmethod
def protocol_as_string(cls, proto):
'''Convert protocol codes to human-readable strings'''
return Avahi.protos.get(proto, 'unknown')
# ==========================================================================
def __init__(self, sysbus, change_cb):
self._change_cb = change_cb
self._services = dict()
self._sysbus = sysbus
self._stypes = set()
self._service_browsers = dict()
# Avahi is an on-demand service. If, for some reason, the avahi-daemon
# were to stop, we need to try to contact it for it to restart. For
# example, when installing the avahi-daemon package on a running system,
# the daemon doesn't get started right away. It needs another process to
# access it over D-Bus to wake it up. The following timer is used to
# periodically query the avahi-daemon until we successfully establish
# first contact.
self._kick_avahi_tmr = gutil.GTimer(60, self._on_kick_avahi)
# Subscribe for Avahi signals (i.e. events). This must be done before
# any Browser or Resolver is created to avoid race conditions and
# missed events.
self._subscriptions = [
self._sysbus.connection.signal_subscribe(
Avahi.DBUS_NAME,
Avahi.DBUS_INTERFACE_SERVICE_BROWSER,
'ItemNew',
None,
None,
0,
self._service_discovered,
),
self._sysbus.connection.signal_subscribe(
Avahi.DBUS_NAME,
Avahi.DBUS_INTERFACE_SERVICE_BROWSER,
'ItemRemove',
None,
None,
0,
self._service_removed,
),
self._sysbus.connection.signal_subscribe(
Avahi.DBUS_NAME, Avahi.DBUS_INTERFACE_SERVICE_BROWSER, 'Failure', None, None, 0, self._failure_handler
),
self._sysbus.connection.signal_subscribe(
Avahi.DBUS_NAME, Avahi.DBUS_INTERFACE_SERVICE_RESOLVER, 'Found', None, None, 0, self._service_identified
),
self._sysbus.connection.signal_subscribe(
Avahi.DBUS_NAME, Avahi.DBUS_INTERFACE_SERVICE_RESOLVER, 'Failure', None, None, 0, self._failure_handler
),
]
self._avahi = self._sysbus.get_proxy(Avahi.DBUS_NAME, '/')
self._avahi_watcher = dasbus.client.observer.DBusObserver(self._sysbus, Avahi.DBUS_NAME)
self._avahi_watcher.service_available.connect(self._avahi_available)
self._avahi_watcher.service_unavailable.connect(self._avahi_unavailable)
self._avahi_watcher.connect_once_available()
def kill(self):
'''@brief Clean up object'''
logging.debug('Avahi.kill()')
self._kick_avahi_tmr.kill()
self._kick_avahi_tmr = None
for subscription in self._subscriptions:
self._sysbus.connection.signal_unsubscribe(subscription)
self._subscriptions = list()
self._disconnect()
self._avahi_watcher.service_available.disconnect()
self._avahi_watcher.service_unavailable.disconnect()
self._avahi_watcher.disconnect()
self._avahi_watcher = None
dasbus.client.proxy.disconnect_proxy(self._avahi)
self._avahi = None
self._change_cb = None
self._sysbus = None
def info(self) -> dict:
'''@brief return debug info about this object'''
info = {
'avahi wake up timer': str(self._kick_avahi_tmr),
'service types': list(self._stypes),
'services': {service.key_str: service.info() for service in self._services.values()},
}
return info
def get_controllers(self) -> list:
'''@brief Get the discovery controllers as a list of dict()
as follows:
[
{
'transport': tcp,
'traddr': str(),
'trsvcid': str(),
'subsysnqn': 'nqn.2014-08.org.nvmexpress.discovery',
'host-traddr': str(),
'host-iface': str(),
'host-nqn': str(),
},
{
'transport': tcp,
'traddr': str(),
'trsvcid': str(),
'subsysnqn': 'nqn.2014-08.org.nvmexpress.discovery',
'host-traddr': str(),
'host-iface': str(),
'host-nqn': str(),
},
[...]
]
'''
return [service.data for service in self._services.values() if service.reachable]
def config_stypes(self, stypes: list):
'''@brief Configure the service types that we want to discover.
@param stypes: A list of services types, e.g. ['_nvme-disc._tcp']
'''
self._stypes = set(stypes)
success = self._configure_browsers()
if not success:
self._kick_avahi_tmr.start()
def kick_start(self):
'''@brief We use this to kick start the Avahi
daemon (i.e. socket activation).
'''
self._kick_avahi_tmr.clear()
def _remove_service(self, service_to_rm: typing.Tuple[int, int, str, str, str]):
service = self._services.pop(service_to_rm)
if service is not None:
service.close()
def _disconnect(self):
logging.debug('Avahi._disconnect()')
for service in self._services.values():
service.close()
self._services.clear()
for browser in self._service_browsers.values():
try:
browser.Free()
dasbus.client.proxy.disconnect_proxy(browser)
except (AttributeError, dasbus.error.DBusError) as ex:
logging.debug('Avahi._disconnect() - Failed to Free() browser. %s', ex)
self._service_browsers = dict()
def _on_kick_avahi(self):
try:
# try to contact avahi-daemon. This is just a wake
# up call in case the avahi-daemon was sleeping.
self._avahi.GetVersionString()
except dasbus.error.DBusError:
return GLib.SOURCE_CONTINUE
return GLib.SOURCE_REMOVE
def _avahi_available(self, _avahi_watcher):
'''@brief Hook up DBus signal handlers for signals from stafd.'''
logging.info('avahi-daemon service available, zeroconf supported.')
success = self._configure_browsers()
if not success:
self._kick_avahi_tmr.start()
def _avahi_unavailable(self, _avahi_watcher):
self._disconnect()
logging.warning('avahi-daemon not available, zeroconf not supported.')
self._kick_avahi_tmr.start()
def _configure_browsers(self):
stypes_cur = set(self._service_browsers.keys())
stypes_to_add = self._stypes - stypes_cur
stypes_to_rm = stypes_cur - self._stypes
logging.debug('Avahi._configure_browsers() - stypes_to_rm = %s', list(stypes_to_rm))
logging.debug('Avahi._configure_browsers() - stypes_to_add = %s', list(stypes_to_add))
for stype_to_rm in stypes_to_rm:
browser = self._service_browsers.pop(stype_to_rm, None)
if browser is not None:
try:
browser.Free()
dasbus.client.proxy.disconnect_proxy(browser)
except (AttributeError, dasbus.error.DBusError) as ex:
logging.debug('Avahi._configure_browsers() - Failed to Free() browser. %s', ex)
# Find the cached services corresponding to stype_to_rm and remove them
services_to_rm = [service.key for service in self._services.values() if service.stype == stype_to_rm]
for service_to_rm in services_to_rm:
self._remove_service(service_to_rm)
for stype in stypes_to_add:
try:
obj_path = self._avahi.ServiceBrowserNew(
Avahi.IF_UNSPEC, Avahi.PROTO_UNSPEC, stype, 'local', Avahi.LOOKUP_USE_MULTICAST
)
self._service_browsers[stype] = self._sysbus.get_proxy(Avahi.DBUS_NAME, obj_path)
except dasbus.error.DBusError as ex:
logging.debug('Avahi._configure_browsers() - Failed to contact avahi-daemon. %s', ex)
logging.warning('avahi-daemon not available, operating w/o mDNS discovery.')
return False
return True
def _service_discovered(
self,
_connection,
_sender_name: str,
_object_path: str,
_interface_name: str,
_signal_name: str,
args: typing.Tuple[int, int, str, str, str, int],
*_user_data,
):
service = Service(args, self._change_cb)
logging.debug('Avahi._service_discovered() - %s', service)
if service.key not in self._services:
try:
obj_path = self._avahi.ServiceResolverNew(
service.interface,
service.protocol,
service.name,
service.stype,
service.domain,
Avahi.PROTO_UNSPEC,
Avahi.LOOKUP_USE_MULTICAST,
)
service.set_resolver(self._sysbus.get_proxy(Avahi.DBUS_NAME, obj_path))
except dasbus.error.DBusError as ex:
logging.warning('Failed to create resolver - %s: %s', service, ex)
self._services[service.key] = service
def _service_removed(
self,
_connection,
_sender_name: str,
_object_path: str,
_interface_name: str,
_signal_name: str,
args: typing.Tuple[int, int, str, str, str, int],
*_user_data,
):
(interface, protocol, name, stype, domain, flags) = args
logging.debug(
'Avahi._service_removed() - %s',
fmt_service_str(interface, protocol, name, stype, domain, flags),
)
service_key = mk_service_key(interface, protocol, name, stype, domain)
self._remove_service(service_key)
if self._change_cb is not None:
self._change_cb()
def _service_identified( # pylint: disable=too-many-locals
self,
_connection,
_sender_name: str,
_object_path: str,
_interface_name: str,
_signal_name: str,
args: typing.Tuple[int, int, str, str, str, str, int, str, int, list, int],
*_user_data,
):
(interface, protocol, name, stype, domain, host, aprotocol, address, port, txt, flags) = args
txt = _txt2dict(txt)
logging.debug(
'Avahi._service_identified() - %s, host=%s, aprotocol=%s, port=%s, address=%s, txt=%s',
fmt_service_str(interface, protocol, name, stype, domain, flags),
host,
Avahi.protocol_as_string(aprotocol),
port,
address,
txt,
)
service_key = mk_service_key(interface, protocol, name, stype, domain)
service = self._services.get(service_key, None)
if service is not None:
transport = _proto2trans(txt.get('p'))
if transport is not None:
service.set_identity(transport, address, port, txt)
else:
logging.error(
'Received invalid/undefined protocol in mDNS TXT field: address=%s, iface=%s, TXT=%s',
address,
socket.if_indextoname(interface).strip(),
txt,
)
self._check_for_duplicate_ips()
def _failure_handler( # pylint: disable=no-self-use
self,
_connection,
_sender_name: str,
_object_path: str,
interface_name: str,
_signal_name: str,
args: typing.Tuple[str],
*_user_data,
):
(error,) = args
if 'ServiceResolver' not in interface_name or 'TimeoutError' not in error:
# ServiceResolver may fire a timeout event after being Free'd(). This seems to be normal.
logging.error('Avahi._failure_handler() - name=%s, error=%s', interface_name, error)
def _check_for_duplicate_ips(self):
'''This is to identify misconfigured networks where the
same IP addresses are discovered on two or more interfaces.'''
ips = {}
for service in self._services.values():
if service.ip is not None:
ips.setdefault(service.ip.compressed, []).append(service.interface_name)
for ip, ifaces in ips.items():
if len(ifaces) > 1:
logging.error('IP address %s was found on multiple interfaces: %s', ip, ','.join(ifaces))