hemna/rbd_iscsi.py

## rbd_iscsi.py
#    Copyright 2018 OpenStack Foundation
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.
"""RADOS iSCSI Block Device Driver"""

import json
import os
import random
import tempfile

from eventlet import greenthread
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import importutils
from oslo_utils import excutils

from cinder import exception
from cinder.i18n import _
from cinder.volume.targets import driver
from cinder.volume.drivers import rbd
from cinder import ssh_utils
from cinder import utils


LOG = logging.getLogger(__name__)

RBD_iSCSI_OPTS = [
    cfg.StrOpt('rbd_igw_host',
               default=None,
               help='The iSCSI gateway host name'),
    cfg.StrOpt('rbd_ssh_ip',
               default=None,
               help='IP Address of RBD host'),
    cfg.StrOpt('rbd_ssh_port',
               default=22,
               help='SSH port of RBD host'),
    cfg.StrOpt('rbd_ssh_user',
               default=None,
               help='Username on RBD host'),
    cfg.StrOpt('rbd_ssh_password',
               default=None,
               help='password on RBD Host'),
    cfg.StrOpt('rbd_ssh_private_key',
               default=None,
               help='filename of private key to use for SSH authentication'),
    cfg.IntOpt('rbd_ssh_conn_timeout',
               default=10,
               help='SSH connection timeout in seconds.'),
    cfg.IntOpt('rbd_ssh_min_pool_conn',
               default=1,
               help='Minimum ssh connections in the pool'),
    cfg.IntOpt('rbd_ssh_max_pool_conn',
               default=5,
               help='Maximumssh connections in the pool'),

]


CONF = cfg.CONF
CONF.register_opts(RBD_iSCSI_OPTS)


class RBDiSCSITarget(driver.Target):
    """Base iSCSI Target for RBD.

    This base target driver users ssh to execute commands on the remote
    RBD host.   It's assumed that the RBD host has iscsi targetcli
    installed and the appropriate helper as well (gwcli or lrbd).
    """

    def __init__(self, *args, **kwargs):
        super(RBDiSCSITarget, self).__init__(*args, **kwargs)
        self.protocol = 'iSCSI'
        self.sshpool = None
        self.rbd_pool = self.configuration.rbd_pool
        self.rbd_user = self.configuration.rbd_user

    def ceph_execute(self, *cmd, **kwargs):
        """Execute a command on the remote ceph IGW host."""
        check_exit_code = kwargs.pop('check_exit_code', False)
        return self._run_ssh(cmd, check_exit_code)

    def _run_ssh(self, cmd_list, check_exit_code=True, attempts=1):
        utils.check_ssh_injection(cmd_list)
        command = ' '. join(cmd_list)

        if not self.sshpool:
            rbd_ssh_ip = self.configuration.rbd_ssh_ip
            LOG.warning("using SSH IP %s", rbd_ssh_ip)
            rbd_ssh_port = self.configuration.rbd_ssh_port
            LOG.warning("using SSH Port %s", rbd_ssh_port)
            password = self.configuration.rbd_ssh_password
            privatekey = self.configuration.rbd_ssh_private_key
            LOG.warning("using SSH KEY %s", privatekey)
            min_size = self.configuration.rbd_ssh_min_pool_conn
            max_size = self.configuration.rbd_ssh_max_pool_conn
            timeout = self.configuration.rbd_ssh_conn_timeout
            LOG.warning("Using ssh pool connection timeout of %s",
                        timeout)
            self.sshpool = ssh_utils.SSHPool(
                rbd_ssh_ip,
                rbd_ssh_port,
                timeout,
                login=self.configuration.rbd_ssh_user,
                password=password,
                privatekey=privatekey,
                min_size=min_size,
                max_size=max_size)
        last_exception = None
        try:
            with self.sshpool.item() as ssh:
                while attempts > 0:
                    attempts -= 1
                    try:
                        (result, stderr) = processutils.ssh_execute(
                            ssh,
                            command,
                            check_exit_code=check_exit_code,
                            timeout=30)
                        return (result, stderr)

                    except Exception as e:
                        LOG.error(e)
                        last_exception = e
                        greenthread.sleep(random.randint(20, 500) / 100.0)
                try:
                    raise processutils.ProcessExecutionError(
                        exit_code=last_exception.exit_code,
                        stdout=last_exception.stdout,
                        stderr=last_exception.stderr,
                        cmd=last_exception.cmd)
                except AttributeError:
                    raise processutils.ProcessExecutionError(
                        exit_code=-1,
                        stdout="",
                        stderr="Error running SSH command",
                        cmd=command)

        except Exception:
            with excutils.save_and_reraise_exception():
                LOG.error("Error running SSH command: %s", command)


class RBDiSCSILRBDTarget(RBDiSCSITarget):
    """LRBD iSCSI target driver.

       This target driver uses ssh to talk to the
       remote ceph host that has lrbd on it for
       creating iSCSI Targets.   lrbd is typically
       on SUSE base systems.
    """

    def __init__(self, *args, **kwargs):
        super(RBDiSCSILRBDTarget, self).__init__(*args, **kwargs)

        self.lrbd_config = {}

    @utils.trace
    def get_igw_iqn(self):
        """Get the IQN of the gateway host."""
        (result, stderr) = self.ceph_execute('sudo',
                                             'grep',
                                             '"InitiatorName="',
                                             '/etc/iscsi/initiatorname.iscsi')
        LOG.warning("Got result of %s", result)
        LOG.warning("Got stderr of %s", stderr)
        return result.replace('InitiatorName=', '').strip()

    @utils.trace
    def get_igw_host_from_iqn(self):
        """Get the IGW's target hostname from IQN."""
        iqn = self.get_igw_iqn()
        if ('targets' in self.lrbd_config and
                len(self.lrbd_config) > 0):
            targets = self.lrbd_config['targets']
            for target in targets:
                if target['target'] == iqn:
                    return target['host']

    def lrbd_get_config(self):
        """Fetch the config from lrbd."""
        (result, stderr) = self.ceph_execute('sudo', 'lrbd', '-o')
        LOG.warning("Got result of %s", result)
        LOG.warning("Got stderr of %s", stderr)
        result = result.strip()
        self.lrbd_config = eval(result)
        LOG.warning("Current Config = %s", self.lrbd_config)

    def lrbd_reload_config(self, remotepath):
        (result, stderr) = self.ceph_execute('sudo', 'lrbd', '-f',
                                             remotepath)
        LOG.warning("Got result of %s", result)
        LOG.warning("Got stderr of %s", stderr)
        # Now force lrbd to use the new config
        (result, stderr) = self.ceph_execute('sudo', 'lrbd')
        LOG.warning("Got result of %s", result)
        LOG.warning("Got stderr of %s", stderr)

    def lrbd_update_config(self):
        """Update the lrbd config with changes."""
        # First save the updated json config to tmp
        tmpdir = tempfile.gettempdir()
        if not os.path.exists(tmpdir):
            os.makedirs(tmpdir)

        with tempfile.NamedTemporaryFile(prefix='lrbd_',
                                         dir=tmpdir) as local_config:
            with open(local_config.name, 'w') as f:
                f.write(json.dumps(self.lrbd_config))

            # Now use paramiko to scp the file to the lrbd host
            ssh = self.sshpool.get()
            sftp = ssh.open_sftp()
            remotepath = "/tmp/lrbd_config.json"
            sftp.put(local_config.name, remotepath)
            sftp.close()
            ssh.close()

            # now force lrbd to reconfig based on the new file.
            self.lrbd_reload_config(remotepath)

    @utils.trace
    def get_gateway_from_host(self, target_host):
        """Find the right gateway descriptor from host."""
        gateways = []
        for pool in self.lrbd_config['pools']:
            if pool["pool"] == self.rbd_pool:
                gateways = pool['gateways']

        for gateway in gateways:
            if gateway['host'] == target_host:
                return gateway

    @utils.trace
    def update_gateway_for_host(self, target_host, gateway):
        pool_idx = 0
        while pool_idx < len(self.lrbd_config['pools']):
            if self.lrbd_config['pools'][pool_idx]["pool"] == self.rbd_pool:
                break
            pool_idx += 1

        pool = self.lrbd_config['pools'][pool_idx]
        gateway_idx = 0
        while (gateway_idx < len(
              self.lrbd_config['pools'][pool_idx]['gateways'])):
            gateway_tmp = pool['gateways'][gateway_idx]
            if gateway_tmp['host'] == target_host:
                break
            gateway_idx += 1

        self.lrbd_config['pools'][pool_idx]['gateways'][gateway_idx] = gateway

    def add_tpg(self, volume, connector):
        """Add new volume export to the pool."""
        self.lrbd_get_config()
        target_host = self.get_igw_host_from_iqn()
        # use the first portal we find
        if ("portals" in self.lrbd_config and
                len(self.lrbd_config["portals"]) > 0):
            portal = self.lrbd_config["portals"][0]
            portal_name = portal["name"]
        else:
            # we don't have any portals
            # Storage admin needs to create
            # the portals they want to export
            # the images
            # https://github.com/SUSE/lrbd/wiki
            raise exception.Invalid("No valid Portals exist on ceph iscsi "
                                    "gw host.  Please create a portal with "
                                    "lrbd.")

        vol_name = utils.convert_str(volume.name)

        tpg = {
            "portal": portal_name,
            "image": vol_name,
            "initiator": connector['initiator']
        }

        LOG.debug("Adding new TPG to lrbd config  %s", tpg)
        gateway = self.get_gateway_from_host(target_host)
        LOG.debug("Gateway %s", gateway)
        gateway['tpg'] = gateway['tpg'] + [tpg]
        LOG.debug("New Gateway %s", gateway)
        self.update_gateway_for_host(target_host, gateway)
        LOG.debug("New Config %s", self.lrbd_config)
        self.lrbd_update_config()

    @utils.trace
    def ensure_export(self, context, volume, volume_path):
        pass

    @utils.trace
    def create_export(self, context, volume, volume_path):
        pass

    @utils.trace
    def remove_export(self, context, volume):
        pass

    @utils.trace
    def initialize_connection(self, volume, connector):

        LOG.warning("Volume = %s", volume)
        LOG.warning("Connector = %s", connector)
        self.add_tpg(volume, connector)
        iscsi_properties = {}

        return {
            'driver_volume_type': self.protocol,
            'data': iscsi_properties
        }

    @utils.trace
    def terminate_connection(self, volume, connector, **kwargs):
        pass


class RBDiSCSIDriver(rbd.RBDDriver):
    """Implements iSCSI driver for ceph (RBD) Volumes.

       This driver uses the base RBDDriver for volume
       provisioning and management.  It uses a custom
       target driver for exporting ceph iSCSI volumes
       using either gwcli or lrbd.
    """

    VERSION = '1.0.0'

    CI_WIKI_NAME = "Cinder_Jenkins"

    def __init__(self, *args, **kwargs):
        super(RBDiSCSIDriver, self).__init__(*args, **kwargs)
        self.configuration.append_config_values(RBD_iSCSI_OPTS)

        self.target_mapping = {
            'rbd_iscsi_lrbd': 'cinder.volume.drivers.rbd_iscsi.'
                              'RBDiSCSILRBDTarget'
        }

        target_helper = self.configuration.safe_get('target_helper')
        if target_helper and target_helper in self.target_mapping:
            target_driver = self.target_mapping[target_helper]
        else:
            LOG.error("Tried to use an invalid target_helper for "
                      "RBDiSCSIDriver.  Must use a target_helper from "
                      "The following list %s" %
                      list(self.target_mapping.keys()))

        LOG.debug('Attempting to initialize RBD iSCSI driver with the '
                  'following target driver: %s', target_driver)

        self.target_driver = importutils.import_object(
            target_driver,
            configuration=self.configuration,
            db=self.db,
            executor=self._execute)

    @utils.trace
    def initialize_connection(self, volume, connector):
        return self.target_driver.initialize_connection(volume, connector)

    @utils.trace
    def terminate_connection(self, volume, connector, **kwargs):
        return self.target_driver.terminate_connection(
            volume, connector, **kwargs)
	# Copyright 2018 OpenStack Foundation
	#
	# Licensed under the Apache License, Version 2.0 (the "License"); you may
	# not use this file except in compliance with the License. You may obtain
	# a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing, software
	# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
	# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
	# License for the specific language governing permissions and limitations
	# under the License.
	"""RADOS iSCSI Block Device Driver"""

	import json
	import os
	import random
	import tempfile

	from eventlet import greenthread
	from oslo_concurrency import processutils
	from oslo_config import cfg
	from oslo_log import log as logging
	from oslo_utils import importutils
	from oslo_utils import excutils

	from cinder import exception
	from cinder.i18n import _
	from cinder.volume.targets import driver
	from cinder.volume.drivers import rbd
	from cinder import ssh_utils
	from cinder import utils


	LOG = logging.getLogger(__name__)

	RBD_iSCSI_OPTS = [
	cfg.StrOpt('rbd_igw_host',
	default=None,
	help='The iSCSI gateway host name'),
	cfg.StrOpt('rbd_ssh_ip',
	default=None,
	help='IP Address of RBD host'),
	cfg.StrOpt('rbd_ssh_port',
	default=22,
	help='SSH port of RBD host'),
	cfg.StrOpt('rbd_ssh_user',
	default=None,
	help='Username on RBD host'),
	cfg.StrOpt('rbd_ssh_password',
	default=None,
	help='password on RBD Host'),
	cfg.StrOpt('rbd_ssh_private_key',
	default=None,
	help='filename of private key to use for SSH authentication'),
	cfg.IntOpt('rbd_ssh_conn_timeout',
	default=10,
	help='SSH connection timeout in seconds.'),
	cfg.IntOpt('rbd_ssh_min_pool_conn',
	default=1,
	help='Minimum ssh connections in the pool'),
	cfg.IntOpt('rbd_ssh_max_pool_conn',
	default=5,
	help='Maximumssh connections in the pool'),

	]


	CONF = cfg.CONF
	CONF.register_opts(RBD_iSCSI_OPTS)


	class RBDiSCSITarget(driver.Target):
	"""Base iSCSI Target for RBD.

	This base target driver users ssh to execute commands on the remote
	RBD host. It's assumed that the RBD host has iscsi targetcli
	installed and the appropriate helper as well (gwcli or lrbd).
	"""

	def __init__(self, args, *kwargs):
	super(RBDiSCSITarget, self).__init__(args, *kwargs)
	self.protocol = 'iSCSI'
	self.sshpool = None
	self.rbd_pool = self.configuration.rbd_pool
	self.rbd_user = self.configuration.rbd_user

	def ceph_execute(self, cmd, *kwargs):
	"""Execute a command on the remote ceph IGW host."""
	check_exit_code = kwargs.pop('check_exit_code', False)
	return self._run_ssh(cmd, check_exit_code)

	def _run_ssh(self, cmd_list, check_exit_code=True, attempts=1):
	utils.check_ssh_injection(cmd_list)
	command = ' '. join(cmd_list)

	if not self.sshpool:
	rbd_ssh_ip = self.configuration.rbd_ssh_ip
	LOG.warning("using SSH IP %s", rbd_ssh_ip)
	rbd_ssh_port = self.configuration.rbd_ssh_port
	LOG.warning("using SSH Port %s", rbd_ssh_port)
	password = self.configuration.rbd_ssh_password
	privatekey = self.configuration.rbd_ssh_private_key
	LOG.warning("using SSH KEY %s", privatekey)
	min_size = self.configuration.rbd_ssh_min_pool_conn
	max_size = self.configuration.rbd_ssh_max_pool_conn
	timeout = self.configuration.rbd_ssh_conn_timeout
	LOG.warning("Using ssh pool connection timeout of %s",
	timeout)
	self.sshpool = ssh_utils.SSHPool(
	rbd_ssh_ip,
	rbd_ssh_port,
	timeout,
	login=self.configuration.rbd_ssh_user,
	password=password,
	privatekey=privatekey,
	min_size=min_size,
	max_size=max_size)
	last_exception = None
	try:
	with self.sshpool.item() as ssh:
	while attempts > 0:
	attempts -= 1
	try:
	(result, stderr) = processutils.ssh_execute(
	ssh,
	command,
	check_exit_code=check_exit_code,
	timeout=30)
	return (result, stderr)

	except Exception as e:
	LOG.error(e)
	last_exception = e
	greenthread.sleep(random.randint(20, 500) / 100.0)
	try:
	raise processutils.ProcessExecutionError(
	exit_code=last_exception.exit_code,
	stdout=last_exception.stdout,
	stderr=last_exception.stderr,
	cmd=last_exception.cmd)
	except AttributeError:
	raise processutils.ProcessExecutionError(
	exit_code=-1,
	stdout="",
	stderr="Error running SSH command",
	cmd=command)

	except Exception:
	with excutils.save_and_reraise_exception():
	LOG.error("Error running SSH command: %s", command)


	class RBDiSCSILRBDTarget(RBDiSCSITarget):
	"""LRBD iSCSI target driver.

	This target driver uses ssh to talk to the
	remote ceph host that has lrbd on it for
	creating iSCSI Targets. lrbd is typically
	on SUSE base systems.
	"""

	def __init__(self, args, *kwargs):
	super(RBDiSCSILRBDTarget, self).__init__(args, *kwargs)

	self.lrbd_config = {}

	@utils.trace
	def get_igw_iqn(self):
	"""Get the IQN of the gateway host."""
	(result, stderr) = self.ceph_execute('sudo',
	'grep',
	'"InitiatorName="',
	'/etc/iscsi/initiatorname.iscsi')
	LOG.warning("Got result of %s", result)
	LOG.warning("Got stderr of %s", stderr)
	return result.replace('InitiatorName=', '').strip()

	@utils.trace
	def get_igw_host_from_iqn(self):
	"""Get the IGW's target hostname from IQN."""
	iqn = self.get_igw_iqn()
	if ('targets' in self.lrbd_config and
	len(self.lrbd_config) > 0):
	targets = self.lrbd_config['targets']
	for target in targets:
	if target['target'] == iqn:
	return target['host']

	def lrbd_get_config(self):
	"""Fetch the config from lrbd."""
	(result, stderr) = self.ceph_execute('sudo', 'lrbd', '-o')
	LOG.warning("Got result of %s", result)
	LOG.warning("Got stderr of %s", stderr)
	result = result.strip()
	self.lrbd_config = eval(result)
	LOG.warning("Current Config = %s", self.lrbd_config)

	def lrbd_reload_config(self, remotepath):
	(result, stderr) = self.ceph_execute('sudo', 'lrbd', '-f',
	remotepath)
	LOG.warning("Got result of %s", result)
	LOG.warning("Got stderr of %s", stderr)
	# Now force lrbd to use the new config
	(result, stderr) = self.ceph_execute('sudo', 'lrbd')
	LOG.warning("Got result of %s", result)
	LOG.warning("Got stderr of %s", stderr)

	def lrbd_update_config(self):
	"""Update the lrbd config with changes."""
	# First save the updated json config to tmp
	tmpdir = tempfile.gettempdir()
	if not os.path.exists(tmpdir):
	os.makedirs(tmpdir)

	with tempfile.NamedTemporaryFile(prefix='lrbd_',
	dir=tmpdir) as local_config:
	with open(local_config.name, 'w') as f:
	f.write(json.dumps(self.lrbd_config))

	# Now use paramiko to scp the file to the lrbd host
	ssh = self.sshpool.get()
	sftp = ssh.open_sftp()
	remotepath = "/tmp/lrbd_config.json"
	sftp.put(local_config.name, remotepath)
	sftp.close()
	ssh.close()

	# now force lrbd to reconfig based on the new file.
	self.lrbd_reload_config(remotepath)

	@utils.trace
	def get_gateway_from_host(self, target_host):
	"""Find the right gateway descriptor from host."""
	gateways = []
	for pool in self.lrbd_config['pools']:
	if pool["pool"] == self.rbd_pool:
	gateways = pool['gateways']

	for gateway in gateways:
	if gateway['host'] == target_host:
	return gateway

	@utils.trace
	def update_gateway_for_host(self, target_host, gateway):
	pool_idx = 0
	while pool_idx < len(self.lrbd_config['pools']):
	if self.lrbd_config['pools'][pool_idx]["pool"] == self.rbd_pool:
	break
	pool_idx += 1

	pool = self.lrbd_config['pools'][pool_idx]
	gateway_idx = 0
	while (gateway_idx < len(
	self.lrbd_config['pools'][pool_idx]['gateways'])):
	gateway_tmp = pool['gateways'][gateway_idx]
	if gateway_tmp['host'] == target_host:
	break
	gateway_idx += 1

	self.lrbd_config['pools'][pool_idx]['gateways'][gateway_idx] = gateway

	def add_tpg(self, volume, connector):
	"""Add new volume export to the pool."""
	self.lrbd_get_config()
	target_host = self.get_igw_host_from_iqn()
	# use the first portal we find
	if ("portals" in self.lrbd_config and
	len(self.lrbd_config["portals"]) > 0):
	portal = self.lrbd_config["portals"][0]
	portal_name = portal["name"]
	else:
	# we don't have any portals
	# Storage admin needs to create
	# the portals they want to export
	# the images
	# https://github.com/SUSE/lrbd/wiki
	raise exception.Invalid("No valid Portals exist on ceph iscsi "
	"gw host. Please create a portal with "
	"lrbd.")

	vol_name = utils.convert_str(volume.name)

	tpg = {
	"portal": portal_name,
	"image": vol_name,
	"initiator": connector['initiator']
	}

	LOG.debug("Adding new TPG to lrbd config %s", tpg)
	gateway = self.get_gateway_from_host(target_host)
	LOG.debug("Gateway %s", gateway)
	gateway['tpg'] = gateway['tpg'] + [tpg]
	LOG.debug("New Gateway %s", gateway)
	self.update_gateway_for_host(target_host, gateway)
	LOG.debug("New Config %s", self.lrbd_config)
	self.lrbd_update_config()

	@utils.trace
	def ensure_export(self, context, volume, volume_path):
	pass

	@utils.trace
	def create_export(self, context, volume, volume_path):
	pass

	@utils.trace
	def remove_export(self, context, volume):
	pass

	@utils.trace
	def initialize_connection(self, volume, connector):

	LOG.warning("Volume = %s", volume)
	LOG.warning("Connector = %s", connector)
	self.add_tpg(volume, connector)
	iscsi_properties = {}

	return {
	'driver_volume_type': self.protocol,
	'data': iscsi_properties
	}

	@utils.trace
	def terminate_connection(self, volume, connector, **kwargs):
	pass


	class RBDiSCSIDriver(rbd.RBDDriver):
	"""Implements iSCSI driver for ceph (RBD) Volumes.

	This driver uses the base RBDDriver for volume
	provisioning and management. It uses a custom
	target driver for exporting ceph iSCSI volumes
	using either gwcli or lrbd.
	"""

	VERSION = '1.0.0'

	CI_WIKI_NAME = "Cinder_Jenkins"

	def __init__(self, args, *kwargs):
	super(RBDiSCSIDriver, self).__init__(args, *kwargs)
	self.configuration.append_config_values(RBD_iSCSI_OPTS)

	self.target_mapping = {
	'rbd_iscsi_lrbd': 'cinder.volume.drivers.rbd_iscsi.'
	'RBDiSCSILRBDTarget'
	}

	target_helper = self.configuration.safe_get('target_helper')
	if target_helper and target_helper in self.target_mapping:
	target_driver = self.target_mapping[target_helper]
	else:
	LOG.error("Tried to use an invalid target_helper for "
	"RBDiSCSIDriver. Must use a target_helper from "
	"The following list %s" %
	list(self.target_mapping.keys()))

	LOG.debug('Attempting to initialize RBD iSCSI driver with the '
	'following target driver: %s', target_driver)

	self.target_driver = importutils.import_object(
	target_driver,
	configuration=self.configuration,
	db=self.db,
	executor=self._execute)

	@utils.trace
	def initialize_connection(self, volume, connector):
	return self.target_driver.initialize_connection(volume, connector)

	@utils.trace
	def terminate_connection(self, volume, connector, **kwargs):
	return self.target_driver.terminate_connection(
	volume, connector, **kwargs)